aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorGravatar janine9vn <[email protected]>2021-04-10 20:17:47 -0400
committerGravatar janine9vn <[email protected]>2021-04-10 20:17:47 -0400
commit0ac739dc029664332f1416ba85424c3b396e5660 (patch)
treee21ce78f68e83c6c2274fd81b02d14fbdcabef4a
parentRealigned to SirLancebot Structure (diff)
Change names
These were missed in a previous commit. It's a simple name change from the original files to better align with Sir Lancebot.
-rw-r--r--bot/exts/internal_eval/helpers.py243
-rw-r--r--bot/exts/internal_eval/internal_eval.py152
2 files changed, 0 insertions, 395 deletions
diff --git a/bot/exts/internal_eval/helpers.py b/bot/exts/internal_eval/helpers.py
deleted file mode 100644
index 5c602e4d..00000000
--- a/bot/exts/internal_eval/helpers.py
+++ /dev/null
@@ -1,243 +0,0 @@
-import ast
-import collections
-import contextlib
-import functools
-import inspect
-import io
-import logging
-import sys
-import traceback
-import types
-import typing
-
-
-log = logging.getLogger("rattlesnake.exts.admin_tools.internal_eval")
-
-# A type alias to annotate the tuples returned from `sys.exc_info()`
-ExcInfo = typing.Tuple[typing.Type[Exception], Exception, types.TracebackType]
-Namespace = typing.Dict[str, typing.Any]
-
-# This will be used as an coroutine function wrapper for the code
-# to be evaluated. The wrapper contains one `pass` statement which
-# will be replaced with `ast` with the code that we want to have
-# evaluated.
-# The function redirects output and captures exceptions that were
-# raised in the code we evaluate. The latter is used to provide a
-# meaningful traceback to the end user.
-EVAL_WRAPPER = """
-async def _eval_wrapper_function():
- try:
- with contextlib.redirect_stdout(_eval_context.stdout):
- pass
- if '_value_last_expression' in locals():
- if inspect.isawaitable(_value_last_expression):
- _value_last_expression = await _value_last_expression
- _eval_context._value_last_expression = _value_last_expression
- else:
- _eval_context._value_last_expression = None
- except Exception:
- _eval_context.exc_info = sys.exc_info()
- finally:
- _eval_context.locals = locals()
-_eval_context.function = _eval_wrapper_function
-"""
-
-
-def format_internal_eval_exception(exc_info: ExcInfo, code: str) -> str:
- """Format an exception caught while evaluation code by inserting lines."""
- exc_type, exc_value, tb = exc_info
- stack_summary = traceback.StackSummary.extract(traceback.walk_tb(tb))
- code = code.split("\n")
-
- output = ["Traceback (most recent call last):"]
- for frame in stack_summary:
- if frame.filename == "<internal eval>":
- line = code[frame.lineno - 1].lstrip()
-
- if frame.name == "_eval_wrapper_function":
- name = "<internal eval>"
- else:
- name = frame.name
- else:
- line = frame.line
- name = frame.name
-
- output.append(
- f' File "{frame.filename}", line {frame.lineno}, in {name}\n'
- f" {line}"
- )
-
- output.extend(traceback.format_exception_only(exc_type, exc_value))
- return "\n".join(output)
-
-
-class EvalContext:
- """
- Represents the current `internal eval` context.
- The context remembers names set during earlier runs of `internal eval`. To
- clear the context, use the `?internal clear` command.
- """
-
- def __init__(self, context_vars: Namespace, local_vars: Namespace) -> None:
- self._locals = dict(local_vars)
- self.context_vars = dict(context_vars)
-
- self.stdout = io.StringIO()
- self._value_last_expression = None
- self.exc_info = None
- self.code = ""
- self.function = None
- self.eval_tree = None
-
- @property
- def dependencies(self) -> typing.Dict[str, typing.Any]:
- """
- Return a mapping of the dependencies for the wrapper function.
- By using a property descriptor, the mapping can't be accidentally
- mutated during evaluation. This ensures the dependencies are always
- available.
- """
- return {
- "print": functools.partial(print, file=self.stdout),
- "contextlib": contextlib,
- "inspect": inspect,
- "sys": sys,
- "_eval_context": self,
- "_": self._value_last_expression,
- }
-
- @property
- def locals(self) -> typing.Dict[str, typing.Any]:
- """Return a mapping of names->values needed for evaluation."""
- return {**collections.ChainMap(self.dependencies, self.context_vars, self._locals)}
-
- @locals.setter
- def locals(self, locals_: typing.Dict[str, typing.Any]) -> None:
- """Update the contextual mapping of names to values."""
- log.trace(f"Updating {self._locals} with {locals_}")
- self._locals.update(locals_)
-
- def prepare_eval(self, code: str) -> typing.Optional[str]:
- """Prepare an evaluation by processing the code and setting up the context."""
- self.code = code
-
- if not self.code:
- log.debug("No code was attached to the evaluation command")
- return "[No code detected]"
-
- try:
- code_tree = ast.parse(code, filename="<internal eval>")
- except SyntaxError:
- log.debug("Got a SyntaxError while parsing the eval code")
- return "".join(traceback.format_exception(*sys.exc_info(), limit=0))
-
- log.trace("Parsing the AST to see if there's a trailing expression we need to capture")
- code_tree = CaptureLastExpression(code_tree).capture()
-
- log.trace("Wrapping the AST in the AST of the wrapper coroutine")
- eval_tree = WrapEvalCodeTree(code_tree).wrap()
-
- self.eval_tree = eval_tree
- return None
-
- async def run_eval(self) -> Namespace:
- """Run the evaluation and return the updated locals."""
- log.trace("Compiling the AST to bytecode using `exec` mode")
- compiled_code = compile(self.eval_tree, filename="<internal eval>", mode="exec")
-
- log.trace("Executing the compiled code with the desired namespace environment")
- exec(compiled_code, self.locals) # noqa: B102,S102
-
- log.trace("Awaiting the created evaluation wrapper coroutine.")
- await self.function()
-
- log.trace("Returning the updated captured locals.")
- return self._locals
-
- def format_output(self) -> str:
- """Format the output of the most recent evaluation."""
- output = []
-
- log.trace(f"Getting output from stdout `{id(self.stdout)}`")
- stdout_text = self.stdout.getvalue()
- if stdout_text:
- log.trace("Appending output captured from stdout/print")
- output.append(stdout_text)
-
- if self._value_last_expression is not None:
- log.trace("Appending the output of a captured trialing expression")
- output.append(f"[Captured] {self._value_last_expression!r}")
-
- if self.exc_info:
- log.trace("Appending exception information")
- output.append(format_internal_eval_exception(self.exc_info, self.code))
-
- log.trace(f"Generated output: {output!r}")
- return "\n".join(output) or "[No output]"
-
-
-class WrapEvalCodeTree(ast.NodeTransformer):
- """Wraps the AST of eval code with the wrapper function."""
-
- def __init__(self, eval_code_tree: ast.AST, *args, **kwargs) -> None:
- super().__init__(*args, **kwargs)
- self.eval_code_tree = eval_code_tree
-
- # To avoid mutable aliasing, parse the WRAPPER_FUNC for each wrapping
- self.wrapper = ast.parse(EVAL_WRAPPER, filename="<internal eval>")
-
- def wrap(self) -> ast.AST:
- """Wrap the tree of the code by the tree of the wrapper function."""
- new_tree = self.visit(self.wrapper)
- return ast.fix_missing_locations(new_tree)
-
- def visit_Pass(self, node: ast.Pass) -> typing.List[ast.AST]: # noqa: N802
- """
- Replace the `_ast.Pass` node in the wrapper function by the eval AST.
- This method works on the assumption that there's a single `pass`
- statement in the wrapper function.
- """
- return list(ast.iter_child_nodes(self.eval_code_tree))
-
-
-class CaptureLastExpression(ast.NodeTransformer):
- """Captures the return value from a loose expression."""
-
- def __init__(self, tree: ast.AST, *args, **kwargs) -> None:
- super().__init__(*args, **kwargs)
- self.tree = tree
- self.last_node = list(ast.iter_child_nodes(tree))[-1]
-
- def visit_Expr(self, node: ast.Expr) -> typing.Union[ast.Expr, ast.Assign]: # noqa: N802
- """
- Replace the Expr node that is last child node of Module with an assignment.
- We use an assignment to capture the value of the last node, if it's a loose
- Expr node. Normally, the value of an Expr node is lost, meaning we don't get
- the output of such a last "loose" expression. By assigning it a name, we can
- retrieve it for our output.
- """
- if node is not self.last_node:
- return node
-
- log.trace("Found a trailing last expression in the evaluation code")
-
- log.trace("Creating assignment statement with trailing expression as the right-hand side")
- right_hand_side = list(ast.iter_child_nodes(node))[0]
-
- assignment = ast.Assign(
- targets=[ast.Name(id='_value_last_expression', ctx=ast.Store())],
- value=right_hand_side,
- lineno=node.lineno,
- col_offset=0,
- )
- ast.fix_missing_locations(assignment)
- return assignment
-
- def capture(self) -> ast.AST:
- """Capture the value of the last expression with an assignment."""
- if not isinstance(self.last_node, ast.Expr):
- # We only have to replace a node if the very last node is an Expr node
- return self.tree
-
- new_tree = self.visit(self.tree)
- return ast.fix_missing_locations(new_tree)
diff --git a/bot/exts/internal_eval/internal_eval.py b/bot/exts/internal_eval/internal_eval.py
deleted file mode 100644
index f6812942..00000000
--- a/bot/exts/internal_eval/internal_eval.py
+++ /dev/null
@@ -1,152 +0,0 @@
-import logging
-import re
-import textwrap
-import typing
-
-import discord
-from discord.ext import commands
-
-from rattlesnake.bot import Rattlesnake
-from rattlesnake.constants import ADMIN_ROLES
-from rattlesnake.utils import in_whitelist
-from .helpers import EvalContext
-
-__all__ = ["InternalEval"]
-
-log = logging.getLogger("rattlesnake.exts.admin_tools.internal_eval")
-
-CODEBLOCK_REGEX = re.compile(r"(^```(py(thon)?)?\n)|(```$)")
-
-
-class InternalEval(commands.Cog):
- """Top secret code evaluation for admins and owners."""
-
- def __init__(self, bot: Rattlesnake):
- self.bot = bot
- self.locals = {}
-
- @staticmethod
- def shorten_output(
- output: str,
- max_length: int = 1900,
- placeholder: str = "\n[output truncated]"
- ) -> str:
- """
- Shorten the `output` so it's shorter than `max_length`.
- There are three tactics for this, tried in the following order:
- - Shorten the output on a line-by-line basis
- - Shorten the output on any whitespace character
- - Shorten the output solely on character count
- """
- max_length = max_length - len(placeholder)
-
- shortened_output = []
- char_count = 0
- for line in output.split("\n"):
- if char_count + len(line) > max_length:
- break
- shortened_output.append(line)
- char_count += len(line) + 1 # account for (possible) line ending
-
- if shortened_output:
- shortened_output.append(placeholder)
- return "\n".join(shortened_output)
-
- shortened_output = textwrap.shorten(output, width=max_length, placeholder=placeholder)
-
- if shortened_output.strip() == placeholder.strip():
- # `textwrap` was unable to find whitespace to shorten on, so it has
- # reduced the output to just the placeholder. Let's shorten based on
- # characters instead.
- shortened_output = output[:max_length] + placeholder
-
- return shortened_output
-
- async def _upload_output(self, output: str) -> typing.Optional[str]:
- """Upload `internal eval` output to our pastebin and return the url."""
- try:
- async with self.bot.http_session.post(
- "https://paste.pythondiscord.com/documents", data=output, raise_for_status=True
- ) as resp:
- data = await resp.json()
-
- if "key" in data:
- return f"https://paste.pythondiscord.com/{data['key']}"
- except Exception:
- # 400 (Bad Request) means there are too many characters
- log.exception("Failed to upload `internal eval` output to paste service!")
-
- async def _send_output(self, ctx: commands.Context, output: str) -> None:
- """Send the `internal eval` output to the command invocation context."""
- upload_message = ""
- if len(output) >= 1980:
- # The output is too long, let's truncate it for in-channel output and
- # upload the complete output to the paste service.
- url = await self._upload_output(output)
-
- if url:
- upload_message = f"\nFull output here: {url}"
- else:
- upload_message = "\n:warning: Failed to upload full output!"
-
- output = self.shorten_output(output)
-
- await ctx.send(f"```py\n{output}```{upload_message}")
-
- async def _eval(self, ctx: commands.Context, code: str) -> None:
- """Evaluate the `code` in the current evaluation context."""
- if code.startswith("exit"):
- self.locals = {}
- await ctx.send("The evaluation context was reset.")
- return
-
- context_vars = {
- "message": ctx.message,
- "author": ctx.message.author,
- "channel": ctx.channel,
- "guild": ctx.guild,
- "ctx": ctx,
- "self": self,
- "bot": self.bot,
- "discord": discord,
- }
-
- eval_context = EvalContext(context_vars, self.locals)
-
- log.trace("Preparing the evaluation by parsing the AST of the code")
- error = eval_context.prepare_eval(code)
-
- if error:
- log.trace("The code can't be evaluated due to an error")
- await ctx.send(f"```py\n{error}\n```")
- return
-
- log.trace("Evaluate the AST we've generated for the evaluation")
- new_locals = await eval_context.run_eval()
-
- log.trace("Updating locals with those set during evaluation")
- self.locals.update(new_locals)
-
- log.trace("Sending the formatted output back to the context")
- await self._send_output(ctx, eval_context.format_output())
-
- @commands.group(name='internal', aliases=('int',))
- @in_whitelist(roles=ADMIN_ROLES)
- async def internal_group(self, ctx: commands.Context) -> None:
- """Internal commands. Top secret!"""
- if not ctx.invoked_subcommand:
- await ctx.send_help(ctx.command)
-
- @internal_group.command(name='eval', aliases=('e',))
- @in_whitelist(roles=ADMIN_ROLES)
- async def eval(self, ctx: commands.Context, *, code: str) -> None:
- """Run eval in a REPL-like format."""
- code = CODEBLOCK_REGEX.sub("", code.strip())
- await self._eval(ctx, code)
-
- @internal_group.command(name='reset', aliases=("clear", "exit", "r", "c"))
- @in_whitelist(roles=ADMIN_ROLES)
- async def reset(self, ctx: commands.Context) -> None:
- """Run eval in a REPL-like format."""
- self.locals = {}
- await ctx.send("The evaluation context was reset.")