diff --git a/CHANGELOG.md b/CHANGELOG.md index 310e0c1..21c4bdd 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 13c12b7..68e7825 100644 --- a/README.md +++ b/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 ``` diff --git a/benchmark/results.json b/benchmark/results.json index 3a29689..2dd9474 100644 --- a/benchmark/results.json +++ b/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, diff --git a/pyproject.toml b/pyproject.toml index 9432311..875ea73 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/autonomyproof/__init__.py b/src/autonomyproof/__init__.py index 4190de4..285ff31 100644 --- a/src/autonomyproof/__init__.py +++ b/src/autonomyproof/__init__.py @@ -4,4 +4,4 @@ __all__ = ["__version__"] -__version__ = "0.19.0" +__version__ = "0.20.0" diff --git a/src/autonomyproof/astutils.py b/src/autonomyproof/astutils.py index 1c86eb7..242f34f 100644 --- a/src/autonomyproof/astutils.py +++ b/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 + ] diff --git a/src/autonomyproof/rules/execution.py b/src/autonomyproof/rules/execution.py index c85c1aa..c3ce266 100644 --- a/src/autonomyproof/rules/execution.py +++ b/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 diff --git a/tests/test_cli.py b/tests/test_cli.py index b585a7d..ed7ae6d 100644 --- a/tests/test_cli.py +++ b/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: diff --git a/tests/test_rules_execution.py b/tests/test_rules_execution.py index 38a670a..78e1aa9 100644 --- a/tests/test_rules_execution.py +++ b/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) == []