feat(sp36): absorb 20 admin endpoints into api_routers/admin.py
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.
This commit is contained in:
Nora
@@ -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 <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()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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) #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@@ -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
|
||||
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 <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()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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": []}
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user