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