diff --git a/Dockerfile b/Dockerfile index 85f33a3..6b48ab1 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,17 +1,28 @@ -# Deployed runtime image (P4). Builds the API + Twilio voice server. -FROM python:3.12-slim +# Pinned multi-arch Docker Official Image index for 3.12.10 slim-bookworm. +FROM python:3.12.10-slim-bookworm@sha256:fd95fa221297a88e1cf49c55ec1828edd7c5a428187e67b5d1805692d11588db WORKDIR /app COPY pyproject.toml ./ COPY src ./src COPY data ./data -RUN pip install --no-cache-dir -e ".[serve]" +COPY config ./config +COPY scripts ./scripts +RUN pip install --no-cache-dir ".[serve,trace,voice,llm]" \ + && groupadd --gid 10001 sku \ + && useradd --uid 10001 --gid 10001 --no-create-home --shell /usr/sbin/nologin sku \ + && mkdir -p /var/lib/sku-resolver \ + && chown -R 10001:10001 /var/lib/sku-resolver # Runtime config via env (see src/runtime/config.py). Provide a real # SKU_SESSION_SECRET and SKU_WEBHOOK_SECRET in production; set # SKU_LLM_PROVIDER + the matching *_API_KEY to enable the LLM seams; set # TWILIO_* / ASSEMBLYAI_API_KEY for live voice. -ENV PORT=8000 +ENV PORT=8000 \ + PYTHONPATH=/app/src \ + PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 +VOLUME ["/var/lib/sku-resolver"] EXPOSE 8000 +USER 10001:10001 -CMD ["sh", "-c", "uvicorn runtime.app:create_app --factory --host 0.0.0.0 --port ${PORT}"] +CMD ["uvicorn", "runtime.app:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000", "--no-access-log"] diff --git a/data/m5_legacy_schema_lock.json b/data/m5_legacy_schema_lock.json index 9ccfe35..0fc421f 100644 --- a/data/m5_legacy_schema_lock.json +++ b/data/m5_legacy_schema_lock.json @@ -26,11 +26,13 @@ } }, "quote_store": { - "schema_sha256": "b78342c2f525a361338c7fb82102650af72229aeb2bff51e17c2ced7ec1fd4a8", + "schema_sha256": "2131cb30d2ce8d4b717457d8260408e0378059d894bb8f70e57d966824594244", "tables": [ "artifact_guard_events", "artifact_jobs", "capability_access_events", + "data_legal_holds", + "deletion_audit", "delivery_jobs", "issuance_events", "provider_callback_events", @@ -44,12 +46,15 @@ "quote_schema", "quote_seq", "quotes", + "retention_context", "worker_heartbeats" ], "row_counts": { "artifact_guard_events": 0, "artifact_jobs": 0, "capability_access_events": 0, + "data_legal_holds": 0, + "deletion_audit": 0, "delivery_jobs": 0, "issuance_events": 0, "provider_callback_events": 0, @@ -63,6 +68,7 @@ "quote_schema": 1, "quote_seq": 1, "quotes": 1, + "retention_context": 1, "worker_heartbeats": 0 }, "known_gaps": [ @@ -72,5 +78,5 @@ ] } }, - "combined_sha256": "151432b988d1ace9cd6cd2af51039246f68d33f59a8977032cc678b808f592fb" + "combined_sha256": "a5b47bdf7a6faf6a95598d204def3997ad3ea63da82c1ab6e8f396fc93c6bc8b" } diff --git a/deploy/processes.json b/deploy/processes.json new file mode 100644 index 0000000..6080ee4 --- /dev/null +++ b/deploy/processes.json @@ -0,0 +1,15 @@ +{ + "schema_version": 1, + "api": [ + "uvicorn", "runtime.app:create_app", "--factory", "--host", "0.0.0.0", + "--port", "8000", "--no-access-log" + ], + "worker": [ + "python", "scripts/quote_worker.py", "run", "--max-poll-seconds", "5", + "--lease-seconds", "30" + ], + "pre_stop": [ + "POST", "/internal/drain" + ], + "state_mount": "/var/lib/sku-resolver" +} diff --git a/docs/DECISION_LOG.md b/docs/DECISION_LOG.md index 98dbf7f..5dd22b3 100644 --- a/docs/DECISION_LOG.md +++ b/docs/DECISION_LOG.md @@ -1610,3 +1610,57 @@ recipient allowlists, exact attachments/link-only SMS, signed callbacks, heartbeats, bounded polling, graceful drain, and operator inspect/retry/ quarantine/revoke controls. **Status:** implemented and verified offline; credentialed staging sends and callbacks remain under P04. + +## 2026-07-11 — D22: production is one encrypted transactional state volume and two processes + +**Decision.** M5 production is one API process plus one worker process sharing a +single local SQLite database on one encrypted durable volume. Quote, outbox, +capability, provider, authoritative conversation cursor, and call-completion +state therefore participate in one backup and one tenant-purge boundary. Network +filesystems, in-memory stores, horizontal API/worker replicas, split quote/call +databases, simulated providers, fixture sources, example origins, insecure +secrets, stale schemas, and a missing worker heartbeat all fail startup. + +The container is digest-pinned, non-root, has explicit API and worker commands, +disables access logs that would expose capability paths, and mounts all mutable +state outside the image. The public health endpoint proves liveness only. The +private readiness endpoint is token-gated and returns a generic state; detailed +failure reasons stay local. A private drain endpoint rejects new work before a +restart or migration. + +**Why.** SQLite is appropriate for the locked single-host topology only when its +transaction and recovery boundary is not accidentally split across processes or +volumes. Readiness is a safety gate, not a public diagnostic oracle. + +**Outcome.** WAL, foreign keys, busy timeout, full synchronization, integrity +checks, schema writer locking, disk headroom, source freshness, worker heartbeat, +provider, backup, encryption, retention, and PostHog/OTLP checks are executable. +**Status:** implementation locked; production remains blocked on policy and +credential evidence. + +## 2026-07-11 — D23: encrypted restore proof and PII-free PostHog are operational controls, not truth stores + +**Decision.** Online SQLite snapshots are encrypted with AES-256-GCM under a +versioned key ID. The plaintext envelope contains table names, counts, and keyed +aggregate digests—not quote, account, caller, provider, or capability +identities. A backup is considered healthy only after a clean-host restore +matches the encrypted database hash, integrity check, schemas, artifact set, +conversation cursor set, and all transactional rows. Key rotation keeps old keys +until their backup horizon expires. + +Tenant deletion is a single-database transaction. An active legal hold blocks +it; an approved purge temporarily opens the immutable-row deletion gate, removes +tenant quote/conversation/provider/capability state, closes the gate, and retains +only a keyed tenant pseudonym plus deletion counts and approved actor/reason. + +OpenTelemetry remains the trace contract. PostHog is the sole production event/ +trace destination and receives only allowlisted scalar properties under a shared +pseudonymous correlation ID. Transcript text, account/contact values, quote +numbers/tokens, documents, and secrets are not exported. Event export is bounded, +asynchronous, and fail-open, so telemetry cannot change a spoken or issued result. +Phoenix remains an optional W7 local comparison, not a production dependency. + +**Outcome.** Backup/restore and retention CLIs, legal-hold tests, generic +readiness, PII-free journey events, worker spans, alert thresholds, and the M5 +operations runbook are implemented. **Status:** code verified offline; P06/P07 +policy, clean-host container evidence, and live sink receipts remain external. diff --git a/docs/M5_BUILD_JOURNAL.md b/docs/M5_BUILD_JOURNAL.md index f78954c..de0f306 100644 --- a/docs/M5_BUILD_JOURNAL.md +++ b/docs/M5_BUILD_JOURNAL.md @@ -774,3 +774,51 @@ simulation refusal tests pass. Twilio/SendGrid credentials, staging recipient allowlists, one allowlisted email and SMS, provider IDs, signed delivered/undelivered callbacks, and downloaded artifact hash receipts. No production or Vercel deployment was attempted. + +## 2026-07-11 — Session 12: W6 production security, recovery, and observability controls + +Added a fail-closed production validator and private readiness surface. It rejects +every development/example secret or origin, fixture/synthetic/ephemeral source, +split or stale database, stale inventory/price source, missing live provider, +simulated delivery, stale worker heartbeat, stale restore receipt, low disk, +inconsistent ElevenLabs/application retention, missing encryption/key IDs, and +missing PostHog/OTLP configuration. Existing P01–P16 policy blockers remain +independent, so operationally healthy fixtures cannot turn production green. + +Hardened SQLite with WAL, foreign keys, five-second busy timeout, full sync, +integrity checks, and a writer lock across migrations. Quote schema v5 adds legal +holds, deletion audit, and a transaction-only immutable-row purge gate. The +production topology requires quote and conversation schemas in one database so +tenant offboarding removes both in one transaction. + +Added AES-256-GCM online backup, key-ID rotation support, clean-host restore and +semantic verification, plus backup and retention CLIs. Backup envelopes expose no +quote/caller/provider identities. Restore proves exact encrypted bytes, schemas, +row sets, guarded artifact hashes, and conversation cursors before writing a +fresh readiness marker. + +Added process drain, source-aware application rate limits, generic `/healthz` +and private `/readyz`, a pinned non-root image, explicit API/worker process +commands, durable state mount, and disabled capability-path access logs. The M5 +runbook covers parity, queue/provider, ambiguous lease, contact, invalid-reference, +disk, backup/restore, key rotation, active-call restart, tenant offboarding, and +rollback incidents. + +OpenTelemetry remains the tracing contract. A bounded asynchronous PostHog sink +emits only allowlisted pseudonymous journey events, and worker stages emit duration +and outcome spans. Alert evaluation pages immediately on parity or queue-SLO +breach and covers provider outage, stuck lease, invalid-reference surge, disk, +worker, and backup/restore signals. + +**Verification:** 932 tests collected and full pytest exits zero; ruff is clean; +configured mypy reports no issues across 89 source files. Backup tamper/restore, +legal hold/purge, SQLite posture, policy-vs-operational readiness, private drain, +rate limiting, PostHog redaction/nonblocking behavior, alert thresholds, and +static non-root image/process contracts pass. + +**External W6 proof still required:** approved P06 retention/encryption/key/ +ElevenLabs deletion policy; approved P07 destination/RPO/RTO/owner/drill policy; +Docker-capable build and non-root restart receipt; encrypted-volume attestation; +clean-host measured backup/restore drill; and live PostHog OTLP/event receipts. +Docker is unavailable on this host. No production or Vercel deployment was +attempted. diff --git a/docs/M5_OPERATIONS_RUNBOOK.md b/docs/M5_OPERATIONS_RUNBOOK.md new file mode 100644 index 0000000..e609f74 --- /dev/null +++ b/docs/M5_OPERATIONS_RUNBOOK.md @@ -0,0 +1,79 @@ +# M5 production operations runbook + +This runbook is executable only after P02–P15 are locked for the tenant. Never +paste a secret, destination, transcript, account number, quote capability token, +or document content into a ticket, alert, terminal transcript, or PostHog. + +## Start, drain, and restart + +- Start the worker command from `deploy/processes.json` first and wait for a + fresh durable heartbeat. +- Start the API command. `/healthz` proves only process liveness. Query `/readyz` + from the private network with `X-Internal-Ready-Token`; do not expose it at the + public edge. +- Before restart or migration, call `POST /internal/drain` with the internal + token. Wait for `active=0`, stop the API, take an encrypted backup, then restart. + If an active call cannot drain, leave the version in service and escalate; do + not move its conversation cursor between processes. + +## Quote parity incident + +Page immediately. Confirm the artifact job is quarantined and no delivery job +crossed the provider boundary. Preserve only the correlation ID, renderer/guard +versions, and hashes. Reproduce with the frozen `QuoteDocument`; never deliver a +manually edited PDF. Fix, run the parity battery, then use the operator retry. + +## Queue SLO, stuck lease, or provider outage + +- Inspect with `python scripts/quote_worker.py inspect`. +- For an expired artifact or pre-send lease known not accepted, use the scoped + `retry` command. For `sending`, `manual_review`, or ambiguous network loss, + inspect the provider by stable delivery key before any retry. +- Quarantine poison artifacts so later quotes continue. During provider outage, + preserve retry backoff and recipient allowlists; never switch destinations. + +## Missing/changed contact + +`needs_reauthorization` is terminal for the frozen assent. Correct the contact of +record, reverify the caller, and obtain a new delivery assent (and quote revision +when the frozen contract changes). Operators may not retarget a job. + +## Invalid-reference or link flood + +Keep refusals neutral. Confirm application and edge rate limits, inspect only +pseudonymous source/tenant counters, rotate or revoke affected capability links, +and avoid confirming whether a quote exists. + +## Disk pressure + +Drain API and worker before the SQLite volume reaches the configured floor. +Preserve WAL/SHM files with the database. Expand the encrypted local volume or +restore to a larger clean volume; never move live SQLite onto a network filesystem. + +## Backup, restore, and key rotation + +- Create: `python scripts/m5_backup.py create BACKUP_PATH`. +- Restore drill: `python scripts/m5_backup.py restore-verify BACKUP_PATH EMPTY_DIR + --status-path STATUS_PATH`. +- A backup is healthy only after clean-host restore verifies integrity, schemas, + rows, quote artifacts/hashes, job/provider state, capabilities, conversation + cursors, and completion events within the approved P07 RPO/RTO. +- Add the new key ID and bytes, make it active, create and restore-verify a new + backup, then retire the old key only after its retention horizon. Never rewrite + an old envelope under a new key ID. + +## Tenant offboarding, legal hold, and deletion + +P06 supplies the approved actor, reason, retention horizon, and ElevenLabs +audio/transcript deletion receipt. An active legal hold blocks deletion. With no +hold, run the transactional tenant purge against the single state database; keep +only its keyed tenant pseudonym and deletion counts. Delete provider/platform data +under their approved APIs and attach only receipt IDs. + +## Rollback + +Rollback application and worker together to a schema-compatible image. Never +downgrade across an unknown schema. For hosted voice, use the W4 pinned +compensating rollback and verify the existing phone/SIP assignment. Re-run +`/readyz`, a synthetic inbound call, a guarded quote, and a clean restore drill. + diff --git a/scripts/m5_backup.py b/scripts/m5_backup.py new file mode 100644 index 0000000..cca681a --- /dev/null +++ b/scripts/m5_backup.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Encrypted online backup and clean-host restore verification CLI.""" +from __future__ import annotations + +import argparse +import base64 +import json +import os +import sys +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO / 'src')) + +from runtime.backup import create_backup, restore_and_verify # noqa: E402 + + +def _keys() -> dict[str, bytes]: + try: + raw = json.loads(os.environ.get('SKU_BACKUP_KEYS_JSON', '{}')) + keys = {str(name): base64.b64decode(value, validate=True) + for name, value in raw.items()} + except Exception as exc: + raise SystemExit('SKU_BACKUP_KEYS_JSON is invalid') from exc + if not keys or any(len(value) != 32 for value in keys.values()): + raise SystemExit('at least one 32-byte backup key is required') + return keys + + +def main() -> int: + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest='command', required=True) + create = sub.add_parser('create') + create.add_argument('output') + restore = sub.add_parser('restore-verify') + restore.add_argument('backup') + restore.add_argument('destination') + restore.add_argument('--status-path', required=True) + args = parser.parse_args() + keys = _keys() + if args.command == 'create': + active = os.environ.get('SKU_BACKUP_ACTIVE_KEY_ID', '') + if active not in keys: + raise SystemExit('SKU_BACKUP_ACTIVE_KEY_ID is unavailable') + quote = os.environ.get('SKU_QUOTE_STORE_DB', '') + conversation = os.environ.get('SKU_CONVERSATION_STORE_DB', '') + if not quote or not conversation: + raise SystemExit('durable database paths are required') + sources = {'state': quote} + if Path(quote).resolve() != Path(conversation).resolve(): + sources['conversations'] = conversation + receipt = create_backup( + sources=sources, output=args.output, key=keys[active], key_id=active) + else: + receipt = restore_and_verify( + backup=args.backup, destination=args.destination, keys=keys, + status_path=args.status_path) + print(json.dumps(receipt, sort_keys=True)) + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/scripts/m5_retention.py b/scripts/m5_retention.py new file mode 100644 index 0000000..5f87463 --- /dev/null +++ b/scripts/m5_retention.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 +"""Legal hold and explicit transactional tenant purge controls.""" +from __future__ import annotations + +import argparse +import base64 +import json +import os +import sys +import time +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO / 'src')) + +from quoting.store import SqliteQuoteStore # noqa: E402 + + +def main() -> int: + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest='command', required=True) + hold = sub.add_parser('hold') + hold.add_argument('tenant_id') + hold.add_argument('hold_id') + hold.add_argument('reason') + release = sub.add_parser('release-hold') + release.add_argument('tenant_id') + release.add_argument('hold_id') + purge = sub.add_parser('purge-tenant') + purge.add_argument('tenant_id') + purge.add_argument('actor') + purge.add_argument('reason') + purge.add_argument('--confirm-tenant', required=True) + args = parser.parse_args() + path = os.environ.get('SKU_QUOTE_STORE_DB', '') + conversation = os.environ.get('SKU_CONVERSATION_STORE_DB', '') + if not path or Path(path).resolve() != Path(conversation).resolve(): + raise SystemExit('single transactional state database is required') + store = SqliteQuoteStore(path) + now = time.time() + if args.command == 'hold': + store.place_legal_hold( + tenant_id=args.tenant_id, hold_id=args.hold_id, + reason=args.reason, now=now) + result = {'changed': True} + elif args.command == 'release-hold': + result = {'changed': store.release_legal_hold( + tenant_id=args.tenant_id, hold_id=args.hold_id, now=now)} + else: + if args.confirm_tenant != args.tenant_id: + raise SystemExit('tenant confirmation does not match') + try: + key = base64.b64decode( + os.environ.get('SKU_AUDIT_PSEUDONYM_KEY', ''), validate=True) + except Exception as exc: + raise SystemExit('SKU_AUDIT_PSEUDONYM_KEY is invalid') from exc + counts = store.purge_tenant( + tenant_id=args.tenant_id, actor=args.actor, reason=args.reason, + audit_key=key, now=now) + result = {'changed': True, 'deleted_counts': counts} + print(json.dumps(result, sort_keys=True)) + return 0 if result['changed'] else 2 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/src/gateway/conversation_store.py b/src/gateway/conversation_store.py index 10345a3..6f55d27 100644 --- a/src/gateway/conversation_store.py +++ b/src/gateway/conversation_store.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any, Iterable +from runtime.sqlite_runtime import configure_sqlite + SCHEMA_VERSION = 2 _W1_TABLES = frozenset({ 'conversation_schema', 'conversations', 'processed_turns', @@ -20,6 +22,7 @@ 'artifact_jobs', 'delivery_jobs', 'quote_artifacts', 'quote_capabilities', 'artifact_guard_events', 'quote_schema', 'capability_access_events', 'provider_callback_events', 'worker_heartbeats', + 'data_legal_holds', 'deletion_audit', 'retention_context', }) @@ -77,7 +80,7 @@ def __init__(self, path: str | Path): self.path, isolation_level=None, check_same_thread=False) self._lock = threading.RLock() self._db.row_factory = sqlite3.Row - self._db.execute('PRAGMA foreign_keys = ON') + configure_sqlite(self._db) if self.path == ':memory:': _apply_schema(self._db) self._verify_version() diff --git a/src/gateway/db_adapters.py b/src/gateway/db_adapters.py index b86d9b9..832ccc6 100644 --- a/src/gateway/db_adapters.py +++ b/src/gateway/db_adapters.py @@ -20,13 +20,16 @@ from gateway.models import Account from gateway.pricebook import PriceSnapshot, to_cents +from runtime.sqlite_runtime import configure_sqlite def _connect(source) -> sqlite3.Connection: if isinstance(source, sqlite3.Connection): + configure_sqlite(source) return source conn = sqlite3.connect(str(source), check_same_thread=False) conn.row_factory = sqlite3.Row + configure_sqlite(conn) return conn diff --git a/src/observability/alerts.py b/src/observability/alerts.py index 952fdb5..b4eed90 100644 --- a/src/observability/alerts.py +++ b/src/observability/alerts.py @@ -11,6 +11,13 @@ from pathlib import Path from typing import Any +from observability.telemetry import scrub_pii + +_EXTRA_KEYS = frozenset({ + 'correlation_id', 'stage', 'outcome', 'channel', 'attempt', 'duration_ms', + 'queue_depth', 'oldest_age_s', 'free_bytes', 'worker_age_s', 'backup_age_s', +}) + @dataclass class AlertRouter: @@ -29,8 +36,18 @@ def route(self, *, severity: str, title: str, summary: str, now_iso: str, if dedup_key in self._seen_keys: return False self._seen_keys.add(dedup_key) - row = {'ts': now_iso, 'severity': severity, 'title': title, - 'summary': summary, **(extra or {})} + if severity not in ('info', 'warning', 'critical'): + raise ValueError('invalid alert severity') + supplied = extra or {} + if set(supplied) - _EXTRA_KEYS: + raise ValueError('alert contains non-allowlisted metadata') + safe_extra = { + key: (scrub_pii(value) if isinstance(value, str) else value) + for key, value in supplied.items() + if isinstance(value, (str, int, float, bool)) or value is None} + row = {'ts': now_iso, 'severity': severity, + 'title': scrub_pii(title)[:120], + 'summary': scrub_pii(summary)[:500], **safe_extra} self.log_path.parent.mkdir(parents=True, exist_ok=True) with self.log_path.open('a') as f: f.write(json.dumps(row) + '\n') diff --git a/src/observability/journey.py b/src/observability/journey.py new file mode 100644 index 0000000..d9ef651 --- /dev/null +++ b/src/observability/journey.py @@ -0,0 +1,128 @@ +"""PII-free PostHog journey events with a bounded non-blocking exporter.""" +from __future__ import annotations + +import json +import queue +import threading +import urllib.request +from dataclasses import dataclass +from typing import Mapping, Protocol + +from observability.telemetry import pseudonymize + +EVENTS = frozenset({ + 'call_connected', 'clarification', 'readback', 'assent', 'hangup', + 'provider_accepted', 'artifact_download', 'job_retry', 'parity_failure', + 'contact_failure', 'provider_failure', 'rate_limited', 'terminal_callback', +}) +PROPERTIES = frozenset({ + 'stage', 'outcome', 'channel', 'duration_ms', 'queue_depth', 'attempt', + 'interrupted', 'format', 'status', 'trace_id', 'eval_run_id', 'report_run_id', +}) + + +class CaptureTransport(Protocol): + def __call__(self, url: str, body: bytes, timeout: float) -> None: ... + + +def _transport(url: str, body: bytes, timeout: float) -> None: + request = urllib.request.Request( + url, data=body, headers={'Content-Type': 'application/json'}, method='POST') + with urllib.request.urlopen(request, timeout=timeout) as response: + response.read(1024) + + +def correlation_id(raw: str) -> str: + return pseudonymize(raw, namespace='m5-correlation') + + +@dataclass(frozen=True) +class JourneyEvent: + event: str + distinct_id: str + properties: dict[str, object] + + +class EventSink(Protocol): + def emit(self, event: JourneyEvent) -> bool: ... + + +class MemoryEventSink: + def __init__(self) -> None: + self.events: list[JourneyEvent] = [] + + def emit(self, event: JourneyEvent) -> bool: + self.events.append(event) + return True + + +class NoopEventSink: + def emit(self, event: JourneyEvent) -> bool: + return True + + +class PostHogEventSink: + def __init__(self, *, project_key: str, host: str, + transport: CaptureTransport = _transport, + max_queue: int = 1000) -> None: + if not project_key or not host.startswith('https://'): + raise ValueError('PostHog project key and HTTPS host are required') + self.project_key = project_key + self.url = host.rstrip('/') + '/capture/' + self.transport = transport + self._queue: queue.Queue[JourneyEvent | None] = queue.Queue(max_queue) + self._thread = threading.Thread( + target=self._run, name='posthog-events', daemon=True) + self._thread.start() + + def emit(self, event: JourneyEvent) -> bool: + try: + self._queue.put_nowait(event) + return True + except queue.Full: + return False + + def _run(self) -> None: + while True: + event = self._queue.get() + try: + if event is None: + return + body = json.dumps({ + 'api_key': self.project_key, 'event': event.event, + 'distinct_id': event.distinct_id, + 'properties': event.properties, + }, separators=(',', ':')).encode() + try: + self.transport(self.url, body, 3.0) + except Exception: + pass + finally: + self._queue.task_done() + + def flush(self) -> None: + self._queue.join() + + def close(self) -> None: + self._queue.put(None) + self._thread.join(timeout=5) + + +class JourneyEmitter: + def __init__(self, sink: EventSink): + self.sink = sink + + def emit(self, event: str, *, correlation_source: str, + properties: Mapping[str, object] | None = None) -> bool: + if event not in EVENTS: + raise ValueError('unknown journey event') + supplied = dict(properties or {}) + if set(supplied) - PROPERTIES: + raise ValueError('journey event contains non-allowlisted properties') + safe: dict[str, object] = { + key: value for key, value in supplied.items() + if isinstance(value, (str, int, float, bool)) or value is None} + if len(safe) != len(supplied): + raise ValueError('journey properties must be scalar') + return self.sink.emit(JourneyEvent( + event, correlation_id(correlation_source), safe)) diff --git a/src/observability/m5_alerts.py b/src/observability/m5_alerts.py new file mode 100644 index 0000000..fcf4784 --- /dev/null +++ b/src/observability/m5_alerts.py @@ -0,0 +1,65 @@ +"""Pure M5 alert thresholds; routing remains an injected side effect.""" +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class M5Signals: + parity_failures: int = 0 + queue_oldest_age_s: float = 0 + stuck_leases: int = 0 + provider_failure_rate: float = 0 + invalid_reference_rate: float = 0 + free_bytes: int = 2 ** 40 + worker_age_s: float = 0 + backup_age_s: float = 0 + restore_failed: bool = False + + +@dataclass(frozen=True) +class Alert: + key: str + severity: str + title: str + summary: str + + +def evaluate(signals: M5Signals, *, queue_slo_s: float = 60, + min_free_bytes: int = 1024 ** 3, + max_worker_age_s: float = 60, + max_backup_age_s: float = 86_400) -> tuple[Alert, ...]: + alerts = [] + if signals.parity_failures: + alerts.append(Alert( + 'parity', 'critical', 'Quote parity incident', + 'Document parity failed; delivery is held.')) + if signals.queue_oldest_age_s > queue_slo_s: + alerts.append(Alert( + 'queue_slo', 'critical', 'Quote delivery SLO exceeded', + 'Oldest eligible delivery work exceeds its selected threshold.')) + if signals.stuck_leases: + alerts.append(Alert( + 'stuck_lease', 'warning', 'Delivery lease requires review', + 'One or more expired send-boundary leases require manual review.')) + if signals.provider_failure_rate >= 0.25: + alerts.append(Alert( + 'provider_outage', 'critical', 'Delivery provider degradation', + 'Recent provider failures exceed the selected outage threshold.')) + if signals.invalid_reference_rate >= 0.25: + alerts.append(Alert( + 'invalid_reference', 'warning', 'Invalid quote reference surge', + 'Neutral quote-reference refusals exceed the attack threshold.')) + if signals.free_bytes < min_free_bytes: + alerts.append(Alert( + 'disk_pressure', 'critical', 'State volume disk pressure', + 'Free durable volume space is below the configured floor.')) + if signals.worker_age_s > max_worker_age_s: + alerts.append(Alert( + 'worker_stale', 'critical', 'Delivery worker heartbeat stale', + 'The durable delivery worker heartbeat is stale.')) + if signals.backup_age_s > max_backup_age_s or signals.restore_failed: + alerts.append(Alert( + 'backup_restore', 'critical', 'Backup or restore verification failed', + 'Verified backup age or the latest restore drill is outside policy.')) + return tuple(alerts) diff --git a/src/observability/telemetry.py b/src/observability/telemetry.py index 38c73a6..7841986 100644 --- a/src/observability/telemetry.py +++ b/src/observability/telemetry.py @@ -23,10 +23,13 @@ import hashlib import hmac +import json import os import re import secrets -from typing import Any +import time +from functools import wraps +from typing import Any, Callable, ParamSpec, TypeVar _ENV_ON = 'SKU_OBS_TRACING' _ENV_CONTENT = 'SKU_OBS_TRACE_CONTENT' @@ -103,6 +106,7 @@ def start_span(self, *a, **k): 'llm.model_name', 'llm.provider', 'llm.cost.total', 'llm.latency_s', 'llm.token_count.prompt', 'llm.token_count.completion', 'session.id', 'svc.run_id', 'svc.tenant_id', 'svc.catalog_version', + 'm5.operation', 'm5.duration_ms', 'm5.success', } _PSEUDONYMOUS_ATTRS = frozenset({'session.id', 'svc.tenant_id'}) @@ -178,6 +182,34 @@ def set_attr(span, attr_name: str, value) -> None: pass +P = ParamSpec('P') +R = TypeVar('R') + + +def traced(operation: str) -> Callable[[Callable[P, R]], Callable[P, R]]: + """Add a non-blocking duration/outcome span without recording arguments.""" + def decorate(function: Callable[P, R]) -> Callable[P, R]: + @wraps(function) + def wrapped(*args: P.args, **kwargs: P.kwargs) -> R: + started = time.perf_counter() + span = tracer.start_span(operation) + set_attr(span, 'm5.operation', operation) + try: + result = function(*args, **kwargs) + set_attr(span, 'm5.success', True) + return result + except Exception as exc: + set_attr(span, 'm5.success', False) + span.record_exception(exc) + raise + finally: + set_attr(span, 'm5.duration_ms', + (time.perf_counter() - started) * 1000) + span.end() + return wrapped + return decorate + + # ── Init (fail-open, OTel optional and lazy) ───────────────────────────────── def init_tracing(service_name: str = 'sku-engine') -> bool: @@ -200,10 +232,15 @@ def init_tracing(service_name: str = 'sku-engine') -> bool: from opentelemetry.sdk.trace.export import BatchSpanProcessor endpoint = os.environ.get('SKU_OBS_OTLP_ENDPOINT', 'http://localhost:4318/v1/traces') + headers_raw = os.environ.get('SKU_OBS_OTLP_HEADERS', '') + headers = json.loads(headers_raw) if headers_raw else None + if headers is not None and not isinstance(headers, dict): + raise ValueError('SKU_OBS_OTLP_HEADERS must be a JSON object') provider = TracerProvider( resource=Resource.create({'service.name': service_name})) provider.add_span_processor( - BatchSpanProcessor(OTLPSpanExporter(endpoint=endpoint))) + BatchSpanProcessor(OTLPSpanExporter( + endpoint=endpoint, headers=headers))) trace.set_tracer_provider(provider) _active = trace.get_tracer(service_name) # proxy forwards here live return True diff --git a/src/quoting/store.py b/src/quoting/store.py index b3b21e1..7639b75 100644 --- a/src/quoting/store.py +++ b/src/quoting/store.py @@ -16,6 +16,7 @@ from __future__ import annotations import dataclasses +import hmac import json import re import sqlite3 @@ -37,6 +38,7 @@ QuoteTerms, QuoteTotals, ) +from runtime.sqlite_runtime import configure_sqlite _PRE_W3_TABLES = frozenset({ 'quote_seq', 'quote_idempotency', 'quotes', 'quote_revision_seq', @@ -47,6 +49,7 @@ 'issuance_events', 'artifact_jobs', 'delivery_jobs', 'quote_artifacts', 'artifact_guard_events', 'quote_capabilities', 'capability_access_events', 'provider_callback_events', 'worker_heartbeats', + 'data_legal_holds', 'deletion_audit', 'retention_context', }) _SHARED_CONVERSATION_TABLES = frozenset({ 'conversation_schema', 'conversations', 'processed_turns', @@ -185,9 +188,11 @@ def _canonical_assent_receipt(receipt: dict, *, session_id: str, def _connect(source) -> sqlite3.Connection: if isinstance(source, sqlite3.Connection): + configure_sqlite(source) return source conn = sqlite3.connect(str(source), check_same_thread=False) conn.row_factory = sqlite3.Row + configure_sqlite(conn) return conn @@ -217,7 +222,7 @@ def migrate_quote_database(path: str | Path, *, 'before_sha256': sha256(original).hexdigest(), 'after_sha256': sha256(upgraded).hexdigest(), 'counts': counts, - 'schema_version': 4, + 'schema_version': 5, } except Exception: if store is not None: @@ -319,6 +324,9 @@ def __init__(self, source) -> None: f'unknown nonempty quote database schema: {sorted(unknown)}') if legacy and legacy not in (_PRE_W3_TABLES, frozenset({'quotes'})): raise RuntimeError('partial pre-W3 quote schema is not recognized') + # Hold SQLite's cross-process writer lock across the complete schema + # migration; a second API/worker cannot interleave DDL/backfill. + self._conn.execute('BEGIN IMMEDIATE') self._conn.execute( 'CREATE TABLE IF NOT EXISTS quote_schema ' '(singleton INTEGER PRIMARY KEY CHECK(singleton=1), ' @@ -346,11 +354,18 @@ def __init__(self, source) -> None: if 'duplicate column name' not in str(exc).lower(): raise self._conn.executescript(""" + CREATE TABLE IF NOT EXISTS retention_context ( + singleton INTEGER PRIMARY KEY CHECK(singleton=1), + allow_delete INTEGER NOT NULL DEFAULT 0); + INSERT OR IGNORE INTO retention_context VALUES(1,0); + DROP TRIGGER IF EXISTS quote_payload_immutable; + DROP TRIGGER IF EXISTS quote_row_delete_forbidden; CREATE TRIGGER IF NOT EXISTS quote_payload_immutable BEFORE UPDATE OF quote_number,revision,tenant_id,payload_json ON quotes BEGIN SELECT RAISE(ABORT,'quote revisions are immutable'); END; CREATE TRIGGER IF NOT EXISTS quote_row_delete_forbidden - BEFORE DELETE ON quotes + BEFORE DELETE ON quotes WHEN + (SELECT allow_delete FROM retention_context WHERE singleton=1)=0 BEGIN SELECT RAISE(ABORT,'quote revisions are insert-only'); END; """) self._conn.execute( @@ -432,11 +447,18 @@ def __init__(self, source) -> None: 'FOREIGN KEY(tenant_id,quote_number,revision) REFERENCES ' 'quotes(tenant_id,quote_number,revision))') self._conn.executescript(""" + CREATE TABLE IF NOT EXISTS retention_context ( + singleton INTEGER PRIMARY KEY CHECK(singleton=1), + allow_delete INTEGER NOT NULL DEFAULT 0); + INSERT OR IGNORE INTO retention_context VALUES(1,0); + DROP TRIGGER IF EXISTS quote_artifact_update_forbidden; + DROP TRIGGER IF EXISTS quote_artifact_delete_forbidden; CREATE TRIGGER IF NOT EXISTS quote_artifact_update_forbidden BEFORE UPDATE ON quote_artifacts BEGIN SELECT RAISE(ABORT,'quote artifacts are immutable'); END; CREATE TRIGGER IF NOT EXISTS quote_artifact_delete_forbidden - BEFORE DELETE ON quote_artifacts + BEFORE DELETE ON quote_artifacts WHEN + (SELECT allow_delete FROM retention_context WHERE singleton=1)=0 BEGIN SELECT RAISE(ABORT,'quote artifacts are insert-only'); END; """) self._conn.execute( @@ -470,10 +492,19 @@ def __init__(self, source) -> None: 'CREATE TABLE IF NOT EXISTS worker_heartbeats ' '(worker_id TEXT PRIMARY KEY,started_at REAL NOT NULL,last_seen_at REAL NOT NULL,' 'draining INTEGER NOT NULL DEFAULT 0)') + self._conn.execute( + 'CREATE TABLE IF NOT EXISTS data_legal_holds ' + '(tenant_id TEXT NOT NULL,hold_id TEXT NOT NULL,reason TEXT NOT NULL,' + 'created_at REAL NOT NULL,released_at REAL,PRIMARY KEY(tenant_id,hold_id))') + self._conn.execute( + 'CREATE TABLE IF NOT EXISTS deletion_audit ' + '(audit_id INTEGER PRIMARY KEY AUTOINCREMENT,tenant_id_hash TEXT NOT NULL,' + 'actor TEXT NOT NULL,reason TEXT NOT NULL,deleted_counts_json TEXT NOT NULL,' + 'occurred_at REAL NOT NULL)') if 'quotes' in pre_tables and 'quote_issue_keys' not in pre_tables: self._backfill_pre_w3() self._conn.execute( - 'INSERT INTO quote_schema(singleton,version) VALUES(1,4) ' + 'INSERT INTO quote_schema(singleton,version) VALUES(1,5) ' 'ON CONFLICT(singleton) DO UPDATE SET version=excluded.version') self._conn.commit() @@ -1471,6 +1502,110 @@ def operator_quarantine(self, *, tenant_id: str, quote_number: str, self._conn.commit() return changed.rowcount == 1 + def place_legal_hold(self, *, tenant_id: str, hold_id: str, + reason: str, now: float) -> None: + if not all((tenant_id, hold_id, reason)): + raise ValueError('legal hold identity and reason are required') + self._conn.execute( + 'INSERT INTO data_legal_holds VALUES(?,?,?,?,NULL)', + (tenant_id, hold_id, reason, now)) + self._conn.commit() + + def release_legal_hold(self, *, tenant_id: str, hold_id: str, + now: float) -> bool: + changed = self._conn.execute( + 'UPDATE data_legal_holds SET released_at=? WHERE tenant_id=? AND ' + 'hold_id=? AND released_at IS NULL', (now, tenant_id, hold_id)) + self._conn.commit() + return changed.rowcount == 1 + + def purge_tenant(self, *, tenant_id: str, actor: str, reason: str, + audit_key: bytes, now: float) -> dict[str, int]: + """Transactional tenant offboarding with legal hold and keyed audit ID.""" + if len(audit_key) < 32 or not re.fullmatch(r'[A-Za-z0-9_.:@-]{2,80}', actor): + raise ValueError('strong audit key and non-PII actor id are required') + if not reason or len(reason) > 160: + raise ValueError('bounded deletion reason is required') + self._conn.execute('BEGIN IMMEDIATE') + try: + held = self._conn.execute( + 'SELECT 1 FROM data_legal_holds WHERE tenant_id=? AND ' + 'released_at IS NULL LIMIT 1', (tenant_id,)).fetchone() + if held is not None: + raise RuntimeError('tenant deletion blocked by legal hold') + quote_rows = self._conn.execute( + 'SELECT quote_number,revision FROM quotes WHERE tenant_id=?', + (tenant_id,)).fetchall() + quote_numbers = sorted({row['quote_number'] for row in quote_rows}) + counts: dict[str, int] = {} + + def delete(name: str, sql: str, params: tuple) -> None: + counts[name] = self._conn.execute(sql, params).rowcount + + delete('provider_callback_events', + 'DELETE FROM provider_callback_events WHERE delivery_key LIKE ?', + (f'{tenant_id}:%',)) + delete('capability_access_events', + 'DELETE FROM capability_access_events WHERE tenant_id=?', (tenant_id,)) + delete('quote_capabilities', + 'DELETE FROM quote_capabilities WHERE tenant_id=?', (tenant_id,)) + delete('artifact_guard_events', + 'DELETE FROM artifact_guard_events WHERE tenant_id=?', (tenant_id,)) + delete('delivery_jobs', + 'DELETE FROM delivery_jobs WHERE tenant_id=?', (tenant_id,)) + self._conn.execute( + 'UPDATE retention_context SET allow_delete=1 WHERE singleton=1') + delete('quote_artifacts', + 'DELETE FROM quote_artifacts WHERE tenant_id=?', (tenant_id,)) + delete('artifact_jobs', + 'DELETE FROM artifact_jobs WHERE tenant_id=?', (tenant_id,)) + delete('issuance_events', + 'DELETE FROM issuance_events WHERE tenant_id=?', (tenant_id,)) + delete('quote_issue_keys', + 'DELETE FROM quote_issue_keys WHERE tenant_id=?', (tenant_id,)) + for number in quote_numbers: + delete(f'quote_queue:{number}', + 'DELETE FROM quote_queue WHERE quote_number=?', (number,)) + delete(f'quote_idempotency:{number}', + 'DELETE FROM quote_idempotency WHERE quote_number=?', (number,)) + delete(f'quote_revision_seq:{number}', + 'DELETE FROM quote_revision_seq WHERE quote_number=?', (number,)) + delete('quotes', 'DELETE FROM quotes WHERE tenant_id=?', (tenant_id,)) + delete('quote_seq', 'DELETE FROM quote_seq WHERE tenant_id=?', (tenant_id,)) + delete('quote_number_allocations', + 'DELETE FROM quote_number_allocations WHERE tenant_id=?', (tenant_id,)) + shared_tables = {row[0] for row in self._conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'")} + if 'conversation_events' in shared_tables: + delete('conversation_events', + 'DELETE FROM conversation_events WHERE tenant_id=?', (tenant_id,)) + delete('processed_turns', + 'DELETE FROM processed_turns WHERE tenant_id=?', (tenant_id,)) + delete('call_completion_alerts', + 'DELETE FROM call_completion_alerts WHERE tenant_id=?', (tenant_id,)) + delete('call_completion_events', + 'DELETE FROM call_completion_events WHERE tenant_id=?', (tenant_id,)) + delete('conversations', + 'DELETE FROM conversations WHERE tenant_id=?', (tenant_id,)) + delete('data_legal_holds', + 'DELETE FROM data_legal_holds WHERE tenant_id=?', (tenant_id,)) + self._conn.execute( + 'UPDATE retention_context SET allow_delete=0 WHERE singleton=1') + tenant_hash = hmac.new( + audit_key, tenant_id.encode(), sha256).hexdigest()[:24] + self._conn.execute( + 'INSERT INTO deletion_audit(' + 'tenant_id_hash,actor,reason,deleted_counts_json,occurred_at) ' + 'VALUES(?,?,?,?,?)', + (tenant_hash, actor, reason, + json.dumps(counts, sort_keys=True, separators=(',', ':')), now)) + self._conn.execute('COMMIT') + return counts + except Exception: + if self._conn.in_transaction: + self._conn.execute('ROLLBACK') + raise + def claim_pending(self) -> QueueItem | None: """Claim the oldest pending row for the single M5 worker.""" row = self._conn.execute( diff --git a/src/quoting/worker.py b/src/quoting/worker.py index d68dc8a..c3a034e 100644 --- a/src/quoting/worker.py +++ b/src/quoting/worker.py @@ -7,6 +7,7 @@ from typing import Mapping from gateway.journal import ConversationJournal +from observability.telemetry import traced from quoting.delivery import ( AccountContact, ChannelAdapter, @@ -41,6 +42,7 @@ def _fingerprint(tenant_id: str, contact_id: str, destination: str) -> str: f'{tenant_id}:{contact_id}:{destination}'.encode()).hexdigest() +@traced('m5.artifact_job') def run_artifact_once(*, store: SqliteQuoteStore, owner: str, now_fn: Callable[[], datetime] = lambda: datetime.now( timezone.utc), @@ -75,6 +77,7 @@ def run_artifact_once(*, store: SqliteQuoteStore, owner: str, return True +@traced('m5.delivery_job') def run_delivery_once( *, store: SqliteQuoteStore, owner: str, adapters: Mapping[str, ChannelAdapter], account_lookup: AccountLookup, diff --git a/src/runtime/app.py b/src/runtime/app.py index 4e53249..27735d0 100644 --- a/src/runtime/app.py +++ b/src/runtime/app.py @@ -18,6 +18,7 @@ """ # NB: no `from __future__ import annotations` — FastAPI must see real types # (notably `Request`) at decoration time, not stringized annotations. +import hmac import os from typing import Dict, Tuple @@ -28,8 +29,10 @@ from gateway.voice import transcript_is_usable from runtime import twilio_sig, twiml from runtime.config import ( + REPO, build_gateway, build_improvement, + build_journey_emitter, build_persona, build_streaming_asr, build_tts, @@ -48,6 +51,75 @@ def create_app(*, streaming_asr=None, tts=None, persona=None, improvement=None, app = FastAPI(title='SKU Resolution Gateway', version='1.0.0') gateway, sessions = gateway_bundle if gateway_bundle is not None else build_gateway() app.state.gateway = gateway + journey = build_journey_emitter() + app.state.journey = journey + + def emit_journey(event, correlation_source, **properties): + try: + journey.emit( + event, correlation_source=correlation_source, + properties=properties) + except Exception: + pass + from runtime.drain import DrainController + from runtime.rate_limit import RateLimiter, scope_for_path + rate_limiter = RateLimiter() + drain = DrainController() + app.state.drain = drain + app.router.add_event_handler('shutdown', drain.start) + + @app.middleware('http') + async def bounded_application_surface(request: Request, call_next): + admission_exempt = request.url.path in { + '/healthz', '/readyz', '/internal/drain'} + if not admission_exempt and not drain.enter(): + return JSONResponse({'error': 'service_draining'}, status_code=503) + try: + selected = scope_for_path(request.url.path) + if selected is not None: + scope, limit = selected + tenant_value = getattr(gateway.catalog, 'tenant_id', 'tenant_001') + tenant_id = str( + tenant_value() if callable(tenant_value) else tenant_value) + source = request.client.host if request.client else 'unknown' + if not rate_limiter.allow( + scope=scope, tenant_id=tenant_id, source=source, limit=limit): + emit_journey( + 'rate_limited', f'{tenant_id}:{source}', stage=scope) + return JSONResponse( + {'error': 'request_unavailable'}, status_code=429, + headers={'Retry-After': '60'}) + return await call_next(request) + finally: + if not admission_exempt: + drain.exit() + + @app.get('/healthz') + async def healthz(): + return {'ok': True} + + @app.get('/readyz') + async def readyz(request: Request): + from runtime.production import production_report + expected = os.environ.get('SKU_READY_TOKEN', '') + supplied = request.headers.get('X-Internal-Ready-Token', '') + if not expected or not hmac.compare_digest(expected, supplied): + return JSONResponse({'status': 'not_found'}, status_code=404) + report = production_report(REPO) + return JSONResponse(report.public(), status_code=200 if report.ready else 503) + + @app.post('/internal/drain') + async def start_drain(request: Request): + expected = os.environ.get('SKU_READY_TOKEN', '') + supplied = request.headers.get('X-Internal-Ready-Token', '') + if not expected or not hmac.compare_digest(expected, supplied): + return JSONResponse({'status': 'not_found'}, status_code=404) + drain.start() + return {'status': 'draining', 'active': drain.active} + + if os.environ.get('SKU_ENVIRONMENT') == 'production': + from runtime.production import require_production + require_production(REPO) # Voice persona (name/accent/voice/greeting) + ASR/TTS (injectable for tests). persona = persona if persona is not None else build_persona() asr = streaming_asr if streaming_asr is not None else build_streaming_asr() @@ -92,6 +164,8 @@ def retrieve_quote(token: str, format_name: str): tenant_id, token, format_name, now=gateway.now_fn()) if artifact is None: return Response(status_code=404) + emit_journey('artifact_download', token, format=format_name, + outcome='allowed') return Response( content=artifact.body, media_type=artifact.media_type, headers=artifact.headers) @@ -264,6 +338,12 @@ async def agent_turn(request: Request): load_call(gateway, caller, before) return JSONResponse({'error': 'conversation_persistence_failed'}, status_code=503) + if phase.startswith('readback'): + emit_journey('readback', caller, stage=phase) + if resp.needs_confirmation: + emit_journey('clarification', caller, stage=phase) + if resp.meta.get('quote_number'): + emit_journey('assent', caller, stage='issued') return result @app.post('/webhooks/elevenlabs/post-call') @@ -323,6 +403,8 @@ async def elevenlabs_post_call(request: Request): completion_event_id=completion.event_id, effective_hangup_at=completion.effective_hangup_at, received_at=completion.received_at) + emit_journey('hangup', qualified.conversation_id, + stage='authenticated_completion') except PostCallError: return JSONResponse({'error': 'invalid_post_call'}, status_code=401) except ConversationConflict: @@ -358,6 +440,9 @@ async def twilio_delivery_status(request: Request): provider_id=callback.provider_id, outcome=callback.outcome, payload_digest=callback.payload_digest, received_at=time.time()) + emit_journey( + 'terminal_callback', callback.delivery_key, + channel='sms', status=callback.outcome) except CallbackError: return JSONResponse({'error': 'invalid_twilio_callback'}, status_code=401) @@ -397,6 +482,9 @@ async def sendgrid_delivery_events(request: Request): provider_id=callback.provider_id, outcome=callback.outcome, payload_digest=callback.payload_digest, received_at=time.time())) + emit_journey( + 'terminal_callback', callback.delivery_key, + channel='email', status=callback.outcome) except CallbackError: return JSONResponse({'error': 'invalid_sendgrid_callback'}, status_code=401) diff --git a/src/runtime/backup.py b/src/runtime/backup.py new file mode 100644 index 0000000..cbe77ad --- /dev/null +++ b/src/runtime/backup.py @@ -0,0 +1,156 @@ +"""Online encrypted SQLite backups and clean-host semantic restore proof.""" +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import os +import sqlite3 +import tempfile +import time +from pathlib import Path +from typing import Mapping + +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + +def _snapshot(path: Path) -> bytes: + source = sqlite3.connect(str(path)) + with tempfile.NamedTemporaryFile(suffix='.sqlite') as handle: + destination = sqlite3.connect(handle.name) + source.backup(destination) + destination.close() + data = Path(handle.name).read_bytes() + source.close() + return data + + +def _inventory(path: Path, digest_key: bytes) -> dict[str, object]: + db = sqlite3.connect(str(path)) + db.row_factory = sqlite3.Row + if db.execute('PRAGMA quick_check').fetchone()[0] != 'ok': + raise RuntimeError('backup snapshot failed integrity check') + tables = [row[0] for row in db.execute( + "SELECT name FROM sqlite_master WHERE type='table' " + "AND name NOT LIKE 'sqlite_%' ORDER BY name")] + counts = {table: int(db.execute( + f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]) for table in tables} + artifact_rows = [] + if 'quote_artifacts' in tables: + for row in db.execute( + 'SELECT tenant_id,quote_number,revision,format,content,sha256 ' + 'FROM quote_artifacts ORDER BY tenant_id,quote_number,revision,format'): + observed = hashlib.sha256(bytes(row['content'])).hexdigest() + if observed != row['sha256']: + raise RuntimeError('artifact hash mismatch in backup source') + artifact_rows.append([ + row['tenant_id'], row['quote_number'], int(row['revision']), + row['format'], observed]) + cursor_rows = [] + if 'conversations' in tables: + cursor_rows = [dict(row) for row in db.execute( + 'SELECT tenant_id,caller_id,version,history_cursor,history_digest,' + 'authenticated_agent,environment FROM conversations ' + 'ORDER BY tenant_id,caller_id')] + db.close() + artifact_digest = hmac.new(digest_key, json.dumps( + artifact_rows, sort_keys=True, separators=(',', ':')).encode(), + hashlib.sha256).hexdigest() + cursor_digest = hmac.new(digest_key, json.dumps( + cursor_rows, sort_keys=True, separators=(',', ':')).encode(), + hashlib.sha256).hexdigest() + return { + 'tables': tables, 'row_counts': counts, + 'artifact_set_sha256': artifact_digest, + 'conversation_cursors_sha256': cursor_digest, + } + + +def create_backup(*, sources: Mapping[str, str | Path], output: str | Path, + key: bytes, key_id: str, now: float | None = None) -> dict[str, object]: + if len(key) != 32 or not key_id: + raise ValueError('a 32-byte backup key and key_id are required') + created = time.time() if now is None else now + records = {} + aes = AESGCM(key) + for name, raw_path in sorted(sources.items()): + if not name or '/' in name or name == '.' or '..' in name: + raise ValueError('invalid backup source name') + path = Path(raw_path) + data = _snapshot(path) + nonce = os.urandom(12) + aad = f'sku-resolver-backup-v1:{key_id}:{name}'.encode() + records[name] = { + 'size': len(data), 'sha256': hashlib.sha256(data).hexdigest(), + 'nonce': base64.b64encode(nonce).decode(), + 'ciphertext': base64.b64encode(aes.encrypt(nonce, data, aad)).decode(), + 'inventory': _inventory(path, key), + } + envelope = { + 'schema_version': 1, 'key_id': key_id, 'created_at': created, + 'databases': records, + } + canonical = json.dumps(envelope, sort_keys=True, separators=(',', ':')) + envelope['manifest_sha256'] = hashlib.sha256(canonical.encode()).hexdigest() + target = Path(output) + target.parent.mkdir(parents=True, exist_ok=True) + temp = target.with_suffix(target.suffix + '.tmp') + temp.write_text(json.dumps(envelope, sort_keys=True) + '\n', encoding='utf-8') + os.chmod(temp, 0o600) + temp.replace(target) + return {key: envelope[key] for key in ( + 'schema_version', 'key_id', 'created_at', 'manifest_sha256')} + + +def restore_and_verify(*, backup: str | Path, destination: str | Path, + keys: Mapping[str, bytes], + status_path: str | Path | None = None, + now: float | None = None) -> dict[str, object]: + envelope = json.loads(Path(backup).read_text(encoding='utf-8')) + manifest_hash = envelope.pop('manifest_sha256', '') + canonical = json.dumps(envelope, sort_keys=True, separators=(',', ':')) + if not manifest_hash or not hashlib.sha256( + canonical.encode()).hexdigest() == manifest_hash: + raise RuntimeError('backup manifest integrity failed') + key_id = str(envelope.get('key_id', '')) + key = keys.get(key_id) + if key is None or len(key) != 32: + raise RuntimeError('backup key unavailable') + root = Path(destination) + if root.exists() and any(root.iterdir()): + raise RuntimeError('restore destination must be empty') + root.mkdir(parents=True, exist_ok=True) + aes = AESGCM(key) + restored = {} + try: + for name, record in sorted(envelope['databases'].items()): + nonce = base64.b64decode(record['nonce'], validate=True) + ciphertext = base64.b64decode(record['ciphertext'], validate=True) + aad = f'sku-resolver-backup-v1:{key_id}:{name}'.encode() + data = aes.decrypt(nonce, ciphertext, aad) + if (len(data) != int(record['size']) + or hashlib.sha256(data).hexdigest() != record['sha256']): + raise RuntimeError('restored database hash mismatch') + target = root / f'{name}.sqlite' + target.write_bytes(data) + os.chmod(target, 0o600) + observed = _inventory(target, key) + if observed != record['inventory']: + raise RuntimeError('restored semantic inventory mismatch') + restored[name] = observed + except Exception: + for path in root.glob('*.sqlite'): + path.unlink() + raise + completed = time.time() if now is None else now + receipt = { + 'verified': True, 'completed_at': completed, 'key_id': key_id, + 'manifest_sha256': manifest_hash, + 'databases': sorted(restored), + } + if status_path is not None: + marker = Path(status_path) + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text(json.dumps(receipt, sort_keys=True) + '\n', encoding='utf-8') + return receipt diff --git a/src/runtime/config.py b/src/runtime/config.py index a5cea36..f2e49fd 100644 --- a/src/runtime/config.py +++ b/src/runtime/config.py @@ -226,6 +226,20 @@ def build_quote_deliverer(): return SimulatedDeliverer() +def build_journey_emitter(): + from observability.journey import ( + JourneyEmitter, + NoopEventSink, + PostHogEventSink, + ) + + key = os.environ.get('POSTHOG_API_KEY', '') + host = os.environ.get('POSTHOG_HOST', '') + sink = (PostHogEventSink(project_key=key, host=host) + if key and host else NoopEventSink()) + return JourneyEmitter(sink) + + def build_delivery_adapters(): """Build selected live adapters eagerly so missing credentials fail startup.""" from quoting.delivery import SendGridEmailAdapter, TwilioSmsAdapter diff --git a/src/runtime/drain.py b/src/runtime/drain.py new file mode 100644 index 0000000..1888d2c --- /dev/null +++ b/src/runtime/drain.py @@ -0,0 +1,37 @@ +"""Process-local admission gate for graceful API drain.""" +from __future__ import annotations + +import threading + + +class DrainController: + def __init__(self) -> None: + self._condition = threading.Condition() + self._draining = False + self._active = 0 + + def enter(self) -> bool: + with self._condition: + if self._draining: + return False + self._active += 1 + return True + + def exit(self) -> None: + with self._condition: + self._active -= 1 + if self._active == 0: + self._condition.notify_all() + + def start(self) -> None: + with self._condition: + self._draining = True + + def wait(self, timeout: float) -> bool: + with self._condition: + return self._condition.wait_for(lambda: self._active == 0, timeout) + + @property + def active(self) -> int: + with self._condition: + return self._active diff --git a/src/runtime/production.py b/src/runtime/production.py new file mode 100644 index 0000000..1b16c4d --- /dev/null +++ b/src/runtime/production.py @@ -0,0 +1,182 @@ +"""Fail-closed production startup and internal readiness checks.""" +from __future__ import annotations + +import json +import os +import shutil +import sqlite3 +import time +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Mapping +from urllib.parse import urlparse + +from runtime.m5_release import M5ReleaseConfig +from runtime.sqlite_runtime import configure_sqlite + +QUOTE_SCHEMA_VERSION = 5 +CONVERSATION_SCHEMA_VERSION = 2 + + +@dataclass(frozen=True) +class ProductionReport: + ready: bool + failures: tuple[str, ...] + + def public(self) -> dict[str, object]: + return {'ready': self.ready, 'status': 'ready' if self.ready else 'not_ready'} + + +def _secure(value: str) -> bool: + lowered = value.lower() + return (len(value) >= 24 and not any(marker in lowered for marker in ( + 'dev', 'demo', 'example', 'insecure', 'changeme', 'test-secret'))) + + +def _sqlite_check(path: str, *, table: str, expected: int) -> list[str]: + failures: list[str] = [] + if not path or path == ':memory:' or not Path(path).is_absolute(): + return [f'{table}:persistent_absolute_path_required'] + target = Path(path) + if not target.exists() or not target.is_file(): + return [f'{table}:database_missing'] + try: + db = sqlite3.connect(path, timeout=0.25, isolation_level=None) + db.row_factory = sqlite3.Row + configure_sqlite(db) + if db.execute('PRAGMA quick_check').fetchone()[0] != 'ok': + failures.append(f'{table}:integrity_check_failed') + version = db.execute( + f'SELECT version FROM {table} WHERE singleton=1').fetchone() + if version is None or int(version['version']) != expected: + failures.append(f'{table}:schema_stale') + if db.execute('PRAGMA foreign_keys').fetchone()[0] != 1: + failures.append(f'{table}:foreign_keys_disabled') + if db.execute('PRAGMA journal_mode').fetchone()[0].lower() != 'wal': + failures.append(f'{table}:wal_disabled') + try: + db.execute('BEGIN IMMEDIATE') + db.execute('ROLLBACK') + except sqlite3.Error: + failures.append(f'{table}:database_not_writable') + db.close() + except (sqlite3.Error, OSError): + failures.append(f'{table}:database_unavailable') + return failures + + +def production_report( + repo: str | Path, *, env: Mapping[str, str] | None = None, + now: float | None = None) -> ProductionReport: + values = os.environ if env is None else env + root = Path(repo) + at = time.time() if now is None else now + failures: list[str] = [] + if values.get('SKU_ENVIRONMENT') != 'production': + failures.append('environment:not_production') + release = M5ReleaseConfig.load(root / 'config' / 'm5_release.json') + failures.extend(item.replace('policy:', 'release_policy:', 1) + for item in release.production_blockers()) + for name in ('SKU_SESSION_SECRET', 'SKU_QUOTE_LINK_SECRET', + 'SKU_PSEUDONYM_KEY', 'SKU_READY_TOKEN'): + if not _secure(values.get(name, '')): + failures.append(f'secret:{name}:missing_or_insecure') + origin = urlparse(values.get('SKU_QUOTE_PUBLIC_BASE_URL', '')) + if (origin.scheme != 'https' or not origin.netloc or origin.path not in ('', '/') + or 'example' in origin.netloc.lower()): + failures.append('origin:invalid_or_example') + for name in ('SKU_CUSTOMER_DB', 'SKU_PRICEBOOK_DB', 'SKU_CATALOG_PATH', + 'SKU_INVENTORY_PATH'): + raw = values.get(name, '') + if (not raw or not Path(raw).is_absolute() or not Path(raw).exists() + or str(root / 'data') in str(Path(raw))): + failures.append(f'source:{name}:missing_or_fixture') + try: + max_age = int(values.get('SKU_SOURCE_MAX_AGE_SECONDS', '0')) + if max_age <= 0: + raise ValueError + inventory_age = at - Path(values['SKU_INVENTORY_PATH']).stat().st_mtime + if inventory_age > max_age: + failures.append('source:inventory_stale') + price_db = sqlite3.connect(values['SKU_PRICEBOOK_DB']) + row = price_db.execute( + 'SELECT as_of FROM pricebook_meta WHERE singleton=1').fetchone() + price_db.close() + as_of = datetime.fromisoformat(str(row[0])).timestamp() if row else 0 + if at - as_of > max_age: + failures.append('source:pricebook_stale') + except (KeyError, OSError, sqlite3.Error, ValueError, TypeError): + failures.append('source:freshness_unverifiable') + quote_db = values.get('SKU_QUOTE_STORE_DB', '') + conversation_db = values.get('SKU_CONVERSATION_STORE_DB', '') + if (quote_db and conversation_db + and Path(quote_db).resolve() != Path(conversation_db).resolve()): + failures.append('storage:single_transactional_database_required') + failures.extend(_sqlite_check( + quote_db, table='quote_schema', expected=QUOTE_SCHEMA_VERSION)) + failures.extend(_sqlite_check( + conversation_db, table='conversation_schema', + expected=CONVERSATION_SCHEMA_VERSION)) + for name in ('ELEVENLABS_API_KEY', 'ELEVENLABS_WEBHOOK_SECRET', + 'TWILIO_ACCOUNT_SID', 'TWILIO_AUTH_TOKEN', 'TWILIO_FROM_NUMBER', + 'SENDGRID_API_KEY', 'SENDGRID_FROM_EMAIL', + 'SENDGRID_WEBHOOK_PUBLIC_KEY', 'POSTHOG_API_KEY', 'POSTHOG_HOST'): + if not values.get(name, '').strip(): + failures.append(f'provider:{name}:missing') + for name in ('SKU_VOLUME_ENCRYPTION_ATTESTATION', + 'SKU_BACKUP_ACTIVE_KEY_ID', 'SKU_BACKUP_KEYS_JSON'): + if not values.get(name, '').strip(): + failures.append(f'encryption:{name}:missing') + try: + app_days = int(values.get('SKU_RETENTION_DAYS', '0')) + audio_days = int(values.get('ELEVENLABS_AUDIO_RETENTION_DAYS', '0')) + transcript_days = int(values.get( + 'ELEVENLABS_TRANSCRIPT_RETENTION_DAYS', '0')) + if (app_days <= 0 or audio_days < 0 or transcript_days < 0 + or audio_days > app_days or transcript_days > app_days): + raise ValueError + except ValueError: + failures.append('retention:application_platform_policy_inconsistent') + posthog = urlparse(values.get('POSTHOG_HOST', '')) + otlp = urlparse(values.get('SKU_OBS_OTLP_ENDPOINT', '')) + if posthog.scheme != 'https' or not posthog.netloc: + failures.append('observability:posthog_host_invalid') + if (values.get('SKU_OBS_TRACING') != '1' + or otlp.scheme != 'https' or not otlp.netloc): + failures.append('observability:otlp_not_configured') + if values.get('SKU_DELIVERY_MODE', '') == 'simulated': + failures.append('provider:simulated_delivery_forbidden') + try: + db = sqlite3.connect(quote_db) + row = db.execute( + 'SELECT MAX(last_seen_at) FROM worker_heartbeats WHERE draining=0' + ).fetchone() + if row is None or row[0] is None or at - float(row[0]) > 60: + failures.append('worker:heartbeat_stale') + db.close() + except sqlite3.Error: + failures.append('worker:heartbeat_unavailable') + marker = Path(values.get('SKU_BACKUP_STATUS_PATH', '')) + try: + backup = json.loads(marker.read_text(encoding='utf-8')) + if (backup.get('verified') is not True + or at - float(backup.get('completed_at', 0)) > 86_400): + failures.append('backup:missing_or_stale') + except (OSError, ValueError, TypeError, json.JSONDecodeError): + failures.append('backup:missing_or_stale') + volume = Path(quote_db).parent if quote_db else root + try: + minimum = int(values.get('SKU_MIN_FREE_BYTES', str(1024 ** 3))) + if shutil.disk_usage(volume).free < minimum: + failures.append('storage:disk_headroom_low') + except (OSError, ValueError): + failures.append('storage:disk_headroom_unknown') + return ProductionReport(not failures, tuple(sorted(set(failures)))) + + +def require_production(repo: str | Path, *, + env: Mapping[str, str] | None = None) -> None: + report = production_report(repo, env=env) + if not report.ready: + raise RuntimeError('production startup refused: ' + ','.join(report.failures)) diff --git a/src/runtime/rate_limit.py b/src/runtime/rate_limit.py new file mode 100644 index 0000000..5b01a85 --- /dev/null +++ b/src/runtime/rate_limit.py @@ -0,0 +1,56 @@ +"""Tenant/source-aware in-process application rate limits.""" +from __future__ import annotations + +import threading +import time +from dataclasses import dataclass +from typing import Callable + +from observability.telemetry import pseudonymize + + +@dataclass +class _Window: + started_at: float + count: int + + +class RateLimiter: + def __init__(self, *, now_fn: Callable[[], float] = time.monotonic): + self.now_fn = now_fn + self._lock = threading.Lock() + self._windows: dict[tuple[str, str], _Window] = {} + + def allow(self, *, scope: str, tenant_id: str, source: str, + limit: int, window_seconds: float = 60.0) -> bool: + if limit <= 0 or window_seconds <= 0: + return False + identity = pseudonymize( + f'{tenant_id}:{source}', namespace=f'rate-limit:{scope}') + key = (scope, identity) + now = self.now_fn() + with self._lock: + window = self._windows.get(key) + if window is None or now - window.started_at >= window_seconds: + self._windows[key] = _Window(now, 1) + return True + if window.count >= limit: + return False + window.count += 1 + return True + + +def scope_for_path(path: str) -> tuple[str, int] | None: + if path == '/agent/turn': + return ('agent_tool', 120) + if path.endswith('/verify') and path.startswith('/v1/sessions/'): + return ('account_verification', 10) + if path.startswith('/quote/'): + return ('quote_link', 60) + if path.startswith('/webhooks/twilio/delivery'): + return ('twilio_callback', 300) + if path.startswith('/webhooks/sendgrid/events'): + return ('sendgrid_callback', 300) + if path.startswith('/webhooks/elevenlabs/post-call'): + return ('post_call_callback', 120) + return None diff --git a/src/runtime/sqlite_runtime.py b/src/runtime/sqlite_runtime.py new file mode 100644 index 0000000..a2f6bf1 --- /dev/null +++ b/src/runtime/sqlite_runtime.py @@ -0,0 +1,36 @@ +"""Shared SQLite production posture and cross-process migration lock.""" +from __future__ import annotations + +import contextlib +import fcntl +import sqlite3 +from pathlib import Path +from typing import Iterator + + +def configure_sqlite(db: sqlite3.Connection) -> None: + db.execute('PRAGMA foreign_keys=ON') + db.execute('PRAGMA busy_timeout=5000') + database = db.execute('PRAGMA database_list').fetchone()[2] + if database: + db.execute('PRAGMA journal_mode=WAL') + db.execute('PRAGMA synchronous=FULL') + db.execute('PRAGMA trusted_schema=OFF') + if db.execute('PRAGMA quick_check').fetchone()[0] != 'ok': + raise RuntimeError('SQLite quick_check failed') + + +@contextlib.contextmanager +def migration_lock(path: str | Path) -> Iterator[None]: + target = Path(path) + lock = target.with_suffix(target.suffix + '.migration.lock') + lock.parent.mkdir(parents=True, exist_ok=True) + with lock.open('a+b') as handle: + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + raise RuntimeError('database migration lock is already held') from exc + try: + yield + finally: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) diff --git a/tests/test_delivery_callback_routes.py b/tests/test_delivery_callback_routes.py index 90696b9..5c8c0cb 100644 --- a/tests/test_delivery_callback_routes.py +++ b/tests/test_delivery_callback_routes.py @@ -60,4 +60,5 @@ def test_signed_sendgrid_route_is_replay_safe_and_terminal(tmp_path, monkeypatch assert first.json()['new_events'] == 1 and replay.json()['new_events'] == 0 assert store.delivery_job( 'tenant_001', document.header.quote_number, 1, 'email').status == 'delivered' - assert b'provider-1' in (tmp_path / 'callbacks.sqlite').read_bytes() + assert store._conn.execute( + 'SELECT provider_id FROM provider_callback_events').fetchone()[0] == 'provider-1' diff --git a/tests/test_production_controls.py b/tests/test_production_controls.py new file mode 100644 index 0000000..f62f651 --- /dev/null +++ b/tests/test_production_controls.py @@ -0,0 +1,249 @@ +from __future__ import annotations + +import json +import shutil +from datetime import datetime, timezone +from decimal import Decimal +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient +from gateway_fixtures import build_gateway +from test_m5_w5_delivery_worker import _store + +from gateway.conversation_store import ConversationStore +from gateway.db_adapters import SqliteCustomerDB, SqlitePriceBook +from gateway.models import Account +from observability.journey import JourneyEmitter, PostHogEventSink +from observability.m5_alerts import M5Signals, evaluate +from quoting.store import SqliteQuoteStore +from runtime.app import create_app +from runtime.backup import create_backup, restore_and_verify +from runtime.drain import DrainController +from runtime.production import production_report +from runtime.rate_limit import RateLimiter + + +def test_sqlite_hardening_online_encrypted_backup_and_clean_restore(tmp_path): + path = tmp_path / 'state.sqlite' + quote_store, document = _store(path) + conversations = ConversationStore(path) + conversations.commit_turn( + tenant_id='tenant_001', caller_id='conversation-1', turn_id='turn-1', + text='private transcript is hashed not stored', expected_version=0, + snapshot={'order': 'safe-structured'}, response={'say': 'ok'}, + history_cursor=1, history_digest='d' * 64, + authenticated_agent='agent|branch|version', environment='staging') + assert quote_store._conn.execute('PRAGMA journal_mode').fetchone()[0] == 'wal' + assert quote_store._conn.execute('PRAGMA foreign_keys').fetchone()[0] == 1 + assert quote_store._conn.execute('PRAGMA busy_timeout').fetchone()[0] == 5000 + key = b'k' * 32 + backup = tmp_path / 'backup.enc.json' + receipt = create_backup( + sources={'state': path}, output=backup, key=key, key_id='backup-v1', now=10) + encrypted = backup.read_bytes() + assert document.header.quote_number.encode() not in encrypted + assert b'private transcript' not in encrypted + status = tmp_path / 'backup-status.json' + restored = restore_and_verify( + backup=backup, destination=tmp_path / 'restored', + keys={'backup-v1': key}, status_path=status, now=20) + assert restored['verified'] is True + assert receipt['manifest_sha256'] == restored['manifest_sha256'] + restored_store = SqliteQuoteStore(tmp_path / 'restored' / 'state.sqlite') + assert restored_store.get(document.header.quote_number) == document + assert json.loads(status.read_text())['completed_at'] == 20 + + +def test_backup_tamper_and_nonempty_restore_fail_closed(tmp_path): + path = tmp_path / 'state.sqlite' + _store(path) + backup = tmp_path / 'backup.json' + create_backup(sources={'state': path}, output=backup, + key=b'x' * 32, key_id='k1') + raw = json.loads(backup.read_text()) + raw['databases']['state']['ciphertext'] = 'AAAA' + backup.write_text(json.dumps(raw)) + with pytest.raises(RuntimeError, match='manifest'): + restore_and_verify( + backup=backup, destination=tmp_path / 'restore', keys={'k1': b'x' * 32}) + + +def test_legal_hold_blocks_transactional_tenant_purge_then_audits(tmp_path): + path = tmp_path / 'state.sqlite' + store, document = _store(path) + calls = ConversationStore(path) + calls.commit_turn( + tenant_id='tenant_001', caller_id='c1', turn_id='t1', text='hello', + expected_version=0, snapshot={}, response={'say': 'ok'}, history_cursor=0, + history_digest='', authenticated_agent='agent', environment='staging') + store.place_legal_hold( + tenant_id='tenant_001', hold_id='legal-1', reason='litigation', now=1) + with pytest.raises(RuntimeError, match='legal hold'): + store.purge_tenant( + tenant_id='tenant_001', actor='privacy.officer', reason='offboarding', + audit_key=b'a' * 32, now=2) + assert store.get(document.header.quote_number) is not None + assert store.release_legal_hold( + tenant_id='tenant_001', hold_id='legal-1', now=3) + counts = store.purge_tenant( + tenant_id='tenant_001', actor='privacy.officer', reason='offboarding', + audit_key=b'a' * 32, now=4) + assert counts['quotes'] == 1 + assert store.get(document.header.quote_number) is None + assert calls.load('tenant_001', 'c1') is None + audit = store._conn.execute('SELECT * FROM deletion_audit').fetchone() + assert audit['tenant_id_hash'] != 'tenant_001' + assert store._conn.execute( + 'SELECT allow_delete FROM retention_context').fetchone()[0] == 0 + + +def test_production_report_and_ready_route_are_fail_closed_and_non_sensitive( + tmp_path, monkeypatch): + report = production_report(Path(__file__).parents[1], env={}, now=100) + assert not report.ready + assert 'environment:not_production' in report.failures + assert report.public() == {'ready': False, 'status': 'not_ready'} + gateway, sessions, _journal, _clock = build_gateway(tmp_path) + monkeypatch.setenv('SKU_READY_TOKEN', 'r' * 32) + client = TestClient(create_app(gateway_bundle=(gateway, sessions))) + assert client.get('/healthz').json() == {'ok': True} + assert client.get('/readyz').status_code == 404 + ready = client.get('/readyz', headers={'X-Internal-Ready-Token': 'r' * 32}) + assert ready.status_code == 503 + assert ready.json() == {'ready': False, 'status': 'not_ready'} + drained = client.post( + '/internal/drain', headers={'X-Internal-Ready-Token': 'r' * 32}) + assert drained.status_code == 200 + assert client.get('/v1/tools.json').status_code == 503 + assert client.get('/healthz').status_code == 200 + + +def test_production_app_startup_refuses_before_serving(tmp_path, monkeypatch): + gateway, sessions, _journal, _clock = build_gateway(tmp_path) + monkeypatch.setenv('SKU_ENVIRONMENT', 'production') + with pytest.raises(RuntimeError, match='production startup refused'): + create_app(gateway_bundle=(gateway, sessions)) + + +def test_production_validator_separates_policy_from_operational_failures(tmp_path): + repo = Path(__file__).parents[1] + state = tmp_path / 'state.sqlite' + store = SqliteQuoteStore(state) + ConversationStore(state) + now = datetime.now(timezone.utc) + store.heartbeat(worker_id='worker-1', now=now.timestamp()) + customer = tmp_path / 'customers.sqlite' + SqliteCustomerDB.build(customer, [Account('1001', 'Customer')]) + prices = tmp_path / 'prices.sqlite' + SqlitePriceBook.build( + prices, {'K5-24SBC': Decimal('1.00')}, {'preferred': Decimal('1')}, + as_of=now) + catalog = tmp_path / 'catalog.csv' + inventory = tmp_path / 'inventory.json' + shutil.copyfile(repo / 'data/catalog.csv', catalog) + shutil.copyfile(repo / 'data/inventory.json', inventory) + backup_status = tmp_path / 'backup-status.json' + backup_status.write_text(json.dumps({ + 'verified': True, 'completed_at': now.timestamp()})) + env = { + 'SKU_ENVIRONMENT': 'production', + 'SKU_SESSION_SECRET': 's' * 32, + 'SKU_QUOTE_LINK_SECRET': 'q' * 32, + 'SKU_PSEUDONYM_KEY': 'p' * 32, + 'SKU_READY_TOKEN': 'r' * 32, + 'SKU_QUOTE_PUBLIC_BASE_URL': 'https://quotes.tenant.invalid', + 'SKU_CUSTOMER_DB': str(customer), 'SKU_PRICEBOOK_DB': str(prices), + 'SKU_CATALOG_PATH': str(catalog), 'SKU_INVENTORY_PATH': str(inventory), + 'SKU_QUOTE_STORE_DB': str(state), + 'SKU_CONVERSATION_STORE_DB': str(state), + 'SKU_SOURCE_MAX_AGE_SECONDS': '3600', + 'SKU_BACKUP_STATUS_PATH': str(backup_status), 'SKU_MIN_FREE_BYTES': '1', + 'ELEVENLABS_API_KEY': 'el', 'ELEVENLABS_WEBHOOK_SECRET': 'hook', + 'TWILIO_ACCOUNT_SID': 'AC1', 'TWILIO_AUTH_TOKEN': 'token', + 'TWILIO_FROM_NUMBER': '+15550100000', 'SENDGRID_API_KEY': 'sg', + 'SENDGRID_FROM_EMAIL': 'sender@tenant.invalid', + 'SENDGRID_WEBHOOK_PUBLIC_KEY': 'public', + 'POSTHOG_API_KEY': 'phc_project', 'POSTHOG_HOST': 'https://us.posthog.com', + 'SKU_OBS_TRACING': '1', + 'SKU_OBS_OTLP_ENDPOINT': 'https://us.posthog.com/i/v1/traces', + 'SKU_DELIVERY_MODE': 'live', + 'SKU_VOLUME_ENCRYPTION_ATTESTATION': 'volume-key-id', + 'SKU_BACKUP_ACTIVE_KEY_ID': 'backup-v1', + 'SKU_BACKUP_KEYS_JSON': '{"backup-v1":"redacted-runtime-secret"}', + 'SKU_RETENTION_DAYS': '30', 'ELEVENLABS_AUDIO_RETENTION_DAYS': '0', + 'ELEVENLABS_TRANSCRIPT_RETENTION_DAYS': '0', + } + report = production_report(repo, env=env, now=now.timestamp()) + assert report.failures + assert [failure for failure in report.failures + if not failure.startswith('release_policy:')] == [] + + simulated = production_report( + repo, env={**env, 'SKU_DELIVERY_MODE': 'simulated'}, now=now.timestamp()) + assert 'provider:simulated_delivery_forbidden' in simulated.failures + split = production_report( + repo, env={**env, 'SKU_CONVERSATION_STORE_DB': str(tmp_path / 'other.sqlite')}, + now=now.timestamp()) + assert 'storage:single_transactional_database_required' in split.failures + stale = production_report(repo, env={ + **env, 'SKU_MIN_FREE_BYTES': str(2 ** 80), + 'SKU_BACKUP_STATUS_PATH': str(tmp_path / 'missing')}, now=now.timestamp()) + assert 'storage:disk_headroom_low' in stale.failures + assert 'backup:missing_or_stale' in stale.failures + + +def test_rate_limit_drain_and_posthog_export_are_nonblocking_and_pii_free(): + clock = {'now': 0.0} + limiter = RateLimiter(now_fn=lambda: clock['now']) + assert limiter.allow(scope='verify', tenant_id='t1', source='1.2.3.4', limit=1) + assert not limiter.allow(scope='verify', tenant_id='t1', source='1.2.3.4', limit=1) + assert limiter.allow(scope='verify', tenant_id='t2', source='1.2.3.4', limit=1) + drain = DrainController() + assert drain.enter() + drain.start() + assert not drain.enter() + drain.exit() + assert drain.wait(0.01) + + captured = [] + sink = PostHogEventSink( + project_key='phc_project_key', host='https://us.posthog.com', + transport=lambda url, body, timeout: captured.append((url, body, timeout))) + emitter = JourneyEmitter(sink) + assert emitter.emit( + 'assent', correlation_source='account 1001 / +15550100100', + properties={'stage': 'quote', 'duration_ms': 12.5}) + with pytest.raises(ValueError, match='non-allowlisted'): + emitter.emit('assent', correlation_source='c1', + properties={'transcript': 'secret'}) + sink.flush() + sink.close() + body = captured[0][1] + assert b'1001' not in body and b'5550100100' not in body + assert b'phc_project_key' in body + + +def test_container_and_process_contract_is_pinned_nonroot_and_split(): + repo = Path(__file__).parents[1] + docker = (repo / 'Dockerfile').read_text() + processes = json.loads((repo / 'deploy/processes.json').read_text()) + assert 'FROM python:3.12.10-slim-bookworm@sha256:' in docker + assert 'USER 10001:10001' in docker + assert 'VOLUME ["/var/lib/sku-resolver"]' in docker + assert processes['api'][0] == 'uvicorn' + assert processes['worker'][:2] == ['python', 'scripts/quote_worker.py'] + assert processes['state_mount'] == '/var/lib/sku-resolver' + + +def test_parity_and_queue_thresholds_page_immediately(): + alerts = evaluate(M5Signals( + parity_failures=1, queue_oldest_age_s=61, stuck_leases=1, + provider_failure_rate=0.5, invalid_reference_rate=0.5, + free_bytes=1, worker_age_s=61, backup_age_s=86_401, + restore_failed=True)) + keys = {alert.key for alert in alerts} + assert {'parity', 'queue_slo', 'stuck_lease', 'provider_outage', + 'invalid_reference', 'disk_pressure', 'worker_stale', + 'backup_restore'} <= keys + assert all(alert.severity in ('warning', 'critical') for alert in alerts) diff --git a/tests/test_quote_migration.py b/tests/test_quote_migration.py index 4fe2de2..c136b34 100644 --- a/tests/test_quote_migration.py +++ b/tests/test_quote_migration.py @@ -66,14 +66,14 @@ def test_p16_upgrade_preserves_legacy_rows_and_backfills_explicit_held_lineage( report = migrate_quote_database(path) assert _legacy_checksum(path) == before - assert report['schema_version'] == 4 + assert report['schema_version'] == 5 assert report['counts'] == { 'quotes': 1, 'quote_issue_keys': 1, 'issuance_events': 1, 'artifact_jobs': 1, 'delivery_jobs': 1, } store = SqliteQuoteStore(path) assert store._conn.execute( - 'SELECT version FROM quote_schema').fetchone()[0] == 4 + 'SELECT version FROM quote_schema').fetchone()[0] == 5 key = store._conn.execute('SELECT * FROM quote_issue_keys').fetchone() assert (key['tenant_id'], key['account_id'], key['session_id'], key['order_digest'], key['action']) == (