guarded_eval.py
898 lines
| 28.8 KiB
| text/x-python
|
PythonLexer
|
r28675 | from inspect import isclass, signature, Signature | ||
|
r27915 | from typing import ( | ||
|
r28680 | Annotated, | ||
AnyStr, | ||||
|
r27915 | Callable, | ||
|
r27920 | Dict, | ||
|
r28678 | Literal, | ||
NamedTuple, | ||||
NewType, | ||||
|
r28680 | Optional, | ||
Protocol, | ||||
|
r27915 | Set, | ||
|
r27926 | Sequence, | ||
|
r27915 | Tuple, | ||
Type, | ||||
|
r28680 | TypeGuard, | ||
|
r27915 | Union, | ||
|
r28678 | get_args, | ||
get_origin, | ||||
|
r28680 | is_typeddict, | ||
|
r28678 | ) | ||
|
r27920 | import ast | ||
|
r27915 | import builtins | ||
|
r27906 | import collections | ||
|
r27920 | import operator | ||
|
r27906 | import sys | ||
from functools import cached_property | ||||
from dataclasses import dataclass, field | ||||
|
r28272 | from types import MethodDescriptorType, ModuleType | ||
|
r27906 | |||
|
r28678 | from IPython.utils.decorators import undoc | ||
|
r27912 | |||
|
r27906 | |||
|
r28679 | if sys.version_info < (3, 11): | ||
|
r28680 | from typing_extensions import Self, LiteralString | ||
|
r28679 | else: | ||
|
r28680 | from typing import Self, LiteralString | ||
|
r28679 | |||
if sys.version_info < (3, 12): | ||||
from typing_extensions import TypeAliasType | ||||
else: | ||||
from typing import TypeAliasType | ||||
|
r27918 | @undoc | ||
|
r27906 | class HasGetItem(Protocol): | ||
|
r27913 | def __getitem__(self, key) -> None: | ||
... | ||||
|
r27906 | |||
|
r27918 | @undoc | ||
|
r27906 | class InstancesHaveGetItem(Protocol): | ||
|
r27915 | def __call__(self, *args, **kwargs) -> HasGetItem: | ||
|
r27913 | ... | ||
|
r27906 | |||
|
r27918 | @undoc | ||
|
r27906 | class HasGetAttr(Protocol): | ||
|
r27913 | def __getattr__(self, key) -> None: | ||
... | ||||
|
r27906 | |||
|
r27918 | @undoc | ||
|
r27906 | class DoesNotHaveGetAttr(Protocol): | ||
pass | ||||
|
r27913 | |||
|
r27906 | # By default `__getattr__` is not explicitly implemented on most objects | ||
MayHaveGetattr = Union[HasGetAttr, DoesNotHaveGetAttr] | ||||
|
r27918 | def _unbind_method(func: Callable) -> Union[Callable, None]: | ||
|
r27906 | """Get unbound method for given bound method. | ||
|
r27929 | Returns None if cannot get unbound method, or method is already unbound. | ||
""" | ||||
|
r27913 | owner = getattr(func, "__self__", None) | ||
|
r27906 | owner_class = type(owner) | ||
|
r27913 | name = getattr(func, "__name__", None) | ||
instance_dict_overrides = getattr(owner, "__dict__", None) | ||||
|
r27906 | if ( | ||
owner is not None | ||||
|
r27913 | and name | ||
and ( | ||||
|
r27906 | not instance_dict_overrides | ||
|
r27913 | or (instance_dict_overrides and name not in instance_dict_overrides) | ||
|
r27906 | ) | ||
): | ||||
return getattr(owner_class, name) | ||||
|
r27915 | return None | ||
|
r27906 | |||
|
r27918 | @undoc | ||
|
r27906 | @dataclass | ||
class EvaluationPolicy: | ||||
|
r27918 | """Definition of evaluation policy.""" | ||
|
r27906 | 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 | ||||
|
r27920 | allow_all_operations: bool = False | ||
|
r27906 | 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 | ||||
|
r27920 | def can_operate(self, dunders: Tuple[str, ...], a, b=None): | ||
if self.allow_all_operations: | ||||
return True | ||||
|
r27906 | def can_call(self, func): | ||
if self.allow_any_calls: | ||||
return True | ||||
if func in self.allowed_calls: | ||||
return True | ||||
|
r27918 | owner_method = _unbind_method(func) | ||
|
r27921 | |||
|
r27906 | if owner_method and owner_method in self.allowed_calls: | ||
return True | ||||
|
r27913 | |||
|
r27926 | 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 | ||||
|
r27918 | def _has_original_dunder_external( | ||
|
r27913 | value, | ||
|
r27926 | module_name: str, | ||
access_path: Sequence[str], | ||||
method_name: str, | ||||
|
r27913 | ): | ||
|
r27926 | if module_name not in sys.modules: | ||
# LBYLB as it is faster | ||||
return False | ||||
|
r27906 | try: | ||
|
r27926 | member_type = _get_external(module_name, access_path) | ||
|
r27906 | value_type = type(value) | ||
if type(value) == member_type: | ||||
return True | ||||
|
r27921 | 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 | ||||
|
r27906 | 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 | ||||
|
r27918 | def _has_original_dunder( | ||
|
r27913 | value, allowed_types, allowed_methods, allowed_external, method_name | ||
|
r27906 | ): | ||
# 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) | ||||
|
r27921 | if method is None: | ||
|
r27906 | return None | ||
if method in allowed_methods: | ||||
return True | ||||
for module_name, *access_path in allowed_external: | ||||
|
r27918 | if _has_original_dunder_external(value, module_name, access_path, method_name): | ||
|
r27906 | return True | ||
return False | ||||
|
r27918 | @undoc | ||
|
r27906 | @dataclass | ||
class SelectivePolicy(EvaluationPolicy): | ||||
|
r27915 | allowed_getitem: Set[InstancesHaveGetItem] = field(default_factory=set) | ||
|
r27906 | allowed_getitem_external: Set[Tuple[str, ...]] = field(default_factory=set) | ||
|
r27920 | |||
|
r27906 | allowed_getattr: Set[MayHaveGetattr] = field(default_factory=set) | ||
allowed_getattr_external: Set[Tuple[str, ...]] = field(default_factory=set) | ||||
|
r27920 | 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 | ||||
) | ||||
|
r27906 | def can_get_attr(self, value, attr): | ||
|
r27918 | has_original_attribute = _has_original_dunder( | ||
|
r27906 | value, | ||
allowed_types=self.allowed_getattr, | ||||
allowed_methods=self._getattribute_methods, | ||||
allowed_external=self.allowed_getattr_external, | ||||
|
r27913 | method_name="__getattribute__", | ||
|
r27906 | ) | ||
|
r27918 | has_original_attr = _has_original_dunder( | ||
|
r27906 | value, | ||
allowed_types=self.allowed_getattr, | ||||
allowed_methods=self._getattr_methods, | ||||
allowed_external=self.allowed_getattr_external, | ||||
|
r27913 | method_name="__getattr__", | ||
|
r27906 | ) | ||
|
r27921 | |||
|
r27926 | accept = False | ||
|
r27929 | # Many objects do not have `__getattr__`, this is fine. | ||
|
r27906 | if has_original_attr is None and has_original_attribute: | ||
|
r27926 | accept = True | ||
else: | ||||
# Accept objects without modifications to `__getattr__` and `__getattribute__` | ||||
accept = has_original_attr and has_original_attribute | ||||
|
r27906 | |||
|
r27926 | 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 | ||||
|
r27929 | # Properties in allowed types are ok (although we do not include any | ||
# properties in our default allow list currently). | ||||
|
r27926 | if type(value) in self.allowed_getattr: | ||
|
r27929 | return True # pragma: no cover | ||
|
r27926 | |||
# 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 | ||||
|
r27906 | |||
def can_get_item(self, value, item): | ||||
"""Allow accessing `__getiitem__` of allow-listed instances unless it was not modified.""" | ||||
|
r27918 | return _has_original_dunder( | ||
|
r27906 | value, | ||
allowed_types=self.allowed_getitem, | ||||
allowed_methods=self._getitem_methods, | ||||
allowed_external=self.allowed_getitem_external, | ||||
|
r27913 | method_name="__getitem__", | ||
|
r27906 | ) | ||
|
r27920 | def can_operate(self, dunders: Tuple[str, ...], a, b=None): | ||
|
r27921 | objects = [a] | ||
if b is not None: | ||||
objects.append(b) | ||||
|
r27920 | return all( | ||
[ | ||||
_has_original_dunder( | ||||
|
r27921 | obj, | ||
|
r27920 | allowed_types=self.allowed_operations, | ||
|
r27921 | allowed_methods=self._operator_dunder_methods(dunder), | ||
|
r27920 | allowed_external=self.allowed_operations_external, | ||
method_name=dunder, | ||||
) | ||||
for dunder in dunders | ||||
|
r27921 | for obj in objects | ||
|
r27920 | ] | ||
) | ||||
|
r27921 | def _operator_dunder_methods(self, dunder: str) -> Set[Callable]: | ||
|
r27920 | 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] | ||||
|
r27906 | @cached_property | ||
def _getitem_methods(self) -> Set[Callable]: | ||||
|
r27913 | return self._safe_get_methods(self.allowed_getitem, "__getitem__") | ||
|
r27906 | |||
@cached_property | ||||
def _getattr_methods(self) -> Set[Callable]: | ||||
|
r27913 | return self._safe_get_methods(self.allowed_getattr, "__getattr__") | ||
|
r27906 | |||
@cached_property | ||||
def _getattribute_methods(self) -> Set[Callable]: | ||||
|
r27913 | return self._safe_get_methods(self.allowed_getattr, "__getattribute__") | ||
|
r27906 | |||
def _safe_get_methods(self, classes, name) -> Set[Callable]: | ||||
return { | ||||
method | ||||
for class_ in classes | ||||
for method in [getattr(class_, name, None)] | ||||
if method | ||||
} | ||||
|
r27918 | class _DummyNamedTuple(NamedTuple): | ||
|
r27921 | """Used internally to retrieve methods of named tuple instance.""" | ||
|
r27906 | |||
class EvaluationContext(NamedTuple): | ||||
|
r27918 | #: Local namespace | ||
locals: dict | ||||
#: Global namespace | ||||
globals: dict | ||||
#: Evaluation policy identifier | ||||
|
r27913 | evaluation: Literal[ | ||
|
r27914 | "forbidden", "minimal", "limited", "unsafe", "dangerous" | ||
|
r27913 | ] = "forbidden" | ||
|
r27918 | #: Whether the evalution of code takes place inside of a subscript. | ||
#: Useful for evaluating ``:-1, 'col'`` in ``df[:-1, 'col']``. | ||||
|
r27906 | in_subscript: bool = False | ||
|
r27918 | class _IdentitySubscript: | ||
"""Returns the key itself when item is requested via subscript.""" | ||||
|
r27906 | def __getitem__(self, key): | ||
return key | ||||
|
r27913 | |||
|
r27918 | IDENTITY_SUBSCRIPT = _IdentitySubscript() | ||
|
r27913 | SUBSCRIPT_MARKER = "__SUBSCRIPT_SENTINEL__" | ||
|
r28432 | UNKNOWN_SIGNATURE = Signature() | ||
|
r28675 | NOT_EVALUATED = object() | ||
|
r27913 | |||
|
r27906 | |||
|
r27918 | class GuardRejection(Exception): | ||
"""Exception raised when guard rejects evaluation attempt.""" | ||||
|
r27906 | pass | ||
|
r27913 | def guarded_eval(code: str, context: EvaluationContext): | ||
|
r27918 | """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 | ||||
|
r27906 | |||
|
r27913 | if context.evaluation == "forbidden": | ||
raise GuardRejection("Forbidden mode") | ||||
|
r27906 | |||
# 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 | ||||
|
r27913 | code = SUBSCRIPT_MARKER + "[" + code + "]" | ||
|
r27918 | context = EvaluationContext(**{**context._asdict(), **{"locals": locals_}}) | ||
|
r27906 | |||
|
r27913 | if context.evaluation == "dangerous": | ||
|
r27918 | return eval(code, context.globals, context.locals) | ||
|
r27906 | |||
|
r27913 | expression = ast.parse(code, mode="eval") | ||
|
r27906 | |||
return eval_node(expression, context) | ||||
|
r27913 | |||
|
r27920 | 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__",), | ||||
} | ||||
|
r28680 | class ImpersonatingDuck: | ||
|
r28431 | """A dummy class used to create objects of other classes without calling their ``__init__``""" | ||
|
r28430 | |||
|
r28680 | # 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() | ||||
|
r28430 | |||
|
r27920 | 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 | ||||
|
r27906 | def eval_node(node: Union[ast.AST, None], context: EvaluationContext): | ||
|
r27918 | """Evaluate AST node in provided context. | ||
|
r27906 | |||
|
r27918 | Applies evaluation restrictions defined in the context. Currently does not support evaluation of functions with keyword arguments. | ||
|
r27906 | |||
|
r27918 | Does not evaluate actions that always have side effects: | ||
|
r27906 | |||
|
r27912 | - class definitions (``class sth: ...``) | ||
- function definitions (``def sth: ...``) | ||||
- variable assignments (``x = 1``) | ||||
|
r27914 | - augmented assignments (``x += 1``) | ||
|
r27912 | - deletions (``del x``) | ||
|
r27906 | |||
Does not evaluate operations which do not return values: | ||||
|
r27918 | |||
|
r27912 | - assertions (``assert x``) | ||
- pass (``pass``) | ||||
- imports (``import x``) | ||||
|
r27918 | - control flow: | ||
- conditionals (``if x:``) except for ternary IfExp (``a if x else b``) | ||||
|
r28390 | - loops (``for`` and ``while``) | ||
|
r27918 | - exception handling | ||
|
r27912 | |||
The purpose of this function is to guard against unwanted side-effects; | ||||
it does not give guarantees on protection from malicious code execution. | ||||
|
r27906 | """ | ||
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) | ||||
|
r27920 | 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: | ||||
|
r27921 | raise ValueError( | ||
f"Comparison `{dunder}` not supported" | ||||
) # pragma: no cover | ||||
|
r27920 | return all_true | ||
|
r27906 | if isinstance(node, ast.Constant): | ||
return node.value | ||||
if isinstance(node, ast.Tuple): | ||||
|
r27913 | return tuple(eval_node(e, context) for e in node.elts) | ||
|
r27906 | if isinstance(node, ast.List): | ||
|
r27913 | return [eval_node(e, context) for e in node.elts] | ||
|
r27906 | if isinstance(node, ast.Set): | ||
|
r27913 | return {eval_node(e, context) for e in node.elts} | ||
|
r27906 | if isinstance(node, ast.Dict): | ||
|
r27913 | return dict( | ||
zip( | ||||
[eval_node(k, context) for k in node.keys], | ||||
[eval_node(v, context) for v in node.values], | ||||
) | ||||
) | ||||
|
r27906 | if isinstance(node, ast.Slice): | ||
return slice( | ||||
eval_node(node.lower, context), | ||||
eval_node(node.upper, context), | ||||
|
r27913 | eval_node(node.step, context), | ||
|
r27906 | ) | ||
if isinstance(node, ast.UnaryOp): | ||||
value = eval_node(node.operand, context) | ||||
|
r27920 | 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", | ||||
) | ||||
|
r27906 | 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( | ||||
|
r27913 | "Subscript access (`__getitem__`) for", | ||
type(value), # not joined to avoid calling `repr` | ||||
f" not allowed in {context.evaluation} mode", | ||||
|
r27906 | ) | ||
if isinstance(node, ast.Name): | ||||
|
r28678 | return _eval_node_name(node.id, context) | ||
|
r27906 | 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( | ||||
|
r27913 | "Attribute access (`__getattr__`) for", | ||
type(value), # not joined to avoid calling `repr` | ||||
f"not allowed in {context.evaluation} mode", | ||||
|
r27906 | ) | ||
if isinstance(node, ast.IfExp): | ||||
test = eval_node(node.test, context) | ||||
if test: | ||||
|
r27913 | return eval_node(node.body, context) | ||
|
r27906 | 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: | ||||
|
r27913 | args = [eval_node(arg, context) for arg in node.args] | ||
|
r27906 | return func(*args) | ||
|
r28675 | 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. | ||||
|
r28678 | overridden_return_type = _eval_return_type(func.__new__, node, context) | ||
|
r28675 | if overridden_return_type is not NOT_EVALUATED: | ||
return overridden_return_type | ||||
|
r28678 | return _create_duck_for_heap_type(func) | ||
|
r28675 | else: | ||
|
r28678 | return_type = _eval_return_type(func, node, context) | ||
|
r28675 | if return_type is not NOT_EVALUATED: | ||
return return_type | ||||
|
r27906 | raise GuardRejection( | ||
|
r27913 | "Call for", | ||
func, # not joined to avoid calling `repr` | ||||
f"not allowed in {context.evaluation} mode", | ||||
|
r27906 | ) | ||
|
r27921 | raise ValueError("Unhandled node", ast.dump(node)) | ||
|
r27906 | |||
|
r28678 | def _eval_return_type(func: Callable, node: ast.Call, context: EvaluationContext): | ||
|
r28675 | """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 | ||||
|
r28676 | if not_empty: | ||
|
r28680 | return _resolve_annotation(sig.return_annotation, sig, func, node, context) | ||
|
r28675 | return NOT_EVALUATED | ||
|
r28680 | 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) | ||||
|
r28678 | def _eval_node_name(node_id: str, context: EvaluationContext): | ||
policy = EVALUATION_POLICIES[context.evaluation] | ||||
|
r28676 | 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") | ||||
|
r28678 | 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): | ||||
|
r28675 | """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. | ||||
""" | ||||
|
r28680 | duck = ImpersonatingDuck() | ||
|
r28675 | try: | ||
# this only works for heap types, not builtins | ||||
duck.__class__ = duck_type | ||||
return duck | ||||
except TypeError: | ||||
pass | ||||
return NOT_EVALUATED | ||||
|
r27906 | SUPPORTED_EXTERNAL_GETITEM = { | ||
|
r27913 | ("pandas", "core", "indexing", "_iLocIndexer"), | ||
("pandas", "core", "indexing", "_LocIndexer"), | ||||
("pandas", "DataFrame"), | ||||
("pandas", "Series"), | ||||
("numpy", "ndarray"), | ||||
("numpy", "void"), | ||||
|
r27906 | } | ||
|
r27921 | |||
|
r27915 | BUILTIN_GETITEM: Set[InstancesHaveGetItem] = { | ||
|
r27906 | dict, | ||
|
r28091 | str, # type: ignore[arg-type] | ||
bytes, # type: ignore[arg-type] | ||||
|
r27906 | list, | ||
tuple, | ||||
collections.defaultdict, | ||||
collections.deque, | ||||
collections.OrderedDict, | ||||
collections.ChainMap, | ||||
collections.UserDict, | ||||
collections.UserList, | ||||
|
r28091 | collections.UserString, # type: ignore[arg-type] | ||
|
r27918 | _DummyNamedTuple, | ||
_IdentitySubscript, | ||||
|
r27906 | } | ||
def _list_methods(cls, source=None): | ||||
"""For use on immutable objects or with methods returning a copy""" | ||||
|
r27913 | return [getattr(cls, k) for k in (source if source else dir(cls))] | ||
|
r27906 | |||
|
r27913 | dict_non_mutating_methods = ("copy", "keys", "values", "items") | ||
list_non_mutating_methods = ("copy", "index", "count") | ||||
|
r27906 | set_non_mutating_methods = set(dir(set)) & set(dir(frozenset)) | ||
|
r27915 | dict_keys: Type[collections.abc.KeysView] = type({}.keys()) | ||
|
r27906 | |||
|
r27921 | NUMERICS = {int, float, complex} | ||
|
r27906 | 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), | ||||
|
r27921 | *NUMERICS, | ||
*[method for numeric_cls in NUMERICS for method in _list_methods(numeric_cls)], | ||||
|
r27906 | 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, | ||||
|
r27913 | collections.Counter.most_common, | ||
|
r27906 | } | ||
|
r27915 | BUILTIN_GETATTR: Set[MayHaveGetattr] = { | ||
*BUILTIN_GETITEM, | ||||
set, | ||||
frozenset, | ||||
object, | ||||
type, # `type` handles a lot of generic cases, e.g. numbers as in `int.real`. | ||||
|
r27921 | *NUMERICS, | ||
|
r27915 | dict_keys, | ||
|
r28272 | MethodDescriptorType, | ||
ModuleType, | ||||
|
r27915 | } | ||
|
r27920 | |||
|
r27921 | BUILTIN_OPERATIONS = {*BUILTIN_GETATTR} | ||
|
r27920 | |||
|
r27906 | EVALUATION_POLICIES = { | ||
|
r27913 | "minimal": EvaluationPolicy( | ||
|
r27906 | allow_builtins_access=True, | ||
allow_locals_access=False, | ||||
allow_globals_access=False, | ||||
allow_item_access=False, | ||||
allow_attr_access=False, | ||||
allowed_calls=set(), | ||||
|
r27913 | allow_any_calls=False, | ||
|
r27920 | allow_all_operations=False, | ||
|
r27906 | ), | ||
|
r27914 | "limited": SelectivePolicy( | ||
|
r27906 | allowed_getitem=BUILTIN_GETITEM, | ||
allowed_getitem_external=SUPPORTED_EXTERNAL_GETITEM, | ||||
|
r27915 | allowed_getattr=BUILTIN_GETATTR, | ||
|
r27906 | allowed_getattr_external={ | ||
# pandas Series/Frame implements custom `__getattr__` | ||||
|
r27913 | ("pandas", "DataFrame"), | ||
("pandas", "Series"), | ||||
|
r27906 | }, | ||
|
r27920 | allowed_operations=BUILTIN_OPERATIONS, | ||
|
r27906 | allow_builtins_access=True, | ||
allow_locals_access=True, | ||||
allow_globals_access=True, | ||||
|
r27913 | allowed_calls=ALLOWED_CALLS, | ||
|
r27906 | ), | ||
|
r27913 | "unsafe": EvaluationPolicy( | ||
|
r27906 | allow_builtins_access=True, | ||
allow_locals_access=True, | ||||
allow_globals_access=True, | ||||
allow_attr_access=True, | ||||
allow_item_access=True, | ||||
|
r27913 | allow_any_calls=True, | ||
|
r27920 | allow_all_operations=True, | ||
|
r27913 | ), | ||
} | ||||
|
r27918 | |||
__all__ = [ | ||||
"guarded_eval", | ||||
"eval_node", | ||||
"GuardRejection", | ||||
"EvaluationContext", | ||||
"_unbind_method", | ||||
] | ||||