diff --git a/src/openultrasast/learning/verifiers.py b/src/openultrasast/learning/verifiers.py new file mode 100644 index 0000000..7474ac7 --- /dev/null +++ b/src/openultrasast/learning/verifiers.py @@ -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", +] diff --git a/src/openultrasast/regress/verdict.py b/src/openultrasast/regress/verdict.py index 60442ac..637df3a 100644 --- a/src/openultrasast/regress/verdict.py +++ b/src/openultrasast/regress/verdict.py @@ -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) diff --git a/tests/test_learning_verifiers.py b/tests/test_learning_verifiers.py new file mode 100644 index 0000000..d805a2b --- /dev/null +++ b/tests/test_learning_verifiers.py @@ -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/')\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"