fix: terminate call-graph alias fixpoint on oscillating rebinds (#1247)

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - keep shard sibling discovery within the requested scan root
 - preserve per-shard metadata when aggregating sharded model families
 - reject plain PyTorch source text from Torch7 content routing
+- prevent picklescan call-graph alias cycles from hanging scans
 
 ## [0.2.45](https://github.com/promptfoo/modelaudit/compare/v0.2.44...v0.2.45) (2026-05-03)
 

packages/modelaudit-picklescan/CHANGELOG.md

@@ -51,6 +51,7 @@ and this package adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Bug Fixes
 
+- prevent call-graph alias cycles from hanging scans
 - detect nested brace-format lookups that reach tracked `defaultdict` factories
 - avoid `str.format` false positives when a `ChainMap` shadows a `defaultdict`
 - block `statistics.quantiles` call-iterator consumption in call-graph analysis

packages/modelaudit-picklescan/src/modelaudit_picklescan/call_graph.py

@@ -30,6 +30,7 @@
 _MAX_VISITED_FUNCTIONS = 64
 _MAX_CALLS_PER_FUNCTION = 128
 _MAX_ASSIGNMENT_ALIASES = 128
+_MAX_ASSIGNMENT_ALIAS_PASSES = 256
 _MAX_FUNCTION_INSTANCE_ALIASES = 32
 _MAX_CLASS_INSTANCE_ALIASES = 128
 _MAX_INHERITED_CLASS_METHODS = 128
@@ -66,6 +67,10 @@ def cache_clear(self) -> None:
 _SOURCE_SENSITIVE_CACHED_FUNCTIONS: set[_CacheClearable] = set()
 
 
+class _CallGraphAnalysisLimitError(RuntimeError):
+    """Raised when bounded call-graph enrichment cannot complete safely."""
+
+
 def _register_source_sensitive_cache(function: _CachedFunctionT) -> _CachedFunctionT:
     _SOURCE_SENSITIVE_CACHED_FUNCTIONS.add(cast(_CacheClearable, function))
     return function
@@ -497,6 +502,8 @@ def shared_source_sensitive_caches() -> Iterator[None]:
 def _safe_call_graph_entrypoints(function_name: str) -> tuple[str, ...]:
     try:
         return _call_graph_entrypoints(function_name)
+    except _CallGraphAnalysisLimitError:
+        raise
     except Exception:
         return ()
 
@@ -1664,9 +1671,22 @@ def _collect_assignment_aliases(
 ) -> dict[str, str]:
     node_list = tuple(nodes)
     assignment_aliases: dict[str, str] = {}
+    # A source-flattened branch may rebind one name indefinitely. Track full
+    # states so each distinct state gets a propagation pass before a cycle ends.
+    seen_states: set[tuple[tuple[str, str], ...]] = set()
+    passes = 0
 
     changed = True
     while changed and len(assignment_aliases) < _MAX_ASSIGNMENT_ALIASES:
+        if passes >= _MAX_ASSIGNMENT_ALIAS_PASSES:
+            raise _CallGraphAnalysisLimitError(
+                f"assignment alias analysis exceeded {_MAX_ASSIGNMENT_ALIAS_PASSES} propagation passes"
+            )
+        passes += 1
+        state = tuple(sorted(assignment_aliases.items()))
+        if state in seen_states:
+            break
+        seen_states.add(state)
         changed = False
         scoped_aliases = {**aliases, **assignment_aliases}
         for node in node_list:

packages/modelaudit-picklescan/tests/test_call_graph_assignment_alias_cycle.py

@@ -0,0 +1,190 @@
+"""Regression: assignment-alias fixpoint must terminate on oscillating binds.
+
+A module that binds the same name in both branches of an ``if``/``else``
+(e.g. the ``imaplib``/``http.server``/``nntplib`` stdlib ``__main__`` blocks)
+makes ``_collect_assignment_aliases`` oscillate the alias between two values
+without ever growing the dict. The dict-size guard never trips, so the
+fixpoint loop spins forever and the whole scan hangs (GitHub issue #1247).
+"""
+
+from __future__ import annotations
+
+import ast
+import threading
+from importlib.util import find_spec
+
+import pytest
+
+import modelaudit_picklescan.call_graph as call_graph
+from modelaudit_picklescan import PickleReport, Severity, scan_bytes
+from modelaudit_picklescan.api import _RUST_EXTENSION_MODULE
+from modelaudit_picklescan.call_graph import (
+    _CallGraphAnalysisLimitError,
+    _collect_assignment_aliases,
+    _collect_local_defs,
+    _module_level_statements,
+)
+
+_OSCILLATING_MODULE_SOURCE = """\
+class A:
+    pass
+
+
+class B:
+    pass
+
+
+if cond:
+    m = A()
+else:
+    m = B()
+"""
+
+_DEPENDENT_CYCLE_SOURCE = """\
+class A:
+    pass
+
+
+class B:
+    pass
+
+
+a = A()
+a = b
+b = B()
+b = a
+c = a
+"""
+
+
+def _run_with_timeout(target: object, timeout: float = 10.0) -> None:
+    thread = threading.Thread(target=target)  # type: ignore[arg-type]
+    thread.daemon = True
+    thread.start()
+    thread.join(timeout)
+    if thread.is_alive():
+        pytest.fail(f"call-graph analysis did not terminate within {timeout}s")
+
+
+def test_collect_assignment_aliases_terminates_on_branch_rebind() -> None:
+    tree = ast.parse(_OSCILLATING_MODULE_SOURCE)
+    statements = _module_level_statements(tree)
+    local_defs = _collect_local_defs(statements)
+    local_class_targets = {"testmod.A", "testmod.B"}
+
+    result: dict[str, dict[str, str]] = {}
+
+    def _collect() -> None:
+        result["aliases"] = _collect_assignment_aliases(
+            statements,
+            "testmod",
+            {},
+            local_defs,
+            local_class_targets,
+        )
+
+    _run_with_timeout(_collect)
+
+    aliases = result["aliases"]
+    # The rebinding name still resolves to one of the two local classes; the
+    # exact branch is order-dependent, but the loop must converge.
+    assert aliases.get("m") in {"testmod.A", "testmod.B"}
+
+
+def test_collect_assignment_aliases_terminates_after_cyclic_dependency_propagation() -> None:
+    tree = ast.parse(_DEPENDENT_CYCLE_SOURCE)
+    statements = _module_level_statements(tree)
+    local_defs = _collect_local_defs(statements)
+    local_class_targets = {"testmod.A", "testmod.B"}
+
+    result: dict[str, dict[str, str]] = {}
+
+    def _collect() -> None:
+        result["aliases"] = _collect_assignment_aliases(
+            statements,
+            "testmod",
+            {},
+            local_defs,
+            local_class_targets,
+        )
+
+    _run_with_timeout(_collect)
+
+    assert result["aliases"].get("c") in local_class_targets
+
+
+def test_collect_assignment_aliases_fails_closed_on_long_period_cycles() -> None:
+    periods = (7, 11, 13, 17, 19)
+    source_lines: list[str] = []
+    local_class_targets: set[str] = set()
+    for ring_index, period in enumerate(periods):
+        for position in range(period):
+            class_name = f"Ring{ring_index}Class{position}"
+            variable_name = f"ring_{ring_index}_{position}"
+            source_lines.extend((f"class {class_name}:", "    pass", "", f"{variable_name} = {class_name}()", ""))
+            local_class_targets.add(f"testmod.{class_name}")
+        for position in range(period):
+            next_position = (position + 1) % period
+            source_lines.append(f"ring_{ring_index}_{position} = ring_{ring_index}_{next_position}")
+
+    tree = ast.parse("\n".join(source_lines))
+    statements = _module_level_statements(tree)
+    local_defs = _collect_local_defs(statements)
+    result: dict[str, bool] = {}
+
+    def _collect() -> None:
+        with pytest.raises(_CallGraphAnalysisLimitError):
+            _collect_assignment_aliases(
+                statements,
+                "testmod",
+                {},
+                local_defs,
+                local_class_targets,
+            )
+        result["limited"] = True
+
+    _run_with_timeout(_collect)
+
+    assert result == {"limited": True}
+
+
+def test_assignment_alias_limit_is_not_hidden_by_safe_entrypoint_wrapper(monkeypatch: pytest.MonkeyPatch) -> None:
+    def _raise_limit(_function_name: str) -> tuple[str, ...]:
+        raise _CallGraphAnalysisLimitError("assignment alias limit")
+
+    monkeypatch.setattr(call_graph, "_call_graph_entrypoints", _raise_limit)
+    call_graph._safe_call_graph_entrypoints.cache_clear()
+
+    with pytest.raises(_CallGraphAnalysisLimitError, match="assignment alias limit"):
+        call_graph._safe_call_graph_entrypoints("long_period.module")
+
+
+def _stack_global_payload(module: str, name: str) -> bytes:
+    def _operand(value: str) -> bytes:
+        data = value.encode()
+        return b"\x8c" + bytes([len(data)]) + data
+
+    return b"\x80\x04" + _operand(module) + _operand(name) + b"\x93" + _operand("proof_of_bypass") + b"\x85R."
+
+
+@pytest.mark.skipif(
+    find_spec(_RUST_EXTENSION_MODULE) is None,
+    reason="Rust picklescan extension is not built",
+)
+@pytest.mark.skipif(
+    find_spec("imaplib") is None,
+    reason="imaplib stdlib module is unavailable",
+)
+def test_scan_imaplib_reference_terminates_and_flags() -> None:
+    """The issue #1247 proof-of-concept must finish and stay flagged."""
+    payload = _stack_global_payload("imaplib", "test")
+    result: dict[str, PickleReport] = {}
+
+    def _scan() -> None:
+        result["report"] = scan_bytes(payload)
+
+    _run_with_timeout(_scan)
+
+    report = result["report"]
+    severities = {finding.severity for finding in report.findings}
+    assert Severity.CRITICAL in severities

tests/conftest.py

@@ -192,6 +192,7 @@ def pytest_runtest_setup(item):
             "test_call_graph_local_imports.py",  # standalone picklescan function-local import RCE regressions
             "test_call_graph_six.py",  # standalone picklescan six.moves alias RCE regressions
             "test_call_graph_tkinter.py",  # standalone picklescan Tcl call-graph RCE regressions
+            "test_call_graph_assignment_alias_cycle.py",  # standalone picklescan alias fixpoint termination regressions
             "test_dill_joblib_enhanced.py",  # Dill/joblib pickle routing regression tests
             "test_pickle_context_filtering.py",  # Pickle context filtering regression tests
             "test_xdist_status.py",  # xdist worker progress reporting tests