From 8507af9309f6456a6ecc3fca86bd1dcb4b9bc901 Mon Sep 17 00:00:00 2001 From: Pengyu Zhang Date: Sat, 22 Aug 2026 15:40:53 -0700 Subject: [PATCH] test(discovery): check planted canaries and gate the run Two things that decide whether a run is allowed to pass, kept apart from the measurement they sit on top of. The canary check is deliberately the crudest test in the harness. It does not know how redaction is implemented, does not walk the asset model, and searches the serialized snapshot - every byte the collector would have written out - for each planted value. A value that leaked into a field nobody modelled is exactly the one a smarter check would miss. Any hit fails the run outright regardless of every other score: a collector that finds every tool and leaks one token has not had a good run. A canary with no recorded value is reported as unplanted rather than clean, because a check that never ran is not a check that passed. The context line around a hit masks the value it reports - printing it into a document that then gets shared would repeat exactly the mistake being reported. The gate fails on a leaked canary, a dirty baseline, an unexplained error, any duplicate, a missed review-queue entry, or recall below the previous accepted run for that OS. Recall is compared against history rather than an absolute threshold because a real endpoint is not perfectly reproducible - a background updater can change a version mid-run - while a leaked credential or an invented asset is not weather and fails immediately. --- Discovery/tests/scoring/__init__.py | 7 ++- Discovery/tests/scoring/canary.py | 53 +++++++++++++++++ Discovery/tests/scoring/schema.py | 8 +++ Discovery/tests/scoring/score.py | 52 ++++++++++++++++- Discovery/tests/test_score.py | 91 ++++++++++++++++++++++++++++- 5 files changed, 204 insertions(+), 7 deletions(-) create mode 100644 Discovery/tests/scoring/canary.py diff --git a/Discovery/tests/scoring/__init__.py b/Discovery/tests/scoring/__init__.py index f78a6e0..203da85 100644 --- a/Discovery/tests/scoring/__init__.py +++ b/Discovery/tests/scoring/__init__.py @@ -5,10 +5,11 @@ about by somebody who has never provisioned a guest. """ +from .canary import check_canaries from .match import load_snapshot, match_all from .schema import SCORE_VERSION from .score import score, score_run -from .snapshot import Asset, Snapshot, added_assets, duplicate_ids +from .snapshot import Asset, Snapshot, added_assets -__all__ = ["Asset", "Snapshot", "added_assets", "duplicate_ids", "load_snapshot", "match_all", - "score", "score_run", "SCORE_VERSION"] +__all__ = ["Asset", "Snapshot", "added_assets", "load_snapshot", "match_all", "score", + "score_run", "check_canaries", "SCORE_VERSION"] diff --git a/Discovery/tests/scoring/canary.py b/Discovery/tests/scoring/canary.py new file mode 100644 index 0000000..e9d203f --- /dev/null +++ b/Discovery/tests/scoring/canary.py @@ -0,0 +1,53 @@ +"""The redaction check: did any planted credential survive serialization? + +Deliberately the crudest test in the harness. It does not know how redaction is +implemented, does not walk the asset model, and does not care which field a +value came from. It searches the serialized snapshot - every byte the collector +would have written to a file or posted to a server - for each planted string. + +A hit fails the run outright, regardless of every other score. A collector that +finds every tool and leaks one token has not had a good run. +""" + +import json +from typing import Any, Dict, Iterable, List + +from .snapshot import Snapshot + + +def check_canaries(snapshot: Snapshot, planted: Dict[str, str], + also: Iterable[Any] = ()) -> Dict[str, Any]: + """Search the serialized snapshot for every planted canary value. + + ``also`` takes any extra documents the run publishes - the report, a fleet + payload - because a value redacted out of the snapshot and copied into the + scorecard has still left the machine. + """ + haystacks = [snapshot.serialized()] + haystacks.extend(json.dumps(document, default=str) for document in also) + leaked: List[Dict[str, str]] = [] + for name, value in sorted(planted.items()): + if not value: + # A canary with no value was never planted, and reporting it clean + # would be reporting a check that never ran. + leaked.append({"canary": name, "where": "unplanted", + "detail": "no value recorded for this run"}) + continue + for index, haystack in enumerate(haystacks): + if value in haystack: + leaked.append({"canary": name, "where": "snapshot" if index == 0 else "document:%d" % index, + "detail": _context(haystack, value)}) + return {"planted": len(planted), "leaked": len(leaked), "clean": not leaked, "hits": leaked} + + +def _context(haystack: str, value: str) -> str: + """A window around the hit, with the canary itself masked. + + The point of the detail line is to say *where* a value surfaced. Printing + the value again into a report that then gets shared would repeat exactly the + mistake being reported. + """ + start = max(0, haystack.find(value) - 60) + end = min(len(haystack), haystack.find(value) + len(value) + 60) + window = haystack[start:end] + return window.replace(value, "") diff --git a/Discovery/tests/scoring/schema.py b/Discovery/tests/scoring/schema.py index 61d88f8..4deb45a 100644 --- a/Discovery/tests/scoring/schema.py +++ b/Discovery/tests/scoring/schema.py @@ -20,6 +20,12 @@ SCORED_FIELDS = ("version", "install_path", "install_method", "config_scope", "transport", "pinned", "liveness") +#: Why a run failed, in the order a reader should care. Canary leaks come first +#: because they invalidate everything below them. +GATE_REASONS = ("canary_leaked", "baseline_dirty", "unexplained_errors", + "duplicates", "recall_regressed", "review_queue_miss") + + def empty_totals() -> Dict[str, Any]: return {"tp": 0, "fp": 0, "fn": 0, "dup": 0, "recall": None, "precision": None} @@ -46,12 +52,14 @@ def blank_score(run: Dict[str, Any]) -> Dict[str, Any]: "totals": empty_totals(), "by_category": {}, "fields": {}, + "canaries": {}, "errors": {}, "review_queue": {}, "misses": [], "inventions": [], "duplicates": [], "excluded": [], + "gate": {"passed": True, "reasons": []}, } diff --git a/Discovery/tests/scoring/score.py b/Discovery/tests/scoring/score.py index dea1f19..0f17db7 100644 --- a/Discovery/tests/scoring/score.py +++ b/Discovery/tests/scoring/score.py @@ -12,11 +12,12 @@ from ..manifest import Entry, Manifest from . import schema +from .canary import check_canaries from .match import Match, expand, load_snapshot, match_all, norm_path from .snapshot import Asset, Snapshot, added_assets, duplicate_ids -def score_run(run_dir: str, manifest: Manifest) -> Dict[str, Any]: +def score_run(run_dir: str, manifest: Manifest, previous: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Score a run directory: the two snapshots plus what actually installed. A run directory is the unit of replay. Everything the scorer needs is in it, @@ -26,11 +27,26 @@ def score_run(run_dir: str, manifest: Manifest) -> Dict[str, Any]: after = load_snapshot(os.path.join(run_dir, "after.json")) with open(os.path.join(run_dir, "manifest.actual.json"), encoding="utf-8") as handle: actual = json.load(handle) - return score(before, after, actual, manifest) + planted = _load_planted(run_dir) + return score(before, after, actual, manifest, planted=planted, previous=previous) + + +def _load_planted(run_dir: str) -> Dict[str, str]: + """Canary values for this run, if the runner recorded them. + + Absent is not the same as none planted: a run whose canary file is missing + cannot claim a clean redaction check, and ``check_canaries`` is what says so. + """ + path = os.path.join(run_dir, "canaries.json") + if not os.path.exists(path): + return {} + with open(path, encoding="utf-8") as handle: + return json.load(handle) def score(before: Snapshot, after: Snapshot, actual: Dict[str, Any], - manifest: Manifest) -> Dict[str, Any]: + manifest: Manifest, planted: Optional[Dict[str, str]] = None, + previous: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Compare what the collector reported against what was actually installed.""" platform = actual.get("os", "linux") for name, snapshot in (("before", before), ("after", after)): @@ -64,8 +80,10 @@ def score(before: Snapshot, after: Snapshot, actual: Dict[str, Any], result["duplicates"] = schema.sort_findings(duplicates) result["excluded"] = schema.sort_findings(excluded) result["fields"] = _field_accuracy(matches, actual, platform) + result["canaries"] = check_canaries(after, planted or {}) result["errors"] = _errors(after, manifest, platform, actual.get("home")) result["review_queue"] = _review_queue(after, manifest, platform, actual.get("home")) + result["gate"] = _gate(result, previous) return result @@ -301,3 +319,31 @@ def _review_queue(after: Snapshot, manifest: Manifest, platform: str, return {"expected": len(wanted), "queued": sum(1 for row in rows if row["queued"]), "size": len(after.review_queue), "entries": rows, "passed": all(row["queued"] for row in rows)} + + +def _gate(result: Dict[str, Any], previous: Optional[Dict[str, Any]]) -> Dict[str, Any]: + """Whether this run is allowed to pass. + + Absolute thresholds do not survive contact with a real endpoint - a + background updater can change a version mid-run - so recall is compared + against the previous accepted run for the same OS. Everything else is + binary, because a leaked credential or an invented asset is not weather. + """ + reasons: List[str] = [] + if not result["canaries"].get("clean", False): + reasons.append("canary_leaked") + if not result["baseline"]["clean"]: + reasons.append("baseline_dirty") + if result["errors"]["unexplained"]: + reasons.append("unexplained_errors") + if result["totals"]["dup"]: + reasons.append("duplicates") + if not result["review_queue"]["passed"]: + reasons.append("review_queue_miss") + baseline_recall = ((previous or {}).get("totals") or {}).get("recall") + recall = result["totals"]["recall"] + if baseline_recall is not None and recall is not None and recall < baseline_recall: + reasons.append("recall_regressed") + return {"passed": not reasons, "reasons": reasons, + "compared_to": (previous or {}).get("run", {}).get("id") if previous else None, + "previous_recall": baseline_recall} diff --git a/Discovery/tests/test_score.py b/Discovery/tests/test_score.py index 96977e4..37e7177 100644 --- a/Discovery/tests/test_score.py +++ b/Discovery/tests/test_score.py @@ -7,12 +7,14 @@ "recall is high" would not notice. """ +import copy import json import os import unittest from . import manifest as manifest_module -from .scoring import score_run +from .scoring import canary, score, score_run +from .scoring.snapshot import Snapshot RECORDED = os.path.join(os.path.dirname(os.path.abspath(__file__)), "recorded", "synthetic-linux") @@ -104,6 +106,93 @@ def test_the_open_world_entry_reaches_the_review_queue(self): self.assertTrue(self.score["review_queue"]["passed"]) self.assertEqual(self.score["review_queue"]["expected"], 1) + def test_the_gate_fails_on_the_duplicates(self): + self.assertFalse(self.score["gate"]["passed"]) + self.assertEqual(self.score["gate"]["reasons"], ["duplicates"]) + + +class Gate(unittest.TestCase): + """Everything the gate refuses, one refusal at a time.""" + + @classmethod + def setUpClass(cls): + cls.manifest = manifest_module.load() + cls.actual = _load("manifest.actual") + cls.planted = _load("canaries") + + def _score(self, before=None, after=None, planted=None, previous=None): + return score(Snapshot(before or _load("before")), Snapshot(after or _load("after")), + self.actual, self.manifest, planted=planted or self.planted, + previous=previous) + + def test_a_leaked_canary_fails_the_run(self): + """Regardless of every other score: a collector that finds every tool + and leaks one token has not had a good run.""" + after = copy.deepcopy(_load("after")) + after["assets"][0]["risk"]["args"] = [self.planted["hook_token"]] + result = self._score(after=after) + self.assertIn("canary_leaked", result["gate"]["reasons"]) + self.assertEqual(result["canaries"]["leaked"], 1) + self.assertNotIn(self.planted["hook_token"], json.dumps(result["canaries"])) + + def test_an_unplanted_canary_is_not_reported_clean(self): + """A check that never ran is not a check that passed.""" + result = self._score(planted=dict(self.planted, hook_token="")) + self.assertFalse(result["canaries"]["clean"]) + + def test_a_dirty_baseline_fails_the_run(self): + """Anything on a clean machine is a false positive with no manifest to + blame, and it invalidates every number computed after it.""" + before = copy.deepcopy(_load("before")) + before["assets"] = [copy.deepcopy(_load("after")["assets"][0])] + result = self._score(before=before) + self.assertIn("baseline_dirty", result["gate"]["reasons"]) + + def test_an_unexplained_error_fails_the_run(self): + after = copy.deepcopy(_load("after")) + after["errors"].append({"probe": "app", "path": "/opt/mystery", "message": "denied"}) + result = self._score(after=after) + self.assertIn("unexplained_errors", result["gate"]["reasons"]) + + def test_recall_is_compared_against_the_last_accepted_run(self): + """A real endpoint is not perfectly reproducible, so the gate compares + against history rather than an absolute threshold.""" + previous = {"run": {"id": "yesterday"}, "totals": {"recall": 1.0}} + result = self._score(previous=previous) + self.assertIn("recall_regressed", result["gate"]["reasons"]) + self.assertEqual(result["gate"]["compared_to"], "yesterday") + + def test_a_repeated_asset_id_is_refused_rather_than_scored(self): + """Any score computed from a halved delta would be wrong in a direction + that flatters the collector.""" + after = copy.deepcopy(_load("after")) + after["assets"].append(copy.deepcopy(after["assets"][0])) + with self.assertRaises(ValueError): + self._score(after=after) + + +class CanarySearch(unittest.TestCase): + def test_the_search_covers_the_whole_document_not_the_modelled_fields(self): + snapshot = Snapshot({"hostname": "h", "assets": [], + "stats": {"note": "leaked-value-here"}}) + result = canary.check_canaries(snapshot, {"c": "leaked-value-here"}) + self.assertEqual(result["leaked"], 1) + + def test_the_context_line_masks_the_value_it_reports(self): + """Printing the value into a report that then gets shared would repeat + exactly the mistake being reported.""" + snapshot = Snapshot({"hostname": "h", "assets": [], "stats": {"n": "sk-xyz"}}) + result = canary.check_canaries(snapshot, {"c": "sk-xyz"}) + self.assertIn("", result["hits"][0]["detail"]) + self.assertNotIn("sk-xyz", result["hits"][0]["detail"]) + + def test_extra_documents_are_searched_too(self): + """A value redacted out of the snapshot and copied into the scorecard + has still left the machine.""" + snapshot = Snapshot({"hostname": "h", "assets": []}) + result = canary.check_canaries(snapshot, {"c": "tok"}, also=[{"report": "tok"}]) + self.assertEqual(result["leaked"], 1) + if __name__ == "__main__": unittest.main()