diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 8a270e8..dce9609 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -147,7 +147,8 @@ 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 only the exact correlation key while retaining later occurrences as evidence; -a changed content hash gets a new key. A `RESOLVED` threat that recurs opens a +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 new incident instead of being silently ignored. ```text @@ -170,6 +171,31 @@ can tell a broken rule from an expected-but-noisy one. Each incident card also prints the documented benign causes for the sensors that fired, so triage is a match against a known list rather than an investigation. +### Acquired tolerance + +Exact-key suppression cannot absorb the dominant benign churn: a vendor updater +rewrites the same plist on every release, so the same reviewed identity re-opens +a fresh HIGH incident per update, forever. Acquired tolerance is the +identity-level memory over the operator's own verdicts, with immune-system +guards because they answer the same adversarial pressures: + +- Only **human `benign-positive` verdicts teach** — `false-positive` labels tune + rules instead, and machine verdicts write no dismissal record, so tolerance + can never feed on itself or pollute `backtest` precision. +- **Antigen-specific**: the identity is the fingerprint minus its trailing + content hash, and only categories whose benign churn is hash-shaped are + eligible (an allowlist). A beacon's endpoint, a path, a marker set are facts — + a new one is a new incident, always. +- **Repeated exposure required**: three distinct dismissed incidents on the + identity inside 180 days, so one hasty dismissal teaches nothing. +- **Inflammation overrides**: never `CRITICAL`, never above the severity the + operator actually reviewed, never for attack-defined evidence (decoys, + latches, canaries), and never while any incident on the identity is active. +- **Visible and disputable**: the incident is still created with its full + evidence, closed as `auto-tolerated` citing the precedent count, counted in a + footer on the active listing, and one `reopen` both re-alerts and revokes the + tolerance (reopening deletes the dismissal rows the count was built on). + `aegis.py replay [days]` re-runs the current correlation logic over recorded history in a throwaway in-memory database. It is strictly read-only — no incident, no notification, no durable write — so a detection change can be @@ -224,9 +250,13 @@ occupied. Destroy verifies deletion but does not claim secure erase on APFS/SSD. on a periodic full scan; vnode notification is not treated as a complete log. - `doctor` exposes permission and sensor degradation. Three consecutive sensor failures open one health incident; recovery resolves it and resets the count. -- Capability-dependent inventories such as Background Task Management remain - DEGRADED when macOS requires interactive authorization; denied data is never - interpreted as an empty or clean snapshot. +- Capability-dependent inventories such as Background Task Management report + `PRIVILEGED` when the OS demands interactive admin authorization the + background observer cannot synthesize (macOS 26 moved `sfltool dumpbtm` + behind `system.privilege.admin`): a named permanent coverage gap — shown in + `doctor`/`status` as `i`, never green — that does not escalate to a + coverage-degraded incident the way a transient failure (still DEGRADED) + does. Denied data is never interpreted as an empty or clean snapshot. - Uninstall retains evidence by default. Purge requires the explicit `--purge`. ## Protective tier (opt-in, by hand) diff --git a/aegis.py b/aegis.py index c405ba1..761403c 100755 --- a/aegis.py +++ b/aegis.py @@ -2559,6 +2559,128 @@ def _category_dismissal_weights(db, now, window=90 * 86400): return weights +# --------------------------------------------------------------------------- # +# ACQUIRED TOLERANCE — identity-level immune memory over typed dismissals. +# +# The exact-fingerprint reattach above tolerates *identical* re-observations, +# and _category_dismissal_weights decays a whole noisy category's risk weight — +# but nothing in between: a vendor updater that rewrites the same plist monthly +# presents a new content hash each time, so the SAME reviewed identity re-opens +# a fresh HIGH incident on every update, forever. That per-identity churn is +# what actually fills the queue (55 of 55 real triaged incidents here were it). +# +# The generalization is deliberately narrow, and every guard is an immunology +# rule because those rules exist for the same adversarial reason: +# · Only HUMAN `benign-positive` verdicts teach (false-positive labels tune +# rules instead; machine verdicts never count — no runaway feedback). +# · Tolerance is ANTIGEN-SPECIFIC: the identity is the fingerprint minus its +# trailing content hash. A beacon's ip:port, a path, a marker set are all +# facts — only hash churn generalizes. New endpoint = new incident. +# · It takes REPEATED EXPOSURE: >= _TOLERANCE_MIN_VERDICTS distinct dismissed +# incidents inside _TOLERANCE_WINDOW, so one hasty dismissal teaches nothing. +# · INFLAMMATION OVERRIDES: never CRITICAL, never above the severity the +# operator actually reviewed, never for attack-defined fingerprints, and +# never while any incident on that identity is active (a `reopen` is a +# dispute — it also deletes the dismissal rows the count is built on). +# · Tolerated is NOT invisible: the incident is still created with its full +# evidence, auto-closed with resolution 'auto-tolerated' citing precedent, +# listed under `incidents all`, and one `reopen` revokes the whole thing. +# --------------------------------------------------------------------------- # + +_TOLERANCE_MIN_VERDICTS = 3 +_TOLERANCE_WINDOW = 180 * 86400 +# A trailing : component of this shape is a content hash, not identity. +_TOLERANCE_HASH_RE = re.compile(r"^[0-9a-f]{12,64}$", re.I) +# Only surfaces whose benign churn is hash-shaped may generalize at all — +# an allowlist, matching the deadfall discipline, never a blocklist. +_TOLERANCE_CATEGORIES = frozenset(( + "persistence", "xpersist", "process", "behavior", "agent-surface", "btm")) +# Attack-defined evidence (deadfall's own trigger prefixes plus honeytokens): +# no verdict history may ever tolerize these. +_NEVER_TOLERATE_PREFIXES = ("decoy:", "latch:", "canary:") + + +def _tolerance_identity(fingerprint): + """The hash-stripped stable identity of a signal fingerprint, or None when + no safe generalization exists. Strips exactly ONE trailing content-hash + component; a fingerprint with no hash suffix (beacons end in a port, some + process rows carry sha None) has nothing to generalize over and the + existing exact-key reattach already covers it.""" + fp = str(fingerprint or "") + if not fp or fp.startswith(_NEVER_TOLERATE_PREFIXES): + return None + parts = fp.split(":") + if len(parts) < 3 or not _TOLERANCE_HASH_RE.match(parts[-1]): + return None + return ":".join(parts[:-1]) + + +def _tolerance_memory(db, now): + """{identity: (distinct_verdicts, max_reviewed_sev_order)} from the + operator's own benign-positive dismissals of signal incidents inside the + window. Only identities past the verdict floor are returned.""" + memory = {} + try: + rows = db.execute( + "SELECT d.incident_id, d.correlation_key, i.severity " + "FROM dismissals d JOIN incidents i ON i.id=d.incident_id " + "WHERE d.reason_code='benign-positive' AND d.dismissed_at>=? " + "AND d.correlation_key LIKE 'signal:%'", + (now - _TOLERANCE_WINDOW,)).fetchall() + except Exception: + return memory + seen = {} + for row in rows: + ident = _tolerance_identity(row["correlation_key"][len("signal:"):]) + if not ident: + continue + bucket = seen.setdefault(ident, {"incidents": set(), "sev": -1}) + bucket["incidents"].add(row["incident_id"]) + bucket["sev"] = max(bucket["sev"], + SEV_ORDER.get(row["severity"], -1)) + for ident, bucket in seen.items(): + if len(bucket["incidents"]) >= _TOLERANCE_MIN_VERDICTS: + memory[ident] = (len(bucket["incidents"]), bucket["sev"]) + return memory + + +def _disputed_identities(db): + """Identities with ANY incident currently in an active state. An operator + who reopened (or has not yet triaged) an incident on an identity is in + dispute with tolerance for it, so tolerance stands down there.""" + idents = set() + marks = ",".join("?" for _ in _ACTIVE_INCIDENT_STATES) + for row in db.execute( + "SELECT correlation_key FROM incidents WHERE status IN (%s) " + "AND correlation_key LIKE 'signal:%%'" % marks, + _ACTIVE_INCIDENT_STATES): + ident = _tolerance_identity(row["correlation_key"][len("signal:"):]) + if ident: + idents.add(ident) + return idents + + +def _auto_tolerate(db, incident_id, verdicts, now): + """Close a just-created incident under acquired tolerance. The guard is in + the WHERE: only an OPEN incident created THIS pass may be auto-closed, so a + reattach to something the operator already saw is never swept from under + them. Deliberately writes NO dismissals row — machine verdicts must never + feed backtest precision, category down-weighting, or future tolerance.""" + cur = db.execute( + "UPDATE incidents SET status='FALSE_POSITIVE'," + "resolution='auto-tolerated',updated_at=?,next_reminder_at=NULL " + "WHERE id=? AND status='OPEN' AND created_at=?", + (now, incident_id, now)) + if cur.rowcount: + db.execute( + "INSERT INTO events(occurred_at,observed_at,source,event_type," + "incident_id,data_json) VALUES(?,?,?,?,?,?)", + (now, now, "incident", "incident.lifecycle", incident_id, + json.dumps({"from": "OPEN", "to": "FALSE_POSITIVE", + "reason_code": "auto-tolerated", + "prior_verdicts": verdicts}))) + + def _accumulate_risk(db, now, new_ids): """Open one 'risk' incident per entity whose recent findings sum past RISK_THRESHOLD from ≥ RISK_MIN_SIGNALS DISTINCT signals spanning at least @@ -2824,15 +2946,31 @@ def has_marker(f, values): # Middle tier: pile-up of weak signals on one entity → one 'risk' incident. _accumulate_risk(db, now, new_ids) - # Every uncorrelated HIGH+ signal still becomes one actionable incident. + # Every uncorrelated HIGH+ signal still becomes one actionable incident — + # but an identity the operator has repeatedly reviewed as benign-positive + # opens pre-closed under acquired tolerance instead of re-alerting. + tolerance = _tolerance_memory(db, now) + disputed = _disputed_identities(db) if tolerance else frozenset() for event_id, f in new_events: if f.get("category") in suppressed_categories or event_id in attached \ or SEV_ORDER.get(f.get("severity"), -1) \ < SEV_ORDER["HIGH"]: continue - _upsert_incident(db, "signal:" + f["fingerprint"], f["title"], - f["severity"], "signal", now, [event_id], - initially_notified) + verdicts = 0 + if tolerance and f.get("category") in _TOLERANCE_CATEGORIES \ + and SEV_ORDER.get(f.get("severity"), -1) < SEV_ORDER["CRITICAL"]: + ident = _tolerance_identity(f.get("fingerprint")) + if ident and ident not in disputed: + count, reviewed_sev = tolerance.get(ident, (0, -1)) + if count and SEV_ORDER.get(f.get("severity"), -1) \ + <= reviewed_sev: + verdicts = count + incident_id = _upsert_incident( + db, "signal:" + f["fingerprint"], f["title"], + f["severity"], "signal", now, [event_id], + initially_notified or bool(verdicts)) + if verdicts: + _auto_tolerate(db, incident_id, verdicts, now) def _record_health(db, health, now): @@ -2842,7 +2980,11 @@ def _record_health(db, health, now): detail = redact_sensitive(item.get("detail") or "")[:500] prior = db.execute("SELECT * FROM sensor_status WHERE sensor_id=?", (sensor_id,)).fetchone() - failed = status != "OK" + # PRIVILEGED is neither working nor broken: a named, OS-imposed + # permanent coverage gap. It must not escalate to a coverage-degraded + # incident (that alarm is for sensors that SHOULD be answering), and it + # must not forge last_ok_at (the surface did not answer). + failed = status not in ("OK", "PRIVILEGED") failures = (prior["consecutive_failures"] if prior else 0) + 1 \ if failed else 0 episode = (prior["episode_started_at"] if prior else None) @@ -2858,7 +3000,7 @@ def _record_health(db, health, now): "item_count=excluded.item_count,detail=excluded.detail," "consecutive_failures=excluded.consecutive_failures," "episode_started_at=excluded.episode_started_at", - (sensor_id, status, now, now if not failed else + (sensor_id, status, now, now if status == "OK" else (prior["last_ok_at"] if prior else None), int(item.get("duration_ms") or 0), int(item.get("item_count") or 0), detail, failures, episode)) @@ -2873,11 +3015,14 @@ def _record_health(db, health, now): "Security coverage degraded: %s" % sensor_id, "HIGH", "sensor-health", now, [cur.lastrowid]) elif not failed: + resolution = ("surface is privileged-only on this OS — recorded " + "as a permanent coverage gap, not a sensor failure" + if status == "PRIVILEGED" else "sensor recovered") db.execute("UPDATE incidents SET status='RESOLVED',resolution=?," "updated_at=?,last_seen=?,next_reminder_at=NULL WHERE " "correlation_key=? AND status IN (%s)" % ",".join("?" for _ in _ACTIVE_INCIDENT_STATES), - ("sensor recovered", now, now, "sensor:" + sensor_id) + + (resolution, now, now, "sensor:" + sensor_id) + _ACTIVE_INCIDENT_STATES) @@ -6210,6 +6355,24 @@ def f(p, h, *old): # instead of the real (slow) sfltool — the same override pattern as LSOF_LISTEN_CMD. BTM_DUMP_CMD = ["sfltool", "dumpbtm"] +# Sentinel a snapshot fn returns when its backing command EXISTS but this OS +# requires interactive admin authorization that a background observer cannot — +# and must not — synthesize. Distinct from None (a transient non-answer: +# timeout, flake) because the two need opposite health handling: a transient +# failure escalates to a coverage-degraded incident after 3 misses, while an +# OS-imposed privilege wall is a PERMANENT, precisely-diagnosed condition — +# alerting on it every scan forever is pure fatigue. It is still surfaced as a +# PRIVILEGED coverage gap in doctor/status (unknown is never green); it just +# stops masquerading as a broken sensor. +SURFACE_PRIVILEGED = object() + +# macOS 26 (Darwin 25) moved `sfltool dumpbtm` behind system.privilege.admin: +# the unprivileged harvest this sensor was built on ("Ventura+, unprivileged") +# no longer exists there. The failure is identified by its authorization +# signature, never inferred from a bare non-zero exit. +_BTM_PRIVILEGED_MARKERS = ("system.privilege.admin", "authorization failed", + "errauthorization") + def _parse_btm(text): """{identifier: {name, team, type, url}} from `sfltool dumpbtm`. A top-level @@ -6271,9 +6434,17 @@ def snapshot_btm(): auto-updaters …), so a false-empty adopted into the baseline would storm ~90 bogus 'new background item' findings the instant sfltool later succeeds. We therefore signal the non-answer as None (skipped by _scan_surfaces) so - sensor health remains DEGRADED instead of silently baselining false-empty.""" - out, _, rc = run(BTM_DUMP_CMD, timeout=30) + sensor health remains DEGRADED instead of silently baselining false-empty. + + A third outcome exists since macOS 26: dumpbtm refuses without interactive + admin authorization. That is not a flake — it will fail identically on + every scan this OS ever runs — so it returns SURFACE_PRIVILEGED and is + recorded as a permanent, named coverage gap rather than a degraded sensor.""" + out, err, rc = run(BTM_DUMP_CMD, timeout=30) if rc != 0 or not out: + blob = ((err or "") + "\n" + (out or "")).lower() + if any(marker in blob for marker in _BTM_PRIVILEGED_MARKERS): + return SURFACE_PRIVILEGED return None # timeout/failure — a non-answer, NOT "zero items" return _parse_btm(out) @@ -9816,8 +9987,19 @@ def _scan_surfaces(baseline, corrupt, first_run, health=None): started = time.monotonic() try: cur = snap_fn() - status = "DEGRADED" if cur is None else "OK" - detail = "sensor returned no reliable snapshot" if cur is None else "" + if cur is SURFACE_PRIVILEGED: + # OS-imposed privilege wall: a named permanent coverage gap, + # not a broken sensor. Never diffed, never adopted, never + # escalated to a coverage-degraded incident. + cur, status = None, "PRIVILEGED" + detail = ("this OS requires interactive admin authorization " + "for the backing command; surface is observable " + "only from a privileged context") + elif cur is None: + status = "DEGRADED" + detail = "sensor returned no reliable snapshot" + else: + status, detail = "OK", "" except Exception as e: cur = None status, detail = "FAILED", str(e) @@ -10463,20 +10645,50 @@ def cmd_report(): def cmd_incidents(show_all=False): incidents = list_incidents(active_only=not show_all) + tolerated = _recent_tolerated_count() if not incidents: print("No %sincidents." % ("recorded " if show_all else "active ")) + if tolerated and not show_all: + print(_tolerated_footer(tolerated)) return 0 print("# Aegis incidents (%d)\n" % len(incidents)) for item in incidents: + status = item["status"] + if item.get("resolution") == "auto-tolerated": + status = "TOLERATED" print(" #%s %-12s %-8s %s\n %s · %s evidence event%s" % ( - item["id"], item["status"], item["severity"], item["title"], + item["id"], status, item["severity"], item["title"], item["correlation_key"], item["evidence_count"], "" if item["evidence_count"] == 1 else "s")) + if tolerated and not show_all: + print("\n" + _tolerated_footer(tolerated)) print("\nDetails/actions: aegis.py incident [ack|investigate|contain|" "recover|monitor|resolve|false-positive|benign-positive|reopen]") return 0 +def _recent_tolerated_count(window=30 * 86400): + """How many signals acquired tolerance auto-closed in the window. Suppression + that cannot be seen is indistinguishable from a detection gap, so the count + is surfaced wherever active incidents are listed.""" + ensure_state() + init_event_store() + db = _event_connection() + try: + return db.execute( + "SELECT COUNT(*) FROM incidents WHERE resolution='auto-tolerated' " + "AND updated_at>=?", (_epoch() - window,)).fetchone()[0] + finally: + db.close() + + +def _tolerated_footer(count): + return ("%d recurring signal(s) auto-tolerated in the last 30d — each " + "matched >=%d of your own benign-positive verdicts on the same " + "identity. Review: aegis.py incidents all; dispute: reopen ." + % (count, _TOLERANCE_MIN_VERDICTS)) + + _INCIDENT_ACTIONS = { "ack": "ACK", "investigate": "INVESTIGATING", "contain": "CONTAINED", "recover": "RECOVERING", "monitor": "MONITORING", "resolve": "RESOLVED", @@ -10684,9 +10896,13 @@ def cmd_incident(incident_id, action=None, reason=None): if not item: print("no such incident: %s" % incident_id) return 1 + status = item["status"] + resolution = item.get("resolution") + if resolution and resolution != status.lower().replace("_", "-"): + status = "%s (%s)" % (status, resolution) print("# Incident #%s — %s\n\n severity: %s\n status: %s\n chain: %s" "\n opened: %s\n updated: %s\n\nEvidence:" % ( - item["id"], item["title"], item["severity"], item["status"], + item["id"], item["title"], item["severity"], status, item["correlation_key"], datetime.fromtimestamp(item["created_at"]).isoformat(), datetime.fromtimestamp(item["updated_at"]).isoformat())) @@ -10884,11 +11100,14 @@ def cmd_doctor(): print(" ? sensors no completed scan recorded") problems.append("no sensor health") for item in health: - mark = "✓" if item["status"] == "OK" else "?" + # PRIVILEGED = a named, OS-imposed permanent coverage gap ("i", like + # rootwatch-absent below): shown so it is never mistaken for coverage, + # but not a problem — nothing here is fixable or unexpectedly broken. + mark = {"OK": "✓", "PRIVILEGED": "i"}.get(item["status"], "?") print(" %s %-27s %-10s failures=%d %s" % ( mark, item["sensor_id"], item["status"], item["consecutive_failures"], item["detail"] or "")) - if item["status"] != "OK": + if item["status"] not in ("OK", "PRIVILEGED"): problems.append(item["sensor_id"]) # INFO, never a problem: rootwatch is opt-in, but its absence is worth one # honest line — without it a same-uid attacker can kill Aegis and the @@ -10990,7 +11209,7 @@ def cmd_status(): if health: print("\n# Sensor coverage") for item in health: - mark = "✓" if item["status"] == "OK" else "?" + mark = {"OK": "✓", "PRIVILEGED": "i"}.get(item["status"], "?") print(" %s %-32s %s%s" % ( mark, item["sensor_id"], item["status"], (" — " + item["detail"]) if item["detail"] else "")) @@ -17082,7 +17301,12 @@ def cmd_guard(action="status", rest=None): ...the two dismissals are recorded separately and feed different tuning queues: false-positive = the DETECTION was wrong (tune the rule), benign-positive = the event was real - but authorized (suppress this instance, rule is fine) + but authorized (suppress this instance, rule is fine). + Acquired tolerance: an identity you have dismissed + benign-positive on 3+ distinct incidents re-opens PRE-CLOSED + (TOLERATED, evidence kept, no alert) when only its content + hash changed — never on higher severity, a new endpoint, or + attack-defined evidence. `reopen` disputes + revokes it replay [days] backtest the CURRENT correlation logic against recorded history (default 30d). Read-only: opens no incident, sends no notification — run it after changing detection logic diff --git a/tests/test_btm_privileged.py b/tests/test_btm_privileged.py new file mode 100644 index 0000000..355c27b --- /dev/null +++ b/tests/test_btm_privileged.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +"""The BTM privileged-surface fix — macOS 26 moved `sfltool dumpbtm` behind +system.privilege.admin, and until this fix the sensor reported the permanent +OS-imposed wall as a broken sensor: DEGRADED on every scan (540+ consecutive +failures) feeding one immortal HIGH coverage-degraded incident. + +Pinned here: + 1. The authorization signature returns SURFACE_PRIVILEGED; a generic failure + still returns None (the transient path keeps escalating — that alarm is + for sensors that SHOULD be answering). + 2. PRIVILEGED health never escalates to an incident, resets the failure + counter, never forges last_ok_at, and RESOLVES an existing degraded + incident with the honest privileged-only resolution. + 3. _scan_surfaces treats the sentinel like a non-answer for diffing: no + findings fabricated, nothing adopted into the baseline. + +Fully sandboxed: durable-state tests redirect STATE_DIR/EVENT_DB to a tmp dir. +""" +import json +import os +import shutil +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import aegis # noqa: E402 + +NOW = 1786700000 # > 2026-01-01 store floor + +# Verbatim shape of the real macOS 26 refusal (captured live 2026-08-12). +_AUTH_STDERR = ( + "2026-08-12 07:54:45.948 sfltool[70850:6343723] Error obtaining right " + "system.privilege.admin: Error Domain=NSOSStatusErrorDomain Code=-60006 " + '"errAuthorizationCanceled: The authorization was cancelled by the user."\n' + "2026-08-12 07:54:45.949 sfltool[70850:6343723] authorization failed") + +_DUMP_OK = """#1: + UUID: AAAA-BBBB + Name: Helper + Type: login item + Identifier: com.vendor.helper + URL: file:///Applications/Vendor.app/ +""" + + +class TestSnapshotBtmOutcomes(unittest.TestCase): + def setUp(self): + self._run = aegis.run + + def tearDown(self): + aegis.run = self._run + + def test_authorization_refusal_is_privileged(self): + aegis.run = lambda cmd, timeout=15, extra_env=None: ("", _AUTH_STDERR, 1) + self.assertIs(aegis.snapshot_btm(), aegis.SURFACE_PRIVILEGED) + + def test_generic_failure_is_still_a_transient_none(self): + # A timeout/flake must keep the old contract: None -> DEGRADED -> the + # coverage alarm still exists for sensors that should be answering. + aegis.run = lambda cmd, timeout=15, extra_env=None: ("", "boom", 1) + self.assertIsNone(aegis.snapshot_btm()) + aegis.run = lambda cmd, timeout=15, extra_env=None: ("", "", 0) + self.assertIsNone(aegis.snapshot_btm()) # empty output = non-answer + + def test_success_still_parses(self): + aegis.run = lambda cmd, timeout=15, extra_env=None: (_DUMP_OK, "", 0) + snap = aegis.snapshot_btm() + self.assertIn("com.vendor.helper", snap) + + +class _Sandbox(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="aegis_btm_") + state = os.path.join(self.tmp, ".aegis") + os.makedirs(state) + self._saved = {} + for k, v in (("STATE_DIR", state), + ("EVENT_DB", os.path.join(state, "aegis.db"))): + self._saved[k] = getattr(aegis, k) + setattr(aegis, k, v) + aegis.init_event_store() + + def tearDown(self): + for k, v in self._saved.items(): + setattr(aegis, k, v) + shutil.rmtree(self.tmp, ignore_errors=True) + + def _health_pass(self, status, at, detail="d"): + db = aegis._event_connection() + try: + with db: + aegis._record_health( + db, [{"sensor_id": "surface.btm", "status": status, + "detail": detail}], at) + finally: + db.close() + + def _sensor_row(self): + db = aegis._event_connection() + try: + return dict(db.execute( + "SELECT * FROM sensor_status WHERE sensor_id='surface.btm'" + ).fetchone()) + finally: + db.close() + + def _sensor_incident(self): + db = aegis._event_connection() + try: + row = db.execute( + "SELECT * FROM incidents WHERE correlation_key=" + "'sensor:surface.btm' ORDER BY id DESC LIMIT 1").fetchone() + return dict(row) if row else None + finally: + db.close() + + +class TestPrivilegedHealth(_Sandbox): + def test_privileged_never_escalates_to_an_incident(self): + for i in range(5): + self._health_pass("PRIVILEGED", NOW + i * 3600) + self.assertIsNone(self._sensor_incident()) + row = self._sensor_row() + self.assertEqual(row["status"], "PRIVILEGED") + self.assertEqual(row["consecutive_failures"], 0) + self.assertIsNone(row["last_ok_at"]) # the surface did NOT answer + + def test_degraded_still_escalates(self): + for i in range(3): + self._health_pass("DEGRADED", NOW + i * 3600) + incident = self._sensor_incident() + self.assertIsNotNone(incident) + self.assertEqual(incident["severity"], "HIGH") + self.assertEqual(self._sensor_row()["consecutive_failures"], 3) + + def test_privileged_resolves_the_degraded_incident_honestly(self): + # The pre-fix world: an incident accumulated by DEGRADED scans, then + # acknowledged by the operator (incident #26's exact state). + for i in range(3): + self._health_pass("DEGRADED", NOW + i * 3600) + incident = self._sensor_incident() + self.assertTrue(aegis.transition_incident( + incident["id"], "ACK", now=NOW + 4 * 3600)) + # First post-fix scan identifies the wall and closes the case. + self._health_pass("PRIVILEGED", NOW + 5 * 3600) + incident = self._sensor_incident() + self.assertEqual(incident["status"], "RESOLVED") + self.assertIn("privileged-only", incident["resolution"]) + + def test_ok_recovery_resolution_is_unchanged(self): + for i in range(3): + self._health_pass("DEGRADED", NOW + i * 3600) + self._health_pass("OK", NOW + 4 * 3600) + incident = self._sensor_incident() + self.assertEqual(incident["status"], "RESOLVED") + self.assertEqual(incident["resolution"], "sensor recovered") + self.assertEqual(self._sensor_row()["last_ok_at"], NOW + 4 * 3600) + + +class TestScanSurfacesSentinel(_Sandbox): + def test_privileged_surface_yields_no_findings_and_no_adoption(self): + saved = aegis.SURFACES + aegis.SURFACES = [ + ("btm", lambda: aegis.SURFACE_PRIVILEGED, + lambda prior, cur: [aegis.finding( + "HIGH", "btm", "fabricated", "must never appear", "btm:x")])] + try: + health = [] + findings, baseline = aegis._scan_surfaces( + {}, corrupt=False, first_run=False, health=health) + finally: + aegis.SURFACES = saved + self.assertEqual(findings, []) + self.assertNotIn("btm", baseline) # never adopted + self.assertEqual(len(health), 1) + self.assertEqual(health[0]["status"], "PRIVILEGED") + self.assertIn("admin authorization", health[0]["detail"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tolerance.py b/tests/test_tolerance.py new file mode 100644 index 0000000..837502f --- /dev/null +++ b/tests/test_tolerance.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +"""Acquired tolerance — identity-level immune memory over typed dismissals. + +The exact-fingerprint reattach tolerates identical re-observations; these tests +pin the layer above it: >= _TOLERANCE_MIN_VERDICTS distinct HUMAN +benign-positive verdicts on one hash-stripped identity auto-close the next +hash-churned re-observation (evidence kept, no alert), and every immunology +guard holds — repeated exposure required, antigen-specific, inflammation +overrides, machine verdicts teach nothing, one reopen disputes the tolerance. + +Fully sandboxed: every test redirects STATE_DIR/EVENT_DB into a tmp dir. +""" +import json +import os +import shutil +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import aegis # noqa: E402 + +NOW = 1786600000 # > 2026-01-01 store floor +PLIST = "/Library/LaunchAgents/com.vendor.updater.plist" + + +def _finding(fp, severity="HIGH", category="persistence", title="Persistence item CHANGED"): + return {"fingerprint": fp, "severity": severity, "category": category, + "title": title, "detail": "test", "confidence": "medium"} + + +class _Sandbox(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="aegis_tol_") + state = os.path.join(self.tmp, ".aegis") + os.makedirs(state) + self._saved = {} + for k, v in (("STATE_DIR", state), + ("EVENT_DB", os.path.join(state, "aegis.db"))): + self._saved[k] = getattr(aegis, k) + setattr(aegis, k, v) + aegis.init_event_store() + + def tearDown(self): + for k, v in self._saved.items(): + setattr(aegis, k, v) + shutil.rmtree(self.tmp, ignore_errors=True) + + def _ingest(self, f, at): + """One finding event -> correlation pass, as the scan pipeline would.""" + db = aegis._event_connection() + try: + with db: + cur = db.execute( + "INSERT INTO events(occurred_at,observed_at,source," + "event_type,data_json) VALUES(?,?,?,?,?)", + (at, at, f["category"], "observation.finding", + json.dumps(f))) + aegis._apply_correlations(db, [(cur.lastrowid, f)], at, + initially_notified=True) + finally: + db.close() + + def _incident_for(self, fp): + db = aegis._event_connection() + try: + row = db.execute( + "SELECT * FROM incidents WHERE correlation_key=? " + "ORDER BY id DESC LIMIT 1", ("signal:" + fp,)).fetchone() + return dict(row) if row else None + finally: + db.close() + + def _teach(self, n, severity="HIGH", base=NOW): + """n distinct incidents on PLIST (hash churn), each dismissed + benign-positive by the 'operator'. Returns the taught identity.""" + for i in range(n): + fp = "persistence:changed:%s:%016x" % (PLIST, i) + self._ingest(_finding(fp, severity=severity), base + i * 60) + incident = self._incident_for(fp) + self.assertEqual(incident["status"], "OPEN") + self.assertTrue(aegis.transition_incident( + incident["id"], "FALSE_POSITIVE", now=base + i * 60 + 30, + reason_code="benign-positive")) + return "persistence:changed:" + PLIST + + +class TestToleranceIdentity(unittest.TestCase): + def test_trailing_content_hash_is_stripped(self): + self.assertEqual( + aegis._tolerance_identity( + "persistence:changed:%s:4ecbaeb7c89892df" % PLIST), + "persistence:changed:" + PLIST) + self.assertEqual( + aegis._tolerance_identity("process:/usr/local/bin/x:adhoc:" + "a" * 64), + "process:/usr/local/bin/x:adhoc") + + def test_non_hash_suffixes_do_not_generalize(self): + # A port, a 'None' sha, a bare word: all facts, none of them churn. + for fp in ("beacon:/Applications/App.app/Contents/MacOS/app:1.2.3.4:443", + "process:/var/folders/9n/:unsigned:None", + "short:ab"): + self.assertIsNone(aegis._tolerance_identity(fp), fp) + + def test_attack_defined_prefixes_never_tolerize(self): + for fp in ("decoy:read:/home/x/.aws/credentials:" + "b" * 16, + "latch:cleared:/Library/LaunchAgents:" + "c" * 16, + "canary:touched:/home/x/canary.docx:" + "d" * 16): + self.assertIsNone(aegis._tolerance_identity(fp), fp) + + +class TestAcquiredTolerance(_Sandbox): + def test_three_verdicts_confer_tolerance(self): + self._teach(3) + fp = "persistence:changed:%s:%016x" % (PLIST, 99) + self._ingest(_finding(fp), NOW + 3600) + incident = self._incident_for(fp) + self.assertEqual(incident["status"], "FALSE_POSITIVE") + self.assertEqual(incident["resolution"], "auto-tolerated") + self.assertIsNone(incident["next_reminder_at"]) + # The evidence is kept and the machine verdict is on the record. + db = aegis._event_connection() + try: + kept = db.execute( + "SELECT COUNT(*) FROM incident_events WHERE incident_id=?", + (incident["id"],)).fetchone()[0] + lifecycle = db.execute( + "SELECT data_json FROM events WHERE incident_id=? AND " + "event_type='incident.lifecycle' ORDER BY id DESC LIMIT 1", + (incident["id"],)).fetchone()[0] + finally: + db.close() + self.assertGreaterEqual(kept, 1) + record = json.loads(lifecycle) + self.assertEqual(record["reason_code"], "auto-tolerated") + self.assertEqual(record["prior_verdicts"], 3) + + def test_repeated_exposure_is_required(self): + self._teach(2) # one short of the floor + fp = "persistence:changed:%s:%016x" % (PLIST, 99) + self._ingest(_finding(fp), NOW + 3600) + self.assertEqual(self._incident_for(fp)["status"], "OPEN") + + def test_machine_verdicts_teach_nothing(self): + # 3 human verdicts + 1 auto-tolerated close must still count as 3: + # a fingerprint on a DIFFERENT identity with 2 human verdicts plus + # dismissal rows written by the machine would otherwise cascade. + self._teach(3) + fp = "persistence:changed:%s:%016x" % (PLIST, 99) + self._ingest(_finding(fp), NOW + 3600) + db = aegis._event_connection() + try: + machine_rows = db.execute( + "SELECT COUNT(*) FROM dismissals WHERE incident_id=?", + (self._incident_for(fp)["id"],)).fetchone()[0] + finally: + db.close() + self.assertEqual(machine_rows, 0) + + def test_false_positive_labels_do_not_teach(self): + # false-positive = broken rule -> tune the rule, never tolerize events. + for i in range(3): + fp = "persistence:changed:%s:%016x" % (PLIST, i) + self._ingest(_finding(fp), NOW + i * 60) + aegis.transition_incident( + self._incident_for(fp)["id"], "FALSE_POSITIVE", + now=NOW + i * 60 + 30, reason_code="false-positive") + fp = "persistence:changed:%s:%016x" % (PLIST, 99) + self._ingest(_finding(fp), NOW + 3600) + self.assertEqual(self._incident_for(fp)["status"], "OPEN") + + def test_severity_escalation_breaks_tolerance(self): + self._teach(3, severity="HIGH") + fp = "persistence:changed:%s:%016x" % (PLIST, 99) + self._ingest(_finding(fp, severity="CRITICAL"), NOW + 3600) + self.assertEqual(self._incident_for(fp)["status"], "OPEN") + + def test_category_outside_allowlist_is_not_tolerated(self): + ip_fp = "net-outbound:/usr/bin/nc:%s" % ("e" * 16) + for i in range(3): + fp = "net-outbound:/usr/bin/nc:%016x" % i + self._ingest(_finding(fp, category="net-outbound", + title="Outbound connection"), NOW + i * 60) + aegis.transition_incident( + self._incident_for(fp)["id"], "FALSE_POSITIVE", + now=NOW + i * 60 + 30, reason_code="benign-positive") + self._ingest(_finding(ip_fp, category="net-outbound", + title="Outbound connection"), NOW + 3600) + self.assertEqual(self._incident_for(ip_fp)["status"], "OPEN") + + def test_reopen_disputes_and_revokes(self): + ident_fps = self._teach(3) + self.assertTrue(ident_fps) + # Tolerance is live; one close happens... + fp1 = "persistence:changed:%s:%016x" % (PLIST, 90) + self._ingest(_finding(fp1), NOW + 3600) + tolerated = self._incident_for(fp1) + self.assertEqual(tolerated["resolution"], "auto-tolerated") + # ...the operator disputes it. The reopened incident is active on the + # identity, so the NEXT churn must open normally and alert. + self.assertTrue(aegis.transition_incident( + tolerated["id"], "OPEN", now=NOW + 3700)) + fp2 = "persistence:changed:%s:%016x" % (PLIST, 91) + self._ingest(_finding(fp2), NOW + 4000) + self.assertEqual(self._incident_for(fp2)["status"], "OPEN") + + def test_stale_verdicts_expire(self): + old = NOW - aegis._TOLERANCE_WINDOW - 86400 + self._teach(3, base=old) + fp = "persistence:changed:%s:%016x" % (PLIST, 99) + self._ingest(_finding(fp), NOW) + self.assertEqual(self._incident_for(fp)["status"], "OPEN") + + def test_tolerated_incident_is_visible_and_counted(self): + self._teach(3) + fp = "persistence:changed:%s:%016x" % (PLIST, 99) + self._ingest(_finding(fp), NOW + 3600) + # It appears in the complete history... + everything = aegis.list_incidents(active_only=False) + self.assertIn("auto-tolerated", + [i.get("resolution") for i in everything]) + # ...and never in the active queue. + active = aegis.list_incidents(active_only=True) + self.assertNotIn("signal:" + fp, + [i["correlation_key"] for i in active]) + + +if __name__ == "__main__": + unittest.main()