diff --git a/src/openultrasast/learning/canaries.py b/src/openultrasast/learning/canaries.py new file mode 100644 index 0000000..13a0082 --- /dev/null +++ b/src/openultrasast/learning/canaries.py @@ -0,0 +1,188 @@ +"""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. + +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 +and oracle code live in the package, outside every directory a round may write. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path + +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 ..tool_hunter import ChatClient +from .verifiers import Verification, VerifierKind, token_oracle + +_TOKEN_PREFIX = "ousast-canary-" +_JUDGE_SYSTEM = ( + "You check one security claim against what a sandbox actually printed. " + 'Answer with JSON only: {"confirms": true|false, "reason": ""}. ' + "Confirm only when the printed output can be explained by the claimed vulnerability and by nothing else." +) + + +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.""" + 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.""" + 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) + return snippet, token + + +@dataclass(frozen=True) +class CanaryVerifier: + """Runs a family's canary in the sandbox and lets only its own token count as proof.""" + + runner: SandboxRunner + limits: SandboxConfig + images: dict[str, str] | None = None + kind: VerifierKind = "canary" + + def verify(self, claim: StaticFinding, *, root: Path, language: str) -> Verification: + family = next((tag.split(":", 1)[1] for tag in claim.tags if tag.startswith("family:")), "") + rendered = render_canary(family, language=language, path=claim.path, function=claim.function_name or "") + if rendered is None: + return Verification(rung="suspicion", reason="no_template") + snippet, token = rendered + images = self.images or dict(DEFAULT_IMAGES) + verdict = RegressionRunner(self.runner).run_recipe( + language, snippet, images.get(language, "ousast-missing-image"), self.limits, repo_root=root, oracle=token_oracle(token) + ) + if verdict.verdict == TRIGGERABLE: + return Verification(rung="proven", reason="oracle_fired", oracle_output=token) + return Verification(rung="suspicion", reason="oracle_quiet" if verdict.reason == "oracle_quiet" else verdict.reason) + + +def confirm_proven( + verification: Verification, claim: StaticFinding, *, client: ChatClient | None, model: str, snippet: str +) -> Verification: + """Req 7.5: a claim is published as proven only when a second, independent judgement agrees.""" + if verification.rung != "proven": + return verification + if client is None or not model: + return Verification(rung="static_corroboration", reason="judge_unavailable", oracle_output=verification.oracle_output) + question = ( + f"Claim: {claim.title}. {claim.rationale[:400]}\n" + f"Location: {claim.path} line {claim.line}, function {claim.function_name}.\n\n" + f"Snippet that was run:\n{snippet}\n\nWhat the sandbox printed:\n{verification.oracle_output}\n\n" + "Does that output demonstrate the claim? Answer as JSON." + ) + messages: list[dict[str, object]] = [{"role": "system", "content": _JUDGE_SYSTEM}, {"role": "user", "content": question}] + try: + response = client.complete(model=model, messages=messages, tools=[], json_object=True) # type: ignore[call-arg] + except TypeError: + response = client.complete(model=model, messages=messages, tools=[]) + except Exception: # noqa: BLE001 — a judge that cannot answer must not promote a claim + return Verification(rung="static_corroboration", reason="judge_unavailable", oracle_output=verification.oracle_output) + if _confirms(response.content): + return verification + return Verification(rung="static_corroboration", reason="judge_declined", oracle_output=verification.oracle_output) + + +def _confirms(content: str | None) -> bool: + if not isinstance(content, str) or "{" not in content: + return False + try: + payload = json.loads(content[content.index("{") : content.rindex("}") + 1]) + except (ValueError, json.JSONDecodeError): + return False + return isinstance(payload, dict) and payload.get("confirms") is True + + +# 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 +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() +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) +except Exception as error: + print("canary could not reach the function:", type(error).__name__) +""" + +_PYTHON_PATH = """# canary: a traversal reads a file the handler was never meant to open +import sys +sys.path.insert(0, {workspace!r}) +CANARY = {token!r} +with open("/scratch/canary.txt", "w", encoding="utf-8") as handle: + handle.write(CANARY) +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) + result = handler("../../scratch/canary.txt") if callable(handler) else None + if CANARY in str(result): + print("CANARY", CANARY) +except Exception as error: + print("canary could not reach the function:", type(error).__name__) +""" + +_PYTHON_DESERIALIZATION = """# canary: an unsafe load reconstructs an object that announces itself +import pickle, sys +sys.path.insert(0, {workspace!r}) +CANARY = {token!r} + +class Marker: + def __reduce__(self): + return (print, ("CANARY", CANARY)) + +blob = pickle.dumps(Marker()) +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) + if callable(handler): + handler(blob) +except Exception as error: + print("canary could not reach the function:", type(error).__name__) +""" + +_TEMPLATES: dict[tuple[str, str], str] = { + ("injection", "python"): _PYTHON_INJECTION, + ("path", "python"): _PYTHON_PATH, + ("deserialization", "python"): _PYTHON_DESERIALIZATION, +} + + +__all__ = ["CanaryVerifier", "canary_token", "confirm_proven", "render_canary"] diff --git a/src/openultrasast/regress/candidate.py b/src/openultrasast/regress/candidate.py index 84c2b9a..d663c38 100644 --- a/src/openultrasast/regress/candidate.py +++ b/src/openultrasast/regress/candidate.py @@ -3,7 +3,7 @@ from __future__ import annotations import json -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from dataclasses import asdict, dataclass, replace from pathlib import Path @@ -131,24 +131,23 @@ class RegressionRunner: sandbox_limits: SandboxConfig, *, repo_root: Path, + oracle: Callable[[SandboxResult], bool] | None = None, ) -> RegressionVerdict: job = recipe_for(language, snippet, image, sandbox_limits, repo_root=repo_root) if job is None: return verdict_from_result(recipe_missing=True) - return self.run_snippet(snippet, job) + return self.run_snippet(snippet, job, oracle=oracle) - def run_snippet(self, snippet: str, job: SandboxJob | None) -> RegressionVerdict: + def run_snippet( + self, snippet: str, job: SandboxJob | None, *, oracle: Callable[[SandboxResult], bool] | None = None + ) -> RegressionVerdict: if job is None: return verdict_from_result(recipe_missing=True) try: check_snippet_safety(snippet) except UnsafeSnippetError: return verdict_from_result(safety_rejected=True) - return _verdict_from_result(self._sandbox.run(job)) - - -def _verdict_from_result(result: SandboxResult) -> RegressionVerdict: - return verdict_from_result(result) + return verdict_from_result(self._sandbox.run(job), oracle=oracle) def snippet_for(language: str, path: str, function_name: str | None) -> str: diff --git a/tests/test_learning_canaries.py b/tests/test_learning_canaries.py new file mode 100644 index 0000000..e21be50 --- /dev/null +++ b/tests/test_learning_canaries.py @@ -0,0 +1,145 @@ +"""learning-harness task 2.7: canary templates and the second judge (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +from openultrasast.config import SandboxConfig +from openultrasast.findings import StaticFinding +from openultrasast.learning.families import load_families +from openultrasast.regress.safety import check_snippet_safety +from openultrasast.sandbox import SandboxResult +from openultrasast.sandbox.runner import FakeSandboxRunner +from openultrasast.tool_hunter import ChatResponse, ScriptedChatClient + +CANARY_FAMILIES = ("injection", "path", "deserialization") + + +def _claim(family: str, function: str = "lookup", line: int = 3) -> StaticFinding: + return StaticFinding( + finding_id=f"detector:app.py:{line}", + path="app.py", + title="t", + severity="high", + 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, + ) + + +def test_every_canary_template_renders_and_passes_the_snippet_safety_check() -> None: + from openultrasast.learning.canaries import render_canary + + for family in CANARY_FAMILIES: + snippet, token = render_canary(family, language="python", path="app.py", function="lookup") + assert token and token in snippet + check_snippet_safety(snippet) # raises when the template would be refused before it ever runs + assert "/workspace" in snippet # the candidate is imported from the read-only source mount + assert render_canary("memory", language="python", path="app.py", function="f") is None + 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: + from openultrasast.learning.canaries import render_canary + + first = render_canary("injection", language="python", path="app.py", function="lookup") + second = render_canary("injection", language="python", path="app.py", function="other") + 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] + + +def test_the_canary_verifier_proves_only_when_the_oracle_fires(tmp_path: Path) -> None: + from openultrasast.learning.canaries import CanaryVerifier + + (tmp_path / "app.py").write_text("def lookup(value):\n return value\n") + fired = CanaryVerifier(runner=_runner(lambda token: SandboxResult(0, f"CANARY {token}", "", False)), limits=SandboxConfig()) + proven = fired.verify(_claim("injection"), root=tmp_path, language="python") + assert proven.rung == "proven" and proven.reason == "oracle_fired" + quiet = CanaryVerifier(runner=_runner(lambda _t: SandboxResult(1, "boom", "Traceback", False)), limits=SandboxConfig()) + unproven = quiet.verify(_claim("injection"), root=tmp_path, language="python") + assert unproven.rung == "suspicion" and unproven.reason == "oracle_quiet" # a crash is not this family's evidence + + +def test_a_family_or_language_with_no_template_stays_at_suspicion(tmp_path: Path) -> None: + from openultrasast.learning.canaries import CanaryVerifier + + (tmp_path / "app.py").write_text("def lookup(value):\n return value\n") + verifier = CanaryVerifier(runner=_runner(lambda token: SandboxResult(0, token, "", False)), limits=SandboxConfig()) + assert verifier.verify(_claim("memory"), root=tmp_path, language="python").reason == "no_template" + assert verifier.verify(_claim("injection"), root=tmp_path, language="rust").reason == "no_template" + + +def test_the_registry_uses_the_canary_verifier_when_one_is_supplied(tmp_path: Path) -> None: + from openultrasast.learning.canaries import CanaryVerifier + from openultrasast.learning.verifiers import NoVerifier, verifier_for + + taxonomy = load_families() + canary = CanaryVerifier(runner=_runner(lambda token: SandboxResult(0, token, "", False)), limits=SandboxConfig()) + assert verifier_for(taxonomy.by_id("injection"), canary=canary) is canary + assert isinstance(verifier_for(taxonomy.by_id("access_control"), canary=canary), type(verifier_for(taxonomy.by_id("access_control")))) + assert isinstance(verifier_for(taxonomy.by_id("memory"), canary=canary), NoVerifier) + + +def test_nothing_is_published_as_proven_until_a_second_judge_confirms() -> None: + from openultrasast.learning.canaries import confirm_proven + from openultrasast.learning.verifiers import Verification + + proven = Verification(rung="proven", reason="oracle_fired", oracle_output="CANARY abc") + agrees = ScriptedChatClient([ChatResponse(content=json.dumps({"confirms": True, "reason": "the row came back"}))]) + kept = confirm_proven(proven, _claim("injection"), client=agrees, model="judge", snippet="print('x')") + assert kept.rung == "proven" + declines = ScriptedChatClient([ChatResponse(content=json.dumps({"confirms": False, "reason": "unrelated output"}))]) + dropped = confirm_proven(proven, _claim("injection"), client=declines, model="judge", snippet="print('x')") + assert dropped.rung == "static_corroboration" and dropped.reason == "judge_declined" + assert confirm_proven(proven, _claim("injection"), client=None, model="", snippet="x").rung == "static_corroboration" + lower = Verification(rung="suspicion", reason="oracle_quiet") + assert confirm_proven(lower, _claim("injection"), client=agrees, model="judge", snippet="x") is lower # nothing to confirm + + +def test_the_judge_sees_the_snippet_and_the_oracle_output_and_never_the_label() -> None: + from openultrasast.learning.canaries import confirm_proven + from openultrasast.learning.verifiers import Verification + + client = ScriptedChatClient([ChatResponse(content=json.dumps({"confirms": True, "reason": "ok"}))]) + confirm_proven( + Verification(rung="proven", reason="oracle_fired", oracle_output="CANARY abc"), + _claim("injection"), + client=client, + model="judge", + snippet="print('CANARY abc')", + ) + prompt = json.dumps(client.calls[0]["messages"]) + assert "CANARY abc" in prompt and "print('CANARY abc')" in prompt + assert client.calls[0].get("json_object") is True and client.calls[0]["tools"] == [] + assert "family:injection" not in prompt # the judge is told the claim, not the answer key + + +def test_templates_and_oracle_code_live_outside_every_proposer_write_root(tmp_path: Path) -> None: + from openultrasast.learning import canaries + from openultrasast.learning.detectors import write_default_configs + + write_roots = set(write_default_configs(tmp_path, load_families(), prompt="p", version="0").values()) + module = Path(canaries.__file__).resolve() + assert not any(module.is_relative_to(root.resolve()) for root in write_roots) + assert "site-packages" in str(module) or "src/openultrasast" in str(module) + + +def _runner(reply: object) -> FakeSandboxRunner: + """A sandbox that answers from whatever canary token the snippet actually planted.""" + + class _Runner(FakeSandboxRunner): # type: ignore[misc] + def run(self, spec: object) -> SandboxResult: + body = "\n".join(str(value) for value in (getattr(spec, "scratch_files", {}) or {}).values()) + marker = next((word.strip("'\"") for word in body.split() if "ousast-canary-" in word), "ousast-canary-none") + return reply(marker) # type: ignore[operator] + + return _Runner()