from inspect import isclass, signature, Signature from typing import ( Annotated, AnyStr, Callable, Dict, Literal, NamedTuple, NewType, Optional, Protocol, Set, Sequence, Tuple, Type, TypeGuard, Union, get_args, get_origin, is_typeddict, ) import ast import builtins import collections import operator import sys from functools import cached_property from dataclasses import dataclass, field from types import MethodDescriptorType, ModuleType from IPython.utils.decorators import undoc if sys.version_info < (3, 11): from typing_extensions import Self, LiteralString else: from typing import Self, LiteralString if sys.version_info < (3, 12): from typing_extensions import TypeAliasType else: from typing import TypeAliasType @undoc class HasGetItem(Protocol): def __getitem__(self, key) -> None: ... @undoc class InstancesHaveGetItem(Protocol): def __call__(self, *args, **kwargs) -> HasGetItem: ... @undoc class HasGetAttr(Protocol): def __getattr__(self, key) -> None: ... @undoc class DoesNotHaveGetAttr(Protocol): pass # By default `__getattr__` is not explicitly implemented on most objects MayHaveGetattr = Union[HasGetAttr, DoesNotHaveGetAttr] def _unbind_method(func: Callable) -> Union[Callable, None]: """Get unbound method for given bound method. Returns None if cannot get unbound method, or method is already unbound. """ owner = getattr(func, "__self__", None) owner_class = type(owner) name = getattr(func, "__name__", None) instance_dict_overrides = getattr(owner, "__dict__", None) if ( owner is not None and name and ( not instance_dict_overrides or (instance_dict_overrides and name not in instance_dict_overrides) ) ): return getattr(owner_class, name) return None @undoc @dataclass class EvaluationPolicy: """Definition of evaluation policy.""" allow_locals_access: bool = False allow_globals_access: bool = False allow_item_access: bool = False allow_attr_access: bool = False allow_builtins_access: bool = False allow_all_operations: bool = False allow_any_calls: bool = False allowed_calls: Set[Callable] = field(default_factory=set) def can_get_item(self, value, item): return self.allow_item_access def can_get_attr(self, value, attr): return self.allow_attr_access def can_operate(self, dunders: Tuple[str, ...], a, b=None): if self.allow_all_operations: return True def can_call(self, func): if self.allow_any_calls: return True if func in self.allowed_calls: return True owner_method = _unbind_method(func) if owner_method and owner_method in self.allowed_calls: return True def _get_external(module_name: str, access_path: Sequence[str]): """Get value from external module given a dotted access path. Raises: * `KeyError` if module is removed not found, and * `AttributeError` if acess path does not match an exported object """ member_type = sys.modules[module_name] for attr in access_path: member_type = getattr(member_type, attr) return member_type def _has_original_dunder_external( value, module_name: str, access_path: Sequence[str], method_name: str, ): if module_name not in sys.modules: # LBYLB as it is faster return False try: member_type = _get_external(module_name, access_path) value_type = type(value) if type(value) == member_type: return True if method_name == "__getattribute__": # we have to short-circuit here due to an unresolved issue in # `isinstance` implementation: https://bugs.python.org/issue32683 return False if isinstance(value, member_type): method = getattr(value_type, method_name, None) member_method = getattr(member_type, method_name, None) if member_method == method: return True except (AttributeError, KeyError): return False def _has_original_dunder( value, allowed_types, allowed_methods, allowed_external, method_name ): # note: Python ignores `__getattr__`/`__getitem__` on instances, # we only need to check at class level value_type = type(value) # strict type check passes → no need to check method if value_type in allowed_types: return True method = getattr(value_type, method_name, None) if method is None: return None if method in allowed_methods: return True for module_name, *access_path in allowed_external: if _has_original_dunder_external(value, module_name, access_path, method_name): return True return False @undoc @dataclass class SelectivePolicy(EvaluationPolicy): allowed_getitem: Set[InstancesHaveGetItem] = field(default_factory=set) allowed_getitem_external: Set[Tuple[str, ...]] = field(default_factory=set) allowed_getattr: Set[MayHaveGetattr] = field(default_factory=set) allowed_getattr_external: Set[Tuple[str, ...]] = field(default_factory=set) allowed_operations: Set = field(default_factory=set) allowed_operations_external: Set[Tuple[str, ...]] = field(default_factory=set) _operation_methods_cache: Dict[str, Set[Callable]] = field( default_factory=dict, init=False ) def can_get_attr(self, value, attr): has_original_attribute = _has_original_dunder( value, allowed_types=self.allowed_getattr, allowed_methods=self._getattribute_methods, allowed_external=self.allowed_getattr_external, method_name="__getattribute__", ) has_original_attr = _has_original_dunder( value, allowed_types=self.allowed_getattr, allowed_methods=self._getattr_methods, allowed_external=self.allowed_getattr_external, method_name="__getattr__", ) accept = False # Many objects do not have `__getattr__`, this is fine. if has_original_attr is None and has_original_attribute: accept = True else: # Accept objects without modifications to `__getattr__` and `__getattribute__` accept = has_original_attr and has_original_attribute if accept: # We still need to check for overriden properties. value_class = type(value) if not hasattr(value_class, attr): return True class_attr_val = getattr(value_class, attr) is_property = isinstance(class_attr_val, property) if not is_property: return True # Properties in allowed types are ok (although we do not include any # properties in our default allow list currently). if type(value) in self.allowed_getattr: return True # pragma: no cover # Properties in subclasses of allowed types may be ok if not changed for module_name, *access_path in self.allowed_getattr_external: try: external_class = _get_external(module_name, access_path) external_class_attr_val = getattr(external_class, attr) except (KeyError, AttributeError): return False # pragma: no cover return class_attr_val == external_class_attr_val return False def can_get_item(self, value, item): """Allow accessing `__getiitem__` of allow-listed instances unless it was not modified.""" return _has_original_dunder( value, allowed_types=self.allowed_getitem, allowed_methods=self._getitem_methods, allowed_external=self.allowed_getitem_external, method_name="__getitem__", ) def can_operate(self, dunders: Tuple[str, ...], a, b=None): objects = [a] if b is not None: objects.append(b) return all( [ _has_original_dunder( obj, allowed_types=self.allowed_operations, allowed_methods=self._operator_dunder_methods(dunder), allowed_external=self.allowed_operations_external, method_name=dunder, ) for dunder in dunders for obj in objects ] ) def _operator_dunder_methods(self, dunder: str) -> Set[Callable]: if dunder not in self._operation_methods_cache: self._operation_methods_cache[dunder] = self._safe_get_methods( self.allowed_operations, dunder ) return self._operation_methods_cache[dunder] @cached_property def _getitem_methods(self) -> Set[Callable]: return self._safe_get_methods(self.allowed_getitem, "__getitem__") @cached_property def _getattr_methods(self) -> Set[Callable]: return self._safe_get_methods(self.allowed_getattr, "__getattr__") @cached_property def _getattribute_methods(self) -> Set[Callable]: return self._safe_get_methods(self.allowed_getattr, "__getattribute__") def _safe_get_methods(self, classes, name) -> Set[Callable]: return { method for class_ in classes for method in [getattr(class_, name, None)] if method } class _DummyNamedTuple(NamedTuple): """Used internally to retrieve methods of named tuple instance.""" class EvaluationContext(NamedTuple): #: Local namespace locals: dict #: Global namespace globals: dict #: Evaluation policy identifier evaluation: Literal[ "forbidden", "minimal", "limited", "unsafe", "dangerous" ] = "forbidden" #: Whether the evalution of code takes place inside of a subscript. #: Useful for evaluating ``:-1, 'col'`` in ``df[:-1, 'col']``. in_subscript: bool = False class _IdentitySubscript: """Returns the key itself when item is requested via subscript.""" def __getitem__(self, key): return key IDENTITY_SUBSCRIPT = _IdentitySubscript() SUBSCRIPT_MARKER = "__SUBSCRIPT_SENTINEL__" UNKNOWN_SIGNATURE = Signature() NOT_EVALUATED = object() class GuardRejection(Exception): """Exception raised when guard rejects evaluation attempt.""" pass def guarded_eval(code: str, context: EvaluationContext): """Evaluate provided code in the evaluation context. If evaluation policy given by context is set to ``forbidden`` no evaluation will be performed; if it is set to ``dangerous`` standard :func:`eval` will be used; finally, for any other, policy :func:`eval_node` will be called on parsed AST. """ locals_ = context.locals if context.evaluation == "forbidden": raise GuardRejection("Forbidden mode") # note: not using `ast.literal_eval` as it does not implement # getitem at all, for example it fails on simple `[0][1]` if context.in_subscript: # syntatic sugar for ellipsis (:) is only available in susbcripts # so we need to trick the ast parser into thinking that we have # a subscript, but we need to be able to later recognise that we did # it so we can ignore the actual __getitem__ operation if not code: return tuple() locals_ = locals_.copy() locals_[SUBSCRIPT_MARKER] = IDENTITY_SUBSCRIPT code = SUBSCRIPT_MARKER + "[" + code + "]" context = EvaluationContext(**{**context._asdict(), **{"locals": locals_}}) if context.evaluation == "dangerous": return eval(code, context.globals, context.locals) expression = ast.parse(code, mode="eval") return eval_node(expression, context) BINARY_OP_DUNDERS: Dict[Type[ast.operator], Tuple[str]] = { ast.Add: ("__add__",), ast.Sub: ("__sub__",), ast.Mult: ("__mul__",), ast.Div: ("__truediv__",), ast.FloorDiv: ("__floordiv__",), ast.Mod: ("__mod__",), ast.Pow: ("__pow__",), ast.LShift: ("__lshift__",), ast.RShift: ("__rshift__",), ast.BitOr: ("__or__",), ast.BitXor: ("__xor__",), ast.BitAnd: ("__and__",), ast.MatMult: ("__matmul__",), } COMP_OP_DUNDERS: Dict[Type[ast.cmpop], Tuple[str, ...]] = { ast.Eq: ("__eq__",), ast.NotEq: ("__ne__", "__eq__"), ast.Lt: ("__lt__", "__gt__"), ast.LtE: ("__le__", "__ge__"), ast.Gt: ("__gt__", "__lt__"), ast.GtE: ("__ge__", "__le__"), ast.In: ("__contains__",), # Note: ast.Is, ast.IsNot, ast.NotIn are handled specially } UNARY_OP_DUNDERS: Dict[Type[ast.unaryop], Tuple[str, ...]] = { ast.USub: ("__neg__",), ast.UAdd: ("__pos__",), # we have to check both __inv__ and __invert__! ast.Invert: ("__invert__", "__inv__"), ast.Not: ("__not__",), } class ImpersonatingDuck: """A dummy class used to create objects of other classes without calling their ``__init__``""" # no-op: override __class__ to impersonate class _Duck: """A dummy class used to create objects pretending to have given attributes""" def __init__(self, attributes: Optional[dict] = None, items: Optional[dict] = None): self.attributes = attributes or {} self.items = items or {} def __getattr__(self, attr: str): return self.attributes[attr] def __hasattr__(self, attr: str): return attr in self.attributes def __dir__(self): return [*dir(super), *self.attributes] def __getitem__(self, key: str): return self.items[key] def __hasitem__(self, key: str): return self.items[key] def _ipython_key_completions_(self): return self.items.keys() def _find_dunder(node_op, dunders) -> Union[Tuple[str, ...], None]: dunder = None for op, candidate_dunder in dunders.items(): if isinstance(node_op, op): dunder = candidate_dunder return dunder def eval_node(node: Union[ast.AST, None], context: EvaluationContext): """Evaluate AST node in provided context. Applies evaluation restrictions defined in the context. Currently does not support evaluation of functions with keyword arguments. Does not evaluate actions that always have side effects: - class definitions (``class sth: ...``) - function definitions (``def sth: ...``) - variable assignments (``x = 1``) - augmented assignments (``x += 1``) - deletions (``del x``) Does not evaluate operations which do not return values: - assertions (``assert x``) - pass (``pass``) - imports (``import x``) - control flow: - conditionals (``if x:``) except for ternary IfExp (``a if x else b``) - loops (``for`` and ``while``) - exception handling The purpose of this function is to guard against unwanted side-effects; it does not give guarantees on protection from malicious code execution. """ policy = EVALUATION_POLICIES[context.evaluation] if node is None: return None if isinstance(node, ast.Expression): return eval_node(node.body, context) if isinstance(node, ast.BinOp): left = eval_node(node.left, context) right = eval_node(node.right, context) dunders = _find_dunder(node.op, BINARY_OP_DUNDERS) if dunders: if policy.can_operate(dunders, left, right): return getattr(left, dunders[0])(right) else: raise GuardRejection( f"Operation (`{dunders}`) for", type(left), f"not allowed in {context.evaluation} mode", ) if isinstance(node, ast.Compare): left = eval_node(node.left, context) all_true = True negate = False for op, right in zip(node.ops, node.comparators): right = eval_node(right, context) dunder = None dunders = _find_dunder(op, COMP_OP_DUNDERS) if not dunders: if isinstance(op, ast.NotIn): dunders = COMP_OP_DUNDERS[ast.In] negate = True if isinstance(op, ast.Is): dunder = "is_" if isinstance(op, ast.IsNot): dunder = "is_" negate = True if not dunder and dunders: dunder = dunders[0] if dunder: a, b = (right, left) if dunder == "__contains__" else (left, right) if dunder == "is_" or dunders and policy.can_operate(dunders, a, b): result = getattr(operator, dunder)(a, b) if negate: result = not result if not result: all_true = False left = right else: raise GuardRejection( f"Comparison (`{dunder}`) for", type(left), f"not allowed in {context.evaluation} mode", ) else: raise ValueError( f"Comparison `{dunder}` not supported" ) # pragma: no cover return all_true if isinstance(node, ast.Constant): return node.value if isinstance(node, ast.Tuple): return tuple(eval_node(e, context) for e in node.elts) if isinstance(node, ast.List): return [eval_node(e, context) for e in node.elts] if isinstance(node, ast.Set): return {eval_node(e, context) for e in node.elts} if isinstance(node, ast.Dict): return dict( zip( [eval_node(k, context) for k in node.keys], [eval_node(v, context) for v in node.values], ) ) if isinstance(node, ast.Slice): return slice( eval_node(node.lower, context), eval_node(node.upper, context), eval_node(node.step, context), ) if isinstance(node, ast.UnaryOp): value = eval_node(node.operand, context) dunders = _find_dunder(node.op, UNARY_OP_DUNDERS) if dunders: if policy.can_operate(dunders, value): return getattr(value, dunders[0])() else: raise GuardRejection( f"Operation (`{dunders}`) for", type(value), f"not allowed in {context.evaluation} mode", ) if isinstance(node, ast.Subscript): value = eval_node(node.value, context) slice_ = eval_node(node.slice, context) if policy.can_get_item(value, slice_): return value[slice_] raise GuardRejection( "Subscript access (`__getitem__`) for", type(value), # not joined to avoid calling `repr` f" not allowed in {context.evaluation} mode", ) if isinstance(node, ast.Name): return _eval_node_name(node.id, context) if isinstance(node, ast.Attribute): value = eval_node(node.value, context) if policy.can_get_attr(value, node.attr): return getattr(value, node.attr) raise GuardRejection( "Attribute access (`__getattr__`) for", type(value), # not joined to avoid calling `repr` f"not allowed in {context.evaluation} mode", ) if isinstance(node, ast.IfExp): test = eval_node(node.test, context) if test: return eval_node(node.body, context) else: return eval_node(node.orelse, context) if isinstance(node, ast.Call): func = eval_node(node.func, context) if policy.can_call(func) and not node.keywords: args = [eval_node(arg, context) for arg in node.args] return func(*args) if isclass(func): # this code path gets entered when calling class e.g. `MyClass()` # or `my_instance.__class__()` - in both cases `func` is `MyClass`. # Should return `MyClass` if `__new__` is not overridden, # otherwise whatever `__new__` return type is. overridden_return_type = _eval_return_type(func.__new__, node, context) if overridden_return_type is not NOT_EVALUATED: return overridden_return_type return _create_duck_for_heap_type(func) else: return_type = _eval_return_type(func, node, context) if return_type is not NOT_EVALUATED: return return_type raise GuardRejection( "Call for", func, # not joined to avoid calling `repr` f"not allowed in {context.evaluation} mode", ) raise ValueError("Unhandled node", ast.dump(node)) def _eval_return_type(func: Callable, node: ast.Call, context: EvaluationContext): """Evaluate return type of a given callable function. Returns the built-in type, a duck or NOT_EVALUATED sentinel. """ try: sig = signature(func) except ValueError: sig = UNKNOWN_SIGNATURE # if annotation was not stringized, or it was stringized # but resolved by signature call we know the return type not_empty = sig.return_annotation is not Signature.empty if not_empty: return _resolve_annotation(sig.return_annotation, sig, func, node, context) return NOT_EVALUATED def _resolve_annotation( annotation, sig: Signature, func: Callable, node: ast.Call, context: EvaluationContext, ): """Resolve annotation created by user with `typing` module and custom objects.""" annotation = ( _eval_node_name(annotation, context) if isinstance(annotation, str) else annotation ) origin = get_origin(annotation) if annotation is Self and hasattr(func, "__self__"): return func.__self__ elif origin is Literal: type_args = get_args(annotation) if len(type_args) == 1: return type_args[0] elif annotation is LiteralString: return "" elif annotation is AnyStr: index = None for i, (key, value) in enumerate(sig.parameters.items()): if value.annotation is AnyStr: index = i break if index is not None and index < len(node.args): return eval_node(node.args[index], context) elif origin is TypeGuard: return bool() elif origin is Union: attributes = [ attr for type_arg in get_args(annotation) for attr in dir(_resolve_annotation(type_arg, sig, func, node, context)) ] return _Duck(attributes=dict.fromkeys(attributes)) elif is_typeddict(annotation): return _Duck( attributes=dict.fromkeys(dir(dict())), items={ k: _resolve_annotation(v, sig, func, node, context) for k, v in annotation.__annotations__.items() }, ) elif hasattr(annotation, "_is_protocol"): return _Duck(attributes=dict.fromkeys(dir(annotation))) elif origin is Annotated: type_arg = get_args(annotation)[0] return _resolve_annotation(type_arg, sig, func, node, context) elif isinstance(annotation, NewType): return _eval_or_create_duck(annotation.__supertype__, node, context) elif isinstance(annotation, TypeAliasType): return _eval_or_create_duck(annotation.__value__, node, context) else: return _eval_or_create_duck(annotation, node, context) def _eval_node_name(node_id: str, context: EvaluationContext): policy = EVALUATION_POLICIES[context.evaluation] if policy.allow_locals_access and node_id in context.locals: return context.locals[node_id] if policy.allow_globals_access and node_id in context.globals: return context.globals[node_id] if policy.allow_builtins_access and hasattr(builtins, node_id): # note: do not use __builtins__, it is implementation detail of cPython return getattr(builtins, node_id) if not policy.allow_globals_access and not policy.allow_locals_access: raise GuardRejection( f"Namespace access not allowed in {context.evaluation} mode" ) else: raise NameError(f"{node_id} not found in locals, globals, nor builtins") def _eval_or_create_duck(duck_type, node: ast.Call, context: EvaluationContext): policy = EVALUATION_POLICIES[context.evaluation] # if allow-listed builtin is on type annotation, instantiate it if policy.can_call(duck_type) and not node.keywords: args = [eval_node(arg, context) for arg in node.args] return duck_type(*args) # if custom class is in type annotation, mock it return _create_duck_for_heap_type(duck_type) def _create_duck_for_heap_type(duck_type): """Create an imitation of an object of a given type (a duck). Returns the duck or NOT_EVALUATED sentinel if duck could not be created. """ duck = ImpersonatingDuck() try: # this only works for heap types, not builtins duck.__class__ = duck_type return duck except TypeError: pass return NOT_EVALUATED SUPPORTED_EXTERNAL_GETITEM = { ("pandas", "core", "indexing", "_iLocIndexer"), ("pandas", "core", "indexing", "_LocIndexer"), ("pandas", "DataFrame"), ("pandas", "Series"), ("numpy", "ndarray"), ("numpy", "void"), } BUILTIN_GETITEM: Set[InstancesHaveGetItem] = { dict, str, # type: ignore[arg-type] bytes, # type: ignore[arg-type] list, tuple, collections.defaultdict, collections.deque, collections.OrderedDict, collections.ChainMap, collections.UserDict, collections.UserList, collections.UserString, # type: ignore[arg-type] _DummyNamedTuple, _IdentitySubscript, } def _list_methods(cls, source=None): """For use on immutable objects or with methods returning a copy""" return [getattr(cls, k) for k in (source if source else dir(cls))] dict_non_mutating_methods = ("copy", "keys", "values", "items") list_non_mutating_methods = ("copy", "index", "count") set_non_mutating_methods = set(dir(set)) & set(dir(frozenset)) dict_keys: Type[collections.abc.KeysView] = type({}.keys()) NUMERICS = {int, float, complex} ALLOWED_CALLS = { bytes, *_list_methods(bytes), dict, *_list_methods(dict, dict_non_mutating_methods), dict_keys.isdisjoint, list, *_list_methods(list, list_non_mutating_methods), set, *_list_methods(set, set_non_mutating_methods), frozenset, *_list_methods(frozenset), range, str, *_list_methods(str), tuple, *_list_methods(tuple), *NUMERICS, *[method for numeric_cls in NUMERICS for method in _list_methods(numeric_cls)], collections.deque, *_list_methods(collections.deque, list_non_mutating_methods), collections.defaultdict, *_list_methods(collections.defaultdict, dict_non_mutating_methods), collections.OrderedDict, *_list_methods(collections.OrderedDict, dict_non_mutating_methods), collections.UserDict, *_list_methods(collections.UserDict, dict_non_mutating_methods), collections.UserList, *_list_methods(collections.UserList, list_non_mutating_methods), collections.UserString, *_list_methods(collections.UserString, dir(str)), collections.Counter, *_list_methods(collections.Counter, dict_non_mutating_methods), collections.Counter.elements, collections.Counter.most_common, } BUILTIN_GETATTR: Set[MayHaveGetattr] = { *BUILTIN_GETITEM, set, frozenset, object, type, # `type` handles a lot of generic cases, e.g. numbers as in `int.real`. *NUMERICS, dict_keys, MethodDescriptorType, ModuleType, } BUILTIN_OPERATIONS = {*BUILTIN_GETATTR} EVALUATION_POLICIES = { "minimal": EvaluationPolicy( allow_builtins_access=True, allow_locals_access=False, allow_globals_access=False, allow_item_access=False, allow_attr_access=False, allowed_calls=set(), allow_any_calls=False, allow_all_operations=False, ), "limited": SelectivePolicy( allowed_getitem=BUILTIN_GETITEM, allowed_getitem_external=SUPPORTED_EXTERNAL_GETITEM, allowed_getattr=BUILTIN_GETATTR, allowed_getattr_external={ # pandas Series/Frame implements custom `__getattr__` ("pandas", "DataFrame"), ("pandas", "Series"), }, allowed_operations=BUILTIN_OPERATIONS, allow_builtins_access=True, allow_locals_access=True, allow_globals_access=True, allowed_calls=ALLOWED_CALLS, ), "unsafe": EvaluationPolicy( allow_builtins_access=True, allow_locals_access=True, allow_globals_access=True, allow_attr_access=True, allow_item_access=True, allow_any_calls=True, allow_all_operations=True, ), } __all__ = [ "guarded_eval", "eval_node", "GuardRejection", "EvaluationContext", "_unbind_method", ]