Files
praxis/server/operator/credentials.py
T
Praxis CI bdcf793db2 feat(P02): complete integration + tech-debt + NFR measurement phase — v0.1.12 tagged
Phase 2 (Integration + Tech-Debt + NFR Measurement) complete.
4 slices, 2 waves, 9 tasks. 4 REQs covered. 60 new tests (469 total).
8 v0.4 P1+ tech-debt findings addressed. Verify: APPROVE_WITH_NOTES.

NFR measurement (p95 latency + guardrail FP/FN), cohort aggregation
assist metrics (5 new metrics, no schema change), assist cost tracking
+ C-3 budget check, tech-debt wave (argon2id offload, cookie-secret
validation, credential enum, f-string SQL, cache persistence, zoneinfo,
audit log, 429 mock).

---ci---
project: praxis
phase: 2
milestone: v0.5
status: complete
requirements:
  covered: [REQ-NFR-ASSIST-01, REQ-IDEATE-04, REQ-IDEATE-06, REQ-IDEATE-07]
  partial: []
---/ci---
2026-08-04 22:11:51 +00:00

86 lines
2.7 KiB
Python

"""GET/POST /api/operator/credentials — VC management (TASK-08-04, D-057).
Auth-gated. GET lists issued VCs from Postgres issued_credentials (operator's
issuance log). POST /{id}/revoke revokes a VC (status='revoked',
revoked_at=now()). Revoked credentials fail verification. No PII beyond what
the credential asserts (D-043).
"""
from __future__ import annotations
import datetime as _dt
import logging
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel
from server.auth.dependencies import current_operator
from server.auth.models import Operator
from server.operator._common import require_pg_store
router = APIRouter(prefix="/api/operator", tags=["operator-credentials"])
log = logging.getLogger(__name__)
class CredentialOut(BaseModel):
id: str
learner_ref: str
vc_type: str | None = None
status: str
issued_at: _dt.datetime | None = None
revoked_at: _dt.datetime | None = None
class CredentialListResponse(BaseModel):
credentials: list[CredentialOut]
class OkResponse(BaseModel):
ok: bool = True
id: str
status: str
@router.get("/credentials", response_model=CredentialListResponse)
async def list_credentials(
request: Request,
op: Operator = Depends(current_operator),
) -> CredentialListResponse:
pg_store = await require_pg_store(request)
rows = await pg_store.list_credentials()
creds = [
CredentialOut(
id=str(r["id"]),
learner_ref=r["learner_ref"],
vc_type=r.get("vc_type"),
status=r.get("status", "active"),
issued_at=r.get("issued_at"),
revoked_at=r.get("revoked_at"),
)
for r in rows
]
return CredentialListResponse(credentials=creds)
@router.post("/credentials/{cred_id}/revoke", response_model=OkResponse)
async def revoke_credential(
cred_id: str,
request: Request,
op: Operator = Depends(current_operator),
) -> OkResponse:
pg_store = await require_pg_store(request)
row = await pg_store.get_credential(cred_id)
if row is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail="credential not found")
await pg_store.set_credential_status(cred_id, "revoked")
# TASK-12-04 (P1+ #5): application-level audit log for credential revocation.
# The revoking operator_id + cred_id are logged. No audit_log table (the
# log is sufficient for pilot — D-056 stateless cookies + revoked_at
# timestamp are the primary audit trail).
log.info("credential revoked: operator=%s cred_id=%s", op.id, cred_id)
return OkResponse(ok=True, id=cred_id, status="revoked")
__all__ = ["router"]