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()