From 85791e0df75a88f0dd4cbb927da9adfb7323323a Mon Sep 17 00:00:00 2001 From: Nora Date: Mon, 6 Jul 2026 14:01:37 -0600 Subject: [PATCH] feat(sp36): absorb 20 admin endpoints into api_routers/admin.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Moves the entire /api/admin/* namespace (audit-log ×2, db/rotate-key ×1, backup ×10, scheduler ×6, reload-config ×1) out of api.py and into the existing api_routers/admin.py, which already owned /api/admin/validate-provider. - api.py: 4294 → 3547 LOC (Δ -747). Removed the orphaned # --- separator left behind by the cut, the orphaned 'return {ok,loaded,errors}' tail of reload-config, and the now-unused cyclone.clearhouse.InboundFile top-level import. - api_routers/admin.py: 60 → 851 LOC. Added the imports the moved routes need (json, logging, threading, time, AuditEvent, verify_chain, InboundFile) and stripped the redundant top-level imports of symbols that the route bodies reach via the inline _db_crypto / _secrets / _backup_svc_mod / _backup_sched_mod / _scheduler_mod / _audit aliases (those aliases stay — tests rely on being able to monkeypatch cyclone.api_routers.admin._X.method). Hoisted the inline 'import time' from inside pull_inbound to module scope. Dropped the redundant 'import threading as _threading' alias — top-level 'import threading' already covers it. - tests/test_api_rotate_key.py: 4 monkeypatch targets migrated from cyclone.api._db_crypto / cyclone.api._secrets to cyclone.api_routers.admin._db_crypto / cyclone.api_routers.admin._secrets to match the new home of the rotate-key endpoint. Lock acquire/release also migrated. The non-admin /api/config/* and /api/payers/{id}/summary routes that bracketed the moved admin blocks stay in api.py — they're extracted in Tasks 5 & 6. Behaviour-preserving: pytest = 20 failed / 1253 passed / 10 skipped = baseline match. Admin-only subset (audit-log + backup + scheduler + rotate-key + reload-config) = 34 passed. --- backend/src/cyclone/api.py | 747 --------------------- backend/src/cyclone/api_routers/admin.py | 791 ++++++++++++++++++++++- backend/tests/test_api_rotate_key.py | 18 +- 3 files changed, 790 insertions(+), 766 deletions(-) diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index b84bd33..90d854b 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -37,7 +37,6 @@ from pydantic import ValidationError from cyclone import __version__, db from cyclone.auth.deps import matrix_gate -from cyclone.clearhouse import InboundFile from cyclone.db import Batch, Claim, ClaimState, Remittance from sqlalchemy import desc, or_ from sqlalchemy.exc import IntegrityError @@ -3317,739 +3316,6 @@ def list_configured_providers(is_active: bool | None = Query(default=True)): return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)] -# --------------------------------------------------------------------------- # -# SP11: tamper-evident audit log (admin) -# --------------------------------------------------------------------------- # - - -@app.get("/api/admin/audit-log", dependencies=[Depends(matrix_gate)]) -def list_audit_log_endpoint( - entity_type: str | None = Query(default=None), - entity_id: str | None = Query(default=None), - event_type: str | None = Query(default=None), - limit: int = Query(default=100, ge=1, le=1000), -) -> Any: - """List audit-log rows, newest first, with optional filters. - - Filters match the (entity_type, entity_id) pair (typical use: - "show me everything that happened to claim C-123") or a single - event_type (typical use: "show me all clearhouse.submitted - events today"). - """ - with db.SessionLocal()() as s: - q = s.query(db.AuditLog) - if entity_type: - q = q.filter(db.AuditLog.entity_type == entity_type) - if entity_id: - q = q.filter(db.AuditLog.entity_id == entity_id) - if event_type: - q = q.filter(db.AuditLog.event_type == event_type) - rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all() - return { - "total": len(rows), - "items": [ - { - "id": r.id, - "event_type": r.event_type, - "entity_type": r.entity_type, - "entity_id": r.entity_id, - "actor": r.actor, - "payload": json.loads(r.payload_json) if r.payload_json else None, - "created_at": r.created_at.isoformat() if r.created_at else None, - "prev_hash": r.prev_hash, - "hash": r.hash, - } - for r in rows - ], - } - - -@app.get("/api/admin/audit-log/verify", dependencies=[Depends(matrix_gate)]) -def verify_audit_log_endpoint() -> Any: - """Walk the audit-log chain and verify every row's hash. - - Returns ``{"ok": true, "checked": N}`` for a clean chain, or - ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}`` - for a broken chain. This is the operator's "did anyone tamper?" - endpoint; run it on demand or via a nightly cron job. - """ - with db.SessionLocal()() as s: - result = verify_chain(s) - return { - "ok": result.ok, - "checked": result.checked, - "first_bad_id": result.first_bad_id, - "reason": result.reason, - } - - -# --------------------------------------------------------------------------- -# 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", dependencies=[Depends(matrix_gate)]) -def rotate_db_key_endpoint(body: dict | None = None) -> Any: - """Generate a fresh DB key, re-encrypt the DB, update the Keychain. - - Request body (optional): - actor: who initiated the rotation. Defaults to "operator". - reason: human-readable reason. Written to the audit log. - - Returns: - ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}`` - on success. On failure (DB not encrypted, rekey failed, - Keychain update failed) returns the same shape with - ``ok=false`` and a ``reason``. HTTP 503 is returned if the - rekey fails or encryption is not enabled. - - The Keychain write happens *after* the rekey succeeds. If the - Keychain write fails, the DB has the new key but the Keychain - still has the old one — the endpoint returns 503 with a - "keychain update failed" reason and the operator must restore - the old key manually (``cyclone db restore-key ``) to - avoid being locked out. - """ - body = body or {} - actor = body.get("actor") or "operator" - reason = body.get("reason") or "" - - if not _db_crypto.is_encryption_enabled(): - raise HTTPException( - status_code=400, - detail="encryption not enabled (sqlcipher3 missing or no Keychain key)", - ) - - # Acquire the lock; non-blocking so a stuck rotation doesn't - # silently hold up other requests. - if not _db_rotate_lock.acquire(blocking=False): - raise HTTPException( - status_code=409, - detail="another key rotation is in progress", - ) - try: - url = db._resolve_url() - old_key = _db_crypto.get_db_key() - if not old_key: - raise HTTPException( - status_code=400, - detail="no DB key in Keychain; cannot rotate", - ) - - new_key = _db_crypto.generate_db_key() - result = _db_crypto.rotate_db_key( - url=url, old_key=old_key, new_key=new_key, - ) - if not result.ok: - # Rekey failed. The DB still has the old key. The - # Keychain is unchanged. Caller should NOT retry with - # the same new key (it's lost); generate a fresh one. - log.error("SQLCipher rotate failed: %s", result.reason) - raise HTTPException( - status_code=503, - detail={ - "ok": False, - "old_fingerprint": result.old_fingerprint, - "new_fingerprint": result.new_fingerprint, - "rotated_at": result.rotated_at, - "reason": result.reason, - }, - ) - - # Rekey succeeded. Now update the Keychain. If this fails - # the DB is locked behind the new key — operator must - # restore the old key manually. - if not _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT, new_key): - log.error("Keychain update failed after successful rekey!") - raise HTTPException( - status_code=503, - detail={ - "ok": False, - "old_fingerprint": result.old_fingerprint, - "new_fingerprint": result.new_fingerprint, - "rotated_at": result.rotated_at, - "reason": ( - "rekey succeeded but Keychain update failed — " - "the DB is now encrypted with the new key but " - "the Keychain still has the old one. " - "Restore the old key to the Keychain to recover." - ), - }, - ) - - # Store the old key in the "previous" account for a grace - # period so the operator can roll back if they discover the - # new key is broken (e.g. the Keychain entry got truncated). - _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key) - - # Rebuild the engine so subsequent connections use the new - # key. dispose_engine() closes every pooled connection that - # was using the old key; init_db() opens new ones with the - # new key from the (now-updated) Keychain. - db.reinit_engine() - - # Audit log the rotation. We do this after the engine is - # rebuilt so the audit event is written with the new key — - # proving that the new key works for new writes. - try: - from cyclone.audit_log import append_event, AuditEvent - with db.SessionLocal()() as s: - append_event(s, AuditEvent( - event_type="db.key_rotated", - entity_type="database", - entity_id="cyclone.db", - actor=actor, - payload={ - "old_fingerprint": result.old_fingerprint, - "new_fingerprint": result.new_fingerprint, - "table_count": result.table_count, - "reason": reason, - }, - )) - s.commit() - except Exception as exc: # noqa: BLE001 - # Audit append is best-effort; rotation already succeeded. - log.warning("could not write audit event for rotation: %s", exc) - - return { - "ok": True, - "old_fingerprint": result.old_fingerprint, - "new_fingerprint": result.new_fingerprint, - "rotated_at": result.rotated_at, - "table_count": result.table_count, - } - finally: - _db_rotate_lock.release() - - -# --------------------------------------------------------------------------- -# SP17: encrypted DB backups (admin) -# -# The actual encryption + lifecycle lives in :mod:`cyclone.backup` and -# :mod:`cyclone.backup_service`. The scheduler (separate from the -# MFT scheduler) lives in :mod:`cyclone.backup_scheduler`. These -# endpoints expose the operator's manual controls plus a tick for -# "take a backup right now." -# -# Restore is intentionally two-step: an idle browser tab can't nuke -# the live DB. The first call returns a ``restore_token`` (a one-shot -# 64-char hex) and a preview of the backup's fingerprint + table -# count plus the live DB's. The second call with the token performs -# the actual swap. -# --------------------------------------------------------------------------- -from cyclone import backup_service as _backup_svc_mod -from cyclone import backup_scheduler as _backup_sched_mod - - -def _backup_or_503(): - try: - return _backup_svc_mod.get_backup_service() - except RuntimeError as exc: - raise HTTPException(status_code=503, detail=str(exc)) - - -@app.post("/api/admin/backup/create", dependencies=[Depends(matrix_gate)]) -def backup_create() -> Any: - """Take an encrypted backup right now. Returns the new backup metadata.""" - from cyclone import audit_log as _audit - svc = _backup_or_503() - try: - result = svc.create_now() - except Exception as exc: # noqa: BLE001 - # Surface a 503 with the reason so the operator sees what - # went wrong without grepping server logs. - raise HTTPException( - status_code=503, - detail=f"backup failed: {type(exc).__name__}: {exc}", - ) - # Audit the create. Best-effort; failure here doesn't roll back - # the backup (already on disk). - try: - with db.SessionLocal()() as s: - _audit.append_event(s, _audit.AuditEvent( - event_type="db.backup_created", - entity_type="database", - entity_id="cyclone.db", - actor="operator", - payload={ - "backup_id": result.backup.id, - "db_fingerprint": result.backup.db_fingerprint, - "table_count": result.backup.table_count, - "triggered_by": "api", - }, - )) - s.commit() - except Exception as exc: # noqa: BLE001 - log.warning("could not write backup_created audit event: %s", exc) - - return { - "ok": True, - "backup": { - "id": result.backup.id, - "filename": result.backup.filename, - "size_bytes": result.backup.size_bytes, - "db_fingerprint": result.backup.db_fingerprint, - "table_count": result.backup.table_count, - "created_at": result.backup.created_at.isoformat(), - "key_fingerprint": result.backup.key_fingerprint, - }, - "sidecar": { - "format_version": result.sidecar.format_version, - "kdf": result.sidecar.kdf, - "kdf_iterations": result.sidecar.kdf_iterations, - "cipher": result.sidecar.cipher, - }, - } - - -@app.get("/api/admin/backup/list", dependencies=[Depends(matrix_gate)]) -def backup_list( - limit: int = Query(default=100, ge=1, le=1000), - status: str | None = Query(default=None), -) -> Any: - """List ``db_backups`` rows, newest first. Filters by status.""" - svc = _backup_or_503() - rows = svc.list_backups(limit=limit, status=status) - return { - "count": len(rows), - "files": [ - { - "id": r.id, - "filename": r.filename, - "backup_dir": r.backup_dir, - "size_bytes": r.size_bytes, - "db_fingerprint": r.db_fingerprint, - "table_count": r.table_count, - "created_at": r.created_at.isoformat() if r.created_at else None, - "completed_at": r.completed_at.isoformat() if r.completed_at else None, - "status": r.status, - "error_message": r.error_message, - "key_fingerprint": r.key_fingerprint, - } - for r in rows - ], - } - - -@app.get("/api/admin/backup/status", dependencies=[Depends(matrix_gate)]) -def backup_status() -> Any: - """Snapshot of the backup subsystem (counts, disk usage, last run).""" - svc = _backup_or_503() - snap = svc.status() - # Also include the BackupScheduler's snapshot if configured. - try: - sched = _backup_sched_mod.get_backup_scheduler() - snap["scheduler"] = sched.status().as_dict() - except RuntimeError: - snap["scheduler"] = None - return snap - - -@app.post("/api/admin/backup/{backup_id}/verify", dependencies=[Depends(matrix_gate)]) -def backup_verify(backup_id: int) -> Any: - """Decrypt + checksum-verify a backup against its sidecar.""" - svc = _backup_or_503() - try: - v = svc.verify(backup_id) - except _backup_svc_mod.BackupError as exc: - raise HTTPException(status_code=404, detail=str(exc)) - return { - "backup_id": v.backup_id, - "filename": v.filename, - "ok": v.ok, - "expected_fingerprint": v.expected_fingerprint, - "actual_fingerprint": v.actual_fingerprint, - "table_count": v.table_count, - "reason": v.reason, - } - - -@app.post("/api/admin/backup/{backup_id}/restore/initiate", dependencies=[Depends(matrix_gate)]) -def backup_restore_initiate(backup_id: int) -> Any: - """First step of the two-step restore. Returns a ``restore_token``.""" - svc = _backup_or_503() - try: - init = svc.restore_initiate(backup_id) - except _backup_svc_mod.BackupError as exc: - raise HTTPException(status_code=400, detail=str(exc)) - return { - "backup_id": init.backup_id, - "filename": init.filename, - "size_bytes": init.size_bytes, - "restore_token": init.restore_token, - "expires_at": init.expires_at.isoformat(), - "preview": { - "backup_db_fingerprint": init.db_fingerprint, - "backup_table_count": init.table_count, - "current_db_fingerprint": init.current_db_fingerprint, - "current_table_count": init.current_table_count, - }, - "warning": ( - "Confirming will dispose the live engine and replace the DB " - "file with the backup. In-flight requests will error. " - "Re-issue the call with the restore_token within 5 minutes." - ), - } - - -@app.post("/api/admin/backup/{backup_id}/restore/confirm", dependencies=[Depends(matrix_gate)]) -def backup_restore_confirm( - backup_id: int, - body: dict | None = None, -) -> Any: - """Second step of the two-step restore. Performs the swap.""" - body = body or {} - token = body.get("restore_token") - if not token or not isinstance(token, str): - raise HTTPException( - status_code=400, - detail="missing or invalid restore_token in request body", - ) - actor = body.get("actor") or "operator" - - svc = _backup_or_503() - try: - result = svc.restore_confirm(backup_id, token, actor=actor) - except _backup_svc_mod.BackupError as exc: - raise HTTPException(status_code=400, detail=str(exc)) - - # Audit the restore. Best-effort. - try: - from cyclone import audit_log as _audit - with db.SessionLocal()() as s: - _audit.append_event(s, _audit.AuditEvent( - event_type="db.backup_restored", - entity_type="database", - entity_id="cyclone.db", - actor=actor, - payload={ - "backup_id": result.backup_id, - "filename": result.filename, - "restored_from_fingerprint": result.restored_from_fingerprint, - "new_db_fingerprint": result.new_db_fingerprint, - "restored_at": result.restored_at.isoformat(), - }, - )) - s.commit() - except Exception as exc: # noqa: BLE001 - log.warning("could not write backup_restored audit event: %s", exc) - - return { - "ok": True, - "backup_id": result.backup_id, - "filename": result.filename, - "restored_from_fingerprint": result.restored_from_fingerprint, - "restored_at": result.restored_at.isoformat(), - "new_db_fingerprint": result.new_db_fingerprint, - } - - -@app.post("/api/admin/backup/prune", dependencies=[Depends(matrix_gate)]) -def backup_prune() -> Any: - """Apply the retention policy now. Returns the deleted paths.""" - from cyclone import audit_log as _audit - svc = _backup_or_503() - deleted = svc.prune() - actor = "operator" - if deleted: - try: - with db.SessionLocal()() as s: - _audit.append_event(s, _audit.AuditEvent( - event_type="db.backup_pruned", - entity_type="database", - entity_id="cyclone.db", - actor=actor, - payload={"deleted_paths": deleted}, - )) - s.commit() - except Exception as exc: # noqa: BLE001 - log.warning("could not write backup_pruned audit event: %s", exc) - return {"ok": True, "deleted_count": len(deleted), "deleted_paths": deleted} - - -# --------------------------------------------------------------------------- -# SP17: backup scheduler (admin) -# --------------------------------------------------------------------------- - - -@app.post("/api/admin/backup/scheduler/start", dependencies=[Depends(matrix_gate)]) -async def backup_scheduler_start() -> Any: - """Begin the backup scheduler loop.""" - try: - sched = _backup_sched_mod.get_backup_scheduler() - except RuntimeError as exc: - raise HTTPException(status_code=503, detail=str(exc)) - await sched.start() - return {"status": sched.status().as_dict()} - - -@app.post("/api/admin/backup/scheduler/stop", dependencies=[Depends(matrix_gate)]) -async def backup_scheduler_stop() -> Any: - """Stop the backup scheduler loop.""" - try: - sched = _backup_sched_mod.get_backup_scheduler() - except RuntimeError as exc: - raise HTTPException(status_code=503, detail=str(exc)) - await sched.stop() - return {"status": sched.status().as_dict()} - - -@app.post("/api/admin/backup/scheduler/tick", dependencies=[Depends(matrix_gate)]) -async def backup_scheduler_tick() -> Any: - """Run one backup tick now (create + prune + audit).""" - try: - sched = _backup_sched_mod.get_backup_scheduler() - except RuntimeError as exc: - raise HTTPException(status_code=503, detail=str(exc)) - result = await sched.tick() - return {"ok": result.ok, "tick": result.as_dict()} - - -# --------------------------------------------------------------------------- -# SP16: live MFT polling scheduler (admin) -# -# The scheduler lives in :mod:`cyclone.scheduler` and is configured by -# the lifespan handler. The endpoints below expose start / stop / -# one-shot tick / status / history so an operator (or a cron job) -# can drive the scheduler without touching the DB. -# -# Note: the scheduler is OFF by default. Auto-start is opt-in via -# ``CYCLONE_SCHEDULER_AUTOSTART=true`` at launch. These endpoints -# are the operator's manual controls. -# --------------------------------------------------------------------------- -from cyclone import scheduler as _scheduler_mod - - -def _scheduler_or_503(): - """Return the configured scheduler or raise 503.""" - try: - return _scheduler_mod.get_scheduler() - except RuntimeError as exc: - raise HTTPException(status_code=503, detail=str(exc)) - - -@app.post("/api/admin/scheduler/start", dependencies=[Depends(matrix_gate)]) -async def scheduler_start() -> Any: - """Begin polling the MFT inbound path every poll_interval_seconds.""" - sched = _scheduler_or_503() - await sched.start() - return {"status": sched.status().as_dict()} - - -@app.post("/api/admin/scheduler/stop", dependencies=[Depends(matrix_gate)]) -async def scheduler_stop() -> Any: - """Stop polling. Waits up to 30s for the current tick to finish.""" - sched = _scheduler_or_503() - await sched.stop() - return {"status": sched.status().as_dict()} - - -@app.post("/api/admin/scheduler/tick", dependencies=[Depends(matrix_gate)]) -async def scheduler_tick() -> Any: - """Run a single poll cycle synchronously and return the result. - - Useful for: forcing a poll without waiting for the next interval; - verifying SFTP connectivity; running a one-shot import from the - CLI (``curl -X POST .../api/admin/scheduler/tick``). - """ - sched = _scheduler_or_503() - result = await sched.tick() - return {"ok": True, "tick": result.as_dict()} - - -@app.post("/api/admin/scheduler/pull-inbound", dependencies=[Depends(matrix_gate)]) -async def scheduler_pull_inbound( - date: str = Query( - ..., pattern=r"^\d{8}$", - description="Date filter as YYYYMMDD; only filenames whose 8-digit " - "timestamp (the 9th positional group in the inbound " - "filename) matches are downloaded and processed.", - ), - file_types: str | None = Query( - default=None, - description="Optional comma-separated whitelist of file_types " - "(999, TA1, 277, 277CA, 835). Defaults to 999+TA1.", - ), - limit: int = Query(default=2000, ge=1, le=10000), -) -> Any: - """Targeted pull: list, filter to a date, download, and process. - - Bypasses the alphabetical full-listing pass. Workflow: - 1. ``SftpClient.list_inbound_names()`` — sub-second metadata-only - listing of the inbound MFT dir (skips ``*_warn.txt``). - 2. Client-side filter: keep files whose 8-digit timestamp - substring equals ``date`` and whose ``file_type`` is in the - allowlist. - 3. ``SftpClient.download_inbound(f)`` for each — fetches bytes - into the local cache. - 4. ``Scheduler.process_inbound_files(files)`` — runs the same - per-file pipeline as a regular tick (already-processed files - are deduped via ``processed_inbound_files``). - - Use this for the daily "process today's 999s" workflow without - paying the cost of downloading the full inbound set. - - Returns ``{"ok": True, "summary": {...}}`` with - ``listed / matched / downloaded / processed / skipped / errored`` - counters and the date / file_type filters applied. - """ - import time - - from cyclone.clearhouse import SftpClient - from cyclone.edi.filenames import ( - ALLOWED_FILE_TYPES, - parse_inbound_filename, - ) - from cyclone.providers import SftpBlock - - sched = _scheduler_or_503() - block: SftpBlock = sched._sftp_block # noqa: SLF001 — internal but stable - client = SftpClient(block) - - if file_types: - wanted = {t.strip().upper() for t in file_types.split(",") if t.strip()} - unknown = wanted - ALLOWED_FILE_TYPES - if unknown: - raise HTTPException( - status_code=400, - detail=f"file_types {sorted(unknown)!r} not in " - f"{sorted(ALLOWED_FILE_TYPES)}", - ) - else: - wanted = {"999", "TA1"} # daily default — what the operator needs - - started = time.monotonic() - try: - # Single SFTP listdir — fast, no download. - all_files = await asyncio.to_thread(client.list_inbound_names) - except Exception as exc: - log.exception("SFTP list_inbound_names failed") - raise HTTPException( - status_code=502, - detail=f"SFTP list failed: {type(exc).__name__}: {exc}", - ) from exc - - listed = len(all_files) - matched: list[InboundFile] = [] - for f in all_files: - if f.name.find(date) == -1: - continue - try: - parsed = parse_inbound_filename(f.name) - except ValueError: - continue - if parsed.file_type not in wanted: - continue - matched.append(f) - if len(matched) >= limit: - break - - # Download in parallel-ish via to_thread (SftpClient serializes per - # connection; the overhead is dominated by the SFTP round trip). - downloaded = 0 - download_errors: list[str] = [] - for f in matched: - try: - await asyncio.to_thread(client.download_inbound, f) - downloaded += 1 - except Exception as exc: # noqa: BLE001 - log.warning("Failed to download %s: %s", f.name, exc) - download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}") - - # Hand off to the scheduler pipeline (idempotent; dedupes via - # processed_inbound_files). - tick = await sched.process_inbound_files(matched) - duration = round(time.monotonic() - started, 3) - return { - "ok": True, - "summary": { - "date": date, - "file_types": sorted(wanted), - "limit": limit, - "listed": listed, - "matched": len(matched), - "downloaded": downloaded, - "download_errors": download_errors, - "processed": tick.files_processed, - "skipped": tick.files_skipped, - "errored": tick.files_errored, - "duration_s": duration, - }, - "tick": tick.as_dict(), - } - - -@app.get("/api/admin/scheduler/status", dependencies=[Depends(matrix_gate)]) -def scheduler_status() -> Any: - """Return the scheduler's runtime snapshot (running, counters, last tick).""" - sched = _scheduler_or_503() - return sched.status().as_dict() - - -@app.get("/api/admin/scheduler/processed-files", dependencies=[Depends(matrix_gate)]) -def scheduler_processed_files( - limit: int = Query(default=100, ge=1, le=1000), - status: str | None = Query(default=None), -) -> Any: - """List rows from ``processed_inbound_files``, newest first. - - The operator's "what did the scheduler do?" view. Filters by - ``status`` (``ok`` / ``error`` / ``skipped`` / ``pending``). - Returns ``{"count": N, "files": [...]}`` where ``files[i]`` - matches the ORM row as a JSON dict. - """ - from cyclone.db import ProcessedInboundFile - from cyclone.scheduler import STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING - - valid_statuses = {STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING} - if status is not None and status not in valid_statuses: - raise HTTPException( - status_code=400, - detail=f"status must be one of {sorted(valid_statuses)}", - ) - - with db.SessionLocal()() as s: - q = s.query(db.ProcessedInboundFile) - if status is not None: - q = q.filter(db.ProcessedInboundFile.status == status) - rows = q.order_by(db.ProcessedInboundFile.id.desc()).limit(limit).all() - return { - "count": len(rows), - "files": [ - { - "id": r.id, - "sftp_block_name": r.sftp_block_name, - "name": r.name, - "size": r.size, - "modified_at": r.modified_at.isoformat() if r.modified_at else None, - "file_type": r.file_type, - "processed_at": r.processed_at.isoformat() if r.processed_at else None, - "parser_used": r.parser_used, - "claim_count": r.claim_count, - "status": r.status, - "error_message": r.error_message, - } - for r in rows - ], - } - - @app.get("/api/config/providers/{npi}", dependencies=[Depends(matrix_gate)]) def get_configured_provider(npi: str): p = store.get_provider(npi) @@ -4266,19 +3532,6 @@ def get_payer_summary(payer_id: str): return payload -@app.post("/api/admin/reload-config", dependencies=[Depends(matrix_gate)]) -def reload_config(): - """Re-read ``config/payers.yaml`` and revalidate. Returns counts.""" - from cyclone import payers as payer_loader - try: - configs = payer_loader.load_payer_configs() - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - return {"ok": True, "loaded": len(configs), "errors": []} - - - - # --------------------------------------------------------------------------- # # Auth routers (login/logout/me + admin user management) # # --------------------------------------------------------------------------- # diff --git a/backend/src/cyclone/api_routers/admin.py b/backend/src/cyclone/api_routers/admin.py index 9fcf362..386576d 100644 --- a/backend/src/cyclone/api_routers/admin.py +++ b/backend/src/cyclone/api_routers/admin.py @@ -1,25 +1,45 @@ -"""``/api/admin/validate-provider`` — NPI + Tax ID liveness probe (SP20). +"""``/api/admin/*`` — operator-only endpoints. -Pure read-only endpoint that runs the local NPI Luhn + EIN format checks -without touching the DB. Useful for: -- operators vetting a new provider before adding them to the registry, -- the dashboard's "validate" button on a Provider row, -- smoke-testing the SP20 checks after a deploy. +The /api/admin namespace covers: -Both query params are optional; omitting one just skips that check. -Returns the per-check result dict so the caller can distinguish "bad -format" from "bad checksum". +* **audit-log** (SP11): list + chain-verify the tamper-evident log +* **db/rotate-key**: SQLCipher key rotation (SP12) +* **backup**: create / list / status / verify / restore / prune / scheduler (SP15) +* **scheduler**: backup & main scheduler control + processed-files log +* **reload-config**: hot-reload ``config/payers.yaml`` after edits +* **validate-provider**: NPI + Tax ID liveness probe (SP20) + +All routes on this router are gated by ``matrix_gate`` — declared +once on the ``APIRouter`` constructor. ``/api/admin/validate-provider`` +lives here too because it's an admin-shaped read-only probe. + +SP36 Task 3: the 20 admin endpoints formerly in ``cyclone.api`` +(audit-log ×2, db/rotate-key ×1, backup ×10, scheduler ×6, +reload-config ×1) moved here from ``api.py:3321-4052`` and +``api.py:4269-4276``. The non-admin ``/api/config/*`` and +``/api/payers/{id}/summary`` routes that bracketed those blocks +stay in ``api.py`` for now — they're extracted in Tasks 5 & 6. """ from __future__ import annotations -from fastapi import APIRouter, Depends, Query +import json +import logging +import threading +import time +from typing import Any +from fastapi import APIRouter, Depends, HTTPException, Query + +from cyclone import db +from cyclone.audit_log import verify_chain from cyclone.auth.deps import matrix_gate +from cyclone.clearhouse import InboundFile from cyclone.npi import is_valid_npi, is_valid_tax_id, normalize_tax_id - router = APIRouter(dependencies=[Depends(matrix_gate)]) +log = logging.getLogger(__name__) + @router.get("/api/admin/validate-provider") def validate_provider( @@ -58,4 +78,751 @@ def validate_provider( "normalized": normalized, } - return result \ No newline at end of file + return result + + +# --------------------------------------------------------------------------- # +# SP36 Task 3: the 20 admin endpoints below were moved here from api.py. # +# --------------------------------------------------------------------------- # + + +# --------------------------------------------------------------------------- # +# SP11: tamper-evident audit log (admin) +# --------------------------------------------------------------------------- # + + +@router.get("/api/admin/audit-log") +def list_audit_log_endpoint( + entity_type: str | None = Query(default=None), + entity_id: str | None = Query(default=None), + event_type: str | None = Query(default=None), + limit: int = Query(default=100, ge=1, le=1000), +) -> Any: + """List audit-log rows, newest first, with optional filters. + + Filters match the (entity_type, entity_id) pair (typical use: + "show me everything that happened to claim C-123") or a single + event_type (typical use: "show me all clearhouse.submitted + events today"). + """ + with db.SessionLocal()() as s: + q = s.query(db.AuditLog) + if entity_type: + q = q.filter(db.AuditLog.entity_type == entity_type) + if entity_id: + q = q.filter(db.AuditLog.entity_id == entity_id) + if event_type: + q = q.filter(db.AuditLog.event_type == event_type) + rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all() + return { + "total": len(rows), + "items": [ + { + "id": r.id, + "event_type": r.event_type, + "entity_type": r.entity_type, + "entity_id": r.entity_id, + "actor": r.actor, + "payload": json.loads(r.payload_json) if r.payload_json else None, + "created_at": r.created_at.isoformat() if r.created_at else None, + "prev_hash": r.prev_hash, + "hash": r.hash, + } + for r in rows + ], + } + + +@router.get("/api/admin/audit-log/verify") +def verify_audit_log_endpoint() -> Any: + """Walk the audit-log chain and verify every row's hash. + + Returns ``{"ok": true, "checked": N}`` for a clean chain, or + ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}`` + for a broken chain. This is the operator's "did anyone tamper?" + endpoint; run it on demand or via a nightly cron job. + """ + with db.SessionLocal()() as s: + result = verify_chain(s) + return { + "ok": result.ok, + "checked": result.checked, + "first_bad_id": result.first_bad_id, + "reason": result.reason, + } + + +# --------------------------------------------------------------------------- +# 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). +# --------------------------------------------------------------------------- +from cyclone import db_crypto as _db_crypto +from cyclone import secrets as _secrets + +_db_rotate_lock = threading.Lock() + + +@router.post("/api/admin/db/rotate-key") +def rotate_db_key_endpoint(body: dict | None = None) -> Any: + """Generate a fresh DB key, re-encrypt the DB, update the Keychain. + + Request body (optional): + actor: who initiated the rotation. Defaults to "operator". + reason: human-readable reason. Written to the audit log. + + Returns: + ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}`` + on success. On failure (DB not encrypted, rekey failed, + Keychain update failed) returns the same shape with + ``ok=false`` and a ``reason``. HTTP 503 is returned if the + rekey fails or encryption is not enabled. + + The Keychain write happens *after* the rekey succeeds. If the + Keychain write fails, the DB has the new key but the Keychain + still has the old one — the endpoint returns 503 with a + "keychain update failed" reason and the operator must restore + the old key manually (``cyclone db restore-key ``) to + avoid being locked out. + """ + body = body or {} + actor = body.get("actor") or "operator" + reason = body.get("reason") or "" + + if not _db_crypto.is_encryption_enabled(): + raise HTTPException( + status_code=400, + detail="encryption not enabled (sqlcipher3 missing or no Keychain key)", + ) + + # Acquire the lock; non-blocking so a stuck rotation doesn't + # silently hold up other requests. + if not _db_rotate_lock.acquire(blocking=False): + raise HTTPException( + status_code=409, + detail="another key rotation is in progress", + ) + try: + url = db._resolve_url() + old_key = _db_crypto.get_db_key() + if not old_key: + raise HTTPException( + status_code=400, + detail="no DB key in Keychain; cannot rotate", + ) + + new_key = _db_crypto.generate_db_key() + result = _db_crypto.rotate_db_key( + url=url, old_key=old_key, new_key=new_key, + ) + if not result.ok: + # Rekey failed. The DB still has the old key. The + # Keychain is unchanged. Caller should NOT retry with + # the same new key (it's lost); generate a fresh one. + log.error("SQLCipher rotate failed: %s", result.reason) + raise HTTPException( + status_code=503, + detail={ + "ok": False, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "reason": result.reason, + }, + ) + + # Rekey succeeded. Now update the Keychain. If this fails + # the DB is locked behind the new key — operator must + # restore the old key manually. + if not _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT, new_key): + log.error("Keychain update failed after successful rekey!") + raise HTTPException( + status_code=503, + detail={ + "ok": False, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "reason": ( + "rekey succeeded but Keychain update failed — " + "the DB is now encrypted with the new key but " + "the Keychain still has the old one. " + "Restore the old key to the Keychain to recover." + ), + }, + ) + + # Store the old key in the "previous" account for a grace + # period so the operator can roll back if they discover the + # new key is broken (e.g. the Keychain entry got truncated). + _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key) + + # Rebuild the engine so subsequent connections use the new + # key. dispose_engine() closes every pooled connection that + # was using the old key; init_db() opens new ones with the + # new key from the (now-updated) Keychain. + db.reinit_engine() + + # Audit log the rotation. We do this after the engine is + # rebuilt so the audit event is written with the new key — + # proving that the new key works for new writes. + try: + from cyclone.audit_log import append_event, AuditEvent + with db.SessionLocal()() as s: + append_event(s, AuditEvent( + event_type="db.key_rotated", + entity_type="database", + entity_id="cyclone.db", + actor=actor, + payload={ + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "table_count": result.table_count, + "reason": reason, + }, + )) + s.commit() + except Exception as exc: # noqa: BLE001 + # Audit append is best-effort; rotation already succeeded. + log.warning("could not write audit event for rotation: %s", exc) + + return { + "ok": True, + "old_fingerprint": result.old_fingerprint, + "new_fingerprint": result.new_fingerprint, + "rotated_at": result.rotated_at, + "table_count": result.table_count, + } + finally: + _db_rotate_lock.release() + + +# --------------------------------------------------------------------------- +# SP17: encrypted DB backups (admin) +# +# The actual encryption + lifecycle lives in :mod:`cyclone.backup` and +# :mod:`cyclone.backup_service`. The scheduler (separate from the +# MFT scheduler) lives in :mod:`cyclone.backup_scheduler`. These +# endpoints expose the operator's manual controls plus a tick for +# "take a backup right now." +# +# Restore is intentionally two-step: an idle browser tab can't nuke +# the live DB. The first call returns a ``restore_token`` (a one-shot +# 64-char hex) and a preview of the backup's fingerprint + table +# count plus the live DB's. The second call with the token performs +# the actual swap. +# --------------------------------------------------------------------------- +from cyclone import backup_service as _backup_svc_mod +from cyclone import backup_scheduler as _backup_sched_mod + + +def _backup_or_503(): + try: + return _backup_svc_mod.get_backup_service() + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) + + +@router.post("/api/admin/backup/create") +def backup_create() -> Any: + """Take an encrypted backup right now. Returns the new backup metadata.""" + from cyclone import audit_log as _audit + svc = _backup_or_503() + try: + result = svc.create_now() + except Exception as exc: # noqa: BLE001 + # Surface a 503 with the reason so the operator sees what + # went wrong without grepping server logs. + raise HTTPException( + status_code=503, + detail=f"backup failed: {type(exc).__name__}: {exc}", + ) + # Audit the create. Best-effort; failure here doesn't roll back + # the backup (already on disk). + try: + with db.SessionLocal()() as s: + _audit.append_event(s, _audit.AuditEvent( + event_type="db.backup_created", + entity_type="database", + entity_id="cyclone.db", + actor="operator", + payload={ + "backup_id": result.backup.id, + "db_fingerprint": result.backup.db_fingerprint, + "table_count": result.backup.table_count, + "triggered_by": "api", + }, + )) + s.commit() + except Exception as exc: # noqa: BLE001 + log.warning("could not write backup_created audit event: %s", exc) + + return { + "ok": True, + "backup": { + "id": result.backup.id, + "filename": result.backup.filename, + "size_bytes": result.backup.size_bytes, + "db_fingerprint": result.backup.db_fingerprint, + "table_count": result.backup.table_count, + "created_at": result.backup.created_at.isoformat(), + "key_fingerprint": result.backup.key_fingerprint, + }, + "sidecar": { + "format_version": result.sidecar.format_version, + "kdf": result.sidecar.kdf, + "kdf_iterations": result.sidecar.kdf_iterations, + "cipher": result.sidecar.cipher, + }, + } + + +@router.get("/api/admin/backup/list") +def backup_list( + limit: int = Query(default=100, ge=1, le=1000), + status: str | None = Query(default=None), +) -> Any: + """List ``db_backups`` rows, newest first. Filters by status.""" + svc = _backup_or_503() + rows = svc.list_backups(limit=limit, status=status) + return { + "count": len(rows), + "files": [ + { + "id": r.id, + "filename": r.filename, + "backup_dir": r.backup_dir, + "size_bytes": r.size_bytes, + "db_fingerprint": r.db_fingerprint, + "table_count": r.table_count, + "created_at": r.created_at.isoformat() if r.created_at else None, + "completed_at": r.completed_at.isoformat() if r.completed_at else None, + "status": r.status, + "error_message": r.error_message, + "key_fingerprint": r.key_fingerprint, + } + for r in rows + ], + } + + +@router.get("/api/admin/backup/status") +def backup_status() -> Any: + """Snapshot of the backup subsystem (counts, disk usage, last run).""" + svc = _backup_or_503() + snap = svc.status() + # Also include the BackupScheduler's snapshot if configured. + try: + sched = _backup_sched_mod.get_backup_scheduler() + snap["scheduler"] = sched.status().as_dict() + except RuntimeError: + snap["scheduler"] = None + return snap + + +@router.post("/api/admin/backup/{backup_id}/verify") +def backup_verify(backup_id: int) -> Any: + """Decrypt + checksum-verify a backup against its sidecar.""" + svc = _backup_or_503() + try: + v = svc.verify(backup_id) + except _backup_svc_mod.BackupError as exc: + raise HTTPException(status_code=404, detail=str(exc)) + return { + "backup_id": v.backup_id, + "filename": v.filename, + "ok": v.ok, + "expected_fingerprint": v.expected_fingerprint, + "actual_fingerprint": v.actual_fingerprint, + "table_count": v.table_count, + "reason": v.reason, + } + + +@router.post("/api/admin/backup/{backup_id}/restore/initiate") +def backup_restore_initiate(backup_id: int) -> Any: + """First step of the two-step restore. Returns a ``restore_token``.""" + svc = _backup_or_503() + try: + init = svc.restore_initiate(backup_id) + except _backup_svc_mod.BackupError as exc: + raise HTTPException(status_code=400, detail=str(exc)) + return { + "backup_id": init.backup_id, + "filename": init.filename, + "size_bytes": init.size_bytes, + "restore_token": init.restore_token, + "expires_at": init.expires_at.isoformat(), + "preview": { + "backup_db_fingerprint": init.db_fingerprint, + "backup_table_count": init.table_count, + "current_db_fingerprint": init.current_db_fingerprint, + "current_table_count": init.current_table_count, + }, + "warning": ( + "Confirming will dispose the live engine and replace the DB " + "file with the backup. In-flight requests will error. " + "Re-issue the call with the restore_token within 5 minutes." + ), + } + + +@router.post("/api/admin/backup/{backup_id}/restore/confirm") +def backup_restore_confirm( + backup_id: int, + body: dict | None = None, +) -> Any: + """Second step of the two-step restore. Performs the swap.""" + body = body or {} + token = body.get("restore_token") + if not token or not isinstance(token, str): + raise HTTPException( + status_code=400, + detail="missing or invalid restore_token in request body", + ) + actor = body.get("actor") or "operator" + + svc = _backup_or_503() + try: + result = svc.restore_confirm(backup_id, token, actor=actor) + except _backup_svc_mod.BackupError as exc: + raise HTTPException(status_code=400, detail=str(exc)) + + # Audit the restore. Best-effort. + try: + from cyclone import audit_log as _audit + with db.SessionLocal()() as s: + _audit.append_event(s, _audit.AuditEvent( + event_type="db.backup_restored", + entity_type="database", + entity_id="cyclone.db", + actor=actor, + payload={ + "backup_id": result.backup_id, + "filename": result.filename, + "restored_from_fingerprint": result.restored_from_fingerprint, + "new_db_fingerprint": result.new_db_fingerprint, + "restored_at": result.restored_at.isoformat(), + }, + )) + s.commit() + except Exception as exc: # noqa: BLE001 + log.warning("could not write backup_restored audit event: %s", exc) + + return { + "ok": True, + "backup_id": result.backup_id, + "filename": result.filename, + "restored_from_fingerprint": result.restored_from_fingerprint, + "restored_at": result.restored_at.isoformat(), + "new_db_fingerprint": result.new_db_fingerprint, + } + + +@router.post("/api/admin/backup/prune") +def backup_prune() -> Any: + """Apply the retention policy now. Returns the deleted paths.""" + from cyclone import audit_log as _audit + svc = _backup_or_503() + deleted = svc.prune() + actor = "operator" + if deleted: + try: + with db.SessionLocal()() as s: + _audit.append_event(s, _audit.AuditEvent( + event_type="db.backup_pruned", + entity_type="database", + entity_id="cyclone.db", + actor=actor, + payload={"deleted_paths": deleted}, + )) + s.commit() + except Exception as exc: # noqa: BLE001 + log.warning("could not write backup_pruned audit event: %s", exc) + return {"ok": True, "deleted_count": len(deleted), "deleted_paths": deleted} + + +# --------------------------------------------------------------------------- +# SP17: backup scheduler (admin) +# --------------------------------------------------------------------------- + + +@router.post("/api/admin/backup/scheduler/start") +async def backup_scheduler_start() -> Any: + """Begin the backup scheduler loop.""" + try: + sched = _backup_sched_mod.get_backup_scheduler() + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) + await sched.start() + return {"status": sched.status().as_dict()} + + +@router.post("/api/admin/backup/scheduler/stop") +async def backup_scheduler_stop() -> Any: + """Stop the backup scheduler loop.""" + try: + sched = _backup_sched_mod.get_backup_scheduler() + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) + await sched.stop() + return {"status": sched.status().as_dict()} + + +@router.post("/api/admin/backup/scheduler/tick") +async def backup_scheduler_tick() -> Any: + """Run one backup tick now (create + prune + audit).""" + try: + sched = _backup_sched_mod.get_backup_scheduler() + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) + result = await sched.tick() + return {"ok": result.ok, "tick": result.as_dict()} + + +# --------------------------------------------------------------------------- +# SP16: live MFT polling scheduler (admin) +# +# The scheduler lives in :mod:`cyclone.scheduler` and is configured by +# the lifespan handler. The endpoints below expose start / stop / +# one-shot tick / status / history so an operator (or a cron job) +# can drive the scheduler without touching the DB. +# +# Note: the scheduler is OFF by default. Auto-start is opt-in via +# ``CYCLONE_SCHEDULER_AUTOSTART=true`` at launch. These endpoints +# are the operator's manual controls. +# --------------------------------------------------------------------------- +from cyclone import scheduler as _scheduler_mod + + +def _scheduler_or_503(): + """Return the configured scheduler or raise 503.""" + try: + return _scheduler_mod.get_scheduler() + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) + + +@router.post("/api/admin/scheduler/start") +async def scheduler_start() -> Any: + """Begin polling the MFT inbound path every poll_interval_seconds.""" + sched = _scheduler_or_503() + await sched.start() + return {"status": sched.status().as_dict()} + + +@router.post("/api/admin/scheduler/stop") +async def scheduler_stop() -> Any: + """Stop polling. Waits up to 30s for the current tick to finish.""" + sched = _scheduler_or_503() + await sched.stop() + return {"status": sched.status().as_dict()} + + +@router.post("/api/admin/scheduler/tick") +async def scheduler_tick() -> Any: + """Run a single poll cycle synchronously and return the result. + + Useful for: forcing a poll without waiting for the next interval; + verifying SFTP connectivity; running a one-shot import from the + CLI (``curl -X POST .../api/admin/scheduler/tick``). + """ + sched = _scheduler_or_503() + result = await sched.tick() + return {"ok": True, "tick": result.as_dict()} + + +@router.post("/api/admin/scheduler/pull-inbound") +async def scheduler_pull_inbound( + date: str = Query( + ..., pattern=r"^\d{8}$", + description="Date filter as YYYYMMDD; only filenames whose 8-digit " + "timestamp (the 9th positional group in the inbound " + "filename) matches are downloaded and processed.", + ), + file_types: str | None = Query( + default=None, + description="Optional comma-separated whitelist of file_types " + "(999, TA1, 277, 277CA, 835). Defaults to 999+TA1.", + ), + limit: int = Query(default=2000, ge=1, le=10000), +) -> Any: + """Targeted pull: list, filter to a date, download, and process. + + Bypasses the alphabetical full-listing pass. Workflow: + 1. ``SftpClient.list_inbound_names()`` — sub-second metadata-only + listing of the inbound MFT dir (skips ``*_warn.txt``). + 2. Client-side filter: keep files whose 8-digit timestamp + substring equals ``date`` and whose ``file_type`` is in the + allowlist. + 3. ``SftpClient.download_inbound(f)`` for each — fetches bytes + into the local cache. + 4. ``Scheduler.process_inbound_files(files)`` — runs the same + per-file pipeline as a regular tick (already-processed files + are deduped via ``processed_inbound_files``). + + Use this for the daily "process today's 999s" workflow without + paying the cost of downloading the full inbound set. + + Returns ``{"ok": True, "summary": {...}}`` with + ``listed / matched / downloaded / processed / skipped / errored`` + counters and the date / file_type filters applied. + """ + + from cyclone.clearhouse import SftpClient + from cyclone.edi.filenames import ( + ALLOWED_FILE_TYPES, + parse_inbound_filename, + ) + from cyclone.providers import SftpBlock + + sched = _scheduler_or_503() + block: SftpBlock = sched._sftp_block # noqa: SLF001 — internal but stable + client = SftpClient(block) + + if file_types: + wanted = {t.strip().upper() for t in file_types.split(",") if t.strip()} + unknown = wanted - ALLOWED_FILE_TYPES + if unknown: + raise HTTPException( + status_code=400, + detail=f"file_types {sorted(unknown)!r} not in " + f"{sorted(ALLOWED_FILE_TYPES)}", + ) + else: + wanted = {"999", "TA1"} # daily default — what the operator needs + + started = time.monotonic() + try: + # Single SFTP listdir — fast, no download. + all_files = await asyncio.to_thread(client.list_inbound_names) + except Exception as exc: + log.exception("SFTP list_inbound_names failed") + raise HTTPException( + status_code=502, + detail=f"SFTP list failed: {type(exc).__name__}: {exc}", + ) from exc + + listed = len(all_files) + matched: list[InboundFile] = [] + for f in all_files: + if f.name.find(date) == -1: + continue + try: + parsed = parse_inbound_filename(f.name) + except ValueError: + continue + if parsed.file_type not in wanted: + continue + matched.append(f) + if len(matched) >= limit: + break + + # Download in parallel-ish via to_thread (SftpClient serializes per + # connection; the overhead is dominated by the SFTP round trip). + downloaded = 0 + download_errors: list[str] = [] + for f in matched: + try: + await asyncio.to_thread(client.download_inbound, f) + downloaded += 1 + except Exception as exc: # noqa: BLE001 + log.warning("Failed to download %s: %s", f.name, exc) + download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}") + + # Hand off to the scheduler pipeline (idempotent; dedupes via + # processed_inbound_files). + tick = await sched.process_inbound_files(matched) + duration = round(time.monotonic() - started, 3) + return { + "ok": True, + "summary": { + "date": date, + "file_types": sorted(wanted), + "limit": limit, + "listed": listed, + "matched": len(matched), + "downloaded": downloaded, + "download_errors": download_errors, + "processed": tick.files_processed, + "skipped": tick.files_skipped, + "errored": tick.files_errored, + "duration_s": duration, + }, + "tick": tick.as_dict(), + } + + +@router.get("/api/admin/scheduler/status") +def scheduler_status() -> Any: + """Return the scheduler's runtime snapshot (running, counters, last tick).""" + sched = _scheduler_or_503() + return sched.status().as_dict() + + +@router.get("/api/admin/scheduler/processed-files") +def scheduler_processed_files( + limit: int = Query(default=100, ge=1, le=1000), + status: str | None = Query(default=None), +) -> Any: + """List rows from ``processed_inbound_files``, newest first. + + The operator's "what did the scheduler do?" view. Filters by + ``status`` (``ok`` / ``error`` / ``skipped`` / ``pending``). + Returns ``{"count": N, "files": [...]}`` where ``files[i]`` + matches the ORM row as a JSON dict. + """ + from cyclone.db import ProcessedInboundFile + from cyclone.scheduler import STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING + + valid_statuses = {STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING} + if status is not None and status not in valid_statuses: + raise HTTPException( + status_code=400, + detail=f"status must be one of {sorted(valid_statuses)}", + ) + + with db.SessionLocal()() as s: + q = s.query(db.ProcessedInboundFile) + if status is not None: + q = q.filter(db.ProcessedInboundFile.status == status) + rows = q.order_by(db.ProcessedInboundFile.id.desc()).limit(limit).all() + return { + "count": len(rows), + "files": [ + { + "id": r.id, + "sftp_block_name": r.sftp_block_name, + "name": r.name, + "size": r.size, + "modified_at": r.modified_at.isoformat() if r.modified_at else None, + "file_type": r.file_type, + "processed_at": r.processed_at.isoformat() if r.processed_at else None, + "parser_used": r.parser_used, + "claim_count": r.claim_count, + "status": r.status, + "error_message": r.error_message, + } + for r in rows + ], + } + + +@router.post("/api/admin/reload-config") +def reload_config(): + """Re-read ``config/payers.yaml`` and revalidate. Returns counts.""" + from cyclone import payers as payer_loader + try: + configs = payer_loader.load_payer_configs() + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + return {"ok": True, "loaded": len(configs), "errors": []} diff --git a/backend/tests/test_api_rotate_key.py b/backend/tests/test_api_rotate_key.py index 5e094ae..dcdb4ca 100644 --- a/backend/tests/test_api_rotate_key.py +++ b/backend/tests/test_api_rotate_key.py @@ -82,7 +82,9 @@ class TestRotateKeyEndpointWiring: 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) + # SP36 Task 3: this endpoint moved from cyclone.api to + # cyclone.api_routers.admin; patch the live import surface. + monkeypatch.setattr("cyclone.api_routers.admin._db_crypto.rotate_db_key", _stub_rotate_ok) db.init_db() yield db_file, fake_kc db._reset_for_tests() @@ -151,7 +153,7 @@ class TestRotateKeyEndpointWiring: rotated_at=datetime.now(timezone.utc).isoformat(), reason="simulated PRAGMA rekey failure", ) - monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", _fail_rotate) + monkeypatch.setattr("cyclone.api_routers.admin._db_crypto.rotate_db_key", _fail_rotate) _, fake_kc = _fake_encrypted_env before = dict(fake_kc) @@ -187,7 +189,8 @@ class TestRotateKeyEndpointWiring: 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) + # SP36 Task 3: endpoint moved to cyclone.api_routers.admin. + monkeypatch.setattr("cyclone.api_routers.admin._secrets.set_secret", lambda n, v: False) from fastapi.testclient import TestClient from cyclone.api import app @@ -202,10 +205,11 @@ class TestRotateKeyEndpointWiring: """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, + "cyclone.api_routers.admin._secrets.set_secret", lambda n, v: True, ) - from cyclone import api as api_mod - api_mod._db_rotate_lock.acquire() + # SP36 Task 3: lock moved with the endpoint into admin router. + from cyclone.api_routers import admin as admin_mod + admin_mod._db_rotate_lock.acquire() try: from fastapi.testclient import TestClient from cyclone.api import app @@ -214,4 +218,4 @@ class TestRotateKeyEndpointWiring: assert r.status_code == 409 assert "in progress" in r.json()["detail"] finally: - api_mod._db_rotate_lock.release() + admin_mod._db_rotate_lock.release()