From b8f8f727c651635b47a1f717dc275b983ecfd00d Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 8 Sep 2026 13:49:55 +0200 Subject: [PATCH] feat(model-grounded-detection): group 3, the evidence ladder and the judge 3.1 adds StaticFinding.rung beside evidence_level rather than renaming it. The existing evidence_level carries `static_corroboration`, emitted by the flat IR overlay -- a path Req 4.4 demotes to the suspicion band. Restating that as model_corroborated would claim a CPG arbitrated something it never saw, and rewriting every legacy finding to suspicion would be a product-visible regression nobody asked for. So `rung` defaults to suspicion and is raised only by ladder.at_rung(finding, verdict), which is the structural form of Req 5.4: nothing rises above suspicion without a Verdict behind it. The witness travels with the rung, because a rung without its evidence is just a louder assertion. It reaches markdown, JSON and SARIF. 3.2 is the judge, and it is defined as much by what it lacks. Three exhaustive outcomes: the model entails (report it, never call the model), the model contradicts (drop the claim, record why), the model cannot decide (report suspicion). The LLM gets one bounded typed question with tools=[] so it cannot search for the site, asked only about the residual the graph cannot settle -- whether a sanitizer on a REAL flow suffices. Its answer can only confirm a flow the CPG already found; it can never conjure one. Req 8.5 is asserted structurally: a test parses judge.py and fails on any AST name like k_runs/majority/vote/average/acceptance/floor/budget, any for/while loop, or an import from the deleted round machinery. That test first matched the word "votes" in the module's own docstring, which is why it now inspects code rather than text. Verified live against a real CPG, not only scripted rows: vuln rung=model_entailed request.args["name"] -> db.execute(...) fixed rung=suspicion no source reaches this sink model calls made: 0 Both sides arbitrated by the graph with the LLM never consulted -- the thing the prior architecture could not do at any price. 678 passing, mypy/ruff clean, gates byte-identical, zero orphaned modules. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- .kiro/specs/model-grounded-detection/tasks.md | 39 ++++- .../measurements/2026-09-08-module-audit.json | 25 ++- src/openultrasast/findings.py | 4 + src/openultrasast/model/audit.py | 1 + src/openultrasast/model/judge.py | 128 ++++++++++++++ src/openultrasast/model/ladder.py | 24 ++- src/openultrasast/reports.py | 3 + tests/test_model_judge.py | 159 ++++++++++++++++++ tests/test_model_ladder.py | 95 +++++++++++ 9 files changed, 471 insertions(+), 7 deletions(-) create mode 100644 src/openultrasast/model/judge.py create mode 100644 tests/test_model_judge.py create mode 100644 tests/test_model_ladder.py diff --git a/.kiro/specs/model-grounded-detection/tasks.md b/.kiro/specs/model-grounded-detection/tasks.md index e505ea5..8aa25b8 100644 --- a/.kiro/specs/model-grounded-detection/tasks.md +++ b/.kiro/specs/model-grounded-detection/tasks.md @@ -170,7 +170,7 @@ queries in group 4). The gates, `redaction.py`, `pairs.py` overlay/inventory pat ## Group 3 — The ladder and the judge -- [ ] 3.1 The evidence ladder as the verdict type +- [x] 3.1 The evidence ladder as the verdict type - `model/ladder.py`: `Rung` enum (suspicion/model_corroborated/model_entailed/execution_confirmed), `Verdict` (rung, family, witness, contradiction). Findings carry the rung in markdown, SARIF and JSON. - Observable: a finding at each rung round-trips through every output format with its rung intact; nothing @@ -178,7 +178,7 @@ queries in group 4). The gates, `redaction.py`, `pairs.py` overlay/inventory pat - _Requirements: 5.1, 5.3, 5.4_ - _Depends: 2.3_ -- [ ] 3.2 The judge: one bounded question, checked against the model +- [x] 3.2 The judge: one bounded question, checked against the model - `model/judge.py::judge(candidate, cpg, spec, *, client, model)` — if the model entails, report `model_entailed` with no LLM call; else ask the LLM one typed question (source expression / guard / family) with no tools; then check the answer against the CPG — contradiction drops the claim (records why), @@ -379,3 +379,38 @@ slice. The larger levers are the missing sinks and guard dominance. **Next**, in this order: 4.1 guard dominance over the tainted path; 4.2 the missing sinks (Django ORM, GraphQL, and the four families with no sink fact at all); then the two remaining safe-shape forms, argv-list and `shell=False`. + +## Implementation Notes — group 3 (2026-09-08) + +**3.1 — the rung is a new field beside `evidence_level`, not a rename of it.** The existing +`evidence_level` carries `static_corroboration`, emitted by the *flat IR overlay*. Req 4.4 demotes that path +to the suspicion band, so restating its value as `model_corroborated` would claim a CPG arbitrated something +it never saw, while rewriting every legacy finding to `suspicion` would be a product-visible regression +nobody asked for. So `StaticFinding.rung` is added, defaults to `suspicion`, and is raised *only* by +`ladder.at_rung(finding, verdict)` — which is the structural form of Req 5.4. The witness travels with the +rung, because a rung without the evidence that justifies it is just a louder assertion. The rung reaches +markdown, JSON (`asdict`, so automatically) and SARIF. `evidence_level` is now redundant in intent and should +be retired once the model layer covers every path; that is a follow-up, not a silent change. + +**3.2 — the judge, and what it does *not* contain.** Three outcomes, exhaustive: the model entails (report it, +**never call the model**), the model contradicts (drop the claim, record why), the model cannot decide (report +`suspicion`, honestly). The LLM is asked exactly one bounded, typed question, with `tools=[]` so it cannot +search for the site, and only about the residual the graph cannot settle — whether a sanitizer on a *real* +flow is sufficient. Its answer can only ever confirm a flow the CPG already found; it can never conjure one. + +Req 8.5 is asserted structurally rather than by prose: a test parses `judge.py` and fails if the AST contains +any name like `k_runs`, `majority`, `vote`, `average`, `acceptance`, `floor` or `budget`, or any `for`/`while` +loop at all, or an import from the deleted round machinery. The first draft of that test matched the word +"votes" inside the module's own docstring, which is the reason it now inspects code rather than text — the +docstring is allowed to name the machinery it replaced. + +**Verified live against a real CPG**, not only against scripted rows: + + vuln rung=model_entailed witness=request.args["name"] -> db.execute("select * from users ... + fixed rung=suspicion contradiction: the model resolved no flow: no source reaches this sink + model calls made: 0 + +Both sides arbitrated by the graph with the LLM never consulted — which is the entire point of the arbiter, +and the thing the prior architecture could not do at any price. + +678 tests passing, mypy and ruff clean, three gates byte-identical, zero orphaned modules. diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index e2f5a5b..5b6b2a0 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,13 +1,13 @@ { "counts": { - "load_bearing": 79, + "load_bearing": 80, "standalone_capability": 7 }, - "generated": "2026-09-08 (regenerated after group 2 added the cpg/ seam)", + "generated": "2026-09-08 (regenerated after group 3 added the ladder and the judge)", "kept_with_a_lever_stripped": { "improve/evolve.py + improve/validator.py": "the design's Removed list named these whole, but each is a mix: the mechanism lever (removed) and the deterministic rule-status improve loop that `ousast improve` still runs from benchmark evidence, which has no LLM in it and is not what this feature supersedes. Deleting the files would have taken a retained capability with them." }, - "modules": 86, + "modules": 87, "note": "Classification of every source module after the group-1 subtraction. load_bearing = reachable from a gate, the CLI, or named by a surviving requirement; standalone_capability = its own entry point or protocol, deliberately off the scan path; orphaned = no importer, no entry point, no surviving requirement -- removed, never carried.", "orphaned": [], "removed_in_this_pass": { @@ -128,6 +128,7 @@ "classification": "load_bearing", "importers": [ "cpg.backend", + "model.judge", "model.taint" ], "module": "cpg", @@ -136,6 +137,7 @@ { "classification": "load_bearing", "importers": [ + "model.judge", "model.taint" ], "module": "cpg.backend", @@ -165,6 +167,7 @@ "hunter_tools", "improve.evolve", "map_gate", + "model.ladder", "pairs", "rank", "regress.candidate", @@ -320,6 +323,8 @@ "classification": "load_bearing", "importers": [ "cli", + "model.candidates", + "model.judge", "model.specs", "model.taint" ], @@ -346,9 +351,16 @@ "module": "model.endpoint", "reason": "model-grounded-detection Req 8: the judge client, wired by model/judge.py in group 3" }, + { + "classification": "load_bearing", + "importers": [], + "module": "model.judge", + "reason": "model-grounded-detection Req 8: one bounded question, checked against the model" + }, { "classification": "load_bearing", "importers": [ + "model.judge", "model.taint" ], "module": "model.ladder", @@ -357,6 +369,7 @@ { "classification": "load_bearing", "importers": [ + "model.judge", "model.taint" ], "module": "model.specs", @@ -364,7 +377,9 @@ }, { "classification": "load_bearing", - "importers": [], + "importers": [ + "model.judge" + ], "module": "model.taint", "reason": "model-grounded-detection Req 6.1: taint reachability over the CPG, the flow-family arbiter" }, @@ -372,6 +387,7 @@ "classification": "load_bearing", "importers": [ "cli", + "model.candidates", "model.specs" ], "module": "model.taxonomy", @@ -496,6 +512,7 @@ "classification": "load_bearing", "importers": [ "harness", + "model.judge", "reports", "semantic.obligations.intent" ], diff --git a/src/openultrasast/findings.py b/src/openultrasast/findings.py index 16ec29b..7e7c00e 100644 --- a/src/openultrasast/findings.py +++ b/src/openultrasast/findings.py @@ -29,6 +29,10 @@ class StaticFinding: ranking_priority: float status: str = "enabled" proposed_snippet: str | None = None + # What ESTABLISHED this finding (model-grounded-detection Req 5). Defaults to `suspicion` and is raised + # only by `model.ladder.at_rung` with a Verdict behind it, so nothing can claim the model arbitrated a + # claim the model never saw. Distinct from `evidence_level`, which is the older overlay pipeline's word. + rung: str = "suspicion" SEVERITY_LABEL = {5: "critical", 4: "high", 3: "medium", 2: "low", 1: "low", 0: "info"} diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index 264cd70..c790415 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -33,6 +33,7 @@ SPEC_OWNED = { "model.audit": "model-grounded-detection Req 3: this classifier; run by the maintainer, not by the scan", "model.ladder": "model-grounded-detection Req 5: the evidence ladder every finding carries", "model.taint": "model-grounded-detection Req 6.1: taint reachability over the CPG, the flow-family arbiter", + "model.judge": "model-grounded-detection Req 8: one bounded question, checked against the model", "cpg": "model-grounded-detection Req 4: the CPG seam package", "cpg.backend": "model-grounded-detection Req 4.1/4.5: the single Joern subprocess boundary", "cpg.capability": "model-grounded-detection Req 4.2/4.3: the engine probe that degrades to suspicion", diff --git a/src/openultrasast/model/judge.py b/src/openultrasast/model/judge.py new file mode 100644 index 0000000..44050ae --- /dev/null +++ b/src/openultrasast/model/judge.py @@ -0,0 +1,128 @@ +"""The LLM proposes; the model disposes (model-grounded-detection, Req 8). + +This module is where the prior architecture's premise is inverted. There, the LLM was an oracle whose noise +had to be averaged over K runs, floored per family, and gated by a sign test. Here it answers one bounded, +typed question about a candidate it was *handed*, and its answer is checked against the CPG. Three outcomes, +and they are exhaustive: + + the model entails it report `model_entailed`, and never call the model at all + the model contradicts it drop the claim, recording the contradiction + the model cannot decide report `suspicion`, honestly + +Nothing here averages, votes, retries for agreement, or maintains a budget: a deterministic arbiter removes +the need for every one of those. The LLM is consulted only for the residual the graph cannot settle — whether +a sanitizer on a real flow is actually sufficient — and even that answer only ever *confirms* a flow the CPG +already found. It can never conjure one. +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Mapping +from typing import Any + +from ..cpg.backend import CpgResult +from ..redaction import redact_secrets +from .ladder import Rung, Verdict +from .specs import TaintSpec +from .taint import verdict as taint_verdict + +logger = logging.getLogger(__name__) + +QUESTION = """You are judging ONE candidate a static model already found. Do not look for other issues. + +A dataflow analysis resolved this path in {language} code: + + source: {source} + sink: {sink} (line {line}, function {function}) + a sanitizer appears on the path: {sanitized} + +The graph has established that the value reaches the sink. The only open question is whether the sanitizer on +that path actually makes it safe for the {family} weakness class. + +Answer ONLY with JSON: {{"vulnerable": true|false, "why": ""}}""" + + +def judge( + cpg: CpgResult, + spec: TaintSpec, + *, + function: str = "", + client: Any | None = None, + model: str = "", + parameter_sources: bool = False, +) -> Verdict: + """Arbitrate one candidate. Always returns a verdict; ``suspicion`` is a real answer, not a failure.""" + answer = taint_verdict(cpg, spec, function=function, parameter_sources=parameter_sources) + + if answer is not None and answer.rung is Rung.ENTAILED: + return answer # the graph decided; the model is not consulted + + if answer is None: + # No resolved flow. Whatever anyone claims about this site, the model saw nothing to support it. + return Verdict( + rung=Rung.SUSPICION, + family=spec.family, + contradiction="the model resolved no flow: no source reaches this sink", + ) + + # A real flow through a sanitizer. Whether that sanitizer suffices is a semantic judgement the graph + # cannot make, so it is the one thing the LLM is asked -- about evidence the model itself produced. + if client is None: + return Verdict( + rung=Rung.SUSPICION, + family=spec.family, + witness=answer.witness, + contradiction="no chat endpoint: the sanitizer's sufficiency was never judged", + ) + + said_vulnerable = _ask(client, model, spec, answer.witness, function) + if said_vulnerable is None: + return Verdict(rung=Rung.SUSPICION, family=spec.family, witness=answer.witness, + contradiction="the judge returned no usable answer") + if not said_vulnerable: + return Verdict(rung=Rung.SUSPICION, family=spec.family, witness=answer.witness, + contradiction="the judge held the sanitizer on this path sufficient") + return answer # CORROBORATED: a real flow the judge agrees is not neutralised + + +def _ask(client: Any, model: str, spec: TaintSpec, witness: str, function: str) -> bool | None: + """One question, one answer. No tools, no retries for agreement, no second opinion.""" + source, _, sink = witness.partition(" -> ") + line = "" + if "(line " in sink: + sink, _, tail = sink.partition(" (line ") + line = tail.split(",")[0].strip(") ") + prompt = redact_secrets( + QUESTION.format( + language=spec.language, source=source.strip(), sink=sink.strip(), line=line or "?", + function=function or "?", sanitized="yes", family=spec.family, + ) + ) + try: + response = client.complete( + model=model, + messages=[{"role": "user", "content": prompt}], + tools=[], # the candidate is given; the model never searches for it + json_object=True, + ) + except Exception as exc: # noqa: BLE001 -- an endpoint failure is a degradation, never a verdict + logger.warning("judge call failed: %s", exc) + return None + return _vulnerable(getattr(response, "content", None)) + + +def _vulnerable(content: object) -> bool | None: + if not isinstance(content, str) or not content.strip(): + return None + try: + payload = json.loads(content) + except json.JSONDecodeError: + return None + if isinstance(payload, Mapping) and isinstance(payload.get("vulnerable"), bool): + return bool(payload["vulnerable"]) + return None + + +__all__ = ["QUESTION", "judge"] diff --git a/src/openultrasast/model/ladder.py b/src/openultrasast/model/ladder.py index 1e85fb7..1b0854a 100644 --- a/src/openultrasast/model/ladder.py +++ b/src/openultrasast/model/ladder.py @@ -18,6 +18,12 @@ from __future__ import annotations from dataclasses import dataclass from enum import StrEnum +from typing import TYPE_CHECKING, TypeVar + +if TYPE_CHECKING: + from ..findings import StaticFinding + +_Finding = TypeVar("_Finding", bound="StaticFinding") class Rung(StrEnum): @@ -45,4 +51,20 @@ class Verdict: contradiction: str = "" # why an LLM claim was dropped, when it was -__all__ = ["Rung", "Verdict", "at_or_above"] +def at_rung(finding: _Finding, verdict: Verdict | None) -> _Finding: + """Return ``finding`` carrying ``verdict``'s rung and witness, or unchanged when there is no verdict. + + This is the only way a finding rises above ``suspicion`` (Req 5.4). The witness travels with the rung + because a rung without the evidence that justifies it is just a louder assertion. + """ + from dataclasses import replace + + if verdict is None or verdict.rung is Rung.SUSPICION: + return finding + rationale = finding.rationale + if verdict.witness and verdict.witness not in rationale: + rationale = f"{rationale} [{verdict.rung.value}: {verdict.witness}]".strip() + return replace(finding, rung=verdict.rung.value, rationale=rationale) + + +__all__ = ["Rung", "Verdict", "at_or_above", "at_rung"] diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index 5f2637c..cb9f1aa 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -46,6 +46,7 @@ def write_markdown_report( f"- Severity: `{finding.severity}`", f"- Confidence: `{finding.confidence}`", f"- Evidence level: `{finding.evidence_level}`", + f"- Rung: `{finding.rung}`", f"- Verification: `{verification.status if verification else 'not_run'}`", f"- Reachability: `{finding.reachability_status}`", f"- Conditions: `{', '.join(finding.reachability_conditions) or 'none'}`", @@ -218,6 +219,7 @@ def write_manifest( "line": finding.line, "severity": finding.severity, "evidence_level": finding.evidence_level, + "rung": finding.rung, "verification_status": verification_by_id[finding.finding_id].status if finding.finding_id in verification_by_id else "not_run", @@ -293,6 +295,7 @@ def _sarif_result( "severity": finding.severity, "confidence": finding.confidence, "evidence_level": finding.evidence_level, + "rung": finding.rung, "verification": asdict(verification) if verification else None, "reachability_status": finding.reachability_status, "reachability_conditions": finding.reachability_conditions, diff --git a/tests/test_model_judge.py b/tests/test_model_judge.py new file mode 100644 index 0000000..89da04c --- /dev/null +++ b/tests/test_model_judge.py @@ -0,0 +1,159 @@ +"""model-grounded-detection task 3.2: the LLM proposes, the model disposes (Req 8). + +The judge is where the prior architecture's whole premise is inverted. There is no K-run averaging, no noise +floor and no acceptance rule; there is one bounded, typed question per candidate, and the answer is CHECKED +against the CPG rather than voted on. The three outcomes are exhaustive: + + the model entails it -> report it, and never call the model at all + the model contradicts it -> drop the claim, and record the contradiction + the model cannot decide -> report `suspicion`, honestly + +Every test here runs with no network and a scripted client. +""" + +from __future__ import annotations + +from pathlib import Path + + +def _spec(): # type: ignore[no-untyped-def] + from openultrasast.model.specs import TaintSpec + + return TaintSpec(family="injection", language="python", sources=("request.args",), + sinks=("os.system",), sanitizers=("escape",)) + + +def _cpg(rows): # type: ignore[no-untyped-def] + from openultrasast.cpg.backend import CpgResult + + return CpgResult(cpg_path=Path("cpg.bin"), run=lambda q, p: rows) + + +ENTAILING_ROW = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", + "source": "request.args['c']", "sanitized": False, "length": 3}] + + +class _Client: + """A scripted chat client that records whether it was called at all.""" + + def __init__(self, answer: str = '{"vulnerable": true, "family": "injection"}') -> None: + self.calls: list[object] = [] + self._answer = answer + + def complete(self, **kwargs: object): # type: ignore[no-untyped-def] + from openultrasast.tool_hunter import ChatResponse + + self.calls.append(kwargs) + return ChatResponse(content=self._answer, tool_calls=[], reasoning=None) + + +def test_an_entailed_candidate_costs_no_model_call() -> None: + """The point of the arbiter: where the graph decides, the LLM is not consulted at all.""" + from openultrasast.model.judge import judge + from openultrasast.model.ladder import Rung + + client = _Client() + answer = judge(_cpg(ENTAILING_ROW), _spec(), function="run", client=client, model="m") + assert answer.rung is Rung.ENTAILED + assert client.calls == [], "an entailed candidate must not reach the model" + + +def test_a_claim_the_model_contradicts_is_dropped_with_its_reason() -> None: + """Req 8.2. The LLM says vulnerable; the CPG shows no flow reaches the sink. The graph wins.""" + from openultrasast.model.judge import judge + from openultrasast.model.ladder import Rung + + client = _Client('{"vulnerable": true, "family": "injection"}') + answer = judge(_cpg([]), _spec(), function="run", client=client, model="m") + assert answer.rung is Rung.SUSPICION + assert answer.contradiction, "a dropped claim must say why it was dropped" + assert "no flow" in answer.contradiction.lower() or "not reach" in answer.contradiction.lower() + + +def test_a_claim_the_model_supports_advances_to_corroborated() -> None: + """Req 8.3: a sanitized path is a real flow whose sufficiency only the LLM can judge.""" + from openultrasast.model.judge import judge + from openultrasast.model.ladder import Rung + + rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", + "source": "request.args['c']", "sanitized": True, "length": 3}] + client = _Client('{"vulnerable": true, "family": "injection"}') + answer = judge(_cpg(rows), _spec(), function="run", client=client, model="m") + assert answer.rung is Rung.CORROBORATED + assert len(client.calls) == 1, "exactly one bounded question, never a loop" + + +def test_the_llm_declining_leaves_a_corroborated_flow_at_suspicion() -> None: + """The model saw a flow but the LLM says the sanitizer suffices; the claim does not advance.""" + from openultrasast.model.judge import judge + from openultrasast.model.ladder import Rung + + rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", + "source": "request.args['c']", "sanitized": True, "length": 3}] + client = _Client('{"vulnerable": false, "family": "injection"}') + answer = judge(_cpg(rows), _spec(), function="run", client=client, model="m") + assert answer.rung is Rung.SUSPICION + + +def test_without_a_client_the_verdict_still_stands_on_its_own() -> None: + """Req 11.3: the endpoint is optional. The model layer does not need it to entail.""" + from openultrasast.model.judge import judge + from openultrasast.model.ladder import Rung + + assert judge(_cpg(ENTAILING_ROW), _spec(), function="run", client=None, model="").rung is Rung.ENTAILED + sanitized = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", + "source": "request.args['c']", "sanitized": True, "length": 3}] + answer = judge(_cpg(sanitized), _spec(), function="run", client=None, model="") + assert answer.rung is Rung.SUSPICION and "endpoint" in answer.contradiction.lower() + + +def test_the_question_is_bounded_typed_and_offers_no_tools() -> None: + """Req 8.1: the LLM answers one question about a site it was handed; it never searches for the site.""" + from openultrasast.model.judge import judge + + rows = [{"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", + "source": "request.args['c']", "sanitized": True, "length": 3}] + client = _Client() + judge(_cpg(rows), _spec(), function="run", client=client, model="m") + call = client.calls[0] + assert call["tools"] == [], "no tools: the candidate is given, not searched for" + assert call["json_object"] is True, "a typed answer, not prose" + text = " ".join(str(m.get("content", "")) for m in call["messages"]) # type: ignore[union-attr] + assert "os.system(cmd)" in text and "request.args['c']" in text, "the question carries the model's own evidence" + + +def test_the_prompt_is_redacted() -> None: + """Req 11.4: every prompt passes through redaction.""" + from openultrasast.model.judge import judge + + rows = [{"sink": "connect(pw)", "sinkLine": "4", "sinkMethod": "run", + "source": "AKIA" + "IOSFODNN7" + "EXAMPLE", "sanitized": True, "length": 2}] + client = _Client() + judge(_cpg(rows), _spec(), function="run", client=client, model="m") + text = " ".join(str(m.get("content", "")) for m in client.calls[0]["messages"]) # type: ignore[index,union-attr] + assert "AKIAIOSFODNN7EXAMPLE" not in text + + +def test_there_is_no_k_run_averaging_anywhere_in_the_judge() -> None: + """Req 8.5, asserted structurally: the noise architecture must not grow back.""" + import ast + + source = Path("src/openultrasast/model/judge.py").read_text() + tree = ast.parse(source) + + # Check CODE, not prose: the docstring is allowed to name the machinery it replaced. + names = { + node.id for node in ast.walk(tree) if isinstance(node, ast.Name) + } | { + node.attr for node in ast.walk(tree) if isinstance(node, ast.Attribute) + } | { + node.name for node in ast.walk(tree) if isinstance(node, ast.FunctionDef) + } + for banned in ("k_runs", "majority", "vote", "votes", "average", "mean", "acceptance", "floor", "budget"): + assert banned not in names, f"{banned} has no place in the judge" + + # No loop over runs anywhere in the module. + assert not [n for n in ast.walk(tree) if isinstance(n, ast.For | ast.While)], "the judge asks once" + + imported = {n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module} + assert not any("rounds" in m or "acceptance" in m for m in imported) diff --git a/tests/test_model_ladder.py b/tests/test_model_ladder.py new file mode 100644 index 0000000..f9e4fb1 --- /dev/null +++ b/tests/test_model_ladder.py @@ -0,0 +1,95 @@ +"""model-grounded-detection task 3.1: the evidence ladder is the verdict type (Req 5). + +The rung says what ESTABLISHED a finding, so a deterministic model result is never confused with an LLM +suspicion. The integrity constraint is Req 5.4: nothing reaches a rung above `suspicion` without a `Verdict`, +and a `Verdict` above `suspicion` is only ever produced by the CPG with no model call in the establishing step. +""" + +from __future__ import annotations + +import json +from pathlib import Path + + +def _finding(**kw: object): # type: ignore[no-untyped-def] + from openultrasast.findings import StaticFinding + + base = dict( + finding_id="r1:app.py:1", path="app.py", title="t", severity="high", confidence="medium", + evidence_level="static_corroboration", rationale="why", line=1, function_name="run", + reachability_status="unknown", reachability_evidence=[], reachability_conditions=[], tags=[], + ranking_priority=1.0, + ) + base.update(kw) + return StaticFinding(**base) # type: ignore[arg-type] + + +def test_the_rungs_are_ordered_and_comparable() -> None: + from openultrasast.model.ladder import Rung, at_or_above + + assert at_or_above(Rung.ENTAILED, Rung.CORROBORATED) + assert at_or_above(Rung.CORROBORATED, Rung.SUSPICION) + assert not at_or_above(Rung.SUSPICION, Rung.CORROBORATED) + assert at_or_above(Rung.EXECUTION_CONFIRMED, Rung.ENTAILED) + + +def test_a_finding_defaults_to_suspicion() -> None: + """Req 5.4: a claim nothing has arbitrated is a suspicion, whatever produced it.""" + from openultrasast.model.ladder import Rung + + assert _finding().rung == Rung.SUSPICION + + +def test_only_a_verdict_raises_a_finding_above_suspicion() -> None: + from openultrasast.model.ladder import Rung, Verdict, at_rung + + finding = _finding() + verdict = Verdict(rung=Rung.ENTAILED, family="injection", witness="src -> sink") + raised = at_rung(finding, verdict) + assert raised.rung == Rung.ENTAILED + assert "src -> sink" in raised.rationale, "the witness travels with the rung it justifies" + assert raised.finding_id == finding.finding_id and raised.line == finding.line + + +def test_a_none_verdict_leaves_the_finding_at_suspicion() -> None: + from openultrasast.model.ladder import Rung, at_rung + + assert at_rung(_finding(), None).rung == Rung.SUSPICION + + +def test_the_legacy_evidence_level_is_left_alone() -> None: + """`static_corroboration` is the *flat IR overlay's* word and Req 4.4 demotes that path to the suspicion + band. Restating it as `model_corroborated` would claim a CPG arbitrated something it never saw.""" + from openultrasast.model.ladder import Rung, Verdict, at_rung + + raised = at_rung(_finding(), Verdict(rung=Rung.CORROBORATED, family="injection", witness="w")) + assert raised.evidence_level == "static_corroboration", "the legacy field keeps its own meaning" + assert raised.rung == Rung.CORROBORATED + + +def test_the_rung_reaches_markdown_json_and_sarif(tmp_path: Path) -> None: + """Req 5.3: every output format carries the rung.""" + from openultrasast.findings import write_findings + from openultrasast.model.ladder import Rung, Verdict, at_rung + from openultrasast.reports import write_markdown_report, write_sarif_report + + finding = at_rung(_finding(), Verdict(rung=Rung.ENTAILED, family="injection", witness="src -> sink")) + md, js, sarif = tmp_path / "r.md", tmp_path / "r.json", tmp_path / "r.sarif" + write_markdown_report([finding], md) + write_findings([finding], js) + write_sarif_report([finding], [], sarif) + + assert "model_entailed" in md.read_text() + assert json.loads(js.read_text())["findings"][0]["rung"] == "model_entailed" + props = json.loads(sarif.read_text())["runs"][0]["results"][0]["properties"] + assert props["rung"] == "model_entailed" + + +def test_every_rung_round_trips_through_json(tmp_path: Path) -> None: + from openultrasast.findings import write_findings + from openultrasast.model.ladder import Rung + + for rung in Rung: + out = tmp_path / f"{rung.value}.json" + write_findings([_finding(rung=rung)], out) + assert json.loads(out.read_text())["findings"][0]["rung"] == rung.value