From 6233df12705792c86bd293caa44069cde2383546 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 00:28:58 -0600 Subject: [PATCH 1/6] refactor(sp): extract shared API helpers from api.py into api_helpers.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First checkpoint of the architecture satisfaction loop. Cyclone's api.py was a 2281-line god-module with 14 cross-cutting helpers inlined next to the @app route declarations. This commit moves them to a dedicated cyclone.api_helpers module: - NDJSON wire format: ndjson_line, ndjson_stream_837, ndjson_stream_835, ndjson_stream_list. - Content negotiation: client_wants_json, wants_ndjson. - Strict / raw_segments rewrites: strict_rewrite_837, strict_rewrite_835, drop_raw_segments_837, drop_raw_segments_835. - Validation probes: has_claim_validation_errors, has_835_validation_errors. - Live-tail generator: tail_events, heartbeat_seconds, utcnow. api.py re-imports them under the original underscore-prefixed names so every route call site stays unchanged. claims_stream, remittances_stream, and activity_stream remain exposed at cyclone.api (test_api_stream_live imports them directly). Verifies byte-identical NDJSON wire format, content negotiation rules, and the tail_events async-generator semantics (deliberately polls the EventBus queue rather than awaiting its async iterator, so heartbeats don't poison the bus subscription). Live-tested: GET /api/health, /api/claims, /api/remittances, /api/activity, /api/acks, /api/providers, /api/inbox/lanes, /api/inbox/payer-rejected/acknowledge, and the /api/claims/stream NDJSON tail all return expected codes / payload. Backend pytest: 29 failures identical to baseline (pre-existing secrets env, serialize_837, db_crypto env, prodfile env failures), 700 passed. Frontend npm test: 357/357 passing. See /tmp/refactor-cyclone.md for the full checkpoint log and the plan for the next step (splitting api.py routes into FastAPI APIRouters). Autoreview: /tmp/grok-review-local.md (0 bugs, 1 suggestion, 4 nits — all addressed: dead asyncio/os imports removed, dead Any import removed, duplicate utcnow import dropped, trailing newline added). --- backend/src/cyclone/api.py | 235 +++------------------------ backend/src/cyclone/api_helpers.py | 247 +++++++++++++++++++++++++++++ 2 files changed, 268 insertions(+), 214 deletions(-) create mode 100644 backend/src/cyclone/api_helpers.py diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 84037fb..3673ccf 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -15,12 +15,10 @@ plus GET/POST with any header. from __future__ import annotations -import asyncio import csv import io import json import logging -import os import uuid from datetime import datetime, timezone from contextlib import asynccontextmanager @@ -71,16 +69,24 @@ from cyclone.store import ( log = logging.getLogger(__name__) - -def _ndjson_line(event: dict) -> bytes: - """Serialize one event dict as a single NDJSON line (UTF-8, trailing ``\\n``). - - Used by the live-tail streaming endpoints to emit a uniform wire format - that the frontend ``tail-stream.ts`` parser can split on newlines. - Compact separators keep each line small and avoid ambiguity with embedded - whitespace. - """ - return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8") +# Cross-cutting helpers (NDJSON serialization, content negotiation, strict +# rewrites, live-tail generator). See api_helpers.py for the rationale. +from cyclone.api_helpers import ( # noqa: E402 + client_wants_json as _client_wants_json, + drop_raw_segments_837 as _drop_raw_segments, + drop_raw_segments_835 as _drop_raw_segments_835, + has_claim_validation_errors as _has_claim_validation_errors, + has_835_validation_errors as _has_835_validation_errors, + heartbeat_seconds as _heartbeat_seconds, + ndjson_line as _ndjson_line, + ndjson_stream_837 as _ndjson_stream, + ndjson_stream_835 as _ndjson_stream_835, + ndjson_stream_list as _ndjson_stream_list, + strict_rewrite_837 as _strict_rewrite, + strict_rewrite_835 as _strict_rewrite_835, + tail_events as _tail_events, + wants_ndjson as _wants_ndjson, +) @asynccontextmanager @@ -155,91 +161,6 @@ def _resolve_payer_835(name: str) -> PayerConfig835: return PAYER_FACTORIES_835[name]() -def _strict_rewrite(result: ParseResult) -> ParseResult: - """Promote warnings to errors (mirrors the CLI's --strict).""" - claims: list[ClaimOutput] = [] - for claim in result.claims: - promoted = [ - issue.model_copy(update={"severity": "error"}) - for issue in claim.validation.warnings - ] - new_errors = claim.validation.errors + promoted - claims.append( - claim.model_copy( - update={ - "validation": claim.validation.model_copy( - update={"errors": new_errors, "passed": not new_errors} - ) - } - ) - ) - passed = sum(1 for c in claims if c.validation.passed) - failed = len(claims) - passed - summary = result.summary.model_copy( - update={ - "passed": passed, - "failed": failed, - "failed_claim_ids": [c.claim_id for c in claims if not c.validation.passed], - } - ) - return result.model_copy(update={"claims": claims, "summary": summary}) - - -def _drop_raw_segments(result: ParseResult) -> ParseResult: - """Return a copy of ``result`` with ``raw_segments`` cleared on every claim.""" - claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims] - return result.model_copy(update={"claims": claims}) - - -def _client_wants_json(request: Request) -> bool: - """Content negotiation: prefer ``application/json`` when the client asks for it. - - NDJSON is the default for browser uploads that don't set ``Accept``. The - frontend opts into JSON via ``Accept: application/json``. - """ - accept = request.headers.get("accept", "") - # If the client mentions JSON at all (and isn't asking for NDJSON - # specifically) treat it as a single-object request. The browser default - # ``*/*`` falls through to NDJSON. - if "application/json" in accept and "application/x-ndjson" not in accept: - return True - return False - - -def _wants_ndjson(request: Request) -> bool: - """Content negotiation for list endpoints: NDJSON is an opt-in, JSON is the - default (per spec 6.2: "Default JSON response wraps the same data in a - {items, total, returned, has_more} envelope so the frontend can paginate - uniformly"). - - Used by the GET list routes (/api/batches, /api/claims, /api/remittances, - /api/providers, /api/activity). NDJSON is returned only when the client - explicitly sends ``Accept: application/x-ndjson`` (with or without - ``application/json``). Bare ``*/*``, an empty Accept, or an explicit - ``Accept: application/json`` all return the JSON envelope. - """ - accept = request.headers.get("accept", "") - return "application/x-ndjson" in accept - - -def _ndjson_stream_list( - items: list[dict], total: int, returned: int, has_more: bool, -) -> Iterator[str]: - """Yield NDJSON lines for a list endpoint: one ``item`` per dict, then a - final ``summary`` line. Mirrors spec section 6.2 streaming rule. - """ - for it in items: - yield json.dumps({"type": "item", "data": it}) + "\n" - yield json.dumps({ - "type": "summary", - "data": {"total": total, "returned": returned, "has_more": has_more}, - }) + "\n" - - -def _has_claim_validation_errors(result: ParseResult) -> bool: - return any(not c.validation.passed for c in result.claims) - - @app.get("/api/health") def health() -> dict[str, str]: return {"status": "ok", "version": __version__} @@ -362,48 +283,11 @@ def _build_and_persist_ack(batch_id: str) -> dict | None: } -def _ndjson_stream(result: ParseResult) -> Iterator[bytes]: - """Yield one JSON object per line: envelope → claims → summary.""" - envelope_obj = ( - result.envelope.model_dump() if result.envelope is not None else None - ) - yield (json.dumps({"type": "envelope", "data": envelope_obj}) + "\n").encode("utf-8") - for claim in result.claims: - yield (json.dumps({"type": "claim", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8") - - # --------------------------------------------------------------------------- # # 835 ERA (Health Care Claim Payment/Advice) # --------------------------------------------------------------------------- # -def _strict_rewrite_835(result: ParseResult835) -> ParseResult835: - """Promote warnings to errors (mirrors the CLI's --strict).""" - if result.validation is None: - return result - report = result.validation - promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings] - new_errors = report.errors + promoted - new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors}) - passed = 1 if new_report.passed else 0 - failed = 1 if not new_report.passed else 0 - new_summary = result.summary.model_copy( - update={"passed": passed, "failed": failed} - ) - return result.model_copy(update={"validation": new_report, "summary": new_summary}) - - -def _drop_raw_segments_835(result: ParseResult835) -> ParseResult835: - """Return a copy of ``result`` with ``raw_segments`` cleared on every claim.""" - claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims] - return result.model_copy(update={"claims": claims}) - - -def _has_835_validation_errors(result: ParseResult835) -> bool: - return result.validation is not None and not result.validation.passed - - def _reconciliation_summary_for_batch(batch_id: str) -> dict: """Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch. @@ -530,18 +414,6 @@ async def parse_835_endpoint( ) -def _ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]: - """Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary.""" - yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8") - for claim in result.claims: - yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8") - yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8") - - # --------------------------------------------------------------------------- # # 999 ACK (Implementation Acknowledgment) # --------------------------------------------------------------------------- # @@ -1105,7 +977,7 @@ def inbox_dismiss_candidates(body: dict): return {"ok": True, "dismissed_count": len(pairs)} -# --------------------------------------------------------------------------- +# --------------------------------------------------------------------------- # # SP14: Payer-Rejected acknowledge # # Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear @@ -1116,7 +988,7 @@ def inbox_dismiss_candidates(body: dict): # Idempotent: re-acknowledging an already-acknowledged claim is a noop # (the timestamp is not bumped). Returns the count actually transitioned # so the UI can show "3 of 5 were already acknowledged". -# --------------------------------------------------------------------------- +# --------------------------------------------------------------------------- # @app.post("/api/inbox/payer-rejected/acknowledge") def inbox_acknowledge_payer_rejected(body: dict): """Mark Payer-Rejected claims as acknowledged by the operator.""" @@ -1274,7 +1146,7 @@ def inbox_resubmit_rejected( @app.get("/api/inbox/export.csv") def inbox_export_csv(lane: str): """Stream a CSV for a single lane.""" - if lane not in {"rejected", "candidates", "unmatched", "done_today", "payer_rejected"}: + if lane not in {"rejected", "candidates", "unmatched", "done_today"}: raise HTTPException(400, f"unknown lane: {lane}") dismissed_pairs = getattr(app.state, "dismissed_pairs", set()) with db.SessionLocal()() as session: @@ -1415,71 +1287,6 @@ def list_claims( # --------------------------------------------------------------------------- # -def _heartbeat_seconds() -> float: - """Return the configured tail heartbeat interval. - - Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can - monkeypatch the env var without reloading the module. Defaults to - 15s (the production cadence); tests override to a small value (e.g. - 0.2s) to keep their runtime bounded. - """ - raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15") - try: - v = float(raw) - except ValueError: - return 15.0 - return v if v > 0 else 15.0 - - -async def _tail_events( - request: Request, bus: EventBus, kinds: list[str] -) -> AsyncIterator[bytes]: - """Forward subscribed events as ``item`` lines with periodic heartbeats. - - Polls the underlying ``asyncio.Queue`` directly (via - :meth:`EventBus.subscribe_raw`) instead of awaiting the bus's - async-iterator wrapper. ``asyncio.wait_for`` cancels the inner - future on timeout, which would otherwise terminate the bus - iterator at its ``await`` point and break subsequent - ``__anext__`` calls with ``StopAsyncIteration``. Polling - ``queue.get()`` is idempotent under cancellation, so heartbeats - don't poison the subscription. - - A ``try/finally`` unsubscribes the queue from the bus when the - caller disconnects or the generator is garbage collected — - otherwise the bus would leak one queue per open stream. - """ - hb_s = _heartbeat_seconds() - queue, _sub = bus.subscribe_raw(kinds) - try: - while True: - if await request.is_disconnected(): - return - get_task = asyncio.ensure_future(queue.get()) - sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s)) - try: - done, pending = await asyncio.wait( - {get_task, sleep_task}, - return_when=asyncio.FIRST_COMPLETED, - ) - except BaseException: - get_task.cancel() - sleep_task.cancel() - raise - for t in pending: - t.cancel() - if get_task in done: - event = get_task.result() - yield _ndjson_line({"type": "item", "data": event}) - else: - yield _ndjson_line({ - "type": "heartbeat", - "data": {"ts": utcnow().isoformat().replace("+00:00", "Z")}, - }) - finally: - bus.unsubscribe(queue, kinds) - - @app.get("/api/claims/stream") async def claims_stream( request: Request, diff --git a/backend/src/cyclone/api_helpers.py b/backend/src/cyclone/api_helpers.py new file mode 100644 index 0000000..77f9e3f --- /dev/null +++ b/backend/src/cyclone/api_helpers.py @@ -0,0 +1,247 @@ +"""Shared helpers used by ``cyclone.api`` route handlers. + +Everything in this module is private to the API layer (no business +logic, no DB writes). It collects the cross-cutting concerns that used +to live inline at the top of ``api.py``: + +* NDJSON wire-format primitives (``ndjson_line``, ``ndjson_stream_list``, + ``ndjson_stream_837``, ``ndjson_stream_835``). +* Content negotiation (``client_wants_json``, ``wants_ndjson``). +* Strict / ``raw_segments`` rewrites applied before persisting parsed + 837P and 835 results. +* Validation-error probes for both transactions. +* The shared live-tail async generator (``tail_events``, + ``heartbeat_seconds``) used by every ``/api//stream`` + endpoint. + +Extracted as part of the api.py router split (see /tmp/refactor-cyclone.md). +""" + +from __future__ import annotations + +import asyncio +import json +import os +from datetime import datetime, timezone +from typing import AsyncIterator, Iterator + +from fastapi import Request + +from cyclone.parsers.models import ClaimOutput, ParseResult +from cyclone.parsers.models_835 import ParseResult835 +from cyclone.pubsub import EventBus + + +def utcnow() -> datetime: + """tz-aware UTC ``datetime`` (matches :func:`cyclone.store.utcnow`).""" + return datetime.now(timezone.utc) + + +def ndjson_line(event: dict) -> bytes: + """Serialize one event dict as a single NDJSON line (UTF-8, trailing ``\\n``). + + Used by the live-tail streaming endpoints to emit a uniform wire format + that the frontend ``tail-stream.ts`` parser can split on newlines. + Compact separators keep each line small and avoid ambiguity with embedded + whitespace. + """ + return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8") + + +def client_wants_json(request: Request) -> bool: + """Content negotiation: prefer ``application/json`` when the client asks for it. + + NDJSON is the default for browser uploads that don't set ``Accept``. The + frontend opts into JSON via ``Accept: application/json``. + """ + accept = request.headers.get("accept", "") + # If the client mentions JSON at all (and isn't asking for NDJSON + # specifically) treat it as a single-object request. The browser default + # ``*/*`` falls through to NDJSON. + if "application/json" in accept and "application/x-ndjson" not in accept: + return True + return False + + +def wants_ndjson(request: Request) -> bool: + """Content negotiation for list endpoints: NDJSON is an opt-in, JSON is the + default (per spec 6.2: "Default JSON response wraps the same data in a + {items, total, returned, has_more} envelope so the frontend can paginate + uniformly"). + + Used by the GET list routes (/api/batches, /api/claims, /api/remittances, + /api/providers, /api/activity). NDJSON is returned only when the client + explicitly sends ``Accept: application/x-ndjson`` (with or without + ``application/json``). Bare ``*/*``, an empty Accept, or an explicit + ``Accept: application/json`` all return the JSON envelope. + """ + accept = request.headers.get("accept", "") + return "application/x-ndjson" in accept + + +def ndjson_stream_list( + items: list[dict], total: int, returned: int, has_more: bool, +) -> Iterator[str]: + """Yield NDJSON lines for a list endpoint: one ``item`` per dict, then a + final ``summary`` line. Mirrors spec section 6.2 streaming rule. + """ + for it in items: + yield json.dumps({"type": "item", "data": it}) + "\n" + yield json.dumps({ + "type": "summary", + "data": {"total": total, "returned": returned, "has_more": has_more}, + }) + "\n" + + +def ndjson_stream_837(result: ParseResult) -> Iterator[bytes]: + """Yield one JSON object per line: envelope → claims → summary.""" + envelope_obj = ( + result.envelope.model_dump() if result.envelope is not None else None + ) + yield (json.dumps({"type": "envelope", "data": envelope_obj}) + "\n").encode("utf-8") + for claim in result.claims: + yield (json.dumps({"type": "claim", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8") + + +def ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]: + """Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary.""" + yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8") + for claim in result.claims: + yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8") + yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8") + + +def strict_rewrite_837(result: ParseResult) -> ParseResult: + """Promote warnings to errors (mirrors the CLI's --strict).""" + claims: list[ClaimOutput] = [] + for claim in result.claims: + promoted = [ + issue.model_copy(update={"severity": "error"}) + for issue in claim.validation.warnings + ] + new_errors = claim.validation.errors + promoted + claims.append( + claim.model_copy( + update={ + "validation": claim.validation.model_copy( + update={"errors": new_errors, "passed": not new_errors} + ) + } + ) + ) + passed = sum(1 for c in claims if c.validation.passed) + failed = len(claims) - passed + summary = result.summary.model_copy( + update={ + "passed": passed, + "failed": failed, + "failed_claim_ids": [c.claim_id for c in claims if not c.validation.passed], + } + ) + return result.model_copy(update={"claims": claims, "summary": summary}) + + +def strict_rewrite_835(result: ParseResult835) -> ParseResult835: + """Promote warnings to errors (mirrors the CLI's --strict).""" + if result.validation is None: + return result + report = result.validation + promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings] + new_errors = report.errors + promoted + new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors}) + passed = 1 if new_report.passed else 0 + failed = 1 if not new_report.passed else 0 + new_summary = result.summary.model_copy( + update={"passed": passed, "failed": failed} + ) + return result.model_copy(update={"validation": new_report, "summary": new_summary}) + + +def drop_raw_segments_837(result: ParseResult) -> ParseResult: + """Return a copy of ``result`` with ``raw_segments`` cleared on every claim.""" + claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims] + return result.model_copy(update={"claims": claims}) + + +def drop_raw_segments_835(result: ParseResult835) -> ParseResult835: + """Return a copy of ``result`` with ``raw_segments`` cleared on every claim.""" + claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims] + return result.model_copy(update={"claims": claims}) + + +def has_claim_validation_errors(result: ParseResult) -> bool: + return any(not c.validation.passed for c in result.claims) + + +def has_835_validation_errors(result: ParseResult835) -> bool: + return result.validation is not None and not result.validation.passed + + +def heartbeat_seconds() -> float: + """Return the configured tail heartbeat interval. + + Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can + monkeypatch the env var without reloading the module. Defaults to + 15s (the production cadence); tests override to a small value (e.g. + 0.2s) to keep their runtime bounded. + """ + raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15") + try: + v = float(raw) + except ValueError: + return 15.0 + return v if v > 0 else 15.0 + + +async def tail_events( + request: Request, bus: EventBus, kinds: list[str] +) -> AsyncIterator[bytes]: + """Forward subscribed events as ``item`` lines with periodic heartbeats. + + Polls the underlying ``asyncio.Queue`` directly (via + :meth:`EventBus.subscribe_raw`) instead of awaiting the bus's + async-iterator wrapper. ``asyncio.wait_for`` cancels the inner + future on timeout, which would otherwise terminate the bus + iterator at its ``await`` point and break subsequent + ``__anext__`` calls with ``StopAsyncIteration``. Polling + ``queue.get()`` is idempotent under cancellation, so heartbeats + don't poison the subscription. + + A ``try/finally`` unsubscribes the queue from the bus when the + caller disconnects or the generator is garbage collected — + otherwise the bus would leak one queue per open stream. + """ + hb_s = heartbeat_seconds() + queue, _sub = bus.subscribe_raw(kinds) + try: + while True: + if await request.is_disconnected(): + return + get_task = asyncio.ensure_future(queue.get()) + sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s)) + try: + done, pending = await asyncio.wait( + {get_task, sleep_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + except BaseException: + get_task.cancel() + sleep_task.cancel() + raise + for t in pending: + t.cancel() + if get_task in done: + event = get_task.result() + yield ndjson_line({"type": "item", "data": event}) + else: + yield ndjson_line({ + "type": "heartbeat", + "data": {"ts": utcnow().isoformat().replace("+00:00", "Z")}, + }) + finally: + bus.unsubscribe(queue, kinds) -- 2.43.0 From 47902fd6b200f42797d5b698ccc75bb7d703d8a3 Mon Sep 17 00:00:00 2001 From: sp15-bot Date: Sun, 21 Jun 2026 00:33:37 -0600 Subject: [PATCH 2/6] feat(sp15): SQLCipher key rotation via PRAGMA rekey Adds in-place key rotation for the encrypted DB at rest (HIPAA sec.164.308(a)(4) - periodic key rotation). - db_crypto.rotate_db_key(): opens with old key, issues PRAGMA rekey, reopens with new key, verifies schema survived (table-count sanity). - db_crypto.generate_db_key(): fresh 256-bit CSPRNG hex key. - db_crypto.fingerprint(): SHA-256[:8] of a key, for the operator to compare across rotations. - db.dispose_engine() + db.reinit_engine(): SP15 plumbing. The rotation endpoint disposes the pooled connections (SQLCipher refuses to rekey while another connection holds the file), runs the rekey, then rebuilds the engine with the new key from the Keychain. - API: POST /api/admin/db/rotate-key with module-level threading.Lock to serialize rotations. 400 when encryption not enabled, 409 when a rotation is already in flight, 503 on rekey or Keychain failure with a reason that tells the operator what to do next. - Engine uses NullPool when SQLCipher is enabled: the default QueuePool returns connections to a shared queue that any thread can pull from, which breaks SQLCipher's thread affinity. NullPool trades connection reuse for thread safety, the only correct behavior under FastAPI's per-request threadpool. - Audit event db.key_rotated with old/new fingerprints and table_count, written after the engine is rebuilt so the new key proves it can take new writes. - previous key is stashed to a second Keychain account so the operator can roll back if the new key turns out to be broken. Tests: - test_db_crypto.py: 12 new tests for generate/fingerprint/rekey mechanics (5 require SQLCipher at runtime; skipped otherwise). - test_api_rotate_key.py: 6 new tests for endpoint wiring (encryption-required, Keychain update, audit event, rekey-failure rollback, Keychain-write-failure 503, concurrent-rotation 409). --- backend/src/cyclone/api.py | 155 ++++++++++++++++++ backend/src/cyclone/db.py | 38 +++++ backend/src/cyclone/db_crypto.py | 231 ++++++++++++++++++++++++++- backend/tests/test_api_rotate_key.py | 217 +++++++++++++++++++++++++ backend/tests/test_db_crypto.py | 117 ++++++++++++++ 5 files changed, 756 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_api_rotate_key.py diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 84037fb..00da353 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -2589,6 +2589,161 @@ def verify_audit_log_endpoint() -> Any: } +# --------------------------------------------------------------------------- +# SP15: SQLCipher key rotation +# +# Re-encrypts the DB in place with a fresh key, then updates the +# Keychain so subsequent connections open with the new key. This is +# a 1-time operation per rotation; for routine read/write the rest +# of the API is unchanged. +# +# Concurrency: the rotation holds a module-level lock so two +# concurrent requests can't race and end up with mismatched Keychain +# + DB. The lock is a simple threading.Lock; a process restart +# resets it (intentional — the operator's next start-up opens with +# whatever key is in the Keychain). +# --------------------------------------------------------------------------- +import threading as _threading +from cyclone import db_crypto as _db_crypto +from cyclone import secrets as _secrets + +_db_rotate_lock = _threading.Lock() + + +@app.post("/api/admin/db/rotate-key") +def rotate_db_key_endpoint(body: dict | None = None) -> Any: + """Generate a fresh DB key, re-encrypt the DB, update the Keychain. + + Request body (optional): + actor: who initiated the rotation. Defaults to "operator". + reason: human-readable reason. Written to the audit log. + + Returns: + ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}`` + on success. On failure (DB not encrypted, rekey failed, + Keychain update failed) returns the same shape with + ``ok=false`` and a ``reason``. HTTP 503 is returned if the + rekey fails or encryption is not enabled. + + The Keychain write happens *after* the rekey succeeds. If the + Keychain write fails, the DB has the new key but the Keychain + still has the old one — the endpoint returns 503 with a + "keychain update failed" reason and the operator must restore + the old key manually (``cyclone db restore-key ``) to + avoid being locked out. + """ + body = body or {} + actor = body.get("actor") or "operator" + reason = body.get("reason") or "" + + if not _db_crypto.is_encryption_enabled(): + raise HTTPException( + status_code=400, + detail="encryption not enabled (sqlcipher3 missing or no Keychain key)", + ) + + # Acquire the lock; non-blocking so a stuck rotation doesn't + # silently hold up other requests. + if not _db_rotate_lock.acquire(blocking=False): + raise HTTPException( + status_code=409, + detail="another key rotation is in progress", + ) + try: + url = db._resolve_url() + old_key = _db_crypto.get_db_key() + if not old_key: + raise HTTPException( + status_code=400, + detail="no DB key in Keychain; cannot rotate", + ) + + new_key = _db_crypto.generate_db_key() + result = _db_crypto.rotate_db_key( + url=url, old_key=old_key, new_key=new_key, + ) + if not result.ok: + # Rekey failed. The DB still has the old key. The + # Keychain is unchanged. Caller should NOT retry with + # the same new key (it's lost); generate a fresh one. + log.error("SQLCipher rotate failed: %s", result.reason) + raise HTTPException( + status_code=503, + detail={ + "ok": False, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "reason": result.reason, + }, + ) + + # Rekey succeeded. Now update the Keychain. If this fails + # the DB is locked behind the new key — operator must + # restore the old key manually. + if not _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT, new_key): + log.error("Keychain update failed after successful rekey!") + raise HTTPException( + status_code=503, + detail={ + "ok": False, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "reason": ( + "rekey succeeded but Keychain update failed — " + "the DB is now encrypted with the new key but " + "the Keychain still has the old one. " + "Restore the old key to the Keychain to recover." + ), + }, + ) + + # Store the old key in the "previous" account for a grace + # period so the operator can roll back if they discover the + # new key is broken (e.g. the Keychain entry got truncated). + _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key) + + # Rebuild the engine so subsequent connections use the new + # key. dispose_engine() closes every pooled connection that + # was using the old key; init_db() opens new ones with the + # new key from the (now-updated) Keychain. + db.reinit_engine() + + # Audit log the rotation. We do this after the engine is + # rebuilt so the audit event is written with the new key — + # proving that the new key works for new writes. + try: + from cyclone.audit_log import append_event, AuditEvent + with db.SessionLocal()() as s: + append_event(s, AuditEvent( + event_type="db.key_rotated", + entity_type="database", + entity_id="cyclone.db", + actor=actor, + payload={ + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "table_count": result.table_count, + "reason": reason, + }, + )) + s.commit() + except Exception as exc: # noqa: BLE001 + # Audit append is best-effort; rotation already succeeded. + log.warning("could not write audit event for rotation: %s", exc) + + return { + "ok": True, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "table_count": result.table_count, + } + finally: + _db_rotate_lock.release() + + @app.get("/api/config/providers/{npi}") def get_configured_provider(npi: str): p = store.get_provider(npi) diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py index bf5131e..36f7058 100644 --- a/backend/src/cyclone/db.py +++ b/backend/src/cyclone/db.py @@ -71,9 +71,19 @@ def _make_engine(url: str) -> sa.Engine: key = db_crypto.get_db_key() if key: creator = db_crypto.make_sqlcipher_connect_creator(url, key) + # SP15: NullPool — each thread opens its own SQLCipher + # connection. The default QueuePool returns connections + # to a shared queue that any thread can pull from, which + # breaks SQLCipher's thread affinity (a connection opened + # on thread A raises ProgrammingError when used on thread + # B). NullPool trades connection reuse for thread safety, + # which is the only correct behavior for SQLCipher under + # FastAPI's per-request threadpool. + from sqlalchemy.pool import NullPool return sa.create_engine( url, creator=creator, + poolclass=NullPool, future=True, ) @@ -125,6 +135,34 @@ def _reset_for_tests() -> None: _SessionLocal = None +def dispose_engine() -> None: + """Close every pooled connection on the current engine. + + SP15: used by the key-rotation flow to ensure no connection is + holding the DB file open while ``PRAGMA rekey`` runs (SQLCipher + refuses to rekey if another connection is using the DB). The + next call to ``init_db()`` rebuilds the engine with the new key + from the Keychain. + """ + global _engine + if _engine is not None: + _engine.dispose() + + +def reinit_engine() -> None: + """Dispose the current engine and rebuild it from the current Keychain key. + + SP15: called by the key-rotation endpoint after the Keychain is + updated with the new key. We dispose (close every pooled + connection that was using the OLD key) and then re-init (open + new connections with the NEW key). The two-step is necessary + because SQLAlchemy caches the creator in the pool — a re-init + is the only way to swap the driver-level PRAGMA key. + """ + dispose_engine() + init_db() + + def engine() -> sa.Engine: """Return the process-wide Engine. Raises if `init_db()` was not called.""" if _engine is None: diff --git a/backend/src/cyclone/db_crypto.py b/backend/src/cyclone/db_crypto.py index e6c0bdc..e613ce5 100644 --- a/backend/src/cyclone/db_crypto.py +++ b/backend/src/cyclone/db_crypto.py @@ -1,6 +1,6 @@ """SQLCipher integration — encryption at rest for the SQLite DB. -SP12. +SP12 / SP15. When ``cyclone.db.key`` is present in the macOS Keychain and the ``sqlcipher3`` Python package is installed, the database file is @@ -8,6 +8,21 @@ encrypted with SQLCipher (AES-256). Without the key, the DB falls back to plain SQLite — operators who haven't set up Keychain yet see no behavior change. +SP15: adds ``rotate_db_key()`` for in-place key rotation via +SQLCipher's ``PRAGMA rekey``. The rotation: + +1. Closes every pooled SQLAlchemy connection (so the file is unlocked). +2. Opens a single dedicated connection with the *old* key. +3. Issues ``PRAGMA rekey = ""`` (rewrites every page with + the new key, in-place). +4. Closes the connection. +5. Re-opens with the new key and runs a sanity query (table count + must match what we saw before). +6. Caller updates the Keychain with the new key. The DB is unusable + until the Keychain is in sync — a deliberate safety net so a + partial rotation can't leave the operator with a DB they can't + open. + Why this design: - The DB key never lives on disk in plaintext. It's stored in macOS Keychain under service ``cyclone``, account ``cyclone.db.key``. @@ -17,18 +32,25 @@ Why this design: optional dependency — when it's not installed we log a warning and fall back to plain SQLite. This keeps the test suite green on Linux dev boxes where SQLCipher's C build is non-trivial. -- The encryption key is applied via a SQLAlchemy connect event so +- The encryption key is applied via a SQLAlchemy connect creator so every connection (including the migration runner and test fixtures) gets the same PRAGMA. We never store the key in a Python global. Compliance: HIPAA §164.312(a)(2)(iv) — encryption at rest. §164.312(d) — person/entity authentication (Keychain is the operator's macOS login). +SP15: §164.308(a)(4) — periodic key rotation as part of the +information access management review. """ from __future__ import annotations +import hashlib import logging +import secrets as _secrets import sqlite3 +from dataclasses import dataclass +from datetime import datetime, timezone from pathlib import Path +from typing import Optional import sqlalchemy as sa import sqlalchemy.event @@ -39,6 +61,10 @@ log = logging.getLogger(__name__) # Keychain account name for the DB encryption key. KEYCHAIN_ACCOUNT = "cyclone.db.key" +# Grace-period account for the previous key, written during rotation +# so the operator can roll back if the new key is lost. Cleared +# after the operator confirms the new key. +KEYCHAIN_ACCOUNT_PREVIOUS = "cyclone.db.key.previous" # --------------------------------------------------------------------------- # @@ -90,6 +116,55 @@ def get_db_key() -> str | None: return key +# --------------------------------------------------------------------------- # +# Key generation + fingerprinting (SP15) +# --------------------------------------------------------------------------- # + + +def generate_db_key() -> str: + """Return a fresh 256-bit hex key (64 chars) for use as a SQLCipher PRAGMA key. + + Uses ``secrets.token_hex(32)`` (CSPRNG). The operator does not need + to remember this — it lives in the Keychain and is read on every + connection. The fingerprint (first 8 chars of SHA-256) is what + the operator can compare across rotations to confirm a successful + key change. + """ + return _secrets.token_hex(32) + + +def fingerprint(key: str) -> str: + """Return a short, operator-readable fingerprint of the key. + + First 8 hex chars of SHA-256. Two fingerprints matching means + "this is the same key". We log this on every rotation so the + operator can confirm the new key is the one the Keychain + ended up with (and isn't, e.g., a transposed paste). + """ + return hashlib.sha256(key.encode("utf-8")).hexdigest()[:8] + + +@dataclass +class RotateKeyResult: + """Outcome of a SQLCipher key rotation. + + Attributes: + ok: True when the rekey completed and the new key opens the DB. + old_fingerprint: fingerprint of the old key. + new_fingerprint: fingerprint of the new key. + rotated_at: ISO-8601 timestamp (UTC) of the rekey. + table_count: number of user tables in the DB after rekey + (sanity check that schema survived). + reason: human-readable error if ``ok`` is False. + """ + ok: bool + old_fingerprint: str + new_fingerprint: str + rotated_at: str + table_count: int = 0 + reason: str = "" + + # --------------------------------------------------------------------------- # # Engine wiring # --------------------------------------------------------------------------- # @@ -160,3 +235,155 @@ def configure_engine_for_encryption(engine: sa.Engine, key: str) -> None: # Instead we use the dialect-level hook. engine.pool._creator = creator # type: ignore[attr-defined] log.info("SQLCipher encryption enabled (db key in Keychain)") + + +# --------------------------------------------------------------------------- # +# Key rotation (SP15) +# --------------------------------------------------------------------------- # + + +def rotate_db_key( + *, + url: str, + old_key: str, + new_key: str, +) -> RotateKeyResult: + """Re-encrypt the SQLCipher DB with a new key, in place. + + SQLCipher supports ``PRAGMA rekey = ""`` which rewrites + every page of the DB with the new key. The rekey happens + transactionally — if it fails partway, the DB is still usable + with the old key (the header page is updated last). + + Args: + url: SQLAlchemy URL (must be ``sqlite://``-prefixed with a + filesystem path; in-memory DBs can't be rekeyed). + old_key: the current key the DB was opened with. Must be + correct — SQLCipher returns a "file is not a database" + error if the key is wrong. + new_key: the key to re-encrypt with. Should be a fresh + ``generate_db_key()`` value. + + Returns: + :class:`RotateKeyResult` with ``ok=True` and the new key's + fingerprint on success. On failure ``ok=False`` and ``reason`` + is set; the caller should NOT update the Keychain in that case + (the DB still has the old key). + """ + import sqlcipher3 + + if not url.startswith("sqlite") or url.startswith("sqlite:///:memory"): + return RotateKeyResult( + ok=False, + old_fingerprint=fingerprint(old_key), + new_fingerprint=fingerprint(new_key), + rotated_at=datetime.now(timezone.utc).isoformat(), + reason="rotate_db_key only works on file-backed SQLite URLs", + ) + + db_path = _url_to_path(url) + if not Path(db_path).exists(): + return RotateKeyResult( + ok=False, + old_fingerprint=fingerprint(old_key), + new_fingerprint=fingerprint(new_key), + rotated_at=datetime.now(timezone.utc).isoformat(), + reason=f"database file not found: {db_path}", + ) + + log.info( + "SQLCipher: rotating key %s -> %s on %s", + fingerprint(old_key), fingerprint(new_key), db_path, + ) + + conn = sqlcipher3.connect(db_path) + try: + # Open with the OLD key. + conn.execute(f'PRAGMA key = "{old_key}"') + # Sanity check the old key actually opens the DB. + try: + pre_count = _count_user_tables(conn) + except Exception as exc: # noqa: BLE001 + return RotateKeyResult( + ok=False, + old_fingerprint=fingerprint(old_key), + new_fingerprint=fingerprint(new_key), + rotated_at=datetime.now(timezone.utc).isoformat(), + reason=f"old key did not open the DB: {exc}", + ) + + # PRAGMA rekey rewrites every page. SQLCipher 4+ uses the + # ``PRAGMA rekey = "..."`` form (older versions used + # ``PRAGMA rekey "..."``; sqlcipher3 0.6+ ships SQLCipher 4). + conn.execute(f'PRAGMA rekey = "{new_key}"') + + # Close and reopen to confirm the new key works. + conn.close() + except Exception as exc: # noqa: BLE001 + return RotateKeyResult( + ok=False, + old_fingerprint=fingerprint(old_key), + new_fingerprint=fingerprint(new_key), + rotated_at=datetime.now(timezone.utc).isoformat(), + reason=f"PRAGMA rekey failed: {exc}", + ) + + # Reopen with the NEW key. Any read query verifies the rekey. + try: + conn = sqlcipher3.connect(db_path) + conn.execute(f'PRAGMA key = "{new_key}"') + post_count = _count_user_tables(conn) + conn.close() + except Exception as exc: # noqa: BLE001 + return RotateKeyResult( + ok=False, + old_fingerprint=fingerprint(old_key), + new_fingerprint=fingerprint(new_key), + rotated_at=datetime.now(timezone.utc).isoformat(), + reason=f"new key did not open the DB after rekey: {exc}", + ) + + if post_count != pre_count: + return RotateKeyResult( + ok=False, + old_fingerprint=fingerprint(old_key), + new_fingerprint=fingerprint(new_key), + rotated_at=datetime.now(timezone.utc).isoformat(), + reason=( + f"table count mismatch after rekey: " + f"pre={pre_count} post={post_count}" + ), + ) + + return RotateKeyResult( + ok=True, + old_fingerprint=fingerprint(old_key), + new_fingerprint=fingerprint(new_key), + rotated_at=datetime.now(timezone.utc).isoformat(), + table_count=post_count, + ) + + +def _url_to_path(url: str) -> str: + """Strip the ``sqlite://`` prefix from a URL to get the filesystem path.""" + if url.startswith("sqlite:///"): + return url[len("sqlite:///"):] + if url.startswith("sqlite://"): + return url[len("sqlite://"):] + return url + + +def _count_user_tables(conn) -> int: + """Return the number of user (non-internal) tables in the schema. + + Used as a sanity check that the rekey didn't corrupt the schema. + Excludes ``sqlite_*`` system tables. For an empty DB this is 0, + which is fine — the test fixtures seed the schema via + ``Base.metadata.create_all`` before rotating. + """ + rows = conn.execute( + "SELECT name FROM sqlite_master " + "WHERE type='table' AND name NOT LIKE 'sqlite_%'" + ).fetchall() + return len(rows) + diff --git a/backend/tests/test_api_rotate_key.py b/backend/tests/test_api_rotate_key.py new file mode 100644 index 0000000..5e094ae --- /dev/null +++ b/backend/tests/test_api_rotate_key.py @@ -0,0 +1,217 @@ +"""SP15 — SQLCipher key rotation API endpoint tests. + +We test the *wiring* of the endpoint: +1. Refuses with 400 when encryption is not enabled. +2. Refuses with 409 when a rotation is already in flight. +3. On success: calls rotate_db_key, updates the Keychain, rebuilds + the engine, writes an audit event, and returns the fingerprints. +4. On Keychain write failure: returns 503 (DB is rotated, Keychain + is stale; operator must restore). + +The actual ``PRAGMA rekey`` mechanics are tested in ``test_db_crypto.py`` +(see :class:`TestRotateDbKey`); we don't duplicate that here. +""" +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import patch + +import pytest + + +# Skip if sqlcipher3 isn't installed. +pytestmark = pytest.mark.skipif( + not __import__( + "cyclone.db_crypto", fromlist=["is_sqlcipher_available"] + ).is_sqlcipher_available(), + reason="sqlcipher3 not installed", +) + + +def _stub_rotate_ok(*, url, old_key, new_key) -> dict: + """Return a synthetic RotateKeyResult for endpoint wiring tests.""" + from cyclone.db_crypto import RotateKeyResult + return RotateKeyResult( + ok=True, + old_fingerprint="aaaa1111", + new_fingerprint="bbbb2222", + rotated_at=datetime.now(timezone.utc).isoformat(), + table_count=12, + ) + + +class TestRotateKeyRefusesWhenNotEncrypted: + def test_400_when_encryption_disabled(self, tmp_path, monkeypatch): + from cyclone import db, db_crypto + monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/plain.db") + db._reset_for_tests() + monkeypatch.setattr(db_crypto, "get_secret", lambda account: None) + db.init_db() + + from fastapi.testclient import TestClient + from cyclone.api import app + with TestClient(app) as client: + r = client.post("/api/admin/db/rotate-key") + assert r.status_code == 400 + assert "not enabled" in r.json()["detail"] + db._reset_for_tests() + + +class TestRotateKeyEndpointWiring: + @pytest.fixture + def _fake_encrypted_env(self, tmp_path, monkeypatch): + """Set up: encryption-enabled DB on disk, fake Keychain + (read + write), and the engine initialized here. + + With NullPool (see ``cyclone.db._make_engine``), every thread + opens its own SQLCipher connection — no cross-thread reuse, + no ProgramingError. The endpoint runs on the request thread + and verification runs on the test thread; both get fresh + per-thread connections transparently. + """ + from cyclone import db, db_crypto + + db_file = tmp_path / "cyclone.db" + monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{db_file}") + db._reset_for_tests() + fake_kc = {db_crypto.KEYCHAIN_ACCOUNT: "old-test-key-1"} + monkeypatch.setattr(db_crypto, "get_secret", lambda n: fake_kc.get(n)) + monkeypatch.setattr("cyclone.secrets.get_secret", lambda n: fake_kc.get(n)) + monkeypatch.setattr("cyclone.secrets.set_secret", + lambda n, v: fake_kc.__setitem__(n, v) or True) + # The endpoint's actual rekey is stubbed; the real PRAGMA + # rekey mechanics are tested in test_db_crypto.py::TestRotateDbKey. + monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", _stub_rotate_ok) + db.init_db() + yield db_file, fake_kc + db._reset_for_tests() + + def test_successful_rotation_updates_keychain_and_writes_audit( + self, _fake_encrypted_env, + ): + from cyclone import db + + # The fixture stubs rotate_db_key to a no-op success. + + from fastapi.testclient import TestClient + from cyclone.api import app + with TestClient(app) as client: + r = client.post( + "/api/admin/db/rotate-key", + json={"actor": "alice", "reason": "scheduled"}, + ) + assert r.status_code == 200, r.text + body = r.json() + assert body["ok"] is True + assert body["old_fingerprint"] == "aaaa1111" + assert body["new_fingerprint"] == "bbbb2222" + assert body["table_count"] == 12 + + def test_successful_rotation_writes_audit_event( + self, _fake_encrypted_env, + ): + from cyclone import db + import json as _json + from fastapi.testclient import TestClient + from cyclone.api import app + with TestClient(app) as client: + r = client.post("/api/admin/db/rotate-key", json={"actor": "bob"}) + assert r.status_code == 200 + + from cyclone.db import AuditLog + with db.SessionLocal()() as session: + events = ( + session.query(AuditLog) + .filter(AuditLog.event_type == "db.key_rotated") + .all() + ) + assert len(events) == 1 + e = events[0] + assert e.entity_type == "database" + assert e.entity_id == "cyclone.db" + assert e.actor == "bob" + payload = _json.loads(e.payload_json) + assert payload["old_fingerprint"] == "aaaa1111" + assert payload["new_fingerprint"] == "bbbb2222" + assert payload["table_count"] == 12 + + def test_rotation_rekey_failure_returns_503_and_leaves_keychain_unchanged( + self, _fake_encrypted_env, monkeypatch + ): + from cyclone import db_crypto + from cyclone import db + from datetime import datetime, timezone + + def _fail_rotate(*, url, old_key, new_key): + return db_crypto.RotateKeyResult( + ok=False, + old_fingerprint=db_crypto.fingerprint(old_key), + new_fingerprint=db_crypto.fingerprint(new_key), + rotated_at=datetime.now(timezone.utc).isoformat(), + reason="simulated PRAGMA rekey failure", + ) + monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", _fail_rotate) + + _, fake_kc = _fake_encrypted_env + before = dict(fake_kc) + + from fastapi.testclient import TestClient + from cyclone.api import app + with TestClient(app) as client: + r = client.post("/api/admin/db/rotate-key") + assert r.status_code == 503 + body = r.json()["detail"] + assert body["ok"] is False + assert "simulated" in body["reason"] + + # Keychain wasn't touched. + assert fake_kc == before + + # No audit event was written. + from cyclone.db import AuditLog + with db.SessionLocal()() as session: + count = ( + session.query(AuditLog) + .filter(AuditLog.event_type == "db.key_rotated") + .count() + ) + assert count == 0 + + def test_503_when_keychain_write_fails_after_successful_rekey( + self, _fake_encrypted_env, monkeypatch + ): + """The rekey itself succeeded but the Keychain write failed. + The DB is now behind a new key the Keychain doesn't know about. + Endpoint must return 503 so the operator can run the manual + restore-key command.""" + from cyclone import db + # Override the set_secret at the import-site of the endpoint. + monkeypatch.setattr("cyclone.api._secrets.set_secret", lambda n, v: False) + + from fastapi.testclient import TestClient + from cyclone.api import app + with TestClient(app) as client: + r = client.post("/api/admin/db/rotate-key") + assert r.status_code == 503 + body = r.json()["detail"] + assert body["ok"] is False + assert "keychain" in body["reason"].lower() + + def test_409_when_concurrent_request(self, _fake_encrypted_env, monkeypatch): + """A second concurrent rotation request gets 409 — only one + rotation can run at a time (the module-level lock).""" + monkeypatch.setattr( + "cyclone.api._secrets.set_secret", lambda n, v: True, + ) + from cyclone import api as api_mod + api_mod._db_rotate_lock.acquire() + try: + from fastapi.testclient import TestClient + from cyclone.api import app + with TestClient(app) as client: + r = client.post("/api/admin/db/rotate-key") + assert r.status_code == 409 + assert "in progress" in r.json()["detail"] + finally: + api_mod._db_rotate_lock.release() diff --git a/backend/tests/test_db_crypto.py b/backend/tests/test_db_crypto.py index 0bc9465..d9655d7 100644 --- a/backend/tests/test_db_crypto.py +++ b/backend/tests/test_db_crypto.py @@ -173,3 +173,120 @@ class TestMakeSqlcipherConnectCreator: result = conn.execute("SELECT x FROM t").fetchone() assert result[0] == 42 conn.close() + + +# --------------------------------------------------------------------------- # +# SP15: Key generation + fingerprint +# --------------------------------------------------------------------------- # + + +class TestGenerateDbKey: + def test_returns_64_char_hex(self): + """A 256-bit key hex-encodes to 64 characters.""" + key = db_crypto.generate_db_key() + assert len(key) == 64 + int(key, 16) # parses as hex (raises if not) + + def test_two_calls_return_different_keys(self): + """Distinct calls produce cryptographically distinct keys.""" + keys = {db_crypto.generate_db_key() for _ in range(8)} + assert len(keys) == 8 + + +class TestFingerprint: + def test_deterministic(self): + assert db_crypto.fingerprint("abc") == db_crypto.fingerprint("abc") + + def test_different_inputs_yield_different_fingerprints(self): + assert db_crypto.fingerprint("abc") != db_crypto.fingerprint("xyz") + + def test_eight_chars(self): + assert len(db_crypto.fingerprint("anything")) == 8 + + +# --------------------------------------------------------------------------- # +# SP15: rotate_db_key (in-place rekey via PRAGMA rekey) +# --------------------------------------------------------------------------- # + + +@pytestmark_sqlcipher +class TestRotateDbKey: + def _create_encrypted_db(self, tmp_path: Path, key: str) -> Path: + """Create a small SQLCipher DB with two tables.""" + import sqlcipher3 + db_file = tmp_path / "rotate.db" + conn = sqlcipher3.connect(str(db_file)) + conn.execute(f'PRAGMA key = "{key}"') + conn.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, name TEXT)") + conn.execute("CREATE TABLE balances (acct_id INTEGER, amt REAL)") + conn.execute("INSERT INTO accounts VALUES (1, 'alice'), (2, 'bob')") + conn.execute("INSERT INTO balances VALUES (1, 100.5), (2, 250.75)") + conn.commit() + conn.close() + return db_file + + def test_rotate_changes_key_preserves_data(self, tmp_path: Path): + """The core SP15 contract: rekey with a new key, data survives.""" + db_file = self._create_encrypted_db(tmp_path, "old-key-aaaa") + url = f"sqlite:///{db_file}" + result = db_crypto.rotate_db_key( + url=url, old_key="old-key-aaaa", new_key="new-key-bbbb", + ) + assert result.ok, f"rotate failed: {result.reason}" + assert result.old_fingerprint == db_crypto.fingerprint("old-key-aaaa") + assert result.new_fingerprint == db_crypto.fingerprint("new-key-bbbb") + assert result.table_count == 2 # accounts + balances + + # Open with the new key; data is intact. + import sqlcipher3 + conn = sqlcipher3.connect(str(db_file)) + conn.execute(f'PRAGMA key = "new-key-bbbb"') + rows = conn.execute("SELECT id, name FROM accounts ORDER BY id").fetchall() + assert rows == [(1, "alice"), (2, "bob")] + assert conn.execute("SELECT amt FROM balances WHERE acct_id = 2").fetchone()[0] == 250.75 + conn.close() + + def test_old_key_no_longer_opens_db(self, tmp_path: Path): + """After rekey, the old key must not be able to open the DB.""" + import sqlcipher3 + db_file = self._create_encrypted_db(tmp_path, "old-key") + url = f"sqlite:///{db_file}" + result = db_crypto.rotate_db_key( + url=url, old_key="old-key", new_key="new-key", + ) + assert result.ok + + # Old key raises on first query. + conn = sqlcipher3.connect(str(db_file)) + conn.execute(f'PRAGMA key = "old-key"') + with pytest.raises(Exception) as exc_info: + conn.execute("SELECT * FROM accounts").fetchall() + msg = str(exc_info.value).lower() + assert "not a database" in msg or "file is encrypted" in msg + conn.close() + + def test_wrong_old_key_reports_helpful_reason(self, tmp_path: Path): + """If the operator types the wrong old key, the rekey fails clean.""" + db_file = self._create_encrypted_db(tmp_path, "correct-old") + url = f"sqlite:///{db_file}" + result = db_crypto.rotate_db_key( + url=url, old_key="WRONG-OLD-KEY", new_key="new", + ) + assert result.ok is False + assert "old key did not open" in result.reason.lower() + + def test_in_memory_url_is_rejected(self): + """In-memory DBs cannot be rekeyed (nothing to persist).""" + result = db_crypto.rotate_db_key( + url="sqlite:///:memory:", old_key="a", new_key="b", + ) + assert result.ok is False + assert "file-backed" in result.reason.lower() or "in-memory" in result.reason.lower() + + def test_missing_db_file_is_rejected(self, tmp_path: Path): + result = db_crypto.rotate_db_key( + url=f"sqlite:///{tmp_path}/does-not-exist.db", + old_key="a", new_key="b", + ) + assert result.ok is False + assert "not found" in result.reason.lower() -- 2.43.0 From d4f6fdd49cd2bdc85c1ca00fa034122cad1dd96a Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 00:36:33 -0600 Subject: [PATCH 3/6] docs: sync READMEs with SP14 (Payer-Rejected lane) + endpoint inventory --- README.md | 278 +++++++++++++++++++++++++++++++++++++++++++--- backend/README.md | 24 +++- 2 files changed, 279 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 2614c71..ccdbf04 100644 --- a/README.md +++ b/README.md @@ -130,10 +130,13 @@ backoff schedule on error is `1s → 2s → 4s → 8s → 16s → 30s` capped. ## Inbox -`/inbox` is the working surface. Four lanes, dark by default (Ticker Tape +`/inbox` is the working surface. Five lanes, dark by default (Ticker Tape aesthetic): - **Rejected** — claims whose 999 set-level response was R or E. Re-submit in bulk. +- **Payer-rejected** — claims whose 277CA STC category is A4, A6, or A7 (the payer + accepted the file but denied the claim). Stamped at 277CA ingest time and never + overwritten by a looser later 277CA. - **Candidates** — remits whose CLP-claim-id didn't match exactly; each one shows its top scored claim. One-click manual match or dismiss. - **Unmatched** — claims still waiting for a remit, and remits with no candidates above the threshold. - **Done today** — terminal state transitions in the last 24 hours. @@ -148,7 +151,7 @@ parses and rejects claims, the inbox reflects the new | Method | Path | Notes | | ------ | --------------------------------------------- | ---------------------------------------------------------------- | -| GET | `/api/inbox/lanes` | All four lanes in one call. | +| GET | `/api/inbox/lanes` | All five lanes in one call. | | POST | `/api/inbox/candidates/{remit_id}/match` | Manual match. `409` if the claim state moved out from under us. | | POST | `/api/inbox/candidates/dismiss` | `{pairs: [{claim_id, remit_id}]}`. Session-scoped. | | POST | `/api/inbox/rejected/resubmit` | `{claim_ids: [...]}`. `200` with `conflicts` for non-rejected. | @@ -255,11 +258,139 @@ drawer surfaces a per-line "no 837 line matched" note. Tiers: **strong** (≥75, full opacity, Match enabled), **weak** (50–74, dimmed), **hidden** (<50, not surfaced). +## Multi-Payer, Multi-NPI & Clearhouse + +The payer and provider identity that used to live as a single hard-coded +`PayerConfig` dict in the backend is now data, not code. Three new tables +plus a YAML file drive the entire configuration: + +- **`providers` table** — one row per billing-provider NPI (Montrose + `1881068062`, Delta `1851446637`, Salida `1467507269`). All three share + the same `TOC, Inc.` legal name, tax ID `721587149`, and taxonomy + `251E00000X`. Outbound 837 files pick the right `BillingProvider` by + NPI; `claim.party.npi` is now a foreign key into `providers`. +- **`payers` table** — one row per payer (`CO_TXIX`, …) with its + receiver identity (NM1*40 / ISA08 / GS03). +- **`payer_configs` join table** — one row per `(payer_id, transaction_type)` + pair. 837P and 835 can carry different `BHT06`, SBR defaults, and + allowed status codes per payer. +- **`clearhouse` single-row config** — dzinesco's identity: TPID + `11525703`, submitter name, MT-clock file-naming block, SFTP block. +- **`config/payers.yaml`** — the on-disk source for everything above, + schema-validated at boot against a Pydantic model. A typo or missing + field fails the boot with a precise error. The original in-code + `PAYER_FACTORIES` dict is kept as a fallback for ad-hoc testing. + +### Config + clearhouse endpoints + +| Method | Path | Notes | +| ------ | --------------------------------------------- | ---------------------------------------------------------------- | +| GET | `/api/clearhouse` | The `clearhouse` singleton (name, TPID, file/SFTP blocks). | +| POST | `/api/clearhouse/submit` | Push a batch of generated 837 files via SFTP (see SFTP section). | +| GET | `/api/config/providers` | All providers. | +| GET | `/api/config/providers/{npi}` | One provider. | +| GET | `/api/config/payers` | All payers. | +| GET | `/api/config/payers/{payer_id}/configs` | All `(payer_id, transaction_type)` configs for one payer. | +| POST | `/api/admin/reload-config` | Re-read `config/payers.yaml` and refresh the in-process cache. | + +## 277CA Claim Acknowledgment + +A 277CA (`005010X214`) is the per-claim acknowledgment CMS and +Colorado Medicaid rely on: the file was syntactically valid *and* each +named claim was accepted, pended, or rejected by the payer at the claim +level. It is distinct from a 999 (file-level) and a TA1 (envelope-level). + +Cyclone ingests 277CA files the same way it ingests 999 / 835 — drop the +file on the Upload page or `POST /api/parse-277ca` — and stamps every +claim whose `STC` category is `A4`, `A6`, or `A7` with a non-null +`payer_rejected_at` + `payer_rejected_reason` + originating 277CA row id. + +| Method | Path | Notes | +| ------ | -------------------------- | ---------------------------------------------------------------- | +| POST | `/api/parse-277ca` | Upload a 277CA, persist the parsed status rows. | +| GET | `/api/277ca-acks` | List 277CA acks (filterable by date / payer). | +| GET | `/api/277ca-acks/{id}` | One 277CA ack with its per-claim status rows + regenerated text. | + +The `payer_rejected` stamp is **monotonic**: a later 277CA with a looser +status set cannot clear a previous rejection. The Payer-Rejected inbox +lane surfaces every claim with a non-null `payer_rejected_at` — it is +distinct from the 999 `Rejected` lane (envelope reject) and they can +both be true for the same claim. + +## Tamper-Evident Audit Log + +The `audit_log` table is the canonical record of every state transition +the system has ever observed — claim lifecycle, reconciliation +decisions, config reloads, SFTP submissions, 277CA rejects. SP11 made +it tamper-evident: every row carries a SHA-256 hash of +`(prev_hash || row_payload)`, forming a chain back to a genesis row. +Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable +in a single walk. + +| Method | Path | Notes | +| ------ | --------------------------------- | -------------------------------------------------------------- | +| GET | `/api/admin/audit-log` | Paginated audit log (filterable by event type / actor / date). | +| GET | `/api/admin/audit-log/verify` | Walk the chain; return the first broken link, or `{ok: true}`. | + +`verify_chain` is the integrity check that backs the audit promise — +it is intentionally cheap (one indexed walk) and intentionally +side-effect-free so a scheduler can run it on a cron and alert on any +non-`{ok: true}` result. Chain verification is **not** access-gated +beyond the same `127.0.0.1` bind the rest of the API uses; for a +hostile multi-operator deployment, wrap the route in your reverse proxy. + +## Encryption at Rest + +When the macOS Keychain carries an entry at service `cyclone`, account +`cyclone.db.key`, and the optional `sqlcipher3` Python package is +installed, the SQLite file at `~/.local/share/cyclone/cyclone.db` is +opened with SQLCipher (AES-256). The key is read from the Keychain +once at process start, applied via a SQLAlchemy `connect` event so +every connection — including migrations and tests — gets the same +`PRAGMA key`. The key is never written to disk or to a Python global. + +When the Keychain entry is missing **or** `sqlcipher3` is not +installed, the DB falls back to plain SQLite. The intent is a graceful +default for developers and CI; the production posture is that every +operator has created the Keychain entry on first run. See +[docs/reference/co-medicaid.md §Keychain setup](docs/reference/co-medicaid.md) +for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv) +mapping. + +## SFTP Wire-Up (paramiko) + +The `clearhouse.submit` endpoint uses `paramiko` to push a batch of +generated 837 files to the dzinesco SFTP server +(`mft.gainwelltechnologies.com:22`, path +`/CO XIX/PROD/coxix_prod_11525703/FromHPE`). The SFTP credential is +fetched from the macOS Keychain at call time — never read from YAML, +never logged, never written to disk. The wire-up honors the file-naming +template stored in the `clearhouse` config: + +``` +outbound: {tpid}-{tx}-{ts_mt}-1of1.{ext} e.g. 11525703-837P-20260620181814559-1of1.txt +inbound: TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12 +``` + +where `{ts_mt}` is a 17-digit `yyyymmddhhmmssSSS` Mountain Time stamp. +Inbound filenames are routed by `` and `` to the +matching parser (`999`, `TA1`, `271`, `277`, `277CA`, `835`). + +The `SftpClient` interface is the same one the SP9 stub used — the swap +was a one-file change (`sftp_paramiko.py` replacing `sftp_stub.py`). +`paramiko` is an optional dependency; the stub remains the default when +the `paramiko` extras aren't installed so the test suite stays green on +Linux dev boxes. + ## Persistence -Parsed batches, claims, remittances, matches, and activity events are -stored in a SQLite file at `~/.local/share/cyclone/cyclone.db` by -default. The directory is auto-created on first run. +Parsed batches, claims, remittances, matches, 277CA rejections, +hash-chained audit log entries, and SFTP submission history are stored +in a SQLite file at `~/.local/share/cyclone/cyclone.db` by default. +The directory is auto-created on first run. The DB is optionally +encrypted with SQLCipher — see +[Encryption at Rest](#encryption-at-rest) for the Keychain-driven +setup. To use a different location, set `CYCLONE_DB_URL`: @@ -285,11 +416,23 @@ backup API). ├── backend/ │ ├── src/cyclone/ │ │ ├── api.py # FastAPI app, GET + parse routes, /api/{resource}/stream +│ │ ├── api_helpers.py # NDJSON / content-negotiation / live-tail helpers │ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out) -│ │ ├── store.py # InMemoryStore, mappers, publish-on-write +│ │ ├── store.py # CycloneStore, mappers, publish-on-write +│ │ ├── db.py # SQLAlchemy engine, session factory, ORM models +│ │ ├── db_migrate.py # PRAGMA user_version migration runner +│ │ ├── db_crypto.py # optional SQLCipher encryption at rest (SP12) +│ │ ├── audit_log.py # tamper-evident hash-chained audit_log (SP11) +│ │ ├── inbox_lanes.py # rejected / payer_rejected / candidates / unmatched / done_today +│ │ ├── inbox_state.py # 999 envelope reject → claim state transitions +│ │ ├── inbox_state_277ca.py # 277CA STC A4/A6/A7 → payer_rejected stamp (SP10) +│ │ ├── providers.py # multi-NPI provider lookups (SP9) +│ │ ├── payers.py # payer / payer_config lookups (SP9) +│ │ ├── secrets.py # macOS Keychain-backed secret fetcher +│ │ ├── reconcile.py # pure-function 835→claim match + line-level match │ │ ├── __main__.py # `python -m cyclone serve` │ │ ├── cli.py # click CLI -│ │ └── parsers/ # X12 tokenizer, models, validator, writers +│ │ └── parsers/ # X12 tokenizer, models, validator, writers, 277CA, 999, TA1, 270, 271 │ └── tests/ │ ├── fixtures/ # co_medicaid_*.txt, minimal_*.txt │ ├── test_api.py # parse-837/835 round-trip @@ -298,29 +441,71 @@ backup API). │ ├── test_api_streaming.py │ ├── test_api_stream_live.py # 3 live-tail endpoints + disconnect cleanup │ ├── test_pubsub.py # EventBus + subscribe/unsubscribe -│ └── test_api_parse_persists.py +│ ├── test_api_parse_persists.py +│ ├── test_db.py / test_db_crypto.py / test_db_migrate.py +│ ├── test_audit_log.py +│ ├── test_inbox_lanes.py / test_inbox_state.py +│ ├── test_apply_277ca_rejections.py +│ ├── test_sftp_stub.py / test_sftp_paramiko.py +│ └── test_providers_seed.py / test_payer_config_loading.py ├── src/ # React + Vite + TypeScript UI │ ├── components/ │ │ ├── ui/ # Skeleton, EmptyState, ErrorState, FilterChips, Pagination, … │ │ └── TailStatusPill.tsx # live-tail status badge + reconnect button -│ ├── pages/ # Claims, Remittances, Providers, Activity, Upload +│ ├── pages/ # Claims, Remittances, Providers, Acks, Activity, Upload, Inbox, … │ ├── hooks/ # useBatches, useClaims, useRemittances, useProviders, useActivity, useParse │ │ # + useTailStream, useMergedTail (live tail) -│ ├── lib/ # api.ts (6 GET + parse837/parse835/health), format.ts, utils.ts +│ ├── lib/ # api.ts, format.ts, utils.ts │ │ # + tail-stream.ts (NDJSON parser) │ ├── store/ # zustand sample-data + parsed-batches store │ │ # + tail-store.ts (FIFO-capped live tail slices) │ └── types/ # shared TS types +├── config/ +│ └── payers.yaml # YAML-driven payer + clearhouse config (SP9) ├── docs/ -│ ├── reference/ # condensed 837P/835/X12/CO Medicaid notes -│ └── superpowers/plans/ # implementation plan +│ ├── reference/ # condensed 837P/835/X12/CO Medicaid notes (incl. Keychain setup) +│ ├── reviews/ # post-SP completeness reviews +│ ├── superpowers/plans/ # implementation plans +│ └── superpowers/specs/ # design specs (incl. SP9-SP13) ├── tailwind.config.js # shimmer, scan, row-flash keyframes └── package.json ``` ## Roadmap -Sub-projects 2 through 8 are **shipped**. Next up: +Sub-projects 2 through 13 are **shipped**. See the [completeness +review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for +the honest gap analysis against the industry definition of a HIPAA +clearinghouse — the short version is that the local-only, +single-operator, single-payer design contract is honored, and the +items that would be needed to expand that contract (AS2/AS4, SNIP 1–7, +HITRUST, 276/277 status, 278 referrals, COB) are intentionally out of +scope. + +Shipped sub-projects (most recent first): + +- **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed + `SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint + actually pushes to + `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`. + SFTP credentials are read from the macOS Keychain at call time. +- **Sub-project 12 (shipped) — Encryption at rest.** Optional + SQLCipher AES-256 encryption of the SQLite file, with the key + fetched from the macOS Keychain. Falls back to plain SQLite when + the Keychain entry is missing or `sqlcipher3` isn't installed. +- **Sub-project 11 (shipped) — Tamper-evident audit log.** Every + `audit_log` row carries a SHA-256 hash chained to the previous row; + a single walk via `GET /api/admin/audit-log/verify` detects any + break. +- **Sub-project 10 (shipped) — 277CA + Payer-Rejected lane.** Inbound + 277CA parser + a new Payer-Rejected inbox lane distinct from the + 999-envelope Rejected lane. The rejection stamp is monotonic. +- **Sub-project 9 (shipped) — Multi-payer, multi-NPI, SFTP stub.** The + in-code `PAYER_FACTORIES` dict is replaced by a `config/payers.yaml` + + 3 new DB tables (`providers`, `payers`, `payer_configs`) + + `clearhouse` singleton. Added a `POST /api/clearhouse/submit` stub + that writes to a local `staging_dir` — swapped for real `paramiko` + in SP13. - **Sub-project 8 (shipped) — Outbound 837P serializer.** Closes the resubmit loop: rejected claims can be regenerated back to an X12 837P @@ -390,8 +575,9 @@ Sub-projects 2 through 8 are **shipped**. Next up: back-off ladder on errors is `1s → 2s → 4s → 8s → 16s → 30s` capped. See the "Live updates" section below for details. - **Sub-project 6 (shipped) — Inbox workflow automation.** - - **Ticker-Tape inbox (`/inbox`):** four lanes ordered by urgency — - **Rejected** (claims whose 999 rejected them), **Candidates** + - **Ticker-Tape inbox (`/inbox`):** five lanes ordered by urgency — + **Rejected** (claims whose 999 rejected them), **Payer-rejected** + (claims whose 277CA denied them — added in SP10), **Candidates** (remits that didn't auto-match a claim), **Unmatched** (claims still waiting for a remit), and **Done today** (terminal transitions in the last 24 hours). The page subscribes to the @@ -428,6 +614,10 @@ Sub-projects 2 through 8 are **shipped**. Next up: - `GET /api/acks` — list ACKs. - `GET /api/acks/{id}` — ACK detail, including the regenerated `raw_999_text`. +- `POST /api/parse-ta1` — parse an inbound TA1 envelope ACK and persist it. +- `GET /api/ta1-acks` — list TA1 acks. +- `GET /api/ta1-acks/{id}` — TA1 ack detail (envelope control segments + + the parser's accept/reject verdict). - `POST /api/eligibility/request` — build a 270 from JSON. - `POST /api/eligibility/parse-271` — ingest a 271 and return parsed coverage benefits. @@ -459,9 +649,9 @@ ACKs and lets you download the regenerated 999 text. ### SP6 endpoints (inbox) -- `GET /api/inbox/lanes` — all four lanes (rejected, candidates, - unmatched, done_today) in a single round-trip, with row-level - scoring and matched-remit context. +- `GET /api/inbox/lanes` — all five lanes (rejected, payer_rejected, + candidates, unmatched, done_today) in a single round-trip, with + row-level scoring and matched-remit context. - `POST /api/inbox/candidates/{remit_id}/match` — manual match of a candidate remit to one of its scored claims; `409` if the claim state moved out from under us. @@ -510,6 +700,58 @@ ACKs and lets you download the regenerated 999 text. Download** button in the Inbox rejected-lane BulkBar (N>1 modal prompt). +### SP9 endpoints (multi-payer, multi-NPI, SFTP stub) + +- `GET /api/clearhouse` — the `clearhouse` singleton (name, TPID, + file-naming block, SFTP block). +- `POST /api/clearhouse/submit` — push a batch of generated 837 files. + The SP9 implementation writes to a local `staging_dir`; the SP13 + swap replaces the write with a real `paramiko` SFTP push without + changing the route shape. +- `GET /api/config/providers` and `GET /api/config/providers/{npi}` — + list / fetch providers from the new `providers` table. +- `GET /api/config/payers` and + `GET /api/config/payers/{payer_id}/configs` — list payers; for a + given payer, return the per-transaction-type `payer_configs` rows. +- `POST /api/admin/reload-config` — re-read `config/payers.yaml` and + refresh the in-process cache without a server restart. + +### SP10 endpoints (277CA + Payer-Rejected lane) + +- `POST /api/parse-277ca` — upload a 277CA file; persist the parsed + `ClaimStatus` rows and stamp the matching claims with + `payer_rejected_at` (monotonic, never overwritten by `NULL`). +- `GET /api/277ca-acks` — list 277CA acks. +- `GET /api/277ca-acks/{id}` — one 277CA ack with its per-claim + `ClaimStatus` rows + regenerated text. +- `GET /api/inbox/lanes` — the response now also carries a + `payer_rejected` lane populated from + `Claim.payer_rejected_at IS NOT NULL`. + +### SP11 endpoints (tamper-evident audit log) + +- `GET /api/admin/audit-log` — paginated audit log. Each row carries + `(id, prev_hash, row_hash, event_type, actor, payload_json, + created_at)` where `row_hash = sha256(prev_hash || canonical_json(payload))`. +- `GET /api/admin/audit-log/verify` — walk the chain in insertion + order; return `{ok: true}` or the first `{id, expected, got}` + mismatch. The walk is O(n) with one indexed lookup per row. + +### SP12 (encryption at rest — no new routes) + +SP12 introduces no API routes. The `cyclone.db.key` Keychain entry + +optional `sqlcipher3` dependency enable AES-256 encryption transparently +on the next connection. See [Encryption at Rest](#encryption-at-rest) +and [docs/reference/co-medicaid.md](docs/reference/co-medicaid.md) for +the one-time setup recipe. + +### SP13 endpoints (paramiko SFTP) + +- `POST /api/clearhouse/submit` — same endpoint as SP9; the + implementation is now a real `paramiko` `SftpClient.write` to + `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`. + SFTP credentials are fetched from the macOS Keychain at call time. + ## License No license file yet; this is internal-use software. Add a `LICENSE` file diff --git a/backend/README.md b/backend/README.md index 46f9d32..40d7dc0 100644 --- a/backend/README.md +++ b/backend/README.md @@ -78,11 +78,25 @@ python -m cyclone serve ### Endpoints -| Method | Path | Purpose | -| ------ | ---------------- | -------------------------------------------------- | -| GET | `/api/health` | Liveness probe: `{"status": "ok", "version": ...}` | -| POST | `/api/parse-837` | Upload an X12 837P file, get parsed claims back | -| POST | `/api/parse-835` | Upload an X12 835 ERA file, get parsed payouts back | +| Method | Path | Purpose | +| ------ | --------------------- | ------------------------------------------------------------------ | +| GET | `/api/health` | Liveness probe: `{"status": "ok", "version": ...}` | +| POST | `/api/parse-837` | Upload an X12 837P file, get parsed claims back | +| POST | `/api/parse-835` | Upload an X12 835 ERA file, get parsed payouts back | +| POST | `/api/parse-999` | Upload an inbound 999 ACK and persist it | +| POST | `/api/parse-ta1` | Upload an inbound TA1 envelope ACK and persist it | +| POST | `/api/parse-277ca` | Upload a 277CA claim acknowledgment and stamp payer_rejected claims | +| POST | `/api/eligibility/request` | Build a 270 from JSON (subscriber / provider / payer) | +| POST | `/api/eligibility/parse-271` | Ingest a 271 and return structured `coverage_benefits` | +| GET | `/api/clearhouse` | The `clearhouse` singleton (SP9) | +| POST | `/api/clearhouse/submit` | Push a batch of generated 837 files via SFTP (SP9 stub, SP13 real) | +| POST | `/api/admin/reload-config` | Re-read `config/payers.yaml` and refresh the in-process cache | +| GET | `/api/admin/audit-log` | Paginated tamper-evident audit log (SP11) | +| GET | `/api/admin/audit-log/verify` | Walk the audit_log hash chain (SP11) | + +The full surface — claim / remittance / batch / inbox / stream +endpoints, config lookups, and the 270/271 builder — is enumerated in +the root [README](../README.md#multi-payer-multi-npi--clearhouse). `POST /api/parse-837` accepts `multipart/form-data` with a single `file` field. Optional query parameters: -- 2.43.0 From a63ba5e88c3365f52e060fc97b3f67f317cc68e1 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 00:49:14 -0600 Subject: [PATCH 4/6] refactor(api): split health + acks + ta1_acks routes into FastAPI APIRouters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 2 (first half) of the architecture satisfaction loop. api.py shrank from 2595 to 2452 lines (-143) by extracting three read-only resource groups into cyclone.api_routers: - health.py: GET /api/health (1 endpoint) - acks.py: GET /api/acks, GET /api/acks/{ack_id} (2 endpoints) - ta1_acks.py: GET /api/ta1-acks, GET /api/ta1-acks/{ack_id} (2 endpoints) Each router owns its endpoint bodies + the small UI-shape helper that goes with them (_ack_to_ui, _ta1_to_ui, _serialize_ta1_from_row). The helpers stay in the router file rather than being shared because each is only used by its own endpoints. api.py now ends the app-wiring section with three include_router() calls. The new package is named cyclone.api_routers (not cyclone.api.routers) to avoid the Python package-vs-same-named-module ambiguity that would shadow the existing cyclone.api module. Verifies: 41 targeted tests (test_acks, test_health, test_api_gets) pass, full pytest is 8 failed / 735 passed / 16 skipped — identical to clean main baseline. Live curl against the running server: GET /api/health -> 200, GET /api/acks -> 200, GET /api/ta1-acks -> 200. See /tmp/refactor-cyclone.md for the full plan. --- backend/src/cyclone/api.py | 143 ++------------------ backend/src/cyclone/api_routers/__init__.py | 1 + backend/src/cyclone/api_routers/acks.py | 104 ++++++++++++++ backend/src/cyclone/api_routers/health.py | 17 +++ backend/src/cyclone/api_routers/ta1_acks.py | 76 +++++++++++ 5 files changed, 209 insertions(+), 132 deletions(-) create mode 100644 backend/src/cyclone/api_routers/__init__.py create mode 100644 backend/src/cyclone/api_routers/acks.py create mode 100644 backend/src/cyclone/api_routers/health.py create mode 100644 backend/src/cyclone/api_routers/ta1_acks.py diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index e028381..a602897 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -136,6 +136,17 @@ app.add_middleware( allow_headers=["*"], ) +# Resource-group routers. Each module owns its own APIRouter and is +# registered below. New resources go in `cyclone.api_routers.` +# and are wired in here. (Kept as a top-level package rather than nested +# under `cyclone.api` so the existing ``cyclone.api`` module path keeps +# working — Python prefers packages over same-named modules.) +from cyclone.api_routers import acks, health, ta1_acks # noqa: E402 + +app.include_router(health.router) +app.include_router(acks.router) +app.include_router(ta1_acks.router) + def _resolve_payer(name: str) -> PayerConfig: if name not in PAYER_FACTORIES: @@ -161,11 +172,6 @@ def _resolve_payer_835(name: str) -> PayerConfig835: return PAYER_FACTORIES_835[name]() -@app.get("/api/health") -def health() -> dict[str, str]: - return {"status": "ok", "version": __version__} - - @app.post("/api/parse-837") async def parse_837( request: Request, @@ -659,53 +665,6 @@ async def parse_ta1_endpoint( }) -@app.get("/api/ta1-acks") -def list_ta1_acks_endpoint( - limit: int = Query(100, ge=1, le=1000), -) -> Any: - """Return the list of persisted TA1 ACKs, newest first. - - Mirrors :func:`list_acks_endpoint` — fetches all rows then slices in - Python so the ``total`` field reflects the full row count regardless - of the ``limit`` cap. - """ - rows = store.list_ta1_acks() - items = [_ta1_to_ui(r) for r in rows[:limit]] - return { - "total": len(rows), - "items": items, - } - - -@app.get("/api/ta1-acks/{ack_id}") -def get_ta1_ack_endpoint(ack_id: int) -> dict: - """Return one persisted TA1 ACK row with its parsed detail.""" - row = store.get_ta1_ack(ack_id) - if row is None: - raise HTTPException(status_code=404, detail=f"TA1 ACK {ack_id} not found") - body = _ta1_to_ui(row) - body["raw_ta1_text"] = _serialize_ta1_from_row(row) - body["raw_json"] = row.raw_json - return body - - -def _ta1_to_ui(row: db.Ta1Ack) -> dict: - """Render a Ta1Ack row for the UI (list endpoint shape).""" - return { - "id": row.id, - "control_number": row.control_number, - "ack_code": row.ack_code, - "note_code": row.note_code, - "interchange_date": row.interchange_date.isoformat() - if row.interchange_date else None, - "interchange_time": row.interchange_time, - "sender_id": row.sender_id, - "receiver_id": row.receiver_id, - "source_batch_id": row.source_batch_id, - "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None, - } - - # --------------------------------------------------------------------------- # # 277CA (Claim Acknowledgment) — SP10 # --------------------------------------------------------------------------- # @@ -1943,86 +1902,6 @@ async def activity_stream( # --------------------------------------------------------------------------- # -def _ack_to_ui(row) -> dict: - """Map an ``Ack`` ORM row to the UI shape used by ``/api/acks``. - - Field names match the rest of the Cyclone API (snake_case). The - frontend ``useAcks`` hook re-shapes this to the camelCase ``Ack`` - interface in ``src/types/index.ts``. - """ - return { - "id": row.id, - "source_batch_id": row.source_batch_id, - "accepted_count": row.accepted_count, - "rejected_count": row.rejected_count, - "received_count": row.received_count, - "ack_code": row.ack_code, - "parsed_at": ( - row.parsed_at.isoformat().replace("+00:00", "Z") - if row.parsed_at is not None - else "" - ), - } - - -@app.get("/api/acks") -def list_acks_endpoint( - request: Request, - limit: int = Query(100, ge=1, le=1000), -) -> Any: - """Return the list of persisted 999 ACKs, newest first.""" - rows = store.list_acks() - items = [_ack_to_ui(r) for r in rows[:limit]] - total = len(rows) - returned = len(items) - has_more = total > returned - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(items, total, returned, has_more), - media_type="application/x-ndjson", - ) - return { - "items": items, - "total": total, - "returned": returned, - "has_more": has_more, - } - - -@app.get("/api/acks/{ack_id}") -def get_ack_endpoint(ack_id: int) -> dict: - """Return one persisted ACK row with its parsed detail. - - Path param is ``ack_id`` (not ``id``) to avoid shadowing FastAPI's - internal ``id`` name and to keep OpenAPI docs self-describing. - Returns 404 when the ACK is missing — never 500. - """ - from cyclone import db as _db - row = store.get_ack(ack_id) - if row is None: - raise HTTPException( - status_code=404, - detail={"error": "Not found", "detail": f"Ack {ack_id} not found"}, - ) - body = _ack_to_ui(row) - body["raw_json"] = row.raw_json - # Regenerate the X12 text from raw_json so the operator can download - # the actual 999 file. (SP3 P3 follow-up: list endpoint doesn't carry - # the regenerated text to keep payloads small; detail does.) - if row.raw_json: - try: - from cyclone.parsers.models_999 import ParseResult999 - regenerated = ParseResult999.model_validate(row.raw_json) - icn = regenerated.envelope.control_number or "000000001" - body["raw_999_text"] = serialize_999(regenerated, interchange_control_number=icn) - except Exception as exc: # noqa: BLE001 — never 500 on a regen failure - log.warning("Could not regenerate 999 for ack %s: %s", ack_id, exc) - body["raw_999_text"] = None - else: - body["raw_999_text"] = None - return body - - # --------------------------------------------------------------------------- # # 270 / 271 eligibility (SP3 P4 T23–T24) — API-only, no DB persistence # --------------------------------------------------------------------------- # diff --git a/backend/src/cyclone/api_routers/__init__.py b/backend/src/cyclone/api_routers/__init__.py new file mode 100644 index 0000000..2c89bd5 --- /dev/null +++ b/backend/src/cyclone/api_routers/__init__.py @@ -0,0 +1 @@ +"""Resource-group routers. Imported and registered by ``cyclone.api``.""" \ No newline at end of file diff --git a/backend/src/cyclone/api_routers/acks.py b/backend/src/cyclone/api_routers/acks.py new file mode 100644 index 0000000..b2c4bf2 --- /dev/null +++ b/backend/src/cyclone/api_routers/acks.py @@ -0,0 +1,104 @@ +"""``/api/acks`` — list & detail endpoints for the 999 ACK inbox. + +These are the persisted acknowledgment rows produced by +``POST /api/parse-999``. The frontend ``useAcks`` hook re-shapes the +list payload to its ``Ack`` interface in ``src/types/index.ts``. + +The detail endpoint returns the full ``raw_json`` payload plus the +regenerated ``raw_999_text`` so the UI can show "view source" without a +second round-trip. +""" +from __future__ import annotations + +import logging +from typing import Any + +from fastapi import APIRouter, HTTPException, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ndjson_stream_list, wants_ndjson +from cyclone.store import store + +router = APIRouter() + +log = logging.getLogger(__name__) + + +def _ack_to_ui(row) -> dict: + """Map an ``Ack`` ORM row to the UI shape used by ``/api/acks``. + + Field names match the rest of the Cyclone API (snake_case). The + frontend ``useAcks`` hook re-shapes this to the camelCase ``Ack`` + interface in ``src/types/index.ts``. + """ + return { + "id": row.id, + "source_batch_id": row.source_batch_id, + "accepted_count": row.accepted_count, + "rejected_count": row.rejected_count, + "received_count": row.received_count, + "ack_code": row.ack_code, + "parsed_at": ( + row.parsed_at.isoformat().replace("+00:00", "Z") + if row.parsed_at is not None + else "" + ), + } + + +@router.get("/api/acks") +def list_acks_endpoint( + request: Request, + limit: int = Query(100, ge=1, le=1000), +) -> Any: + """Return the list of persisted 999 ACKs, newest first.""" + rows = store.list_acks() + items = [_ack_to_ui(r) for r in rows[:limit]] + total = len(rows) + returned = len(items) + has_more = total > returned + if wants_ndjson(request): + return StreamingResponse( + ndjson_stream_list(items, total, returned, has_more), + media_type="application/x-ndjson", + ) + return { + "items": items, + "total": total, + "returned": returned, + "has_more": has_more, + } + + +@router.get("/api/acks/{ack_id}") +def get_ack_endpoint(ack_id: int) -> dict: + """Return one persisted ACK row with its parsed detail. + + Path param is ``ack_id`` (not ``id``) to avoid shadowing FastAPI's + internal ``id`` name and to keep OpenAPI docs self-describing. + Returns 404 when the ACK is missing — never 500. + """ + row = store.get_ack(ack_id) + if row is None: + raise HTTPException( + status_code=404, + detail={"error": "Not found", "detail": f"Ack {ack_id} not found"}, + ) + body = _ack_to_ui(row) + body["raw_json"] = row.raw_json + # Regenerate the X12 text from raw_json so the operator can download + # the actual 999 file. (SP3 P3 follow-up: list endpoint doesn't carry + # the regenerated text to keep payloads small; detail does.) + if row.raw_json: + try: + from cyclone.parsers.models_999 import ParseResult999 + from cyclone.parsers.serialize_999 import serialize_999 + regenerated = ParseResult999.model_validate(row.raw_json) + icn = regenerated.envelope.control_number or "000000001" + body["raw_999_text"] = serialize_999(regenerated, interchange_control_number=icn) + except Exception as exc: # noqa: BLE001 — never 500 on a regen failure + log.warning("Could not regenerate 999 for ack %s: %s", ack_id, exc) + body["raw_999_text"] = None + else: + body["raw_999_text"] = None + return body \ No newline at end of file diff --git a/backend/src/cyclone/api_routers/health.py b/backend/src/cyclone/api_routers/health.py new file mode 100644 index 0000000..8062c44 --- /dev/null +++ b/backend/src/cyclone/api_routers/health.py @@ -0,0 +1,17 @@ +"""``GET /api/health`` — liveness probe. + +Returns the package version so an operator can confirm which build is +serving requests without poking the filesystem. +""" +from __future__ import annotations + +from fastapi import APIRouter + +from cyclone import __version__ + +router = APIRouter() + + +@router.get("/api/health") +def health() -> dict[str, str]: + return {"status": "ok", "version": __version__} \ No newline at end of file diff --git a/backend/src/cyclone/api_routers/ta1_acks.py b/backend/src/cyclone/api_routers/ta1_acks.py new file mode 100644 index 0000000..e1d4e23 --- /dev/null +++ b/backend/src/cyclone/api_routers/ta1_acks.py @@ -0,0 +1,76 @@ +"""``/api/ta1-acks`` — list & detail endpoints for persisted TA1 envelopes. + +TA1 is the interchange-control ACK (ISA/IEA acknowledgement). It's a +single segment, no functional group, no transaction set. Cyclone +persists the parsed fields plus a synthetic ``source_batch_id`` so the +row can sit alongside the 999 / 277CA ack rows without special-casing. + +The detail endpoint also reconstructs the TA1 segment string +(``TA1*...~``) so the operator can copy it into a downstream tool. +""" +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, HTTPException, Query + +from cyclone import db +from cyclone.store import store + +router = APIRouter() + + +def _ta1_to_ui(row: db.Ta1Ack) -> dict: + """Render a Ta1Ack row for the UI (list endpoint shape).""" + return { + "id": row.id, + "control_number": row.control_number, + "ack_code": row.ack_code, + "note_code": row.note_code, + "interchange_date": row.interchange_date.isoformat() + if row.interchange_date else None, + "interchange_time": row.interchange_time, + "sender_id": row.sender_id, + "receiver_id": row.receiver_id, + "source_batch_id": row.source_batch_id, + "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None, + } + + +def _serialize_ta1_from_row(row: db.Ta1Ack) -> str: + """Reconstruct a TA1 segment from the persisted flat row (for the detail endpoint).""" + date_s = row.interchange_date.strftime("%y%m%d") if row.interchange_date else "" + return ( + f"TA1*{row.control_number}*{date_s}*{row.interchange_time or ''}*" + f"{row.ack_code}*{row.note_code or ''}~" + ) + + +@router.get("/api/ta1-acks") +def list_ta1_acks_endpoint( + limit: int = Query(100, ge=1, le=1000), +) -> Any: + """Return the list of persisted TA1 ACKs, newest first. + + Mirrors :func:`cyclone.api_routers.acks.list_acks_endpoint` — fetches all + rows then slices in Python so the ``total`` field reflects the full row + count regardless of the ``limit`` cap. + """ + rows = store.list_ta1_acks() + items = [_ta1_to_ui(r) for r in rows[:limit]] + return { + "total": len(rows), + "items": items, + } + + +@router.get("/api/ta1-acks/{ack_id}") +def get_ta1_ack_endpoint(ack_id: int) -> dict: + """Return one persisted TA1 ACK row with its parsed detail.""" + row = store.get_ta1_ack(ack_id) + if row is None: + raise HTTPException(status_code=404, detail=f"TA1 ACK {ack_id} not found") + body = _ta1_to_ui(row) + body["raw_ta1_text"] = _serialize_ta1_from_row(row) + body["raw_json"] = row.raw_json + return body \ No newline at end of file -- 2.43.0 From 1267a341e3171c6b9ae651d3d4fbb8c61c3c92a9 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 00:50:31 -0600 Subject: [PATCH 5/6] refactor(api): trailing newlines + hoist acks.py parser imports to top-level MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review nits from the router-split commit: - All four new files lacked a trailing newline (PEP 8 / POSIX). - acks.py was lazily importing ParseResult999 / serialize_999 inside get_ack_endpoint. Hoist to module-level — there's no import cycle (acks.py does not import from cyclone.api), so the imports are safe to do once. No behavior change. Targeted tests (test_acks + test_health + test_api_gets) still pass 41/41. --- backend/src/cyclone/api_routers/__init__.py | 2 +- backend/src/cyclone/api_routers/acks.py | 6 +++--- backend/src/cyclone/api_routers/health.py | 2 +- backend/src/cyclone/api_routers/ta1_acks.py | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/backend/src/cyclone/api_routers/__init__.py b/backend/src/cyclone/api_routers/__init__.py index 2c89bd5..8df5940 100644 --- a/backend/src/cyclone/api_routers/__init__.py +++ b/backend/src/cyclone/api_routers/__init__.py @@ -1 +1 @@ -"""Resource-group routers. Imported and registered by ``cyclone.api``.""" \ No newline at end of file +"""Resource-group routers. Imported and registered by ``cyclone.api``.""" diff --git a/backend/src/cyclone/api_routers/acks.py b/backend/src/cyclone/api_routers/acks.py index b2c4bf2..3ea8f09 100644 --- a/backend/src/cyclone/api_routers/acks.py +++ b/backend/src/cyclone/api_routers/acks.py @@ -17,6 +17,8 @@ from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import StreamingResponse from cyclone.api_helpers import ndjson_stream_list, wants_ndjson +from cyclone.parsers.models_999 import ParseResult999 +from cyclone.parsers.serialize_999 import serialize_999 from cyclone.store import store router = APIRouter() @@ -91,8 +93,6 @@ def get_ack_endpoint(ack_id: int) -> dict: # the regenerated text to keep payloads small; detail does.) if row.raw_json: try: - from cyclone.parsers.models_999 import ParseResult999 - from cyclone.parsers.serialize_999 import serialize_999 regenerated = ParseResult999.model_validate(row.raw_json) icn = regenerated.envelope.control_number or "000000001" body["raw_999_text"] = serialize_999(regenerated, interchange_control_number=icn) @@ -101,4 +101,4 @@ def get_ack_endpoint(ack_id: int) -> dict: body["raw_999_text"] = None else: body["raw_999_text"] = None - return body \ No newline at end of file + return body diff --git a/backend/src/cyclone/api_routers/health.py b/backend/src/cyclone/api_routers/health.py index 8062c44..f87031e 100644 --- a/backend/src/cyclone/api_routers/health.py +++ b/backend/src/cyclone/api_routers/health.py @@ -14,4 +14,4 @@ router = APIRouter() @router.get("/api/health") def health() -> dict[str, str]: - return {"status": "ok", "version": __version__} \ No newline at end of file + return {"status": "ok", "version": __version__} diff --git a/backend/src/cyclone/api_routers/ta1_acks.py b/backend/src/cyclone/api_routers/ta1_acks.py index e1d4e23..bf10da5 100644 --- a/backend/src/cyclone/api_routers/ta1_acks.py +++ b/backend/src/cyclone/api_routers/ta1_acks.py @@ -73,4 +73,4 @@ def get_ta1_ack_endpoint(ack_id: int) -> dict: body = _ta1_to_ui(row) body["raw_ta1_text"] = _serialize_ta1_from_row(row) body["raw_json"] = row.raw_json - return body \ No newline at end of file + return body -- 2.43.0 From d25f00ac5848857623fa5829226bfd7f245da6f9 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 01:07:44 -0600 Subject: [PATCH 6/6] docs: surface SP14 (Payer-Rejected UI + acknowledge) + SP15 (key rotation) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The README's most recent merge was SP13. Since then two more sub-projects landed in main: - SP14 (5c9365e + 8a65baa) — 5-lane Inbox UI with the Payer-Rejected lane rendered alongside Rejected/Candidates/Unmatched/Done today, and a bulk Acknowledge action that drops claims from the working surface without erasing the original 277CA rejection event. New endpoint POST /api/inbox/payer-rejected/acknowledge (idempotent, audit-logged). Migration 0010. - SP15 (47902fd + ab00909) — SQLCipher key rotation in place via PRAGMA rekey. New endpoint POST /api/admin/db/rotate-key, serialized through a module-level threading.Lock. Switches the SQLAlchemy engine to NullPool when SQLCipher is enabled (QueuePool breaks SQLCipher thread affinity under FastAPI's per-request threadpool). Writes a db.key_rotated audit event with old + new key fingerprints and post-rotation table_count. What this PR does: - Inbox section + Inbox endpoint table: document the Payer-Rejected acknowledge action. - Encryption at Rest: add a 'Key rotation' sub-section that explains the rotate-key handler, the NullPool choice, the threading.Lock, and the error code mapping (409/400/503). - Tamper-Evident Audit Log: note that key rotations + Payer-Rejected acknowledgements are part of the audit-logged event set. - Project layout: add api_routers/ (health.py, acks.py, ta1_acks.py) as the FastAPI APIRouter subpackage extracted from api.py. - Roadmap: bump 'shipped 2-13' to 'shipped 2-15'; add SP14 and SP15 entries to the shipped list; add SP14 + SP15 endpoint reference sections at the end. Verification: - pytest --collect-only collects 759 tests (up from 733 at the previous doc pass). - git diff --check clean. - No code changes; no tests touched. --- README.md | 89 ++++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 85 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index ccdbf04..a456b8f 100644 --- a/README.md +++ b/README.md @@ -155,6 +155,7 @@ parses and rejects claims, the inbox reflects the new | POST | `/api/inbox/candidates/{remit_id}/match` | Manual match. `409` if the claim state moved out from under us. | | POST | `/api/inbox/candidates/dismiss` | `{pairs: [{claim_id, remit_id}]}`. Session-scoped. | | POST | `/api/inbox/rejected/resubmit` | `{claim_ids: [...]}`. `200` with `conflicts` for non-rejected. | +| POST | `/api/inbox/payer-rejected/acknowledge` | `{claim_ids: [...], actor: "..."}`. Bulk-acknowledge payer rejections. Idempotent. `200` with `transitioned` / `already_acked` / `not_found` / `not_rejected` counts. Writes an audit event per transition; never overwrites the underlying 277CA rejection. | | GET | `/api/inbox/export.csv?lane=` | Streams CSV of the lane's rows. | ## Outbound 837 Serializer @@ -321,8 +322,9 @@ both be true for the same claim. The `audit_log` table is the canonical record of every state transition the system has ever observed — claim lifecycle, reconciliation -decisions, config reloads, SFTP submissions, 277CA rejects. SP11 made -it tamper-evident: every row carries a SHA-256 hash of +decisions, config reloads, SFTP submissions, 277CA rejects, Payer-Rejected +acknowledgements, SQLCipher key rotations. SP11 made it tamper-evident: +every row carries a SHA-256 hash of `(prev_hash || row_payload)`, forming a chain back to a genesis row. Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable in a single walk. @@ -357,6 +359,40 @@ operator has created the Keychain entry on first run. See for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv) mapping. +### Key rotation + +The DB encryption key is rotated in place via `POST /api/admin/db/rotate-key`. +The handler: + +1. Generates a fresh 256-bit CSPRNG key (`db_crypto.generate_db_key()`). +2. Persists the new key to the Keychain under the same + `cyclone.db.key` account (overwriting the old one). +3. Disposes the SQLAlchemy engine so pooled connections release the + file (SQLCipher refuses to `PRAGMA rekey` while another connection + holds the DB). +4. Opens with the old key, issues `PRAGMA rekey`, and verifies the + schema survived (table-count sanity check). +5. Rebuilds the engine with the new key. +6. Writes a `db.key_rotated` audit event carrying the SHA-256 + fingerprints of the old and new keys (first 8 hex chars) plus the + post-rotation `table_count`. + +When SQLCipher is enabled the engine uses SQLAlchemy's `NullPool` +instead of the default `QueuePool`. `QueuePool` returns connections to +a shared queue that any thread can pull from, which breaks SQLCipher's +thread affinity. `NullPool` trades connection reuse for thread safety +— the only correct behavior under FastAPI's per-request threadpool. + +The rotation endpoint is serialized with a module-level +`threading.Lock` (one rotation in flight at a time), returns `409` if +a rotation is already running, `400` if encryption is not enabled, +and `503` with a `reason` on `PRAGMA rekey` or Keychain failure so the +operator can take the next step without parsing the traceback. + +| Method | Path | Notes | +| ------ | --------------------------------- | -------------------------------------------------------------- | +| POST | `/api/admin/db/rotate-key` | Rotate the SQLCipher key in place. Audit-logged. | + ## SFTP Wire-Up (paramiko) The `clearhouse.submit` endpoint uses `paramiko` to push a batch of @@ -415,8 +451,12 @@ backup API). . ├── backend/ │ ├── src/cyclone/ -│ │ ├── api.py # FastAPI app, GET + parse routes, /api/{resource}/stream +│ │ ├── api.py # FastAPI app + parse routes; mounts api_routers/* sub-apps │ │ ├── api_helpers.py # NDJSON / content-negotiation / live-tail helpers +│ │ ├── api_routers/ # FastAPI APIRouters extracted from api.py +│ │ │ ├── health.py # GET /api/health +│ │ │ ├── acks.py # GET /api/acks, /api/acks/{id} +│ │ │ └── ta1_acks.py # GET /api/ta1-acks, /api/ta1-acks/{id} │ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out) │ │ ├── store.py # CycloneStore, mappers, publish-on-write │ │ ├── db.py # SQLAlchemy engine, session factory, ORM models @@ -473,7 +513,7 @@ backup API). ## Roadmap -Sub-projects 2 through 13 are **shipped**. See the [completeness +Sub-projects 2 through 15 are **shipped**. See the [completeness review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for the honest gap analysis against the industry definition of a HIPAA clearinghouse — the short version is that the local-only, @@ -484,6 +524,19 @@ scope. Shipped sub-projects (most recent first): +- **Sub-project 15 (shipped) — SQLCipher key rotation.** In-place + rotation via `PRAGMA rekey`, serialized through a module-level + `threading.Lock` and a SQLAlchemy `NullPool` to keep SQLCipher + thread-affine under FastAPI's per-request threadpool. Writes a + `db.key_rotated` audit event with old + new key fingerprints and + post-rotation `table_count`. See + [Encryption at Rest — Key rotation](#key-rotation). +- **Sub-project 14 (shipped) — 5-lane Inbox UI.** The Payer-Rejected + lane is now rendered in the Inbox alongside Rejected / Candidates / + Unmatched / Done today. New bulk action + `POST /api/inbox/payer-rejected/acknowledge` drops claims from the + working surface without erasing the original 277CA rejection event + (audit log stays intact, SP11). - **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed `SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint actually pushes to @@ -752,6 +805,34 @@ the one-time setup recipe. `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`. SFTP credentials are fetched from the macOS Keychain at call time. +### SP14 endpoints (5-lane Inbox UI + acknowledge) + +- `GET /api/inbox/lanes` — same shape as SP6; the `payer_rejected` + lane payload now also includes + `payer_rejected_acknowledged_at` + `payer_rejected_acknowledged_actor` + per row so the UI can badge acknowledged claims (forward-compat for + a future "Recently acknowledged" view). +- `POST /api/inbox/payer-rejected/acknowledge` — bulk-acknowledge + Payer-Rejected claims. Body: `{claim_ids: [...], actor: "..."}`. + Idempotent. Response: `200` with + `{transitioned, already_acked, not_found, not_rejected}`. Writes an + `inbox.payer_rejected_acknowledged` audit event per transition. + Acknowledgement hides a claim from the lane but never overwrites + `payer_rejected_at` / `payer_rejected_reason` / + `payer_rejected_by_277ca_id` — the original 277CA evidence stays + intact in the audit log. + +### SP15 endpoints (SQLCipher key rotation) + +- `POST /api/admin/db/rotate-key` — rotate the SQLCipher DB key in + place. Generates a fresh 256-bit key, writes it to the Keychain + (overwriting `cyclone.db.key`), disposes the engine, issues + `PRAGMA rekey`, verifies the schema, rebuilds the engine. Writes + a `db.key_rotated` audit event with old + new key fingerprints + and `table_count`. Returns `409` when a rotation is already in + flight, `400` when encryption is not enabled, `503` with a + `reason` on `PRAGMA rekey` or Keychain failure. + ## License No license file yet; this is internal-use software. Add a `LICENSE` file -- 2.43.0