Cross-function taint phase 1: single-file return propagation (0.20.0)

CHANGELOG.md

@@ -4,6 +4,15 @@ All notable changes to this project are documented here. The format is based on
 [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.20.0]
+
+### Added
+- **Cross-function taint (phase 1: single-file).** New `astutils.function_defs` /
+  `return_values` build a minimal same-file call graph, and **AG040** now follows model output
+  across a local helper's `return` — e.g. `def get_resp(p): return llm.invoke(p)` then
+  `exec(get_resp(p))` is now caught, where before the taint was lost at the function boundary.
+  Bounded depth, conservative, still single-file (no cross-module reachability yet).
+
 ## [0.19.0]
 
 ### Added

README.md

@@ -73,14 +73,16 @@ ATLAS/ATT&CK, and CVE** mappings where a genuine one exists.
 
 ### How the analysis works (and its limits)
 
-AutonomyProof is AST-based static analysis. It resolves imports and follows **single-function**
-source tracking — so it sees HTTP through session variables (`c = httpx.Client(); c.get(url)`),
-one-line SSRF indirection, and whether a URL comes from a hardcoded constant / trusted config
-(`settings.X`, `os.environ`) versus a tool parameter. SSRF classification uses real
-`ipaddress` range checks, not string matching.
-
-It does **not** yet do cross-function taint or whole-program call-graph reachability, so a value
-laundered through several functions can still be missed. This is deliberately conservative and
+AutonomyProof is AST-based static analysis. It resolves imports and follows source tracking —
+so it sees HTTP through session variables (`c = httpx.Client(); c.get(url)`), one-line SSRF
+indirection, and whether a URL comes from a hardcoded constant / trusted config (`settings.X`,
+`os.environ`) versus a tool parameter. SSRF classification uses real `ipaddress` range checks,
+not string matching.
+
+It also follows **cross-function taint within a file**: a value returned by a local helper is
+tracked into its caller — e.g. model output laundered through a helper into `eval`/`exec` is
+caught (AG040). It does **not** yet do whole-program / cross-file call-graph reachability, so a
+value laundered across modules can still be missed. This is deliberately conservative and
 improving; treat findings as "this authority is reachable in the code," not a proof of
 exploitability.
 
@@ -143,7 +145,7 @@ re-run `autonomyproof baseline .` and commit the updated file in the same PR.
 Use the action directly:
 
 ```yaml
-- uses: autonomyproof/autonomyproof-cli@v0.19.0
+- uses: autonomyproof/autonomyproof-cli@v0.20.0
   with:
     target: .
     fail-on: high
@@ -170,7 +172,7 @@ Gate locally before a commit ever leaves your machine:
 # .pre-commit-config.yaml
 repos:
   - repo: https://github.com/autonomyproof/autonomyproof-cli
-    rev: v0.19.0
+    rev: v0.20.0
     hooks:
       - id: autonomyproof
 ```

benchmark/results.json

@@ -1,11 +1,11 @@
 {
   "repo_count": 41,
-  "total_files": 34817,
-  "total_findings": 8942,
+  "total_files": 34820,
+  "total_findings": 8943,
   "by_rule": {
     "AG001": 37,
     "AG002": 168,
-    "AG003": 3334,
+    "AG003": 3335,
     "AG004": 58,
     "AG005": 918,
     "AG006": 61,
@@ -97,11 +97,11 @@
     {
       "repo": "openai-agents",
       "status": "ok",
-      "files_scanned": 907,
-      "findings": 396,
+      "files_scanned": 910,
+      "findings": 397,
       "by_rule": {
         "AG002": 11,
-        "AG003": 34,
+        "AG003": 35,
         "AG004": 4,
         "AG005": 9,
         "AG007": 2,

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "autonomyproof"
-version = "0.19.0"
+version = "0.20.0"
 description = "Open-source local scanner that finds unsafe capabilities and missing guardrails in Python AI-agent code."
 readme = "README.md"
 requires-python = ">=3.11"

src/autonomyproof/__init__.py

@@ -4,4 +4,4 @@
 
 __all__ = ["__version__"]
 
-__version__ = "0.19.0"
+__version__ = "0.20.0"

src/autonomyproof/astutils.py

@@ -222,3 +222,30 @@ def string_literals(node: ast.AST) -> list[str]:
         for child in ast.walk(node)
         if isinstance(child, ast.Constant) and isinstance(child.value, str)
     ]
+
+
+def function_defs(tree: ast.AST) -> dict[str, ast.FunctionDef | ast.AsyncFunctionDef]:
+    """Map each same-file function name to its definition (first definition wins).
+
+    This is the minimal call graph for cross-function taint: it lets a rule resolve a
+    plain ``helper(x)`` call to the ``def helper`` in the same file and reason about what
+    that helper returns.
+    """
+    defs: dict[str, ast.FunctionDef | ast.AsyncFunctionDef] = {}
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
+            defs.setdefault(node.name, node)
+    return defs
+
+
+def return_values(func: ast.FunctionDef | ast.AsyncFunctionDef) -> list[ast.expr]:
+    """Return the value expressions of ``func``'s own ``return`` statements.
+
+    Scope-respecting: it does not descend into nested functions, so a return inside a
+    closure defined within ``func`` is not attributed to ``func``.
+    """
+    return [
+        node.value
+        for node in _walk_scope(func)
+        if isinstance(node, ast.Return) and node.value is not None
+    ]

src/autonomyproof/rules/execution.py

@@ -5,7 +5,7 @@
 import ast
 from collections.abc import Iterable
 
-from autonomyproof.astutils import is_true_literal, keyword
+from autonomyproof.astutils import function_defs, is_true_literal, keyword, return_values
 from autonomyproof.models import Finding, Mappings, Severity
 from autonomyproof.rules.base import Rule, RuleContext
 
@@ -198,25 +198,39 @@ class InsecureModelOutputRule(Rule):
     )
 
     def check(self, ctx: RuleContext) -> Iterable[Finding]:
+        fmap = function_defs(ctx.analysis.tree)
         for call in ctx.analysis.calls:
             name = ctx.analysis.resolve_call(call)
             if name not in _CODE_EXEC_SINKS and name not in _SHELL_EXEC_SINKS:
                 continue
             if not call.args:
                 continue
-            if self._from_model(ctx, call.args[0], call):
+            if self._from_model(ctx, call.args[0], call, fmap):
                 yield self.make_finding(
                     ctx, call, evidence=f"Model output flows into {name}() and is executed"
                 )
 
     def _from_model(
-        self, ctx: RuleContext, node: ast.expr, origin: ast.AST, depth: int = 0
+        self,
+        ctx: RuleContext,
+        node: ast.expr,
+        origin: ast.AST,
+        fmap: dict[str, ast.FunctionDef | ast.AsyncFunctionDef],
+        depth: int = 0,
     ) -> bool:
         base = _terminal(node)
         if _is_model_call(base):
             return True
+        # Cross-function (single file): a call to a local helper whose body returns model
+        # output taints the result too — following the taint across the function boundary.
+        if isinstance(base, ast.Call) and isinstance(base.func, ast.Name) and depth < 4:
+            helper = fmap.get(base.func.id)
+            if helper is not None and any(
+                self._from_model(ctx, ret, ret, fmap, depth + 1) for ret in return_values(helper)
+            ):
+                return True
         if isinstance(base, ast.Name) and depth < 4:
             assigned = ctx.analysis.resolve_local_value(base.id, origin)
             if assigned is not None:
-                return self._from_model(ctx, assigned, origin, depth + 1)
+                return self._from_model(ctx, assigned, origin, fmap, depth + 1)
         return False

tests/test_cli.py

@@ -33,7 +33,7 @@ def _home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
 def test_version(runner: CliRunner) -> None:
     result = runner.invoke(cli.main, ["--version"])
     assert result.exit_code == 0
-    assert "0.19.0" in result.output
+    assert "0.20.0" in result.output
 
 
 def test_init_creates_and_is_idempotent(runner: CliRunner) -> None:

tests/test_rules_execution.py

@@ -154,3 +154,61 @@ def test_ag040_deep_alias_chain_not_tracked_clean() -> None:
         "    d = c\n    e = d\n    exec(e)\n"
     )
     assert run_rule(InsecureModelOutputRule(), code) == []
+
+
+# --- AG040 cross-function (single-file) taint --------------------------------
+def test_ag040_xfn_helper_returns_model() -> None:
+    code = (
+        "def get_response(p):\n    return llm.invoke(p)\ndef run(p):\n    exec(get_response(p))\n"
+    )
+    assert run_rule(InsecureModelOutputRule(), code)
+
+
+def test_ag040_xfn_helper_via_variable() -> None:
+    code = (
+        "def get_response(p):\n    r = llm.predict(p)\n    return r\n"
+        "def run(p):\n    code = get_response(p)\n    exec(code)\n"
+    )
+    assert run_rule(InsecureModelOutputRule(), code)
+
+
+def test_ag040_xfn_helper_content_accessor() -> None:
+    code = (
+        "def get_response(p):\n    return llm.invoke(p)\n"
+        "def run(p):\n    exec(get_response(p).content)\n"
+    )
+    assert run_rule(InsecureModelOutputRule(), code)
+
+
+def test_ag040_xfn_helper_with_bare_return() -> None:
+    code = (
+        "def get_response(p):\n    if not p:\n        return\n    return llm.invoke(p)\n"
+        "def run(p):\n    exec(get_response(p))\n"
+    )
+    assert run_rule(InsecureModelOutputRule(), code)
+
+
+def test_ag040_xfn_helper_returns_nonmodel_clean() -> None:
+    code = (
+        "def get_response(p):\n    return requests.get(p).text\n"
+        "def run(p):\n    exec(get_response(p))\n"
+    )
+    assert run_rule(InsecureModelOutputRule(), code) == []
+
+
+def test_ag040_xfn_unknown_function_clean() -> None:
+    # The callee is not defined in this file, so nothing can be proven about its return.
+    assert run_rule(InsecureModelOutputRule(), "def run(p):\n    exec(make_code(p))\n") == []
+
+
+def test_ag040_xfn_deep_chain_not_tracked_clean() -> None:
+    # Cross-function tracking is bounded; a 5-hop helper chain exceeds the depth limit.
+    code = (
+        "def h5(p):\n    return llm.invoke(p)\n"
+        "def h4(p):\n    return h5(p)\n"
+        "def h3(p):\n    return h4(p)\n"
+        "def h2(p):\n    return h3(p)\n"
+        "def h1(p):\n    return h2(p)\n"
+        "def run(p):\n    exec(h1(p))\n"
+    )
+    assert run_rule(InsecureModelOutputRule(), code) == []