From 624e1837d0b0e98cb8c913d7b8555423ad3e77e0 Mon Sep 17 00:00:00 2001 From: Richard Hope Date: Fri, 31 Jul 2026 17:14:12 +1000 Subject: [PATCH] notifications: dispatch newly-opened findings to email and webhook channels (#60) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The channel model and its CRUD shipped with M3 (#59); this is the delivery half. **Shape: a transactional outbox.** Reconciliation writes one `notification_delivery` row per (channel, finding, scan) in the transaction that finishes the scan, and sends nothing. The maintenance loop — one replica, under the advisory lock the scheduler already uses — picks up due rows and attempts them. That is what makes "failures retried and logged, never lost silently" a property rather than an intention: * a webhook that hangs for its full timeout delays an alert, not an engine's result submission; * a receiver that is down is retried with exponential backoff (60s, doubling); * after the attempt ceiling the row goes `failed` **and stays**, holding the error that ended it, so "what did we never send?" is a query; * failures time cannot fix — no URL, no relay configured, HTTP 401/403/400 — skip the retries, because five attempts only delay the operator finding out. A `(channel, finding, scan)` unique constraint makes enqueueing idempotent. The stalled-scan sweep re-finalizes scans routinely, and it must not re-announce the same secret every beat. **What gets announced** is what this scan opened: first seen here, or seen again after being resolved. Not every open finding on every scan — an operator told weekly about the same secret stops reading the alerts. Suppressed findings are excluded: honouring a suppression in the console and mailing anyway would be worse than not having suppressions (ADR 0008). **Egress is treated as a boundary** (docs/security.md § Notification egress). The payload is built from an explicit field list, never by serialising the ORM row, so a new column on Finding cannot silently start being exported; it carries the engine-redacted snippet (ADR 0004) and no analyst notes. Requests are signed (HMAC-SHA256 over `timestamp.body`) when the channel has a secret, redirects are not followed — a 302 would relocate where findings go without editing the channel — and response bodies never reach a log line. Email is plain text because resource locators come from scanned systems. Payload, headers, signature verification and receiver expectations are documented in `docs/notifications.md`. Tested: filter/suppression/disabled-channel exclusion, idempotent re-enqueue, delivery, retry scheduling, backoff timing, exhaustion, permanent failures, a crashing transport, signature correctness, redirect refusal, and SMTP end to end against an in-process server that actually speaks SMTP. Closes #60 Co-Authored-By: Claude Opus 5 (1M context) --- .env.example | 22 + apps/api/src/iceberg_api/maintenance.py | 23 +- .../versions/0006_notification_delivery.py | 100 ++ .../src/iceberg_api/notifications/dispatch.py | 344 +++++++ .../src/iceberg_api/notifications/payload.py | 108 +++ .../iceberg_api/notifications/transports.py | 239 +++++ apps/api/src/iceberg_api/scans/service.py | 8 + apps/api/tests/test_maintenance.py | 66 +- apps/api/tests/test_notification_dispatch.py | 853 ++++++++++++++++++ deploy/compose/docker-compose.yml | 10 + docs/backlog.md | 4 +- docs/notifications.md | 177 ++++ docs/security.md | 16 + packages/core/src/iceberg_core/config.py | 30 + packages/core/src/iceberg_core/enums.py | 12 + .../core/src/iceberg_core/models/__init__.py | 3 +- .../src/iceberg_core/models/notifications.py | 66 +- 17 files changed, 2060 insertions(+), 21 deletions(-) create mode 100644 apps/api/src/iceberg_api/migrations/versions/0006_notification_delivery.py create mode 100644 apps/api/src/iceberg_api/notifications/dispatch.py create mode 100644 apps/api/src/iceberg_api/notifications/payload.py create mode 100644 apps/api/src/iceberg_api/notifications/transports.py create mode 100644 apps/api/tests/test_notification_dispatch.py create mode 100644 docs/notifications.md diff --git a/.env.example b/.env.example index fd07125..bd44c8c 100644 --- a/.env.example +++ b/.env.example @@ -66,6 +66,28 @@ ICEBERG_BOOTSTRAP_ADMIN_EMAIL=you@example.test # corpus, lower it to see what the rules are nearly matching. ICEBERG_CONFIDENCE_THRESHOLD=0.5 +# ─── Notifications (api role) ───────────────────────────────────────────────── +# Channels are configured in the console; these are the deployment facts behind +# them (docs/notifications.md). Delivery runs in the API's maintenance loop, so +# none of this reaches an engine. +# +# Leave ICEBERG_SMTP_HOST empty to disable email: an email channel's deliveries +# then fail loudly naming this variable, rather than quietly going nowhere. +ICEBERG_SMTP_HOST= +ICEBERG_SMTP_PORT=587 +# Omit both for an unauthenticated relay. +ICEBERG_SMTP_USERNAME= +ICEBERG_SMTP_PASSWORD= +# Certificates are verified. Only turn this off for a relay on localhost. +ICEBERG_SMTP_STARTTLS=true +ICEBERG_SMTP_FROM=icebergsst@localhost + +# Attempts before a delivery is given up on. The row is kept either way — giving +# up is recorded, never silent. +ICEBERG_NOTIFICATION_MAX_ATTEMPTS=5 +# First retry delay in seconds; doubles each attempt. +ICEBERG_NOTIFICATION_RETRY_BACKOFF_SECONDS=60 + # ─── Engine role ────────────────────────────────────────────────────────────── # The per-engine API credential. In production this is minted at deploy time # (docs/security.md § Bootstrap), never defaulted into an image. diff --git a/apps/api/src/iceberg_api/maintenance.py b/apps/api/src/iceberg_api/maintenance.py index f1d3cfa..0005519 100644 --- a/apps/api/src/iceberg_api/maintenance.py +++ b/apps/api/src/iceberg_api/maintenance.py @@ -24,9 +24,11 @@ import structlog from iceberg_core.config import ApiSettings, get_api_settings from iceberg_core.db import session_scope +from iceberg_core.secrets import SecretStore, build_secret_store from iceberg_api import suppressions from iceberg_api.dispatch import Dispatcher, build_dispatcher +from iceberg_api.notifications import dispatch as notification_dispatch from iceberg_api.scans import service from iceberg_api.scheduler import postgres_advisory_lock, tick from iceberg_api.scheduler_launcher import build_launcher @@ -34,8 +36,14 @@ logger = structlog.get_logger() -def run_once(dispatcher: Dispatcher, *, now: datetime | None = None) -> None: - """One maintenance round: schedules, reclaim, then the safety sweeps. +def run_once( + dispatcher: Dispatcher, + *, + now: datetime | None = None, + settings: ApiSettings | None = None, + store: SecretStore | None = None, +) -> None: + """One maintenance round: schedules, reclaim, the safety sweeps, then alerts. 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 @@ -43,8 +51,13 @@ def run_once(dispatcher: Dispatcher, *, now: datetime | None = None) -> None: gone after the first schedule fired, and every replica would run the rest of the round at once. The guard session runs no other statements, so its 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. """ at = now or datetime.now(UTC) + resolved = settings or get_api_settings() + secret_store = store or build_secret_store(resolved) with session_scope() as guard: if not postgres_advisory_lock(guard): # Another replica is doing this round — the whole round, sweeps @@ -69,6 +82,12 @@ def run_once(dispatcher: Dispatcher, *, now: datetime | None = None) -> None: # that source, which for a weekly cadence is six days of silence (ADR 0008). with session_scope() as db: suppressions.release_lapsed(db, now=at) + # Announcements queued by reconciliation (#60). Sending here rather than at + # ingest means a webhook that hangs for its full timeout delays an alert + # instead of an engine's result submission, and a receiver that is down + # 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) def _already_leader(db: object) -> bool: diff --git a/apps/api/src/iceberg_api/migrations/versions/0006_notification_delivery.py b/apps/api/src/iceberg_api/migrations/versions/0006_notification_delivery.py new file mode 100644 index 0000000..5563251 --- /dev/null +++ b/apps/api/src/iceberg_api/migrations/versions/0006_notification_delivery.py @@ -0,0 +1,100 @@ +"""Notification delivery outbox + +Dispatch (#60) writes its intention to announce a finding in the same transaction +that opens the finding, and sends afterwards from the maintenance loop. That needs +a table: a row per (channel, finding, scan), carrying attempt count, when it next +becomes due, and the error that ended it if it never went out. + +The unique constraint is the load-bearing part. Enqueueing is re-run whenever a +stalled scan is re-finalized, so without it the safety sweep would announce the +same secret again every beat. + +Revision ID: 0006 +Revises: 0005 +Create Date: 2026-07-31 07:10:00.000000+00:00 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes # AutoString appears in autogenerated column ops +from alembic import op + +revision: str = "0006" +down_revision: str | None = "0005" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "notification_delivery", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("channel_id", sa.Uuid(), nullable=False), + sa.Column("finding_id", sa.Uuid(), nullable=False), + sa.Column("scan_id", sa.Uuid(), nullable=False), + sa.Column( + "status", + sa.Enum( + "pending", + "delivered", + "failed", + name="notification_delivery_status", + native_enum=False, + length=32, + ), + nullable=False, + ), + sa.Column("attempts", sa.Integer(), nullable=False), + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("delivered_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error", sqlmodel.sql.sqltypes.AutoString(length=500), nullable=True), + # CASCADE throughout: a delivery record is about a finding, not part of + # its history, so removing the finding, the channel, or the scan should + # take the record rather than be blocked by it (contrast finding's own + # RESTRICT scan references, added in 0004). + sa.ForeignKeyConstraint( + ["channel_id"], + ["notification_channel.id"], + name=op.f("fk_notification_delivery_channel_id_notification_channel"), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["finding_id"], + ["finding.id"], + name=op.f("fk_notification_delivery_finding_id_finding"), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["scan_id"], + ["scan.id"], + name=op.f("fk_notification_delivery_scan_id_scan"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_notification_delivery")), + # What makes re-enqueueing idempotent, and therefore what stops the + # stalled-scan sweep re-announcing every secret it re-finalizes. + sa.UniqueConstraint( + "channel_id", + "finding_id", + "scan_id", + name="uq_notification_delivery_channel_finding_scan", + ), + ) + # The delivery loop's only query: pending rows that are due. + op.create_index( + "ix_notification_delivery_status_next_attempt_at", + "notification_delivery", + ["status", "next_attempt_at"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index( + "ix_notification_delivery_status_next_attempt_at", + table_name="notification_delivery", + ) + op.drop_table("notification_delivery") diff --git a/apps/api/src/iceberg_api/notifications/dispatch.py b/apps/api/src/iceberg_api/notifications/dispatch.py new file mode 100644 index 0000000..bcc54bd --- /dev/null +++ b/apps/api/src/iceberg_api/notifications/dispatch.py @@ -0,0 +1,344 @@ +"""Deciding who to tell, and then telling them (#60). + +Two halves, deliberately in different transactions. + +:func:`enqueue_for_scan` runs at the end of reconciliation, in the transaction +that finished the scan. It writes one ``NotificationDelivery`` row per (channel, +finding) that qualifies and sends nothing. A transactional outbox: if the scan +commits, the intention to announce commits with it, and no webhook timeout can +roll back a scan or drop an alert. + +:func:`deliver_pending` runs in the maintenance loop, under the same advisory +lock as everything else there, so one replica delivers even when five are up. It +attempts due rows, marks them delivered, or schedules a retry with exponential +backoff until the attempt ceiling — at which point the row goes ``failed`` and +stays, holding the error that ended it. Nothing is deleted, so "what were we +never able to send?" is a query rather than a log search. + +Which findings qualify is narrower than it might look. A finding is announced +when **this scan opened it** — first seen here, or seen again after having been +resolved. Not every open finding every scan: an operator who is told about the +same secret weekly stops reading the alerts, and the finding is on the console +either way. +""" + +import uuid +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta + +import structlog +from iceberg_core.config import ApiSettings +from iceberg_core.enums import ( + FindingEventKind, + FindingState, + NotificationChannelType, + NotificationDeliveryStatus, + Severity, +) +from iceberg_core.models import ( + Finding, + FindingEvent, + NotificationChannel, + NotificationDelivery, + Scan, + Source, +) +from iceberg_core.secrets import SecretStore +from sqlmodel import Session, col, select + +from iceberg_api.notifications.payload import email_subject, finding_opened +from iceberg_api.notifications.schemas import EventFilter +from iceberg_api.notifications.transports import ( + DeliveryError, + Transport, + build_transports, +) + +logger = structlog.get_logger() + +#: Severity order for the ``min_severity`` filter. Defined here rather than on the +#: enum because "critical is worse than high" is a policy this filter applies, not +#: a property of the label. +_SEVERITY_RANK: dict[Severity, int] = { + Severity.LOW: 0, + Severity.MEDIUM: 1, + Severity.HIGH: 2, + Severity.CRITICAL: 3, +} + + +@dataclass(frozen=True, slots=True) +class DeliveryOutcome: + """What one delivery round did.""" + + delivered: int = 0 + retrying: int = 0 + failed: int = 0 + + +def newly_opened_findings(db: Session, scan: Scan) -> list[Finding]: + """Findings this scan opened: first seen here, or re-opened here. + + Suppressed findings are excluded. A suppression is an analyst saying "stop + telling me about this" (ADR 0008), and it would be a poor system that honoured + that in the UI and mailed them anyway. + """ + first_seen = select(Finding).where( + col(Finding.first_seen_scan_id) == scan.id, + col(Finding.state) == FindingState.OPEN, + col(Finding.suppressed_at).is_(None), + ) + # A re-opened finding was first seen by some earlier scan, so it is identified + # by the event ingest wrote when the secret came back — which names this scan. + reopened = ( + select(Finding) + .join(FindingEvent, col(FindingEvent.finding_id) == col(Finding.id)) + .where( + col(Finding.last_seen_scan_id) == scan.id, + col(Finding.state) == FindingState.OPEN, + col(Finding.suppressed_at).is_(None), + col(FindingEvent.kind) == FindingEventKind.REOPENED, + col(FindingEvent.comment) == f"seen again by scan {scan.id}", + ) + ) + findings = {finding.id: finding for finding in db.exec(first_seen)} + findings.update({finding.id: finding for finding in db.exec(reopened)}) + return list(findings.values()) + + +def channel_wants(channel: NotificationChannel, finding: Finding) -> bool: + """Whether this channel's event filter selects this finding.""" + event_filter = EventFilter.model_validate(channel.event_filter) + if event_filter.min_severity is not None and ( + _SEVERITY_RANK[finding.severity] < _SEVERITY_RANK[event_filter.min_severity] + ): + return False + return not (event_filter.source_ids and finding.source_id not in event_filter.source_ids) + + +def enqueue_for_scan(db: Session, scan: Scan, *, now: datetime | None = None) -> int: + """Record an announcement per (enabled channel, newly-opened finding). + + Does not commit — the caller owns the transaction, which is the whole point + of an outbox. Returns how many rows were written. + """ + channels = list(db.exec(select(NotificationChannel).where(col(NotificationChannel.enabled)))) + if not channels: + return 0 + + findings = newly_opened_findings(db, scan) + if not findings: + return 0 + + at = now or datetime.now(UTC) + queued = 0 + for finding in findings: + for channel in channels: + if not channel_wants(channel, finding): + continue + # The unique constraint is the real guard against double-announcing; + # this check just avoids a savepoint in the common case. Re-running + # enqueue for the same scan (the stalled-scan sweep does exactly that) + # must be a no-op. + already = db.exec( + select(NotificationDelivery).where( + col(NotificationDelivery.channel_id) == channel.id, + col(NotificationDelivery.finding_id) == finding.id, + col(NotificationDelivery.scan_id) == scan.id, + ) + ).first() + if already is not None: + continue + db.add( + NotificationDelivery( + channel_id=channel.id, + finding_id=finding.id, + scan_id=scan.id, + status=NotificationDeliveryStatus.PENDING, + next_attempt_at=at, + ) + ) + queued += 1 + + if queued: + logger.info( + "notifications_enqueued", + scan_id=str(scan.id), + findings=len(findings), + channels=len(channels), + queued=queued, + ) + return queued + + +def deliver_pending( + db: Session, + settings: ApiSettings, + store: SecretStore, + *, + now: datetime | None = None, + transports: dict[NotificationChannelType, Transport] | None = None, +) -> DeliveryOutcome: + """Attempt every due delivery, up to the configured batch size.""" + at = now or datetime.now(UTC) + resolved = transports if transports is not None else build_transports(settings, store) + + due = list( + db.exec( + select(NotificationDelivery) + .where( + col(NotificationDelivery.status) == NotificationDeliveryStatus.PENDING, + col(NotificationDelivery.next_attempt_at) <= at, + ) + .order_by(col(NotificationDelivery.next_attempt_at)) + .limit(settings.notification_batch_size) + ) + ) + if not due: + return DeliveryOutcome() + + delivered = retrying = failed = 0 + for delivery in due: + result = _attempt(db, delivery, settings, resolved, at=at) + match result: + case NotificationDeliveryStatus.DELIVERED: + delivered += 1 + case NotificationDeliveryStatus.FAILED: + failed += 1 + case _: + retrying += 1 + db.commit() + + logger.info( + "notifications_delivered", + attempted=len(due), + delivered=delivered, + retrying=retrying, + failed=failed, + ) + return DeliveryOutcome(delivered=delivered, retrying=retrying, failed=failed) + + +def _attempt( + db: Session, + delivery: NotificationDelivery, + settings: ApiSettings, + transports: dict[NotificationChannelType, Transport], + *, + at: datetime, +) -> NotificationDeliveryStatus: + """One attempt. Always leaves the row in a defensible state.""" + delivery.attempts += 1 + + channel = db.get(NotificationChannel, delivery.channel_id) + finding = db.get(Finding, delivery.finding_id) + scan = db.get(Scan, delivery.scan_id) + source = db.get(Source, finding.source_id) if finding is not None else None + + if channel is None or finding is None or scan is None or source is None: + # Deleted between enqueue and delivery. Nothing to announce and nothing to + # retry, so this is a terminal state rather than an error worth alarming on. + return _fail(db, delivery, "referenced row no longer exists", at=at) + if not channel.enabled: + # Disabling a channel is an operator saying stop, including for what is + # already queued. + return _fail(db, delivery, "channel disabled before delivery", at=at) + + transport = transports.get(channel.type) + if transport is None: # pragma: no cover — every enum member has a transport + return _fail(db, delivery, f"no transport for channel type {channel.type}", at=at) + + payload = finding_opened(finding, source=source, scan=scan, channel=channel) + try: + transport.send(channel, payload, subject=email_subject(finding, source)) + except DeliveryError as exc: + return _retry_or_fail(db, delivery, exc, settings, at=at) + except Exception as exc: # a transport bug must not strand the row + logger.exception("notification_transport_crashed", delivery_id=str(delivery.id)) + crash = DeliveryError(f"transport error: {exc.__class__.__name__}") + return _retry_or_fail(db, delivery, crash, settings, at=at) + + delivery.status = NotificationDeliveryStatus.DELIVERED + delivery.delivered_at = at + delivery.last_error = None + db.add(delivery) + logger.info( + "notification_sent", + delivery_id=str(delivery.id), + channel=channel.name, + channel_type=channel.type.value, + finding_id=str(finding.id), + attempts=delivery.attempts, + ) + return NotificationDeliveryStatus.DELIVERED + + +def _retry_or_fail( + db: Session, + delivery: NotificationDelivery, + error: DeliveryError, + settings: ApiSettings, + *, + at: datetime, +) -> NotificationDeliveryStatus: + exhausted = delivery.attempts >= settings.notification_max_attempts + if error.permanent or exhausted: + return _fail(db, delivery, str(error), at=at, exhausted=exhausted) + + # Exponential: 60s, 120s, 240s… from the first failure, so a receiver that is + # restarting is not hammered and one that is down for ten minutes is still + # caught by the ceiling. + delay = settings.notification_retry_backoff_seconds * (2 ** (delivery.attempts - 1)) + delivery.next_attempt_at = at + timedelta(seconds=delay) + delivery.last_error = str(error) + db.add(delivery) + logger.warning( + "notification_retry_scheduled", + delivery_id=str(delivery.id), + attempts=delivery.attempts, + retry_in_seconds=delay, + error=str(error), + ) + return NotificationDeliveryStatus.PENDING + + +def _fail( + db: Session, + delivery: NotificationDelivery, + reason: str, + *, + at: datetime, + exhausted: bool = False, +) -> NotificationDeliveryStatus: + delivery.status = NotificationDeliveryStatus.FAILED + delivery.last_error = reason + delivery.next_attempt_at = at + db.add(delivery) + logger.error( + "notification_failed", + delivery_id=str(delivery.id), + attempts=delivery.attempts, + exhausted=exhausted, + reason=reason, + ) + return NotificationDeliveryStatus.FAILED + + +def pending_count(db: Session, *, channel_id: uuid.UUID | None = None) -> int: + """How much is queued. Used by the tests and worth having for an operator.""" + statement = select(NotificationDelivery).where( + col(NotificationDelivery.status) == NotificationDeliveryStatus.PENDING + ) + if channel_id is not None: + statement = statement.where(col(NotificationDelivery.channel_id) == channel_id) + return len(list(db.exec(statement))) + + +__all__ = [ + "DeliveryOutcome", + "channel_wants", + "deliver_pending", + "enqueue_for_scan", + "newly_opened_findings", + "pending_count", +] diff --git a/apps/api/src/iceberg_api/notifications/payload.py b/apps/api/src/iceberg_api/notifications/payload.py new file mode 100644 index 0000000..8cbce9f --- /dev/null +++ b/apps/api/src/iceberg_api/notifications/payload.py @@ -0,0 +1,108 @@ +"""What an announcement contains (#60). + +This module is the whole answer to "what leaves the deployment when a finding +opens", which is why it is separate from the transports that send it and from the +loop that schedules it. A webhook receiver is operator-supplied and outside the +trust boundary (docs/security.md § Notification egress), so the payload is built +from an explicit field list rather than by serialising the ORM row: adding a +column to ``Finding`` must never silently start exporting it. + +Two things are deliberately absent: + +* **The secret.** Only the redacted snippet the engine produced (ADR 0004) and + the peppered hash. Neither is reversible, and the plaintext never existed on + this side of the boundary to begin with. +* **Anything an analyst wrote.** Notes and assignee are internal triage state; + a chat relay does not need them, and they can contain anything a person typed. + +The shape is a documented contract — ``docs/notifications.md`` publishes it and +``apps/api/tests/test_notification_dispatch.py`` pins it, so a receiver that +parses it keeps working. +""" + +from typing import Any + +from iceberg_core.models import Finding, NotificationChannel, Scan, Source + +#: Bumped when a field is removed or changes meaning. Additive changes do not +#: bump it: a receiver that ignores unknown keys keeps working, and one that +#: does not was going to break anyway. +PAYLOAD_VERSION = "1" + +#: An event name rather than a bare object, so a receiver routing several kinds +#: of message can switch on it and so adding kinds later is not a breaking change. +EVENT_FINDING_OPENED = "finding.opened" + +#: How much snippet to send. Already redacted; capped so a channel cannot be used +#: to stream a large document out through a series of findings. +_MAX_SNIPPET = 500 + + +def finding_opened( + finding: Finding, + *, + source: Source, + scan: Scan, + channel: NotificationChannel, + console_url: str | None = None, +) -> dict[str, Any]: + """The JSON body for a newly-opened finding. + + ``console_url`` is a deep link when the deployment knows its own address; the + useful thing to put in an alert is a way to go and look at it. + """ + return { + "version": PAYLOAD_VERSION, + "event": EVENT_FINDING_OPENED, + "channel": {"id": str(channel.id), "name": channel.name}, + "finding": { + "id": str(finding.id), + "fingerprint": finding.fingerprint, + "rule_id": finding.rule_id, + "rulepack_version": finding.rulepack_version, + "severity": finding.severity.value, + "confidence": finding.confidence, + "state": finding.state.value, + # Redacted inside the engine before it ever reached the API (ADR 0004). + "redacted_snippet": finding.redacted_snippet[:_MAX_SNIPPET], + "resource_locator": finding.resource_locator, + "first_seen_at": finding.created_at.isoformat(), + "url": console_url, + }, + "source": {"id": str(source.id), "name": source.name, "type": source.type.value}, + "scan": {"id": str(scan.id), "trigger": scan.trigger.value}, + } + + +def email_subject(finding: Finding, source: Source) -> str: + """The subject line. Severity first, because it is what triages the inbox.""" + return f"[IcebergSST] {finding.severity.value.upper()} secret in {source.name}" + + +def email_body(payload: dict[str, Any]) -> str: + """A plain-text rendering of the same payload. + + Plain text on purpose: an HTML mail containing attacker-influenced content — + a resource locator is a path from a scanned system — is a small XSS surface in + whatever client opens it, for no gain over a readable summary. + """ + finding = payload["finding"] + source = payload["source"] + lines = [ + f"A {finding['severity']} secret was found in {source['name']} ({source['type']}).", + "", + f"Rule: {finding['rule_id']} (pack {finding['rulepack_version']})", + f"Severity: {finding['severity']}", + f"Confidence: {finding['confidence']}", + f"Fingerprint: {finding['fingerprint']}", + f"Location: {finding['resource_locator']}", + "", + "Redacted context:", + f" {finding['redacted_snippet']}", + "", + ] + if finding.get("url"): + lines.append(f"Triage it here: {finding['url']}") + lines.append("") + lines.append("The secret itself is never included in this message.") + return "\n".join(lines) diff --git a/apps/api/src/iceberg_api/notifications/transports.py b/apps/api/src/iceberg_api/notifications/transports.py new file mode 100644 index 0000000..02bde00 --- /dev/null +++ b/apps/api/src/iceberg_api/notifications/transports.py @@ -0,0 +1,239 @@ +"""Getting an announcement out of the deployment (#60). + +Two transports, one interface, so the delivery loop knows nothing about SMTP or +HTTP and tests can substitute a recorder. Both are synchronous and blocking: the +loop that calls them already runs in a worker thread, and a background job is the +right place to wait on somebody else's server. + +The failure contract is the important part. A transport either returns, meaning +delivered, or raises :class:`DeliveryError`, meaning *try again later*. Anything +else escaping would be a bug in a transport, and the loop treats it the same way +— a failed attempt, never a lost row. + +Egress is a security boundary here, not a detail (docs/security.md § Notification +egress). The receiver is an arbitrary operator-supplied URL, so: + +* redirects are not followed — a 302 to somewhere else is a way to move where + findings land without anyone changing the channel; +* the response body is never read into the error, because it is somebody else's + data and ends up in our logs; +* the request is signed when the channel has a secret, so a receiver can tell a + real announcement from anything else that can reach its URL. +""" + +import hashlib +import hmac +import json +import smtplib +import ssl +from datetime import UTC, datetime +from email.message import EmailMessage +from typing import Any, Protocol + +import httpx2 +import structlog +from iceberg_core.config import ApiSettings +from iceberg_core.enums import NotificationChannelType +from iceberg_core.models import NotificationChannel +from iceberg_core.secrets import SecretStore, SecretStoreError + +from iceberg_api.notifications.payload import email_body +from iceberg_api.notifications.schemas import SECRET_REF_KEY + +logger = structlog.get_logger() + +#: Carries the HMAC. Named for the product so it cannot collide with a header the +#: receiver already uses. +SIGNATURE_HEADER = "X-Iceberg-Signature" +#: Signed alongside the body, so a captured request cannot be replayed forever. +TIMESTAMP_HEADER = "X-Iceberg-Timestamp" +EVENT_HEADER = "X-Iceberg-Event" + +#: Errors are stored on the delivery row and printed in logs; keep them small. +_MAX_ERROR = 400 + + +class DeliveryError(Exception): + """A delivery attempt failed. Retryable unless ``permanent`` is set.""" + + def __init__(self, message: str, *, permanent: bool = False) -> None: + super().__init__(message[:_MAX_ERROR]) + #: A misconfigured channel will fail identically forever — retrying a + #: malformed URL 5 times only delays the operator finding out. + self.permanent = permanent + + +class Transport(Protocol): + """Sends one announcement. Returns on success, raises DeliveryError on failure.""" + + def send( + self, + channel: NotificationChannel, + payload: dict[str, Any], + *, + subject: str, + ) -> None: ... + + +def _channel_secret(channel: NotificationChannel, store: SecretStore) -> str | None: + """The channel's signing key, or None. Opened per attempt, never cached.""" + ref = channel.config.get(SECRET_REF_KEY) + if not ref: + return None + try: + return store.open(ref).get_secret_value() + except SecretStoreError as exc: + # A ref that will not open is a configuration problem — usually a master + # key that was rotated without re-sealing — and no number of retries will + # fix it. + raise DeliveryError(f"channel secret could not be opened: {exc}", permanent=True) from exc + + +def sign(body: bytes, secret: str, timestamp: str) -> str: + """``sha256=`` over ``timestamp.body``. + + The timestamp is inside the MAC rather than beside it, so it cannot be edited + without invalidating the signature. Receivers should reject a stale one. + """ + mac = hmac.new(secret.encode(), f"{timestamp}.".encode() + body, hashlib.sha256) + return f"sha256={mac.hexdigest()}" + + +class WebhookTransport: + """POSTs the payload as JSON to the channel's URL.""" + + def __init__( + self, + settings: ApiSettings, + store: SecretStore, + *, + transport: httpx2.BaseTransport | None = None, + ) -> None: + self._settings = settings + self._store = store + #: Injected in tests so the suite never opens a socket. + self._transport = transport + + def send( + self, + channel: NotificationChannel, + payload: dict[str, Any], + *, + subject: str, + ) -> None: + url = channel.config.get("url") + if not url: + raise DeliveryError("webhook channel has no url", permanent=True) + + body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() + timestamp = str(int(datetime.now(UTC).timestamp())) + headers = { + "Content-Type": "application/json", + EVENT_HEADER: str(payload.get("event", "")), + TIMESTAMP_HEADER: timestamp, + **{str(k): str(v) for k, v in channel.config.get("headers", {}).items()}, + } + secret = _channel_secret(channel, self._store) + if secret is not None: + headers[SIGNATURE_HEADER] = sign(body, secret, timestamp) + + try: + with httpx2.Client( + timeout=self._settings.webhook_timeout_seconds, + # See the module docstring: a redirect would silently relocate + # where findings are sent. + follow_redirects=False, + transport=self._transport, + ) as client: + response = client.post(url, content=body, headers=headers) + except httpx2.HTTPError as exc: + raise DeliveryError(f"webhook request failed: {exc.__class__.__name__}") from exc + + if response.status_code >= 400: + # 4xx is usually configuration and 5xx usually transient, but a 429 is + # explicitly "try later" and a 404 can be a receiver mid-deploy. Only + # the codes that time cannot change count as permanent. + permanent = response.status_code in {400, 401, 403, 405, 410, 413, 414, 415} + raise DeliveryError( + f"webhook returned HTTP {response.status_code}", permanent=permanent + ) + + +class SmtpTransport: + """Sends the payload as a plain-text email to the channel's recipients.""" + + def __init__(self, settings: ApiSettings) -> None: + self._settings = settings + + def send( + self, + channel: NotificationChannel, + payload: dict[str, Any], + *, + subject: str, + ) -> None: + settings = self._settings + if not settings.smtp_host: + # Configured channel, unconfigured deployment. Permanent, because + # retrying cannot supply a relay — and the error says exactly what to + # set, which is the point of failing loudly instead of dropping mail. + raise DeliveryError("no SMTP relay configured; set ICEBERG_SMTP_HOST", permanent=True) + + recipients = [str(address) for address in channel.config.get("recipients", [])] + if not recipients: + raise DeliveryError("email channel has no recipients", permanent=True) + + message = EmailMessage() + message["Subject"] = subject + message["From"] = settings.smtp_from + message["To"] = ", ".join(recipients) + message.set_content(email_body(payload)) + + try: + with self._connect() as server: + server.send_message(message) + except (OSError, smtplib.SMTPException) as exc: + # Includes authentication failures, which are configuration — but a + # relay can also reject credentials while it is degraded, so these + # stay retryable and exhaust normally rather than giving up at once. + raise DeliveryError(f"smtp send failed: {exc.__class__.__name__}: {exc}") from exc + + def _connect(self) -> smtplib.SMTP: + settings = self._settings + server = smtplib.SMTP( + host=str(settings.smtp_host), + port=settings.smtp_port, + timeout=settings.smtp_timeout_seconds, + ) + if settings.smtp_starttls: + # A default context verifies the certificate and hostname. A relay + # with a self-signed certificate should be fixed, not accommodated: + # this connection carries where our secrets are. + server.starttls(context=ssl.create_default_context()) + if settings.smtp_username and settings.smtp_password: + server.login(settings.smtp_username, settings.smtp_password.get_secret_value()) + return server + + +def build_transports( + settings: ApiSettings, + store: SecretStore, +) -> dict[NotificationChannelType, Transport]: + """The transport for each channel type. One place the loop looks things up.""" + return { + NotificationChannelType.WEBHOOK: WebhookTransport(settings, store), + NotificationChannelType.EMAIL: SmtpTransport(settings), + } + + +__all__ = [ + "EVENT_HEADER", + "SIGNATURE_HEADER", + "TIMESTAMP_HEADER", + "DeliveryError", + "SmtpTransport", + "Transport", + "WebhookTransport", + "build_transports", + "sign", +] diff --git a/apps/api/src/iceberg_api/scans/service.py b/apps/api/src/iceberg_api/scans/service.py index fcd5370..3ad68a6 100644 --- a/apps/api/src/iceberg_api/scans/service.py +++ b/apps/api/src/iceberg_api/scans/service.py @@ -35,6 +35,7 @@ from sqlmodel import Session, col, select, update from iceberg_api.dispatch import Dispatcher +from iceberg_api.notifications import dispatch as notification_dispatch from iceberg_api.scans.reconcile import reconcile_scan #: How long a lease is good for without a heartbeat. Long enough for a slow @@ -355,6 +356,13 @@ def finalize_and_reconcile( if scan is not None: # pragma: no branch — the UPDATE just matched it db.refresh(scan) reconcile_scan(db, scan, now=now) + # Queue announcements for what this scan opened (#60). Writing the + # outbox rows here — after reconciliation, so a finding auto-resolved + # in the same pass is not announced — keeps "the scan finished" and + # "somebody will be told" in one transaction. Sending happens in the + # maintenance loop; nothing here talks to SMTP or a webhook. + if notification_dispatch.enqueue_for_scan(db, scan, now=now): + db.commit() return final diff --git a/apps/api/tests/test_maintenance.py b/apps/api/tests/test_maintenance.py index b7e5bcc..9d95ddd 100644 --- a/apps/api/tests/test_maintenance.py +++ b/apps/api/tests/test_maintenance.py @@ -7,6 +7,7 @@ from conftest import RecordingDispatcher from iceberg_api import maintenance from iceberg_api.scans import service +from iceberg_core.config import ApiSettings from iceberg_core.db import set_db_engine from iceberg_core.enums import ( ScanStatus, @@ -24,6 +25,8 @@ Source, Suppression, ) +from iceberg_core.secrets import SecretStore +from pydantic import SecretStr from sqlalchemy import Engine as SAEngine from sqlmodel import Session, select @@ -36,6 +39,25 @@ def process_engine_fixture(db_engine: SAEngine) -> Iterator[None]: set_db_engine(None) +@pytest.fixture(name="run_round") +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. + """ + settings = ApiSettings( + database_url="postgresql+psycopg://unused/unused", + master_key=SecretStr("unused"), + ) + + def run(dispatcher: RecordingDispatcher, **kwargs: object) -> None: + maintenance.run_once(dispatcher, settings=settings, store=secret_store, **kwargs) # type: ignore[arg-type] + + return run + + def _source(session: Session, name: str = "confluence-prod") -> Source: source = Source( name=name, @@ -48,7 +70,9 @@ def _source(session: Session, name: str = "confluence-prod") -> Source: def test_a_round_launches_due_scans_and_advances_the_schedule( - session: Session, dispatcher: RecordingDispatcher + session: Session, + dispatcher: RecordingDispatcher, + run_round: Callable[..., None], ) -> None: source = _source(session) schedule = Schedule( @@ -59,7 +83,7 @@ def test_a_round_launches_due_scans_and_advances_the_schedule( session.add(schedule) session.commit() - maintenance.run_once(dispatcher) + run_round(dispatcher) scans = session.exec(select(Scan)).all() assert len(scans) == 1 @@ -69,7 +93,9 @@ def test_a_round_launches_due_scans_and_advances_the_schedule( assert schedule.last_run_at is not None -def test_a_round_reclaims_expired_leases(session: Session, dispatcher: RecordingDispatcher) -> None: +def test_a_round_reclaims_expired_leases( + session: Session, dispatcher: RecordingDispatcher, run_round: Callable[..., None] +) -> None: source = _source(session) scan = service.launch_scan(session, source, trigger=ScanTrigger.MANUAL, dispatcher=dispatcher) task = session.exec(select(ScanTask).where(ScanTask.scan_id == scan.id)).one() @@ -79,7 +105,7 @@ def test_a_round_reclaims_expired_leases(session: Session, dispatcher: Recording service.claim_task(session, task.id, engine.id, lease_seconds=1) dispatcher.enqueued.clear() - maintenance.run_once(dispatcher, now=datetime.now(UTC) + timedelta(minutes=10)) + run_round(dispatcher, now=datetime.now(UTC) + timedelta(minutes=10)) session.refresh(task) assert task.status is ScanTaskStatus.QUEUED @@ -87,7 +113,9 @@ def test_a_round_reclaims_expired_leases(session: Session, dispatcher: Recording def test_a_source_with_an_active_scan_is_skipped_not_double_scanned( - session: Session, dispatcher: RecordingDispatcher + session: Session, + dispatcher: RecordingDispatcher, + run_round: Callable[..., None], ) -> None: """The cadence said "scan now" and one is already running; the next beat will do.""" source = _source(session) @@ -100,7 +128,7 @@ def test_a_source_with_an_active_scan_is_skipped_not_double_scanned( session.add(schedule) session.commit() - maintenance.run_once(dispatcher) + run_round(dispatcher) assert len(session.exec(select(Scan)).all()) == 1 session.refresh(schedule) @@ -109,7 +137,9 @@ def test_a_source_with_an_active_scan_is_skipped_not_double_scanned( def test_a_disabled_source_is_not_scanned_on_its_cadence( - session: Session, dispatcher: RecordingDispatcher + session: Session, + dispatcher: RecordingDispatcher, + run_round: Callable[..., None], ) -> None: source = _source(session) source.enabled = False @@ -121,13 +151,15 @@ def test_a_disabled_source_is_not_scanned_on_its_cadence( session.add(schedule) session.commit() - maintenance.run_once(dispatcher) + run_round(dispatcher) assert session.exec(select(Scan)).all() == [] def test_a_round_redispatches_a_queued_task_whose_message_was_lost( - session: Session, dispatcher: RecordingDispatcher + session: Session, + dispatcher: RecordingDispatcher, + run_round: Callable[..., None], ) -> None: """A crash between commit and enqueue leaves a queued row with no message; the sweep is the only thing that will ever deliver it.""" @@ -137,17 +169,19 @@ def test_a_round_redispatches_a_queued_task_whose_message_was_lost( dispatcher.enqueued.clear() # the launch enqueue is the message that "got lost" later = datetime.now(UTC) + timedelta(minutes=10) - maintenance.run_once(dispatcher, now=later) + run_round(dispatcher, now=later) assert dispatcher.enqueued == [task.id] # Paced: the same round does not spam the queue on the next beat. dispatcher.enqueued.clear() - maintenance.run_once(dispatcher, now=later + timedelta(seconds=30)) + run_round(dispatcher, now=later + timedelta(seconds=30)) assert dispatcher.enqueued == [] def test_a_round_finalizes_a_scan_stranded_by_a_crash( - session: Session, dispatcher: RecordingDispatcher + session: Session, + dispatcher: RecordingDispatcher, + run_round: Callable[..., None], ) -> None: """All tasks terminal but the scan still active: the finalize sweep settles it, so the source is not blocked forever by the one-active-scan index.""" @@ -157,7 +191,7 @@ def test_a_round_finalizes_a_scan_stranded_by_a_crash( service.complete_task(session, task, status=ScanTaskStatus.FAILED, error="engine died") session.commit() # ...and the follow-up finalisation never ran - maintenance.run_once(dispatcher) + run_round(dispatcher) session.refresh(scan) assert scan.status is ScanStatus.FAILED @@ -167,6 +201,7 @@ def test_a_round_finalizes_a_scan_stranded_by_a_crash( def test_a_round_releases_findings_whose_suppression_expired( session: Session, dispatcher: RecordingDispatcher, + run_round: Callable[..., None], make_finding: Callable[..., Finding], ) -> None: """Expiry is a property of the clock, so it cannot wait for the next scan. @@ -188,7 +223,7 @@ def test_a_round_releases_findings_whose_suppression_expired( session.add(finding) session.commit() - maintenance.run_once(dispatcher) + run_round(dispatcher) session.refresh(finding) released = session.get(Finding, finding.id) @@ -200,6 +235,7 @@ def test_a_round_releases_findings_whose_suppression_expired( def test_a_lapsed_suppression_hands_over_to_another_that_still_covers_the_finding( session: Session, dispatcher: RecordingDispatcher, + run_round: Callable[..., None], make_finding: Callable[..., Finding], ) -> None: """A finding covered by both an expiring rule and a permanent one must stay @@ -224,7 +260,7 @@ def test_a_lapsed_suppression_hands_over_to_another_that_still_covers_the_findin session.add(finding) session.commit() - maintenance.run_once(dispatcher) + run_round(dispatcher) session.refresh(finding) assert finding.suppressed_at is not None # still hidden diff --git a/apps/api/tests/test_notification_dispatch.py b/apps/api/tests/test_notification_dispatch.py new file mode 100644 index 0000000..0752bfb --- /dev/null +++ b/apps/api/tests/test_notification_dispatch.py @@ -0,0 +1,853 @@ +"""Notification dispatch (#60). + +Three properties matter here, and they are what these tests are organised around. + +**Nothing is announced that should not be.** The event filter, disabled channels +and suppressions all have to be honoured before a row is written, because once it +is written something will try very hard to deliver it. + +**Nothing is lost, and nothing is sent twice.** The outbox row commits with the +scan; a failure schedules a retry rather than disappearing; giving up is recorded +as ``failed`` with the reason. Re-running enqueue — which the stalled-scan sweep +does routinely — must not announce the same secret again. + +**Nothing leaks.** The payload is a fixed field list, and the assertion that the +plaintext secret is absent is the one that would matter most if it ever failed. +""" + +import json +import threading +import uuid +from collections.abc import Callable, Iterator +from datetime import UTC, datetime, timedelta +from socketserver import StreamRequestHandler, ThreadingTCPServer +from typing import Any + +import httpx2 +import pytest +from iceberg_api.notifications import dispatch +from iceberg_api.notifications.payload import PAYLOAD_VERSION, finding_opened +from iceberg_api.notifications.transports import ( + SIGNATURE_HEADER, + TIMESTAMP_HEADER, + DeliveryError, + SmtpTransport, + WebhookTransport, + sign, +) +from iceberg_core.config import ApiSettings +from iceberg_core.enums import ( + FindingEventKind, + FindingState, + NotificationChannelType, + NotificationDeliveryStatus, + ScanStatus, + ScanTrigger, + Severity, + SourceType, +) +from iceberg_core.models import ( + Finding, + FindingEvent, + NotificationChannel, + NotificationDelivery, + Scan, + Source, +) +from iceberg_core.secrets import SecretStore +from pydantic import SecretStr +from sqlmodel import Session, col, select + +PLAINTEXT_SECRET = "AKIAIOSFODNN7EXAMPLE" + + +def _utc(value: datetime) -> datetime: + """SQLite hands timestamps back without a zone; Postgres keeps it. + + Same helper as ``test_scheduler.py``. Comparisons in application code happen + in SQL, where the driver binds consistently — this is only needed to compare a + value read back in Python. + """ + return value if value.tzinfo else value.replace(tzinfo=UTC) + + +# ── Fixtures ────────────────────────────────────────────────────────────────── + + +@pytest.fixture(name="source") +def source_fixture(session: Session) -> Source: + source = Source( + name=f"confluence-{uuid.uuid4().hex[:6]}", + type=SourceType.CONFLUENCE, + connection={"base_url": "https://example.atlassian.net/wiki"}, + ) + session.add(source) + session.commit() + return source + + +@pytest.fixture(name="scan") +def scan_fixture(session: Session, source: Source) -> Scan: + scan = Scan(source_id=source.id, trigger=ScanTrigger.MANUAL, status=ScanStatus.COMPLETED) + session.add(scan) + session.commit() + return scan + + +@pytest.fixture(name="make_open_finding") +def make_open_finding_fixture( + session: Session, source: Source, scan: Scan +) -> Callable[..., Finding]: + """A finding this scan opened — the thing dispatch announces.""" + + def factory(**fields: Any) -> Finding: + defaults: dict[str, Any] = { + "fingerprint": uuid.uuid4().hex, + "rule_id": "aws-access-key", + "rulepack_version": "2026.07.1", + "resource_locator": {"path": "/space/DOCS/page-1"}, + "redacted_snippet": "AKIA****************", + "secret_hash": uuid.uuid4().hex, + "severity": Severity.HIGH, + "confidence": 0.9, + "state": FindingState.OPEN, + } + finding = Finding( + source_id=source.id, + first_seen_scan_id=scan.id, + last_seen_scan_id=scan.id, + **(defaults | fields), + ) + session.add(finding) + session.commit() + return finding + + return factory + + +@pytest.fixture(name="make_channel") +def make_channel_fixture(session: Session) -> Callable[..., NotificationChannel]: + def factory(**fields: Any) -> NotificationChannel: + defaults: dict[str, Any] = { + "name": f"channel-{uuid.uuid4().hex[:6]}", + "type": NotificationChannelType.WEBHOOK, + "config": {"url": "https://receiver.example.com/hook"}, + "event_filter": {}, + "enabled": True, + } + channel = NotificationChannel(**(defaults | fields)) + session.add(channel) + session.commit() + return channel + + return factory + + +class RecordingTransport: + """Stands in for SMTP/HTTP. Records what it was asked to send, or fails.""" + + def __init__(self, error: DeliveryError | None = None) -> None: + self.error = error + self.sent: list[tuple[NotificationChannel, dict[str, Any], str]] = [] + + def send(self, channel: NotificationChannel, payload: dict[str, Any], *, subject: str) -> None: + if self.error is not None: + raise self.error + self.sent.append((channel, payload, subject)) + + +@pytest.fixture(name="dispatch_settings") +def dispatch_settings_fixture() -> ApiSettings: + return ApiSettings( + database_url="postgresql+psycopg://unused/unused", + master_key=SecretStr("unused"), + notification_max_attempts=3, + notification_retry_backoff_seconds=60, + ) + + +def _transports(transport: RecordingTransport) -> dict[NotificationChannelType, Any]: + return dict.fromkeys(NotificationChannelType, transport) + + +# ── Who gets told ───────────────────────────────────────────────────────────── + + +def test_a_new_finding_is_queued_for_an_enabled_channel( + session: Session, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + channel = make_channel() + finding = make_open_finding() + + queued = dispatch.enqueue_for_scan(session, scan) + session.commit() + + assert queued == 1 + delivery = session.exec(select(NotificationDelivery)).one() + assert delivery.channel_id == channel.id + assert delivery.finding_id == finding.id + assert delivery.scan_id == scan.id + assert delivery.status is NotificationDeliveryStatus.PENDING + assert delivery.attempts == 0 + + +def test_a_finding_below_the_channel_minimum_severity_is_not_queued( + session: Session, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + make_channel(event_filter={"min_severity": Severity.CRITICAL.value}) + make_open_finding(severity=Severity.HIGH) + + assert dispatch.enqueue_for_scan(session, scan) == 0 + + +def test_a_channel_scoped_to_other_sources_is_not_queued( + session: Session, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + make_channel(event_filter={"source_ids": [str(uuid.uuid4())]}) + make_open_finding() + + assert dispatch.enqueue_for_scan(session, scan) == 0 + + +def test_a_disabled_channel_is_not_queued( + session: Session, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + make_channel(enabled=False) + make_open_finding() + + assert dispatch.enqueue_for_scan(session, scan) == 0 + + +def test_a_suppressed_finding_is_not_announced( + session: Session, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + """A suppression is an analyst saying stop telling me about this (ADR 0008). + + Honouring it in the console and mailing them anyway would be worse than not + having suppressions at all. + """ + make_channel() + make_open_finding(suppressed_at=datetime.now(UTC)) + + assert dispatch.enqueue_for_scan(session, scan) == 0 + + +def test_a_finding_this_scan_did_not_open_is_not_announced( + session: Session, + source: Source, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + """Only what this scan opened. An operator told weekly about the same secret + stops reading the alerts.""" + make_channel() + earlier = Scan(source_id=source.id, trigger=ScanTrigger.MANUAL, status=ScanStatus.COMPLETED) + session.add(earlier) + session.commit() + finding = make_open_finding() + finding.first_seen_scan_id = earlier.id + session.add(finding) + session.commit() + + assert dispatch.enqueue_for_scan(session, scan) == 0 + + +def test_a_reopened_finding_is_announced_again( + session: Session, + source: Source, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + """A secret that came back is news, even though the finding is not new.""" + make_channel() + earlier = Scan(source_id=source.id, trigger=ScanTrigger.MANUAL, status=ScanStatus.COMPLETED) + session.add(earlier) + session.commit() + + finding = make_open_finding() + finding.first_seen_scan_id = earlier.id + session.add(finding) + session.add( + FindingEvent( + finding_id=finding.id, + kind=FindingEventKind.REOPENED, + from_value=FindingState.RESOLVED.value, + to_value=FindingState.OPEN.value, + comment=f"seen again by scan {scan.id}", + ) + ) + session.commit() + + assert dispatch.enqueue_for_scan(session, scan) == 1 + + +def test_enqueueing_the_same_scan_twice_announces_once( + session: Session, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + """The stalled-scan sweep re-finalizes scans; it must not re-announce them.""" + make_channel() + make_open_finding() + + dispatch.enqueue_for_scan(session, scan) + session.commit() + second = dispatch.enqueue_for_scan(session, scan) + session.commit() + + assert second == 0 + assert len(list(session.exec(select(NotificationDelivery)))) == 1 + + +# ── Delivery, retry, and giving up ──────────────────────────────────────────── + + +def test_a_due_delivery_is_sent_and_marked_delivered( + session: Session, + scan: Scan, + dispatch_settings: ApiSettings, + secret_store: SecretStore, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + make_channel() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + transport = RecordingTransport() + + outcome = dispatch.deliver_pending( + session, dispatch_settings, secret_store, transports=_transports(transport) + ) + + assert outcome.delivered == 1 + assert len(transport.sent) == 1 + delivery = session.exec(select(NotificationDelivery)).one() + assert delivery.status is NotificationDeliveryStatus.DELIVERED + assert delivery.delivered_at is not None + assert delivery.last_error is None + + +def test_a_delivered_row_is_not_sent_again( + session: Session, + scan: Scan, + dispatch_settings: ApiSettings, + secret_store: SecretStore, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + make_channel() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + transport = RecordingTransport() + transports = _transports(transport) + + dispatch.deliver_pending(session, dispatch_settings, secret_store, transports=transports) + dispatch.deliver_pending(session, dispatch_settings, secret_store, transports=transports) + + assert len(transport.sent) == 1 + + +def test_a_transient_failure_schedules_a_retry_and_keeps_the_row( + session: Session, + scan: Scan, + dispatch_settings: ApiSettings, + secret_store: SecretStore, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + make_channel() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + at = datetime.now(UTC) + transport = RecordingTransport(error=DeliveryError("receiver is down")) + + outcome = dispatch.deliver_pending( + session, + dispatch_settings, + secret_store, + now=at, + transports=_transports(transport), + ) + + assert outcome.retrying == 1 + delivery = session.exec(select(NotificationDelivery)).one() + assert delivery.status is NotificationDeliveryStatus.PENDING + assert delivery.attempts == 1 + assert delivery.last_error == "receiver is down" + # Backoff pushed it out of the current round, so the next beat does not + # hammer a receiver that is restarting. + assert _utc(delivery.next_attempt_at) == at + timedelta( + seconds=dispatch_settings.notification_retry_backoff_seconds + ) + + +def test_a_retry_is_not_attempted_before_it_is_due( + session: Session, + scan: Scan, + dispatch_settings: ApiSettings, + secret_store: SecretStore, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + make_channel() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + at = datetime.now(UTC) + failing = RecordingTransport(error=DeliveryError("receiver is down")) + dispatch.deliver_pending( + session, dispatch_settings, secret_store, now=at, transports=_transports(failing) + ) + + recovered = RecordingTransport() + dispatch.deliver_pending( + session, + dispatch_settings, + secret_store, + now=at + timedelta(seconds=1), + transports=_transports(recovered), + ) + + assert recovered.sent == [] + + +def test_giving_up_is_recorded_rather_than_silent( + session: Session, + scan: Scan, + dispatch_settings: ApiSettings, + secret_store: SecretStore, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + """After the attempt ceiling the row goes `failed` — and stays, with the error. + + "Never lost silently" means an operator can ask what was never delivered and + get an answer, which is a query against these rows. + """ + make_channel() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + transport = RecordingTransport(error=DeliveryError("receiver is down")) + + at = datetime.now(UTC) + for attempt in range(dispatch_settings.notification_max_attempts): + dispatch.deliver_pending( + session, + dispatch_settings, + secret_store, + # Far enough ahead that each round finds the row due again. + now=at + timedelta(hours=attempt), + transports=_transports(transport), + ) + + delivery = session.exec(select(NotificationDelivery)).one() + assert delivery.status is NotificationDeliveryStatus.FAILED + assert delivery.attempts == dispatch_settings.notification_max_attempts + assert delivery.last_error == "receiver is down" + + +def test_a_permanent_failure_is_not_retried( + session: Session, + scan: Scan, + dispatch_settings: ApiSettings, + secret_store: SecretStore, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + """A malformed URL fails identically forever; five attempts only delay the news.""" + make_channel() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + transport = RecordingTransport(error=DeliveryError("no url", permanent=True)) + + outcome = dispatch.deliver_pending( + session, dispatch_settings, secret_store, transports=_transports(transport) + ) + + assert outcome.failed == 1 + delivery = session.exec(select(NotificationDelivery)).one() + assert delivery.status is NotificationDeliveryStatus.FAILED + assert delivery.attempts == 1 + + +def test_a_channel_disabled_after_queueing_is_not_delivered( + session: Session, + scan: Scan, + dispatch_settings: ApiSettings, + secret_store: SecretStore, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + channel = make_channel() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + channel.enabled = False + session.add(channel) + session.commit() + transport = RecordingTransport() + + dispatch.deliver_pending( + session, dispatch_settings, secret_store, transports=_transports(transport) + ) + + assert transport.sent == [] + delivery = session.exec(select(NotificationDelivery)).one() + assert delivery.status is NotificationDeliveryStatus.FAILED + + +def test_a_transport_that_crashes_does_not_strand_the_row( + session: Session, + scan: Scan, + dispatch_settings: ApiSettings, + secret_store: SecretStore, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + """A bug in a transport is a failed attempt, not a lost announcement.""" + + class Exploding: + def send(self, channel: object, payload: object, *, subject: str) -> None: + raise RuntimeError("boom") + + make_channel() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + + outcome = dispatch.deliver_pending( + session, + dispatch_settings, + secret_store, + transports=dict.fromkeys(NotificationChannelType, Exploding()), + ) + + assert outcome.retrying == 1 + delivery = session.exec(select(NotificationDelivery)).one() + assert delivery.status is NotificationDeliveryStatus.PENDING + assert delivery.last_error is not None and "RuntimeError" in delivery.last_error + + +# ── The payload ─────────────────────────────────────────────────────────────── + + +def test_the_payload_carries_no_secret_and_a_documented_shape( + session: Session, + source: Source, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + channel = make_channel() + finding = make_open_finding(redacted_snippet="AKIA****************") + + payload = finding_opened(finding, source=source, scan=scan, channel=channel) + + assert payload["version"] == PAYLOAD_VERSION + assert payload["event"] == "finding.opened" + assert set(payload) == {"version", "event", "channel", "finding", "source", "scan"} + assert payload["finding"]["severity"] == Severity.HIGH.value + assert payload["finding"]["redacted_snippet"] == "AKIA****************" + # The property that matters most: nothing reversible leaves the deployment. + assert PLAINTEXT_SECRET not in json.dumps(payload) + # Internal triage state is not somebody else's business. + assert "notes" not in payload["finding"] + assert "assignee_id" not in payload["finding"] + + +def test_the_payload_never_carries_analyst_notes( + session: Session, + source: Source, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + channel = make_channel() + finding = make_open_finding(notes="contact bob@example.com, rotate in prod") + + payload = finding_opened(finding, source=source, scan=scan, channel=channel) + + assert "bob@example.com" not in json.dumps(payload) + + +# ── The webhook transport ───────────────────────────────────────────────────── + + +def _webhook_channel(url: str = "https://receiver.example.com/hook") -> NotificationChannel: + return NotificationChannel( + name="hook", + type=NotificationChannelType.WEBHOOK, + config={"url": url}, + ) + + +def test_the_webhook_posts_json_and_signs_it( + dispatch_settings: ApiSettings, secret_store: SecretStore +) -> None: + seen: dict[str, Any] = {} + + def handler(request: httpx2.Request) -> httpx2.Response: + seen["headers"] = dict(request.headers) + seen["body"] = request.read() + return httpx2.Response(204) + + ref = secret_store.seal("payload-signing-secret") + channel = _webhook_channel() + channel.config = {**channel.config, "secret_ref": ref} + transport = WebhookTransport( + dispatch_settings, secret_store, transport=httpx2.MockTransport(handler) + ) + + transport.send(channel, {"event": "finding.opened"}, subject="ignored") + + assert seen["headers"]["content-type"] == "application/json" + signature = seen["headers"][SIGNATURE_HEADER.lower()] + timestamp = seen["headers"][TIMESTAMP_HEADER.lower()] + assert signature == sign(seen["body"], "payload-signing-secret", timestamp) + + +def test_an_unsigned_channel_sends_no_signature_header( + dispatch_settings: ApiSettings, secret_store: SecretStore +) -> None: + seen: dict[str, Any] = {} + + def handler(request: httpx2.Request) -> httpx2.Response: + seen["headers"] = dict(request.headers) + return httpx2.Response(200) + + transport = WebhookTransport( + dispatch_settings, secret_store, transport=httpx2.MockTransport(handler) + ) + + transport.send(_webhook_channel(), {"event": "finding.opened"}, subject="ignored") + + assert SIGNATURE_HEADER.lower() not in seen["headers"] + + +@pytest.mark.parametrize( + ("status_code", "permanent"), + [(500, False), (429, False), (503, False), (401, True), (403, True), (400, True)], +) +def test_webhook_status_codes_are_classified( + dispatch_settings: ApiSettings, + secret_store: SecretStore, + status_code: int, + permanent: bool, +) -> None: + """A 5xx or a 429 is worth retrying; an auth failure is configuration.""" + transport = WebhookTransport( + dispatch_settings, + secret_store, + transport=httpx2.MockTransport(lambda request: httpx2.Response(status_code)), + ) + + with pytest.raises(DeliveryError) as raised: + transport.send(_webhook_channel(), {"event": "x"}, subject="ignored") + + assert raised.value.permanent is permanent + + +def test_the_webhook_does_not_follow_a_redirect( + dispatch_settings: ApiSettings, secret_store: SecretStore +) -> None: + """A 302 would be a way to move where findings are sent without touching the + channel, so the redirect is a failed delivery rather than a followed one.""" + visited: list[str] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + visited.append(str(request.url)) + return httpx2.Response(302, headers={"Location": "https://attacker.example.com/collect"}) + + transport = WebhookTransport( + dispatch_settings, secret_store, transport=httpx2.MockTransport(handler) + ) + + transport.send(_webhook_channel(), {"event": "x"}, subject="ignored") + + assert visited == ["https://receiver.example.com/hook"] + + +# ── The SMTP transport, against a server that speaks SMTP ───────────────────── + + +class _CapturingSmtpHandler(StreamRequestHandler): + """Enough of RFC 5321 to accept one message and hand it back.""" + + def handle(self) -> None: + received: list[str] = [] + self.wfile.write(b"220 fake.test ESMTP\r\n") + while True: + line = self.rfile.readline() + if not line: + return + command = line.decode(errors="replace").strip() + upper = command.upper() + if upper.startswith(("EHLO", "HELO")): + # No STARTTLS advertised: the test relay is a localhost stand-in. + self.wfile.write(b"250-fake.test\r\n250 SIZE 10240000\r\n") + elif upper.startswith(("MAIL FROM", "RCPT TO")): + received.append(command) + self.wfile.write(b"250 OK\r\n") + elif upper == "DATA": + self.wfile.write(b"354 End data with .\r\n") + body: list[str] = [] + while True: + data_line = self.rfile.readline().decode(errors="replace") + if data_line in {".\r\n", ".\n", ""}: + break + body.append(data_line) + received.append("".join(body)) + self.wfile.write(b"250 OK\r\n") + elif upper == "QUIT": + self.wfile.write(b"221 Bye\r\n") + break + else: + self.wfile.write(b"250 OK\r\n") + self.server.captured.append(received) # type: ignore[attr-defined] + + +class _FakeSmtpServer(ThreadingTCPServer): + allow_reuse_address = True + daemon_threads = True + + def __init__(self) -> None: + super().__init__(("127.0.0.1", 0), _CapturingSmtpHandler) + self.captured: list[list[str]] = [] + + +@pytest.fixture(name="smtp_server") +def smtp_server_fixture() -> Iterator[_FakeSmtpServer]: + server = _FakeSmtpServer() + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def test_an_email_is_sent_to_the_channel_recipients(smtp_server: _FakeSmtpServer) -> None: + """The SMTP path, end to end against a server that actually speaks SMTP.""" + settings = ApiSettings( + database_url="postgresql+psycopg://unused/unused", + master_key=SecretStr("unused"), + smtp_host="127.0.0.1", + smtp_port=smtp_server.server_address[1], + # The stand-in relay is on localhost and speaks no TLS — the one case the + # setting exists for. + smtp_starttls=False, + smtp_from="iceberg@example.com", + smtp_timeout_seconds=5.0, + ) + channel = NotificationChannel( + name="secops", + type=NotificationChannelType.EMAIL, + config={"recipients": ["secops@example.com", "oncall@example.com"]}, + ) + payload = { + "finding": { + "severity": "high", + "rule_id": "aws-access-key", + "rulepack_version": "2026.07.1", + "confidence": 0.9, + "fingerprint": "abc123", + "resource_locator": {"path": "/space/DOCS/page-1"}, + "redacted_snippet": "AKIA****************", + "url": None, + }, + "source": {"name": "confluence-prod", "type": "confluence"}, + } + + SmtpTransport(settings).send(channel, payload, subject="[IcebergSST] HIGH secret") + + conversation = smtp_server.captured[0] + assert any("secops@example.com" in line for line in conversation) + assert any("oncall@example.com" in line for line in conversation) + message = conversation[-1] + assert "[IcebergSST] HIGH secret" in message + assert "AKIA****************" in message + assert PLAINTEXT_SECRET not in message + + +def test_email_without_a_configured_relay_fails_loudly() -> None: + """A configured channel and an unconfigured deployment should say so, not + silently drop the mail.""" + settings = ApiSettings( + database_url="postgresql+psycopg://unused/unused", + master_key=SecretStr("unused"), + smtp_host=None, + ) + channel = NotificationChannel( + name="secops", + type=NotificationChannelType.EMAIL, + config={"recipients": ["secops@example.com"]}, + ) + + with pytest.raises(DeliveryError) as raised: + SmtpTransport(settings).send(channel, {}, subject="x") + + assert raised.value.permanent is True + assert "ICEBERG_SMTP_HOST" in str(raised.value) + + +def test_pending_count_reports_what_is_queued( + session: Session, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + channel = make_channel() + make_open_finding() + make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + + assert dispatch.pending_count(session) == 2 + assert dispatch.pending_count(session, channel_id=channel.id) == 2 + assert dispatch.pending_count(session, channel_id=uuid.uuid4()) == 0 + + +def test_deliveries_are_removed_with_their_finding( + session: Session, + scan: Scan, + make_open_finding: Callable[..., Finding], + make_channel: Callable[..., NotificationChannel], +) -> None: + """CASCADE, so retention pruning findings (#73) is not blocked by delivery rows.""" + make_channel() + finding = make_open_finding() + dispatch.enqueue_for_scan(session, scan) + session.commit() + + session.delete(finding) + session.commit() + + remaining = session.exec( + select(NotificationDelivery).where(col(NotificationDelivery.finding_id) == finding.id) + ).all() + assert remaining == [] diff --git a/deploy/compose/docker-compose.yml b/deploy/compose/docker-compose.yml index 2f1edb0..46d7c5c 100644 --- a/deploy/compose/docker-compose.yml +++ b/deploy/compose/docker-compose.yml @@ -85,6 +85,16 @@ services: # api-only on purpose: engines receive the threshold in their lease, the # same way they receive the pepper and the suppressions (#70). ICEBERG_CONFIDENCE_THRESHOLD: ${ICEBERG_CONFIDENCE_THRESHOLD:-0.5} + # Notification dispatch (#60) — api-only for the same reason: delivery runs + # in the maintenance loop, which needs the database (docs/notifications.md). + ICEBERG_SMTP_HOST: ${ICEBERG_SMTP_HOST:-} + ICEBERG_SMTP_PORT: ${ICEBERG_SMTP_PORT:-587} + ICEBERG_SMTP_USERNAME: ${ICEBERG_SMTP_USERNAME:-} + ICEBERG_SMTP_PASSWORD: ${ICEBERG_SMTP_PASSWORD:-} + ICEBERG_SMTP_STARTTLS: ${ICEBERG_SMTP_STARTTLS:-true} + ICEBERG_SMTP_FROM: ${ICEBERG_SMTP_FROM:-icebergsst@localhost} + ICEBERG_NOTIFICATION_MAX_ATTEMPTS: ${ICEBERG_NOTIFICATION_MAX_ATTEMPTS:-5} + ICEBERG_NOTIFICATION_RETRY_BACKOFF_SECONDS: ${ICEBERG_NOTIFICATION_RETRY_BACKOFF_SECONDS:-60} ports: - "${ICEBERG_API_PORT:-8000}:8000" depends_on: diff --git a/docs/backlog.md b/docs/backlog.md index 2af694c..88a6332 100644 --- a/docs/backlog.md +++ b/docs/backlog.md @@ -51,7 +51,9 @@ Confluence-scan-to-triaged-finding flow). M3–M4 complete the product. ## M4 — Notifications & prod deploy - **Epic: Notifications** — email/SMTP + webhook **dispatch**, new-finding events. The channel model - and its CRUD API shipped with M3. + and its CRUD API shipped with M3. Dispatch is a transactional outbox + (`notification_delivery`): reconciliation queues, the maintenance loop delivers and retries. + See [`notifications.md`](./notifications.md). - **Epic: Helm chart** — api Deploy, engine Deploy + HPA, pg/redis, secrets, ingress, values. - **Epic: Hardening** — security review, rate limiting, audit logging, data-retention policy, key + pepper rotation runbook, docs polish. diff --git a/docs/notifications.md b/docs/notifications.md new file mode 100644 index 0000000..0accdd8 --- /dev/null +++ b/docs/notifications.md @@ -0,0 +1,177 @@ +# Notifications + +How IcebergSST tells you a secret turned up. Channels are configured in the console (or through +`/notifications/channels`); this document is the delivery contract — what gets sent, when, to +whom, and what happens when it fails. + +Configuration is **admin-only** and audit-logged, because a webhook channel is a deliberate way +for finding metadata to leave the deployment (see [`security.md`](./security.md) § Notification +egress). + +## When something is announced + +A finding is announced when **the scan that just completed opened it**: + +- it was seen for the first time in that scan, or +- it had been resolved and this scan saw it again (the secret came back). + +Not every open finding on every scan. Being told weekly about the same secret trains people to +ignore the alerts, and the finding is on the console either way. + +Two things are never announced: + +- **Suppressed findings.** A suppression is an analyst saying "stop telling me about this" + (ADR 0008). Honouring that in the UI and mailing them anyway would be worse than not having + suppressions. +- **Findings from a scan that did not complete.** Announcement happens after reconciliation, and + reconciliation refuses to run for a partial or failed scan (ADR 0009 §4). + +Each channel's **event filter** then decides whether that channel cares: + +| Field | Meaning | +|---|---| +| `min_severity` | Announce this severity and above. Omitted means every severity. | +| `source_ids` | Restrict to these sources. Empty means all of them. | + +## How delivery works + +Dispatch is a **transactional outbox**, which is what makes "never lost silently" true rather +than aspirational. + +1. Reconciliation writes one `notification_delivery` row per (channel, finding) in the same + transaction that finishes the scan. If the scan commits, the intention to announce commits + with it. Nothing is sent at this point, so a hung webhook cannot delay an engine's results. +2. The maintenance loop — one replica at a time, under the same advisory lock as the scheduler — + picks up rows that are due and attempts them. +3. Success marks the row `delivered`. Failure schedules a retry with exponential backoff. After + the attempt ceiling the row becomes `failed` **and stays**, holding the error that ended it, + so "what were we never able to send?" is a query rather than a log search. + +A `(channel, finding, scan)` uniqueness constraint makes enqueueing idempotent. This matters: +the safety sweep that re-finalizes a scan stranded by a crash re-runs enqueueing, and it must +not announce the same secret on every beat. + +Retries are skipped, permanently, for failures that time cannot fix — a channel with no URL, a +relay that is not configured, an HTTP 401/403/400. Retrying those five times only delays the +operator finding out. + +### Tuning + +| Setting | Default | What it does | +|---|---|---| +| `ICEBERG_NOTIFICATION_MAX_ATTEMPTS` | `5` | Attempts before a delivery is marked `failed`. | +| `ICEBERG_NOTIFICATION_RETRY_BACKOFF_SECONDS` | `60` | First retry delay; doubles each attempt (≈16 min by the fifth). | +| `ICEBERG_NOTIFICATION_BATCH_SIZE` | `50` | Deliveries attempted per maintenance round. | +| `ICEBERG_WEBHOOK_TIMEOUT_SECONDS` | `10` | Per-attempt ceiling for a webhook POST. | + +## Webhook payload + +`POST` to the channel's URL, `Content-Type: application/json`. + +```json +{ + "version": "1", + "event": "finding.opened", + "channel": {"id": "…", "name": "security-alerts"}, + "finding": { + "id": "…", + "fingerprint": "…", + "rule_id": "aws-access-key", + "rulepack_version": "2026.07.1", + "severity": "high", + "confidence": 0.92, + "state": "open", + "redacted_snippet": "AKIA****************", + "resource_locator": {"path": "/space/DOCS/page-1"}, + "first_seen_at": "2026-07-31T06:12:44+00:00", + "url": null + }, + "source": {"id": "…", "name": "confluence-prod", "type": "confluence"}, + "scan": {"id": "…", "trigger": "scheduled"} +} +``` + +`version` is bumped when a field is **removed or changes meaning**. Adding a field does not bump +it, so receivers should ignore keys they do not recognise. + +**What is deliberately not in there:** the secret (only the snippet the engine already redacted — +ADR 0004 — and never anything reversible), and analyst triage state such as notes and assignee. +The payload is built from an explicit field list, so adding a column to `Finding` cannot silently +start exporting it. + +### Headers + +| Header | Value | +|---|---| +| `X-Iceberg-Event` | The `event` value, so a receiver can route without parsing the body. | +| `X-Iceberg-Timestamp` | Unix seconds, and part of the signed material. | +| `X-Iceberg-Signature` | `sha256=` — present only when the channel has a secret. | + +Custom headers configured on the channel are sent as-is, except that the header names carrying +credentials (`Authorization`, `Proxy-Authorization`, `Cookie`) are refused at write time — channel +config is stored as plain JSON, so a token there would be a plaintext secret at rest. + +### Verifying the signature + +The MAC covers `"." + ` with the channel secret: + +```python +import hashlib, hmac + +def verify(body: bytes, timestamp: str, signature: str, secret: str) -> bool: + expected = hmac.new(secret.encode(), f"{timestamp}.".encode() + body, hashlib.sha256) + return hmac.compare_digest(f"sha256={expected.hexdigest()}", signature) +``` + +Verify against the **raw bytes**, not a re-serialised object. Reject a timestamp outside your +tolerance (a few minutes) so a captured request cannot be replayed indefinitely, and use a +constant-time comparison. + +### Receiver expectations + +- Answer within the timeout, and answer `2xx`. Any `4xx`/`5xx` is a failed delivery. +- **Redirects are not followed.** A `302` is a failed delivery, because following one would be a + way to move where findings are sent without anyone editing the channel. +- Be idempotent on `finding.id` + `scan.id`. A retry after a response that was lost in transit + will deliver the same announcement twice; that is the correct behaviour for an at-least-once + system. + +## Email + +Plain text, one message per finding, to the channel's recipients. Plain text on purpose: an HTML +mail containing attacker-influenced content — a resource locator is a path from a scanned system — +is an injection surface in whatever client opens it, for no gain over a readable summary. + +The subject is `[IcebergSST] secret in `, so severity sorts and filters in +an inbox. + +Email needs a relay configured on the **api** role: + +| Setting | Default | Notes | +|---|---|---| +| `ICEBERG_SMTP_HOST` | *(unset)* | Unset disables email delivery entirely. | +| `ICEBERG_SMTP_PORT` | `587` | Submission. | +| `ICEBERG_SMTP_USERNAME` / `ICEBERG_SMTP_PASSWORD` | *(unset)* | Omit for an unauthenticated relay. | +| `ICEBERG_SMTP_STARTTLS` | `true` | Certificates are verified. Turn off only for a relay on localhost. | +| `ICEBERG_SMTP_FROM` | `icebergsst@localhost` | Envelope and header sender. | +| `ICEBERG_SMTP_TIMEOUT_SECONDS` | `10` | Per-attempt ceiling. | + +With no relay configured, an email channel's deliveries fail **permanently** with a message +naming `ICEBERG_SMTP_HOST` rather than retrying — a configured channel on an unconfigured +deployment should be loud, not quietly undelivered. + +## Operating + +Deliveries are rows, so the questions have answers: + +```sql +-- what never went out, and why +SELECT channel_id, finding_id, attempts, last_error +FROM notification_delivery WHERE status = 'failed'; + +-- what is queued right now +SELECT count(*) FROM notification_delivery WHERE status = 'pending'; +``` + +Logs carry the same events: `notifications_enqueued`, `notification_sent`, +`notification_retry_scheduled`, `notification_failed`. diff --git a/docs/security.md b/docs/security.md index 08914ef..9052f1f 100644 --- a/docs/security.md +++ b/docs/security.md @@ -77,6 +77,22 @@ the header names that carry authentication (`Authorization`, `Proxy-Authorizatio text in `notification_channel.config` would be precisely the finding this product exists to report in other people's systems. +At **dispatch** time (see [`notifications.md`](./notifications.md) for the full contract) the same +boundary applies to what goes over the wire: + +- The payload is built from an **explicit field list**, never by serialising the ORM row, so a new + column on `Finding` cannot silently start being exported. It carries the redacted snippet + (ADR 0004) and the peppered hash — nothing reversible — and no analyst notes or assignee. +- Requests are **signed** (`X-Iceberg-Signature`, HMAC-SHA256 over `timestamp.body`) when the + channel has a secret, so a receiver can distinguish a real announcement from anything else that + can reach its URL. The timestamp is inside the MAC, so replay has a bounded window. +- **Redirects are not followed.** A `302` would relocate where findings are sent without anyone + editing the channel, so it is a failed delivery. +- Response bodies are never read into an error message or a log line: they are somebody else's + data. +- Email is sent as **plain text**. Resource locators come from scanned systems, so an HTML message + would carry attacker-influenced content into whatever client opens it. + ## The browser surface (M3) The console renders attacker-influenced strings — Confluence page titles, resource paths, redacted snippets — which makes it the highest-value stored-XSS target in the deployment. The mitigations, diff --git a/packages/core/src/iceberg_core/config.py b/packages/core/src/iceberg_core/config.py index dbb22d8..0d99512 100644 --- a/packages/core/src/iceberg_core/config.py +++ b/packages/core/src/iceberg_core/config.py @@ -105,6 +105,36 @@ class ApiSettings(SecretStoreSettings): bootstrap_admin_subject: str | None = None bootstrap_admin_email: str | None = None + # ─── Notification dispatch (#60) ────────────────────────────────────────── + # Channel *configuration* lives in the database, because an analyst edits it. + # These are deployment facts — which relay to use, how hard to try — so they + # are configuration, and they are api-role only: engines never dispatch. + #: Unset disables email delivery. Email channels then fail their deliveries + #: with a clear error rather than appearing to work. + smtp_host: str | None = None + smtp_port: int = Field(default=587, ge=1, le=65535) + smtp_username: str | None = None + smtp_password: SecretStr | None = None + #: STARTTLS on a submission port. Turn off only for a relay on localhost. + smtp_starttls: bool = True + smtp_from: str = "icebergsst@localhost" + #: A dispatch must never be the reason an API worker is tied up. + smtp_timeout_seconds: float = Field(default=10.0, gt=0, le=120) + + #: Per-attempt ceiling for a webhook POST. Deliberately short: the receiver is + #: operator-supplied and may be a black hole. + webhook_timeout_seconds: float = Field(default=10.0, gt=0, le=120) + + #: Attempts before a delivery is marked `failed`. The row is kept either way — + #: giving up is recorded, not silent. + notification_max_attempts: int = Field(default=5, ge=1, le=20) + #: First retry delay; doubles each attempt, so the default 60s reaches roughly + #: 16 minutes by the fifth. Long enough to ride out a receiver's restart. + notification_retry_backoff_seconds: int = Field(default=60, ge=1) + #: Deliveries attempted per maintenance round. Bounds how long one round can + #: take when a channel is timing out. + notification_batch_size: int = Field(default=50, ge=1, le=1000) + @field_validator("database_url", "redis_url") @classmethod def _require_url_scheme(cls, value: str) -> str: diff --git a/packages/core/src/iceberg_core/enums.py b/packages/core/src/iceberg_core/enums.py index caa2f9a..e999a8b 100644 --- a/packages/core/src/iceberg_core/enums.py +++ b/packages/core/src/iceberg_core/enums.py @@ -102,3 +102,15 @@ class EngineStatus(StrEnum): class NotificationChannelType(StrEnum): EMAIL = "email" WEBHOOK = "webhook" + + +class NotificationDeliveryStatus(StrEnum): + """Where one announcement to one channel has got to (#60). + + ``failed`` is terminal and means *give up*, not *lost*: the row stays, with + the error that ended it, so an operator can see what was never delivered. + """ + + PENDING = "pending" + DELIVERED = "delivered" + FAILED = "failed" diff --git a/packages/core/src/iceberg_core/models/__init__.py b/packages/core/src/iceberg_core/models/__init__.py index 19f1f3c..235f7d0 100644 --- a/packages/core/src/iceberg_core/models/__init__.py +++ b/packages/core/src/iceberg_core/models/__init__.py @@ -51,7 +51,7 @@ ) from iceberg_core.models.findings import Finding, FindingEvent, Suppression from iceberg_core.models.identity import User -from iceberg_core.models.notifications import NotificationChannel +from iceberg_core.models.notifications import NotificationChannel, NotificationDelivery from iceberg_core.models.scans import Engine, Scan, ScanTask from iceberg_core.models.sources import Schedule, Source @@ -88,6 +88,7 @@ "FindingEvent", "IcebergModel", "NotificationChannel", + "NotificationDelivery", "Scan", "ScanTask", "Schedule", diff --git a/packages/core/src/iceberg_core/models/notifications.py b/packages/core/src/iceberg_core/models/notifications.py index b6bf62f..217eea9 100644 --- a/packages/core/src/iceberg_core/models/notifications.py +++ b/packages/core/src/iceberg_core/models/notifications.py @@ -1,11 +1,19 @@ """Notification channels — a deliberate egress path for finding metadata.""" +import uuid +from datetime import datetime from typing import Any +from sqlalchemy import Index, UniqueConstraint from sqlmodel import Field -from iceberg_core.enums import NotificationChannelType -from iceberg_core.models.base import TimestampedModel, enum_type, json_type +from iceberg_core.enums import NotificationChannelType, NotificationDeliveryStatus +from iceberg_core.models.base import ( + TimestampedModel, + enum_type, + json_type, + utc_timestamp_type, +) class NotificationChannel(TimestampedModel, table=True): @@ -31,3 +39,57 @@ class NotificationChannel(TimestampedModel, table=True): event_filter: dict[str, Any] = Field(default_factory=dict, sa_type=json_type()) enabled: bool = Field(default=True) + + +class NotificationDelivery(TimestampedModel, table=True): + """One announcement, to one channel, about one finding (#60). + + A transactional outbox. The row is written in the same transaction that opens + the finding, so "the finding exists" and "somebody will be told" commit or + fail together — the alternative is sending inside the request and losing the + alert to a webhook timeout, which is the "never lost silently" half of the + acceptance criteria. + + Delivery is then somebody else's problem: the maintenance loop picks up rows + that are due, attempts them, and either marks them delivered or schedules a + retry. Because the attempt happens outside the ingest transaction, a slow SMTP + server delays an alert rather than a scan. + """ + + __tablename__ = "notification_delivery" + __table_args__ = ( + # At most one announcement per channel per finding per scan. This is what + # makes the whole path idempotent: the sweep that re-finalizes a stalled + # scan re-runs enqueueing, and a retried round must not tell an operator + # twice about one secret. + UniqueConstraint( + "channel_id", + "finding_id", + "scan_id", + name="uq_notification_delivery_channel_finding_scan", + ), + # The delivery loop's query: what is pending and due. + Index("ix_notification_delivery_status_next_attempt_at", "status", "next_attempt_at"), + ) + + channel_id: uuid.UUID = Field(foreign_key="notification_channel.id", ondelete="CASCADE") + finding_id: uuid.UUID = Field(foreign_key="finding.id", ondelete="CASCADE") + #: The scan that opened the finding. CASCADE rather than the RESTRICT the + #: finding uses for its own scan references: this row is a delivery record, + #: not part of the finding's history, so retention pruning old scans (#73) + #: should take it rather than be blocked by it. + scan_id: uuid.UUID = Field(foreign_key="scan.id", ondelete="CASCADE") + + status: NotificationDeliveryStatus = Field( + default=NotificationDeliveryStatus.PENDING, + sa_type=enum_type(NotificationDeliveryStatus, name="notification_delivery_status"), + ) + attempts: int = Field(default=0, ge=0) + #: When this row becomes eligible. Set to the enqueue time, then pushed out by + #: exponential backoff after each failure. + next_attempt_at: datetime = Field(sa_type=utc_timestamp_type()) + delivered_at: datetime | None = Field(default=None, sa_type=utc_timestamp_type()) + + #: Why the last attempt failed. Truncated, and never carries a response body: + #: a webhook's reply is somebody else's data and could echo anything back. + last_error: str | None = Field(default=None, max_length=500)