Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Tyler
|
d25f00ac58 | ||
|
Tyler
|
1267a341e3 | ||
|
Tyler
|
a63ba5e88c | ||
|
Tyler
|
d4f6fdd49c | ||
|
Tyler
|
ab00909715 | ||
 sp15-bot
|
47902fd6b2 | ||
|
Tyler
|
6233df1270 |
@@ -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=<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 `<FileType>` and `<OrigTx>` 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
|
||||
|
||||
+19
-5
@@ -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:
|
||||
|
||||
+187
-346
@@ -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.<name>`
|
||||
# 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 <old_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)
|
||||
|
||||
@@ -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/<resource>/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)
|
||||
@@ -0,0 +1 @@
|
||||
"""Resource-group routers. Imported and registered by ``cyclone.api``."""
|
||||
@@ -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
|
||||
@@ -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__}
|
||||
@@ -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
|
||||
@@ -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:
|
||||
|
||||
@@ -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 = "<new_key>"`` (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 = "<new_key>"`` 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)
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user