import ast import copy import traceback from dataclasses import dataclass, field import core.builtin_helpers import core.utils from core.ast_helpers import UnreferencedNamesVisitor, NamesWithAttributesVisitor from core.builtin_concepts import BuiltinConcepts, ParserResultConcept from core.concept import ConceptParts, Concept from core.global_symbols import NotInit, NotFound from core.rule import Rule from core.sheerka.ExecutionContext import ExecutionContext from core.tokenizer import Token, TokenKind from evaluators.BaseEvaluator import OneReturnValueEvaluator from parsers.PythonParser import PythonNode, get_python_node TO_DISABLED = ["breakpoint", "callable", "compile", "delattr", "eval", "exec", "exit", "input", "locals", "open", "print", "quit", "setattr"] def inject_context(context): """ function Decorator used to inject the context in methods that needed :param context: :return: """ def wrapped(func): def inner(*args, **kwargs): return func(context, *args, **kwargs) return inner return wrapped class Expando: def __init__(self, bag): for k, v in bag.items(): setattr(self, k, v) def __repr__(self): return f"{dir(self)}" @dataclass class PythonEvalError: error: Exception traceback: str = field(repr=False) concepts: dict = field(repr=False) def __eq__(self, other): if id(self) == id(other): return True if not isinstance(other, PythonEvalError): return False return isinstance(self.error, type(other.error)) and \ self.traceback == other.traceback and \ self.concepts == other.concepts def __hash__(self): return hash(self.error) class PythonEvaluator(OneReturnValueEvaluator): NAME = "Python" """ Evaluate a Python node, ie, evaluate some Python code """ isinstance = None def __init__(self): super().__init__(self.NAME, [BuiltinConcepts.EVALUATION], 50) @staticmethod def initialize(sheerka): from core.sheerka.services.SheerkaAdmin import SheerkaAdmin PythonEvaluator.isinstance = sheerka.services[SheerkaAdmin.NAME].extended_isinstance def matches(self, context, return_value): if not return_value.status or not isinstance(return_value.value, ParserResultConcept): return False body = return_value.value.value return isinstance(body, PythonNode) or hasattr(body, "python_node") def eval(self, context, return_value): sheerka = context.sheerka node = get_python_node(return_value.value.value) debugger = context.get_debugger(PythonEvaluator.NAME, "eval") debugger.debug_entering(node=node) exception_debugger = context.get_debugger("Exceptions", PythonEvaluator.NAME + ".eval") get_trace_back = context.debug_enabled or exception_debugger.is_enabled() context.log(f"Evaluating python node {node}.", self.name) # If we evaluate a Concept metadata which is NOT the body ex (pre, post, where...) # We need to disable the function that may alter the state # It's a poor way to have source code security check attr_under_eval = context.get_parents(lambda ec: ec.action == BuiltinConcepts.EVALUATING_ATTRIBUTE) if attr_under_eval: attr_under_eval = attr_under_eval[0] expression_only = attr_under_eval.action_context != ConceptParts.BODY if expression_only and isinstance(node.ast_, ast.Module): # Module execution is forbidden in where, pre, post and ret concept parts security_error = sheerka.new(BuiltinConcepts.PYTHON_SECURITY_ERROR, prop=attr_under_eval.action_context, body=node.source) return sheerka.ret(self.name, False, security_error, parents=[return_value]) else: expression_only = False # get globals my_globals = self.get_globals(context, node, expression_only) debugger.debug_var("globals", my_globals) all_possible_globals = self.get_all_possible_globals(context, my_globals) concepts_entries = None # entries in globals_ that refers to Concept objects evaluated = NotInit errors = [] expect_success = context.in_context(BuiltinConcepts.EVAL_UNTIL_SUCCESS_REQUESTED) for globals_ in all_possible_globals: try: # eval my_locals = {} if isinstance(node.ast_, ast.Expression): context.log("Evaluating using 'eval'.", self.name) evaluated = eval(node.get_compiled(), globals_, my_locals) else: context.log("Evaluating using 'exec'.", self.name) evaluated = self.exec_with_return(node.ast_, globals_, my_locals) # TODO find a better implementation using SheerkaMemory sheerka.locals.update(my_locals) if not expect_success or evaluated: break # in this first version, we stop once a success is found except Exception as ex: if concepts_entries is None: concepts_entries = self.get_concepts_entries_from_globals(my_globals) eval_error = PythonEvalError(ex, traceback.format_exc() if get_trace_back else None, self.get_concepts_values_from_globals(globals_, concepts_entries)) errors.append(eval_error) exception_debugger.debug_var("exception", eval_error.error, is_error=True) exception_debugger.debug_var("trace", eval_error.traceback, is_error=True) if evaluated == NotInit: if len(errors) == 1: context.log_error(errors[0].error, who=self.name, exc=errors[0].traceback) return sheerka.ret(self.name, False, sheerka.err(errors[0]), parents=[return_value]) if len(errors) > 1: for eval_error in errors: context.log_error(eval_error.error, who=self.name, exc=eval_error.traceback) return sheerka.ret(self.name, False, sheerka.err(errors), parents=[return_value]) context.log(f"{evaluated=}", self.name) debugger.debug_var("ret", evaluated) if sheerka.isinstance(evaluated, BuiltinConcepts.RETURN_VALUE): return sheerka.ret(self.name, evaluated.status, evaluated.body, parents=[return_value, evaluated]) else: return sheerka.ret(self.name, True, evaluated, parents=[return_value]) def get_globals(self, context, node, expression_only): """ Creates the globals variables :param context: :param node: :param expression_only: :return: """ unreferenced_names_visitor = UnreferencedNamesVisitor(context) names = unreferenced_names_visitor.get_names(node.ast_) if context.debug_enabled: context.debug(self.NAME, "eval", "names", names) return self.get_globals_by_names(context, names, node, expression_only) def get_sheerka_method(self, context, name, expression_only): try: method = context.sheerka.sheerka_methods[name] context.log(f"Resolving '{name}'. It's a sheerka method.", self.name) if expression_only and method.has_side_effect: context.log(f"...but with side effect when {expression_only=}. Discarding.", self.name) return None else: return inject_context(context)(method.method) if name in context.sheerka.methods_with_context \ else method.method except KeyError: return None def get_globals_by_names(self, context, names, node, expression_only): my_globals = { "Concept": core.concept.Concept, "BuiltinConcepts": core.builtin_concepts.BuiltinConcepts, "ExecutionContext": ExecutionContext, "in_context": context.in_context, } for name in names: if name in my_globals: continue if expression_only and name in TO_DISABLED: my_globals[name] = None continue # need to add it manually to avoid conflict with sheerka.isinstance if name == "isinstance": my_globals["isinstance"] = PythonEvaluator.isinstance continue # support reference to sheerka if name == "sheerka": bag = {} visitor = NamesWithAttributesVisitor() for sequence in visitor.get_sequences(node.ast_, "sheerka"): if (len(sequence) > 1 and (method := self.get_sheerka_method(context, sequence[1], expression_only)) is not None): bag[sequence[1]] = method my_globals["sheerka"] = Expando(bag) continue # search in local variables. To remove when local variables will be merged with memory if name in context.sheerka.locals: my_globals[name] = context.sheerka.locals[name] continue # search in short term memory if (obj := context.get_from_short_term_memory(name)) is not NotFound: context.log(f"Resolving '{name}'. Using value found in STM.", self.name) my_globals[name] = obj continue # search in sheerka methods if (method := self.get_sheerka_method(context, name, expression_only)) is not None: my_globals[name] = method continue # search in context.obj (to replace by short time memory ?) if context.obj: if name == "self": my_globals["self"] = context.obj continue try: attribute = context.obj.variables()[name] if attribute != NotInit: my_globals[name] = attribute continue context.log(f"Resolving '{name}'. It's obj attribute (obj={context.obj}).", self.name) except KeyError: pass # search in current node (if the name was found during the parsing) if name in node.objects: context.log(f"Resolving '{name}'. Using value from node.", self.name) obj = self.resolve_object(context, node.objects[name]) # at last, try to instantiate a new concept else: context.log(f"Resolving '{name}'. Instantiating new concept.", self.name) obj = self.resolve_object(context, name) if obj is None: context.log(f"...'{name}' is not found or cannot be instantiated. Skipping.", self.name) continue my_globals[name] = obj return my_globals @staticmethod def get_all_possible_globals(context, my_globals): """ From a dictionary of globals (str, obj) Creates as many globals as there are combination between a concept and its body Example: if the entry 'foo': Concept("foo", body="something") 2 globals will be created one with foo: Concept("foo") # we keep the concept as an object one with foo: 'something' # we substitute its value :param context: :param my_globals: :return: """ # first pass, get all the non concept or concept with no body # Note that we consider that all concepts are evaluated # In the future, it may be a good optimisation to defer the evaluation of the body # until the python evaluation fails fixed_values = {} concepts_with_body = {} for k, v in my_globals.items(): if not isinstance(v, Concept) or context.sheerka.objvalue(v) == v: fixed_values[k] = v else: concepts_with_body[k] = v # make the product the rest as cartesian product res = [fixed_values] for k, v in concepts_with_body.items(): res = core.utils.dict_product(res, [{k: v}, {k: context.sheerka.objvalue(v)}]) return res @staticmethod def get_concepts_entries_from_globals(my_globals): return [k for k, v in my_globals.items() if isinstance(v, Concept)] @staticmethod def get_concepts_values_from_globals(my_globals, names): return {name: my_globals[name] for name in names} @staticmethod def resolve_object(context, name): """ Try to find a concept by its name, id or the pattern c:key|id: :param context: :param name: :return: """ if isinstance(name, Rule): return context.sheerka.resolve_rule(context, name) if isinstance(name, Concept): name = core.builtin_helpers.ensure_evaluated(context, name) return name if isinstance(name, Token) and name.type == TokenKind.RULE: return context.sheerka.resolve_rule(context, name) if isinstance(name, tuple): raise Exception() # try to resolve by name concept = context.sheerka.fast_resolve(name) if concept is None: return None if hasattr(concept, "__iter__"): raise NotImplementedError("Too many concepts") concept = core.builtin_helpers.ensure_evaluated(context, concept) return concept @staticmethod def expr_to_expression(expr): expr.lineno = 0 expr.col_offset = 0 result = ast.Expression(expr.value, lineno=0, col_offset=0) return result @staticmethod def exec_with_return(code_ast, my_globals, my_locals): init_ast = copy.deepcopy(code_ast) init_ast.body = code_ast.body[:-1] last_ast = copy.deepcopy(code_ast) last_ast.body = code_ast.body[-1:] exec(compile(init_ast, "", "exec"), my_globals, my_locals) if type(last_ast.body[0]) == ast.Expr: return eval(compile(PythonEvaluator.expr_to_expression(last_ast.body[0]), "", "eval"), my_globals, my_locals) else: exec(compile(last_ast, "", "exec"), my_globals, my_locals)