fix(learning-harness): the three defects the group 2 review found

Task 2.3: two of the five unscorable reasons had no code path, so an unparsable
row was reported as a corpus labelling defect when the true cause was a tooling
gap. They have different owners and different fixes, and conflating them is the
dishonesty this spec exists to remove. The design's unscorable_reason is
reinstated with the parse result as an input, and a handler that arrived without
its registration is now named as missing context for the families whose bug
shape is a missing check.

Task 2.6: a verification stamped its own word for proof onto the shared evidence
level, which is not a value the project's ladder defines, and three existing
helpers raised on it. Nothing in production called it yet, so the suite stayed
green while the wiring in the next group would have broken a scan. A
verification now writes only its tag, and a caller that wants to move a claim up
the ladder goes through an explicit mapping that can only produce a level the
ladder has.

Task 2.7: the injection canary planted its row in a private in-memory database
the target could never reach, so it could not fire; the fake sandbox handed the
token back regardless of the snippet, which is why nothing noticed. It is now an
execution canary whose marker only an interpreter could write, and each template
is tested by running it for real against a vulnerable fixture and its fixed
twin. That test immediately caught a second flaw: a handler that merely echoes
its input printed the oracle token and satisfied the oracle without any
vulnerability, so an injected payload now carries a value unrelated to the token
the oracle looks for.

Also from the review: a configuration with no tools is refused instead of
producing a detector that silently reports nothing, the classifier abstains when
its fallback call fails too, and the archive docstring says which version wins.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
2026-09-06 10:32:02 +02:00
co-authored by Claude Opus 5
parent e44d45f488
commit b0dba26468
10 changed files with 268 additions and 37 deletions
+62 -23
View File
@@ -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"]
+5 -4
View File
@@ -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:
+2
View File
@@ -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)}")
+5 -1
View File
@@ -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":
+35 -3
View File
@@ -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",
]
+31 -2
View File
@@ -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",
]