From 3ff8ea285242c61236c1442daf26e52cfd862a1e Mon Sep 17 00:00:00 2001 From: "C.D.Lee" <265386841+opencdlee-dotcom@users.noreply.github.com> Date: Wed, 23 Sep 2026 21:16:13 -0700 Subject: [PATCH] Incidents opened by fixed code stayed open for a week, because nothing asked the current code again MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every machine exit waits for the sensor to say something new: re-grade reads the newest evidence, re-verify re-asks one signature, cleared-state and removed-file need the sensor to look again. An incident opened by code that has since been fixed hears none of that unless the same subject is re-observed, so it waits out the 7-day age-out. Live: #527, a Spotify risk case built from listener findings recorded `unsigned` while codesign was not answering (Spotify is Developer ID), and #537, a staging plugin-container the current code grades MEDIUM on its build-output rung. _rejudge_open_incidents(db, now) runs in record_security_state after the removed-file exit. For each OPEN/ACK signal, risk or chain incident created before this scan, every observation finding it holds is re-derived through _reobserve, the path `backtest replay --reobserve` scores with, with the live suppression memory and dismissal weights, the learning period off and the seen-ledger empty. It stands if any evidence is CRITICAL, attack-defined or never-tolerate, or if any finding is replayed as recorded (gone from disk, unmodelled sensor, missing field). Then, by kind: - signal: every finding dropped, or routed below the interrupt tier by route_findings; - risk: re-scored by _risk_buckets / _risk_score over the findings inside one RISK_WINDOW at each moment one was observed, as _accumulate_risk scores it, and no window may still cross. Summed whole, a program that picks a new port per launch is 68 distinct signals (#527); per window it peaks at 0.9; - correlation: a chain the current join rules would not form (_unjoinable_chain_resolution); lineage chains are left alone. A closed incident goes to FALSE_POSITIVE with `re-judged by current code (logic :): — reopens on new evidence`. It writes no dismissals row and no custody ledger row (the ladder is asked with _custody_remember off). It runs at most hourly, at once when _REJUDGE_LOGIC_VERSION or the running aegis.py changes, and is capped by _REJUDGE_MAX_INCIDENTS (25) and _REJUDGE_BUDGET (30 s) per run. A capped run logs how many it left, and the next resumes after the last incident it examined. Also touched: _accumulate_risk. Its per-entity summing and scoring are extracted as _risk_buckets / _risk_score, with no behaviour change, so a pile is re-scored by the code that opened it. _backtest_replay's counter setup is now _reobserve_stats(), shared with the healer. Sandboxed real scan of this code against a copy of the live state: #527 and #537 closed as re-judged. #526 (subject gone), #539 and #540 (rustup, still HIGH with no rung) stay. No incident was opened, no outward effect was taken, and the real custody ledger read 151 lines before and after. Tests: tests/test_rejudge_open_incidents.py (16). Co-Authored-By: Claude Opus 5.5 (1M context) --- aegis.py | 319 ++++++++++++++++++++++++--- tests/test_rejudge_open_incidents.py | 314 ++++++++++++++++++++++++++ 2 files changed, 601 insertions(+), 32 deletions(-) create mode 100644 tests/test_rejudge_open_incidents.py diff --git a/aegis.py b/aegis.py index 1923977..0a15ab6 100755 --- a/aegis.py +++ b/aegis.py @@ -4869,13 +4869,38 @@ def _accumulate_risk(db, now, new_ids): "SELECT id, observed_at, data_json FROM events " "WHERE event_type='observation.finding' AND observed_at>=?", (now - RISK_WINDOW,)).fetchall() - demote = _category_dismissal_weights(db, now) - by_entity = {} + findings = [] for row in rows: try: - f = json.loads(row["data_json"]) + findings.append((row["id"], json.loads(row["data_json"]))) except Exception: continue + by_entity = _risk_buckets(findings, _category_dismissal_weights(db, now), + new_ids) + for ek, b in by_entity.items(): + score, min_signals = _risk_score(b) + if b["new"] and len(b["fps"]) >= min_signals \ + and score >= RISK_THRESHOLD: + # Severity follows the score instead of a hardcoded "HIGH": + # barely past threshold is a MEDIUM worth a look, not an + # interrupt-grade verdict the number never supported. + sev = "HIGH" if score >= 2 * RISK_THRESHOLD else "MEDIUM" + _upsert_incident( + db, "risk:%s" % ek, + "Accumulated risk on %s (%d signals across %d sensor%s, score %.1f)" + % (b["entity"][:80], len(b["fps"]), len(b["cats"]), + "" if len(b["cats"]) == 1 else "s", score), + sev, "risk", now, sorted(b["ids"])) + + +def _risk_buckets(findings, demote, new_ids=()): + """{entity key: {weight, fps, ids, cats, entity, new}} for [(event id, + finding)]: what _accumulate_risk sums per entity before it scores, with + `demote` the per-category dismissal weights. One function, so a pile + re-scored later (_rejudge_open_incidents) is scored the way it was + opened.""" + by_entity = {} + for event_id, f in findings: # Single primary entity on purpose, unlike the chain rules: accumulation # counts weak signals piling up on ONE object, so bucketing a finding # under its every identity would both double-count it and pool unrelated @@ -4951,29 +4976,22 @@ def _accumulate_risk(db, now, new_ids): continue # count each distinct signal once, not once per rescan b["fps"].add(fp) b["weight"] += w - b["ids"].add(row["id"]) + b["ids"].add(event_id) b["cats"].add(_RISK_SENSOR_GROUP.get(category, category)) - if row["id"] in new_ids: + if event_id in new_ids: b["new"] = True - for ek, b in by_entity.items(): - # Corroboration across sensors is the higher-precision evidence: it needs - # fewer signals AND scores higher against the same constant threshold. - # One sensor keeps the original bar, so no existing detection regresses. - multi = len(b["cats"]) >= 2 - min_signals = RISK_MIN_SIGNALS_MULTI_SENSOR if multi else RISK_MIN_SIGNALS - score = b["weight"] * (RISK_CORROBORATION_BONUS if multi else 1.0) - if b["new"] and len(b["fps"]) >= min_signals \ - and score >= RISK_THRESHOLD: - # Severity follows the score instead of a hardcoded "HIGH": - # barely past threshold is a MEDIUM worth a look, not an - # interrupt-grade verdict the number never supported. - sev = "HIGH" if score >= 2 * RISK_THRESHOLD else "MEDIUM" - _upsert_incident( - db, "risk:%s" % ek, - "Accumulated risk on %s (%d signals across %d sensor%s, score %.1f)" - % (b["entity"][:80], len(b["fps"]), len(b["cats"]), - "" if len(b["cats"]) == 1 else "s", score), - sev, "risk", now, sorted(b["ids"])) + return by_entity + + +def _risk_score(b): + """(score, signals needed) for one _risk_buckets entry.""" + # Corroboration across sensors is the higher-precision evidence: it needs + # fewer signals AND scores higher against the same constant threshold. + # One sensor keeps the original bar, so no existing detection regresses. + multi = len(b["cats"]) >= 2 + min_signals = RISK_MIN_SIGNALS_MULTI_SENSOR if multi else RISK_MIN_SIGNALS + return (b["weight"] * (RISK_CORROBORATION_BONUS if multi else 1.0), + min_signals) # --------------------------------------------------------------------------- # @@ -6316,6 +6334,233 @@ def _close_removed_drop_incidents(db, observed, now): return len(closed) +# Re-judging open incidents with the current code (_rejudge_open_incidents). +# Bump _REJUDGE_LOGIC_VERSION when what may close changes; a new aegis.py is a +# new logic too (its code sha is part of the key), so an install re-judges on +# its first scan instead of an hour later. +_REJUDGE_LOGIC_VERSION = 1 +_REJUDGE_INTERVAL = 3600 +_REJUDGE_MAX_INCIDENTS = 25 # per run; the rest wait for the next one +_REJUDGE_BUDGET = 30.0 # seconds of re-derivation per run +_REJUDGE_KINDS = ("signal", "risk", "correlation") + + +def _rejudge_logic(): + """What judged: this closer's version and the code it ran in.""" + return "%d:%s" % (_REJUDGE_LOGIC_VERSION, + (_running_code_sha() or "unknown")[:12]) + + +def _rejudge_subject(f): + """` `, the way a resolution names a finding.""" + return ("%s %s" % (f.get("category") or "finding", os.path.basename( + str(f.get("path") or f.get("program") or "")))).strip() + + +def _rejudge_change(f, g): + """What the current code says differently about `f`, re-derived as `g`.""" + parts = [] + if g.get("trust") and (g.get("trust") or None) != (f.get("trust") or None): + parts.append("now classifies %s" % g["trust"]) + rung = g.get("custody") or g.get("provenance") + if rung and rung != (f.get("custody") or f.get("provenance")): + parts.append("custody %s" % rung) + if g["severity"] != f["severity"]: + parts.append("%s -> %s" % (f["severity"], g["severity"])) + return "%s: %s" % (_rejudge_subject(f), ", ".join(parts) or "as recorded") + + +def _rejudge_summary(reasons, limit=4): + distinct = [] + for why in reasons: + if why not in distinct: + distinct.append(why) + extra = len(distinct) - limit + return "; ".join(distinct[:limit]) + ("; +%d more" % extra + if extra > 0 else "") + + +def _rejudge_incident(db, row, memory, demote, memo, stats): + """Why the current code would not raise incident `row` today, as one + line, or None when it would, or when that cannot be shown. + + Every observation finding it holds is re-derived by `_reobserve`, the + path `backtest replay --reobserve` scores with. Any finding replayed as + recorded -- a subject gone from disk, a sensor it does not model, a field + the record never carried -- and the incident stands: absence of an answer + is not an answer. So does any CRITICAL, attack-defined or never-tolerate + evidence, whatever the current grade. Then, by kind: a signal case needs + every finding dropped or routed below the interrupt tier; a risk pile is + scored again (_risk_buckets, _risk_score) the way _accumulate_risk scores + it -- over the findings inside one RISK_WINDOW, at each moment one of them + was observed -- and no window may still cross; a co-occurrence chain must + be one the current join rules would not form + (_unjoinable_chain_resolution).""" + evidence = [] + for ev in db.execute( + "SELECT e.id, e.observed_at, e.data_json FROM incident_events ie " + "JOIN events e ON e.id=ie.event_id WHERE ie.incident_id=? AND " + "e.event_type='observation.finding' ORDER BY e.id", (row["id"],)): + try: + f = json.loads(ev["data_json"]) + except (ValueError, TypeError): + return None + if not isinstance(f, dict) or not f.get("fingerprint") \ + or f.get("severity") not in SEV_ORDER: + return None + if SEV_ORDER[f["severity"]] >= SEV_ORDER["CRITICAL"] \ + or f.get("attack_defined") \ + or str(f["fingerprint"]).startswith(_NEVER_TOLERATE_PREFIXES): + return None + evidence.append((ev["id"], ev["observed_at"] or 0, f)) + if not evidence: + return None + judged = [] + for event_id, at, f in evidence: + g, as_recorded = _reobserve(dict(f), memo, stats) + if as_recorded: + return None + judged.append((event_id, at, f, g)) + kind, key = row["kind"], row["correlation_key"] or "" + if kind == "signal": + reasons, routed = [], set() + for _event_id, _at, f, g in judged: + if g is None: + reasons.append("%s: no longer emitted" % _rejudge_subject(f)) + continue + if g["fingerprint"] in routed: + continue # one fact, re-derived once per scan held + routed.add(g["fingerprint"]) + verdict = route_findings([dict(g)], memory=memory, + seen={})[g["fingerprint"]] + if verdict["route"] == ROUTE_INTERRUPT: + return None + reasons.append("%s (%s: %s)" % (_rejudge_change(f, g), + verdict["route"], verdict["why"])) + return _rejudge_summary(reasons) + kept = [(event_id, at, g) for event_id, at, _f, g in judged + if g is not None] + if kind == "risk" and key.startswith("risk:"): + # Each scan scores the findings of the last RISK_WINDOW, so the + # evidence of an incident that re-crossed for days is many piles, + # never one: summed whole, a program that picks a new port per launch + # is 68 distinct signals (#527). Every window the evidence holds is + # asked, and the highest is reported. + peak = (0.0, 0, RISK_MIN_SIGNALS) + for anchor in sorted({at for _event_id, at, _g in kept}): + b = _risk_buckets( + [(event_id, g) for event_id, at, g in kept + if anchor - RISK_WINDOW <= at <= anchor], + demote).get(key[len("risk:"):]) + if not b: + continue + score, needed = _risk_score(b) + if len(b["fps"]) >= needed and score >= RISK_THRESHOLD: + return None + peak = max(peak, (score, len(b["fps"]), needed)) + changes = [_rejudge_change(f, g) if g is not None + else "%s: no longer emitted" % _rejudge_subject(f) + for _event_id, _at, f, g in judged] + return _rejudge_summary(changes + [ + "the pile now peaks at %.1f from %d signal(s) in any %d-minute " + "window, under the risk threshold (%.1f from %d)" + % (peak[0], peak[1], RISK_WINDOW // 60, RISK_THRESHOLD, peak[2])]) + if kind == "correlation" and key.startswith("chain:") \ + and not key.startswith("chain:lineage:"): + why = _unjoinable_chain_resolution(_chain_entity_key(key), + [g for _event_id, _at, g in kept]) + if not why: + return None + return why.replace("superseded: ", "", 1).replace( + " — reopens on new evidence", "") + return None + + +def _rejudge_open_incidents(db, now): + """Close OPEN/ACK incidents the CURRENT code would not raise; the number + closed, or None when this run was throttled. + + Every other exit waits for the sensor to say something: re-grade reads + the newest evidence, re-verify re-asks one signature, cleared-state and + removed-file need the sensor to look again. An incident opened by code + that has since been fixed hears none of that unless the same subject is + re-observed, so it waited out the age-out clock. Live, 2026-09-23: #527, + a Spotify risk case built from listener findings recorded `unsigned` + while codesign was not answering (Spotify is Developer ID), and #537, a + staging plugin-container the current code grades MEDIUM on its + build-output rung. The harness already re-derives recorded evidence with + the current code; this uses the same path to heal, not only to score. + + Judged with the live teaching (the suppression memory, the dismissal + weights), the learning period off -- the question is whether the code + would raise this, not whether the machine is still learning -- and the + seen-ledger empty, since having told the operator once is not a verdict. + Read-only on everything but the incident row: the custody ledger is not + written while the ladder is asked. See _rejudge_incident for what stands. + + The discipline is the other machine exits': FALSE_POSITIVE with a + resolution that says why and which logic judged it, no dismissals row + (a machine verdict never feeds precision or tolerance), and new evidence + opens the case again. Never an incident created by the scan that runs it. + + Hourly at most, and at once when the logic changes (a new version or a + new aegis.py). Bounded per run by _REJUDGE_MAX_INCIDENTS and + _REJUDGE_BUDGET; a run that stops early says how many it left, and the + next resumes after the last incident it examined.""" + logic = _rejudge_logic() + meta = {r["key"]: r["value"] for r in db.execute( + "SELECT key, value FROM meta WHERE key IN " + "('rejudge_logic','rejudge_at','rejudge_cursor')")} + try: + last, cursor = int(meta.get("rejudge_at") or 0), \ + int(meta.get("rejudge_cursor") or 0) + except ValueError: + last, cursor = 0, 0 + if meta.get("rejudge_logic") == logic and now - last < _REJUDGE_INTERVAL: + return None + marks = ",".join("?" for _ in _REJUDGE_KINDS) + rows = db.execute( + "SELECT id, kind, correlation_key FROM incidents WHERE status IN " + "('OPEN','ACK') AND kind IN (%s) AND severity<>'CRITICAL' AND " + "created_at cursor] + \ + [r for r in rows if r["id"] <= cursor] + memory = _suppression_memory(db, now) + memory = tuple(memory[:3]) + (False,) + tuple(memory[4:]) + demote = _category_dismissal_weights(db, now) + memo, stats = {}, _reobserve_stats() + deadline = time.monotonic() + _REJUDGE_BUDGET + closed, examined = [], [] + with _replay_overrides(_custody_remember=lambda *_a, **_k: False): + for row in rows: + if len(examined) >= _REJUDGE_MAX_INCIDENTS \ + or time.monotonic() > deadline: + break + examined.append(row["id"]) + why = _rejudge_incident(db, row, memory, demote, memo, stats) + if why: + closed.append((row["id"], why)) + for incident_id, why in closed: + db.execute( + "UPDATE incidents SET status='FALSE_POSITIVE',resolution=?," + "updated_at=?,next_reminder_at=NULL WHERE id=? AND status IN " + "('OPEN','ACK')", + ("re-judged by current code (logic %s): %s — reopens on new " + "evidence" % (logic, why), now, incident_id)) + left = len(rows) - len(examined) + resume = (examined[-1] if examined else cursor) if left else 0 + for key, value in (("rejudge_logic", logic), ("rejudge_at", now), + ("rejudge_cursor", resume)): + db.execute("INSERT INTO meta(key,value) VALUES(?,?) ON CONFLICT(key) " + "DO UPDATE SET value=excluded.value", (key, str(value))) + if rows: + log_run("re-judged %d open incident(s) with current code (logic %s), " + "closed %d%s" % (len(examined), logic, len(closed), + "; %d left for the next run" % left + if left else "")) + return len(closed) + + def _age_out_incidents(db, now, days=_AGE_OUT_DAYS): """Close OPEN incidents that stopped producing evidence, and say so. @@ -7205,6 +7450,10 @@ def record_security_state(findings, sensor_health=(), now=None, log_run("closed %d removed-file incident(s)" % removed) except Exception as e: log_run("removed-file close skipped: %s" % e) + try: + _rejudge_open_incidents(db, now) + except Exception as e: + log_run("re-judge skipped: %s" % e) try: global _LAST_AGED_OUT aged = _age_out_incidents(db, now) @@ -30827,6 +31076,19 @@ def _reobserve_grade(base, path, endpoints, parents, rotating, memo): return graded, rung +def _reobserve_stats(): + """The counters _reobserve keeps, zeroed. Besides the counts: {reason: + count} for findings replayed as recorded because the record cannot be + rebuilt, {((sev, rung), (sev, rung) or None): count} for what re-deriving + changed on a persistence finding, and {reason: count} for re-derived + findings the sensor would not emit, where it says why.""" + stats = dict.fromkeys(("reobserved", "trust_changed", "custody_changed", + "changed", "severity_changed", "no_longer_emitted", + "gone"), 0) + stats.update(not_rederivable={}, persistence_changes={}, dropped_why={}) + return stats + + def _reobserve(f, memo, stats): """(finding, as_recorded): `f` re-observed with the CURRENT classifier and custody ladder — None when today's sensor would not emit it — and None; @@ -31031,14 +31293,7 @@ def _backtest_replay(days=30, reobserve=False, now=None): # reason that has nothing to do with the code being measured. memory = tuple(memory[:3]) + (False,) + tuple(memory[4:]) batches, bad, by_scan_id = _replay_batches(rows) - stats = dict.fromkeys(("reobserved", "trust_changed", "custody_changed", - "changed", "severity_changed", "no_longer_emitted", - "gone"), 0) - # {reason: count} for findings replayed as recorded because the record - # cannot be rebuilt, {((sev, rung), (sev, rung) or None): count} for - # what re-deriving changed on a persistence finding, and {reason: count} - # for re-derived findings the sensor would not emit, where it says why. - stats.update(not_rederivable={}, persistence_changes={}, dropped_why={}) + stats = _reobserve_stats() asked = sum(1 for _n, batch in batches for _i, f in batch if f["category"] in _REOBSERVE_CATEGORIES) routes, route_of, scratch_of, folded_into, finding_of = {}, {}, {}, {}, {} diff --git a/tests/test_rejudge_open_incidents.py b/tests/test_rejudge_open_incidents.py new file mode 100644 index 0000000..c5a1f9b --- /dev/null +++ b/tests/test_rejudge_open_incidents.py @@ -0,0 +1,314 @@ +"""Current code re-judges what it already opened. + +An incident opened under old code stays OPEN after the code that opened it is +fixed, because nothing asks again unless the same subject is re-observed: it +waits out the seven-day age-out. On the reference Mac, 2026-09-23: #527 was a +Spotify risk case built from listener findings recorded `unsigned` while the +codesign probe was not answering (Spotify is Developer ID), and #537 a staging +plugin-container the current code grades MEDIUM on its build-output rung. + +`backtest replay --reobserve` already re-derives recorded evidence with the +current code. _rejudge_open_incidents uses the same `_reobserve` path to heal: +an incident closes only when EVERY finding it holds re-derives (none replayed +as recorded, none gone) and the current code would no longer raise it — each +finding dropped or routed below the interrupt tier (signal), the pile-up +below the threshold (risk), the chain no longer formed (correlation). It is a +machine exit like re-grade and re-verify: FALSE_POSITIVE with a resolution +that says why, no dismissals row, reopened by new evidence. +""" +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__))) +import aegis # noqa: E402 +from conftest import SUSPICIOUS_TRUST # noqa: E402 +from test_backtest_replay import NOW, ReplaySandbox # noqa: E402 + + +class RejudgeSandbox(ReplaySandbox): + """A sandboxed store holding incidents opened by findings about binaries + in the sandbox. The process gate is held open and the rungs ahead of the + receipt are pinned: those are platform-shaped and not under test. Whether + the binary now has a package receipt is the one thing a test flips.""" + + def setUp(self): + super().setUp() + self.receipt = None + self.stub("classify_signature", lambda path: { + "trust": SUSPICIOUS_TRUST, "team": None, "authority": None}) + self.stub("_exec_alert", lambda path, trust: ( + "HIGH", "running from user-writable path")) + self.stub("_package_receipt", lambda path: self.receipt) + self.stub("_build_output_rung", lambda path: None) + self.bin = self.binary("tool") + + def binary(self, name): + path = os.path.join(self.tmp, name) + with open(path, "wb") as f: + f.write(name.encode() + b"\x00" * 16) + return path + + def db(self): + return aegis._event_connection() + + def record(self, findings, kind="signal", key=None, created=NOW - 86400, + severity="HIGH"): + """An OPEN incident holding `findings` as evidence.""" + db = self.db() + with db: + ids = [self.event(db, f, created) for f in findings] + key = key or "signal:" + (findings[0].get("case_fingerprint") + or findings[0]["fingerprint"]) + iid = db.execute( + "INSERT INTO incidents(kind,correlation_key,title,severity," + "status,created_at,first_seen,last_seen,updated_at) " + "VALUES(?,?,'t',?,'OPEN',?,?,?,?)", + (kind, key, severity, created, created, created, + created)).lastrowid + for ev in ids: + db.execute("INSERT INTO incident_events(incident_id,event_id) " + "VALUES(?,?)", (iid, ev)) + db.close() + return iid + + def rejudge(self, now=NOW): + db = self.db() + try: + with db: + return aegis._rejudge_open_incidents(db, now) + finally: + db.close() + + def status(self, iid): + db = self.db() + try: + row = db.execute("SELECT status, resolution FROM incidents " + "WHERE id=?", (iid,)).fetchone() + return row["status"], row["resolution"] + finally: + db.close() + + def dismissals(self): + db = self.db() + try: + return db.execute("SELECT COUNT(*) FROM dismissals").fetchone()[0] + finally: + db.close() + + +class ASignalIncidentCurrentCodeWouldNotRaise(RejudgeSandbox): + def test_closes_when_every_finding_re_derives_below_the_interrupt_tier(self): + """Recorded HIGH with no rung. The binary now has a package receipt: + the grader demotes it and the provenance gate routes it to the + digest, so the current code would not have raised this.""" + iid = self.record([self.process(self.bin, "a" * 64)]) + self.receipt = "brew:tool" + self.assertEqual(1, self.rejudge()) + status, resolution = self.status(iid) + self.assertEqual("FALSE_POSITIVE", status) + self.assertTrue(resolution.startswith("re-judged by current code"), + resolution) + self.assertIn("tool", resolution) + self.assertIn("package-managed", resolution) + self.assertIn("reopens on new evidence", resolution) + self.assertIn("logic %d" % aegis._REJUDGE_LOGIC_VERSION, resolution) + self.assertEqual(0, self.dismissals()) + + def test_stays_while_the_current_code_still_interrupts(self): + iid = self.record([self.process(self.bin, "a" * 64)]) + self.assertEqual(0, self.rejudge()) + self.assertEqual("OPEN", self.status(iid)[0]) + + def test_stays_when_any_finding_is_replayed_as_recorded(self): + """A subject gone from disk cannot be asked again, so nothing says + the current code would not raise it: conservative, it stays.""" + gone = os.path.join(self.tmp, "gone-tool") + iid = self.record([self.process(self.bin, "a" * 64), + self.process(gone, "b" * 64)]) + self.receipt = "brew:tool" + self.assertEqual(0, self.rejudge()) + self.assertEqual("OPEN", self.status(iid)[0]) + + def test_stays_on_critical_evidence(self): + f = dict(self.process(self.bin, "a" * 64), severity="CRITICAL") + iid = self.record([f], severity="CRITICAL") + self.receipt = "brew:tool" + self.rejudge() + self.assertEqual("OPEN", self.status(iid)[0]) + + def test_stays_on_attack_defined_evidence(self): + f = dict(self.process(self.bin, "a" * 64), attack_defined=True) + iid = self.record([f]) + self.receipt = "brew:tool" + self.rejudge() + self.assertEqual("OPEN", self.status(iid)[0]) + + def test_stays_on_a_never_tolerate_fingerprint(self): + f = aegis.finding("HIGH", "decoy", "A process read a decoy", "d", + "decoy:read:%s" % self.bin, path=self.bin) + iid = self.record([f]) + self.receipt = "brew:tool" + self.rejudge() + self.assertEqual("OPEN", self.status(iid)[0]) + + def test_stays_when_opened_by_this_scan(self): + iid = self.record([self.process(self.bin, "a" * 64)], created=NOW) + self.receipt = "brew:tool" + self.assertEqual(0, self.rejudge()) + self.assertEqual("OPEN", self.status(iid)[0]) + + def test_a_new_interrupting_finding_reopens_the_case(self): + f = self.process(self.bin, "a" * 64) + iid = self.record([f]) + self.receipt = "brew:tool" + self.assertEqual(1, self.rejudge()) + # The receipt is gone again and the binary changed: new evidence on + # the same case, and it interrupts. + self.receipt = None + fresh = self.process(self.bin, "c" * 64) + fresh["case_fingerprint"] = f["case_fingerprint"] + aegis.record_security_state([fresh], now=NOW + 60) + db = self.db() + try: + active = db.execute( + "SELECT id FROM incidents WHERE correlation_key=? AND status " + "IN ('OPEN','ACK')", ("signal:" + f["case_fingerprint"],) + ).fetchall() + finally: + db.close() + self.assertEqual(1, len(active)) + self.assertEqual("FALSE_POSITIVE", self.status(iid)[0]) + + +class ARiskIncidentIsRescored(RejudgeSandbox): + """Three sensors on one binary, recorded with no rung, pile past the + threshold. Re-derived, the pile is scored again with the current weights + (a vouched rung weighs nothing).""" + + def pile(self): + sha = "d" * 64 + findings = [ + self.process(self.bin, sha), + aegis.finding( + "HIGH", "net-beacon", + "Persistent outbound connection (beacon shape)", "d", + "beacon:%s:203.0.113.7:443" % self.bin, path=self.bin, + program=self.bin, trust=SUSPICIOUS_TRUST, + remote="203.0.113.7", port=443), + aegis.finding( + "MEDIUM", "net-listener", "New network listener", "d", + "listener:%s:4444" % self.bin, path=self.bin, + program=self.bin, trust=SUSPICIOUS_TRUST, port=4444), + ] + db = self.db() + with db: + ids = [self.event(db, f, NOW - 86400) for f in findings] + aegis._accumulate_risk(db, NOW - 86400, set(ids)) + row = db.execute("SELECT id FROM incidents WHERE kind='risk'" + ).fetchone() + db.close() + self.assertIsNotNone(row, "the pile did not open a risk incident") + return row[0] + + def test_closes_once_it_no_longer_crosses_the_threshold(self): + iid = self.pile() + self.receipt = "brew:tool" + self.assertEqual(1, self.rejudge()) + status, resolution = self.status(iid) + self.assertEqual("FALSE_POSITIVE", status) + self.assertIn("threshold", resolution) + self.assertEqual(0, self.dismissals()) + + def test_stays_while_it_still_crosses(self): + iid = self.pile() + self.assertEqual(0, self.rejudge()) + self.assertEqual("OPEN", self.status(iid)[0]) + + +class AChainIsAskedWhetherTheJoinRulesWouldFormIt(RejudgeSandbox): + """Two legs on one binary, recorded with no rung. Once custody explains + every leg, the current join rules would not form the chain + (_unjoinable_chain_resolution, the rule the chain-leg migration uses).""" + + def chain(self): + legs = [self.process(self.bin, "a" * 64), + aegis.finding( + "HIGH", "net-beacon", + "Persistent outbound connection (beacon shape)", "d", + "beacon:%s:203.0.113.7:443" % self.bin, path=self.bin, + program=self.bin, trust=SUSPICIOUS_TRUST, + remote="203.0.113.7", port=443)] + entity = aegis.hashlib.sha256(aegis._canon_entity_path( + self.bin).encode("utf-8", "replace")).hexdigest()[:16] + return self.record(legs, kind="correlation", + key="chain:persistence-execution:%s" % entity) + + def test_closes_once_every_leg_is_explained(self): + iid = self.chain() + self.receipt = "brew:tool" + self.assertEqual(1, self.rejudge()) + status, resolution = self.status(iid) + self.assertEqual("FALSE_POSITIVE", status) + self.assertIn("provenance-explained", resolution) + + def test_stays_while_a_leg_is_unexplained(self): + iid = self.chain() + self.assertEqual(0, self.rejudge()) + self.assertEqual("OPEN", self.status(iid)[0]) + + +class ItRunsHourlyAndOnANewLogic(RejudgeSandbox): + def test_throttled_to_once_an_hour_unless_the_logic_changed(self): + first = self.record([self.process(self.bin, "a" * 64)]) + self.receipt = "brew:tool" + self.assertEqual(1, self.rejudge(NOW)) + self.assertEqual("FALSE_POSITIVE", self.status(first)[0]) + second = self.record([self.process(self.bin, "e" * 64)]) + self.assertIsNone(self.rejudge(NOW + 60)) # throttled + self.assertEqual("OPEN", self.status(second)[0]) + self.stub("_REJUDGE_LOGIC_VERSION", + aegis._REJUDGE_LOGIC_VERSION + 1) + self.assertEqual(1, self.rejudge(NOW + 120)) # new logic + self.assertEqual("FALSE_POSITIVE", self.status(second)[0]) + third = self.record([self.process(self.bin, "f" * 64)]) + self.assertIsNone(self.rejudge(NOW + 180)) + self.assertEqual(1, self.rejudge(NOW + 120 + 3600)) # an hour on + self.assertEqual("FALSE_POSITIVE", self.status(third)[0]) + + def test_a_run_is_capped_and_says_what_it_left(self): + self.stub("_REJUDGE_MAX_INCIDENTS", 1) + ids = [self.record([self.process(self.bin, c * 64)]) for c in "ab"] + self.receipt = "brew:tool" + self.assertEqual(1, self.rejudge(NOW)) + self.assertEqual(["FALSE_POSITIVE", "OPEN"], + [self.status(i)[0] for i in ids]) + with open(aegis.RUN_LOG, encoding="utf-8") as f: + log = f.read() + self.assertIn("re-judge", log) + self.assertIn("1 left for the next run", log) + # The next run picks up where this one stopped. + self.assertEqual(1, self.rejudge(NOW + 3600)) + self.assertEqual("FALSE_POSITIVE", self.status(ids[1])[0]) + + +class TheScanRunsIt(RejudgeSandbox): + def test_record_security_state_closes_what_the_current_code_would_not_raise(self): + iid = self.record([self.process(self.bin, "a" * 64)]) + self.receipt = "brew:tool" + aegis.record_security_state([], now=NOW) + self.assertEqual("FALSE_POSITIVE", self.status(iid)[0]) + with open(aegis.RUN_LOG, encoding="utf-8") as f: + self.assertIn("re-judged", f.read()) + + def test_it_writes_no_custody_ledger_row(self): + """Re-judging asks the ladder; it must not teach it.""" + self.record([self.process(self.bin, "a" * 64)]) + self.receipt = "brew:tool" + self.rejudge() + self.assertFalse(os.path.exists(aegis.CUSTODY_FILE)) + + +if __name__ == "__main__": + unittest.main()