diff --git a/src/fastapi_endpoint_detector/parser/secure_ast_extractor.py b/src/fastapi_endpoint_detector/parser/secure_ast_extractor.py index 7fd8608..5e1b177 100644 --- a/src/fastapi_endpoint_detector/parser/secure_ast_extractor.py +++ b/src/fastapi_endpoint_detector/parser/secure_ast_extractor.py @@ -36,6 +36,8 @@ class SecureASTExtractorError(Exception): CompositionMode = Literal["copy", "live"] EndpointImpact = Literal["none", "prior", "subsequent", "all"] _ORDER_SCALE = 1_000_000 +_EAGER_DEFINITION_MAX_WORK = 2_048 +_EAGER_DEFINITION_MAX_DEPTH = 32 _HTTP_ROUTE_METADATA_KEYWORDS = frozenset( { @@ -156,6 +158,90 @@ def _registration_call_shape( return _SHARED_REGISTRATION_CALL_SHAPES[operation] +def _route_decorator_metadata_is_statically_safe( # noqa: PLR0911, PLR0912 + arguments: dict[str, ast.expr], +) -> bool: + """Accept only exact metadata shapes harmless during decorator application.""" + optional_string_keywords = { + "description", + "name", + "operation_id", + "summary", + } + boolean_keywords = { + "include_in_schema", + "response_model_by_alias", + "response_model_exclude_defaults", + "response_model_exclude_none", + "response_model_exclude_unset", + } + optional_mapping_keywords = { + "openapi_extra", + "response_model_exclude", + "response_model_include", + "responses", + } + + for keyword, value in arguments.items(): + if keyword == "path": + if not isinstance(value, ast.Constant) or type(value.value) is not str: + return False + continue + if keyword == "methods": + if not isinstance(value, (ast.List, ast.Set, ast.Tuple)) or any( + not isinstance(item, ast.Constant) or type(item.value) is not str + for item in value.elts + ): + return False + continue + is_none = isinstance(value, ast.Constant) and value.value is None + if keyword in {"callbacks", "dependencies"}: + if not (is_none or (isinstance(value, (ast.List, ast.Tuple)) and not value.elts)): + return False + continue + if keyword in optional_string_keywords: + if not (is_none or (isinstance(value, ast.Constant) and type(value.value) is str)): + return False + continue + if keyword == "response_description": + if not (isinstance(value, ast.Constant) and type(value.value) is str): + return False + continue + if keyword == "deprecated": + if not (is_none or (isinstance(value, ast.Constant) and type(value.value) is bool)): + return False + continue + if keyword in boolean_keywords: + if not (isinstance(value, ast.Constant) and type(value.value) is bool): + return False + continue + if keyword == "status_code": + if not (is_none or (isinstance(value, ast.Constant) and type(value.value) is int)): + return False + continue + if keyword == "tags": + if is_none: + continue + if not isinstance(value, (ast.List, ast.Tuple)) or any( + not isinstance(item, ast.Constant) or type(item.value) is not str + for item in value.elts + ): + return False + continue + if keyword in optional_mapping_keywords: + if not ( + is_none or (isinstance(value, ast.Dict) and not value.keys and not value.values) + ): + return False + continue + if keyword == "response_model" and is_none: + continue + # Class/model/callable metadata is intentionally opaque, including + # response_class and generate_unique_id_function. + return False + return True + + @dataclass(frozen=True) class _Route: owner: ObjectKey @@ -218,6 +304,322 @@ class _DirectEffectResult: status: Literal["modeled", "limited", "unrecognized"] +@dataclass(frozen=True) +class _EagerDefinitionRisk: + evidence: ast.AST + reason: str + receiver_only: bool = False + + +class _EagerDefinitionRiskAnalyzer: + """Bound eager definition analysis shared by factory and bootstrap slices.""" + + def __init__( + self, + *, + resolves_tracked_object: Callable[[ast.expr], bool], + is_exact_route_decorator: Callable[[ast.expr], bool], + evaluate_annotations: bool, + budget: list[int], + reason_prefix: str, + ) -> None: + self._resolves_tracked_object = resolves_tracked_object + self._is_exact_route_decorator = is_exact_route_decorator + self._evaluate_annotations = evaluate_annotations + self._budget = budget + self._reason_prefix = reason_prefix + + def _reason(self, detail: str) -> str: + return f"{self._reason_prefix} {detail}" + + @staticmethod + def _type_parameter_names(node: ast.AST) -> frozenset[str]: + return frozenset( + name + for parameter in getattr(node, "type_params", ()) + if isinstance((name := getattr(parameter, "name", None)), str) + ) + + def _spend(self, evidence: ast.AST) -> _EagerDefinitionRisk | None: + self._budget[0] -= 1 + if self._budget[0] < 0: + return _EagerDefinitionRisk( + evidence, + self._reason("eager-definition analysis budget is exhausted"), + ) + return None + + def expression_risk( # noqa: PLR0911 - explicit conservative allowlist + self, + expression: ast.AST, + *, + exempt_registration: bool = False, + shadowed: frozenset[str] = frozenset(), + ) -> _EagerDefinitionRisk | None: + """Accept only eager expressions that cannot dispatch user protocols.""" + pending: list[tuple[ast.AST, bool, frozenset[str]]] = [ + (expression, exempt_registration, shadowed) + ] + while pending: + current, exempt, current_shadowed = pending.pop() + if exhausted := self._spend(expression): + return exhausted + if isinstance(current, ast.Constant): + continue + if isinstance(current, ast.Name): + if current.id not in current_shadowed and self._resolves_tracked_object(current): + return _EagerDefinitionRisk( + current, + self._reason("eager definition may rebind or escape a route object"), + ) + continue + if isinstance(current, ast.NamedExpr): + return _EagerDefinitionRisk( + current, + self._reason("eager named expression invalidates route-object bindings"), + ) + if isinstance(current, ast.Call): + if not exempt: + return _EagerDefinitionRisk( + current, + self._reason("eager definition invokes an unresolved call"), + ) + pending.extend( + (argument, False, current_shadowed) for argument in reversed(current.args) + ) + pending.extend( + (keyword.value, False, current_shadowed) + for keyword in reversed(current.keywords) + ) + continue + if isinstance(current, (ast.Tuple, ast.List)): + if any(isinstance(item, ast.Starred) for item in current.elts): + return _EagerDefinitionRisk( + current, + self._reason("eager expression may dispatch a user protocol"), + ) + pending.extend((item, False, current_shadowed) for item in reversed(current.elts)) + continue + if isinstance(current, ast.Lambda): + pending.extend( + (item, False, current_shadowed) + for item in reversed(current.args.kw_defaults) + if item is not None + ) + pending.extend( + (item, False, current_shadowed) for item in reversed(current.args.defaults) + ) + continue + return _EagerDefinitionRisk( + current, + self._reason("eager expression may dispatch a user protocol"), + ) + return None + + def function_risk( + self, + definition: ast.FunctionDef | ast.AsyncFunctionDef, + *, + allow_exact_decorators: bool, + shadowed: frozenset[str] = frozenset(), + ) -> _EagerDefinitionRisk | None: + receiver_only_risk: _EagerDefinitionRisk | None = None + for decorator in definition.decorator_list: + if not self._is_exact_route_decorator(decorator): + return _EagerDefinitionRisk( + decorator, + self._reason("function decorator application is unresolved"), + ) + if risk := self.expression_risk( + decorator, + exempt_registration=True, + shadowed=shadowed, + ): + return risk + if not allow_exact_decorators: + if receiver_only_risk is not None: + return _EagerDefinitionRisk( + decorator, + self._reason("multiple function decorator applications are unresolved"), + ) + receiver_only_risk = _EagerDefinitionRisk( + decorator, + self._reason("function decorator application is unresolved"), + receiver_only=True, + ) + for default in [ + *definition.args.defaults, + *(item for item in definition.args.kw_defaults if item is not None), + ]: + if risk := self.expression_risk(default, shadowed=shadowed): + return risk + if self._evaluate_annotations: + annotation_shadowed = shadowed | self._type_parameter_names(definition) + for annotation in [ + *(item.annotation for item in definition.args.args), + *(item.annotation for item in definition.args.posonlyargs), + *(item.annotation for item in definition.args.kwonlyargs), + definition.args.vararg.annotation if definition.args.vararg is not None else None, + definition.args.kwarg.annotation if definition.args.kwarg is not None else None, + definition.returns, + ]: + if annotation is not None and ( + risk := self.expression_risk( + annotation, + shadowed=annotation_shadowed, + ) + ): + return risk + return receiver_only_risk + + @staticmethod + def _class_external_names(definition: ast.ClassDef) -> set[str]: + names: set[str] = set() + pending: list[ast.AST] = list(reversed(definition.body)) + while pending: + current = pending.pop() + if isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + continue + if isinstance(current, (ast.Global, ast.Nonlocal)): + names.update(current.names) + continue + pending.extend(reversed(list(ast.iter_child_nodes(current)))) + return names + + @staticmethod + def _descriptor_safe(expression: ast.expr) -> bool: + if isinstance(expression, ast.Constant): + return True + if isinstance(expression, (ast.Tuple, ast.List)): + return all( + not isinstance(item, ast.Starred) + and _EagerDefinitionRiskAnalyzer._descriptor_safe(item) + for item in expression.elts + ) + return False + + def class_risk( # noqa: PLR0911, PLR0912 - explicit class execution cases + self, + definition: ast.ClassDef, + *, + depth: int = 0, + ) -> _EagerDefinitionRisk | None: + if depth >= _EAGER_DEFINITION_MAX_DEPTH: + return _EagerDefinitionRisk( + definition, + self._reason("eager-definition recursion budget is exhausted"), + ) + if definition.decorator_list: + return _EagerDefinitionRisk( + definition.decorator_list[0], + self._reason("class decorator application is unresolved"), + ) + if definition.bases or definition.keywords: + evidence: ast.AST = definition.bases[0] if definition.bases else definition.keywords[0] + return _EagerDefinitionRisk( + evidence, + self._reason("class base or metaclass execution is unresolved"), + ) + + class_bound: set[str] = set(self._type_parameter_names(definition)) + class_external = self._class_external_names(definition) + control_flow = ( + ast.If, + ast.For, + ast.AsyncFor, + ast.While, + ast.With, + ast.AsyncWith, + ast.Try, + ast.Match, + ) + for statement in definition.body: + if exhausted := self._spend(statement): + return exhausted + direct_bound = _scope_bound_names( + statement, + evaluate_annotations=self._evaluate_annotations, + ) + runtime_bound = direct_bound + if isinstance(statement, ast.AnnAssign) and statement.value is None: + runtime_bound = direct_bound - _bound_names(statement.target) + if runtime_bound & class_external: + return _EagerDefinitionRisk( + statement, + self._reason("class body may rebind outer route state"), + ) + if isinstance(statement, (ast.Global, ast.Nonlocal, ast.Pass)): + continue + if isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Constant): + continue + if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): + if risk := self.function_risk( + statement, + allow_exact_decorators=False, + shadowed=frozenset(class_bound), + ): + return risk + class_bound.update(runtime_bound - class_external) + continue + if isinstance(statement, ast.ClassDef): + if risk := self.class_risk(statement, depth=depth + 1): + return risk + class_bound.update(runtime_bound - class_external) + continue + if isinstance(statement, control_flow) or _is_try_star(statement): + evidence = _nested_eager_definition(statement) or statement + return _EagerDefinitionRisk( + evidence, + self._reason("class control flow is unresolved"), + ) + if isinstance(statement, ast.Import | ast.ImportFrom): + return _EagerDefinitionRisk( + statement, + self._reason("class import execution is unresolved"), + ) + if _is_type_alias(statement): + class_bound.update(runtime_bound - class_external) + continue + if isinstance(statement, (ast.Assign, ast.AnnAssign)): + targets = ( + [statement.target] + if isinstance(statement, ast.AnnAssign) + else statement.targets + ) + if any(not isinstance(target, ast.Name) for target in targets): + return _EagerDefinitionRisk( + statement, + self._reason("class assignment target is unresolved"), + ) + value = statement.value + shadowed = frozenset(class_bound) + if value is not None and (risk := self.expression_risk(value, shadowed=shadowed)): + return risk + if value is not None and not self._descriptor_safe(value): + return _EagerDefinitionRisk( + value, + self._reason("class descriptor initialization is unresolved"), + ) + if ( + isinstance(statement, ast.AnnAssign) + and self._evaluate_annotations + and ( + risk := self.expression_risk( + statement.annotation, + shadowed=shadowed, + ) + ) + ): + return risk + class_bound.update(runtime_bound - class_external) + continue + return _EagerDefinitionRisk( + statement, + self._reason("class body execution is unresolved"), + ) + return None + + @dataclass class _Module: name: str @@ -739,7 +1141,7 @@ def _load_modules( with tokenize.open(path) as source_file: source = source_file.read() tree = ast.parse(source, filename=str(path)) - except (OSError, RecursionError, SyntaxError, UnicodeError) as error: + except (MemoryError, OSError, RecursionError, SyntaxError, UnicodeError) as error: parse_failures.append( EndpointDiscoveryCondition( source_path=path.resolve(), @@ -1221,6 +1623,7 @@ def _summarize_factory( # noqa: PLR0911, PLR0912, PLR0915 - safe subset isinstance(function, ast.AsyncFunctionDef) or function.decorator_list or function_identity in stack + or len(stack) >= _EAGER_DEFINITION_MAX_DEPTH ): return None resolver = argument_resolver or ( @@ -1276,6 +1679,8 @@ def _summarize_factory( # noqa: PLR0911, PLR0912, PLR0915 - safe subset for history in candidate_module.objects.values() for item in history } + eager_work = [_EAGER_DEFINITION_MAX_WORK] + evaluate_annotations = not _uses_future_annotations(module.tree) def resolve_literal_name(name: str) -> str | None: if name in local_strings: @@ -1500,6 +1905,31 @@ def inspect(current: ast.AST) -> bool: # noqa: PLR0911 return inspect(node) + def exact_route_decorator(decorator: ast.expr) -> bool: + if not ( + isinstance(decorator, ast.Call) + and isinstance(decorator.func, ast.Attribute) + and decorator.func.attr in {*self.HTTP_METHODS, "api_route"} + ): + return False + receiver = decorator.func.value + owner = object_for(receiver) + if owner is None: + return False + arguments = _validated_call_arguments( + decorator, + *_decorator_call_shape(decorator.func.attr, owner, receiver), + ) + return arguments is not None and _route_decorator_metadata_is_statically_safe(arguments) + + eager_analyzer = _EagerDefinitionRiskAnalyzer( + resolves_tracked_object=lambda expression: object_for(expression) is not None, + is_exact_route_decorator=exact_route_decorator, + evaluate_annotations=evaluate_annotations, + budget=eager_work, + reason_prefix="factory", + ) + for statement in meaningful: if isinstance(statement, ast.Return): break @@ -1637,7 +2067,10 @@ def resolve_nested_argument( local_strings[assigned] = literal(value, statement.lineno) continue if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): - local_functions[statement.name] = statement + if risk := eager_analyzer.function_risk(statement, allow_exact_decorators=True): + if not allow_conditional: + return None + conditionalize(risk.evidence, risk.reason) handler = self._handler(module, statement) for decorator in statement.decorator_list: if not ( @@ -1679,6 +2112,13 @@ def resolve_nested_argument( "factory decorated route arguments are ambiguous", ) continue + if not _route_decorator_metadata_is_statically_safe(arguments): + limit_registration( + owner, + decorator, + "factory decorated route metadata is not statically safe", + ) + continue path = literal(arguments.get("path"), statement.lineno) if path is None: limit_registration( @@ -1712,6 +2152,20 @@ def resolve_nested_argument( call_order, ) ) + local_objects.pop(statement.name, None) + local_router_views.discard(statement.name) + local_strings.pop(statement.name, None) + local_functions[statement.name] = statement + continue + if isinstance(statement, ast.ClassDef): + if risk := eager_analyzer.class_risk(statement): + if not allow_conditional: + return None + conditionalize(risk.evidence, risk.reason) + local_functions.pop(statement.name, None) + local_objects.pop(statement.name, None) + local_router_views.discard(statement.name) + local_strings.pop(statement.name, None) continue if ( isinstance(statement, ast.Expr) @@ -1732,6 +2186,16 @@ def resolve_nested_argument( and isinstance(statement.value, ast.Call) and isinstance(statement.value.func, ast.Attribute) ): + # Definitions below control flow execute eager headers or class + # bodies conditionally and are outside this straight-line slice. + if nested_definition := _nested_eager_definition(statement): + if not allow_conditional: + return None + conditionalize( + nested_definition, + "factory eager definition under control flow is unresolved", + ) + continue # Ignore unrelated setup, but reject control-flow or helpers that # can rebind/mutate an object whose public routes we are proving. if touches_modeled_binding(statement): @@ -1998,6 +2462,7 @@ def _apply_bootstrap_registration( # noqa: PLR0915 ) -> None: """Interpret one explicitly attested, bounded registration call slice.""" budget = [max(32, min(512, len(modules) * 8))] + eager_work = [_EAGER_DEFINITION_MAX_WORK] operation_order = [_line_end_order(2**31 - 2)] def next_order() -> int: @@ -2032,6 +2497,7 @@ def apply( # noqa: PLR0912, PLR0915 if ( budget[0] <= 0 or identity in stack + or len(stack) >= _EAGER_DEFINITION_MAX_DEPTH or isinstance(current, ast.AsyncFunctionDef) or current.decorator_list or current.args.vararg is not None @@ -2053,7 +2519,7 @@ def apply( # noqa: PLR0912, PLR0915 local_modules: dict[str, _Module] = {} local_functions: dict[str, tuple[_Module, ast.FunctionDef | ast.AsyncFunctionDef]] = {} local_handlers: dict[str, ast.FunctionDef | ast.AsyncFunctionDef] = {} - global_names: set[str] = set() + global_names = _scope_global_names(current.body) bound_names = { parameter.arg for parameter in [ @@ -2150,11 +2616,114 @@ def touches_tracked(node: ast.AST) -> bool: isinstance(item, ast.Name) and item.id in tracked for item in ast.walk(node) ) + evaluate_annotations = not _uses_future_annotations(current_module.tree) + + def exact_route_decorator(decorator: ast.expr) -> bool: + if not ( + isinstance(decorator, ast.Call) + and isinstance(decorator.func, ast.Attribute) + and decorator.func.attr in {*self.HTTP_METHODS, "api_route"} + ): + return False + receiver = decorator.func.value + owner = object_for(receiver, decorator.lineno) + if owner is None: + return False + arguments = _validated_call_arguments( + decorator, + *_decorator_call_shape(decorator.func.attr, owner, receiver), + ) + return arguments is not None and _route_decorator_metadata_is_statically_safe( + arguments + ) + + eager_analyzer = _EagerDefinitionRiskAnalyzer( + resolves_tracked_object=lambda expression: ( + object_for( + expression, + getattr(expression, "lineno", current.lineno), + ) + is not None + ), + is_exact_route_decorator=exact_route_decorator, + evaluate_annotations=evaluate_annotations, + budget=eager_work, + reason_prefix="bootstrap", + ) + + def limit_ordered( + owners: set[_Object], + evidence: ast.AST, + reason: str, + ) -> None: + order = next_order() + condition = EndpointDiscoveryCondition( + source_path=current_module.path, + source_line=getattr(evidence, "lineno", current.lineno), + reason=reason, + ) + for owner in sorted(owners, key=lambda item: item.key): + limitation = _OrderedInventoryLimitation( + origin_module=current_module.name, + order=order, + condition=condition, + endpoint_impact="all", + ) + limitations = current_module.ordered_inventory_limitations.setdefault( + owner.key, [] + ) + if limitation not in limitations: + limitations.append(limitation) + + def limit_eager_risk(risk: _EagerDefinitionRisk) -> None: + owners = set(local_objects.values()) | {root} + if risk.receiver_only and isinstance(risk.evidence, ast.Call): + assert isinstance(risk.evidence.func, ast.Attribute) + resolved_owner = object_for( + risk.evidence.func.value, + risk.evidence.lineno, + ) + if resolved_owner is not None: + owners = {resolved_owner} + limit_ordered(owners, risk.evidence, risk.reason) + + def displace_global_binding( + name: str, + evidence: ast.AST, + replacement: _Object | None = None, + *, + replacement_is_router_view: bool = False, + ) -> None: + displaced = local_objects.get(name) + if displaced is None and name in global_names and name not in bound_names: + displaced = object_for( + ast.Name(id=name, ctx=ast.Load()), + getattr(evidence, "lineno", current.lineno), + ) + same_binding = displaced is replacement and ( + (name in local_router_views) == replacement_is_router_view + ) + if name in global_names and displaced is not None and not same_binding: + limit_ordered( + {displaced}, + evidence, + "bootstrap global binding displaces a route object", + ) + + def clear_binding(name: str) -> None: + local_objects.pop(name, None) + local_router_views.discard(name) + local_strings.pop(name, None) + local_modules.pop(name, None) + local_handlers.pop(name, None) + local_functions.pop(name, None) + for statement in current.body: if isinstance(statement, (ast.Global, ast.Nonlocal)): - global_names.update(statement.names) continue if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): + if risk := eager_analyzer.function_risk(statement, allow_exact_decorators=True): + limit_eager_risk(risk) decorator_handler = self._handler(current_module, statement) for decorator in statement.decorator_list: if not ( @@ -2180,9 +2749,13 @@ def touches_tracked(node: ast.AST) -> bool: operation, decorator_owner, decorator.func.value ), ) + metadata_is_safe = ( + arguments is not None + and _route_decorator_metadata_is_statically_safe(arguments) + ) path = ( None - if arguments is None + if arguments is None or not metadata_is_safe else literal(arguments.get("path"), decorator.lineno) ) methods_expr = _keyword_expr(decorator, "methods") @@ -2205,11 +2778,16 @@ def touches_tracked(node: ast.AST) -> bool: ) ) if arguments is None or path is None or not decorator_methods: + reason = ( + "bootstrap decorated route metadata is not statically safe" + if arguments is not None and not metadata_is_safe + else "bootstrap decorated route is unresolved" + ) limit( current_module, decorator_owner, decorator, - "bootstrap decorated route is unresolved", + reason, inventory_only=True, ) continue @@ -2222,14 +2800,19 @@ def touches_tracked(node: ast.AST) -> bool: next_order(), ) ) - local_objects.pop(statement.name, None) - local_router_views.discard(statement.name) - local_strings.pop(statement.name, None) - local_modules.pop(statement.name, None) + displace_global_binding(statement.name, statement) + clear_binding(statement.name) bound_names.add(statement.name) local_handlers[statement.name] = statement local_functions[statement.name] = (current_module, statement) continue + if isinstance(statement, ast.ClassDef): + if risk := eager_analyzer.class_risk(statement): + limit_eager_risk(risk) + displace_global_binding(statement.name, statement) + clear_binding(statement.name) + bound_names.add(statement.name) + continue if isinstance(statement, ast.ImportFrom): target_name = self._absolute_import(current_module, statement) for imported_alias in statement.names: @@ -2238,52 +2821,63 @@ def touches_tracked(node: ast.AST) -> bool: limit(current_module, owner, statement, "wildcard bootstrap import") continue local = imported_alias.asname or imported_alias.name - bound_names.add(local) - local_router_views.discard(local) + imported_object: _Object | None = None + imported_submodule: _Module | None = None + imported_target: ( + tuple[_Module, ast.FunctionDef | ast.AsyncFunctionDef] | None + ) = None submodule_name = aliases.get(f"{target_name}.{imported_alias.name}") if submodule_name in modules: - local_modules[local] = modules[submodule_name] - continue - imported_module = modules.get(aliases.get(target_name, target_name)) - if imported_module is None: - continue - exported = self._resolve_exported_object( - imported_module, - imported_alias.name, - 2**31 - 1, - aliases, - modules, - frozenset(), - min(len(modules) + 1, 64), - ) - if exported is not None: - local_objects[local] = exported - continue - imported_function = self._function_at( - imported_module, imported_alias.name, 2**31 - 1 - ) - if imported_function is not None: - local_functions[local] = ( - imported_module, - imported_function, - ) + imported_submodule = modules[submodule_name] + else: + imported_module = modules.get(aliases.get(target_name, target_name)) + if imported_module is not None: + imported_object = self._resolve_exported_object( + imported_module, + imported_alias.name, + 2**31 - 1, + aliases, + modules, + frozenset(), + min(len(modules) + 1, 64), + ) + imported_function = self._function_at( + imported_module, + imported_alias.name, + 2**31 - 1, + ) + if imported_function is not None: + imported_target = imported_module, imported_function + displace_global_binding(local, statement, imported_object) + clear_binding(local) + bound_names.add(local) + if imported_object is not None: + local_objects[local] = imported_object + elif imported_submodule is not None: + local_modules[local] = imported_submodule + elif imported_target is not None: + local_functions[local] = imported_target continue if isinstance(statement, ast.Import): for imported_alias in statement.names: local = imported_alias.asname or imported_alias.name.split(".")[0] - bound_names.add(local) - local_router_views.discard(local) target_name = aliases.get(imported_alias.name, imported_alias.name) imported_module = modules.get(target_name) + displace_global_binding(local, statement) + clear_binding(local) + bound_names.add(local) if imported_module is not None: local_modules[local] = imported_module - local_functions.pop(local, None) - local_objects.pop(local, None) continue if isinstance(statement, (ast.Assign, ast.AnnAssign)): name = _assignment_name(statement) value = statement.value - if name is None or value is None: + if name is None: + rebound_globals = _scope_bound_names(statement) & global_names + for rebound in sorted(rebound_globals): + displace_global_binding(rebound, statement) + clear_binding(rebound) + bound_names.add(rebound) if touches_tracked(statement): for owner in set(local_objects.values()): limit( @@ -2293,14 +2887,24 @@ def touches_tracked(node: ast.AST) -> bool: "unsupported bootstrap assignment", ) continue - bound_names.add(name) - local_functions.pop(name, None) - local_modules.pop(name, None) - local_handlers.pop(name, None) + if value is None: + clear_binding(name) + bound_names.add(name) + continue aliased = object_for(value, statement.lineno) aliases_router_view = aliased is not None and denotes_router_view( value, statement.lineno ) + displace_global_binding( + name, + statement, + aliased, + replacement_is_router_view=aliases_router_view, + ) + bound_names.add(name) + local_functions.pop(name, None) + local_modules.pop(name, None) + local_handlers.pop(name, None) if aliased is not None: if name in global_names: limit( @@ -2349,6 +2953,19 @@ def touches_tracked(node: ast.AST) -> bool: ) break if not (isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Call)): + rebound_globals = _scope_bound_names(statement) & global_names + for rebound in sorted(rebound_globals): + displace_global_binding(rebound, statement) + clear_binding(rebound) + bound_names.add(rebound) + if nested_definition := _nested_eager_definition(statement): + limit_eager_risk( + _EagerDefinitionRisk( + nested_definition, + "bootstrap eager definition under control flow is unresolved", + ) + ) + continue if touches_tracked(statement): for owner in set(local_objects.values()): limit( @@ -2358,6 +2975,11 @@ def touches_tracked(node: ast.AST) -> bool: "unsupported bootstrap control flow", ) continue + rebound_globals = _scope_bound_names(statement) & global_names + for rebound in sorted(rebound_globals): + displace_global_binding(rebound, statement) + clear_binding(rebound) + bound_names.add(rebound) call = statement.value line = statement.lineno registration_receiver = ( @@ -3750,6 +4372,21 @@ def visit_BinOp(self, node: ast.BinOp) -> None: self.visit(operand) +def _nested_eager_definition( + node: ast.AST, +) -> ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef | None: + """Find a definition executed below control flow without entering deferred bodies.""" + pending = list(reversed(list(ast.iter_child_nodes(node)))) + while pending: + current = pending.pop() + if isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + return current + if isinstance(current, ast.Lambda) or _is_type_alias(current): + continue + pending.extend(reversed(list(ast.iter_child_nodes(current)))) + return None + + def _function_bindings( function: ast.FunctionDef | ast.AsyncFunctionDef, ) -> set[str]: @@ -4073,6 +4710,21 @@ def visit_TypeAlias(self, _child: ast.AST) -> None: return visitor.found +def _scope_global_names(statements: list[ast.stmt]) -> set[str]: + """Return lexical global declarations without entering nested scopes.""" + names: set[str] = set() + pending: list[ast.AST] = list(reversed(statements)) + while pending: + current = pending.pop() + if isinstance(current, ast.Global): + names.update(current.names) + continue + if isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)): + continue + pending.extend(reversed(list(ast.iter_child_nodes(current)))) + return names + + def _scope_bound_names( node: ast.AST, *, diff --git a/tests/unit/test_secure_ast_extractor.py b/tests/unit/test_secure_ast_extractor.py index 70b04f4..d685db7 100644 --- a/tests/unit/test_secure_ast_extractor.py +++ b/tests/unit/test_secure_ast_extractor.py @@ -3439,9 +3439,9 @@ def test_factory_and_bootstrap_local_literals_use_bounded_evaluation(tmp_path: P "from fastapi import FastAPI\n" "def create():\n" " app = FastAPI()\n" - f" path = {bomb}\n" - " @app.get(path)\n" " def route(): pass\n" + f" path = {bomb}\n" + " app.add_api_route(path, route)\n" " return app\n", encoding="utf-8", ) @@ -3449,9 +3449,7 @@ def test_factory_and_bootstrap_local_literals_use_bounded_evaluation(tmp_path: P assert factory_inventory.endpoints == [] assert factory_inventory.status == InventoryStatus.CONDITIONAL - assert any( - "factory decorated route path" in item.reason for item in factory_inventory.limitations - ) + assert any("factory imperative route" in item.reason for item in factory_inventory.limitations) (tmp_path / "bootstrap.py").write_text( "from fastapi import FastAPI\n" @@ -3583,10 +3581,19 @@ def test_malformed_native_registration_call_shapes_fail_closed( inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + if context == "implicit-factory" and operation == "decorator": + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.UNAVAILABLE + return assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] - assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.ESTABLISHED + expected_status = ( + EndpointDiscoveryStatus.CONDITIONAL + if context == "explicit-factory" and operation == "decorator" + else EndpointDiscoveryStatus.ESTABLISHED + ) + assert inventory.endpoints[0].discovery_status == expected_status assert inventory.status == InventoryStatus.CONDITIONAL - assert len(inventory.limitations) == 1 + assert inventory.limitations assert "could not be modeled" not in inventory.limitations[0].reason @@ -3799,10 +3806,18 @@ def test_factory_app_router_view_repeated_router_fails_closed( inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + if context == "implicit-factory" and "@" in invalid_registration: + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.UNAVAILABLE + return assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /positive"] - assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.ESTABLISHED + assert inventory.endpoints[0].discovery_status == ( + EndpointDiscoveryStatus.CONDITIONAL + if "@" in invalid_registration + else EndpointDiscoveryStatus.ESTABLISHED + ) assert inventory.status == InventoryStatus.CONDITIONAL - assert len(inventory.limitations) == 1 + assert inventory.limitations @pytest.mark.parametrize("context", ["implicit-factory", "explicit-factory"]) @@ -3859,9 +3874,13 @@ def test_bootstrap_app_router_view_repeated_router_fails_closed( inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /positive"] - assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.ESTABLISHED + assert inventory.endpoints[0].discovery_status == ( + EndpointDiscoveryStatus.CONDITIONAL + if "@" in invalid_registration + else EndpointDiscoveryStatus.ESTABLISHED + ) assert inventory.status == InventoryStatus.CONDITIONAL - assert len(inventory.limitations) == 1 + assert inventory.limitations @pytest.mark.parametrize("registration", ["imperative", "decorator"]) @@ -3986,10 +4005,22 @@ def test_invalid_native_registration_surfaces_are_additive( inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + eager_decorator = case in { + "unknown-decorator-keyword", + "router-router-decorator", + } + if context == "implicit-factory" and eager_decorator: + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.UNAVAILABLE + return assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] - assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.ESTABLISHED + assert inventory.endpoints[0].discovery_status == ( + EndpointDiscoveryStatus.CONDITIONAL + if context == "explicit-factory" and eager_decorator + else EndpointDiscoveryStatus.ESTABLISHED + ) assert inventory.status == InventoryStatus.CONDITIONAL - assert len(inventory.limitations) == 1 + assert inventory.limitations @pytest.mark.parametrize( @@ -4077,10 +4108,18 @@ def test_fastapi_api_route_callbacks_are_an_additive_limitation( inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + if context == "implicit-factory": + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.UNAVAILABLE + return assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] - assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.ESTABLISHED + assert inventory.endpoints[0].discovery_status == ( + EndpointDiscoveryStatus.CONDITIONAL + if context == "explicit-factory" + else EndpointDiscoveryStatus.ESTABLISHED + ) assert inventory.status == InventoryStatus.CONDITIONAL - assert len(inventory.limitations) == 1 + assert inventory.limitations @pytest.mark.parametrize( @@ -4394,3 +4433,1277 @@ def fail_parse(*_args: object, **_kwargs: object) -> object: assert inventory.status == InventoryStatus.UNAVAILABLE assert inventory.endpoints == [] assert "could not be read, decoded, or parsed" in inventory.limitations[0].reason + + +def test_secure_module_real_parser_complexity_error_is_fail_closed(tmp_path: Path) -> None: + nesting = 200 + expression = "(lambda value=" * nesting + "1" + ": value)" * nesting + source = tmp_path / "main.py" + source.write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + f" def nested(value={expression}): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(source, bootstrap_entry="main:run").extract_inventory() + + assert inventory.status == InventoryStatus.UNAVAILABLE + assert inventory.endpoints == [] + assert "could not be read, decoded, or parsed" in inventory.limitations[0].reason + + +@pytest.mark.parametrize( + "context", + ["implicit-factory", "explicit-factory", "bootstrap"], +) +def test_route_decorator_malformed_dependencies_fail_closed(tmp_path: Path, context: str) -> None: + registrations = ( + "@app.get('/before')\n" + "def before(): pass\n" + "@app.get('/bad', dependencies=['oops'])\n" + "def bad(): pass\n" + "@app.get('/after')\n" + "def after(): pass\n" + ) + kwargs: dict[str, str] = {} + if context == "bootstrap": + source = "from fastapi import FastAPI\napp = FastAPI()\ndef run():\n" + "".join( + f" {line}\n" for line in registrations.splitlines() + ) + kwargs["bootstrap_entry"] = "main:run" + else: + source = ( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + + "".join(f" {line}\n" for line in registrations.splitlines()) + + " return app\n" + + "app = create()\n" + ) + if context == "explicit-factory": + kwargs["app_entry"] = "main:create" + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + if context == "implicit-factory": + assert inventory.status == InventoryStatus.UNAVAILABLE + assert inventory.endpoints == [] + return + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "GET /after", + "GET /before", + ] + assert all( + endpoint.discovery_status == EndpointDiscoveryStatus.CONDITIONAL + for endpoint in inventory.endpoints + ) + assert inventory.status == InventoryStatus.CONDITIONAL + assert any("metadata is not statically safe" in item.reason for item in inventory.limitations) + assert any( + "decorator application is unresolved" in item.reason for item in inventory.limitations + ) + + +@pytest.mark.parametrize( + "metadata", + [ + "name=metadata", + "tags=metadata", + "include_in_schema=1", + "status_code='200'", + "responses={'200': {}}", + "response_model=Model", + "response_class=Model", + "generate_unique_id_function=metadata", + ], +) +def test_factory_route_decorator_unsafe_adjacent_metadata_is_conditional( + tmp_path: Path, metadata: str +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "metadata = object()\n" + "class Model: pass\n" + "def create():\n" + " app = FastAPI()\n" + f" @app.get('/bad', {metadata})\n" + " def bad(): pass\n" + " @app.get('/safe')\n" + " def safe(): pass\n" + " return app\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, app_entry="main:create").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert inventory.status == InventoryStatus.CONDITIONAL + + +def test_bootstrap_imported_iterable_metadata_does_not_execute_and_conditions_prior_route( + tmp_path: Path, +) -> None: + imported_marker = tmp_path / "imported" + iterated_marker = tmp_path / "iterated" + (tmp_path / "metadata.py").write_text( + "from pathlib import Path\n" + f"Path({str(imported_marker)!r}).write_text('imported')\n" + "class ClearingDependencies:\n" + " def __iter__(self):\n" + f" Path({str(iterated_marker)!r}).write_text('iterated')\n" + " from main import app\n" + " app.routes.clear()\n" + " return iter(())\n" + "dependencies = ClearingDependencies()\n", + encoding="utf-8", + ) + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n" + "from metadata import dependencies\n" + "app = FastAPI()\n" + "router = APIRouter()\n" + "@app.get('/stale')\n" + "def stale(): pass\n" + "def run():\n" + " @router.get('/route', dependencies=dependencies)\n" + " def route(): pass\n" + " app.include_router(router)\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert not imported_marker.exists() + assert not iterated_marker.exists() + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /stale"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert inventory.status == InventoryStatus.CONDITIONAL + assert any( + "decorator application is unresolved" in item.reason for item in inventory.limitations + ) + + +@pytest.mark.parametrize( + "context", + ["implicit-factory", "explicit-factory", "bootstrap"], +) +def test_route_decorator_nonliteral_path_fails_closed_without_execution( + tmp_path: Path, context: str +) -> None: + imported_marker = tmp_path / "imported" + accessed_marker = tmp_path / "accessed" + (tmp_path / "payload.py").write_text( + "from pathlib import Path\n" + f"Path({str(imported_marker)!r}).write_text('imported')\n" + "class Trigger:\n" + " def __getattribute__(self, name):\n" + f" Path({str(accessed_marker)!r}).write_text('accessed')\n" + " from main import app\n" + " app.routes.clear()\n" + " return lambda *args: True\n" + "trigger = Trigger()\n", + encoding="utf-8", + ) + registration = "@app.get('/stale')\ndef stale(): pass\n@router.get(trigger)\ndef bad(): pass\n" + kwargs: dict[str, str] = {} + if context == "bootstrap": + source = ( + "from fastapi import APIRouter, FastAPI\n" + "from payload import trigger\n" + "app = FastAPI()\n" + "router = APIRouter()\n" + "@app.get('/stale')\n" + "def stale(): pass\n" + "def run():\n" + " @router.get(trigger)\n" + " def bad(): pass\n" + ) + kwargs["bootstrap_entry"] = "main:run" + else: + source = ( + "from fastapi import APIRouter, FastAPI\n" + "from payload import trigger\n" + "def create():\n" + " app = FastAPI()\n" + " router = APIRouter()\n" + + "".join(f" {line}\n" for line in registration.splitlines()) + + " return app\n" + + "app = create()\n" + ) + if context == "explicit-factory": + kwargs["app_entry"] = "main:create" + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + assert not imported_marker.exists() + assert not accessed_marker.exists() + if context == "implicit-factory": + assert inventory.status == InventoryStatus.UNAVAILABLE + assert inventory.endpoints == [] + return + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /stale"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert inventory.status == InventoryStatus.CONDITIONAL + assert any( + "decorator application is unresolved" in item.reason for item in inventory.limitations + ) + + +@pytest.mark.parametrize("explicit", [False, True], ids=["implicit", "explicit"]) +@pytest.mark.parametrize( + "definition", + [ + "def nested(value=app.routes.clear()): pass", + "def nested(*, value=app.routes.clear()): pass", + "def nested(value=(app := None)): pass", + "@decorate(app)\n def nested(): pass", + "def nested(value: annotate(app)): pass", + "async def nested(value=app.routes.clear()): pass", + ], + ids=["default", "kw-default", "rebind", "decorator", "annotation", "async-default"], +) +def test_factory_nested_definition_eager_effects_fail_closed( + tmp_path: Path, explicit: bool, definition: str +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + " @app.get('/before')\n" + " def before(): pass\n" + f" {definition}\n" + " @app.get('/after')\n" + " def after(): pass\n" + " return app\n" + "app = create()\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor( + tmp_path, app_entry="main:create" if explicit else None + ).extract_inventory() + + if not explicit: + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.UNAVAILABLE + return + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "GET /after", + "GET /before", + ] + assert all( + endpoint.discovery_status == EndpointDiscoveryStatus.CONDITIONAL + for endpoint in inventory.endpoints + ) + assert inventory.status == InventoryStatus.CONDITIONAL + eager_lines = {6, 7} if definition.startswith("@") else {6} + assert any( + item.source_path.name == "main.py" and item.source_line in eager_lines + for item in inventory.limitations + ) + + +def test_factory_definition_binding_shadows_returned_route_object(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + " @app.get('/stale')\n" + " def app(): pass\n" + " return app\n" + "app = create()\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path).extract_inventory() + + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.UNAVAILABLE + + +@pytest.mark.parametrize( + "definition", + [ + "def nested(value=app.routes.clear()): pass", + "def nested(*, value=app.routes.clear()): pass", + "def nested(value=(app := None)): pass", + "@decorate(app)\n def nested(): pass", + "def nested(value: annotate(app)): pass", + "async def nested(value=app.routes.clear()): pass", + ], + ids=["default", "kw-default", "rebind", "decorator", "annotation", "async-default"], +) +def test_bootstrap_nested_definition_eager_effects_condition_known_routes( + tmp_path: Path, definition: str +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " @app.get('/before')\n" + " def before(): pass\n" + f" {definition}\n" + " @app.get('/after')\n" + " def after(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "GET /after", + "GET /before", + ] + assert all( + endpoint.discovery_status == EndpointDiscoveryStatus.CONDITIONAL + for endpoint in inventory.endpoints + ) + assert inventory.status == InventoryStatus.CONDITIONAL + eager_lines = {6, 7} if definition.startswith("@") else {6} + assert any( + item.source_path.name == "main.py" and item.source_line in eager_lines + for item in inventory.limitations + ) + + +def test_bootstrap_exact_nested_routes_remain_established(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " @app.get('/sync')\n" + " def sync_handler(value='safe'): pass\n" + " @app.websocket('/async')\n" + " async def async_handler(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "GET /sync", + "WEBSOCKET /async", + ] + assert inventory.status == InventoryStatus.ESTABLISHED + assert inventory.limitations == () + + +def test_postponed_annotations_and_deferred_bodies_remain_unvisited(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from __future__ import annotations\n" + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " def nested(value: app.routes.clear()):\n" + " app.routes.clear()\n" + " class Plain:\n" + " marker: app.routes.clear()\n" + " def method(self):\n" + " app.routes.clear()\n" + " @app.get('/safe')\n" + " async def safe(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] + assert inventory.status == InventoryStatus.ESTABLISHED + assert inventory.limitations == () + + +@pytest.mark.parametrize( + ("class_definition", "evidence_line", "reason_category"), + [ + ("@decorate(app)\n class Routes: pass", 6, "class decorator"), + ("class Routes(base(app)): pass", 6, "class base"), + ("class Routes(metaclass=meta(app)): pass", 6, "class base"), + ("class Routes:\n app.routes.clear()", 7, "class body execution"), + ( + "class Routes:\n class Nested:\n app.routes.clear()", + 8, + "class body execution", + ), + ("class Routes:\n marker = descriptor", 7, "descriptor"), + ("class Routes:\n marker: annotate(app)", 7, "unresolved call"), + ], + ids=[ + "decorator", + "base", + "metaclass", + "body", + "nested-body", + "descriptor", + "active-annotation", + ], +) +def test_bootstrap_nested_class_eager_effects_are_source_backed( + tmp_path: Path, + class_definition: str, + evidence_line: int, + reason_category: str, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " @app.get('/before')\n" + " def before(): pass\n" + f" {class_definition}\n" + " @app.get('/after')\n" + " def after(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert inventory.status == InventoryStatus.CONDITIONAL + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "GET /after", + "GET /before", + ] + assert all( + endpoint.discovery_status == EndpointDiscoveryStatus.CONDITIONAL + for endpoint in inventory.endpoints + ) + assert any( + item.source_path.name == "main.py" + and item.source_line == evidence_line + and reason_category in item.reason + for item in inventory.limitations + ) + + +@pytest.mark.parametrize("explicit", [False, True], ids=["implicit", "explicit"]) +@pytest.mark.parametrize( + ("class_definition", "evidence_line", "reason_category"), + [ + ("@decorate(app)\n class Routes: pass", 6, "class decorator"), + ("class Routes(base(app)): pass", 6, "class base"), + ("class Routes(metaclass=meta(app)): pass", 6, "class base"), + ("class Routes:\n app.routes.clear()", 7, "class body execution"), + ( + "class Routes:\n class Nested:\n app.routes.clear()", + 8, + "class body execution", + ), + ("class Routes:\n marker = descriptor", 7, "descriptor"), + ("class Routes:\n marker: annotate(app)", 7, "unresolved call"), + ], + ids=[ + "decorator", + "base", + "metaclass", + "body", + "nested-body", + "descriptor", + "active-annotation", + ], +) +def test_factory_nested_class_eager_effects_fail_closed( + tmp_path: Path, + explicit: bool, + class_definition: str, + evidence_line: int, + reason_category: str, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + " @app.get('/before')\n" + " def before(): pass\n" + f" {class_definition}\n" + " @app.get('/after')\n" + " def after(): pass\n" + " return app\n" + "app = create()\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor( + tmp_path, app_entry="main:create" if explicit else None + ).extract_inventory() + + if not explicit: + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.UNAVAILABLE + return + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "GET /after", + "GET /before", + ] + assert all( + endpoint.discovery_status == EndpointDiscoveryStatus.CONDITIONAL + for endpoint in inventory.endpoints + ) + assert inventory.status == InventoryStatus.CONDITIONAL + assert any( + item.source_path.name == "main.py" + and item.source_line == evidence_line + and reason_category in item.reason + for item in inventory.limitations + ) + + +def test_bootstrap_class_body_registration_then_include_is_not_established_empty( + tmp_path: Path, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "router = APIRouter()\n" + "def run():\n" + " class Routes:\n" + " @router.get('/class-body')\n" + " def route(): pass\n" + " app.include_router(router)\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.CONDITIONAL + assert any( + item.source_path.name == "main.py" and item.source_line == 6 and "decorator" in item.reason + for item in inventory.limitations + ) + + +def test_class_local_shadowing_and_deferred_method_do_not_taint_outer_app( + tmp_path: Path, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " class Local:\n" + " app = None\n" + " def method(self):\n" + " app.routes.clear()\n" + " @app.get('/safe')\n" + " def safe(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] + assert inventory.status == InventoryStatus.ESTABLISHED + assert inventory.limitations == () + + +@pytest.mark.skipif(sys.version_info < (3, 12), reason="PEP 695 requires Python 3.12") +def test_type_parameter_bounds_are_lazy_and_shadow_outer_app(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " def nested[app: app.routes.clear()](value: app): pass\n" + " class Local[app: app.routes.clear()]:\n" + " marker: app\n" + " @app.get('/safe')\n" + " def safe(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] + assert inventory.status == InventoryStatus.ESTABLISHED + assert inventory.limitations == () + + +@pytest.mark.parametrize("context", ["factory", "bootstrap"]) +def test_eager_definition_work_budget_fails_closed_without_recursion_error( + tmp_path: Path, context: str +) -> None: + expression = "[" + ", ".join(repr("x") for _ in range(2_200)) + "]" + if context == "factory": + source = ( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + f" def nested(value={expression}): pass\n" + " @app.get('/after')\n" + " def after(): pass\n" + " return app\n" + "app = create()\n" + ) + kwargs = {"app_entry": "main:create"} + else: + source = ( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + f" def nested(value={expression}): pass\n" + " @app.get('/after')\n" + " def after(): pass\n" + ) + kwargs = {"bootstrap_entry": "main:run"} + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /after"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert inventory.status == InventoryStatus.CONDITIONAL + assert any( + item.source_path.name == "main.py" and item.source_line == 4 and "budget" in item.reason + for item in inventory.limitations + ) + + +@pytest.mark.parametrize("context", ["factory", "bootstrap"]) +def test_eager_definition_depth_budget_has_factory_bootstrap_parity( + tmp_path: Path, context: str +) -> None: + nested = "" + for depth in range(34): + nested += " " * (depth + 1) + f"class Level{depth}:\n" + nested += " " * 35 + "pass\n" + if context == "factory": + source = ( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + + nested + + " @app.get('/safe')\n" + + " def safe(): pass\n" + + " return app\n" + + "app = create()\n" + ) + kwargs = {"app_entry": "main:create"} + else: + source = ( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + nested + " @app.get('/safe')\n" + " def safe(): pass\n" + ) + kwargs = {"bootstrap_entry": "main:run"} + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert any("recursion budget" in item.reason for item in inventory.limitations) + + +@pytest.mark.parametrize("context", ["implicit", "explicit", "bootstrap"]) +def test_descriptor_lookup_in_eager_default_never_establishes_stale_routes( + tmp_path: Path, context: str +) -> None: + executed = tmp_path / "descriptor-lookup-executed" + (tmp_path / "trigger.py").write_text( + "from pathlib import Path\n" + "class Trigger:\n" + " def __getattribute__(self, name):\n" + " from main import app\n" + " app.routes.clear()\n" + " return None\n" + "trigger = Trigger()\n" + f"Path({str(executed)!r}).touch()\n", + encoding="utf-8", + ) + common = "from fastapi import FastAPI\nfrom trigger import trigger\n" + if context == "bootstrap": + source = ( + common + + "app = FastAPI()\n" + + "def run():\n" + + " @app.get('/before')\n" + + " def before(): pass\n" + + " def nested(value=trigger.fire): pass\n" + + " @app.get('/after')\n" + + " def after(): pass\n" + ) + kwargs = {"bootstrap_entry": "main:run"} + else: + source = ( + common + + "def create():\n" + + " app = FastAPI()\n" + + " @app.get('/before')\n" + + " def before(): pass\n" + + " def nested(value=trigger.fire): pass\n" + + " @app.get('/after')\n" + + " def after(): pass\n" + + " return app\n" + + "app = create()\n" + ) + kwargs = {"app_entry": "main:create"} if context == "explicit" else {} + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + if context == "implicit": + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.UNAVAILABLE + else: + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "GET /after", + "GET /before", + ] + assert inventory.status == InventoryStatus.CONDITIONAL + assert all( + endpoint.discovery_status == EndpointDiscoveryStatus.CONDITIONAL + for endpoint in inventory.endpoints + ) + assert any( + item.source_line == 7 and "protocol" in item.reason for item in inventory.limitations + ) + assert not executed.exists() + + +@pytest.mark.parametrize( + "expression", + [ + "trigger[0]", + "trigger + 1", + "trigger == 1", + "f'{trigger}'", + "[item for item in trigger]", + ], + ids=["subscript", "operator", "comparison", "formatting", "iteration"], +) +@pytest.mark.parametrize("context", ["factory", "bootstrap"]) +def test_protocol_dispatch_eager_defaults_are_conditional( + tmp_path: Path, expression: str, context: str +) -> None: + common = "from fastapi import FastAPI\ntrigger = object()\n" + if context == "factory": + source = ( + common + + "def create():\n" + + " app = FastAPI()\n" + + f" def nested(value={expression}): pass\n" + + " @app.get('/safe')\n" + + " def safe(): pass\n" + + " return app\n" + + "app = create()\n" + ) + kwargs = {"app_entry": "main:create"} + else: + source = ( + common + + "app = FastAPI()\n" + + "def run():\n" + + f" def nested(value={expression}): pass\n" + + " @app.get('/safe')\n" + + " def safe(): pass\n" + ) + kwargs = {"bootstrap_entry": "main:run"} + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert any("protocol" in item.reason for item in inventory.limitations) + + +@pytest.mark.parametrize("context", ["implicit", "explicit", "bootstrap"]) +def test_descriptor_assignment_target_never_establishes_stale_routes( + tmp_path: Path, context: str +) -> None: + executed = tmp_path / "descriptor-assignment-executed" + (tmp_path / "trigger.py").write_text( + "from pathlib import Path\n" + "class Trigger:\n" + " def __setattr__(self, name, value):\n" + " from main import app\n" + " app.routes.clear()\n" + "trigger = Trigger()\n" + f"Path({str(executed)!r}).touch()\n", + encoding="utf-8", + ) + common = "from fastapi import FastAPI\nfrom trigger import trigger\n" + body = ( + " @app.get('/before')\n" + " def before(): pass\n" + " class Local:\n" + " trigger.fire = None\n" + " @app.get('/after')\n" + " def after(): pass\n" + ) + if context == "bootstrap": + source = common + "app = FastAPI()\ndef run():\n" + body + kwargs = {"bootstrap_entry": "main:run"} + else: + source = ( + common + + "def create():\n" + + " app = FastAPI()\n" + + body + + " return app\n" + + "app = create()\n" + ) + kwargs = {"app_entry": "main:create"} if context == "explicit" else {} + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + if context == "implicit": + assert inventory.status == InventoryStatus.UNAVAILABLE + assert inventory.endpoints == [] + else: + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "GET /after", + "GET /before", + ] + assert inventory.status == InventoryStatus.CONDITIONAL + assert any( + item.source_line == 8 and "assignment target" in item.reason + for item in inventory.limitations + ) + assert not executed.exists() + + +@pytest.mark.parametrize("context", ["implicit", "explicit"]) +def test_factory_control_nested_eager_definitions_fail_closed(tmp_path: Path, context: str) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + " router = APIRouter()\n" + " if True:\n" + " class Routes:\n" + " @router.get('/conditional')\n" + " def route(): pass\n" + " app.include_router(router)\n" + " return app\n" + "app = create()\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor( + tmp_path, + app_entry="main:create" if context == "explicit" else None, + ).extract_inventory() + + assert inventory.endpoints == [] + assert inventory.status == ( + InventoryStatus.CONDITIONAL if context == "explicit" else InventoryStatus.UNAVAILABLE + ) + if context == "explicit": + assert any( + item.source_line == 6 and "under control flow" in item.reason + for item in inventory.limitations + ) + + +@pytest.mark.parametrize("context", ["implicit", "explicit"]) +@pytest.mark.parametrize( + "control", + [ + "if True:\n def nested(value=trigger.fire): pass", + "for _ in [1]:\n def nested(value=trigger.fire): pass", + "with manager:\n def nested(value=trigger.fire): pass", + "try:\n def nested(value=trigger.fire): pass\nexcept Exception:\n pass", + "match 1:\n case 1:\n def nested(value=trigger.fire): pass", + ], + ids=["if", "loop", "with", "try", "match"], +) +def test_factory_control_statement_definitions_are_detected( + tmp_path: Path, context: str, control: str +) -> None: + source = ( + "from fastapi import FastAPI\n" + "trigger = object()\n" + "manager = trigger\n" + "def create():\n" + " app = FastAPI()\n" + + "\n".join(f" {line}" for line in control.splitlines()) + + "\n @app.get('/safe')\n" + + " def safe(): pass\n" + + " return app\n" + + "app = create()\n" + ) + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor( + tmp_path, + app_entry="main:create" if context == "explicit" else None, + ).extract_inventory() + + if context == "implicit": + assert inventory.status == InventoryStatus.UNAVAILABLE + assert inventory.endpoints == [] + else: + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert any("under control flow" in item.reason for item in inventory.limitations) + + +@pytest.mark.parametrize("context", ["factory", "bootstrap"]) +def test_class_external_definition_binding_conditions_routes(tmp_path: Path, context: str) -> None: + if context == "factory": + source = ( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + " class Scope:\n" + " nonlocal app\n" + " def app(): pass\n" + " @app.get('/after')\n" + " def after(): pass\n" + " return app\n" + "app = create()\n" + ) + kwargs = {"app_entry": "main:create"} + evidence_line = 6 + else: + source = ( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " class Scope:\n" + " global app\n" + " def app(): pass\n" + " @app.get('/after')\n" + " def after(): pass\n" + ) + kwargs = {"bootstrap_entry": "main:run"} + evidence_line = 6 + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + assert inventory.status == InventoryStatus.CONDITIONAL + assert all( + endpoint.discovery_status == EndpointDiscoveryStatus.CONDITIONAL + for endpoint in inventory.endpoints + ) + assert any( + item.source_line == evidence_line and "rebind outer route state" in item.reason + for item in inventory.limitations + ) + + +@pytest.mark.parametrize( + "binding", + [ + "def app(): pass", + "class app: pass", + "import replacement as app", + "from replacement import value as app", + "app = None", + "del app", + "(app := None)", + "if True:\n def app(): pass", + ], + ids=[ + "function", + "class", + "import", + "from-import", + "assignment", + "delete", + "walrus", + "compound", + ], +) +def test_bootstrap_class_global_direct_and_compound_bindings_are_detected( + tmp_path: Path, binding: str +) -> None: + (tmp_path / "replacement.py").write_text("value = None\n", encoding="utf-8") + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " class Scope:\n" + " global app\n" + f" {binding}\n" + " @app.get('/after')\n" + " def after(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert inventory.status == InventoryStatus.CONDITIONAL + assert all( + endpoint.discovery_status == EndpointDiscoveryStatus.CONDITIONAL + for endpoint in inventory.endpoints + ) + assert any( + item.source_line == 6 and "rebind outer route state" in item.reason + for item in inventory.limitations + ) + + +@pytest.mark.skipif(sys.version_info < (3, 12), reason="PEP 695 requires Python 3.12") +def test_bootstrap_class_global_type_alias_binding_is_detected(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def run():\n" + " class Scope:\n" + " global app\n" + " type app = None\n" + " @app.get('/after')\n" + " def after(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert inventory.status == InventoryStatus.CONDITIONAL + assert any( + item.source_line == 6 and "rebind outer route state" in item.reason + for item in inventory.limitations + ) + + +@pytest.mark.parametrize("reexported", [False, True], ids=["direct", "reexported"]) +@pytest.mark.parametrize("context", ["implicit", "explicit", "bootstrap"]) +def test_eager_walrus_invalidates_imported_router_binding( + tmp_path: Path, reexported: bool, context: str +) -> None: + (tmp_path / "provider.py").write_text( + "from fastapi import APIRouter\n" + "router = APIRouter()\n" + "empty_router = APIRouter()\n" + "@router.get('/stale')\n" + "def stale(): pass\n", + encoding="utf-8", + ) + import_module = "public" if reexported else "provider" + if reexported: + (tmp_path / "public.py").write_text( + "from provider import empty_router, router\n", encoding="utf-8" + ) + imports = f"from {import_module} import empty_router, router\n" + if context == "bootstrap": + source = ( + "from fastapi import FastAPI\n" + + imports + + "app = FastAPI()\n" + + "def run():\n" + + " def nested(value=(router := empty_router)): pass\n" + + " app.include_router(router)\n" + ) + kwargs = {"bootstrap_entry": "main:run"} + else: + source = ( + "from fastapi import FastAPI\n" + + imports + + "def create():\n" + + " app = FastAPI()\n" + + " def nested(value=(router := empty_router)): pass\n" + + " app.include_router(router)\n" + + " return app\n" + + "app = create()\n" + ) + kwargs = {"app_entry": "main:create"} if context == "explicit" else {} + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + if context == "implicit": + assert inventory.status == InventoryStatus.UNAVAILABLE + assert inventory.endpoints == [] + else: + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /stale"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert any( + item.source_line == 5 and "named expression" in item.reason + for item in inventory.limitations + ) + + +def test_bootstrap_include_before_class_uncertainty_preserves_copied_endpoint( + tmp_path: Path, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "router = APIRouter()\n" + "@router.get('/before')\n" + "def before(): pass\n" + "def run():\n" + " app.include_router(router)\n" + " class Routes:\n" + " @router.get('/late')\n" + " def late(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /before"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.ESTABLISHED + assert inventory.status == InventoryStatus.ESTABLISHED + assert inventory.limitations == () + + +@pytest.mark.parametrize("context", ["factory", "bootstrap"]) +@pytest.mark.parametrize( + ("class_binding", "expected_status"), + [ + ("app: object", InventoryStatus.CONDITIONAL), + ("app: object = None", InventoryStatus.ESTABLISHED), + ], + ids=["annotation-only-uses-outer", "value-shadows-outer"], +) +def test_class_local_method_eager_headers_respect_runtime_annotation_binding( + tmp_path: Path, + context: str, + class_binding: str, + expected_status: InventoryStatus, +) -> None: + body = ( + " class Local:\n" + f" {class_binding}\n" + " def method(self, value=app, *, other: app = app): pass\n" + " @app.get('/safe')\n" + " def safe(): pass\n" + ) + if context == "factory": + source = ( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + body + " return app\n" + "app = create()\n" + ) + kwargs = {"app_entry": "main:create"} + else: + source = "from fastapi import FastAPI\napp = FastAPI()\ndef run():\n" + body + kwargs = {"bootstrap_entry": "main:run"} + (tmp_path / "main.py").write_text(source, encoding="utf-8") + + inventory = SecureASTExtractor(tmp_path, **kwargs).extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /safe"] + assert inventory.status == expected_status + if expected_status == InventoryStatus.CONDITIONAL: + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert any( + item.source_line == 6 and "escape a route object" in item.reason + for item in inventory.limitations + ) + else: + assert inventory.limitations == () + + +@pytest.mark.parametrize( + "binding", + [ + "def app(): pass", + "class app: pass", + "import replacement as app", + "from replacement import value as app", + "app = None", + "del app", + "(app := None)", + "if True:\n app = None", + ], + ids=[ + "function", + "class", + "import", + "from-import", + "assignment", + "delete", + "walrus", + "compound", + ], +) +def test_bootstrap_lexical_global_binding_displacement_conditions_prior_routes( + tmp_path: Path, + binding: str, +) -> None: + (tmp_path / "replacement.py").write_text("value = None\n", encoding="utf-8") + indented_binding = "\n".join(f" {line}" for line in binding.splitlines()) + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "@app.get('/before')\n" + "def before(): pass\n" + "def run():\n" + " if False:\n" + " global app\n" + f"{indented_binding}\n" + " @app.get('/after')\n" + " def after(): pass\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /before"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert inventory.status == InventoryStatus.CONDITIONAL + assert any( + item.source_line == 8 and "global binding displaces" in item.reason + for item in inventory.limitations + ) + + +def test_bootstrap_global_router_displacement_after_include_preserves_copy_cutoff( + tmp_path: Path, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "router = APIRouter()\n" + "@router.get('/copied')\n" + "def copied(): pass\n" + "def run():\n" + " global router\n" + " app.include_router(router)\n" + " router = None\n", + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /copied"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.ESTABLISHED + assert inventory.status == InventoryStatus.ESTABLISHED + assert inventory.limitations == () + + +@pytest.mark.parametrize( + ("definition", "evidence_line"), + [ + ( + " @router.get('/route', app.routes.clear())\n def route(): pass\n", + 7, + ), + ( + " class Routes:\n" + " @router.get('/route')\n" + " def route(value=app.routes.clear()): pass\n", + 9, + ), + ], + ids=["decorator-argument", "class-method-default"], +) +def test_bootstrap_route_shaped_eager_risk_keeps_root_owner( + tmp_path: Path, + definition: str, + evidence_line: int, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "router = APIRouter()\n" + "@app.get('/stale')\n" + "def stale(): pass\n" + "def run():\n" + definition, + encoding="utf-8", + ) + + inventory = SecureASTExtractor(tmp_path, bootstrap_entry="main:run").extract_inventory() + + assert [endpoint.identifier for endpoint in inventory.endpoints] == ["GET /stale"] + assert inventory.endpoints[0].discovery_status == EndpointDiscoveryStatus.CONDITIONAL + assert inventory.status == InventoryStatus.CONDITIONAL + assert any(item.source_line == evidence_line for item in inventory.limitations)