diff --git a/README.md b/README.md index 2614c71..a456b8f 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,10 +151,11 @@ 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. | +| 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 @@ -255,11 +259,174 @@ 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, 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. + +| 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. + +### 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 +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`: @@ -284,12 +451,28 @@ 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 # 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 +481,84 @@ 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 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, +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 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 + `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 +628,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 +667,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 +702,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 +753,86 @@ 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. + +### 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 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: diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 84037fb..a602897 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 @@ -130,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: @@ -155,96 +172,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__} - - @app.post("/api/parse-837") async def parse_837( request: Request, @@ -362,48 +289,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 +420,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) # --------------------------------------------------------------------------- # @@ -787,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 # --------------------------------------------------------------------------- # @@ -1105,7 +936,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 +947,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 +1105,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 +1246,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, @@ -2136,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 # --------------------------------------------------------------------------- # @@ -2589,6 +2275,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/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) diff --git a/backend/src/cyclone/api_routers/__init__.py b/backend/src/cyclone/api_routers/__init__.py new file mode 100644 index 0000000..8df5940 --- /dev/null +++ b/backend/src/cyclone/api_routers/__init__.py @@ -0,0 +1 @@ +"""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 new file mode 100644 index 0000000..3ea8f09 --- /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.parsers.models_999 import ParseResult999 +from cyclone.parsers.serialize_999 import serialize_999 +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: + 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 diff --git a/backend/src/cyclone/api_routers/health.py b/backend/src/cyclone/api_routers/health.py new file mode 100644 index 0000000..f87031e --- /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__} 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..bf10da5 --- /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 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()