test(discovery): check planted canaries and gate the run

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"]

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, "<CANARY>")

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": []},
     }
 
 

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}

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("<CANARY>", 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()