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