feat(learning-harness): verifier registry and the sandbox oracle callback

Task 2.6. A detector's claim is a suspicion until something other than the model agrees with it, and what counts as agreement depends on the family. The regress verdict now takes an optional oracle over the sandbox result: when a family supplies one it decides triggerability instead of the exit code, because a crash and a demonstrated injection are different evidence and one family's oracle must not answer for another. Without an oracle the exit code decides exactly as before, and an inconclusive run stays inconclusive either way. The registry answers by the verifier kind the taxonomy declares. Access control gets static corroboration when the obligations checker looks at the same function and agrees the obligation is undischarged, and refuses to corroborate a handler that owes nothing. A family whose verifier does not exist yet reports at suspicion rather than pretending, which is the honest default rather than a gap. A verification is written onto a copy of the claim as a tag, never onto the original. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
2026-10-16 03:49:26 +00:00 · 2026-09-06 10:04:07 +02:00
parent 5628052b51
commit 9801d5407b
3 changed files with 242 additions and 1 deletions
@@ -0,0 +1,111 @@
+"""What may raise a claim above suspicion, keyed by family (learning-harness, Req 7.1, 7.3, 7.4).
+
+A detector's claim is a suspicion until something other than the model agrees with it. What counts as
+agreement depends on the family: a canary oracle observed in the sandbox, corroboration from a static
+checker, or nothing at all. A family whose verifier does not exist yet reports at suspicion rather than
+pretending, and that is the honest default rather than a gap.
+
+The canary oracle is a function over a sandbox result, so the sandbox keeps deciding what happened and
+this module only says what would count. Exit codes are not oracles: a crash and a demonstrated
+injection are different evidence, and one family's oracle must not silently answer for another.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass, replace
+from pathlib import Path
+from typing import Literal, Protocol
+
+from ..findings import StaticFinding
+from ..sandbox import SandboxResult
+from .families import Family
+
+Rung = Literal["suspicion", "static_corroboration", "proven"]
+VerifierKind = Literal["canary", "static", "none"]
+
+
+@dataclass(frozen=True)
+class Verification:
+    """What a verifier saw, and how high it lets the claim go."""
+
+    rung: Rung
+    reason: str
+    oracle_output: str = ""
+
+
+class Verifier(Protocol):
+    @property
+    def kind(self) -> VerifierKind: ...
+
+    def verify(self, claim: StaticFinding, *, root: Path, language: str) -> Verification: ...
+
+
+def token_oracle(token: str) -> Callable[[SandboxResult], bool]:
+    """The oracle a canary snippet is written against: the token appears only if the vulnerability fired."""
+
+    def fired(result: SandboxResult) -> bool:
+        return token in result.stdout or token in result.stderr
+
+    return fired
+
+
+@dataclass(frozen=True)
+class NoVerifier:
+    """Nothing can confirm this family yet, so the claim stays where the detector left it."""
+
+    kind: VerifierKind = "none"
+
+    def verify(self, claim: StaticFinding, *, root: Path, language: str) -> Verification:
+        del claim, root, language
+        return Verification(rung="suspicion", reason="no_verifier")
+
+
+@dataclass(frozen=True)
+class StaticVerifier:
+    """Another checker looked at the same function and agrees the obligation is undischarged."""
+
+    kind: VerifierKind = "static"
+
+    def verify(self, claim: StaticFinding, *, root: Path, language: str) -> Verification:
+        del language
+        from ..hunter_tools import PathEscapesRepo, obligations
+
+        if not claim.function_name:
+            return Verification(rung="suspicion", reason="no_function")
+        try:
+            rows = obligations(root, path=claim.path, function=claim.function_name)
+        except (PathEscapesRepo, OSError):
+            return Verification(rung="suspicion", reason="checker_unavailable")
+        if not rows:
+            return Verification(rung="suspicion", reason="checker_disagrees")
+        summary = "; ".join(f"{row.get('operation')} missing {row.get('missing')}" for row in rows)
+        return Verification(rung="static_corroboration", reason="checker_agrees", oracle_output=summary)
+
+
+def verifier_for(family: Family, *, canary: Verifier | None = None) -> Verifier:
+    """The verifier a family declares. A canary family with no implementation registered stays at suspicion."""
+    if family.verifier == "static":
+        return StaticVerifier()
+    if family.verifier == "canary":
+        return canary if canary is not None else NoVerifier(kind="canary")
+    return NoVerifier()
+
+
+def apply_verification(claim: StaticFinding, verification: Verification) -> StaticFinding:
+    """A copy of the claim carrying what the verifier said. The original is never mutated."""
+    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)
+
+
+__all__ = [
+    "NoVerifier",
+    "Rung",
+    "StaticVerifier",
+    "Verification",
+    "Verifier",
+    "VerifierKind",
+    "apply_verification",
+    "token_oracle",
+    "verifier_for",
+]
@@ -2,6 +2,7 @@

 from __future__ import annotations

+from collections.abc import Callable
 from dataclasses import dataclass

 from ..sandbox import SandboxResult
@@ -21,6 +22,8 @@ SANITIZER_ABORT = "sanitizer_abort"
 CRASH = "crash"
 NONZERO_EXIT = "nonzero_exit"
 EXIT_ZERO = "exit_zero"
+ORACLE_FIRED = "oracle_fired"
+ORACLE_QUIET = "oracle_quiet"

 WORTH_FIXING_REACHABILITY = frozenset({"reachable", "inferred-file-surface"})

@@ -65,8 +68,14 @@ def verdict_from_result(
    sandbox_missing: bool = False,
    safety_rejected: bool = False,
    covering_test: bool = False,
+    oracle: Callable[[SandboxResult], bool] | None = None,
 ) -> RegressionVerdict:
-    """Map a sandbox observation onto one of the four regression verdicts."""
+    """Map a sandbox observation onto one of the four regression verdicts.
+
+    ``oracle`` is how a family says what "the vulnerability fired" means for it: when one is given it
+    decides triggerability instead of the exit code, because a crash and a demonstrated injection are
+    not the same evidence. An inconclusive run stays inconclusive either way.
+    """
    if recipe_missing:
        return RegressionVerdict(verdict=INCONCLUSIVE, reason=MISSING_RECIPE)
    if sandbox_missing:
@@ -79,6 +88,12 @@ def verdict_from_result(
        return RegressionVerdict(verdict=INCONCLUSIVE, reason=TIMEOUT)
    if covering_test and result.exit_code == 0:
        return RegressionVerdict(verdict=ALREADY_COVERED, reason=COVERED_TEST_PASSED)
+    if oracle is not None:
+        return (
+            RegressionVerdict(verdict=TRIGGERABLE, reason=ORACLE_FIRED)
+            if oracle(result)
+            else RegressionVerdict(verdict=NOT_TRIGGERABLE, reason=ORACLE_QUIET)
+        )
    if result.exit_code != 0:
        return RegressionVerdict(verdict=TRIGGERABLE, reason=_nonzero_reason(result))
    return RegressionVerdict(verdict=NOT_TRIGGERABLE, reason=EXIT_ZERO)
@@ -0,0 +1,115 @@
+"""learning-harness task 2.6: the verifier registry and the oracle callback (offline)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from openultrasast.findings import StaticFinding
+from openultrasast.learning.families import load_families
+from openultrasast.regress.verdict import NOT_TRIGGERABLE, TRIGGERABLE, verdict_from_result
+from openultrasast.sandbox import SandboxResult
+
+APP = (
+    "from flask import request\n\n\n"
+    "@app.route('/books/<title>')\n@login_required\ndef guarded(title):\n"
+    "    return Book.query.filter_by(book_title=title).first()\n\n\n"
+    "@app.route('/books/any/<title>')\ndef leaky(title):\n"
+    "    owner = request.args.get('owner')\n"
+    "    return Book.query.filter_by(user_id=owner, book_title=title).first()\n\n\n"
+    "@app.route('/health')\ndef health():\n    return {'ok': True}\n"
+)
+
+
+def _claim(function: str, line: int, family: str) -> StaticFinding:
+    return StaticFinding(
+        finding_id=f"detector:app.py:{line}",
+        path="app.py",
+        title="t",
+        severity="medium",
+        confidence="low",
+        evidence_level="suspicion",
+        rationale="r",
+        line=line,
+        function_name=function,
+        reachability_status="unknown",
+        reachability_evidence=[],
+        reachability_conditions=[],
+        tags=[f"family:{family}"],
+        ranking_priority=0.0,
+    )
+
+
+@pytest.fixture
+def repo(tmp_path: Path) -> Path:
+    (tmp_path / "app.py").write_text(APP)
+    return tmp_path
+
+
+def test_an_oracle_decides_the_verdict_and_the_exit_code_no_longer_does() -> None:
+    from openultrasast.learning.verifiers import token_oracle
+
+    oracle = token_oracle("CANARY:9f2a")
+    fired = SandboxResult(exit_code=0, stdout="ok\nCANARY:9f2a\n", stderr="", timed_out=False)
+    quiet = SandboxResult(exit_code=1, stdout="boom", stderr="Traceback", timed_out=False)
+    assert verdict_from_result(fired, oracle=oracle).verdict == TRIGGERABLE
+    assert verdict_from_result(quiet, oracle=oracle).verdict == NOT_TRIGGERABLE  # a crash is not this family's oracle
+    assert verdict_from_result(quiet).verdict == TRIGGERABLE  # without an oracle the exit code still decides
+    assert verdict_from_result(fired, oracle=oracle).reason == "oracle_fired"
+
+
+def test_an_oracle_never_overrides_an_inconclusive_run() -> None:
+    from openultrasast.learning.verifiers import token_oracle
+
+    oracle = token_oracle("CANARY:9f2a")
+    timed_out = SandboxResult(exit_code=0, stdout="CANARY:9f2a", stderr="", timed_out=True)
+    assert verdict_from_result(timed_out, oracle=oracle).verdict == "inconclusive"
+    assert verdict_from_result(None, oracle=oracle).verdict == "inconclusive"
+    assert verdict_from_result(sandbox_missing=True, oracle=oracle).reason == "sandbox_unavailable"
+
+
+def test_the_registry_answers_by_the_family_s_declared_verifier_kind() -> None:
+    from openultrasast.learning.verifiers import NoVerifier, StaticVerifier, verifier_for
+
+    taxonomy = load_families()
+    assert isinstance(verifier_for(taxonomy.by_id("access_control")), StaticVerifier)
+    assert isinstance(verifier_for(taxonomy.by_id("memory")), NoVerifier)
+    assert isinstance(verifier_for(taxonomy.by_id("output_encoding")), NoVerifier)
+    # a canary family with no canary implementation registered yet reports at suspicion rather than pretending
+    injection = verifier_for(taxonomy.by_id("injection"))
+    assert isinstance(injection, NoVerifier) and injection.kind == "canary"
+
+
+def test_a_family_with_no_verifier_leaves_the_claim_at_suspicion(repo: Path) -> None:
+    from openultrasast.learning.verifiers import verifier_for
+
+    taxonomy = load_families()
+    verification = verifier_for(taxonomy.by_id("memory")).verify(_claim("leaky", 12, "memory"), root=repo, language="python")
+    assert verification.rung == "suspicion" and verification.reason == "no_verifier"
+
+
+def test_the_static_verifier_raises_an_access_control_claim_the_checker_agrees_with(repo: Path) -> None:
+    from openultrasast.learning.verifiers import verifier_for
+
+    taxonomy = load_families()
+    verifier = verifier_for(taxonomy.by_id("access_control"))
+    agreed = verifier.verify(_claim("leaky", 12, "access_control"), root=repo, language="python")
+    assert agreed.rung == "static_corroboration" and "identity_constraint" in agreed.oracle_output
+    # the checker owes nothing in a handler that touches no protected resource, so it refuses to corroborate
+    healthy = verifier.verify(_claim("health", 16, "access_control"), root=repo, language="python")
+    assert healthy.rung == "suspicion" and healthy.reason == "checker_disagrees"
+    elsewhere = verifier.verify(_claim("missing", 99, "access_control"), root=repo, language="python")
+    assert elsewhere.rung == "suspicion"
+
+
+def test_a_verification_can_be_written_onto_the_claim_as_a_tag(repo: Path) -> None:
+    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"
+    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"