"""Static checks for Python state shared between AnySystem processes. The checker parses source files only. It never imports or executes a student's solution. Local imports which can be resolved next to the entrypoint are checked recursively; calls through external imports are limited to explicitly reviewed APIs. """ from __future__ import annotations import ast from pathlib import Path from typing import Callable, Iterable MUTABLE_NODES = ( ast.List, ast.ListComp, ast.Dict, ast.DictComp, ast.Set, ast.SetComp, ast.GeneratorExp, ) MUTABLE_CONSTRUCTORS = { "builtins.bytearray", "builtins.dict", "builtins.list", "builtins.set", "collections.ChainMap", "collections.Counter", "collections.OrderedDict", "collections.defaultdict", "collections.deque", "weakref.WeakKeyDictionary", "weakref.WeakValueDictionary", } IMMUTABLE_SCALAR_CONSTRUCTORS = { "builtins.bool", "builtins.bytes", "builtins.complex", "builtins.float", "builtins.int", "builtins.range", "builtins.str", } KNOWN_IMMUTABLE_CALLS = { "os.getcwd", "os.path.abspath", "os.path.basename", "os.path.dirname", "os.path.join", "os.path.normpath", "os.path.realpath", "random.randint", "random.randrange", } TYPE_SUBSCRIPT_NAMES = { "builtins.dict", "builtins.frozenset", "builtins.list", "builtins.set", "builtins.tuple", "builtins.type", "collections.abc.Mapping", "collections.abc.Sequence", "typing.Annotated", "typing.Callable", "typing.ClassVar", "typing.Dict", "typing.FrozenSet", "typing.List", "typing.Literal", "typing.Mapping", "typing.Optional", "typing.Sequence", "typing.Set", "typing.Tuple", "typing.Type", "typing.Union", } TYPE_FACTORIES = { "collections.namedtuple", "typing.NewType", "typing.ParamSpec", "typing.TypeVar", } ENUM_BASE_NAMES = { "enum.Enum", "enum.Flag", "enum.IntEnum", "enum.IntFlag", "enum.StrEnum", } MUTATOR_METHODS = frozenset( "__delitem__ __setitem__ add append clear discard extend insert move_to_end " "pop popitem remove reverse rotate setdefault sort update".split() ) UNBOUND_MUTATOR_METHODS = {"__delattr__", "__delitem__", "__setattr__", "__setitem__"} OPERATOR_MUTATORS = frozenset( f"operator.{name}" for name in "delitem iadd iand iconcat imatmul imul ior isub ixor setitem".split() ) STATEFUL_DECORATORS = {"functools.cache", "functools.lru_cache"} # Calls through imported objects are denied unless explicitly reviewed for the # course. This is deliberately an allowlist: method names alone cannot # distinguish random.seed() from APIs which construct values, transform their # arguments, or provide the limited random operations used by the assignments. ALLOWED_IMPORTED_CALLS = frozenset( """ anysystem.Message anysystem.Message.from_json anysystem.python.anysystem.Message array.array ast.literal_eval bisect.bisect bisect.bisect_left bisect.bisect_right bisect.insort builtins.print collections.defaultdict collections.deque collections.namedtuple copy.copy copy.deepcopy dataclasses.asdict dataclasses.field dataclasses.fields dataclasses.is_dataclass functools.cmp_to_key functools.reduce hashlib.blake2b hashlib.md5 hashlib.sha1 hashlib.sha512 hashlib.sha256 hashlib.sha3_256 hashlib.shake_256 heapq.heapify heapq.heappop heapq.heappush heapq.merge importlib.import_module inspect.signature itertools.count json.dumps json.loads logging.critical logging.debug logging.error logging.exception logging.info logging.log logging.warning math.ceil math.log math.log2 math.sqrt os.getcwd os.getenv os.path.abspath os.path.basename os.path.dirname os.path.join os.path.normpath os.path.realpath pathlib.Path pickle.dumps pickle.loads portion.closedopen portion.empty pprint.pprint pydantic.Field random.Random random.choice random.choices random.randint random.random random.randrange random.sample random.shuffle random.uniform re.fullmatch sys.intern threading.Lock types.MappingProxyType time.time typing.cast uuid.uuid4 """.split() ) IMPORTED_ARGUMENT_MUTATORS = set( "bisect.insort heapq.heapify heapq.heappop heapq.heappush random.shuffle".split() ) ALLOWED_MUTABLE_METADATA = {"__all__"} TRUSTED_STAR_EXPORTS = { "dataclasses": {"dataclass", "field"}, "enum": {"Enum", "Flag", "IntEnum", "IntFlag", "StrEnum", "auto"}, "functools": {"cache", "lru_cache"}, "types": {"MappingProxyType"}, "typing": { *"Annotated Any Callable ClassVar Dict FrozenSet List Literal Mapping NewType " "Optional ParamSpec Sequence Set Tuple Type TypeVar Union".split(), }, } TRUSTED_BUILTINS = set( "bool bytearray bytes complex dict float frozenset int list range set str tuple type".split() ) def _resolved_attribute(node: ast.AST) -> tuple[ast.AST, str] | None: if isinstance(node, ast.Attribute): return node.value, node.attr if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "getattr" and len(node.args) >= 2 and isinstance(node.args[1], ast.Constant) and isinstance(node.args[1].value, str) ): return node.args[0], node.args[1].value return None def _call_name(node: ast.AST) -> str | None: if isinstance(node, ast.Name): return node.id resolved = _resolved_attribute(node) return resolved[1] if resolved is not None else None def _symbol(node: ast.AST, symbols: dict[str, str]) -> str | None: if isinstance(node, ast.Name): return symbols.get(node.id) resolved = _resolved_attribute(node) if resolved is not None: value, attribute = resolved parent = _symbol(value, symbols) return f"{parent}.{attribute}" if parent is not None else None return None def _root_name(node: ast.AST) -> str | None: while (resolved := _resolved_attribute(node)) is not None: node = resolved[0] return node.id if isinstance(node, ast.Name) else None def _assignment_parts(node: ast.AST) -> tuple[ast.AST, list[ast.AST]] | None: if isinstance(node, ast.Assign): return node.value, node.targets if isinstance(node, (ast.AnnAssign, ast.NamedExpr)) and node.value is not None: return node.value, [node.target] return None def _call_returns_shared( node: ast.AST, is_shared: Callable[[ast.AST], bool], shared_functions: set[str], returned_parameters: dict[str, set[int]], ) -> bool: if not isinstance(node, ast.Call): return False name = node.func.id if isinstance(node.func, ast.Name) else None if name in shared_functions: return True return any( index < len(node.args) and is_shared(node.args[index]) for index in returned_parameters.get(name or "", set()) ) def _is_instance_path(node: ast.AST, instance_names: set[str]) -> bool: while True: if isinstance(node, ast.Subscript): node = node.value elif (resolved := _resolved_attribute(node)) is not None: node = resolved[0] else: break return isinstance(node, ast.Name) and node.id in instance_names def _static_string(node: ast.AST, names: dict[str, str]) -> str | None: if isinstance(node, ast.Constant) and isinstance(node.value, str): return node.value if isinstance(node, ast.Name): return names.get(node.id) if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): left = _static_string(node.left, names) right = _static_string(node.right, names) return left + right if left is not None and right is not None else None if isinstance(node, ast.JoinedStr): parts: list[str] = [] for value in node.values: if isinstance(value, ast.Constant) and isinstance(value.value, str): parts.append(value.value) elif isinstance(value, ast.FormattedValue): part = _static_string(value.value, names) if part is None: return None parts.append(part) else: return None return "".join(parts) return None def _static_string_values(tree: ast.Module) -> dict[str, str]: assignments: list[tuple[str, ast.AST]] = [] for node in ast.walk(tree): if isinstance(node, ast.Assign): for target in node.targets: assignments.extend((name, node.value) for name in _target_names(target)) elif isinstance(node, (ast.AnnAssign, ast.NamedExpr)) and node.value is not None: assignments.extend((name, node.value) for name in _target_names(node.target)) values: dict[str, str] = {} ambiguous: set[str] = set() changed = True while changed: changed = False for name, value_node in assignments: value = _static_string(value_node, values) if value is None or name in ambiguous: continue if name in values and values[name] != value: values.pop(name) ambiguous.add(name) changed = True elif name not in values: values[name] = value changed = True return values def _dynamic_import_names(tree: ast.Module) -> tuple[set[str], set[str]]: modules = {"importlib"} functions: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name == "importlib": modules.add(alias.asname or "importlib") elif isinstance(node, ast.ImportFrom) and node.module == "importlib": for alias in node.names: if alias.name == "import_module": functions.add(alias.asname or alias.name) assignments = [node for node in ast.walk(tree) if isinstance(node, ast.Assign)] changed = True while changed: changed = False for assignment in assignments: names = set().union(*(_target_names(target) for target in assignment.targets)) if isinstance(assignment.value, ast.Name): if assignment.value.id in modules: before = len(modules) modules.update(names) changed |= len(modules) != before if assignment.value.id in functions: before = len(functions) functions.update(names) changed |= len(functions) != before elif ( isinstance(assignment.value, ast.Attribute) and assignment.value.attr == "import_module" and isinstance(assignment.value.value, ast.Name) and assignment.value.value.id in modules ): before = len(functions) functions.update(names) changed |= len(functions) != before return modules, functions def _is_dynamic_import_call( node: ast.Call, importlib_modules: set[str], import_module_functions: set[str] ) -> bool: return ( isinstance(node.func, ast.Name) and (node.func.id == "__import__" or node.func.id in import_module_functions) ) or ( (resolved := _resolved_attribute(node.func)) is not None and resolved[1] == "import_module" and isinstance(resolved[0], ast.Name) and resolved[0].id in importlib_modules ) def _class_has_mutation_escape(class_node: ast.ClassDef) -> bool: for statement in class_node.body: if not isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): continue positional = [*statement.args.posonlyargs, *statement.args.args] if not positional: continue instance_names = {positional[0].arg} for node in _function_nodes(statement): if isinstance(node, (ast.Assign, ast.AnnAssign, ast.AugAssign, ast.Delete)): targets = ( node.targets if isinstance(node, (ast.Assign, ast.Delete)) else [node.target] ) if any(_is_instance_path(target, instance_names) for target in targets): return True if not isinstance(node, ast.Call): continue if ( isinstance(node.func, ast.Name) and node.func.id in {"delattr", "setattr"} and node.args and _is_instance_path(node.args[0], instance_names) ): return True if ( (resolved := _resolved_attribute(node.func)) is not None and resolved[1] in UNBOUND_MUTATOR_METHODS and node.args and _is_instance_path(node.args[0], instance_names) ): return True if ( (resolved := _resolved_attribute(node.func)) is not None and resolved[1] in MUTATOR_METHODS and _is_instance_path(resolved[0], instance_names) ): return True return False def _target_names(node: ast.AST) -> set[str]: if isinstance(node, ast.Name): return {node.id} if isinstance(node, (ast.Tuple, ast.List)): result: set[str] = set() for element in node.elts: result.update(_target_names(element)) return result return set() def _dict_is_deeply_immutable( node: ast.Dict, safe_names: set[str], symbols: dict[str, str] ) -> bool: return all( key is not None and _is_immutable_expr(key, safe_names, symbols) and _is_immutable_expr(value, safe_names, symbols) for key, value in zip(node.keys, node.values) ) def _sequence_is_deeply_immutable( node: ast.List | ast.Set | ast.Tuple, safe_names: set[str], symbols: dict[str, str], ) -> bool: return all(_is_immutable_expr(element, safe_names, symbols) for element in node.elts) def _is_immutable_expr( node: ast.AST | None, safe_names: set[str], symbols: dict[str, str] ) -> bool: if node is None or isinstance(node, ast.Constant): return True if isinstance(node, ast.Name): return node.id in safe_names if isinstance(node, ast.Attribute): root = _root_name(node) if root is not None and root in symbols and symbols[root] != f"builtins.{root}": return node.attr.isupper() return _is_immutable_expr(node.value, safe_names, symbols) if isinstance(node, ast.Subscript): return _symbol(node.value, symbols) in TYPE_SUBSCRIPT_NAMES if isinstance(node, ast.Tuple): return _sequence_is_deeply_immutable(node, safe_names, symbols) if isinstance(node, ast.UnaryOp): return _is_immutable_expr(node.operand, safe_names, symbols) if isinstance(node, ast.BinOp): return _is_immutable_expr(node.left, safe_names, symbols) and _is_immutable_expr( node.right, safe_names, symbols ) if isinstance(node, ast.BoolOp): return all(_is_immutable_expr(value, safe_names, symbols) for value in node.values) if isinstance(node, ast.Compare): return _is_immutable_expr(node.left, safe_names, symbols) and all( _is_immutable_expr(value, safe_names, symbols) for value in node.comparators ) if isinstance(node, ast.IfExp): return _is_immutable_expr(node.body, safe_names, symbols) and _is_immutable_expr( node.orelse, safe_names, symbols ) if isinstance(node, ast.JoinedStr): return all( not isinstance(value, ast.FormattedValue) or _is_immutable_expr(value.value, safe_names, symbols) for value in node.values ) if isinstance(node, ast.Lambda): return True if isinstance(node, ast.Call): name = _symbol(node.func, symbols) if name in TYPE_FACTORIES: return True local_name = _call_name(node.func) if local_name is not None and f"frozen-dataclass:{local_name}" in safe_names: return all( _is_immutable_expr(arg, safe_names, symbols) for arg in node.args ) and all( _is_immutable_expr(keyword.value, safe_names, symbols) for keyword in node.keywords ) if name in IMMUTABLE_SCALAR_CONSTRUCTORS: return all(_is_immutable_expr(arg, safe_names, symbols) for arg in node.args) if name in KNOWN_IMMUTABLE_CALLS: return True if name in {"builtins.tuple", "builtins.frozenset"}: if not node.args: return True if len(node.args) != 1 or node.keywords: return False value = node.args[0] return isinstance(value, (ast.List, ast.Set, ast.Tuple)) and ( _sequence_is_deeply_immutable(value, safe_names, symbols) ) if name == "types.MappingProxyType": return ( len(node.args) == 1 and not node.keywords and isinstance(node.args[0], ast.Dict) and _dict_is_deeply_immutable(node.args[0], safe_names, symbols) ) return False def _is_mutable_expr( node: ast.AST | None, safe_names: set[str], symbols: dict[str, str] ) -> bool: if node is None: return False if isinstance(node, MUTABLE_NODES): return True if isinstance(node, ast.Call) and _symbol(node.func, symbols) in MUTABLE_CONSTRUCTORS: return True return not _is_immutable_expr(node, safe_names, symbols) def _scope_statements(body: Iterable[ast.stmt]) -> Iterable[ast.stmt]: for statement in body: yield statement nested: list[list[ast.stmt]] = [] if isinstance(statement, (ast.If, ast.For, ast.AsyncFor, ast.While)): nested.extend([statement.body, statement.orelse]) elif isinstance(statement, (ast.With, ast.AsyncWith)): nested.append(statement.body) elif isinstance(statement, (ast.Try, ast.TryStar)): nested.extend([statement.body, statement.orelse, statement.finalbody]) nested.extend(handler.body for handler in statement.handlers) elif isinstance(statement, ast.Match): nested.extend(case.body for case in statement.cases) for statements in nested: yield from _scope_statements(statements) def _function_nodes(function: ast.FunctionDef | ast.AsyncFunctionDef) -> list[ast.AST]: result: list[ast.AST] = [] stack: list[ast.AST] = list(reversed(function.body)) while stack: node = stack.pop() result.append(node) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)): continue stack.extend(reversed(list(ast.iter_child_nodes(node)))) return result def _shared_class_nodes( body: Iterable[ast.stmt], prefix: str = "" ) -> Iterable[tuple[str, ast.ClassDef]]: for statement in _scope_statements(body): if not isinstance(statement, ast.ClassDef): continue qualified_name = f"{prefix}.{statement.name}" if prefix else statement.name yield qualified_name, statement yield from _shared_class_nodes(statement.body, qualified_name) def _is_classvar_annotation(node: ast.AST, symbols: dict[str, str]) -> bool: value = node.value if isinstance(node, ast.Subscript) else node return _symbol(value, symbols) == "typing.ClassVar" class _FileChecker: def __init__( self, tree: ast.Module, path: Path, _target_classes: set[str], root: Path | None = None, ): self.tree = tree self.path = path self.root = (root or path.parent).resolve() self.violations: list[tuple[int, int, str]] = [] self.safe_names: set[str] = { "False", "None", "True", "bool", "bytes", "complex", "float", "frozenset", "int", "range", "str", "tuple", } self.symbols = {name: f"builtins.{name}" for name in TRUSTED_BUILTINS} self.import_names: set[str] = set() self.definition_names: set[str] = set() self.module_binding_names: set[str] = set() self.module_mutable_names: set[str] = set() self.classes: dict[str, ast.ClassDef] = {} self.shared_classes = list(_shared_class_nodes(tree.body)) self.functions: dict[str, ast.FunctionDef | ast.AsyncFunctionDef] = {} self.lambdas: dict[str, ast.Lambda] = {} self.shared_closure_factories: set[str] = set() self.shared_returning_functions: set[str] = set() self.returned_parameters: dict[str, set[int]] = {} self.stateful_decorator_factories: set[str] = set() self.shared_instance_attributes: dict[str, set[str]] = {} def add(self, node: ast.AST, message: str) -> None: self.violations.append( (getattr(node, "lineno", 1), getattr(node, "col_offset", 0) + 1, message) ) def collect_symbols(self) -> None: shadowed_names: set[str] = set() for statement in _scope_statements(self.tree.body): if isinstance(statement, ast.Import): for alias in statement.names: name = alias.asname or alias.name.split(".")[0] self.import_names.add(name) self.module_binding_names.add(name) if _candidate_module(self.root, alias.name.split(".")) is None: self.symbols[name] = alias.name if alias.asname else name else: self.symbols.pop(name, None) elif isinstance(statement, ast.ImportFrom): local_module = ( statement.level > 0 or statement.module is None or _candidate_module(self.root, statement.module.split(".")) is not None ) for alias in statement.names: if alias.name == "*": if statement.module is not None and not local_module: for export in TRUSTED_STAR_EXPORTS.get( statement.module, set() ): self.import_names.add(export) self.module_binding_names.add(export) self.symbols[export] = f"{statement.module}.{export}" if statement.module == "typing": self.safe_names.add(export) else: name = alias.asname or alias.name self.import_names.add(name) self.module_binding_names.add(name) if statement.module is not None and not local_module: self.symbols[name] = f"{statement.module}.{alias.name}" else: self.symbols.pop(name, None) if ( name.isupper() or statement.module in {"collections.abc", "typing"} or ( statement.module == "math" and alias.name in {"e", "inf", "nan", "pi", "tau"} ) ): self.safe_names.add(name) elif isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): self.definition_names.add(statement.name) self.module_binding_names.add(statement.name) shadowed_names.add(statement.name) self.functions[statement.name] = statement elif isinstance(statement, ast.ClassDef): self.definition_names.add(statement.name) self.module_binding_names.add(statement.name) shadowed_names.add(statement.name) self.classes[statement.name] = statement elif isinstance(statement, ast.Assign): for target in statement.targets: names = _target_names(target) self.module_binding_names.update(names) shadowed_names.update(names) if isinstance(statement.value, ast.Lambda): for name in names: self.lambdas[name] = statement.value elif isinstance(statement, (ast.AnnAssign, ast.AugAssign)): names = _target_names(statement.target) self.module_binding_names.update(names) shadowed_names.update(names) for name in shadowed_names: self.symbols.pop(name, None) self.safe_names.difference_update(shadowed_names) self.safe_names.update(self.definition_names) for _, class_node in self.shared_classes: frozen = any( isinstance(decorator, ast.Call) and _symbol(decorator.func, self.symbols) == "dataclasses.dataclass" and any( keyword.arg == "frozen" and isinstance(keyword.value, ast.Constant) and keyword.value.value is True for keyword in decorator.keywords ) for decorator in class_node.decorator_list ) immutable_defaults = all( not isinstance(statement, ast.AnnAssign) or statement.value is None or _is_immutable_expr(statement.value, self.safe_names, self.symbols) for statement in class_node.body ) if frozen and immutable_defaults and not _class_has_mutation_escape(class_node): self.safe_names.add(f"frozen-dataclass:{class_node.name}") def mark_shared_closure_factories(self) -> None: def consider(value: ast.AST) -> None: if not isinstance(value, ast.Call): return name = _call_name(value.func) function = self.functions.get(name or "") if function is not None and any( isinstance(node, ast.Nonlocal) for node in ast.walk(function) ): self.shared_closure_factories.add(function.name) scopes = [self.tree.body] scopes.extend(class_node.body for class_node in self.classes.values()) for body in scopes: for statement in _scope_statements(body): if isinstance(statement, ast.Assign): consider(statement.value) elif isinstance(statement, ast.AnnAssign) and statement.value is not None: consider(statement.value) for node in ast.walk(self.tree): if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): continue for decorator in node.decorator_list: name = _call_name(decorator.func if isinstance(decorator, ast.Call) else decorator) function = self.functions.get(name or "") if function is not None and any( isinstance(child, ast.Nonlocal) for child in ast.walk(function) ): self.shared_closure_factories.add(function.name) def collect_function_summaries(self) -> None: """Summarize simple wrappers without executing student code.""" callables: dict[ str, ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda ] = {**self.functions, **self.lambdas} changed = True while changed: changed = False for function_name, function in callables.items(): positional = [*function.args.posonlyargs, *function.args.args] local_names = { argument.arg for argument in [*positional, *function.args.kwonlyargs] } if function.args.vararg is not None: local_names.add(function.args.vararg.arg) if function.args.kwarg is not None: local_names.add(function.args.kwarg.arg) if isinstance(function, ast.Lambda): nodes = [function.body] return_values = [function.body] else: nodes = _function_nodes(function) return_values = [ node.value for node in nodes if isinstance(node, ast.Return) and node.value is not None ] assignments = [ node for node in nodes if _assignment_parts(node) is not None ] for assignment in assignments: _, targets = _assignment_parts(assignment) or (assignment, []) for target in targets: local_names.update(_target_names(target)) shared_aliases: set[str] = set() parameter_aliases = { argument.arg: {index} for index, argument in enumerate(positional) } def returned_parameter_indexes(node: ast.AST) -> set[int]: if isinstance(node, ast.Name): return parameter_aliases.get(node.id, set()) resolved = _resolved_attribute(node) if resolved is not None: return returned_parameter_indexes(resolved[0]) if isinstance(node, ast.Subscript): return returned_parameter_indexes(node.value) if isinstance(node, ast.Call): name = node.func.id if isinstance(node.func, ast.Name) else None return set().union( *( returned_parameter_indexes(node.args[index]) for index in self.returned_parameters.get( name or "", set() ) if index < len(node.args) ), set(), ) return set() def returns_shared(node: ast.AST) -> bool: if _is_immutable_expr(node, self.safe_names, self.symbols): return False if isinstance(node, ast.Name): return ( node.id in self.module_binding_names and node.id not in local_names ) or node.id in shared_aliases resolved = _resolved_attribute(node) if resolved is not None: return returns_shared(resolved[0]) if isinstance(node, ast.Subscript): return returns_shared(node.value) return _call_returns_shared( node, returns_shared, self.shared_returning_functions, self.returned_parameters, ) or ( isinstance(node, ast.Call) and _symbol(node.func, self.symbols) == "importlib.import_module" ) aliases_changed = True while aliases_changed: aliases_changed = False for assignment in assignments: value, targets = _assignment_parts(assignment) or (assignment, []) indexes = returned_parameter_indexes(value) shared = returns_shared(value) before = ( len(shared_aliases), sum(map(len, parameter_aliases.values())), ) for target in targets: for name in _target_names(target): if shared: shared_aliases.add(name) if indexes: parameter_aliases.setdefault(name, set()).update(indexes) aliases_changed |= before != ( len(shared_aliases), sum(map(len, parameter_aliases.values())), ) indexes = set().union( *(returned_parameter_indexes(value) for value in return_values), set() ) known_indexes = self.returned_parameters.setdefault(function_name, set()) before = len(known_indexes) known_indexes.update(indexes) changed |= len(known_indexes) != before if any(returns_shared(value) for value in return_values): if function_name not in self.shared_returning_functions: self.shared_returning_functions.add(function_name) changed = True for value in return_values: if isinstance(value, ast.Call): called = _symbol(value.func, self.symbols) local_called = ( value.func.id if isinstance(value.func, ast.Name) else None ) if ( called in STATEFUL_DECORATORS or local_called in self.stateful_decorator_factories ) and function_name not in self.stateful_decorator_factories: self.stateful_decorator_factories.add(function_name) changed = True def decorator_is_stateful( self, decorator: ast.AST, symbols: dict[str, str] ) -> bool: expression = decorator.func if isinstance(decorator, ast.Call) else decorator return _symbol(expression, symbols) in STATEFUL_DECORATORS or ( isinstance(expression, ast.Name) and expression.id in self.stateful_decorator_factories ) def check_module_bindings(self) -> None: for statement in _scope_statements(self.tree.body): value: ast.AST | None = None targets: set[str] = set() if isinstance(statement, ast.Assign): value = statement.value for target in statement.targets: targets.update(_target_names(target)) elif isinstance(statement, ast.AnnAssign): value = statement.value targets.update(_target_names(statement.target)) elif isinstance(statement, ast.AugAssign): value = statement.value targets.update(_target_names(statement.target)) else: continue if value is None: continue if _is_immutable_expr(value, self.safe_names, self.symbols): self.safe_names.update(targets) continue for name in sorted(targets - ALLOWED_MUTABLE_METADATA): self.module_mutable_names.add(name) self.add( statement, f"module-level value '{name}' is mutable or stateful and can be shared between processes", ) def check_class_bindings(self) -> None: for class_name, class_node in self.shared_classes: class_safe_names = set(self.safe_names) is_dataclass = any( _symbol( decorator.func if isinstance(decorator, ast.Call) else decorator, self.symbols, ) == "dataclasses.dataclass" for decorator in class_node.decorator_list ) is_pydantic_model = any( ( isinstance(base, ast.Name) and _symbol(base, self.symbols) in {"pydantic.BaseModel", "pydantic.main.BaseModel"} ) or ( isinstance(base, ast.Attribute) and _symbol(base, self.symbols) in {"pydantic.BaseModel", "pydantic.main.BaseModel"} ) for base in class_node.bases ) is_enum = any( ( isinstance(base, ast.Name) and _symbol(base, self.symbols) in ENUM_BASE_NAMES ) or ( isinstance(base, ast.Attribute) and _symbol(base, self.symbols) in ENUM_BASE_NAMES ) for base in class_node.bases ) class_symbols = dict(self.symbols) class_import_symbols: dict[str, str] = {} class_shadowed_names: set[str] = set() for class_statement in _scope_statements(class_node.body): names: set[str] = set() if isinstance( class_statement, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef) ): names.add(class_statement.name) elif isinstance(class_statement, ast.Assign): for target in class_statement.targets: names.update(_target_names(target)) elif isinstance(class_statement, (ast.AnnAssign, ast.AugAssign)): names.update(_target_names(class_statement.target)) elif isinstance(class_statement, ast.Import): for alias in class_statement.names: name = alias.asname or alias.name.split(".")[0] if _candidate_module(self.root, alias.name.split(".")) is None: class_import_symbols[name] = ( alias.name if alias.asname else name ) elif isinstance(class_statement, ast.ImportFrom): local_module = ( class_statement.level > 0 or class_statement.module is None or _candidate_module( self.root, class_statement.module.split(".") ) is not None ) if class_statement.module is not None and not local_module: for alias in class_statement.names: if alias.name == "*": for export in TRUSTED_STAR_EXPORTS.get( class_statement.module, set() ): class_import_symbols[export] = ( f"{class_statement.module}.{export}" ) else: name = alias.asname or alias.name class_import_symbols[name] = ( f"{class_statement.module}.{alias.name}" ) class_shadowed_names.update(names) class_symbols.update( (name, symbol) for name, symbol in class_import_symbols.items() if name not in class_shadowed_names ) for name in class_shadowed_names: class_symbols.pop(name, None) class_safe_names.discard(name) for class_statement in class_node.body: if not isinstance( class_statement, (ast.FunctionDef, ast.AsyncFunctionDef) ): continue for decorator in class_statement.decorator_list: expression = decorator.func if isinstance(decorator, ast.Call) else decorator name = _symbol(expression, class_symbols) if self.decorator_is_stateful(decorator, class_symbols): self.add( decorator, f"'{(name or _call_name(expression) or 'decorator').rsplit('.', 1)[-1]}' keeps a cache shared between process instances", ) for statement in _scope_statements(class_node.body): def targets_class_namespace(target: ast.AST) -> bool: if isinstance(target, ast.Subscript): return ( isinstance(target.value, ast.Call) and isinstance(target.value.func, ast.Name) and target.value.func.id in {"globals", "locals"} ) if isinstance(target, (ast.Tuple, ast.List)): return any(targets_class_namespace(item) for item in target.elts) return False if isinstance(statement, ast.Assign) and any( targets_class_namespace(target) for target in statement.targets ): self.add(statement, "assignment mutates class-level shared state") elif isinstance(statement, (ast.AnnAssign, ast.AugAssign)) and ( targets_class_namespace(statement.target) ): self.add(statement, "assignment mutates class-level shared state") elif isinstance(statement, ast.Delete) and any( targets_class_namespace(target) for target in statement.targets ): self.add(statement, "deletion mutates class-level shared state") elif isinstance(statement, ast.Expr) and isinstance( statement.value, ast.Call ): call = statement.value if ( isinstance(call.func, ast.Attribute) and call.func.attr in MUTATOR_METHODS | {"__setitem__"} and isinstance(call.func.value, ast.Call) and isinstance(call.func.value.func, ast.Name) and call.func.value.func.id in {"globals", "locals"} ): self.add(call, "call mutates class-level shared state") value: ast.AST | None = None targets: set[str] = set() if isinstance(statement, ast.Assign): value = statement.value for target in statement.targets: targets.update(_target_names(target)) elif isinstance(statement, ast.AnnAssign): value = statement.value targets.update(_target_names(statement.target)) else: continue if value is None: continue if _is_immutable_expr(value, class_safe_names, class_symbols): class_safe_names.update(targets) for name in targets: class_symbols.pop(name, None) continue if ( is_enum and isinstance(value, ast.Call) and _symbol(value.func, class_symbols) == "enum.auto" and not value.args and not value.keywords ): class_safe_names.update(targets) continue if ( is_dataclass and isinstance(statement, ast.AnnAssign) and not _is_classvar_annotation(statement.annotation, class_symbols) and isinstance(value, ast.Call) and _symbol(value.func, class_symbols) == "dataclasses.field" and not value.args and all( keyword.arg != "default" or not _is_mutable_expr( keyword.value, class_safe_names, class_symbols ) for keyword in value.keywords ) ): continue if ( is_pydantic_model and isinstance(statement, ast.AnnAssign) and not _is_classvar_annotation(statement.annotation, class_symbols) and isinstance(value, ast.Call) and _symbol(value.func, class_symbols) == "pydantic.Field" and not value.args and all( keyword.arg != "default" or not _is_mutable_expr( keyword.value, class_safe_names, class_symbols ) for keyword in value.keywords ) ): continue for name in sorted(targets): self.add( statement, f"class attribute '{class_name}.{name}' is mutable or stateful and is shared by all process instances", ) def check_module_mutations(self) -> None: statements = list(_scope_statements(self.tree.body)) class_aliases = set(self.classes) shared_aliases: set[str] = set() global_objects = self.module_binding_names def is_class_ref(node: ast.AST) -> bool: return isinstance(node, ast.Name) and node.id in class_aliases def is_shared_ref(node: ast.AST) -> bool: if isinstance(node, ast.Name): return node.id in global_objects or node.id in shared_aliases if isinstance(node, ast.Attribute): return ( is_class_ref(node.value) or is_shared_ref(node.value) or ( isinstance(node.value, ast.Name) and node.value.id in global_objects ) ) if isinstance(node, ast.Subscript): return is_shared_ref(node.value) if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "getattr" and node.args ): return is_class_ref(node.args[0]) or is_shared_ref(node.args[0]) if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "eval" and node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str) ): try: expression = ast.parse(node.args[0].value, mode="eval").body except SyntaxError: return False return is_shared_ref(expression) return ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "vars" and len(node.args) == 1 and (is_class_ref(node.args[0]) or is_shared_ref(node.args[0])) ) or ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in {"globals", "locals"} ) assignments = [ statement for statement in statements if isinstance(statement, (ast.Assign, ast.AnnAssign, ast.NamedExpr)) ] changed = True while changed: changed = False for assignment in assignments: value, targets = _assignment_parts(assignment) or (assignment, []) names = set().union(*map(_target_names, targets), set()) before = (len(class_aliases), len(shared_aliases)) if is_class_ref(value): class_aliases.update(names) if is_shared_ref(value): shared_aliases.update(names) if before != (len(class_aliases), len(shared_aliases)): changed = True def target_is_shared(target: ast.AST) -> bool: if isinstance(target, ast.Attribute): return ( is_class_ref(target.value) or is_shared_ref(target.value) or ( isinstance(target.value, ast.Name) and target.value.id in global_objects ) ) if isinstance(target, ast.Subscript): return is_shared_ref(target.value) if isinstance(target, (ast.Tuple, ast.List)): return any(target_is_shared(element) for element in target.elts) return False for statement in statements: if isinstance(statement, ast.Assign): if any(target_is_shared(target) for target in statement.targets): self.add(statement, "assignment creates or mutates module-level shared state") elif isinstance(statement, (ast.AnnAssign, ast.AugAssign)): if target_is_shared(statement.target): self.add(statement, "assignment creates or mutates module-level shared state") elif isinstance(statement, ast.Delete): if any(target_is_shared(target) for target in statement.targets): self.add(statement, "deletion mutates module-level shared state") elif isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Call): call = statement.value if isinstance(call.func, ast.Name) and call.func.id in { "exec", "globals", "locals", }: self.add(call, f"'{call.func.id}' can access or modify module-level shared state") elif ( isinstance(call.func, ast.Name) and call.func.id == "eval" and is_shared_ref(call) ): self.add(call, "'eval' accesses module-level shared state") elif ( isinstance(call.func, ast.Name) and call.func.id in {"setattr", "delattr"} and call.args and ( is_class_ref(call.args[0]) or is_shared_ref(call.args[0]) or ( isinstance(call.args[0], ast.Name) and call.args[0].id in global_objects ) ) ): self.add(call, f"'{call.func.id}' mutates module-level shared state") elif ( _symbol(call.func, self.symbols) in OPERATOR_MUTATORS and call.args and is_shared_ref(call.args[0]) ): self.add(call, "operator call mutates module-level shared state") elif ( (resolved := _resolved_attribute(call.func)) is not None and resolved[1] in UNBOUND_MUTATOR_METHODS and call.args and ( is_class_ref(call.args[0]) or is_shared_ref(call.args[0]) or ( isinstance(call.args[0], ast.Name) and call.args[0].id in global_objects ) ) ): self.add(call, f"'{resolved[1]}' mutates module-level shared state") elif ( (resolved := _resolved_attribute(call.func)) is not None and resolved[1] in MUTATOR_METHODS and is_shared_ref(resolved[0]) ): if not ( _symbol(resolved[0], self.symbols) == "sys.path" and resolved[1] in {"append", "insert"} ): self.add( call, f"'{resolved[1]}' mutates module-level shared state", ) elif ( (canonical := _symbol(call.func, self.symbols)) is not None and (_root_name(call.func) or "") in self.import_names and canonical not in ALLOWED_IMPORTED_CALLS and canonical not in STATEFUL_DECORATORS and canonical not in TYPE_FACTORIES and canonical not in MUTABLE_CONSTRUCTORS and canonical not in IMMUTABLE_SCALAR_CONSTRUCTORS ): self.add( call, f"call to imported API '{canonical}' is not known to be free of shared module-state mutation", ) def check_defaults(self) -> None: for node in ast.walk(self.tree): if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)): continue defaults = list(node.args.defaults) defaults.extend(default for default in node.args.kw_defaults if default is not None) for default in defaults: if _is_mutable_expr(default, self.safe_names, self.symbols): self.add(default, "mutable default argument is shared between function calls") if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): for decorator in node.decorator_list: expression = decorator.func if isinstance(decorator, ast.Call) else decorator name = _symbol(expression, self.symbols) if self.decorator_is_stateful(decorator, self.symbols): self.add( decorator, f"'{(name or _call_name(expression) or 'decorator').rsplit('.', 1)[-1]}' keeps a cache shared between process instances", ) def check_dynamic_imports(self) -> None: string_values = _static_string_values(self.tree) importlib_modules, import_module_functions = _dynamic_import_names(self.tree) for node in ast.walk(self.tree): if ( isinstance(node, ast.Call) and node.args and _is_dynamic_import_call( node, importlib_modules, import_module_functions ) and _static_string(node.args[0], string_values) is None ): self.add( node, "dynamic import name cannot be resolved statically; shared state in the imported module cannot be checked", ) def collect_shared_instance_attributes(self) -> None: for class_name, class_node in self.classes.items(): attributes = self.shared_instance_attributes.setdefault(class_name, set()) functions = [ statement for statement in class_node.body if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)) ] changed = True while changed: changed = False for function in functions: positional = [*function.args.posonlyargs, *function.args.args] if not positional: continue instance_names = {positional[0].arg} nodes = _function_nodes(function) assignments = [ node for node in nodes if isinstance(node, (ast.Assign, ast.AnnAssign, ast.NamedExpr)) ] local_names = { argument.arg for argument in [*positional, *function.args.kwonlyargs] } local_import_names: set[str] = set() for node in nodes: if isinstance(node, ast.Assign): for target in node.targets: local_names.update(_target_names(target)) elif isinstance(node, (ast.AnnAssign, ast.NamedExpr)): local_names.update(_target_names(node.target)) elif isinstance(node, (ast.Import, ast.ImportFrom)): names = { alias.asname or alias.name.split(".")[0] for alias in node.names if alias.name != "*" } local_names.update(names) local_import_names.update(names) shared_aliases: set[str] = set() def is_instance(node: ast.AST) -> bool: return isinstance(node, ast.Name) and node.id in instance_names def is_shared(node: ast.AST) -> bool: if isinstance(node, ast.Name): return ( node.id in self.module_binding_names and node.id not in local_names ) or node.id in shared_aliases | local_import_names if (resolved := _resolved_attribute(node)) is not None: return ( is_instance(resolved[0]) and resolved[1] in attributes ) or is_shared(resolved[0]) if isinstance(node, ast.Subscript): return is_shared(node.value) return _call_returns_shared( node, is_shared, self.shared_returning_functions, self.returned_parameters, ) or ( isinstance(node, ast.Call) and _symbol(node.func, self.symbols) == "importlib.import_module" ) aliases_changed = True while aliases_changed: aliases_changed = False for assignment in assignments: value, targets = _assignment_parts(assignment) or (assignment, []) if is_instance(value): names = set().union(*(_target_names(t) for t in targets)) before = len(instance_names) instance_names.update(names) aliases_changed |= len(instance_names) != before if not is_shared(value): continue for target in targets: names = _target_names(target) before = len(shared_aliases) shared_aliases.update(names) aliases_changed |= len(shared_aliases) != before if ( isinstance(target, ast.Attribute) and is_instance(target.value) and target.attr not in attributes ): attributes.add(target.attr) changed = True def check_function( self, function: ast.FunctionDef | ast.AsyncFunctionDef, owner: str | None, shared_closure: bool, ) -> None: nodes = _function_nodes(function) all_class_names = set(self.classes) instance_aliases: set[str] = set() class_aliases: set[str] = set() shared_aliases: set[str] = set() shared_instance_attributes = self.shared_instance_attributes.get(owner or "", set()) positional = [*function.args.posonlyargs, *function.args.args] local_names = { argument.arg for argument in [*positional, *function.args.kwonlyargs] } if function.args.vararg is not None: local_names.add(function.args.vararg.arg) if function.args.kwarg is not None: local_names.add(function.args.kwarg.arg) local_import_names: set[str] = set() local_import_symbols: dict[str, str] = {} for node in nodes: if isinstance(node, ast.Assign): for target in node.targets: local_names.update(_target_names(target)) elif isinstance(node, (ast.AnnAssign, ast.AugAssign, ast.NamedExpr)): local_names.update(_target_names(node.target)) elif isinstance(node, ast.Import): names = { alias.asname or alias.name.split(".")[0] for alias in node.names } local_names.update(names) local_import_names.update(names) for alias in node.names: name = alias.asname or alias.name.split(".")[0] if _candidate_module(self.root, alias.name.split(".")) is None: local_import_symbols[name] = alias.name if alias.asname else name elif isinstance(node, ast.ImportFrom): names = { alias.asname or alias.name for alias in node.names if alias.name != "*" } local_names.update(names) local_import_names.update(names) local_module = ( node.level > 0 or node.module is None or _candidate_module(self.root, node.module.split(".")) is not None ) if node.module is not None and not local_module: for alias in node.names: if alias.name != "*": local_import_symbols[alias.asname or alias.name] = ( f"{node.module}.{alias.name}" ) function_symbols = dict(self.symbols) for name in local_names - local_import_names: function_symbols.pop(name, None) function_symbols.update(local_import_symbols) imported_callable_names = (self.import_names - local_names) | local_import_names if owner is not None and positional: first = positional[0].arg if first == "cls" or any( isinstance(decorator, ast.Name) and decorator.id == "classmethod" for decorator in function.decorator_list ): class_aliases.add(first) else: instance_aliases.add(first) def is_instance_ref(node: ast.AST) -> bool: return isinstance(node, ast.Name) and node.id in instance_aliases def is_class_ref(node: ast.AST) -> bool: if isinstance(node, ast.Name): return ( node.id in all_class_names and node.id not in local_names ) or node.id in class_aliases if isinstance(node, ast.Attribute): return node.attr == "__class__" and is_instance_ref(node.value) return ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "type" and len(node.args) == 1 and is_instance_ref(node.args[0]) ) global_objects = self.module_binding_names def is_shared_ref(node: ast.AST) -> bool: if isinstance(node, ast.Name): return ( node.id in global_objects and node.id not in local_names ) or node.id in shared_aliases | local_import_names if (resolved := _resolved_attribute(node)) is not None: value, attribute = resolved if is_instance_ref(value) and attribute in shared_instance_attributes: return True if attribute == "__dict__" and ( is_class_ref(value) or (isinstance(value, ast.Name) and value.id in global_objects) ): return True return is_class_ref(value) or is_shared_ref(value) or ( isinstance(value, ast.Name) and value.id in global_objects ) if isinstance(node, ast.Subscript): return is_shared_ref(node.value) if _call_returns_shared( node, is_shared_ref, self.shared_returning_functions, self.returned_parameters, ): return True if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "getattr" and node.args ): return ( is_class_ref(node.args[0]) or is_shared_ref(node.args[0]) or ( isinstance(node.args[0], ast.Name) and node.args[0].id in global_objects ) ) if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "eval" and node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str) ): try: expression = ast.parse(node.args[0].value, mode="eval").body except SyntaxError: return False return is_shared_ref(expression) return ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "vars" and len(node.args) == 1 and ( is_class_ref(node.args[0]) or is_shared_ref(node.args[0]) or ( isinstance(node.args[0], ast.Name) and node.args[0].id in global_objects ) ) ) or ( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "globals" ) assignments = [ node for node in nodes if isinstance(node, (ast.Assign, ast.AnnAssign, ast.NamedExpr)) ] unbound_mutator_aliases: set[str] = set() bound_mutator_aliases: set[str] = set() imported_callable_aliases: dict[str, str] = {} changed = True while changed: changed = False for assignment in assignments: value, targets = _assignment_parts(assignment) or (assignment, []) names: set[str] = set() for target in targets: names.update(_target_names(target)) before = ( len(instance_aliases), len(class_aliases), len(shared_aliases), len(unbound_mutator_aliases), len(bound_mutator_aliases), len(imported_callable_aliases), ) if is_instance_ref(value): instance_aliases.update(names) if is_class_ref(value): class_aliases.update(names) if is_shared_ref(value): shared_aliases.update(names) symbol = _symbol(value, function_symbols) aliased_symbol = ( imported_callable_aliases.get(value.id) if isinstance(value, ast.Name) else None ) if aliased_symbol is not None: symbol = aliased_symbol if ( symbol is not None and ( aliased_symbol is not None or (_root_name(value) or "") in imported_callable_names ) ): for name in names: imported_callable_aliases[name] = symbol if symbol in OPERATOR_MUTATORS or ( isinstance(value, ast.Name) and value.id in unbound_mutator_aliases ): unbound_mutator_aliases.update(names) if ( (resolved := _resolved_attribute(value)) is not None and resolved[1] in MUTATOR_METHODS and is_shared_ref(resolved[0]) ) or ( isinstance(value, ast.Name) and value.id in bound_mutator_aliases ): bound_mutator_aliases.update(names) if before != ( len(instance_aliases), len(class_aliases), len(shared_aliases), len(unbound_mutator_aliases), len(bound_mutator_aliases), len(imported_callable_aliases), ): changed = True def target_is_shared(target: ast.AST) -> bool: if isinstance(target, ast.Attribute): return ( is_class_ref(target.value) or is_shared_ref(target.value) or ( isinstance(target.value, ast.Name) and target.value.id in global_objects ) ) if isinstance(target, ast.Subscript): return is_shared_ref(target.value) if isinstance(target, (ast.Tuple, ast.List)): return any(target_is_shared(element) for element in target.elts) return False for node in nodes: if isinstance(node, ast.Global): self.add(node, "'global' state can be shared between process instances") elif isinstance(node, ast.Nonlocal) and shared_closure: self.add(node, "shared closure state can be accessed by multiple process instances") elif isinstance(node, ast.Assign): if any(target_is_shared(target) for target in node.targets): self.add(node, "assignment mutates state outside the process instance") elif isinstance(node, (ast.AnnAssign, ast.AugAssign)): if target_is_shared(node.target): self.add(node, "assignment mutates state outside the process instance") elif isinstance(node, ast.Delete): if any(target_is_shared(target) for target in node.targets): self.add(node, "deletion mutates state outside the process instance") elif isinstance(node, ast.Call): if isinstance(node.func, ast.Name) and node.func.id in {"exec", "globals"}: self.add(node, f"'{node.func.id}' can access or modify module-level shared state") elif ( isinstance(node.func, ast.Name) and node.func.id == "eval" and is_shared_ref(node) ): self.add(node, "'eval' accesses module-level shared state") elif ( isinstance(node.func, ast.Name) and node.func.id in {"setattr", "delattr"} and node.args and ( is_class_ref(node.args[0]) or is_shared_ref(node.args[0]) or ( isinstance(node.args[0], ast.Name) and node.args[0].id in global_objects ) ) ): self.add(node, f"'{node.func.id}' mutates shared object state") elif ( ( _symbol(node.func, function_symbols) in OPERATOR_MUTATORS or ( isinstance(node.func, ast.Name) and node.func.id in unbound_mutator_aliases ) ) and node.args and is_shared_ref(node.args[0]) ): self.add(node, "operator call mutates shared object state") elif ( isinstance(node.func, ast.Name) and node.func.id in bound_mutator_aliases ): self.add(node, "aliased method mutates shared object state") elif ( ( imported_callable_aliases.get( node.func.id, _symbol(node.func, function_symbols) ) if isinstance(node.func, ast.Name) else _symbol(node.func, function_symbols) ) in IMPORTED_ARGUMENT_MUTATORS and node.args and is_shared_ref(node.args[0]) ): self.add(node, "imported function mutates shared object state") elif ( (resolved := _resolved_attribute(node.func)) is not None and resolved[1] in UNBOUND_MUTATOR_METHODS and node.args and ( is_class_ref(node.args[0]) or is_shared_ref(node.args[0]) or ( isinstance(node.args[0], ast.Name) and node.args[0].id in global_objects ) ) ): self.add(node, f"'{resolved[1]}' mutates shared object state") elif ( (resolved := _resolved_attribute(node.func)) is not None and resolved[1] in MUTATOR_METHODS and is_shared_ref(resolved[0]) ): self.add(node, f"'{resolved[1]}' mutates state outside the process instance") elif ( ( canonical := ( imported_callable_aliases.get( node.func.id, _symbol(node.func, function_symbols) ) if isinstance(node.func, ast.Name) else _symbol(node.func, function_symbols) ) ) is not None and ( (_root_name(node.func) or "") in imported_callable_names or ( isinstance(node.func, ast.Name) and node.func.id in imported_callable_aliases ) ) and canonical not in ALLOWED_IMPORTED_CALLS and canonical not in STATEFUL_DECORATORS and canonical not in TYPE_FACTORIES and canonical not in MUTABLE_CONSTRUCTORS and canonical not in IMMUTABLE_SCALAR_CONSTRUCTORS ): self.add( node, f"call to imported API '{canonical}' is not known to be free of shared module-state mutation", ) def check_functions(self) -> None: def visit_body( body: Iterable[ast.stmt], owner: str | None = None, shared_closure: bool = False, ) -> None: for statement in body: if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): function_closure = shared_closure or ( owner is None and statement.name in self.shared_closure_factories ) self.check_function(statement, owner, function_closure) visit_body(statement.body, None, function_closure) elif isinstance(statement, ast.ClassDef): visit_body(statement.body, statement.name, shared_closure) else: for child in ast.iter_child_nodes(statement): if isinstance(child, ast.stmt): visit_body([child], owner, shared_closure) visit_body(self.tree.body) def run(self) -> list[tuple[int, int, str]]: self.collect_symbols() self.mark_shared_closure_factories() self.check_module_bindings() self.collect_function_summaries() self.check_class_bindings() self.check_module_mutations() self.check_defaults() self.check_dynamic_imports() self.collect_shared_instance_attributes() self.check_functions() return sorted(set(self.violations)) def _candidate_module(root: Path, parts: list[str]) -> Path | None: if not parts: return None module = root.joinpath(*parts) candidates = [module.with_suffix(".py"), module / "__init__.py"] root = root.resolve() for candidate in candidates: try: resolved = candidate.resolve() resolved.relative_to(root) except (OSError, ValueError): continue if resolved.is_file(): return resolved return None def _local_imports(tree: ast.Module, current: Path, root: Path) -> set[Path]: result: set[Path] = set() importlib_modules, import_module_functions = _dynamic_import_names(tree) string_values = _static_string_values(tree) for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: candidate = _candidate_module(root, alias.name.split(".")) if candidate is not None: result.add(candidate) elif isinstance(node, ast.ImportFrom): if node.level: base = current.parent for _ in range(node.level - 1): base = base.parent else: base = root module_parts = node.module.split(".") if node.module else [] candidate = _candidate_module(base, module_parts) if candidate is not None: result.add(candidate) for alias in node.names: if alias.name == "*": continue candidate = _candidate_module(base, [*module_parts, alias.name]) if candidate is not None: result.add(candidate) elif isinstance(node, ast.Call) and node.args: module_name = _static_string(node.args[0], string_values) if ( _is_dynamic_import_call( node, importlib_modules, import_module_functions ) and module_name is not None and not module_name.startswith(".") ): candidate = _candidate_module(root, module_name.split(".")) if candidate is not None: result.add(candidate) return result def validate_source(source: str, filename: str, class_names: list[str]) -> list[str]: tree = ast.parse(source, filename=filename) path = Path(filename) return [ f"{filename}:{line}:{column}: {message}" for line, column, message in _FileChecker( tree, path, set(class_names), path.parent ).run() ] def validate_solution(path: str, class_names: list[str]) -> list[str]: entrypoint = Path(path).resolve() root = entrypoint.parent pending = [entrypoint] visited: set[Path] = set() violations: list[str] = [] while pending: current = pending.pop() if current in visited: continue visited.add(current) source = current.read_text(encoding="utf-8") tree = ast.parse(source, filename=str(current)) checker = _FileChecker( tree, current, set(class_names) if current == entrypoint else set(), root, ) violations.extend( f"{current}:{line}:{column}: {message}" for line, column, message in checker.run() ) pending.extend(sorted(_local_imports(tree, current, root) - visited)) return sorted(set(violations))