From 2f628b921b94aa698fa507cc0759ec68e4951a9f Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:22:45 +0200 Subject: [PATCH] feat(prover): delivery C: the ordered trace with argument provenance for every examined pair Codex delivery under spec access-control-prover, requirement 9. `semantic/obligations/trace.py` serialises each (entry, operation) pair the prover examined, discharged or unguarded: the entry with its boundary, access and parameters; each hop in order with its call line, the guards before the call and the values forwarded; the operation with the provenance of each argument traced through the function's binds; the guards after it; the verdict. One canonical text form and one JSON form, no language, framework, repository, size or count fields. The prover result, the arm record and each unguarded finding carry the traces. On the ec2-plugin fix: 32 traces, 4 unguarded. Gate: 3534 tests pass; the reserved-repository guard passes on the tracked export. Co-Authored-By: Claude Fable 5.1 --- benchmarks/valid/prover_arm.py | 8 +- src/openultrasast/hunt/prover.py | 27 ++- .../semantic/obligations/trace.py | 229 ++++++++++++++++++ tests/test_hunt_prover.py | 5 + tests/test_obligation_trace.py | 198 +++++++++++++++ tests/test_prover_arm.py | 3 + 6 files changed, 467 insertions(+), 3 deletions(-) create mode 100644 src/openultrasast/semantic/obligations/trace.py create mode 100644 tests/test_obligation_trace.py diff --git a/benchmarks/valid/prover_arm.py b/benchmarks/valid/prover_arm.py index ab2adfb..d75a246 100644 --- a/benchmarks/valid/prover_arm.py +++ b/benchmarks/valid/prover_arm.py @@ -50,6 +50,7 @@ def _prove(change, repos, *, max_depth): "status": "prover_covered", "instrument": instrument, "findings": [], + "traces": [], "coverage": {}, "degradations": [], "push": {"admission": {"defects": [], "dispositions": []}, "quick_tier": []}, @@ -112,7 +113,12 @@ def _prove(change, repos, *, max_depth): raise InstrumentFailure("zero-byte read") result = prove_change(root, change, ctx, max_depth=max_depth) instrument["hook_bytes_read"] = sum(instrument["file_bytes"].values()) - record.update(findings=list(result.findings), coverage=asdict(result.coverage), degradations=list(result.degradations)) + record.update( + findings=list(result.findings), + traces=list(result.traces), + coverage=asdict(result.coverage), + degradations=list(result.degradations), + ) record["push"]["admission"]["defects"] = [ {**finding, "defect_id": finding["id"], "source": "prover", "locations": [[finding["path"], finding["line"]]]} for finding in result.findings diff --git a/src/openultrasast/hunt/prover.py b/src/openultrasast/hunt/prover.py index c5c91e0..4033e7b 100644 --- a/src/openultrasast/hunt/prover.py +++ b/src/openultrasast/hunt/prover.py @@ -13,13 +13,14 @@ from .. import mapping, source from ..preprocess import FileTarget, count_loc, detect_language, detect_tags from ..semantic.facts import load_facts from ..semantic.ir import FileIR, FunctionIR, trailing_name -from ..semantic.obligations.check import _on_caller_hop, _relevant, check_obligations +from ..semantic.obligations.check import _on_caller_hop, _relevant, _route_path, check_obligations from ..semantic.obligations.dominance import OrderDominance from ..semantic.obligations.enforcement import witnesses from ..semantic.obligations.facts import load_obligation_facts from ..semantic.obligations.operations import Discharge, Operation, find_discharges, find_operations, valid_witness from ..semantic.obligations.paths import compose_paths from ..semantic.obligations.policy import DEFAULT_POLICY_RELPATH, load_declared_policy +from ..semantic.obligations.trace import build_trace from .producers import Context, _safe_path @@ -40,6 +41,7 @@ class ProverResult: findings: tuple[dict[str, Any], ...] coverage: Coverage degradations: tuple[dict[str, Any], ...] + traces: tuple[dict[str, Any], ...] = () def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = None, *, max_depth: int = 3) -> ProverResult: @@ -232,6 +234,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No ): extra_witnesses.append(replace(witness, kind="path_guard")) findings: dict[str, dict[str, Any]] = {} + traces: list[dict[str, Any]] = [] checked: set[Operation] = set() undischarged: set[Operation] = set() for record in paths: @@ -255,6 +258,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No ) degradations.extend(result.degradations) target_path, _, target_name = record.hops[-1][0].rpartition("::") + pair_traces: dict[Operation, dict[str, Any]] = {} for op in operations.get((target_path, target_name), ()): checked.add(op) # Silence from a public premise is not evidence that a check dominated the operation. @@ -268,6 +272,24 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No } if (set(op.requires) - dominated) if op.fact_id in handler_ids else not dominated: undischarged.add(op) + missing = next((finding.missing for finding in result.findings if finding.operation == op), None) + trace_entry = entry_index[record.entry] + route = _route_path(op, ctx.texts[op.path]) or trace_entry.name + declared_access = policy.route_access(route) if policy is not None and route else None + trace = build_trace( + path=record, + entry=trace_entry, + access=declared_access, + operation=op, + irs=path_irs, + facts=facts, + witnesses=result.discharges, + dominance=dominance, + dominating=tuple(sorted(dominated)), + missing=missing, + ).record() + traces.append(trace) + pair_traces[op] = trace for finding in result.findings: op = finding.operation identity = f"obligation:{op.kind}:{op.path}:{op.line}:{finding.missing}" @@ -300,6 +322,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No operation=dict(id=op.fact_id, kind=op.kind, calls=calls, line=op.line), missing=finding.missing, label=finding.label, + trace=pair_traces[op], ), ) coverage = Coverage( @@ -312,7 +335,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No len(checked - undischarged), dict(reasons), ) - return ProverResult(tuple(findings.values()), coverage, tuple(degradations)) + return ProverResult(tuple(findings.values()), coverage, tuple(degradations), tuple(traces)) class _HandlerDominance: diff --git a/src/openultrasast/semantic/obligations/trace.py b/src/openultrasast/semantic/obligations/trace.py new file mode 100644 index 0000000..f41e1f9 --- /dev/null +++ b/src/openultrasast/semantic/obligations/trace.py @@ -0,0 +1,229 @@ +"""Ordered security structure and bounded argument provenance for examined paths.""" + +from __future__ import annotations + +import ast +import json +import re +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass, replace +from typing import Any + +from ..ir import FileIR, FunctionIR, trailing_name +from .dominance import Dominance +from .facts import ObligationFacts +from .operations import Discharge, Operation, valid_witness +from .paths import PathRecord +from .shapes import _call_matches + + +@dataclass(frozen=True) +class Guard: + kind: str + scope: str + line: int | None + provenance: str + fact_id: str + + +@dataclass(frozen=True) +class Hop: + function: str + call_line: int | None + guards_before_call: tuple[Guard, ...] + forwarded: tuple[str, ...] + + +@dataclass(frozen=True) +class Trace: + entry_kind: str + boundary: str + access: str + parameters: tuple[str, ...] + hops: tuple[Hop, ...] + operation_kind: str + fact_id: str + operation_line: int + arguments: tuple[tuple[str, str], ...] + guards_after: tuple[Guard, ...] + requires: tuple[str, ...] + dominating: tuple[str, ...] + missing: str | None + + def record(self) -> dict[str, Any]: + """Return JSON values with keys sorted recursively, retaining sequence order.""" + result: dict[str, Any] = json.loads(json.dumps(asdict(self), sort_keys=True)) + return result + + def text(self) -> str: + """Emit entry, hops, operation, later guards and verdict in that order.""" + record = self.record() + components = [("entry", {key: record[key] for key in ("entry_kind", "boundary", "access", "parameters")})] + components.extend(("hop", hop) for hop in record["hops"]) + components.append(("operation", {key: record[key] for key in ("operation_kind", "fact_id", "operation_line", "arguments")})) + components.extend(("guard_after", guard) for guard in record["guards_after"]) + components.append(("verdict", {key: record[key] for key in ("requires", "dominating", "missing")})) + return "\n".join(f"{label} {json.dumps(value, sort_keys=True, separators=(',', ':'))}" for label, value in components) + + +def _arguments(function: FunctionIR, line: int, names: Sequence[str], text: str) -> tuple[str, ...]: + calls = [call for call in function.calls if call.line == line and any(_call_matches(call.name, name) for name in names)] + if calls and calls[0].arg_texts: + return calls[0].arg_texts + # Some grammars omit arguments. Read only the selected call, balancing nested delimiters and strings. + lines = text.splitlines() + if not 0 < line <= len(lines): + return () + source = "\n".join(lines[line - 1 : function.end_line]) + for name in names: + match = re.search(rf"(? str: + """Follow the most recent preceding assignment, never guessing an unresolved origin.""" + value = value.strip() + while True: + bind = next( + ( + item + for item in sorted(function.binds, key=lambda item: (item.line, item.order), reverse=True) + if item.line < before and item.name == value + ), + None, + ) + if bind is None: + break + value, before = bind.value_text.strip(), bind.line + if value in {"true", "false", "null"}: + return "constant" + try: + ast.literal_eval(value) + except (ValueError, SyntaxError): + pass + else: + return "constant" + if any(_source_matches(value, source) for fact in facts.dischargers for source in fact.identity_sources): + return "authenticated_context" + if any(_source_matches(value, source) for fact in facts.dischargers for source in fact.request_sources): + return "request_input" + if value in parameters: + return parameters[value] + member = re.fullmatch(r"(\$?[A-Za-z_]\w*)(?:(?:\.|->)[A-Za-z_]\w*|\[[^\[\]]+\])+", value) + if member is not None: + return _origin(member.group(1), function, before, parameters, facts) + return "unknown" + + +def _source_matches(value: str, source: str) -> bool: + suffix = r"(?![\w$])" if source[-1:].isalnum() or source.endswith("_") else "" + return bool(source and re.search(r"(? Trace: + """Build one trace without changing the checker's verdict or coverage.""" + hops: list[Hop] = [] + after_by_hop: list[list[Guard]] = [] + incoming: tuple[str, ...] = () + parameters: tuple[str, ...] = () + arguments: tuple[tuple[str, str], ...] = () + boundary = str(getattr(entry, "trust_boundary", "unknown")) + for index, (node, call_line) in enumerate(path.hops): + file, _, name = node.rpartition("::") + ir, text = irs[file] + function = next(fn for fn in ir.functions if fn.name == name) + scoped = facts.for_language(ir.language) + if index == 0: + parameters = function.params + origins = {param: "request_input" if boundary in {"http_request", "network_request"} else "unknown" for param in parameters} + else: + origins = dict(zip(tuple(param for param in function.params if param not in {"self", "cls", "this"}), incoming, strict=False)) + final = index == len(path.hops) - 1 + line = operation.line if final else call_line + measured = operation if final else replace(operation, path=file, function=name, line=line or function.end_line) + before: list[Guard] = [] + after: list[Guard] = [] + after_by_hop.append(after) + local = sorted( + (w for w in witnesses if (w.path, w.function) == (file, name)), + key=lambda w: (w.line or 0, w.kind, w.scope, w.provenance, w.fact_id), + ) + for witness in local: + guard = Guard(witness.kind, witness.scope, witness.line, witness.provenance, witness.fact_id) + if valid_witness(witness) and dominance.dominates(witness, measured, None): + before.append(guard) + elif witness.line is not None and line is not None and witness.line > line: + after.append(guard) + if final: + names = next(fact.calls for fact in scoped.operations if fact.id == operation.fact_id) + else: + names = (path.hops[index + 1][0].rpartition("::")[2],) + values = _arguments(function, line, names, text) if line is not None and names else () + incoming = tuple(_origin(value, function, line or function.end_line + 1, origins, scoped) for value in values) + if final: + arguments = tuple(zip(values, incoming, strict=True)) + hops.append( + Hop( + name, + call_line, + tuple(before), + tuple(value for value, origin in zip(values, incoming, strict=True) if not final and origin == "request_input"), + ) + ) + return Trace( + str(getattr(entry, "kind", "unknown")), + boundary, + access if access is not None else str(getattr(entry, "access_level", "unknown")), + parameters, + tuple(hops), + operation.kind, + operation.fact_id, + operation.line, + arguments, + tuple(guard for later in reversed(after_by_hop) for guard in later), + operation.requires, + tuple(sorted(dominating)), + missing, + ) diff --git a/tests/test_hunt_prover.py b/tests/test_hunt_prover.py index 271a025..b505274 100644 --- a/tests/test_hunt_prover.py +++ b/tests/test_hunt_prover.py @@ -53,6 +53,9 @@ def test_stapler_verdict(tmp_path, guard, findings): ctx = Context(root, change) result = prove_change(root, change, ctx) assert len(result.findings) == findings + assert len(result.traces) == 1 + assert result.traces[0]["missing"] == ("path_guard" if findings else None) + assert result.traces[0]["dominating"] == ([] if findings else ["path_guard"]) assert result.coverage.operations_checked == 1 assert result.coverage.operations_discharged == 1 - findings assert result.coverage.reached == 1 @@ -64,6 +67,7 @@ def test_stapler_verdict(tmp_path, guard, findings): assert finding["entry"]["kind"] == "route" and finding["hops"] == [["Fixture.java::doTestConnection", None]] assert finding["operation"]["calls"] == ["lookupCredentials"] and finding["missing"] == "path_guard" assert finding["evidence"] + assert finding["trace"] is result.traces[0] def test_unrecognised_is_coverage_only(tmp_path): @@ -169,6 +173,7 @@ def test_declared_public_route_suppresses_premise(tmp_path): result = prove_change(root, change) assert not result.findings and result.coverage.operations_checked == 1 assert result.coverage.operations_discharged == 0 + assert result.traces[0]["access"] == "public" def test_nonpublic_unreached_function_in_partly_covered_repo(tmp_path): diff --git a/tests/test_obligation_trace.py b/tests/test_obligation_trace.py new file mode 100644 index 0000000..9fefa0e --- /dev/null +++ b/tests/test_obligation_trace.py @@ -0,0 +1,198 @@ +"""Ordered traces preserve argument origins across a bounded caller chain.""" + +import json +from types import SimpleNamespace + +from openultrasast.semantic.ir import parse_python_ast +from openultrasast.semantic.obligations.dominance import OrderDominance +from openultrasast.semantic.obligations.facts import DischargerFact, ObligationFacts, OperationFact +from openultrasast.semantic.obligations.operations import Discharge, Operation +from openultrasast.semantic.obligations.paths import PathRecord +from openultrasast.semantic.obligations.trace import build_trace + + +def fixture(): + text = """def entry(item): + guard() + middle(item) +def middle(value): + alias = value + final(alias) +def final(key): + owner = request.user + limit = 7 + mystery = unrecognised() + read(key, owner, limit, mystery) + guard() +""" + ir = parse_python_ast("app.py", text) + facts = ObligationFacts( + "1", + (OperationFact("read", "protected_read", "python", ("read",), None, ("path_guard",), "high"),), + ( + DischargerFact( + "identity", "identity_constraint", "python", identity_sources=("request.user",), request_sources=("request.args",) + ), + ), + ) + return text, ir, facts + + +def trace(): + text, ir, facts = fixture() + return build_trace( + path=PathRecord((("app.py::entry", 3), ("app.py::middle", 6), ("app.py::final", None)), "app.py::entry", True), + entry=SimpleNamespace(kind="route", trust_boundary="http_request", access_level="unknown"), + operation=Operation("app.py", 11, "final", "protected_read", "read", None, ("path_guard",)), + irs={"app.py": (ir, text)}, + facts=facts, + witnesses=( + Discharge("app.py", 2, "entry", "path_guard", "guard", "authenticated_context", "statement"), + Discharge("app.py", 12, "final", "path_guard", "guard", "authenticated_context", "statement"), + ), + dominance=OrderDominance({"app.py": text}), + dominating=("path_guard",), + missing=None, + ) + + +def test_four_origins_and_three_hop_order(): + result = trace() + assert result.arguments == (("key", "request_input"), ("owner", "authenticated_context"), ("limit", "constant"), ("mystery", "unknown")) + assert [hop.function for hop in result.hops] == ["entry", "middle", "final"] + assert result.hops[0].forwarded == ("item",) + assert result.hops[1].forwarded == ("alias",) + assert result.hops[0].guards_before_call[0].line == 2 + assert result.guards_after[0].line == 12 + + +def test_deterministic_text_and_canonical_json(): + result = trace() + assert result.text() == trace().text() + assert [line.split(" ", 1)[0] for line in result.text().splitlines()] == [ + "entry", + "hop", + "hop", + "hop", + "operation", + "guard_after", + "verdict", + ] + record = result.record() + assert json.loads(json.dumps(record)) == record + assert list(record) == sorted(record) + assert record == trace().record() + assert not {"language", "framework", "repository", "size", "count"} & record.keys() + + +def test_recent_bind_sources_and_unknown_composites(): + text, _, facts = fixture() + text = text.replace( + " read(key, owner, limit, mystery)", + """ key = request.args['id'] + owner = external + limit = [mystery] + read(key, owner, limit, mystery)""", + ) + ir = parse_python_ast("app.py", text) + result = build_trace( + path=PathRecord((("app.py::final", None),), "app.py::final", True), + entry=SimpleNamespace(kind="route", trust_boundary="internal", access_level="unknown"), + operation=Operation("app.py", 14, "final", "protected_read", "read", None, ("path_guard",)), + irs={"app.py": (ir, text)}, + facts=facts, + witnesses=(), + dominance=OrderDominance({"app.py": text}), + dominating=(), + missing="path_guard", + ) + assert result.arguments == (("key", "request_input"), ("owner", "unknown"), ("limit", "unknown"), ("mystery", "unknown")) + + +def test_call_text_fallback_balances_nested_arguments(): + from dataclasses import replace + + text, ir, facts = fixture() + text = text.replace("read(key, owner, limit, mystery)", "read(key, nested(owner, limit), 'a,b', mystery)") + parsed = parse_python_ast("app.py", text) + functions = tuple(replace(fn, calls=tuple(replace(call, arg_texts=()) for call in fn.calls)) for fn in parsed.functions) + ir = replace(ir, functions=functions) + result = build_trace( + path=PathRecord((("app.py::final", None),), "app.py::final", True), + entry=SimpleNamespace(kind="route", trust_boundary="network_request", access_level="unknown"), + operation=Operation("app.py", 11, "final", "protected_read", "read", None, ("path_guard",)), + irs={"app.py": (ir, text)}, + facts=facts, + witnesses=(), + dominance=OrderDominance({"app.py": text}), + dominating=(), + missing="path_guard", + ) + assert result.arguments == ( + ("key", "request_input"), + ("nested(owner, limit)", "unknown"), + ("'a,b'", "constant"), + ("mystery", "unknown"), + ) + + +def test_request_alias_and_member_access_across_methods(): + text, _, facts = fixture() + text = ( + text.replace("def middle(value):", "def middle(self, value):") + .replace("def final(key):", "def final(self, key):") + .replace("read(key, owner, limit, mystery)", "read(key.id, owner, limit, mystery)") + ) + ir = parse_python_ast("app.py", text) + result = build_trace( + path=PathRecord((("app.py::entry", 3), ("app.py::middle", 6), ("app.py::final", None)), "app.py::entry", True), + entry=SimpleNamespace(kind="route", trust_boundary="http_request", access_level="unknown"), + operation=Operation("app.py", 11, "final", "protected_read", "read", None, ("path_guard",)), + irs={"app.py": (ir, text)}, + facts=facts, + witnesses=(), + dominance=OrderDominance({"app.py": text}), + dominating=(), + missing="path_guard", + ) + assert result.arguments[0] == ("key.id", "request_input") + + +def test_source_name_suffix_is_unknown(): + text, _, facts = fixture() + text = text.replace("owner = request.user", "owner = request.users") + ir = parse_python_ast("app.py", text) + result = build_trace( + path=PathRecord((("app.py::final", None),), "app.py::final", True), + entry=SimpleNamespace(kind="route", trust_boundary="internal", access_level="unknown"), + operation=Operation("app.py", 11, "final", "protected_read", "read", None, ("path_guard",)), + irs={"app.py": (ir, text)}, + facts=facts, + witnesses=(), + dominance=OrderDominance({"app.py": text}), + dominating=(), + missing="path_guard", + ) + assert result.arguments[1] == ("owner", "unknown") + + +def test_guards_after_follow_return_order(): + text, _, facts = fixture() + text = text.replace(" middle(item)", " middle(item)\n guard()").replace(" final(alias)", " final(alias)\n guard()") + ir = parse_python_ast("app.py", text) + result = build_trace( + path=PathRecord((("app.py::entry", 3), ("app.py::middle", 7), ("app.py::final", None)), "app.py::entry", True), + entry=SimpleNamespace(kind="route", trust_boundary="http_request", access_level="unknown"), + operation=Operation("app.py", 13, "final", "protected_read", "read", None, ("path_guard",)), + irs={"app.py": (ir, text)}, + facts=facts, + witnesses=( + Discharge("app.py", 4, "entry", "path_guard", "entry-after", "authenticated_context", "statement"), + Discharge("app.py", 8, "middle", "path_guard", "middle-after", "authenticated_context", "statement"), + Discharge("app.py", 14, "final", "path_guard", "final-after", "authenticated_context", "statement"), + ), + dominance=OrderDominance({"app.py": text}), + dominating=(), + missing="path_guard", + ) + assert [guard.fact_id for guard in result.guards_after] == ["final-after", "middle-after", "entry-after"] diff --git a/tests/test_prover_arm.py b/tests/test_prover_arm.py index 4b2fee8..a995cd0 100644 --- a/tests/test_prover_arm.py +++ b/tests/test_prover_arm.py @@ -67,6 +67,9 @@ def test_run_resume_summary_and_locations(pool, monkeypatch, capsys): assert prover_arm.main(["run", str(manifest), "--repos", str(repos), "--out", str(out)]) == 0 rows = score.load_records(out) assert len(rows) == 2 and all(r["done"] and r["seconds"] > 0 for r in rows) + assert all(len(row["traces"]) == 1 for row in rows) + assert rows[0]["findings"][0]["trace"] == rows[0]["traces"][0] + assert rows[1]["traces"][0]["missing"] is None assert score.catches(rows[0]) == ["alerts"] and not score.false_alarm(rows[1]) summary = json.loads((out / "summary.json").read_text()) assert summary["recall"]["count"] == summary["recall"]["n"] == 1