From 64839f742f38595c1c5f0c8a07788699f9a254c2 Mon Sep 17 00:00:00 2001 From: "C.D.Lee" <265386841+opencdlee-dotcom@users.noreply.github.com> Date: Wed, 23 Sep 2026 15:04:38 -0700 Subject: [PATCH 1/2] A Developer ID signature earned no rung, and custody was asked only after the interrupt 190 of the 200 HIGH interrupts the operator closed as noise in 30 days carried no custody rung. Two reasons, both in the grading/routing tier. _grade_binary had no publisher rung: publisher-stable is reachable only on a re-sign in place, so a valid Developer ID binary under $HOME earned nothing on first sight and beaconed HIGH on the one sensor whose predicate is an OR. It now awards `publisher-signed` (vouched tier, one step down) when publisher_sig() accepts the verdict and it names a team or authority; asked after the vouch and the receipt, before build-output, recorded in the custody ledger. It reads the stat-cached verdict the sensor already asked for, and a probe that did not answer this scan is not asked a second time. route_findings now consults custody before the notify floor: a finding below CRITICAL, not attack-defined, not a decoy/latch/canary trip, whose rung is in the self or vouched tier routes to the digest with why `provenance:` and carries `routed` for `incident` and `report`. Two regression tests pinned the old MEDIUM for a publisher-signed, non-attack finding (a /bin/ls listener, an un-notarized Developer ID app); both now pin the rung and the digest route. Co-Authored-By: Claude Opus 5.5 (1M context) --- ARCHITECTURE.md | 30 ++- aegis.py | 161 ++++++++++++-- tests/conftest.py | 2 + tests/test_provenance_gate.py | 381 ++++++++++++++++++++++++++++++++++ tests/test_regression.py | 30 ++- 5 files changed, 582 insertions(+), 22 deletions(-) create mode 100644 tests/test_provenance_gate.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 26b5a88..d0340ad 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -363,11 +363,31 @@ quiet rather than silence. verdicts, and a ledger that crossed machines would let one compromised body launder bytes into every other body's known-good set. +9. **Publisher signature** (binary-keyed findings) → **MEDIUM**, asked after + the vouch and the receipt and before `build-output`. `publisher-stable` + was the only rung that read a signature, and only on a re-sign in place, + so a valid Developer ID binary earned nothing on first sight and beaconed + HIGH from `$HOME` on the one sensor whose predicate is an OR. The binary + earns `publisher-signed` when `publisher_sig()` accepts its verdict + (apple / app-store / developer-id on macOS, a `strict: detritus` verdict + included; os-signed / signed-valid on Windows; os-managed on Linux) and + the verdict names a team or authority; the note names that signer and the + control behind it (Apple's notarization and revocation on macOS). Ad-hoc, + broken, unsigned and `signed-other` earn nothing. It reads the stat-cached + verdict the sensor already asked for, so it costs no second probe. + Guards, because grading is where an attacker would want to stand: -- **Grades, never mutes.** A downgraded finding is still created, still in the - report, still accumulates risk and joins correlation chains. Custody writes - no dismissal and cannot feed acquired tolerance. +- **Demotion never suppresses; the routing gate decides what interrupts.** A + downgraded finding is still created, still in the report, still accumulates + risk and joins correlation chains, and custody writes no dismissal and + cannot feed acquired tolerance. What keeps a proven origin quiet is the + ROUTING gate (`_provenance_gate`, asked by `route_findings` before the + notify floor): a finding below CRITICAL whose rung is in the self or + vouched tier, that is not attack-defined and not a decoy/latch/canary trip, + routes to the digest, where it remains visible (it carries + `routed: digest: provenance `, which `incident` and `report` print) + and counts toward risk at its tier weight. Weak rungs only demote. - **Attack-defined content never downgrades.** A conceal imperative stays HIGH even when self-attested — an agent prompt-injected into persisting a hostile instruction attests its own write. Custody grades *churn-shaped structure*, @@ -398,8 +418,8 @@ call sequence: bidirectional: a covered change drops to INFO, an uncovered change is promoted to at least HIGH — under enforcement, the *absence of a record* outranks provenance, which is the entire point of opting in. -4. **The notify floor** (`emit`) routes by the final severity; it never - changes one. +4. **The notify floor** (`emit`) routes by the final severity and, through + the provenance gate, by custody; it never changes a severity. 5. **The incident ratchet** (`_severity_max`) only ever steps UP: once an incident opened HIGH, a later regrade of the same subject cannot quietly lower it — de-escalation is the operator's verdict to give, not custody's. diff --git a/aegis.py b/aegis.py index 4a2d512..81d5998 100755 --- a/aegis.py +++ b/aegis.py @@ -2298,6 +2298,8 @@ def classify_signature(path): # reading them describes neither version. Either way the answer is still # returned for this call; it is only not remembered. failed = result.pop("probe_failed", False) + if failed and stat_sig is not None: + _SIG_UNANSWERED[path] = stat_sig if (stat_sig is not None and not failed and _sig_stat(path) == stat_sig): _sigcache.pop(path, None) # overwrite any prior entry for this path @@ -2367,6 +2369,13 @@ def _classify_linux(path): # a DEGRADED sensor rather than letting the silence read as "everything signed". _SIG_PROBE_FAILURES = 0 +# {path: stat signature} for every probe that did not answer this scan. A +# non-answer is never cached (see classify_signature), which is right for the +# verdict and would make the custody ladder's publisher rung ask codesign (or +# a cold PowerShell) a SECOND time about a binary its own sensor just failed +# to classify. Read only by _publisher_signer; reset with the count above. +_SIG_UNANSWERED = {} + # Every PowerShell-backed probe in this file is sized against MEASUREMENTS of a # cold powershell.exe, never a guess. A cap at or near the boot cost is not a # timeout, it is a coin flip on whether the interpreter finished booting -- and @@ -4617,6 +4626,32 @@ def _signal_decision(f, memory): return None, 0 +def _provenance_gate(f): + """The custody rung that keeps `f` out of the interrupt tier, or None. + + Custody used to be asked last and could only demote, one step; nothing + consulted it before an interrupt, and 190 of the 200 HIGH interrupts the + operator closed as noise (30 days to 2026-09-23) carried no rung at all. + A finding whose ORIGIN is proven -- a self-tier rung (the operator + authored it) or a vouched-tier rung (a signed publisher, a package + receipt, an OS update) -- is an attribute-only observation about a known + thing, and belongs in the digest, where it stays visible and still counts + toward risk at its tier weight. + + Three things are never gated, whatever their custody: CRITICAL, evidence + that is attack-DEFINED (a hostile argv, an IOC hit, a conceal imperative + -- origin is not innocence), and a tripped decoy/latch/canary, which no + provenance explains. Weak rungs (`build-output`, `supervised`, ...) + explain nothing here; they only ever buy the grader's one step.""" + if SEV_ORDER.get(f.get("severity"), -1) >= SEV_ORDER["CRITICAL"]: + return None + if str(f.get("fingerprint") or "").startswith(_NEVER_TOLERATE_PREFIXES): + return None + if not _custody_explained(f): + return None + return f.get("custody") or f.get("provenance") + + def route_findings(findings, first_run=False, adopt=frozenset(), memory=None, seen=None): """{fingerprint: {"route", "why", "decision", "verdicts"}} for a batch. @@ -4639,7 +4674,11 @@ def route_findings(findings, first_run=False, adopt=frozenset(), memory=None, on the very first scan. `adopt` is the same rule for a surface an upgrade sees for the first time. Confidence is the second routing axis: a high-impact-but-noisy hit (explicit confidence='low') is logged and - correlated but routed to the digest instead of interrupting.""" + correlated but routed to the digest instead of interrupting. Provenance + is the third: a finding whose origin custody has proven routes to the + digest at any severity below CRITICAL (see _provenance_gate), and says so + on the finding itself as `routed`, which the incident view and the report + print.""" if seen is None: seen = load_json(SEEN, {}) allow = set(load_json(ALLOWLIST, [])) @@ -4649,6 +4688,8 @@ def route_findings(findings, first_run=False, adopt=frozenset(), memory=None, if fp in out: continue decision, verdicts = _signal_decision(f, memory) + gate = _provenance_gate(f) + f.pop("routed", None) if fp in allow: route, why, decision = ROUTE_SILENT, "allowlisted", "allowlisted" elif fp in seen: @@ -4658,6 +4699,9 @@ def route_findings(findings, first_run=False, adopt=frozenset(), memory=None, route, why = ROUTE_DIGEST, "adopted" elif CONFIDENCE_ORDER.get(f.get("confidence", "medium"), 1) <= 0: route, why = ROUTE_DIGEST, "low-confidence" + elif gate: + route, why = ROUTE_DIGEST, "provenance:%s" % gate + f["routed"] = "digest: provenance %s" % gate elif SEV_ORDER[f["severity"]] < SEV_ORDER[NOTIFY_MIN_SEV]: route, why = ROUTE_DIGEST, "below-floor" elif decision: @@ -16355,6 +16399,14 @@ def _custody_carry_note(carried): "publisher can ship a bad build, and a stolen cert " "signs too, so this quiets it rather than clearing " "it.)"), + "publisher-signed": ("The binary carries a valid signature from a " + "publisher the platform itself trusts, and the " + "signer names itself — the ORIGIN of these bytes is " + "a known vendor, not a drop. Origin, not innocence: " + "a publisher can ship a bad build and a stolen " + "certificate signs cleanly until it is revoked, so " + "this demotes one step and routes to the digest; " + "attack-defined behaviour keeps its full severity."), "package-managed": ("This binary is owned by a package-manager transaction " "on this machine, proven by its receipt on disk — it " "arrived through an install you ran, not a drop. " @@ -16392,13 +16444,16 @@ def _custody_carry_note(carried): # authority as its baseline -> a vendor update # package-managed the binary is owned by a package-manager transaction on # this machine, proven by its RECEIPT on disk +# publisher-signed the binary carries a valid signature from a publisher the +# platform trusts (publisher_sig) that names its signer +# -> a vendor's build, on FIRST sight # # They are deliberately weaker than _SELF_CUSTODY: those three are claims of # AUTHORSHIP, these are claims of ORIGIN. So they demote one step (HIGH -> # MEDIUM), never straight to LOW, except `relocated`, which is a proof that # the executed content is byte-identical and therefore carries no new code. _VOUCHED_CUSTODY = ("relocated", "publisher-stable", "package-managed", - "os-vendor") + "os-vendor", "publisher-signed") # Recognised-but-weak: git knows the edit is local and unpushed. The note has # always read "routine if you made it" while the finding stayed HIGH anyway — @@ -16447,10 +16502,13 @@ def _demote(severity, provenance, attack_defined=False): The three invariants this function exists to hold, all inherited from the delegate-surface grader that came before it: - * grading DEMOTES, it never suppresses — every finding stays in the - report at its new level, no finding is dropped, no incident is - auto-closed, and nothing here ever writes a dismissal, so a demotion - can never feed acquired tolerance; + * demotion never suppresses — every finding stays in the report at + its new level, no finding is dropped, no incident is auto-closed, and + nothing here ever writes a dismissal, so a demotion can never feed + acquired tolerance. (Keeping a proven-origin finding out of the + interrupt tier is the ROUTING gate's decision, _provenance_gate, not + this function's: the finding still reaches the digest and the risk + tier at its custody weight.); * attack-DEFINED evidence is never demoted, whoever authored it. A hostile argv, an IOC hit, a dylib-injection env, a conceal imperative keeps its severity even under perfect custody: knowing who wrote a @@ -16675,6 +16733,68 @@ def _package_receipt(path): return None +def _publisher_signer(path): + """The signature verdict for `path` when a publisher the platform trusts + vouches for these bytes AND names itself, else None. + + `publisher-stable` was the only rung that read a signature, and only on a + re-sign in place, so a valid Developer ID binary earned nothing on first + sight and beaconed HIGH from $HOME. This is the first-sight half. It reads + classify_signature, whose verdict is stat-cached per path, so a binary the + sensor has already classified costs no second probe. `publisher_sig` is + the per-body vocabulary (developer-id / app-store / apple on macOS; a + `strict: detritus` verdict is still developer-id); an ad-hoc, broken, + unsigned or unanchored (`signed-other`) signature is never it. A verdict + with no team and no authority names nobody the operator could check, so + it earns nothing either. Never raises: this runs inside a sensor's + grading path, where an error must cost a rung, never a scan. + + A probe that already failed this scan on these exact bytes is not asked + again: the answer would be the same non-answer, bought with a second + codesign run (or a second cold PowerShell on Windows).""" + if not path: + return None + failed_on = _SIG_UNANSWERED.get(path) + if failed_on is not None and failed_on == _sig_stat(path): + return None + try: + sig = classify_signature(path) + except Exception: + return None + if publisher_sig(sig.get("trust")) and (sig.get("team") + or sig.get("authority")): + return sig + return None + + +def _publisher_line(sig): + """The one line that names who signed and which control stands behind + it, so the operator can check the claim rather than take it.""" + trust, team = sig.get("trust"), sig.get("team") + authority = sig.get("authority") or "" + if trust == "os-managed": + return ("Owned by the distro package %s; the distribution's signed " + "package pipeline is the control that stands behind this " + "rung." % authority) + if trust in ("os-signed", "signed-valid"): + return ("Authenticode-signed by %s; the certificate chain and its " + "revocation are the control that stands behind this rung." + % authority) + if trust == "developer-id" and team: + name = authority.split(": ", 1)[-1] + if name.endswith(" (%s)" % team): + name = name[:-len(" (%s)" % team)] + who = "Developer ID team %s (%s)" % (team, name) + elif trust == "app-store": + who = "the Mac App Store%s" % (" for team %s" % team if team else "") + elif trust == "apple": + who = "Apple (%s)" % (authority or "platform binary") + else: + who = authority or "team %s" % team + return ("Signed by %s; Apple's notarization/revocation is the control " + "that stands behind this rung." % who) + + def _grade_binary(severity, path, attack_defined=False, endpoint=None, sha=None, parents=None): """(graded_severity, rung, note) for a finding keyed on a BINARY's identity. @@ -16708,9 +16828,16 @@ def _grade_binary(severity, path, attack_defined=False, endpoint=None, # the caller names an endpoint, the vouch must cover that exact endpoint — # an identity vouch never widens into "may talk to anywhere". if _vouch_covers(path, endpoint): - rung = "operator-vouched" + rung, signer = "operator-vouched", None elif _package_receipt(path): - rung = "package-managed" + rung, signer = "package-managed", None + else: + # A publisher the platform itself trusts, on FIRST sight. Asked only + # when neither stronger rung answered, so a vouched or receipted + # binary never costs a signature lookup here. + rung, signer = None, _publisher_signer(path) + if rung or signer: + rung = rung or "publisher-signed" # Last, and weakest: a generated artifact of a repo this machine commits # to. Before this rung existed, the two questions above were the ONLY ones # asked of a binary, and a developer's machine answers no to both for @@ -16747,7 +16874,10 @@ def _grade_binary(severity, path, attack_defined=False, endpoint=None, # after the grading decision, never before it, so the ledger only ever # holds rungs that were actually awarded. _custody_remember(sha or _graded_sha(path), rung, path) - return _demote(severity, rung), rung, _PROVENANCE_NOTE.get(rung) + note = _PROVENANCE_NOTE.get(rung) + if signer: + note = "%s %s" % (_publisher_line(signer), note) + return _demote(severity, rung), rung, note # --- the vouch tier: a workload the operator signed for, by hand ------------- @@ -20551,9 +20681,10 @@ def _brief_report(findings, new_findings, incidents, sensor_health, first_run, if top: lines.append("## New since last scan") for f in top: - lines.append("- %s **%s** — %s" % ( + lines.append("- %s **%s** — %s%s" % ( SEV_ICON[f["severity"]], f["title"], - (f["detail"] or "").splitlines()[0][:160])) + (f["detail"] or "").splitlines()[0][:160], + " (%s)" % f["routed"] if f.get("routed") else "")) lines.append("") if crit: lines.append("## Open CRITICAL incidents") @@ -20633,8 +20764,9 @@ def _full_report(payload): if f["category"] != cur: cur = f["category"] lines.append("## %s" % cur) - lines.append("- %s **%s** — %s" % ( - SEV_ICON[f["severity"]], f["title"], f["detail"])) + lines.append("- %s **%s** — %s%s" % ( + SEV_ICON[f["severity"]], f["title"], f["detail"], + " (%s)" % f["routed"] if f.get("routed") else "")) return "\n".join(lines) + "\n" @@ -20883,6 +21015,7 @@ def _cmd_scan_locked(quiet=False): ensure_state() # per-scan; a stale count or flag would mislead every later run _SIG_PROBE_FAILURES = 0 + _SIG_UNANSWERED.clear() _PROC_ENUM_FAILED = False _PROC_ARGV_PARTIAL = False _reset_unexamined() @@ -22240,6 +22373,8 @@ def cmd_incident(incident_id, action=None, reason=None): try: data = json.loads(evidence["data_json"]) summary = data.get("title") or data.get("status") or evidence["event_type"] + if data.get("routed"): + summary += " · %s" % data["routed"] except Exception: summary = evidence["event_type"] print(" - %s · %s · %s" % diff --git a/tests/conftest.py b/tests/conftest.py index 868421a..f8f4dc1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -339,6 +339,8 @@ def publisher_trust_for(is_win, is_linux): # SIP + the sealed system volume: an OS platform-binary update is a # macOS-shaped fact, and _os_program_update refuses off macOS by design. "OneOsUpdateIsOneFinding", "TheGuardRefusesEverythingItShould", + # the publisher custody rung read through codesign's Developer ID verdict + "DeveloperIdRung", )) diff --git a/tests/test_provenance_gate.py b/tests/test_provenance_gate.py new file mode 100644 index 0000000..7e83f6b --- /dev/null +++ b/tests/test_provenance_gate.py @@ -0,0 +1,381 @@ +#!/usr/bin/env python3 +"""Provenance becomes a gate: a proven origin keeps an attribute-only finding +out of the interrupt tier. + +Measured on the live store, 30 days to 2026-09-23: 190 of the 200 HIGH +interrupts the operator closed as noise carried `custody: null`. The custody +ladder existed and the interrupts never met it, for two reasons pinned here. + + * `_grade_binary` had no publisher rung. A valid Developer ID, App Store or + Apple signature earned nothing on first sight -- `publisher-stable` is + reachable only from a re-sign in place -- so a Developer-ID binary under + $HOME beaconed HIGH on the one sensor whose predicate is an OR + (`suspicious_sig(trust) or is_risky_location(path)`). In the corpus: + Codex's SkyComputerUseService, an App-Translocated Obsidian Helper, the + Chrome-app shim under ~/Applications. + * Custody could only DEMOTE, one step, and nothing consulted it before the + interrupt. + +Every class pins the quieting AND the property that stops it becoming a blind +spot: attack-defined evidence, CRITICAL, and the tripwire fingerprints +(decoy:/latch:/canary:) interrupt under any custody; a weak rung still only +demotes; a vouch and a package receipt still answer first; and an ad-hoc, +broken, unsigned or unanchored signature earns nothing. +""" +import contextlib +import io +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from conftest import (PUBLISHER_TRUST, SUSPICIOUS_TRUST, # noqa: E402 + aegis) +from test_codesign_detritus import (CLEAN_VERIFY, DEV_ID_DV, # noqa: E402 + TIMED_OUT) +from test_regression import Sandbox # noqa: E402 +from test_routing_gate import GateSandbox # noqa: E402 + +# Gated on BOTH, as test_signature_corpus is: CI runs a Linux leg with +# aegis.IS_MAC forced on, and that leg has no codesign to be right about. +REAL_MAC = sys.platform == "darwin" and aegis.IS_MAC + +SPOTIFY = "/Applications/Spotify.app/Contents/MacOS/Spotify" +SKY = "/Users/op/.codex/computer-use/SkyComputerUseService" + + +PUBLISHER_NAME = "Example Publisher Ltd" + + +def _publisher_verdict(): + """A classify_signature() answer publisher_sig() accepts on THIS body + (conftest.PUBLISHER_TRUST), naming its signer. The Developer ID shape is + macOS vocabulary and lives in DeveloperIdRung, which is gated.""" + return {"trust": PUBLISHER_TRUST, "team": None, + "authority": PUBLISHER_NAME} + + +def _process(sev="HIGH", custody="publisher-signed", fp=None, **extra): + return aegis.finding( + sev, "process", "Suspicious running process", "d", + fp or "process:%s:adhoc:%s" % (SKY, "a" * 64), + path=SKY, custody=custody, **extra) + + +class TheGate(GateSandbox): + """route_findings consults custody before the notify floor.""" + + def route(self, f): + return aegis.route_findings([f], memory=None, seen={})[f["fingerprint"]] + + def test_a_proven_origin_high_goes_to_the_digest(self): + v = self.route(_process()) + self.assertEqual(("digest", "provenance:publisher-signed"), + (v["route"], v["why"])) + + def test_every_self_and_vouched_rung_gates(self): + for rung in aegis._SELF_CUSTODY + aegis._VOUCHED_CUSTODY: + with self.subTest(rung=rung): + v = self.route(_process(custody=rung)) + self.assertEqual(("digest", "provenance:" + rung), + (v["route"], v["why"])) + + def test_the_agent_surface_spelling_gates_too(self): + """The delegate-surface diff names the rung `provenance`.""" + v = self.route(_process(custody=None, provenance="self-committed")) + self.assertEqual("provenance:self-committed", v["why"]) + + def test_attack_defined_interrupts_under_publisher_custody(self): + """Origin is not innocence: a vouched binary running an osascript + password prompt is still an alarm.""" + v = self.route(_process(attack_defined=True)) + self.assertEqual(("interrupt", "new"), (v["route"], v["why"])) + + def test_critical_interrupts_under_any_custody(self): + for rung in (aegis._SELF_CUSTODY + aegis._VOUCHED_CUSTODY + + aegis._WEAK_CUSTODY): + with self.subTest(rung=rung): + v = self.route(_process("CRITICAL", custody=rung)) + self.assertEqual("interrupt", v["route"]) + + def test_tripwire_fingerprints_interrupt_under_any_custody(self): + self.assertTrue({"decoy:", "latch:", "canary:"} + <= set(aegis._NEVER_TOLERATE_PREFIXES)) + for prefix in aegis._NEVER_TOLERATE_PREFIXES: + with self.subTest(prefix=prefix): + v = self.route(_process(fp=prefix + "tripped:" + SKY)) + self.assertEqual("interrupt", v["route"]) + + def test_a_weak_rung_only_demotes(self): + """build-output, supervised, copy-of-graded ... explain nothing at the + gate; their one step happens in the grader, as before.""" + for rung in aegis._WEAK_CUSTODY: + with self.subTest(rung=rung): + self.assertEqual("interrupt", + self.route(_process(custody=rung))["route"]) + + def test_no_custody_still_interrupts(self): + self.assertEqual("interrupt", self.route(_process(custody=None))["route"]) + + def test_the_gate_is_asked_before_the_floor(self): + """A demoted MEDIUM names the reason it is quiet, not merely that it + is below the floor.""" + v = self.route(_process("MEDIUM")) + self.assertEqual(("digest", "provenance:publisher-signed"), + (v["route"], v["why"])) + + def test_the_operator_allowlist_and_seen_ledger_still_answer_first(self): + f = _process() + v = aegis.route_findings([f], memory=None, + seen={f["fingerprint"]: "t"})[f["fingerprint"]] + self.assertEqual("seen", v["route"]) + + +class TheRouteIsRecorded(GateSandbox): + """The operator can see WHY a HIGH did not interrupt.""" + + def test_the_scan_path_notifies_nobody_and_says_why(self): + f = self.process(SKY, "c" * 64, custody="publisher-signed") + routing, new_high = self.scan_path([f]) + self.assertEqual("provenance:publisher-signed", + routing[f["fingerprint"]]["why"]) + self.assertEqual([], new_high) + self.assertEqual([], self.notified) + self.assertEqual("digest: provenance publisher-signed", f["routed"]) + # Still a finding on the durable record -- digest, not erasure. + with open(aegis.FINDINGS_LOG, encoding="utf-8") as log: + self.assertIn("digest: provenance publisher-signed", log.read()) + opened = [i for i in self.incidents() + if i["correlation_key"] == "signal:" + f["case_fingerprint"]] + self.assertEqual(1, len(opened)) + out = io.StringIO() + with contextlib.redirect_stdout(out): + aegis.cmd_incident(opened[0]["id"]) + self.assertIn("digest: provenance publisher-signed", out.getvalue()) + + def test_the_full_report_names_the_route(self): + f = _process() + aegis.route_findings([f], memory=None, seen={}) + md = aegis._full_report({"findings": [f]}) + self.assertIn("digest: provenance publisher-signed", md) + + def test_an_interrupt_carries_no_route_label(self): + f = _process(custody=None) + aegis.route_findings([f], memory=None, seen={}) + self.assertNotIn("routed", f) + + +class _RungSandbox(Sandbox): + """_grade_binary with the classifier and the receipt probe stubbed.""" + + def setUp(self): + super().setUp() + for name in ("classify_signature", "_vouch_covers", "_package_receipt", + "_build_output_rung", "run"): + self._saved[name] = getattr(aegis, name) + aegis._CUSTODY_CARRY_CACHE.clear() + aegis._GRADED_SHA_CACHE.clear() + aegis._SIG_UNANSWERED.clear() + aegis._package_receipt = lambda path: None + self.bin = os.path.join(self.tmp, "SkyComputerUseService") + with open(self.bin, "wb") as fh: + fh.write(b"MACH-O-ish publisher-signed bytes " * 32) + self.asked = [] + + def tearDown(self): + aegis._CUSTODY_CARRY_CACHE.clear() + aegis._GRADED_SHA_CACHE.clear() + aegis._SIG_UNANSWERED.clear() + super().tearDown() + + def verdict(self, **sig): + def fake(path): + self.asked.append(path) + return dict(sig) + aegis.classify_signature = fake + + +class PublisherRung(_RungSandbox): + """The rung on every body, in that body's own trust vocabulary.""" + + def test_a_publisher_signature_earns_the_rung(self): + self.verdict(**_publisher_verdict()) + sev, rung, note = aegis._grade_binary("HIGH", self.bin) + self.assertEqual(("MEDIUM", "publisher-signed"), (sev, rung)) + self.assertIn(PUBLISHER_NAME, note, "the note must name the signer") + self.assertEqual([self.bin], self.asked, + "the classifier is asked once per grading") + + def test_the_rung_is_recorded_in_the_custody_ledger(self): + self.verdict(**_publisher_verdict()) + aegis._grade_binary("HIGH", self.bin) + aegis._CUSTODY_CARRY_CACHE.clear() # read the file, not the memo + carried = aegis._custody_carried(aegis.sha256(self.bin)) + self.assertIsNotNone(carried) + self.assertEqual("publisher-signed", carried[0]) + + def test_unproven_signatures_earn_nothing(self): + for trust in (SUSPICIOUS_TRUST, "broken", "unsigned", "signed-other", + "unknown", "missing", "unmanaged"): + with self.subTest(trust=trust): + self.verdict(trust=trust, team="2FNC3A47ZF", + authority=PUBLISHER_NAME) + sev, rung, _note = aegis._grade_binary("HIGH", self.bin) + self.assertEqual(("HIGH", None), (sev, rung)) + + def test_a_publisher_verdict_that_names_no_signer_earns_nothing(self): + self.verdict(trust=PUBLISHER_TRUST, team=None, authority=None) + self.assertEqual(("HIGH", None), + aegis._grade_binary("HIGH", self.bin)[:2]) + + def test_a_vouched_path_still_wins_first(self): + aegis._vouch_covers = lambda path, endpoint=None: True + self.verdict(**_publisher_verdict()) + self.assertEqual(("LOW", "operator-vouched"), + aegis._grade_binary("HIGH", self.bin)[:2]) + self.assertEqual([], self.asked, "a stronger rung answered; the " + "classifier must not be asked") + + def test_a_package_receipt_still_wins_before_it(self): + aegis._package_receipt = lambda path: "homebrew:tool@1.0" + self.verdict(**_publisher_verdict()) + self.assertEqual("package-managed", + aegis._grade_binary("HIGH", self.bin)[1]) + self.assertEqual([], self.asked) + + def test_it_is_asked_before_build_output(self): + aegis._build_output_rung = lambda path: True + self.verdict(**_publisher_verdict()) + self.assertEqual("publisher-signed", + aegis._grade_binary("HIGH", self.bin)[1]) + self.verdict(trust=SUSPICIOUS_TRUST, team=None, authority=None) + self.assertEqual("build-output", + aegis._grade_binary("HIGH", self.bin)[1]) + + def test_attack_defined_evidence_earns_nothing(self): + self.verdict(**_publisher_verdict()) + self.assertEqual(("HIGH", None, None), + aegis._grade_binary("HIGH", self.bin, + attack_defined=True)) + + def test_the_rung_is_a_vouched_origin_one_step_down(self): + self.assertIn("publisher-signed", aegis._VOUCHED_CUSTODY) + self.assertNotIn("publisher-signed", + aegis._SELF_CUSTODY + aegis._WEAK_CUSTODY) + self.assertEqual("MEDIUM", aegis._demote("HIGH", "publisher-signed")) + self.assertEqual("HIGH", aegis._demote("CRITICAL", "publisher-signed")) + self.assertEqual(0.25, aegis._RISK_CUSTODY_WEIGHT["publisher-signed"]) + self.assertTrue(aegis._PROVENANCE_NOTE.get("publisher-signed")) + + +class DeveloperIdRung(_RungSandbox): + """The macOS half: a Developer ID verdict, and what codesign costs. + + Listed in conftest._MAC_ONLY_CLASSES: every assertion here is about the + codesign vocabulary or _classify_mac, which no other body produces.""" + + SPOTIFY_ID = {"trust": "developer-id", "team": "2FNC3A47ZF", + "authority": "Developer ID Application: Spotify (2FNC3A47ZF)"} + + def test_a_developer_id_earns_the_rung_and_names_the_team(self): + self.verdict(**self.SPOTIFY_ID) + sev, rung, note = aegis._grade_binary("HIGH", self.bin) + self.assertEqual(("MEDIUM", "publisher-signed"), (sev, rung)) + self.assertTrue(note.startswith( + "Signed by Developer ID team 2FNC3A47ZF (Spotify); Apple's " + "notarization/revocation is the control that stands behind " + "this rung."), note) + + def test_a_strict_detritus_verdict_still_qualifies(self): + """S1: Finder detritus with an intact seal keeps the chain's trust.""" + self.verdict(strict="detritus", **self.SPOTIFY_ID) + self.assertEqual("publisher-signed", + aegis._grade_binary("HIGH", self.bin)[1]) + + def test_adhoc_and_broken_earn_nothing(self): + for trust in ("adhoc", "broken"): + with self.subTest(trust=trust): + self.verdict(**dict(self.SPOTIFY_ID, trust=trust)) + self.assertEqual(("HIGH", None), + aegis._grade_binary("HIGH", self.bin)[:2]) + + def test_grading_after_the_sensor_costs_no_second_codesign(self): + """The sensor classified the binary first; the rung reads the + stat-keyed verdict cache, never codesign again.""" + aegis.classify_signature = self._saved["classify_signature"] + calls = [] + real = self._saved["run"] + + def fake(cmd, timeout=15, extra_env=None, stdin_data=None): + if list(cmd)[:1] == ["codesign"]: + calls.append(list(cmd)[1]) + return ("", DEV_ID_DV, 0) if "-dv" in cmd else CLEAN_VERIFY + return real(cmd, timeout=timeout, extra_env=extra_env, + stdin_data=stdin_data) + aegis.run = fake + self.assertEqual("developer-id", + aegis.classify_signature(self.bin)["trust"]) + probes = len(calls) + self.assertEqual("publisher-signed", + aegis._grade_binary("HIGH", self.bin)[1]) + self.assertEqual(probes, len(calls), calls) + + def test_a_probe_that_did_not_answer_is_not_asked_again(self): + """A non-answer is never cached, so without a per-scan record the + rung would run codesign a second time on exactly the binary the + sensor just failed to classify -- under load, a second timeout.""" + aegis.classify_signature = self._saved["classify_signature"] + calls = [] + real = self._saved["run"] + answer = [TIMED_OUT] + + def fake(cmd, timeout=15, extra_env=None, stdin_data=None): + if list(cmd)[:1] == ["codesign"]: + calls.append(list(cmd)[1]) + if "-dv" in cmd: + return answer[0] + return CLEAN_VERIFY + return real(cmd, timeout=timeout, extra_env=extra_env, + stdin_data=stdin_data) + aegis.run = fake + self.assertEqual("unknown", aegis.classify_signature(self.bin)["trust"]) + probes = len(calls) + sev, rung, _note = aegis._grade_binary("HIGH", self.bin) + self.assertEqual(("HIGH", None), (sev, rung)) + self.assertEqual(probes, len(calls), calls) + # Different bytes are a different question, and are asked. + with open(self.bin, "ab") as fh: + fh.write(b" rebuilt") + answer[0] = ("", DEV_ID_DV, 0) + self.assertEqual("publisher-signed", + aegis._grade_binary("HIGH", self.bin)[1]) + self.assertGreater(len(calls), probes) + + +@unittest.skipUnless(REAL_MAC, "grades a real codesign answer; macOS only") +class LivePublisherRung(Sandbox): + """A fixture cannot test a parser: the rung, against the real tool.""" + + def setUp(self): + super().setUp() + aegis._CUSTODY_CARRY_CACHE.clear() + aegis._GRADED_SHA_CACHE.clear() + + def tearDown(self): + aegis._CUSTODY_CARRY_CACHE.clear() + aegis._GRADED_SHA_CACHE.clear() + super().tearDown() + + def test_spotify_grades_publisher_signed(self): + if not os.path.exists(SPOTIFY): + self.skipTest("Spotify is not installed here") + sev, rung, note = aegis._grade_binary("HIGH", SPOTIFY) + self.assertEqual(("MEDIUM", "publisher-signed"), (sev, rung), note) + team = aegis.classify_signature(SPOTIFY)["team"] + self.assertTrue(team) + self.assertIn("Developer ID team %s" % team, note) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_regression.py b/tests/test_regression.py index a7fa1ea..5627fd9 100644 --- a/tests/test_regression.py +++ b/tests/test_regression.py @@ -2265,11 +2265,25 @@ def test_new_suspicious_listener_is_high(self): self.assertEqual(fs[0]["severity"], "HIGH") self.assertEqual(fs[0]["port"], "4444") - def test_new_signed_listener_is_medium_not_notify(self): + def test_new_signed_listener_is_graded_by_publisher_not_notify(self): + """Pinned MEDIUM until 2026-09-23 because a signature bought nothing + but not being `hostile`. The doctrine it guarded -- a signed listener + never interrupts -- still holds; what changed is that a publisher's + signature is now a custody rung (`publisher-signed`, one step down) + and the routing gate says why it is quiet.""" fs = aegis.diff_listeners({}, {"/bin/ls:8000": "/bin/ls"}) self.assertEqual(len(fs), 1) - self.assertEqual(fs[0]["severity"], "MEDIUM", - "a signed listener must stay below the notify floor") + self.assertLess(aegis.SEV_ORDER[fs[0]["severity"]], + aegis.SEV_ORDER[aegis.NOTIFY_MIN_SEV], + "a signed listener must stay below the notify floor") + route = aegis.route_findings(fs, seen={})[fs[0]["fingerprint"]] + self.assertEqual("digest", route["route"]) + if sys.platform == "darwin": + # A real codesign answer: /bin/ls is Apple-signed. CI's Linux leg + # with the mac flags forced on has no codesign, so no rung. + self.assertEqual(("LOW", "publisher-signed"), + (fs[0]["severity"], fs[0]["custody"])) + self.assertEqual("provenance:publisher-signed", route["why"]) def test_preexisting_listener_not_realerted(self): cur = {"/bin/ls:8000": "/bin/ls"} @@ -2337,9 +2351,17 @@ def test_unnotarized_devid_app_is_medium_with_verdict(self): aegis.classify_signature = saved_cs aegis.gatekeeper_verdict = saved_gk self.assertEqual(len(fs), 1, fs) - self.assertEqual(fs[0]["severity"], "MEDIUM") + # MEDIUM until 2026-09-23. The Developer ID signature is now the + # publisher custody rung, one step down and routed to the digest; the + # Gatekeeper verdict and the notary fingerprint are unchanged. + self.assertEqual(fs[0]["severity"], "LOW") + self.assertEqual(fs[0]["provenance"], "publisher-signed") + self.assertIn("Developer ID team T (X)", fs[0]["detail"]) self.assertEqual(fs[0]["gatekeeper"], "rejected") self.assertTrue(fs[0]["fingerprint"].startswith("hotdir:notary:")) + route = aegis.route_findings(fs, seen={})[fs[0]["fingerprint"]] + self.assertEqual(("digest", "provenance:publisher-signed"), + (route["route"], route["why"])) def test_notarized_app_is_silent(self): app, _exe = self._mk_app("Fine.app") From 54133d8d55b659ccde7f171d3962579ed2d79785 Mon Sep 17 00:00:00 2001 From: "C.D.Lee" <265386841+opencdlee-dotcom@users.noreply.github.com> Date: Wed, 23 Sep 2026 19:32:21 -0700 Subject: [PATCH 2/2] The publisher rung trusted a signature where the platform had said no, and a digest-routed incident could still page later Three verifier decisions on PR #60. 1. Gatekeeper's no withholds the rung. The hot-dir sensor calls _grade_binary(publisher_ok=False) after spctl rejects a bundle, so an un-notarized Developer ID app is MEDIUM with no rung again (the test is restored). A notarized bundle stays silent and its binary still earns the rung from every other sensor. 2. The OS vendor's own signature (`apple`, Windows `os-signed`) earns publisher-signed only inside TRUSTED_PREFIXES, judged on the resolved path, and is never written to the custody ledger, so a copy of those bytes cannot inherit copy-of-graded. /bin/ls in place earns it; the same bytes in a temp dir earn nothing (live codesign test). 3. An incident opened only by digest-routed evidence records why in incidents.digest_only (schema v2, ALTER migration) and is never claimed by a reminder; evidence that would itself interrupt clears the mark. test_routing_gate's "digest-routed signal still gets its reminder" pinned the old doctrine and now pins the new one. Also fixes two test leaks the simbody legs found: _RungSandbox overwrote Sandbox's saved classify_signature on a Windows body (leaking its stub), and Sandbox never cleared the in-process custody-carry memo, so a rung awarded to one clang-built fixture reached every later fixture with the same bytes on Linux. Co-Authored-By: Claude Opus 5.5 (1M context) --- ARCHITECTURE.md | 22 ++- aegis.py | 103 +++++++++++--- tests/test_provenance_gate.py | 248 ++++++++++++++++++++++++++++++++-- tests/test_regression.py | 45 ++++-- tests/test_routing_gate.py | 23 ++-- 5 files changed, 395 insertions(+), 46 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d0340ad..014ad6a 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -163,7 +163,12 @@ Allowed states are `OPEN`, `ACK`, `INVESTIGATING`, `CONTAINED`, `RECOVERING`, `MONITORING`, `RESOLVED`, and `FALSE_POSITIVE`. Transitions are validated so a closed incident cannot silently return to containment. An unresolved incident gets at most three reminders (about +1 hour, +24 hours, and +72 hours); afterward -the durable open state is the reminder. A reviewed `FALSE_POSITIVE` suppresses +the durable open state is the reminder. An incident opened only by evidence the +routing gate sent to the digest (the provenance gate, low confidence, below the +floor) records why in `digest_only` and gets none: no reminder ever turns it +into a notification. It stays open, counted and listed, and becomes an ordinary +notified incident only when evidence that would itself interrupt attaches. A +reviewed `FALSE_POSITIVE` suppresses only the exact correlation key while retaining later occurrences as evidence; a changed content hash gets a new key (unless *acquired tolerance*, below, has earned the right to pre-close it). A `RESOLVED` threat that recurs opens a @@ -374,7 +379,13 @@ quiet rather than silence. the verdict names a team or authority; the note names that signer and the control behind it (Apple's notarization and revocation on macOS). Ad-hoc, broken, unsigned and `signed-other` earn nothing. It reads the stat-cached - verdict the sensor already asked for, so it costs no second probe. + verdict the sensor already asked for, so it costs no second probe. Two + limits. The OS vendor's own signature (`apple`, Windows `os-signed`) earns + it only inside `TRUSTED_PREFIXES`, on the resolved path, and is never + carried to a copy: the same signed bytes in `/tmp`, `$HOME` or `%TEMP%` are + the living-off-the-land shape. And a caller whose evidence says the + platform's control refused the bytes withholds it (`publisher_ok=False`): + the hot-dir sensor after Gatekeeper rejects a bundle. Guards, because grading is where an attacker would want to stand: @@ -550,10 +561,13 @@ above. An allowlisted fingerprint still opened and refreshed incidents and drove reminders, because `emit` skipped it while every finding flowed into the incident tier untouched. And one genuine new HIGH marked every incident created that scan as already-notified, so a digest-routed sibling lost the -reminder that was its only path to a human. +reminder that was its only path to a human. (That reminder was itself removed +on 2026-09-23: a digest-only incident is never reminded — see Incident +workflow. The per-finding "notified" mark stands.) `route_findings` is the one place the order is written down: allowlisted → -seen → adopted → low-confidence → below-floor → tolerated/learning → new. +seen → adopted → low-confidence → provenance → below-floor → +tolerated/learning → new. The scan path computes it once with the incident tier's memory and hands the same verdicts to `emit`, to `record_security_state` (which now marks "notified" per finding and closes an allowlisted incident as `allowlisted`, diff --git a/aegis.py b/aegis.py index 81d5998..b148fb6 100755 --- a/aegis.py +++ b/aegis.py @@ -3004,7 +3004,8 @@ def _epoch(value=None): last_notified_at INTEGER, resolution TEXT, subject_json TEXT, - last_novel_at INTEGER + last_novel_at INTEGER, + digest_only TEXT ); CREATE INDEX IF NOT EXISTS idx_incidents_active ON incidents(status, next_reminder_at); @@ -3064,7 +3065,7 @@ def _epoch(value=None): # Bumped ONLY when _EVENT_SCHEMA_SQL or the ALTER migrations below change. # Stamped into the store's own `PRAGMA user_version`, which is what stops the # schema being re-executed on every connection open (see _event_connection). -_EVENT_SCHEMA_VERSION = 1 +_EVENT_SCHEMA_VERSION = 2 # Two retention buckets, not one. `events` used to keep the newest 50,000 rows # full stop, and _record_health writes ONE row per sensor per scan (~45). At the @@ -3161,6 +3162,10 @@ def _event_connection(): # new is eligible immediately, the correct verdict on it. db.execute("UPDATE incidents SET last_novel_at=created_at " "WHERE last_novel_at IS NULL") + if "digest_only" not in cols: + # NULL for every existing row: an incident opened before the + # mark existed keeps the reminders it was promised. + db.execute("ALTER TABLE incidents ADD COLUMN digest_only TEXT") db.execute("PRAGMA user_version=%d" % _EVENT_SCHEMA_VERSION) db.commit() except sqlite3.DatabaseError as e: @@ -3495,8 +3500,21 @@ def _mark_novelty(db, incident_id, event_ids, now): def _upsert_incident(db, key, title, severity, kind, now, event_ids, - initially_notified=False, subject=None): + initially_notified=False, subject=None, route=None): + """Open or update the incident for `key`; returns its id. + + `route` is the routing verdict (route_findings) for the evidence being + attached, when the caller has one. It carries the one reminder rule: an + incident OPENED by evidence the gate sent to the digest -- the provenance + gate, low confidence, anything below the interrupt tier -- records why in + `digest_only`, is never claimed by a reminder, and so is never turned into + a notification later by the clock. Evidence that would itself interrupt + clears the mark when it attaches, and the incident becomes an ordinary + notified one from that moment. Callers without a routing (the correlation + and risk tiers, legacy callers) pass None and keep the old behaviour.""" subject_json = json.dumps(subject, sort_keys=True) if subject else None + routed = (route or {}).get("route") + digest_why = (route or {}).get("why") if routed == ROUTE_DIGEST else None marks = ",".join("?" for _ in _ACTIVE_INCIDENT_STATES) row = db.execute( "SELECT * FROM incidents WHERE correlation_key=? AND status IN (%s) " @@ -3514,6 +3532,13 @@ def _upsert_incident(db, key, title, severity, kind, now, event_ids, "updated_at=?, subject_json=COALESCE(subject_json,?) " "WHERE id=?", (new_sev, new_status, now, now, subject_json, incident_id)) + if row["digest_only"] and routed == ROUTE_INTERRUPT: + # The one way out: this evidence interrupted on its own merits, + # so the operator has just been told, and the case is an ordinary + # notified incident from here on. + db.execute("UPDATE incidents SET digest_only=NULL, " + "last_notified_at=?, next_reminder_at=? WHERE id=?", + (now, now + _REMINDER_DELAYS[0], incident_id)) _mark_novelty(db, incident_id, event_ids, now) else: # FALSE_POSITIVE is a reviewed verdict on the SIGNALS that were seen, not @@ -3550,10 +3575,11 @@ def _upsert_incident(db, key, title, severity, kind, now, event_ids, "INSERT INTO incidents(kind,correlation_key,title,severity,status," "created_at,first_seen,last_seen,updated_at,reminder_count," "next_reminder_at,last_notified_at,subject_json," - "last_novel_at) " - "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + "last_novel_at,digest_only) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (kind, key, title, severity, "OPEN", now, now, now, now, 0, - now + _REMINDER_DELAYS[0], last_notified, subject_json, now)) + None if digest_why else now + _REMINDER_DELAYS[0], + last_notified, subject_json, now, digest_why)) incident_id = cur.lastrowid for event_id in event_ids: db.execute("INSERT OR IGNORE INTO incident_events(incident_id,event_id) " @@ -5426,7 +5452,8 @@ def has_marker(f, values): incident_id = _upsert_incident( db, "signal:" + (f.get("case_fingerprint") or f["fingerprint"]), f["title"], f["severity"], "signal", now, [event_id], - notified or bool(decision), subject=f.get("subject")) + notified or bool(decision), subject=f.get("subject"), + route=verdict) if decision == "allowlisted": _close_allowlisted(db, incident_id, now) elif decision == "tolerated": @@ -6943,6 +6970,10 @@ def claim_due_incident_reminders(now=None): rows = db.execute( "SELECT * FROM incidents WHERE status IN (%s) AND " "next_reminder_at IS NOT NULL AND next_reminder_at<=? " + # The reminder rule's enforcement point: a digest-only + # incident is never claimed, whoever armed its timer (a + # hand `reopen` does) -- see _upsert_incident. + "AND digest_only IS NULL " "ORDER BY severity DESC,next_reminder_at" % marks, _ACTIVE_INCIDENT_STATES + (now,)).fetchall() for row in rows: @@ -9707,7 +9738,13 @@ def _check_hot_app(path, st, cutoff): # "Verify you built/trust it" is a question custody can often answer # outright: an app this machine built, or a copy of one, is the ordinary # reading of signed-but-un-notarized on a developer's Mac. - graded, rung, rung_note = _grade_binary("MEDIUM", exe, sha=sha) + # + # Gatekeeper just refused this bundle, and Gatekeeper's notarization check + # is the control the publisher rung leans on: the platform said no, so the + # signature earns nothing here (publisher_ok=False). The same bytes, once + # notarized, earn it from every other sensor. + graded, rung, rung_note = _grade_binary("MEDIUM", exe, sha=sha, + publisher_ok=False) return [finding( graded, "hot-dir", "Un-notarized app in watched folder", "%s [%s] is signed but NOT notarized (Gatekeeper: %s%s), modified %s — " @@ -16733,6 +16770,17 @@ def _package_receipt(path): return None +# The OS vendor's own platform signature, per body. Its origin claim is only +# as good as WHERE the bytes run: inside the platform's protected install tree +# (TRUSTED_PREFIXES -- the sealed system volume on macOS; SystemRoot and the +# admin-only Program Files trees on Windows) it is the OS; a copy of the same +# signed bytes in /tmp, $HOME or %TEMP% is the living-off-the-land shape, and +# the signature says nothing about who put them there. A third-party vendor's +# signature (developer-id, app-store, signed-valid) names that vendor wherever +# it runs; Linux's os-managed is already a claim about the package's own path. +_PLATFORM_SIGNATURES = ("apple", "os-signed") + + def _publisher_signer(path): """The signature verdict for `path` when a publisher the platform trusts vouches for these bytes AND names itself, else None. @@ -16751,7 +16799,11 @@ def _publisher_signer(path): A probe that already failed this scan on these exact bytes is not asked again: the answer would be the same non-answer, bought with a second - codesign run (or a second cold PowerShell on Windows).""" + codesign run (or a second cold PowerShell on Windows). + + A platform signature (_PLATFORM_SIGNATURES) earns the rung only inside + the system tree, judged on the resolved path so a symlink or a `..` + cannot walk a copy in.""" if not path: return None failed_on = _SIG_UNANSWERED.get(path) @@ -16761,10 +16813,13 @@ def _publisher_signer(path): sig = classify_signature(path) except Exception: return None - if publisher_sig(sig.get("trust")) and (sig.get("team") - or sig.get("authority")): - return sig - return None + if not (publisher_sig(sig.get("trust")) + and (sig.get("team") or sig.get("authority"))): + return None + if sig.get("trust") in _PLATFORM_SIGNATURES and \ + not _is_trusted_prefix(os.path.realpath(path)): + return None + return sig def _publisher_line(sig): @@ -16796,7 +16851,7 @@ def _publisher_line(sig): def _grade_binary(severity, path, attack_defined=False, endpoint=None, - sha=None, parents=None): + sha=None, parents=None, publisher_ok=True): """(graded_severity, rung, note) for a finding keyed on a BINARY's identity. process / net-listener / net-outbound / net-beacon all raise on the same @@ -16820,6 +16875,12 @@ def _grade_binary(severity, path, attack_defined=False, endpoint=None, the caller has one (check_processes, and only on a host with a vouch). It is the only input here that is not about the file: the `supervised` rung asks who STARTED it, which no path, receipt or repo can answer. + + `publisher_ok=False` is a caller saying the platform's own control has + already refused these bytes -- the hot-dir sensor after Gatekeeper + rejected the bundle. The publisher rung leans on exactly that control + (notarization and revocation), so it is not asked; the rest of the ladder + is. """ if attack_defined: return severity, None, None @@ -16835,7 +16896,8 @@ def _grade_binary(severity, path, attack_defined=False, endpoint=None, # A publisher the platform itself trusts, on FIRST sight. Asked only # when neither stronger rung answered, so a vouched or receipted # binary never costs a signature lookup here. - rung, signer = None, _publisher_signer(path) + rung, signer = None, (_publisher_signer(path) if publisher_ok + else None) if rung or signer: rung = rung or "publisher-signed" # Last, and weakest: a generated artifact of a repo this machine commits @@ -16872,8 +16934,11 @@ def _grade_binary(severity, path, attack_defined=False, endpoint=None, _custody_carry_note(carried)) # A rung earned HERE is what a later copy elsewhere will inherit. Recorded # after the grading decision, never before it, so the ledger only ever - # holds rungs that were actually awarded. - _custody_remember(sha or _graded_sha(path), rung, path) + # holds rungs that were actually awarded. Except a platform signature: it + # earned the rung only BECAUSE of where it runs, so a copy of those bytes + # elsewhere must not inherit even `copy-of-graded` from it. + if not (signer and signer.get("trust") in _PLATFORM_SIGNATURES): + _custody_remember(sha or _graded_sha(path), rung, path) note = _PROVENANCE_NOTE.get(rung) if signer: note = "%s %s" % (_publisher_line(signer), note) @@ -22363,6 +22428,10 @@ def cmd_incident(incident_id, action=None, reason=None): item["correlation_key"], datetime.fromtimestamp(item["created_at"]).isoformat(), datetime.fromtimestamp(item["updated_at"]).isoformat())) + if item.get("digest_only"): + print(" routed: digest only (%s) — no reminders; it notifies only " + "if evidence that would itself interrupt attaches" + % item["digest_only"]) # The honest statement of a standing fact: one line with a count, not one # evidence row per scan (still-true re-observations fold into the count). if (item.get("occurrences") or 0) > 1: diff --git a/tests/test_provenance_gate.py b/tests/test_provenance_gate.py index 7e83f6b..1f8d8e0 100644 --- a/tests/test_provenance_gate.py +++ b/tests/test_provenance_gate.py @@ -25,17 +25,17 @@ import contextlib import io import os +import shutil import sys import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from conftest import (PUBLISHER_TRUST, SUSPICIOUS_TRUST, # noqa: E402 - aegis) +from conftest import SUSPICIOUS_TRUST, aegis # noqa: E402 from test_codesign_detritus import (CLEAN_VERIFY, DEV_ID_DV, # noqa: E402 TIMED_OUT) from test_regression import Sandbox # noqa: E402 -from test_routing_gate import GateSandbox # noqa: E402 +from test_routing_gate import NOW, GateSandbox # noqa: E402 # Gated on BOTH, as test_signature_corpus is: CI runs a Linux leg with # aegis.IS_MAC forced on, and that leg has no codesign to be right about. @@ -48,11 +48,32 @@ PUBLISHER_NAME = "Example Publisher Ltd" +def _vendor_trust(): + """A publisher verdict in THIS body's vocabulary that belongs to a + third-party vendor, and so is valid wherever the binary runs: Developer + ID on macOS, a non-Microsoft Authenticode chain on Windows, a distro + package on Linux (whose claim is already about the package's own path). + Not conftest.PUBLISHER_TRUST: that is `apple` / `os-signed`, the + platform's own signature, which earns the rung only in the system tree.""" + if aegis.IS_LINUX: + return "os-managed" + if aegis.IS_WIN: + return "signed-valid" + return "developer-id" + + +def _platform_trust(): + """The verdict that means "the OS vendor's own platform binary", or None + on a body that has no such spelling.""" + if aegis.IS_LINUX: + return None + return "os-signed" if aegis.IS_WIN else "apple" + + def _publisher_verdict(): - """A classify_signature() answer publisher_sig() accepts on THIS body - (conftest.PUBLISHER_TRUST), naming its signer. The Developer ID shape is - macOS vocabulary and lives in DeveloperIdRung, which is gated.""" - return {"trust": PUBLISHER_TRUST, "team": None, + """A classify_signature() answer publisher_sig() accepts on THIS body at + any location, naming its signer.""" + return {"trust": _vendor_trust(), "team": None, "authority": PUBLISHER_NAME} @@ -171,9 +192,13 @@ class _RungSandbox(Sandbox): def setUp(self): super().setUp() + # setdefault, never assignment: on a Windows body Sandbox has already + # pinned classify_signature and saved the REAL one; overwriting that + # entry would "restore" Sandbox's stub at tearDown and leak it into + # every later test (found by the simbody win leg on PR #60). for name in ("classify_signature", "_vouch_covers", "_package_receipt", "_build_output_rung", "run"): - self._saved[name] = getattr(aegis, name) + self._saved.setdefault(name, getattr(aegis, name)) aegis._CUSTODY_CARRY_CACHE.clear() aegis._GRADED_SHA_CACHE.clear() aegis._SIG_UNANSWERED.clear() @@ -225,7 +250,7 @@ def test_unproven_signatures_earn_nothing(self): self.assertEqual(("HIGH", None), (sev, rung)) def test_a_publisher_verdict_that_names_no_signer_earns_nothing(self): - self.verdict(trust=PUBLISHER_TRUST, team=None, authority=None) + self.verdict(trust=_vendor_trust(), team=None, authority=None) self.assertEqual(("HIGH", None), aegis._grade_binary("HIGH", self.bin)[:2]) @@ -253,6 +278,25 @@ def test_it_is_asked_before_build_output(self): self.assertEqual("build-output", aegis._grade_binary("HIGH", self.bin)[1]) + def test_the_platforms_no_withholds_the_rung(self): + """A caller whose own evidence says the platform's control refused + these bytes (the hot-dir sensor, after Gatekeeper rejected the + bundle) withholds the rung, and the classifier is not even asked.""" + self.verdict(**_publisher_verdict()) + self.assertEqual(("HIGH", None), + aegis._grade_binary("HIGH", self.bin, + publisher_ok=False)[:2]) + self.assertEqual([], self.asked) + + def test_a_vendor_signature_under_home_earns_the_rung(self): + """The corpus shape: Codex's helper under ~/.codex, Developer ID.""" + aegis._build_output_rung = lambda path: None + home_bin = os.path.join(aegis.HOME, ".codex", "computer-use", + "SkyComputerUseService") + self.verdict(**_publisher_verdict()) + self.assertEqual(("MEDIUM", "publisher-signed"), + aegis._grade_binary("HIGH", home_bin)[:2]) + def test_attack_defined_evidence_earns_nothing(self): self.verdict(**_publisher_verdict()) self.assertEqual(("HIGH", None, None), @@ -269,6 +313,58 @@ def test_the_rung_is_a_vouched_origin_one_step_down(self): self.assertTrue(aegis._PROVENANCE_NOTE.get("publisher-signed")) +@unittest.skipIf(_platform_trust() is None, + "no platform-binary signature on this body") +class PlatformSignatureStaysInTheSystemTree(_RungSandbox): + """`apple` / `os-signed` is the OS vendor's own binary. Where it runs + from the sealed system tree its origin is proven; a copy of it anywhere + else -- /tmp, $HOME, /Users/Shared, %TEMP% -- is the living-off-the-land + shape, and the signature on the bytes says nothing about who put them + there.""" + + def setUp(self): + super().setUp() + aegis._build_output_rung = lambda path: None + # Built from the tree itself rather than a known binary: on a + # simulated body WIN_SYSTEMROOT is empty and the prefixes are the + # host's, and the rule under test is "inside TRUSTED_PREFIXES". + self.system_bin = os.path.join(aegis.TRUSTED_PREFIXES[0], + "aegis-platform-probe") + self.remembered = [] + self._saved.setdefault("_custody_remember", aegis._custody_remember) + aegis._custody_remember = ( + lambda sha, rung, path: self.remembered.append((rung, path))) + + def platform(self): + self.verdict(trust=_platform_trust(), team=None, + authority="Software Signing") + + def test_inside_the_system_tree_it_earns_the_rung(self): + self.platform() + self.assertEqual(("MEDIUM", "publisher-signed"), + aegis._grade_binary("HIGH", self.system_bin)[:2]) + + def test_a_copy_outside_the_system_tree_earns_nothing(self): + self.platform() + for path in (self.bin, os.path.join(aegis.HOME, "Library", "nc")): + with self.subTest(path=path): + self.assertEqual(("HIGH", None), + aegis._grade_binary("HIGH", path)[:2]) + + def test_the_system_tree_rung_is_not_carried_to_a_copy(self): + """Carried custody would otherwise hand a copy `copy-of-graded` on + the strength of the original's location, so the rung is not written + to the ledger. A vendor's signature, which is not about location, + still is.""" + self.platform() + self.assertEqual("publisher-signed", + aegis._grade_binary("HIGH", self.system_bin)[1]) + self.assertEqual([], self.remembered) + self.verdict(**_publisher_verdict()) + aegis._grade_binary("HIGH", self.bin, sha="f" * 64) + self.assertEqual([("publisher-signed", self.bin)], self.remembered) + + class DeveloperIdRung(_RungSandbox): """The macOS half: a Developer ID verdict, and what codesign costs. @@ -367,6 +463,21 @@ def tearDown(self): aegis._GRADED_SHA_CACHE.clear() super().tearDown() + def test_a_copied_platform_binary_earns_nothing(self): + """The real tool, both halves: /bin/ls in place is Apple's and earns + the rung; the same bytes copied into a user-writable temp dir keep + Apple's signature and earn nothing -- not even `copy-of-graded`.""" + self.assertEqual(("LOW", "publisher-signed"), + aegis._grade_binary("MEDIUM", "/bin/ls")[:2]) + copy = os.path.join(self.tmp, "ls") + shutil.copyfile("/bin/ls", copy) + os.chmod(copy, 0o755) + self.assertEqual("apple", aegis.classify_signature(copy)["trust"], + "fixture: the copy should still verify as Apple's") + self.assertTrue(aegis.is_risky_location(copy), copy) + self.assertEqual(("HIGH", None), + aegis._grade_binary("HIGH", copy)[:2]) + def test_spotify_grades_publisher_signed(self): if not os.path.exists(SPOTIFY): self.skipTest("Spotify is not installed here") @@ -377,5 +488,124 @@ def test_spotify_grades_publisher_signed(self): self.assertIn("Developer ID team %s" % team, note) +class DigestOnlyIncidentsAreNeverEscalated(GateSandbox): + """An incident opened only by digest-routed evidence -- the provenance + gate, low confidence, anything below the interrupt tier -- is recorded + as such, and no reminder ever turns it into a notification. It notifies + only when evidence that would itself interrupt attaches to it.""" + + def case(self, f): + rows = [i for i in self.incidents() + if i["correlation_key"] == "signal:" + f["case_fingerprint"]] + self.assertEqual(1, len(rows), rows) + return rows[0] + + def assert_never_reminded(self): + for days in (0.05, 1, 3, 30): + self.assertEqual( + [], aegis.claim_due_incident_reminders( + now=NOW + int(days * 86400)), days) + + def test_a_provenance_digest_incident_is_never_reminded(self): + f = self.process(SKY, "c" * 64, custody="publisher-signed") + _routing, new_high = self.scan_path([f]) + self.assertEqual([], new_high) + inc = self.case(f) + self.assertEqual("provenance:publisher-signed", inc["digest_only"]) + self.assertIsNone(inc["next_reminder_at"]) + self.assertIsNone(inc["last_notified_at"]) + self.assert_never_reminded() + self.assertEqual([], self.notified) + out = io.StringIO() + with contextlib.redirect_stdout(out): + aegis.cmd_incident(inc["id"]) + self.assertIn("digest only (provenance:publisher-signed)", + out.getvalue()) + + def test_a_low_confidence_digest_incident_is_never_reminded(self): + f = self.process(SKY, "e" * 64, confidence="low") + _routing, new_high = self.scan_path([f]) + self.assertEqual([], new_high) + self.assertEqual("low-confidence", self.case(f)["digest_only"]) + self.assert_never_reminded() + self.assertEqual([], self.notified) + + def test_a_reopen_by_hand_does_not_arm_reminders(self): + f = self.process(SKY, "c" * 64, custody="publisher-signed") + self.scan_path([f]) + inc = self.case(f) + self.assertTrue(aegis.transition_incident(inc["id"], "ACK", now=NOW)) + self.assertTrue(aegis.transition_incident(inc["id"], "RESOLVED", + now=NOW)) + self.assertTrue(aegis.transition_incident(inc["id"], "OPEN", now=NOW)) + self.assert_never_reminded() + + def test_a_seen_reobservation_does_not_escalate(self): + f = self.process(SKY, "c" * 64, custody="publisher-signed") + self.scan_path([f]) + self.scan_path([dict(f)]) + self.assertEqual("provenance:publisher-signed", + self.case(f)["digest_only"]) + self.assert_never_reminded() + + def test_an_interrupting_finding_escalates_it(self): + quiet = self.process(SKY, "c" * 64, custody="publisher-signed") + self.scan_path([quiet]) + loud = self.process(SKY, "d" * 64) # same case, no custody + routing, new_high = self.scan_path([loud]) + self.assertEqual("interrupt", routing[loud["fingerprint"]]["route"]) + self.assertEqual(1, len(new_high)) + inc = self.case(loud) + self.assertIsNone(inc["digest_only"]) + self.assertEqual(NOW, inc["last_notified_at"]) + self.assertEqual(NOW + aegis._REMINDER_DELAYS[0], + inc["next_reminder_at"]) + due = aegis.claim_due_incident_reminders( + now=NOW + aegis._REMINDER_DELAYS[0]) + self.assertEqual([inc["id"]], [d["id"] for d in due]) + + def test_an_interrupt_opened_incident_still_reminds(self): + f = self.process(SKY, "f" * 64) + self.scan_path([f]) + inc = self.case(f) + self.assertIsNone(inc["digest_only"]) + self.assertEqual(NOW + aegis._REMINDER_DELAYS[0], + inc["next_reminder_at"]) + due = aegis.claim_due_incident_reminders( + now=NOW + aegis._REMINDER_DELAYS[0]) + self.assertEqual([inc["id"]], [d["id"] for d in due]) + + def test_digest_evidence_does_not_mute_a_notified_incident(self): + loud = self.process(SKY, "d" * 64) + self.scan_path([loud]) + quiet = self.process(SKY, "c" * 64, custody="publisher-signed") + self.scan_path([quiet]) + inc = self.case(loud) + self.assertIsNone(inc["digest_only"]) + self.assertEqual(NOW + aegis._REMINDER_DELAYS[0], + inc["next_reminder_at"]) + + def test_a_store_that_predates_the_mark_gains_it(self): + aegis.EVENT_DB = os.path.join(self.tmp, "old.db") + import sqlite3 + raw = sqlite3.connect(aegis.EVENT_DB) + raw.executescript( + "CREATE TABLE incidents (id INTEGER PRIMARY KEY, kind TEXT, " + "correlation_key TEXT, title TEXT, severity TEXT, status TEXT, " + "created_at INTEGER, first_seen INTEGER, last_seen INTEGER, " + "updated_at INTEGER, reminder_count INTEGER DEFAULT 0, " + "next_reminder_at INTEGER, last_notified_at INTEGER, " + "resolution TEXT, subject_json TEXT, last_novel_at INTEGER);" + "PRAGMA user_version=1;") + raw.commit() + raw.close() + db = aegis._event_connection() + try: + cols = {r[1] for r in db.execute("PRAGMA table_info(incidents)")} + finally: + db.close() + self.assertIn("digest_only", cols) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_regression.py b/tests/test_regression.py index 5627fd9..f04853e 100644 --- a/tests/test_regression.py +++ b/tests/test_regression.py @@ -328,6 +328,14 @@ def _sandboxed_win_persistence(): "trust": "unsigned", "team": None, "authority": None} self.notifications = [] + # The ledger FILE is sandboxed above; its in-process memos are not a + # path, so they are cleared here and again at tearDown. Without this a + # rung one test awarded to a clang-built fixture was served, from the + # memo, to every later test whose fixture had the same bytes -- on + # Linux clang output is deterministic, so that was all of them. + aegis._CUSTODY_CARRY_CACHE.clear() + aegis._GRADED_SHA_CACHE.clear() + aegis._SIG_UNANSWERED.clear() self._saved["notify"] = aegis.notify aegis.notify = lambda title, msg: self.notifications.append((title, msg)) # check_processes()/check_behavior() read the live process table, @@ -346,6 +354,9 @@ def _sandboxed_win_persistence(): def tearDown(self): for k, v in self._saved.items(): setattr(aegis, k, v) + aegis._CUSTODY_CARRY_CACHE.clear() + aegis._GRADED_SHA_CACHE.clear() + aegis._SIG_UNANSWERED.clear() for root, dirs, files in os.walk(self.tmp, topdown=True): os.chmod(root, 0o700) @@ -2351,17 +2362,35 @@ def test_unnotarized_devid_app_is_medium_with_verdict(self): aegis.classify_signature = saved_cs aegis.gatekeeper_verdict = saved_gk self.assertEqual(len(fs), 1, fs) - # MEDIUM until 2026-09-23. The Developer ID signature is now the - # publisher custody rung, one step down and routed to the digest; the - # Gatekeeper verdict and the notary fingerprint are unchanged. - self.assertEqual(fs[0]["severity"], "LOW") - self.assertEqual(fs[0]["provenance"], "publisher-signed") - self.assertIn("Developer ID team T (X)", fs[0]["detail"]) + self.assertEqual(fs[0]["severity"], "MEDIUM") self.assertEqual(fs[0]["gatekeeper"], "rejected") self.assertTrue(fs[0]["fingerprint"].startswith("hotdir:notary:")) + # The publisher rung leans on Apple's notarization, and Gatekeeper + # just refused this bundle: the control said no, so no rung and no + # provenance gate -- the finding routes on its severity alone. + self.assertIsNone(fs[0]["provenance"]) route = aegis.route_findings(fs, seen={})[fs[0]["fingerprint"]] - self.assertEqual(("digest", "provenance:publisher-signed"), - (route["route"], route["why"])) + self.assertEqual("below-floor", route["why"]) + + def test_a_notarized_devid_app_is_silent_and_its_binary_earns_the_rung(self): + """The same folder, the other answer: Gatekeeper accepts it, the + hot-dir sensor says nothing, and when the process or beacon sensor + grades its executable the Developer ID signature is a rung.""" + app, exe = self._mk_app("Signed.app") + saved_cs, saved_gk = aegis.classify_signature, aegis.gatekeeper_verdict + aegis.classify_signature = lambda p: { + "trust": "developer-id", "team": "T", + "authority": "Developer ID Application: X (T)"} + aegis.gatekeeper_verdict = lambda p: ("accepted", + "Notarized Developer ID") + try: + fs = [f for f in aegis.check_hot_dirs() if f.get("path") == app] + graded = aegis._grade_binary("HIGH", exe) + finally: + aegis.classify_signature = saved_cs + aegis.gatekeeper_verdict = saved_gk + self.assertEqual(fs, []) + self.assertEqual(("MEDIUM", "publisher-signed"), graded[:2]) def test_notarized_app_is_silent(self): app, _exe = self._mk_app("Fine.app") diff --git a/tests/test_routing_gate.py b/tests/test_routing_gate.py index 6c575a5..4cb683c 100644 --- a/tests/test_routing_gate.py +++ b/tests/test_routing_gate.py @@ -184,11 +184,16 @@ def test_an_allowlisted_fingerprint_closes_its_incident(self): class TestNotifiedIsPerFinding(GateSandbox): - def test_a_digest_routed_signal_still_gets_its_reminder(self): + def test_a_digest_routed_signal_is_not_recorded_as_told(self): """initially_notified was bool(new_high) for the whole scan: one genuine new HIGH marked every incident created that scan as already - notified, so a low-confidence HIGH routed to the digest never got the - reminder that was its only path to a human.""" + notified. That half still holds. The reminder half was reversed on + 2026-09-23 (precision plan S5): this test used to require that the + digest-routed incident be REMINDED, as its only path to a human. A + digest-only incident is now never claimed by a reminder -- it stays + OPEN, counted and listed, and notifies only when evidence that would + itself interrupt attaches (see test_provenance_gate.py + DigestOnlyIncidentsAreNeverEscalated).""" loud = self.process("/opt/a/bin/a", "1" * 64) quiet = self.process("/opt/b/bin/b", "2" * 64, confidence="low") routing, new_high = self.scan_path([loud, quiet]) @@ -199,11 +204,13 @@ def test_a_digest_routed_signal_still_gets_its_reminder(self): by_key["signal:" + loud["case_fingerprint"]]["last_notified_at"]) self.assertIsNone( by_key["signal:" + quiet["case_fingerprint"]]["last_notified_at"]) - # Reminders are scheduled for every OPEN incident; the difference the - # gate makes is that the quiet one is not falsely recorded as told. - due = aegis.claim_due_incident_reminders(NOW + 7200) - self.assertIn("signal:" + quiet["case_fingerprint"], - [i["correlation_key"] for i in due]) + self.assertEqual( + "low-confidence", + by_key["signal:" + quiet["case_fingerprint"]]["digest_only"]) + due = [i["correlation_key"] + for i in aegis.claim_due_incident_reminders(NOW + 7200)] + self.assertIn("signal:" + loud["case_fingerprint"], due) + self.assertNotIn("signal:" + quiet["case_fingerprint"], due) class TestRoutePrecedence(GateSandbox):