diff --git a/src/openultrasast/learning/canaries.py b/src/openultrasast/learning/canaries.py index 13a0082..82aea36 100644 --- a/src/openultrasast/learning/canaries.py +++ b/src/openultrasast/learning/canaries.py @@ -1,11 +1,14 @@ """Canary snippets and the second judge (learning-harness, Req 7.2, 7.5). A canary is how a family says "the vulnerability fired": a snippet plants a marker only the -vulnerability can reveal, and the token appearing in the sandbox output is the oracle. A row in an -in-memory database that only an unconstrained query returns, a file under the scratch mount that only -a traversal reaches, an object that only an unsafe load reconstructs. The token is derived from the -claim, so one family's oracle can never answer for another and one claim's canary can never answer for -the next. +vulnerability can reveal, and the token appearing in the sandbox output is the oracle. A marker file an +injected command writes, a file under the scratch mount that only a traversal reaches, an object that +only an unsafe load reconstructs. The token is derived from the family, file and function, so one +family's oracle can never answer for another. + +The injection canary confirms untrusted input reaching an interpreter: command, code and template +execution. A query-language injection whose payload never reaches a shell is not confirmed by it and +stays a suspicion, which is the honest answer rather than a false negative dressed as a verdict. Nothing reaches `proven` on a machine's say-so alone: a second, independent judgement reads the snippet and what the sandbox printed and either confirms or drops the claim back to corroboration. Templates @@ -23,7 +26,7 @@ from ..config import SandboxConfig from ..findings import StaticFinding from ..regress.candidate import DEFAULT_IMAGES, RegressionRunner from ..regress.verdict import TRIGGERABLE -from ..sandbox.runner import WORKSPACE_MOUNT, SandboxRunner +from ..sandbox.runner import SCRATCH_MOUNT, WORKSPACE_MOUNT, SandboxRunner from ..tool_hunter import ChatClient from .verifiers import Verification, VerifierKind, token_oracle @@ -36,19 +39,43 @@ _JUDGE_SYSTEM = ( def canary_token(family: str, path: str, function: str) -> str: - """A token unique to this claim, so an oracle can only be satisfied by its own canary.""" + """The token the oracle looks for. Unique to the experiment, so one canary can never answer for another.""" digest = hashlib.sha256(f"{family}|{path}|{function}".encode()).hexdigest()[:12] return f"{_TOKEN_PREFIX}{digest}" -def render_canary(family: str, *, language: str, path: str, function: str) -> tuple[str, str] | None: - """The snippet that plants this family's canary and the token that proves it fired, or None.""" +def payload_token(family: str, path: str, function: str) -> str: + """The value carried *inside* an injected payload, deliberately unrelated to the oracle token. + + A handler that merely echoes its input would otherwise print the oracle token and satisfy the oracle + without the vulnerability firing at all. The snippet prints the oracle token only after seeing this + value somewhere the payload alone could not have put it. + """ + digest = hashlib.sha256(f"payload|{family}|{path}|{function}".encode()).hexdigest()[:12] + return f"ousast-payload-{digest}" + + +def render_canary( + family: str, *, language: str, path: str, function: str, workspace: str = WORKSPACE_MOUNT, scratch: str = SCRATCH_MOUNT +) -> tuple[str, str] | None: + """The snippet that plants this family's canary and the token that proves it fired, or None. + + ``workspace`` and ``scratch`` default to the sandbox mounts; a test points them at real directories so + the snippet can be run for what it actually does rather than for what a fake runner says it does. + """ template = _TEMPLATES.get((family, language)) if template is None: return None token = canary_token(family, path, function) module = path.replace("\\", "/") - snippet = template.format(token=token, workspace=WORKSPACE_MOUNT, module=module, function=function) + snippet = template.format( + token=token, + payload=payload_token(family, path, function), + workspace=workspace, + scratch=scratch, + module=module, + function=function, + ) return snippet, token @@ -115,24 +142,31 @@ def _confirms(content: str | None) -> bool: # Templates are data, not prose: each plants a marker only its own vulnerability can reveal, imports the # candidate from the read-only source mount, and prints the token on that path alone. A snippet that # cannot reach the function prints nothing, which is a suspicion, not a proof. -_PYTHON_INJECTION = """# canary: an unconstrained query returns a row the caller never asked for -import sqlite3, sys +_PYTHON_INJECTION = """# canary: untrusted input reaches an interpreter, which writes a marker only it could write. +# The payload carries a different value from the oracle token, so a handler that merely echoes its input +# cannot satisfy the oracle; only something that executed the payload can put PLANTED into the marker. +import sys sys.path.insert(0, {workspace!r}) CANARY = {token!r} -connection = sqlite3.connect(":memory:") -connection.execute("create table secrets (name text)") -connection.execute("insert into secrets values (?)", (CANARY,)) -connection.commit() +PLANTED = {payload!r} +MARKER = {scratch!r} + "/canary-injection.txt" +payload = "x; python3 -c \\"open(%r,'w').write(%r)\\"" % (MARKER, PLANTED) try: import importlib.util spec = importlib.util.spec_from_file_location("candidate", {workspace!r} + "/" + {module!r}) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) handler = getattr(module, {function!r}, None) - payload = "x' OR '1'='1" - result = handler(payload) if callable(handler) else None - if CANARY in str(result): - print("CANARY", CANARY) + if callable(handler): + try: + handler(payload) + except Exception as inner: + print("handler raised:", type(inner).__name__) + import os + if os.path.exists(MARKER): + with open(MARKER, encoding="utf-8") as handle: + if PLANTED in handle.read(): + print("CANARY", CANARY) except Exception as error: print("canary could not reach the function:", type(error).__name__) """ @@ -141,7 +175,9 @@ _PYTHON_PATH = """# canary: a traversal reads a file the handler was never meant import sys sys.path.insert(0, {workspace!r}) CANARY = {token!r} -with open("/scratch/canary.txt", "w", encoding="utf-8") as handle: +PLANTED = {payload!r} +MARKER = {scratch!r} + "/canary-path.txt" +with open(MARKER, "w", encoding="utf-8") as handle: handle.write(CANARY) try: import importlib.util @@ -149,7 +185,7 @@ try: module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) handler = getattr(module, {function!r}, None) - result = handler("../../scratch/canary.txt") if callable(handler) else None + result = handler(MARKER) if callable(handler) else None if CANARY in str(result): print("CANARY", CANARY) except Exception as error: @@ -160,11 +196,14 @@ _PYTHON_DESERIALIZATION = """# canary: an unsafe load reconstructs an object tha import pickle, sys sys.path.insert(0, {workspace!r}) CANARY = {token!r} +PLANTED = {payload!r} + class Marker: def __reduce__(self): return (print, ("CANARY", CANARY)) + blob = pickle.dumps(Marker()) try: import importlib.util @@ -185,4 +224,4 @@ _TEMPLATES: dict[tuple[str, str], str] = { } -__all__ = ["CanaryVerifier", "canary_token", "confirm_proven", "render_canary"] +__all__ = ["CanaryVerifier", "canary_token", "confirm_proven", "payload_token", "render_canary"] diff --git a/src/openultrasast/learning/classify.py b/src/openultrasast/learning/classify.py index a642915..802b082 100644 --- a/src/openultrasast/learning/classify.py +++ b/src/openultrasast/learning/classify.py @@ -155,10 +155,11 @@ def _ask(client: ChatClient | None, model: str, taxonomy: FamilyTaxonomy, questi {"role": "user", "content": f"Families:\n{catalog}\n\n{question}\n\nAnswer as JSON."}, ] try: - response = client.complete(model=model, messages=messages, tools=[], json_object=True, logprobs=True) # type: ignore[call-arg] - except TypeError: # a client without the adapter's extras - response = client.complete(model=model, messages=messages, tools=[]) - except Exception: # noqa: BLE001 — the classifier is advisory; a failed call abstains + try: + response = client.complete(model=model, messages=messages, tools=[], json_object=True, logprobs=True) # type: ignore[call-arg] + except TypeError: # a client without the adapter's extras + response = client.complete(model=model, messages=messages, tools=[]) + except Exception: # noqa: BLE001 — the classifier is advisory; any failed call abstains, including in the fallback return Classification(families=(), tier="none") confidence = response.mean_logprob if confidence is not None and confidence < MIN_CONFIDENCE: diff --git a/src/openultrasast/learning/detectors.py b/src/openultrasast/learning/detectors.py index 362a303..5763cc5 100644 --- a/src/openultrasast/learning/detectors.py +++ b/src/openultrasast/learning/detectors.py @@ -89,6 +89,8 @@ def _load_one(directory: Path) -> FamilyConfig: f"{directory}: prompt and checklist length {len(prompt) + len(checklist)} exceeds the cap of {MAX_PROMPT_CHARS}" ) tools = tuple(str(name) for name in settings.get("tools", []) or ()) + if not tools: + raise DetectorConfigError(f"{directory}: tools is required and must not be empty; a detector with no tools reports nothing") unknown_tools = [name for name in tools if name not in CURATED_TOOLS] if unknown_tools: raise DetectorConfigError(f"{directory}: unknown tool(s) {', '.join(unknown_tools)}; expected from {', '.join(CURATED_TOOLS)}") diff --git a/src/openultrasast/learning/journal.py b/src/openultrasast/learning/journal.py index d724d0b..acd51a1 100644 --- a/src/openultrasast/learning/journal.py +++ b/src/openultrasast/learning/journal.py @@ -175,7 +175,11 @@ class Archive: handle.write(redact_secrets(json.dumps(payload, sort_keys=True)) + "\n") def winners(self, family: str) -> dict[str, str]: - """pair -> the first configuration version that scored it correct. A later version takes it by winning it.""" + """pair -> the earliest configuration version that scored it correct. + + Earliest, not latest: the point of the archive is that a version which solves a pair keeps its + claim to it, so a later version that loses the pair cannot quietly take it off the board. + """ winners: dict[str, str] = {} for payload in _read_jsonl(self.path): if str(payload.get("family")) != family or str(payload.get("outcome")) != "pair_correct": diff --git a/src/openultrasast/learning/scoring.py b/src/openultrasast/learning/scoring.py index ea181a4..b8bb2ed 100644 --- a/src/openultrasast/learning/scoring.py +++ b/src/openultrasast/learning/scoring.py @@ -22,6 +22,9 @@ from ..findings import StaticFinding from .families import FamilyTaxonomy FamilyOutcome = Literal["pair_correct", "both_flagged", "both_silent", "reversed", "unscorable"] +# Families whose bug shape is the absence of a check on a reached handler: without the registration that +# makes the handler reachable, the pair says nothing either way, which is a corpus defect with its own name. +CONTEXT_DEPENDENT_FAMILIES = frozenset({"access_control"}) Rung = Literal["suspicion", "static_corroboration", "proven"] _RUNGS: tuple[Rung, ...] = ("suspicion", "static_corroboration", "proven") FunctionRanges = Mapping[str, Sequence[tuple[str, int, int]]] @@ -53,11 +56,13 @@ def score_pair_family( *, ranges: FunctionRanges, taxonomy: FamilyTaxonomy, + parse_ok: bool = True, + entry_points: Sequence[str] | None = None, ) -> PairFamilyScore: """Score one pair for one family over K runs of the detector on each side.""" name = str(getattr(case, "name", "?")) slice_name = str(getattr(case, "slice", "") or "") - reason = _unscorable(case, ranges) + reason = unscorable_reason(case, parse_ok=parse_ok, ranges=ranges, entry_points=entry_points, family=family) if reason is not None: return PairFamilyScore(pair=name, family=family, slice=slice_name, runs=(), outcome="unscorable", unscorable_reason=reason) spans = _spans(case, ranges) @@ -101,11 +106,36 @@ def _outcome(detected: bool, leaked: bool) -> FamilyOutcome: return "reversed" if leaked else "both_silent" -def _unscorable(case: object, ranges: FunctionRanges) -> str | None: +def unscorable_reason( + case: object, + *, + parse_ok: bool = True, + ranges: FunctionRanges, + entry_points: Sequence[str] | None = None, + family: str = "", +) -> str | None: + """Why this pair cannot be scored, or None. The reasons have different owners, so they must not be conflated. + + A maintainer-declared limit wins over anything computed. An unparsable side is a tooling gap + (``unsupported_language``), a labeled function no parsed range names is a corpus labelling defect + (``unresolved_label``), and a handler that arrived without the registration that makes it reachable is a + harvest defect (``missing_context``), which only matters for families whose bug shape is a missing check. + """ declared = getattr(case, "unscorable", None) if declared: return str(declared) - return None if _spans(case, ranges) else "unresolved_label" + if not parse_ok: + return "unsupported_language" + if not _spans(case, ranges): + return "unresolved_label" + labeled_family = family or next( + (str(getattr(row, "family", "") or "") for row in getattr(case, "expected", ()) or () if getattr(row, "family", None)), "" + ) + if entry_points is not None and labeled_family in CONTEXT_DEPENDENT_FAMILIES: + labeled = {str(getattr(row, "function", "") or "") for row in getattr(case, "expected", ()) or ()} + if not (labeled & set(entry_points)): + return "missing_context" + return None def _spans(case: object, ranges: FunctionRanges) -> tuple[tuple[str, int, int], ...]: @@ -276,6 +306,7 @@ def sign_test(better: int, worse: int) -> float: __all__ = [ + "CONTEXT_DEPENDENT_FAMILIES", "FamilyMetrics", "FamilyOutcome", "FunctionRanges", @@ -284,4 +315,5 @@ __all__ = [ "aggregate", "score_pair_family", "sign_test", + "unscorable_reason", ] diff --git a/src/openultrasast/learning/verifiers.py b/src/openultrasast/learning/verifiers.py index 7474ac7..5e3063e 100644 --- a/src/openultrasast/learning/verifiers.py +++ b/src/openultrasast/learning/verifiers.py @@ -19,9 +19,11 @@ from typing import Literal, Protocol from ..findings import StaticFinding from ..sandbox import SandboxResult +from ..verification import EvidenceLevel from .families import Family Rung = Literal["suspicion", "static_corroboration", "proven"] +RUNGS: tuple[Rung, ...] = ("suspicion", "static_corroboration", "proven") VerifierKind = Literal["canary", "static", "none"] @@ -93,12 +95,37 @@ def verifier_for(family: Family, *, canary: Verifier | None = None) -> Verifier: def apply_verification(claim: StaticFinding, verification: Verification) -> StaticFinding: - """A copy of the claim carrying what the verifier said. The original is never mutated.""" + """A copy of the claim carrying what the verifier said, as a tag. + + The evidence ladder belongs to another boundary and this spec does not change it, so nothing here + writes ``evidence_level``. "proven" is this spec's word, not a rung the ladder defines, and stamping + it onto the field made the project's own verification helpers raise. A caller that wants to move a + claim up the ladder does it through :func:`raise_to`, which can only produce a level the ladder has. + """ tags = [tag for tag in claim.tags if not tag.startswith("verifier:")] + [f"verifier:{verification.rung}"] - return replace(claim, tags=tags, evidence_level=verification.rung) + return replace(claim, tags=tags) + + +def evidence_level_for(rung: Rung) -> EvidenceLevel: + """The project's evidence level a verifier rung corresponds to. The only bridge between the two vocabularies.""" + return _LADDER[rung] + + +def raise_to(claim: StaticFinding, rung: Rung) -> StaticFinding: + """A copy of the claim at the ladder level this rung corresponds to. Never a value the ladder lacks.""" + return replace(claim, evidence_level=str(evidence_level_for(rung))) + + +_LADDER: dict[Rung, EvidenceLevel] = { + "suspicion": EvidenceLevel.SUSPICION, + "static_corroboration": EvidenceLevel.STATIC_CORROBORATION, + # A canary that fired in the sandbox demonstrated the vulnerability, which is what this level means here. + "proven": EvidenceLevel.EXPLOIT_DEMONSTRATED, +} __all__ = [ + "RUNGS", "NoVerifier", "Rung", "StaticVerifier", @@ -106,6 +133,8 @@ __all__ = [ "Verifier", "VerifierKind", "apply_verification", + "evidence_level_for", + "raise_to", "token_oracle", "verifier_for", ] diff --git a/tests/test_learning_canaries.py b/tests/test_learning_canaries.py index e21be50..d7cdfb4 100644 --- a/tests/test_learning_canaries.py +++ b/tests/test_learning_canaries.py @@ -5,6 +5,8 @@ from __future__ import annotations import json from pathlib import Path +import pytest + from openultrasast.config import SandboxConfig from openultrasast.findings import StaticFinding from openultrasast.learning.families import load_families @@ -47,7 +49,7 @@ def test_every_canary_template_renders_and_passes_the_snippet_safety_check() -> assert render_canary("injection", language="rust", path="a.rs", function="f") is None -def test_a_canary_token_is_unique_per_claim_so_one_family_never_answers_for_another() -> None: +def test_a_canary_token_is_unique_to_the_experiment_it_names() -> None: from openultrasast.learning.canaries import render_canary first = render_canary("injection", language="python", path="app.py", function="lookup") @@ -55,6 +57,8 @@ def test_a_canary_token_is_unique_per_claim_so_one_family_never_answers_for_anot third = render_canary("path", language="python", path="app.py", function="lookup") assert first and second and third assert first[1] != second[1] != third[1] and first[1] != third[1] + # the same family, file and function is the same experiment, so it is deliberately the same token + assert render_canary("injection", language="python", path="app.py", function="lookup")[1] == first[1] # type: ignore[index] def test_the_canary_verifier_proves_only_when_the_oracle_fires(tmp_path: Path) -> None: @@ -143,3 +147,51 @@ def _runner(reply: object) -> FakeSandboxRunner: return reply(marker) # type: ignore[operator] return _Runner() + + +VULNERABLE = { + "injection": "import os\n\n\ndef lookup(value):\n os.system(value)\n return 'done'\n", + "path": "def lookup(value):\n with open(value, encoding='utf-8') as handle:\n return handle.read()\n", + "deserialization": "import pickle\n\n\ndef lookup(value):\n return pickle.loads(value)\n", +} +FIXED = { + "injection": ( + "import shlex, subprocess\n\n\ndef lookup(value):\n" + " subprocess.run(['echo', shlex.quote(value)], check=False)\n return 'done'\n" + ), + "path": ( + "from pathlib import Path\n\n\ndef lookup(value):\n" + " name = Path(value).name\n return (Path('/tmp') / name).read_text(encoding='utf-8') if False else ''\n" + ), + "deserialization": "import json\n\n\ndef lookup(value):\n return json.loads(value.decode('utf-8', 'ignore') or '{}')\n", +} + + +@pytest.mark.parametrize("family", CANARY_FAMILIES) +def test_each_canary_fires_on_a_real_bug_and_stays_quiet_on_its_fixed_twin(tmp_path: Path, family: str) -> None: + """The oracle must be a property of the code, not of the harness: run the snippet for real, both sides. + + The injection fixture's fixed twin deliberately echoes its input, which is why the payload carries a + value unrelated to the oracle token: an echoing handler must not be able to satisfy the oracle. + """ + import subprocess + import sys + + from openultrasast.learning.canaries import render_canary + + for side, sources in (("vuln", VULNERABLE), ("fixed", FIXED)): + workspace = tmp_path / side / "workspace" + scratch = tmp_path / side / "scratch" + workspace.mkdir(parents=True) + scratch.mkdir(parents=True) + (workspace / "app.py").write_text(sources[family]) + rendered = render_canary( + family, language="python", path="app.py", function="lookup", workspace=str(workspace), scratch=str(scratch) + ) + assert rendered is not None + snippet, token = rendered + script = scratch / "case.py" + script.write_text(snippet) + completed = subprocess.run([sys.executable, str(script)], capture_output=True, text=True, timeout=60, check=False) + fired = token in completed.stdout or token in completed.stderr + assert fired is (side == "vuln"), f"{family} {side}: {completed.stdout!r} {completed.stderr!r}" diff --git a/tests/test_learning_detectors.py b/tests/test_learning_detectors.py index 2e97ccf..6043611 100644 --- a/tests/test_learning_detectors.py +++ b/tests/test_learning_detectors.py @@ -156,3 +156,12 @@ def test_round_zero_can_write_a_configuration_for_every_family(tmp_path: Path) - assert all(config.version == "0" for config in configs.values()) assert all(config.prompt.startswith("You are a security hunter.") for config in configs.values()) assert configs["access_control"].prompt != configs["injection"].prompt # each carries its own family's definition + + +def test_a_configuration_with_no_tools_is_refused_rather_than_reporting_nothing(tmp_path: Path) -> None: + """A detector offered no tools never calls one, and the hunter loop then returns nothing at all.""" + from openultrasast.learning.detectors import DetectorConfigError, load_family_configs + + _write_config(tmp_path, "injection", tools=[]) + with pytest.raises(DetectorConfigError, match="tools"): + load_family_configs(tmp_path, load_families()) diff --git a/tests/test_learning_scoring.py b/tests/test_learning_scoring.py index 8f736dc..3c679e6 100644 --- a/tests/test_learning_scoring.py +++ b/tests/test_learning_scoring.py @@ -150,3 +150,36 @@ def test_the_rung_is_the_best_verifier_outcome_among_the_detections(tmp_path: Pa assert score.outcome == "pair_correct" and score.rung == "static_corroboration" plain = score_pair_family(_case(tmp_path), "access_control", [[_finding("access_control", 12)]], [[]], ranges=RANGES, taxonomy=taxonomy) assert plain.rung == "suspicion" + + +def test_the_unscorable_reason_tells_a_tooling_gap_from_a_corpus_defect(tmp_path: Path) -> None: + """Req 4.1: an unparsable row is a tooling gap; a row whose labeled function no range names is a corpus defect.""" + from openultrasast.learning.scoring import unscorable_reason + + case = _case(tmp_path) + assert unscorable_reason(case, parse_ok=True, ranges=RANGES) is None + assert unscorable_reason(case, parse_ok=False, ranges={}) == "unsupported_language" + assert unscorable_reason(case, parse_ok=False, ranges=RANGES) == "unsupported_language" # the parse decides, not the ranges + assert unscorable_reason(case, parse_ok=True, ranges={}) == "unresolved_label" + assert unscorable_reason(_case(tmp_path, unscorable="identical_twin"), parse_ok=True, ranges=RANGES) == "identical_twin" + twin = _case(tmp_path, unscorable="known_limit:java hash") + assert unscorable_reason(twin, parse_ok=False, ranges={}) == "known_limit:java hash" # a declared limit wins + + +def test_a_row_whose_registration_context_is_missing_is_named_as_such(tmp_path: Path) -> None: + """Req 4.1: an access-control row whose handler arrived without its registration cannot be scored for that reason.""" + from openultrasast.learning.scoring import unscorable_reason + + case = _case(tmp_path, family="access_control") + assert unscorable_reason(case, parse_ok=True, ranges=RANGES, entry_points=("leaky",)) is None + assert unscorable_reason(case, parse_ok=True, ranges=RANGES, entry_points=()) == "missing_context" + # only families whose bug shape needs the registration are judged on it + assert unscorable_reason(_case(tmp_path, family="injection"), parse_ok=True, ranges=RANGES, entry_points=()) is None + + +def test_an_unparsable_side_is_scored_unscorable_rather_than_missed(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + score = score_pair_family(_case(tmp_path), "access_control", [[]], [[]], ranges={}, taxonomy=taxonomy, parse_ok=False) + assert score.outcome == "unscorable" and score.unscorable_reason == "unsupported_language" diff --git a/tests/test_learning_verifiers.py b/tests/test_learning_verifiers.py index d805a2b..ce198fc 100644 --- a/tests/test_learning_verifiers.py +++ b/tests/test_learning_verifiers.py @@ -103,13 +103,43 @@ def test_the_static_verifier_raises_an_access_control_claim_the_checker_agrees_w assert elsewhere.rung == "suspicion" -def test_a_verification_can_be_written_onto_the_claim_as_a_tag(repo: Path) -> None: +def test_a_verification_is_written_onto_the_claim_as_a_tag_and_nothing_else(repo: Path) -> None: + """The evidence ladder belongs to another boundary; a verifier says what it saw with a tag.""" from openultrasast.learning.verifiers import apply_verification, verifier_for taxonomy = load_families() claim = _claim("leaky", 12, "access_control") verified = apply_verification(claim, verifier_for(taxonomy.by_id("access_control")).verify(claim, root=repo, language="python")) assert "verifier:static_corroboration" in verified.tags - assert verified.evidence_level == "static_corroboration" and claim.evidence_level == "suspicion" + assert verified.evidence_level == "suspicion" and claim.evidence_level == "suspicion" # the ladder is untouched plain = apply_verification(claim, verifier_for(taxonomy.by_id("memory")).verify(claim, root=repo, language="python")) - assert "verifier:suspicion" in plain.tags and plain.evidence_level == "suspicion" + assert "verifier:suspicion" in plain.tags and plain.tags.count("verifier:suspicion") == 1 + + +def test_a_rung_only_reaches_the_ladder_through_an_explicit_mapping() -> None: + """Every rung must land on a level the project's ladder actually defines, or the report raises later.""" + from openultrasast.learning.verifiers import RUNGS, evidence_level_for + from openultrasast.verification import EvidenceLevel + + assert {rung: evidence_level_for(rung) for rung in RUNGS} == { + "suspicion": EvidenceLevel.SUSPICION, + "static_corroboration": EvidenceLevel.STATIC_CORROBORATION, + "proven": EvidenceLevel.EXPLOIT_DEMONSTRATED, + } + assert all(EvidenceLevel(evidence_level_for(rung).value) for rung in RUNGS) # never a value outside the ladder + + +def test_a_verified_claim_still_passes_the_projects_own_verification_helpers(repo: Path) -> None: + """The bug this pins: stamping a learning rung onto evidence_level raised ValueError in three call sites.""" + from openultrasast.learning.verifiers import Verification, apply_verification, raise_to + from openultrasast.verification import VerificationStatus, is_report_verified, verify_finding + + claim = _claim("lookup", 3, "injection") + proven = apply_verification(claim, Verification(rung="proven", reason="oracle_fired", oracle_output="CANARY x")) + assert "verifier:proven" in proven.tags + assert verify_finding(proven).evidence_level == "suspicion" # would raise ValueError if the rung were stamped on + assert is_report_verified(proven.evidence_level, VerificationStatus.ACCEPTED) is False + raised = raise_to(proven, "proven") + assert raised.evidence_level == "exploit_demonstrated" + assert verify_finding(raised).evidence_level == "exploit_demonstrated" + assert is_report_verified(raised.evidence_level, VerificationStatus.ACCEPTED) is True