Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from copy import copy
2from inspect import get_annotations, isclass, signature, Signature, getmodule
3from typing import (
4 Annotated,
5 AnyStr,
6 Literal,
7 NamedTuple,
8 NewType,
9 Protocol,
10 TypeGuard,
11 Union,
12 get_args,
13 get_origin,
14 is_typeddict,
15)
16from collections.abc import Callable, Sequence
17import ast
18import builtins
19import collections
20import dataclasses
21import operator
22import sys
23import typing
24import warnings
25from functools import cached_property
26from dataclasses import dataclass, field
27from types import MethodDescriptorType, ModuleType, MethodType
29from IPython.utils.decorators import undoc
31import types
32from typing import Self, LiteralString
34if sys.version_info < (3, 12):
35 from typing_extensions import TypeAliasType
36else:
37 from typing import TypeAliasType
40@undoc
41class HasGetItem(Protocol):
42 def __getitem__(self, key) -> None:
43 ...
46@undoc
47class InstancesHaveGetItem(Protocol):
48 def __call__(self, *args, **kwargs) -> HasGetItem:
49 ...
52@undoc
53class HasGetAttr(Protocol):
54 def __getattr__(self, key) -> None:
55 ...
58@undoc
59class DoesNotHaveGetAttr(Protocol):
60 pass
63# By default `__getattr__` is not explicitly implemented on most objects
64MayHaveGetattr = HasGetAttr | DoesNotHaveGetAttr
67def _unbind_method(func: Callable) -> Callable | None:
68 """Get unbound method for given bound method.
70 Returns None if cannot get unbound method, or method is already unbound.
71 """
72 owner = getattr(func, "__self__", None)
73 owner_class = type(owner)
74 name = getattr(func, "__name__", None)
75 instance_dict_overrides = getattr(owner, "__dict__", None)
76 if (
77 owner is not None
78 and name
79 and (
80 not instance_dict_overrides
81 or (instance_dict_overrides and name not in instance_dict_overrides)
82 )
83 ):
84 return getattr(owner_class, name)
85 return None
88@undoc
89@dataclass
90class EvaluationPolicy:
91 """Definition of evaluation policy."""
93 allow_locals_access: bool = False
94 allow_globals_access: bool = False
95 allow_item_access: bool = False
96 allow_attr_access: bool = False
97 allow_builtins_access: bool = False
98 allow_all_operations: bool = False
99 allow_any_calls: bool = False
100 allow_auto_import: bool = False
101 allowed_calls: set[Callable] = field(default_factory=set)
103 def can_get_item(self, value, item):
104 return self.allow_item_access
106 def can_get_attr(self, value, attr):
107 return self.allow_attr_access
109 def can_operate(self, dunders: tuple[str, ...], a, b=None):
110 if self.allow_all_operations:
111 return True
113 def can_call(self, func):
114 if self.allow_any_calls:
115 return True
117 if func in self.allowed_calls:
118 return True
120 owner_method = _unbind_method(func)
122 if owner_method and owner_method in self.allowed_calls:
123 return True
126def _get_external(module_name: str, access_path: Sequence[str]):
127 """Get value from external module given a dotted access path.
129 Only gets value if the module is already imported.
131 Raises:
132 * `KeyError` if module is removed not found, and
133 * `AttributeError` if access path does not match an exported object
134 """
135 try:
136 member_type = sys.modules[module_name]
137 # standard module
138 for attr in access_path:
139 member_type = getattr(member_type, attr)
140 return member_type
141 except (KeyError, AttributeError):
142 # handle modules in namespace packages
143 module_path = ".".join([module_name, *access_path])
144 if module_path in sys.modules:
145 return sys.modules[module_path]
146 raise
149def _has_original_dunder_external(
150 value,
151 module_name: str,
152 access_path: Sequence[str],
153 method_name: str,
154):
155 if module_name not in sys.modules:
156 full_module_path = ".".join([module_name, *access_path])
157 if full_module_path not in sys.modules:
158 # LBYLB as it is faster
159 return False
160 try:
161 member_type = _get_external(module_name, access_path)
162 value_type = type(value)
163 if type(value) == member_type:
164 return True
165 if isinstance(member_type, ModuleType):
166 value_module = getmodule(value_type)
167 if not value_module or not value_module.__name__:
168 return False
169 if (
170 value_module.__name__ == member_type.__name__
171 or value_module.__name__.startswith(member_type.__name__ + ".")
172 ):
173 return True
174 if method_name == "__getattribute__":
175 # we have to short-circuit here due to an unresolved issue in
176 # `isinstance` implementation: https://bugs.python.org/issue32683
177 return False
178 if not isinstance(member_type, ModuleType) and isinstance(value, member_type):
179 method = getattr(value_type, method_name, None)
180 member_method = getattr(member_type, method_name, None)
181 if member_method == method:
182 return True
183 if isinstance(member_type, ModuleType):
184 method = getattr(value_type, method_name, None)
185 for base_class in value_type.__mro__[1:]:
186 base_module = getmodule(base_class)
187 if base_module and (
188 base_module.__name__ == member_type.__name__
189 or base_module.__name__.startswith(member_type.__name__ + ".")
190 ):
191 # Check if the method comes from this trusted base class
192 base_method = getattr(base_class, method_name, None)
193 if base_method is not None and base_method == method:
194 return True
195 except (AttributeError, KeyError):
196 return False
199def _has_original_dunder(
200 value, allowed_types, allowed_methods, allowed_external, method_name
201):
202 # note: Python ignores `__getattr__`/`__getitem__` on instances,
203 # we only need to check at class level
204 value_type = type(value)
206 # strict type check passes → no need to check method
207 if value_type in allowed_types:
208 return True
210 method = getattr(value_type, method_name, None)
212 if method is None:
213 return None
215 if method in allowed_methods:
216 return True
218 for module_name, *access_path in allowed_external:
219 if _has_original_dunder_external(value, module_name, access_path, method_name):
220 return True
222 return False
225def _coerce_path_to_tuples(
226 allow_list: set[tuple[str, ...] | str],
227) -> set[tuple[str, ...]]:
228 """Replace dotted paths on the provided allow-list with tuples."""
229 return {
230 path if isinstance(path, tuple) else tuple(path.split("."))
231 for path in allow_list
232 }
235@undoc
236@dataclass
237class SelectivePolicy(EvaluationPolicy):
238 allowed_getitem: set[InstancesHaveGetItem] = field(default_factory=set)
239 allowed_getitem_external: set[tuple[str, ...] | str] = field(default_factory=set)
241 allowed_getattr: set[MayHaveGetattr] = field(default_factory=set)
242 allowed_getattr_external: set[tuple[str, ...] | str] = field(default_factory=set)
244 allowed_operations: set = field(default_factory=set)
245 allowed_operations_external: set[tuple[str, ...] | str] = field(default_factory=set)
247 allow_getitem_on_types: bool = field(default_factory=bool)
249 _operation_methods_cache: dict[str, set[Callable]] = field(
250 default_factory=dict, init=False
251 )
253 def can_get_attr(self, value, attr):
254 allowed_getattr_external = _coerce_path_to_tuples(self.allowed_getattr_external)
256 has_original_attribute = _has_original_dunder(
257 value,
258 allowed_types=self.allowed_getattr,
259 allowed_methods=self._getattribute_methods,
260 allowed_external=allowed_getattr_external,
261 method_name="__getattribute__",
262 )
263 has_original_attr = _has_original_dunder(
264 value,
265 allowed_types=self.allowed_getattr,
266 allowed_methods=self._getattr_methods,
267 allowed_external=allowed_getattr_external,
268 method_name="__getattr__",
269 )
271 accept = False
273 # Many objects do not have `__getattr__`, this is fine.
274 if has_original_attr is None and has_original_attribute:
275 accept = True
276 else:
277 # Accept objects without modifications to `__getattr__` and `__getattribute__`
278 accept = has_original_attr and has_original_attribute
280 if accept:
281 # We still need to check for overridden properties.
283 value_class = type(value)
284 if not hasattr(value_class, attr):
285 return True
287 class_attr_val = getattr(value_class, attr)
288 is_property = isinstance(class_attr_val, property)
290 if not is_property:
291 return True
293 # Properties in allowed types are ok (although we do not include any
294 # properties in our default allow list currently).
295 if type(value) in self.allowed_getattr:
296 return True # pragma: no cover
298 # Properties in subclasses of allowed types may be ok if not changed
299 for module_name, *access_path in allowed_getattr_external:
300 try:
301 external_class = _get_external(module_name, access_path)
302 external_class_attr_val = getattr(external_class, attr)
303 except (KeyError, AttributeError):
304 return False # pragma: no cover
305 return class_attr_val == external_class_attr_val
307 return False
309 def can_get_item(self, value, item):
310 """Allow accessing `__getiitem__` of allow-listed instances unless it was not modified."""
311 allowed_getitem_external = _coerce_path_to_tuples(self.allowed_getitem_external)
312 if self.allow_getitem_on_types:
313 # e.g. Union[str, int] or Literal[True, 1]
314 if isinstance(value, (typing._SpecialForm, typing._BaseGenericAlias)):
315 return True
316 # PEP 560 e.g. list[str]
317 if isinstance(value, type) and hasattr(value, "__class_getitem__"):
318 return True
319 return _has_original_dunder(
320 value,
321 allowed_types=self.allowed_getitem,
322 allowed_methods=self._getitem_methods,
323 allowed_external=allowed_getitem_external,
324 method_name="__getitem__",
325 )
327 def can_operate(self, dunders: tuple[str, ...], a, b=None):
328 allowed_operations_external = _coerce_path_to_tuples(
329 self.allowed_operations_external
330 )
331 objects = [a]
332 if b is not None:
333 objects.append(b)
334 return all(
335 [
336 _has_original_dunder(
337 obj,
338 allowed_types=self.allowed_operations,
339 allowed_methods=self._operator_dunder_methods(dunder),
340 allowed_external=allowed_operations_external,
341 method_name=dunder,
342 )
343 for dunder in dunders
344 for obj in objects
345 ]
346 )
348 def _operator_dunder_methods(self, dunder: str) -> set[Callable]:
349 if dunder not in self._operation_methods_cache:
350 self._operation_methods_cache[dunder] = self._safe_get_methods(
351 self.allowed_operations, dunder
352 )
353 return self._operation_methods_cache[dunder]
355 @cached_property
356 def _getitem_methods(self) -> set[Callable]:
357 return self._safe_get_methods(self.allowed_getitem, "__getitem__")
359 @cached_property
360 def _getattr_methods(self) -> set[Callable]:
361 return self._safe_get_methods(self.allowed_getattr, "__getattr__")
363 @cached_property
364 def _getattribute_methods(self) -> set[Callable]:
365 return self._safe_get_methods(self.allowed_getattr, "__getattribute__")
367 def _safe_get_methods(self, classes, name) -> set[Callable]:
368 return {
369 method
370 for class_ in classes
371 for method in [getattr(class_, name, None)]
372 if method
373 }
376class _DummyNamedTuple(NamedTuple):
377 """Used internally to retrieve methods of named tuple instance."""
380EvaluationPolicyName = Literal["forbidden", "minimal", "limited", "unsafe", "dangerous"]
383@dataclass
384class EvaluationContext:
385 #: Local namespace
386 locals: dict
387 #: Global namespace
388 globals: dict
389 #: Evaluation policy identifier
390 evaluation: EvaluationPolicyName = "forbidden"
391 #: Whether the evaluation of code takes place inside of a subscript.
392 #: Useful for evaluating ``:-1, 'col'`` in ``df[:-1, 'col']``.
393 in_subscript: bool = False
394 #: Auto import method
395 auto_import: Callable[[Sequence[str]], ModuleType] | None = None
396 #: Overrides for evaluation policy
397 policy_overrides: dict = field(default_factory=dict)
398 #: Transient local namespace used to store mocks
399 transient_locals: dict = field(default_factory=dict)
400 #: Transients of class level
401 class_transients: dict | None = None
402 #: Instance variable name used in the method definition
403 instance_arg_name: str | None = None
404 #: Currently associated value
405 #: Useful for adding items to _Duck on annotated assignment
406 current_value: ast.AST | None = None
408 def replace(self, /, **changes):
409 """Return a new copy of the context, with specified changes"""
410 return dataclasses.replace(self, **changes)
413class _IdentitySubscript:
414 """Returns the key itself when item is requested via subscript."""
416 def __getitem__(self, key):
417 return key
420IDENTITY_SUBSCRIPT = _IdentitySubscript()
421SUBSCRIPT_MARKER = "__SUBSCRIPT_SENTINEL__"
422UNKNOWN_SIGNATURE = Signature()
423NOT_EVALUATED = object()
426class GuardRejection(Exception):
427 """Exception raised when guard rejects evaluation attempt."""
429 pass
432def guarded_eval(code: str, context: EvaluationContext):
433 """Evaluate provided code in the evaluation context.
435 If evaluation policy given by context is set to ``forbidden``
436 no evaluation will be performed; if it is set to ``dangerous``
437 standard :func:`eval` will be used; finally, for any other,
438 policy :func:`eval_node` will be called on parsed AST.
439 """
440 locals_ = context.locals
442 if context.evaluation == "forbidden":
443 raise GuardRejection("Forbidden mode")
445 # note: not using `ast.literal_eval` as it does not implement
446 # getitem at all, for example it fails on simple `[0][1]`
448 if context.in_subscript:
449 # syntactic sugar for ellipsis (:) is only available in subscripts
450 # so we need to trick the ast parser into thinking that we have
451 # a subscript, but we need to be able to later recognise that we did
452 # it so we can ignore the actual __getitem__ operation
453 if not code:
454 return tuple()
455 locals_ = locals_.copy()
456 locals_[SUBSCRIPT_MARKER] = IDENTITY_SUBSCRIPT
457 code = SUBSCRIPT_MARKER + "[" + code + "]"
458 context = context.replace(locals=locals_)
460 if context.evaluation == "dangerous":
461 return eval(code, context.globals, context.locals)
463 node = ast.parse(code, mode="exec")
465 return eval_node(node, context)
468BINARY_OP_DUNDERS: dict[type[ast.operator], tuple[str]] = {
469 ast.Add: ("__add__",),
470 ast.Sub: ("__sub__",),
471 ast.Mult: ("__mul__",),
472 ast.Div: ("__truediv__",),
473 ast.FloorDiv: ("__floordiv__",),
474 ast.Mod: ("__mod__",),
475 ast.Pow: ("__pow__",),
476 ast.LShift: ("__lshift__",),
477 ast.RShift: ("__rshift__",),
478 ast.BitOr: ("__or__",),
479 ast.BitXor: ("__xor__",),
480 ast.BitAnd: ("__and__",),
481 ast.MatMult: ("__matmul__",),
482}
484COMP_OP_DUNDERS: dict[type[ast.cmpop], tuple[str, ...]] = {
485 ast.Eq: ("__eq__",),
486 ast.NotEq: ("__ne__", "__eq__"),
487 ast.Lt: ("__lt__", "__gt__"),
488 ast.LtE: ("__le__", "__ge__"),
489 ast.Gt: ("__gt__", "__lt__"),
490 ast.GtE: ("__ge__", "__le__"),
491 ast.In: ("__contains__",),
492 # Note: ast.Is, ast.IsNot, ast.NotIn are handled specially
493}
495UNARY_OP_DUNDERS: dict[type[ast.unaryop], tuple[str, ...]] = {
496 ast.USub: ("__neg__",),
497 ast.UAdd: ("__pos__",),
498 # we have to check both __inv__ and __invert__!
499 ast.Invert: ("__invert__", "__inv__"),
500 ast.Not: ("__not__",),
501}
503GENERIC_CONTAINER_TYPES = (dict, list, set, tuple, frozenset)
506class ImpersonatingDuck:
507 """A dummy class used to create objects of other classes without calling their ``__init__``"""
509 # no-op: override __class__ to impersonate
512class _Duck:
513 """A dummy class used to create objects pretending to have given attributes"""
515 def __init__(self, attributes: dict | None = None, items: dict | None = None):
516 self.attributes = attributes if attributes is not None else {}
517 self.items = items if items is not None else {}
519 def __getattr__(self, attr: str):
520 return self.attributes[attr]
522 def __hasattr__(self, attr: str):
523 return attr in self.attributes
525 def __dir__(self):
526 return [*dir(super), *self.attributes]
528 def __getitem__(self, key: str):
529 return self.items[key]
531 def __hasitem__(self, key: str):
532 return self.items[key]
534 def _ipython_key_completions_(self):
535 return self.items.keys()
538def _find_dunder(node_op, dunders) -> tuple[str, ...] | None:
539 dunder = None
540 for op, candidate_dunder in dunders.items():
541 if isinstance(node_op, op):
542 dunder = candidate_dunder
543 return dunder
546def get_policy(context: EvaluationContext) -> EvaluationPolicy:
547 policy = copy(EVALUATION_POLICIES[context.evaluation])
549 for key, value in context.policy_overrides.items():
550 if hasattr(policy, key):
551 setattr(policy, key, value)
552 return policy
555def _validate_policy_overrides(
556 policy_name: EvaluationPolicyName, policy_overrides: dict
557) -> bool:
558 policy = EVALUATION_POLICIES[policy_name]
560 all_good = True
561 for key, value in policy_overrides.items():
562 if not hasattr(policy, key):
563 warnings.warn(
564 f"Override {key!r} is not valid with {policy_name!r} evaluation policy"
565 )
566 all_good = False
567 return all_good
570def _is_type_annotation(obj) -> bool:
571 """
572 Returns True if obj is a type annotation, False otherwise.
573 """
574 if isinstance(obj, type):
575 return True
576 if isinstance(obj, types.GenericAlias):
577 return True
578 if hasattr(types, "UnionType") and isinstance(obj, types.UnionType):
579 return True
580 if isinstance(obj, (typing._SpecialForm, typing._BaseGenericAlias)):
581 return True
582 if isinstance(obj, typing.TypeVar):
583 return True
584 # Types that support __class_getitem__
585 if isinstance(obj, type) and hasattr(obj, "__class_getitem__"):
586 return True
587 # Fallback: check if get_origin returns something
588 if hasattr(typing, "get_origin") and get_origin(obj) is not None:
589 return True
591 return False
594def _handle_assign(node: ast.Assign, context: EvaluationContext):
595 value = eval_node(node.value, context)
596 transient_locals = context.transient_locals
597 policy = get_policy(context)
598 class_transients = context.class_transients
599 for target in node.targets:
600 if isinstance(target, (ast.Tuple, ast.List)):
601 # Handle unpacking assignment
602 values = list(value)
603 targets = target.elts
604 starred = [i for i, t in enumerate(targets) if isinstance(t, ast.Starred)]
606 # Unified handling: treat no starred as starred at end
607 star_or_last_idx = starred[0] if starred else len(targets)
609 # Before starred
610 for i in range(star_or_last_idx):
611 # Check for self.x assignment
612 if _is_instance_attribute_assignment(targets[i], context):
613 class_transients[targets[i].attr] = values[i]
614 else:
615 transient_locals[targets[i].id] = values[i]
617 # Starred if exists
618 if starred:
619 end = len(values) - (len(targets) - star_or_last_idx - 1)
620 if _is_instance_attribute_assignment(
621 targets[star_or_last_idx], context
622 ):
623 class_transients[targets[star_or_last_idx].attr] = values[
624 star_or_last_idx:end
625 ]
626 else:
627 transient_locals[targets[star_or_last_idx].value.id] = values[
628 star_or_last_idx:end
629 ]
631 # After starred
632 for i in range(star_or_last_idx + 1, len(targets)):
633 if _is_instance_attribute_assignment(targets[i], context):
634 class_transients[targets[i].attr] = values[
635 len(values) - (len(targets) - i)
636 ]
637 else:
638 transient_locals[targets[i].id] = values[
639 len(values) - (len(targets) - i)
640 ]
641 elif isinstance(target, ast.Subscript):
642 if isinstance(target.value, ast.Name):
643 name = target.value.id
644 container = transient_locals.get(name)
645 if container is None:
646 container = context.locals.get(name)
647 if container is None:
648 container = context.globals.get(name)
649 if container is None:
650 raise NameError(
651 f"{name} not found in locals, globals, nor builtins"
652 )
653 storage_dict = transient_locals
654 storage_key = name
655 elif isinstance(
656 target.value, ast.Attribute
657 ) and _is_instance_attribute_assignment(target.value, context):
658 attr = target.value.attr
659 container = class_transients.get(attr, None)
660 if container is None:
661 raise NameError(f"{attr} not found in class transients")
662 storage_dict = class_transients
663 storage_key = attr
664 else:
665 return
667 key = eval_node(target.slice, context)
668 attributes = (
669 dict.fromkeys(dir(container))
670 if policy.can_call(container.__dir__)
671 else {}
672 )
673 items = {}
675 if policy.can_get_item(container, None):
676 try:
677 items = dict(container.items())
678 except Exception:
679 pass
681 items[key] = value
682 duck_container = _Duck(attributes=attributes, items=items)
683 storage_dict[storage_key] = duck_container
684 elif _is_instance_attribute_assignment(target, context):
685 class_transients[target.attr] = value
686 else:
687 transient_locals[target.id] = value
688 return None
691def _handle_annassign(node, context):
692 context_with_value = context.replace(current_value=getattr(node, "value", None))
693 annotation_result = eval_node(node.annotation, context_with_value)
694 if _is_type_annotation(annotation_result):
695 annotation_value = _resolve_annotation(annotation_result, context)
696 # Use Value for generic types
697 use_value = (
698 isinstance(annotation_value, GENERIC_CONTAINER_TYPES) and node.value is not None
699 )
700 else:
701 annotation_value = annotation_result
702 use_value = False
704 # LOCAL VARIABLE
705 if getattr(node, "simple", False) and isinstance(node.target, ast.Name):
706 name = node.target.id
707 if use_value:
708 return _handle_assign(
709 ast.Assign(targets=[node.target], value=node.value), context
710 )
711 context.transient_locals[name] = annotation_value
712 return None
714 # INSTANCE ATTRIBUTE
715 if _is_instance_attribute_assignment(node.target, context):
716 attr = node.target.attr
717 if use_value:
718 return _handle_assign(
719 ast.Assign(targets=[node.target], value=node.value), context
720 )
721 context.class_transients[attr] = annotation_value
722 return None
724 return None
726def _extract_args_and_kwargs(node: ast.Call, context: EvaluationContext):
727 args = [eval_node(arg, context) for arg in node.args]
728 kwargs = {
729 k: v
730 for kw in node.keywords
731 for k, v in (
732 {kw.arg: eval_node(kw.value, context)}
733 if kw.arg
734 else eval_node(kw.value, context)
735 ).items()
736 }
737 return args, kwargs
740def _is_instance_attribute_assignment(
741 target: ast.AST, context: EvaluationContext
742) -> bool:
743 """Return True if target is an attribute access on the instance argument."""
744 return (
745 context.class_transients is not None
746 and context.instance_arg_name is not None
747 and isinstance(target, ast.Attribute)
748 and isinstance(getattr(target, "value", None), ast.Name)
749 and getattr(target.value, "id", None) == context.instance_arg_name
750 )
753def _get_coroutine_attributes() -> dict[str, object | None]:
754 async def _dummy():
755 return None
757 coro = _dummy()
758 try:
759 return {attr: getattr(coro, attr, None) for attr in dir(coro)}
760 finally:
761 coro.close()
764def eval_node(node: ast.AST | None, context: EvaluationContext):
765 """Evaluate AST node in provided context.
767 Applies evaluation restrictions defined in the context. Currently does not support evaluation of functions with keyword arguments.
769 Does not evaluate actions that always have side effects:
771 - class definitions (``class sth: ...``)
772 - function definitions (``def sth: ...``)
773 - variable assignments (``x = 1``)
774 - augmented assignments (``x += 1``)
775 - deletions (``del x``)
777 Does not evaluate operations which do not return values:
779 - assertions (``assert x``)
780 - pass (``pass``)
781 - imports (``import x``)
782 - control flow:
784 - conditionals (``if x:``) except for ternary IfExp (``a if x else b``)
785 - loops (``for`` and ``while``)
786 - exception handling
788 The purpose of this function is to guard against unwanted side-effects;
789 it does not give guarantees on protection from malicious code execution.
790 """
791 policy = get_policy(context)
793 if node is None:
794 return None
795 if isinstance(node, (ast.Interactive, ast.Module)):
796 result = None
797 for child_node in node.body:
798 result = eval_node(child_node, context)
799 return result
800 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
801 is_async = isinstance(node, ast.AsyncFunctionDef)
802 func_locals = context.transient_locals.copy()
803 func_context = context.replace(transient_locals=func_locals)
804 is_property = False
805 is_static = False
806 is_classmethod = False
807 for decorator_node in node.decorator_list:
808 try:
809 decorator = eval_node(decorator_node, context)
810 except NameError:
811 # if the decorator is not yet defined this is fine
812 # especialy because we don't handle imports yet
813 continue
814 if decorator is property:
815 is_property = True
816 elif decorator is staticmethod:
817 is_static = True
818 elif decorator is classmethod:
819 is_classmethod = True
821 if func_context.class_transients is not None:
822 if not is_static and not is_classmethod:
823 func_context.instance_arg_name = (
824 node.args.args[0].arg if node.args.args else None
825 )
827 return_type = eval_node(node.returns, context=context)
829 for child_node in node.body:
830 eval_node(child_node, func_context)
832 if is_property:
833 if return_type is not None:
834 if _is_type_annotation(return_type):
835 context.transient_locals[node.name] = _resolve_annotation(
836 return_type, context
837 )
838 else:
839 context.transient_locals[node.name] = return_type
840 else:
841 return_value = _infer_return_value(node, func_context)
842 context.transient_locals[node.name] = return_value
844 return None
846 def dummy_function(*args, **kwargs):
847 pass
849 if return_type is not None:
850 if _is_type_annotation(return_type):
851 dummy_function.__annotations__["return"] = return_type
852 else:
853 dummy_function.__inferred_return__ = return_type
854 else:
855 inferred_return = _infer_return_value(node, func_context)
856 if inferred_return is not None:
857 dummy_function.__inferred_return__ = inferred_return
859 dummy_function.__name__ = node.name
860 dummy_function.__node__ = node
861 dummy_function.__is_async__ = is_async
862 context.transient_locals[node.name] = dummy_function
863 return None
864 if isinstance(node, ast.Lambda):
866 def dummy_function(*args, **kwargs):
867 pass
869 dummy_function.__inferred_return__ = eval_node(node.body, context)
870 return dummy_function
871 if isinstance(node, ast.ClassDef):
872 # TODO support class decorators?
873 class_locals = {}
874 outer_locals = context.locals.copy()
875 outer_locals.update(context.transient_locals)
876 class_context = context.replace(
877 transient_locals=class_locals, locals=outer_locals
878 )
879 class_context.class_transients = class_locals
880 for child_node in node.body:
881 eval_node(child_node, class_context)
882 bases = tuple([eval_node(base, context) for base in node.bases])
883 dummy_class = type(node.name, bases, class_locals)
884 context.transient_locals[node.name] = dummy_class
885 return None
886 if isinstance(node, ast.Await):
887 value = eval_node(node.value, context)
888 if hasattr(value, "__awaited_type__"):
889 return value.__awaited_type__
890 return value
891 if isinstance(node, ast.While):
892 loop_locals = context.transient_locals.copy()
893 loop_context = context.replace(transient_locals=loop_locals)
895 result = None
896 for stmt in node.body:
897 result = eval_node(stmt, loop_context)
899 policy = get_policy(context)
900 merged_locals = _merge_dicts_by_key(
901 [loop_locals, context.transient_locals.copy()], policy
902 )
903 context.transient_locals.update(merged_locals)
905 return result
906 if isinstance(node, ast.For):
907 try:
908 iterable = eval_node(node.iter, context)
909 except Exception:
910 iterable = None
912 sample = None
913 if iterable is not None:
914 try:
915 if policy.can_call(getattr(iterable, "__iter__", None)):
916 sample = next(iter(iterable))
917 except Exception:
918 sample = None
920 loop_locals = context.transient_locals.copy()
921 loop_context = context.replace(transient_locals=loop_locals)
923 if sample is not None:
924 try:
925 fake_assign = ast.Assign(
926 targets=[node.target], value=ast.Constant(value=sample)
927 )
928 _handle_assign(fake_assign, loop_context)
929 except Exception:
930 pass
932 result = None
933 for stmt in node.body:
934 result = eval_node(stmt, loop_context)
936 policy = get_policy(context)
937 merged_locals = _merge_dicts_by_key(
938 [loop_locals, context.transient_locals.copy()], policy
939 )
940 context.transient_locals.update(merged_locals)
942 return result
943 if isinstance(node, ast.If):
944 branches = []
945 current = node
946 result = None
947 while True:
948 branch_locals = context.transient_locals.copy()
949 branch_context = context.replace(transient_locals=branch_locals)
950 for stmt in current.body:
951 result = eval_node(stmt, branch_context)
952 branches.append(branch_locals)
953 if not current.orelse:
954 break
955 elif len(current.orelse) == 1 and isinstance(current.orelse[0], ast.If):
956 # It's an elif - continue loop
957 current = current.orelse[0]
958 else:
959 # It's an else block - process and break
960 else_locals = context.transient_locals.copy()
961 else_context = context.replace(transient_locals=else_locals)
962 for stmt in current.orelse:
963 result = eval_node(stmt, else_context)
964 branches.append(else_locals)
965 break
966 branches.append(context.transient_locals.copy())
967 policy = get_policy(context)
968 merged_locals = _merge_dicts_by_key(branches, policy)
969 context.transient_locals.update(merged_locals)
970 return result
971 if isinstance(node, ast.Assign):
972 return _handle_assign(node, context)
973 if isinstance(node, ast.AnnAssign):
974 return _handle_annassign(node, context)
975 if isinstance(node, ast.Expression):
976 return eval_node(node.body, context)
977 if isinstance(node, ast.Expr):
978 return eval_node(node.value, context)
979 if isinstance(node, ast.Pass):
980 return None
981 if isinstance(node, ast.Import):
982 # TODO: populate transient_locals
983 return None
984 if isinstance(node, (ast.AugAssign, ast.Delete)):
985 return None
986 if isinstance(node, (ast.Global, ast.Nonlocal)):
987 return None
988 if isinstance(node, ast.BinOp):
989 left = eval_node(node.left, context)
990 right = eval_node(node.right, context)
991 if (
992 isinstance(node.op, ast.BitOr)
993 and _is_type_annotation(left)
994 and _is_type_annotation(right)
995 ):
996 left_duck = (
997 _Duck(dict.fromkeys(dir(left)))
998 if policy.can_call(left.__dir__)
999 else _Duck()
1000 )
1001 right_duck = (
1002 _Duck(dict.fromkeys(dir(right)))
1003 if policy.can_call(right.__dir__)
1004 else _Duck()
1005 )
1006 value_node = context.current_value
1007 if value_node is not None and isinstance(value_node, ast.Dict):
1008 if dict in [left, right]:
1009 return _merge_values(
1010 [left_duck, right_duck, ast.literal_eval(value_node)],
1011 policy=get_policy(context),
1012 )
1013 return _merge_values([left_duck, right_duck], policy=get_policy(context))
1014 dunders = _find_dunder(node.op, BINARY_OP_DUNDERS)
1015 if dunders:
1016 if policy.can_operate(dunders, left, right):
1017 return getattr(left, dunders[0])(right)
1018 else:
1019 raise GuardRejection(
1020 f"Operation (`{dunders}`) for",
1021 type(left),
1022 f"not allowed in {context.evaluation} mode",
1023 )
1024 if isinstance(node, ast.Compare):
1025 left = eval_node(node.left, context)
1026 all_true = True
1027 negate = False
1028 for op, right in zip(node.ops, node.comparators):
1029 right = eval_node(right, context)
1030 dunder = None
1031 dunders = _find_dunder(op, COMP_OP_DUNDERS)
1032 if not dunders:
1033 if isinstance(op, ast.NotIn):
1034 dunders = COMP_OP_DUNDERS[ast.In]
1035 negate = True
1036 if isinstance(op, ast.Is):
1037 dunder = "is_"
1038 if isinstance(op, ast.IsNot):
1039 dunder = "is_"
1040 negate = True
1041 if not dunder and dunders:
1042 dunder = dunders[0]
1043 if dunder:
1044 a, b = (right, left) if dunder == "__contains__" else (left, right)
1045 if dunder == "is_" or dunders and policy.can_operate(dunders, a, b):
1046 result = getattr(operator, dunder)(a, b)
1047 if negate:
1048 result = not result
1049 if not result:
1050 all_true = False
1051 left = right
1052 else:
1053 raise GuardRejection(
1054 f"Comparison (`{dunder}`) for",
1055 type(left),
1056 f"not allowed in {context.evaluation} mode",
1057 )
1058 else:
1059 raise ValueError(
1060 f"Comparison `{dunder}` not supported"
1061 ) # pragma: no cover
1062 return all_true
1063 if isinstance(node, ast.Constant):
1064 return node.value
1065 if isinstance(node, ast.Tuple):
1066 return tuple(eval_node(e, context) for e in node.elts)
1067 if isinstance(node, ast.List):
1068 return [eval_node(e, context) for e in node.elts]
1069 if isinstance(node, ast.Set):
1070 return {eval_node(e, context) for e in node.elts}
1071 if isinstance(node, ast.Dict):
1072 return dict(
1073 zip(
1074 [eval_node(k, context) for k in node.keys],
1075 [eval_node(v, context) for v in node.values],
1076 )
1077 )
1078 if isinstance(node, ast.Slice):
1079 return slice(
1080 eval_node(node.lower, context),
1081 eval_node(node.upper, context),
1082 eval_node(node.step, context),
1083 )
1084 if isinstance(node, ast.UnaryOp):
1085 value = eval_node(node.operand, context)
1086 dunders = _find_dunder(node.op, UNARY_OP_DUNDERS)
1087 if dunders:
1088 if policy.can_operate(dunders, value):
1089 try:
1090 return getattr(value, dunders[0])()
1091 except AttributeError:
1092 raise TypeError(
1093 f"bad operand type for unary {node.op}: {type(value)}"
1094 )
1095 else:
1096 raise GuardRejection(
1097 f"Operation (`{dunders}`) for",
1098 type(value),
1099 f"not allowed in {context.evaluation} mode",
1100 )
1101 if isinstance(node, ast.Subscript):
1102 value = eval_node(node.value, context)
1103 slice_ = eval_node(node.slice, context)
1104 if policy.can_get_item(value, slice_):
1105 return value[slice_]
1106 raise GuardRejection(
1107 "Subscript access (`__getitem__`) for",
1108 type(value), # not joined to avoid calling `repr`
1109 f" not allowed in {context.evaluation} mode",
1110 )
1111 if isinstance(node, ast.Name):
1112 return _eval_node_name(node.id, context)
1113 if isinstance(node, ast.Attribute):
1114 if (
1115 context.class_transients is not None
1116 and isinstance(node.value, ast.Name)
1117 and node.value.id == context.instance_arg_name
1118 ):
1119 return context.class_transients.get(node.attr)
1120 value = eval_node(node.value, context)
1121 if policy.can_get_attr(value, node.attr):
1122 return getattr(value, node.attr)
1123 try:
1124 cls = (
1125 value if isinstance(value, type) else getattr(value, "__class__", None)
1126 )
1127 if cls is not None:
1128 hints = _collect_annotations(cls)
1129 if node.attr in hints:
1130 return _resolve_annotation(hints[node.attr], context)
1131 except Exception:
1132 # Fall through to the guard rejection
1133 pass
1134 raise GuardRejection(
1135 "Attribute access (`__getattr__`) for",
1136 type(value), # not joined to avoid calling `repr`
1137 f"not allowed in {context.evaluation} mode",
1138 )
1139 if isinstance(node, ast.IfExp):
1140 test = eval_node(node.test, context)
1141 if test:
1142 return eval_node(node.body, context)
1143 else:
1144 return eval_node(node.orelse, context)
1145 if isinstance(node, ast.Call):
1146 func = eval_node(node.func, context)
1147 if policy.can_call(func):
1148 args, kwargs = _extract_args_and_kwargs(node, context)
1149 return func(*args, **kwargs)
1150 if isclass(func):
1151 # this code path gets entered when calling class e.g. `MyClass()`
1152 # or `my_instance.__class__()` - in both cases `func` is `MyClass`.
1153 # Should return `MyClass` if `__new__` is not overridden,
1154 # otherwise whatever `__new__` return type is.
1155 overridden_return_type = _eval_return_type(func.__new__, node, context)
1156 if overridden_return_type is not NOT_EVALUATED:
1157 return overridden_return_type
1158 return _create_duck_for_heap_type(func)
1159 else:
1160 inferred_return = getattr(func, "__inferred_return__", NOT_EVALUATED)
1161 return_type = _eval_return_type(func, node, context)
1162 if getattr(func, "__is_async__", False):
1163 awaited_type = (
1164 inferred_return if inferred_return is not None else return_type
1165 )
1166 coroutine_duck = _Duck(attributes=_get_coroutine_attributes())
1167 coroutine_duck.__awaited_type__ = awaited_type
1168 return coroutine_duck
1169 if inferred_return is not NOT_EVALUATED:
1170 return inferred_return
1171 if return_type is not NOT_EVALUATED:
1172 return return_type
1173 raise GuardRejection(
1174 "Call for",
1175 func, # not joined to avoid calling `repr`
1176 f"not allowed in {context.evaluation} mode",
1177 )
1178 if isinstance(node, ast.Assert):
1179 # message is always the second item, so if it is defined user would be completing
1180 # on the message, not on the assertion test
1181 if node.msg:
1182 return eval_node(node.msg, context)
1183 return eval_node(node.test, context)
1184 return None
1187def _merge_dicts_by_key(dicts: list, policy: EvaluationPolicy):
1188 """Merge multiple dictionaries, combining values for each key."""
1189 if len(dicts) == 1:
1190 return dicts[0]
1192 all_keys = set()
1193 for d in dicts:
1194 all_keys.update(d.keys())
1196 merged = {}
1197 for key in all_keys:
1198 values = [d[key] for d in dicts if key in d]
1199 if values:
1200 merged[key] = _merge_values(values, policy)
1202 return merged
1205def _merge_values(values, policy: EvaluationPolicy):
1206 """Recursively merge multiple values, combining attributes and dict items."""
1207 if len(values) == 1:
1208 return values[0]
1210 types = {type(v) for v in values}
1211 merged_items = None
1212 key_values = {}
1213 attributes = set()
1214 for v in values:
1215 if policy.can_call(v.__dir__):
1216 attributes.update(dir(v))
1217 try:
1218 if policy.can_call(v.items):
1219 try:
1220 for k, val in v.items():
1221 key_values.setdefault(k, []).append(val)
1222 except Exception as e:
1223 pass
1224 elif policy.can_call(v.keys):
1225 try:
1226 for k in v.keys():
1227 key_values.setdefault(k, []).append(None)
1228 except Exception as e:
1229 pass
1230 except Exception as e:
1231 pass
1233 if key_values:
1234 merged_items = {
1235 k: _merge_values(vals, policy) if vals[0] is not None else None
1236 for k, vals in key_values.items()
1237 }
1239 if len(types) == 1:
1240 t = next(iter(types))
1241 if t is not dict and not (
1242 hasattr(next(iter(values)), "__getitem__")
1243 and (
1244 hasattr(next(iter(values)), "items")
1245 or hasattr(next(iter(values)), "keys")
1246 )
1247 ):
1248 if t in (list, set, tuple):
1249 return t
1250 return values[0]
1252 return _Duck(attributes=dict.fromkeys(attributes), items=merged_items)
1255def _infer_return_value(node: ast.FunctionDef, context: EvaluationContext):
1256 """Infer the return value(s) of a function by evaluating all return statements."""
1257 return_values = _collect_return_values(node.body, context)
1259 if not return_values:
1260 return None
1261 if len(return_values) == 1:
1262 return return_values[0]
1264 policy = get_policy(context)
1265 return _merge_values(return_values, policy)
1268def _collect_return_values(body, context):
1269 """Recursively collect return values from a list of AST statements."""
1270 return_values = []
1271 for stmt in body:
1272 if isinstance(stmt, ast.Return):
1273 if stmt.value is None:
1274 continue
1275 try:
1276 value = eval_node(stmt.value, context)
1277 if value is not None and value is not NOT_EVALUATED:
1278 return_values.append(value)
1279 except Exception:
1280 pass
1281 if isinstance(
1282 stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)
1283 ):
1284 continue
1285 elif hasattr(stmt, "body") and isinstance(stmt.body, list):
1286 return_values.extend(_collect_return_values(stmt.body, context))
1287 if isinstance(stmt, ast.Try):
1288 for h in stmt.handlers:
1289 if hasattr(h, "body"):
1290 return_values.extend(_collect_return_values(h.body, context))
1291 if hasattr(stmt, "orelse"):
1292 return_values.extend(_collect_return_values(stmt.orelse, context))
1293 if hasattr(stmt, "finalbody"):
1294 return_values.extend(_collect_return_values(stmt.finalbody, context))
1295 if hasattr(stmt, "orelse") and isinstance(stmt.orelse, list):
1296 return_values.extend(_collect_return_values(stmt.orelse, context))
1297 return return_values
1300def _eval_return_type(func: Callable, node: ast.Call, context: EvaluationContext):
1301 """Evaluate return type of a given callable function.
1303 Returns the built-in type, a duck or NOT_EVALUATED sentinel.
1304 """
1305 try:
1306 sig = signature(func)
1307 except ValueError:
1308 sig = UNKNOWN_SIGNATURE
1309 # if annotation was not stringized, or it was stringized
1310 # but resolved by signature call we know the return type
1311 not_empty = sig.return_annotation is not Signature.empty
1312 if not_empty:
1313 return _resolve_annotation(sig.return_annotation, context, sig, func, node)
1314 return NOT_EVALUATED
1317def _collect_annotations(cls: type) -> dict:
1318 """Collect annotations of a class and its bases without resolving them.
1320 `typing.get_type_hints()` is not usable here because it resolves stringized
1321 annotations with `eval()`; under PEP 563 every annotation of a module is a
1322 string, so that would run arbitrary code from `__annotations__`. Strings are
1323 left as-is and later resolved by `_eval_annotation` under the policy.
1324 """
1325 annotations: dict = {}
1326 for base in reversed(cls.__mro__):
1327 annotations.update(get_annotations(base, eval_str=False))
1328 return annotations
1331def _eval_annotation(
1332 annotation: str,
1333 context: EvaluationContext,
1334):
1335 if not isinstance(annotation, str):
1336 return annotation
1337 return eval_node(ast.parse(annotation, mode="eval").body, context)
1340class _GetItemDuck(dict):
1341 """A dict subclass that always returns the factory instance and claims to have any item."""
1343 def __init__(self, factory, *args, **kwargs):
1344 super().__init__(*args, **kwargs)
1345 self._factory = factory
1347 def __getitem__(self, key):
1348 return self._factory()
1350 def __contains__(self, key):
1351 return True
1354def _resolve_annotation(
1355 annotation: object | str,
1356 context: EvaluationContext,
1357 sig: Signature | None = None,
1358 func: Callable | None = None,
1359 node: ast.Call | None = None,
1360):
1361 """Resolve annotation created by user with `typing` module and custom objects."""
1362 if annotation is None:
1363 return None
1364 annotation = _eval_annotation(annotation, context)
1365 origin = get_origin(annotation)
1366 if annotation is Self and func and hasattr(func, "__self__"):
1367 return func.__self__
1368 elif origin is Literal:
1369 type_args = get_args(annotation)
1370 if len(type_args) == 1:
1371 return type_args[0]
1372 elif annotation is LiteralString:
1373 return ""
1374 elif annotation is AnyStr:
1375 index = None
1376 if func and hasattr(func, "__node__"):
1377 def_node = func.__node__
1378 for i, arg in enumerate(def_node.args.args):
1379 if not arg.annotation:
1380 continue
1381 annotation = _eval_annotation(arg.annotation.id, context)
1382 if annotation is AnyStr:
1383 index = i
1384 break
1385 is_bound_method = (
1386 isinstance(func, MethodType) and getattr(func, "__self__") is not None
1387 )
1388 if index and is_bound_method:
1389 index -= 1
1390 elif sig:
1391 for i, (key, value) in enumerate(sig.parameters.items()):
1392 if value.annotation is AnyStr:
1393 index = i
1394 break
1395 if index is None:
1396 return None
1397 if index < 0 or index >= len(node.args):
1398 return None
1399 return eval_node(node.args[index], context)
1400 elif origin is TypeGuard:
1401 return False
1402 elif origin is set or origin is list:
1403 # only one type argument allowed
1404 attributes = [
1405 attr
1406 for attr in dir(
1407 _resolve_annotation(get_args(annotation)[0], context, sig, func, node)
1408 )
1409 ]
1410 duck = _Duck(attributes=dict.fromkeys(attributes))
1411 return _Duck(
1412 attributes=dict.fromkeys(dir(origin())),
1413 # items are not strrictly needed for set
1414 items=_GetItemDuck(lambda: duck),
1415 )
1416 elif origin is tuple:
1417 # multiple type arguments
1418 return tuple(
1419 _resolve_annotation(arg, context, sig, func, node)
1420 for arg in get_args(annotation)
1421 )
1422 elif origin is Union:
1423 # multiple type arguments
1424 attributes = [
1425 attr
1426 for type_arg in get_args(annotation)
1427 for attr in dir(_resolve_annotation(type_arg, context, sig, func, node))
1428 ]
1429 return _Duck(attributes=dict.fromkeys(attributes))
1430 elif is_typeddict(annotation):
1431 return _Duck(
1432 attributes=dict.fromkeys(dir(dict())),
1433 items={
1434 k: _resolve_annotation(v, context, sig, func, node)
1435 for k, v in annotation.__annotations__.items()
1436 },
1437 )
1438 elif hasattr(annotation, "_is_protocol"):
1439 return _Duck(attributes=dict.fromkeys(dir(annotation)))
1440 elif origin is Annotated:
1441 type_arg = get_args(annotation)[0]
1442 return _resolve_annotation(type_arg, context, sig, func, node)
1443 elif isinstance(annotation, NewType):
1444 return _eval_or_create_duck(annotation.__supertype__, context)
1445 elif isinstance(annotation, TypeAliasType):
1446 return _eval_or_create_duck(annotation.__value__, context)
1447 else:
1448 return _eval_or_create_duck(annotation, context)
1451def _eval_node_name(node_id: str, context: EvaluationContext):
1452 policy = get_policy(context)
1453 if node_id in context.transient_locals:
1454 return context.transient_locals[node_id]
1455 if policy.allow_locals_access and node_id in context.locals:
1456 return context.locals[node_id]
1457 if policy.allow_globals_access and node_id in context.globals:
1458 return context.globals[node_id]
1459 if policy.allow_builtins_access and hasattr(builtins, node_id):
1460 # note: do not use __builtins__, it is implementation detail of cPython
1461 return getattr(builtins, node_id)
1462 if policy.allow_auto_import and context.auto_import:
1463 return context.auto_import(node_id)
1464 if not policy.allow_globals_access and not policy.allow_locals_access:
1465 raise GuardRejection(
1466 f"Namespace access not allowed in {context.evaluation} mode"
1467 )
1468 else:
1469 raise NameError(f"{node_id} not found in locals, globals, nor builtins")
1472def _eval_or_create_duck(duck_type, context: EvaluationContext):
1473 policy = get_policy(context)
1474 # if allow-listed builtin is on type annotation, instantiate it
1475 if policy.can_call(duck_type):
1476 return duck_type()
1477 # if custom class is in type annotation, mock it
1478 return _create_duck_for_heap_type(duck_type)
1481def _create_duck_for_heap_type(duck_type):
1482 """Create an imitation of an object of a given type (a duck).
1484 Returns the duck or NOT_EVALUATED sentinel if duck could not be created.
1485 """
1486 duck = ImpersonatingDuck()
1487 try:
1488 # this only works for heap types, not builtins
1489 duck.__class__ = duck_type
1490 return duck
1491 except TypeError:
1492 pass
1493 return NOT_EVALUATED
1496SUPPORTED_EXTERNAL_GETITEM = {
1497 ("pandas", "core", "indexing", "_iLocIndexer"),
1498 ("pandas", "core", "indexing", "_LocIndexer"),
1499 ("pandas", "DataFrame"),
1500 ("pandas", "Series"),
1501 ("numpy", "ndarray"),
1502 ("numpy", "void"),
1503}
1506BUILTIN_GETITEM: set[InstancesHaveGetItem] = {
1507 dict,
1508 str, # type: ignore[arg-type]
1509 bytes, # type: ignore[arg-type]
1510 list,
1511 tuple,
1512 type, # for type annotations like list[str]
1513 _Duck,
1514 collections.defaultdict,
1515 collections.deque,
1516 collections.OrderedDict,
1517 collections.ChainMap,
1518 collections.UserDict,
1519 collections.UserList,
1520 collections.UserString, # type: ignore[arg-type]
1521 _DummyNamedTuple,
1522 _IdentitySubscript,
1523}
1526def _list_methods(cls, source=None):
1527 """For use on immutable objects or with methods returning a copy"""
1528 return [getattr(cls, k) for k in (source if source else dir(cls))]
1531dict_non_mutating_methods = ("copy", "keys", "values", "items")
1532list_non_mutating_methods = ("copy", "index", "count")
1533set_non_mutating_methods = set(dir(set)) & set(dir(frozenset))
1536dict_keys: type[collections.abc.KeysView] = type({}.keys())
1537dict_values: type = type({}.values())
1538dict_items: type = type({}.items())
1540NUMERICS = {int, float, complex}
1542ALLOWED_CALLS = {
1543 bytes,
1544 *_list_methods(bytes),
1545 bytes.__iter__,
1546 dict,
1547 *_list_methods(dict, dict_non_mutating_methods),
1548 dict.__iter__,
1549 dict_keys.__iter__,
1550 dict_values.__iter__,
1551 dict_items.__iter__,
1552 dict_keys.isdisjoint,
1553 list,
1554 *_list_methods(list, list_non_mutating_methods),
1555 list.__iter__,
1556 set,
1557 *_list_methods(set, set_non_mutating_methods),
1558 set.__iter__,
1559 frozenset,
1560 *_list_methods(frozenset),
1561 frozenset.__iter__,
1562 range,
1563 range.__iter__,
1564 str,
1565 *_list_methods(str),
1566 str.__iter__,
1567 tuple,
1568 *_list_methods(tuple),
1569 tuple.__iter__,
1570 bool,
1571 *_list_methods(bool),
1572 *NUMERICS,
1573 *[method for numeric_cls in NUMERICS for method in _list_methods(numeric_cls)],
1574 collections.deque,
1575 *_list_methods(collections.deque, list_non_mutating_methods),
1576 collections.deque.__iter__,
1577 collections.defaultdict,
1578 *_list_methods(collections.defaultdict, dict_non_mutating_methods),
1579 collections.defaultdict.__iter__,
1580 collections.OrderedDict,
1581 *_list_methods(collections.OrderedDict, dict_non_mutating_methods),
1582 collections.OrderedDict.__iter__,
1583 collections.UserDict,
1584 *_list_methods(collections.UserDict, dict_non_mutating_methods),
1585 collections.UserDict.__iter__,
1586 collections.UserList,
1587 *_list_methods(collections.UserList, list_non_mutating_methods),
1588 collections.UserList.__iter__,
1589 collections.UserString,
1590 *_list_methods(collections.UserString, dir(str)),
1591 collections.UserString.__iter__,
1592 collections.Counter,
1593 *_list_methods(collections.Counter, dict_non_mutating_methods),
1594 collections.Counter.__iter__,
1595 collections.Counter.elements,
1596 collections.Counter.most_common,
1597 object.__dir__,
1598 type.__dir__,
1599 _Duck.__dir__,
1600}
1602BUILTIN_GETATTR: set[MayHaveGetattr] = {
1603 *BUILTIN_GETITEM,
1604 set,
1605 frozenset,
1606 object,
1607 type, # `type` handles a lot of generic cases, e.g. numbers as in `int.real`.
1608 *NUMERICS,
1609 dict_keys,
1610 MethodDescriptorType,
1611 ModuleType,
1612}
1615BUILTIN_OPERATIONS = {*BUILTIN_GETATTR}
1617EVALUATION_POLICIES = {
1618 "minimal": EvaluationPolicy(
1619 allow_builtins_access=True,
1620 allow_locals_access=False,
1621 allow_globals_access=False,
1622 allow_item_access=False,
1623 allow_attr_access=False,
1624 allowed_calls=set(),
1625 allow_any_calls=False,
1626 allow_all_operations=False,
1627 ),
1628 "limited": SelectivePolicy(
1629 allowed_getitem=BUILTIN_GETITEM,
1630 allowed_getitem_external=SUPPORTED_EXTERNAL_GETITEM,
1631 allowed_getattr=BUILTIN_GETATTR,
1632 allowed_getattr_external={
1633 # pandas Series/Frame implements custom `__getattr__`
1634 ("pandas", "DataFrame"),
1635 ("pandas", "Series"),
1636 },
1637 allowed_operations=BUILTIN_OPERATIONS,
1638 allow_builtins_access=True,
1639 allow_locals_access=True,
1640 allow_globals_access=True,
1641 allow_getitem_on_types=True,
1642 allowed_calls=ALLOWED_CALLS,
1643 ),
1644 "unsafe": EvaluationPolicy(
1645 allow_builtins_access=True,
1646 allow_locals_access=True,
1647 allow_globals_access=True,
1648 allow_attr_access=True,
1649 allow_item_access=True,
1650 allow_any_calls=True,
1651 allow_all_operations=True,
1652 ),
1653}
1656__all__ = [
1657 "guarded_eval",
1658 "eval_node",
1659 "GuardRejection",
1660 "EvaluationContext",
1661 "_unbind_method",
1662]