api: configurable data retention for findings and audit events (#73)

.env.example

@@ -103,6 +103,19 @@ ICEBERG_ENGINE_AUTH_RATE_LIMIT=30
 # THIS MUST BE SET, or every request is charged to the balancer's address.
 ICEBERG_TRUSTED_PROXY_HOPS=0
 
+# ─── Data retention (api role, #73) ───────────────────────────────────────────
+# All three windows are OFF by default, and that is deliberate: this database
+# records where your secrets are and who did what about them, so deleting any of
+# it should be a decision, not an upgrade side effect (docs/retention.md).
+#
+# 0 = keep forever. Open findings, analyst decisions (false_positive,
+# accepted_risk, manual resolutions) and suppressed findings are never eligible
+# whatever these say.
+ICEBERG_RETENTION_RESOLVED_FINDINGS_DAYS=0
+ICEBERG_RETENTION_FINDING_EVENTS_DAYS=0
+# Usually the window a compliance regime names. You cannot get these rows back.
+ICEBERG_RETENTION_AUDIT_EVENTS_DAYS=0
+
 # ─── Engine role ──────────────────────────────────────────────────────────────
 # The per-engine API credential. In production this is minted at deploy time
 # (docs/security.md § Bootstrap), never defaulted into an image.

apps/api/src/iceberg_api/cli.py

@@ -32,7 +32,7 @@
 )
 from sqlmodel import col, select
 
-from iceberg_api import audit
+from iceberg_api import audit, retention
 from iceberg_api.dispatch import build_dispatcher
 from iceberg_api.engines.auth import mint_token
 from iceberg_api.scans import service
@@ -58,6 +58,10 @@ def _build_parser() -> argparse.ArgumentParser:
 
     commands.add_parser("reclaim", help="return expired-lease tasks to the queue")
     commands.add_parser("scheduler-tick", help="run one scheduler round now")
+    commands.add_parser(
+        "retention-purge",
+        help="apply the configured retention windows now (see docs/retention.md)",
+    )
 
     migrate_parser = commands.add_parser("migrate", help="apply migrations up to a revision")
     migrate_parser.add_argument(
@@ -161,6 +165,17 @@ def main(argv: Sequence[str] | None = None) -> int:
         case "migrate":
             migrate(args.revision)
             print(f"migrated to {args.revision}", file=sys.stderr)
+        case "retention-purge":
+            # The maintenance loop runs this on its own cadence; this exists for
+            # the first purge after a window is configured, when an operator wants
+            # to see the number rather than discover it in the audit log.
+            with session_scope() as db:
+                purged = retention.purge(db, settings)
+            print(
+                f"findings={purged.findings} finding_events={purged.finding_events} "
+                f"audit_events={purged.audit_events}",
+                file=sys.stderr,
+            )
     return 0
 
 

apps/api/src/iceberg_api/maintenance.py

@@ -26,7 +26,7 @@
 from iceberg_core.db import session_scope
 from iceberg_core.secrets import SecretStore, build_secret_store
 
-from iceberg_api import suppressions
+from iceberg_api import retention, suppressions
 from iceberg_api.dispatch import Dispatcher, build_dispatcher
 from iceberg_api.notifications import dispatch as notification_dispatch
 from iceberg_api.scans import service
@@ -43,7 +43,7 @@ def run_once(
     settings: ApiSettings | None = None,
     store: SecretStore | None = None,
 ) -> None:
-    """One maintenance round: schedules, reclaim, the safety sweeps, then alerts.
+    """One round: schedules, reclaim, the safety sweeps, alerts, then retention.
 
     Leadership is held on a session of its own for the whole round. Holding it on
     the working session would not work: ``pg_try_advisory_xact_lock`` releases at
@@ -53,7 +53,8 @@ def run_once(
     transaction — and the lock — spans everything below.
 
     ``settings``/``store`` are injectable so a test can drive a round without a
-    configured SMTP relay; both default to the process configuration.
+    configured SMTP relay and with its own retention windows; both default to the
+    process configuration.
     """
     at = now or datetime.now(UTC)
     resolved = settings or get_api_settings()
@@ -88,6 +89,12 @@ def run_once(
         # gets retried on the next beat instead of losing the alert.
         with session_scope() as db:
             notification_dispatch.deliver_pending(db, resolved, secret_store, now=at)
+        # Retention (#73). A no-op unless the deployment configured a window —
+        # this database is evidence, so deleting any of it is opt-in. Last in the
+        # round because it is the only job that can be slow on a database that has
+        # never been purged, and nothing else waits on it.
+        with session_scope() as db:
+            retention.purge(db, resolved, now=at)
 
 
 def _already_leader(db: object) -> bool:

apps/api/src/iceberg_api/retention.py

@@ -0,0 +1,202 @@
+"""Data retention (#73).
+
+Findings and their append-only event trail grow forever, and so does the
+administrative audit log. This deletes what a deployment has decided it no longer
+needs — and nothing else, which is most of the design.
+
+**Every window is off by default.** This database is evidence: it records where
+secrets were found and who did what about them. Deleting some of it has to be a
+decision somebody made, not something that starts happening because they
+upgraded. With no windows configured this module does nothing at all.
+
+**What is never eligible**, whatever the window says:
+
+* **Open findings.** An unresolved secret is not old news, it is a secret nobody
+  has dealt with.
+* **Findings an analyst decided about** — ``false_positive``, ``accepted_risk``,
+  or a manual resolution. Those are judgements, and a judgement that expires
+  silently means the same finding comes back next scan with nobody remembering
+  it was already considered. Only ``auto`` resolutions — an inference from
+  absence, made by a scan — are eligible.
+* **Events belonging to a finding that still exists and is open.** Its trail is
+  how an analyst understands the state they are looking at.
+* **A suppressed finding's row**, because the suppression is the reason it is not
+  in anyone's list; deleting it would resurrect it on the next scan as new.
+
+**The purge is itself audited.** Deleting evidence is an administrative action,
+so each round that removes anything writes an ``audit_event`` recording what was
+deleted, how much, and the window that justified it — and that row is subject to
+the audit window like any other, which is the honest behaviour rather than a
+special case that hides purges from the retention it enforces.
+
+Deletion is batched (``retention_batch_size``) so the first purge on a database
+that has never had one cannot hold locks for minutes. A round that hits the batch
+ceiling simply continues on the next beat.
+"""
+
+from dataclasses import asdict, dataclass
+from datetime import UTC, datetime, timedelta
+
+import structlog
+from iceberg_core.config import ApiSettings
+from iceberg_core.enums import FindingResolution, FindingState
+from iceberg_core.models import (
+    AUDIT_RETENTION_PURGED,
+    AUDIT_TARGET_RETENTION,
+    AuditEvent,
+    Finding,
+    FindingEvent,
+)
+from sqlmodel import Session, col, delete, select
+
+from iceberg_api import audit
+
+logger = structlog.get_logger()
+
+#: Resolutions a purge may remove. Deliberately only this one — see the module
+#: docstring. A tuple rather than a set so the membership test reads in the query.
+PURGEABLE_RESOLUTIONS = (FindingResolution.AUTO,)
+
+
+@dataclass(frozen=True, slots=True)
+class PurgeResult:
+    """What one retention round deleted."""
+
+    findings: int = 0
+    finding_events: int = 0
+    audit_events: int = 0
+
+    def total(self) -> int:
+        return self.findings + self.finding_events + self.audit_events
+
+
+def purge(
+    db: Session,
+    settings: ApiSettings,
+    *,
+    now: datetime | None = None,
+) -> PurgeResult:
+    """Delete what has aged out. Commits, and audits itself if it removed anything."""
+    at = now or datetime.now(UTC)
+    batch = settings.retention_batch_size
+
+    findings = _purge_findings(db, settings.retention_resolved_findings_days, at=at, batch=batch)
+    # After the findings, so events cascaded away with their finding are not
+    # counted twice — and so the count reported here is events pruned *from
+    # surviving findings*, which is the number an operator is asking about.
+    events = _purge_finding_events(db, settings.retention_finding_events_days, at=at, batch=batch)
+    audits = _purge_audit_events(db, settings.retention_audit_events_days, at=at, batch=batch)
+
+    result = PurgeResult(findings=findings, finding_events=events, audit_events=audits)
+    if result.total() == 0:
+        db.commit()
+        return result
+
+    audit.record(
+        db,
+        # No actor: nobody triggered this, a policy did.
+        actor_id=None,
+        action=AUDIT_RETENTION_PURGED,
+        target_type=AUDIT_TARGET_RETENTION,
+        target_id=None,
+        detail={
+            **{key: str(value) for key, value in asdict(result).items()},
+            # The window that justified it, so the row still explains itself when
+            # somebody reads it a year later and the settings have changed.
+            "resolved_findings_days": str(settings.retention_resolved_findings_days),
+            "finding_events_days": str(settings.retention_finding_events_days),
+            "audit_events_days": str(settings.retention_audit_events_days),
+        },
+    )
+    db.commit()
+    logger.info("retention_purged", **asdict(result))
+    return result
+
+
+def _cutoff(days: int, at: datetime) -> datetime | None:
+    """The instant before which rows are eligible, or None when disabled."""
+    return None if days <= 0 else at - timedelta(days=days)
+
+
+def _purge_findings(db: Session, days: int, *, at: datetime, batch: int) -> int:
+    """Auto-resolved findings that have been resolved longer than the window.
+
+    Eligibility is checked in the query rather than in Python: a finding that
+    changes state between a read and a delete would otherwise be removed on the
+    strength of a state it no longer has.
+    """
+    cutoff = _cutoff(days, at)
+    if cutoff is None:
+        return 0
+
+    eligible = list(
+        db.exec(
+            select(col(Finding.id))
+            .where(
+                col(Finding.state) == FindingState.RESOLVED,
+                col(Finding.resolution).in_(PURGEABLE_RESOLUTIONS),
+                # `updated_at` moves when the finding is resolved, so it is the
+                # age of the *decision*, not of the first sighting. A finding
+                # found two years ago and resolved yesterday is a day old here.
+                col(Finding.updated_at) < cutoff,
+                # A suppressed finding is hidden, not gone: deleting it would
+                # bring it back as new on the next scan (ADR 0008).
+                col(Finding.suppressed_at).is_(None),
+            )
+            .limit(batch)
+        )
+    )
+    if not eligible:
+        return 0
+
+    # FindingEvent cascades on finding_id, so the trail goes with the finding it
+    # describes rather than being orphaned.
+    db.exec(delete(Finding).where(col(Finding.id).in_(eligible)))
+    return len(eligible)
+
+
+def _purge_finding_events(db: Session, days: int, *, at: datetime, batch: int) -> int:
+    """Old trail entries for findings that are no longer open.
+
+    An open finding keeps its whole history however old: it is what an analyst
+    reads to understand the state in front of them.
+    """
+    cutoff = _cutoff(days, at)
+    if cutoff is None:
+        return 0
+
+    still_open = select(col(Finding.id)).where(col(Finding.state) == FindingState.OPEN)
+    eligible = list(
+        db.exec(
+            select(col(FindingEvent.id))
+            .where(
+                col(FindingEvent.created_at) < cutoff,
+                col(FindingEvent.finding_id).not_in(still_open),
+            )
+            .limit(batch)
+        )
+    )
+    if not eligible:
+        return 0
+
+    db.exec(delete(FindingEvent).where(col(FindingEvent.id).in_(eligible)))
+    return len(eligible)
+
+
+def _purge_audit_events(db: Session, days: int, *, at: datetime, batch: int) -> int:
+    """Administrative audit rows past their window."""
+    cutoff = _cutoff(days, at)
+    if cutoff is None:
+        return 0
+
+    eligible = list(
+        db.exec(select(col(AuditEvent.id)).where(col(AuditEvent.created_at) < cutoff).limit(batch))
+    )
+    if not eligible:
+        return 0
+
+    db.exec(delete(AuditEvent).where(col(AuditEvent.id).in_(eligible)))
+    return len(eligible)
+
+
+__all__ = ["PURGEABLE_RESOLUTIONS", "PurgeResult", "purge"]

apps/api/tests/test_maintenance.py

@@ -43,9 +43,11 @@ def process_engine_fixture(db_engine: SAEngine) -> Iterator[None]:
 def run_round_fixture(secret_store: SecretStore) -> Callable[..., None]:
     """One maintenance round with settings supplied rather than read from the env.
 
-    A round delivers queued notifications (#60), so it needs settings and a secret
-    store. Injecting them keeps these tests from depending on a configured
-    deployment — and from needing an SMTP relay to prove the scheduler fires.
+    A round delivers queued notifications (#60) and applies retention (#73), so
+    it needs settings and a secret store. Injecting them keeps these tests from
+    depending on a configured deployment — no SMTP relay is needed to prove the
+    scheduler fires, and the retention windows default to off, so none of these
+    rounds can delete anything.
     """
     settings = ApiSettings(
         database_url="postgresql+psycopg://unused/unused",
@@ -94,7 +96,9 @@ def test_a_round_launches_due_scans_and_advances_the_schedule(
 
 
 def test_a_round_reclaims_expired_leases(
-    session: Session, dispatcher: RecordingDispatcher, run_round: Callable[..., None]
+    session: Session,
+    dispatcher: RecordingDispatcher,
+    run_round: Callable[..., None],
 ) -> None:
     source = _source(session)
     scan = service.launch_scan(session, source, trigger=ScanTrigger.MANUAL, dispatcher=dispatcher)