docs(milestone): complete v0.4-operator-tier — v0.1.9 tagged, milestone release, merged to main
v0.4 (Operator Tier — Cohort Dashboard + Auth + Postgres) milestone complete. Phases: ✓ P0 pre-execution (planning) → v0.1.6 ✓ P1 operator foundation (Postgres+auth+VC migration) → v0.1.7 ✓ P2 cohort dashboard + aggregation → v0.1.8 ✓ P3 final review + ship → v0.1.9 (= v0.4 milestone release) Requirements covered (8/8): REQ-MT-01 (Postgres store), REQ-MT-02 (aggregation pipeline), REQ-AUTH-01 (operator auth), REQ-DASH-01 (cohort dashboard), REQ-NFR-AUTH-01 (auth NFRs), REQ-NFR-MT-01 (Postgres-in-LXC), REQ-NFR-DASH-01 (k-anonymity ≥10), REQ-NFR-DASH-02 (freshness ≤24h) Grill MUSTs honored (6/6): G-008, G-011, G-027, G-031, G-038, G-041 Tests: 317 pytest pass, 36 skip (Postgres-requiring), 0 fail; 17/17 vitest pass Review: APPROVE_WITH_NOTES (6/6 personas, 0 P0, 8 P1+ carry-forward) Audit: HEALTHY (reconstruction PASS, 8/8 REQ, 6/6 grill) ---ci--- project: praxis phase: 3 milestone: v0.4 status: complete phase_role: final milestone_complete: true milestone_merged_to_main: true tag: v0.1.9 requirements: covered: [REQ-MT-01, REQ-MT-02, REQ-AUTH-01, REQ-DASH-01, REQ-NFR-AUTH-01, REQ-NFR-MT-01, REQ-NFR-DASH-01, REQ-NFR-DASH-02] partial: [] ---/ci---
This commit is contained in:
Praxis CI
+166
-15
@@ -14,6 +14,7 @@ no audio/no tokens at runtime, not a crash.
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import Any
|
||||
|
||||
from loguru import logger
|
||||
@@ -27,14 +28,32 @@ try:
|
||||
except ImportError: # pragma: no cover
|
||||
pass
|
||||
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.responses import JSONResponse
|
||||
from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection
|
||||
from slowapi.errors import RateLimitExceeded
|
||||
from slowapi import _rate_limit_exceeded_handler
|
||||
|
||||
from db.pg_migrate import apply_pg_migrations
|
||||
from db.pg_store import PgStore
|
||||
from db.store import PraxisStore
|
||||
from server.auth.cookies import get_session_middleware_kwargs
|
||||
from server.auth.rate_limit import limiter
|
||||
from server.auth.routes import router as auth_router
|
||||
from server.cohort.nightly import NightlyScheduler
|
||||
from server.operator.cohort import router as cohort_router
|
||||
from server.operator.credentials import router as credentials_router
|
||||
from server.operator.failure_patterns import router as failure_router
|
||||
from server.operator.mastery import router as mastery_router
|
||||
from server.pipeline import build_pipeline
|
||||
from server.vc.issuer_keys import _load_root_key
|
||||
from server.vc.migrate_keys import migrate_issuer_keys
|
||||
from server.vc.verification import verify_credential
|
||||
from starlette.middleware.sessions import SessionMiddleware
|
||||
from starlette.responses import FileResponse
|
||||
from starlette.staticfiles import StaticFiles
|
||||
from starlette.exceptions import HTTPException as StarletteHTTPException
|
||||
|
||||
_store = PraxisStore()
|
||||
|
||||
@@ -47,6 +66,64 @@ HOST = _env("PRAXIS_HOST", "0.0.0.0")
|
||||
PORT = int(_env("PRAXIS_PORT", "8789"))
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
"""v0.4 — create the asyncpg Postgres pool on startup, close on shutdown.
|
||||
|
||||
Graceful degradation (D-050, REQ-NFR-MT-01): if PRAXIS_PG_DSN is unset,
|
||||
the server starts without Postgres — the learner voice loop (SQLite)
|
||||
is unaffected. app.state.pg_pool / app.state.pg_store are None in that
|
||||
case and auth/operator routes return 503.
|
||||
"""
|
||||
dsn = os.environ.get("PRAXIS_PG_DSN", "").strip()
|
||||
if not dsn:
|
||||
logger.warning(
|
||||
"PRAXIS_PG_DSN not set — starting without Postgres (dev/no-pool mode). "
|
||||
"Operator auth + cohort endpoints will be unavailable (503). "
|
||||
"Learner voice loop (SQLite) is unaffected."
|
||||
)
|
||||
app.state.pg_pool = None
|
||||
app.state.pg_store = None
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
return
|
||||
import asyncpg
|
||||
|
||||
logger.info("Creating asyncpg Postgres pool (min=1, max=10, D-050)")
|
||||
pool = await asyncpg.create_pool(
|
||||
dsn=dsn,
|
||||
min_size=1,
|
||||
max_size=10,
|
||||
command_timeout=10,
|
||||
)
|
||||
app.state.pg_pool = pool
|
||||
app.state.pg_store = PgStore(pool)
|
||||
nightly = NightlyScheduler()
|
||||
app.state.nightly_scheduler = nightly
|
||||
try:
|
||||
applied = await apply_pg_migrations(pool)
|
||||
if applied:
|
||||
logger.info(f"Postgres migrations applied: {applied}")
|
||||
else:
|
||||
logger.info("Postgres migrations up to date")
|
||||
# VC key migration (TASK-06-03, R-VC-MIG-01, G-027) — runs once on
|
||||
# first boot, idempotent. Non-fatal on failure (v0.3 SQLite path
|
||||
# remains intact for verification).
|
||||
await _maybe_migrate_issuer_keys()
|
||||
# v0.4 P2 (D-054, REQ-NFR-DASH-02): start the nightly reconciliation
|
||||
# scheduler at 03:00 CT. Cancelled on shutdown.
|
||||
await nightly.start(app.state.pg_store)
|
||||
logger.info("Nightly cohort reconciliation scheduler started (03:00 CT)")
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
await nightly.stop()
|
||||
finally:
|
||||
await pool.close()
|
||||
logger.info("Postgres pool closed")
|
||||
|
||||
|
||||
class WebRTCOffer(BaseModel):
|
||||
"""Client→server WebRTC offer (SDP + type)."""
|
||||
|
||||
@@ -54,13 +131,19 @@ class WebRTCOffer(BaseModel):
|
||||
type: str = "offer"
|
||||
|
||||
|
||||
app = FastAPI(title="Praxis v0.1 voice server", version="0.1.0")
|
||||
app = FastAPI(title="Praxis v0.1 voice server", version="0.1.0", lifespan=lifespan)
|
||||
# slowapi rate-limit state + 429 handler (D-041, TASK-03-03).
|
||||
app.state.limiter = limiter
|
||||
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"], # dev — the client is a separate Vite origin
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
# SessionMiddleware (signed cookies, D-056) — added AFTER CORS so it is
|
||||
# the outermost middleware (signs cookies before CORS headers are added).
|
||||
app.add_middleware(SessionMiddleware, **get_session_middleware_kwargs())
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
@@ -123,28 +206,96 @@ async def webrtc_offer(offer: WebRTCOffer) -> dict[str, str]:
|
||||
|
||||
@app.get("/vc/verify/{credential_id}")
|
||||
async def vc_verify(credential_id: str) -> dict[str, Any]:
|
||||
"""Public, unauthenticated VC verification endpoint (D-043).
|
||||
"""Public, unauthenticated VC verification endpoint (D-043, G-011).
|
||||
|
||||
Returns {valid, status, issuer, credential, mastery, credentialTier,
|
||||
verifiedAt}. 404 if the credential id is not found. No PII beyond what
|
||||
the credential asserts.
|
||||
Two-store fallback (G-011, binding contract):
|
||||
(a) If Postgres is available (app.state.pg_store), use it for issuer
|
||||
key lookup (active + superseded keys).
|
||||
(b) If the credential is not in Postgres issued_credentials, fall back
|
||||
to SQLite (v0.3 credentials remain in SQLite — D-051).
|
||||
(c) If Postgres is NOT available, use the v0.3 SQLite path for both.
|
||||
The VC key migration (TASK-04-03) runs once on first boot (idempotent)
|
||||
inside the lifespan — see _maybe_migrate_issuer_keys.
|
||||
"""
|
||||
await _store.init()
|
||||
result = await verify_credential(_store, credential_id)
|
||||
pg_store = getattr(app.state, "pg_store", None)
|
||||
result = await verify_credential(
|
||||
_store, credential_id,
|
||||
pg_store=pg_store, sqlite_store=_store,
|
||||
)
|
||||
if result is None:
|
||||
raise HTTPException(status_code=404, detail="credential not found")
|
||||
return result
|
||||
|
||||
|
||||
# ── Static client serving (D-023, REQ-DEPLOY-13) ────────────────────
|
||||
# Mount client/dist as StaticFiles at "/" AFTER all API routes so they
|
||||
# take precedence. html=True serves index.html for "/" (SPA root).
|
||||
# The client has no React Router (single-view state machine: start→live
|
||||
# →debrief), so no SPA fallback fallback route is needed per RESEARCH.md Q3.
|
||||
async def _maybe_migrate_issuer_keys() -> None:
|
||||
"""Run the VC key migration on first boot (TASK-06-03, R-VC-MIG-01).
|
||||
|
||||
Idempotent — no-op if Postgres already has an active issuer key. G-027:
|
||||
if SQLite has no v0.3 active key (fresh deploy), skips archive and only
|
||||
generates a fresh v0.4 keypair.
|
||||
"""
|
||||
pg_store = getattr(app.state, "pg_store", None)
|
||||
if pg_store is None:
|
||||
return
|
||||
try:
|
||||
await _store.init()
|
||||
root_key = _load_root_key()
|
||||
result = await migrate_issuer_keys(_store, pg_store, root_key)
|
||||
if result["new_key_id"] is not None:
|
||||
logger.info(
|
||||
f"VC key migration: archived v0.3 key={result['archived_key_id']}, "
|
||||
f"generated fresh v0.4 key={result['new_key_id']}"
|
||||
)
|
||||
else:
|
||||
logger.info("VC key migration: active key already present (no-op)")
|
||||
except Exception as exc:
|
||||
logger.error(f"VC key migration failed (non-fatal — v0.3 path intact): {exc}")
|
||||
|
||||
|
||||
# ── Operator auth routes (TASK-06-02, D-057) ───────────────────────────
|
||||
# Mounted BEFORE the StaticFiles mount so /api/operator/* is matched by
|
||||
# the router (routes-before-static-mount constraint, carry-forward v0.2).
|
||||
app.include_router(auth_router)
|
||||
|
||||
# ── Operator API cohort endpoints (TASK-10-02, D-053, D-057) ──────────
|
||||
# Auth-gated via Depends(current_operator) inside each router. Mounted
|
||||
# BEFORE the SPA StaticFiles fallback so /api/operator/* is matched by the
|
||||
# API routers, not the SPA fallback.
|
||||
app.include_router(cohort_router)
|
||||
app.include_router(mastery_router)
|
||||
app.include_router(failure_router)
|
||||
app.include_router(credentials_router)
|
||||
|
||||
|
||||
# ── SPA StaticFiles fallback (G-041 binding, TASK-10-01, R-DASH-03/05) ─
|
||||
# Custom StaticFiles subclass that returns index.html for non-file paths
|
||||
# (SPA client-side routing). G-041 OVERRIDES the plan's catch-all route —
|
||||
# a @app.get("/{path:path}") catch-all before StaticFiles would shadow
|
||||
# asset serving (assertion 8 in TASK-10-04). This subclass serves assets
|
||||
# normally (JS/CSS) and falls back to index.html for client-side routes
|
||||
# (/operator/dashboard, /operator/login). API routes registered above take
|
||||
# precedence over the mount.
|
||||
class SpaStaticFiles(StaticFiles):
|
||||
async def get_response(self, path: str, scope):
|
||||
try:
|
||||
return await super().get_response(path, scope)
|
||||
except (StarletteHTTPException, HTTPException) as e:
|
||||
if getattr(e, "status_code", None) == 404:
|
||||
import os
|
||||
index = os.path.join(self.directory, "index.html")
|
||||
if os.path.isfile(index):
|
||||
return FileResponse(index)
|
||||
raise
|
||||
|
||||
|
||||
# Mount client/dist at "/" AFTER all API routes so they take precedence.
|
||||
# html=True serves index.html for "/" (SPA root). The SpaStaticFiles
|
||||
# subclass serves index.html for unknown paths (React Router routes).
|
||||
_CLIENT_DIST = _env("PRAXIS_CLIENT_DIST", "client/dist")
|
||||
if os.path.isdir(_CLIENT_DIST):
|
||||
app.mount("/", StaticFiles(directory=_CLIENT_DIST, html=True), name="client")
|
||||
logger.info(f"Serving client from {_CLIENT_DIST}")
|
||||
app.mount("/", SpaStaticFiles(directory=_CLIENT_DIST, html=True), name="spa")
|
||||
logger.info(f"Serving client from {_CLIENT_DIST} (SPA fallback enabled)")
|
||||
else:
|
||||
logger.warning(f"Client dist not found at {_CLIENT_DIST} — API-only mode")
|
||||
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
"""Signed cookie configuration (TASK-03-02, D-041, D-056, R-AUTH-01, G-031).
|
||||
|
||||
Returns kwargs for Starlette SessionMiddleware (itsdangerous HMAC-SHA256
|
||||
signed cookies — D-056, stateless, no sessions table). The cookie name is
|
||||
`praxis_op` (distinct from any future learner cookie).
|
||||
|
||||
R-AUTH-01 / G-031 reframe: the PRIMARY mitigation for a sniffed operator
|
||||
cookie is the k-anonymity defense-in-depth — the cohort dashboard reads
|
||||
only k-anonymized aggregates, so a sniffed cookie leaks NO learner PII.
|
||||
The `PRAXIS_COOKIE_SECURE` flag is the SECONDARY mitigation (operational
|
||||
convenience for when TLS arrives). It defaults to true; the HTTP pilot
|
||||
(LXC, no TLS — D-030) sets it to false with a logged WARNING.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import secrets
|
||||
|
||||
from loguru import logger
|
||||
|
||||
_COOKIE_MAX_AGE_S = 28800 # 8h (D-041)
|
||||
|
||||
|
||||
def _env_bool(key: str, default: bool) -> bool:
|
||||
raw = os.environ.get(key, "").strip().lower()
|
||||
if raw in ("true", "1", "yes", "on"):
|
||||
return True
|
||||
if raw in ("false", "0", "no", "off"):
|
||||
return False
|
||||
return default
|
||||
|
||||
|
||||
def get_session_middleware_kwargs() -> dict:
|
||||
"""Return kwargs for Starlette SessionMiddleware.
|
||||
|
||||
If PRAXIS_COOKIE_SECRET is unset, generate an ephemeral random secret
|
||||
and log a WARNING (dev only — sessions won't survive a restart and this
|
||||
MUST NOT be used in pilot/production).
|
||||
"""
|
||||
secret = os.environ.get("PRAXIS_COOKIE_SECRET", "").strip()
|
||||
if not secret:
|
||||
secret = secrets.token_urlsafe(48)
|
||||
logger.warning(
|
||||
"PRAXIS_COOKIE_SECRET not set — generated an ephemeral random secret. "
|
||||
"Sessions will NOT survive a server restart. This is dev-only; set "
|
||||
"PRAXIS_COOKIE_SECRET (>=32 bytes) for pilot/production."
|
||||
)
|
||||
secure = _env_bool("PRAXIS_COOKIE_SECURE", True)
|
||||
if not secure:
|
||||
logger.warning(
|
||||
"Cookie Secure flag disabled (PRAXIS_COOKIE_SECURE=false) — HTTP pilot "
|
||||
"mode (R-AUTH-01). Do not use in production. NOTE (G-031): the primary "
|
||||
"R-AUTH-01 mitigation is k-anon defense-in-depth (cohort dashboard reads "
|
||||
"only k-anonymized aggregates → sniffed cookie leaks no PII); this flag "
|
||||
"is the secondary mitigation."
|
||||
)
|
||||
return {
|
||||
"secret_key": secret,
|
||||
"session_cookie": "praxis_op",
|
||||
"max_age": _COOKIE_MAX_AGE_S,
|
||||
"https_only": secure,
|
||||
"same_site": "strict",
|
||||
"path": "/",
|
||||
}
|
||||
|
||||
|
||||
__all__ = ["get_session_middleware_kwargs"]
|
||||
@@ -0,0 +1,56 @@
|
||||
"""current_operator dependency (TASK-03-04, D-057).
|
||||
|
||||
Server-side auth enforcement: every `/api/operator/*` protected route uses
|
||||
`Depends(current_operator)`. The dependency NEVER trusts the client (D-057)
|
||||
— it reads the signed-cookie session, fetches the operator from Postgres,
|
||||
and 401s on any gap (missing/invalid/expired cookie, unknown id, inactive
|
||||
operator). The cookie is the authz *token*; the Postgres lookup is the
|
||||
authz *decision*.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import HTTPException, Request, status
|
||||
|
||||
from server.auth.models import Operator
|
||||
|
||||
|
||||
async def current_operator(request: Request) -> Operator:
|
||||
"""Resolve the authenticated operator from the signed-cookie session.
|
||||
|
||||
Raises 401 on: missing session, missing operator_id, no Postgres store
|
||||
(503 actually — operator tier unavailable), unknown operator id, or an
|
||||
inactive operator (session is cleared in the latter case so the client
|
||||
cookie is invalidated).
|
||||
"""
|
||||
pg_store = getattr(request.app.state, "pg_store", None)
|
||||
if pg_store is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="operator tier unavailable (no Postgres)",
|
||||
)
|
||||
session = request.session
|
||||
op_id = session.get("operator_id") if session else None
|
||||
if not op_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="not authenticated",
|
||||
)
|
||||
row = await pg_store.get_operator_by_id(op_id)
|
||||
if row is None or not row.get("is_active"):
|
||||
# Inactive/unknown → clear the session so the cookie is invalidated.
|
||||
if session:
|
||||
session.clear()
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="not authenticated",
|
||||
)
|
||||
return Operator(
|
||||
id=str(row["id"]),
|
||||
username=row["username"],
|
||||
display_name=row.get("display_name"),
|
||||
role=row.get("role", "operator"),
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["current_operator"]
|
||||
@@ -0,0 +1,18 @@
|
||||
"""Auth data models (TASK-03-04)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Operator:
|
||||
"""The authenticated operator injected into protected routes (D-057)."""
|
||||
|
||||
id: str
|
||||
username: str
|
||||
display_name: str | None
|
||||
role: str
|
||||
|
||||
|
||||
__all__ = ["Operator"]
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Argon2id password hashing (TASK-03-01, D-041, REQ-NFR-AUTH-01).
|
||||
|
||||
Uses argon2-cffi PasswordHasher with defaults that exceed OWASP minimums
|
||||
(time_cost=3, memory_cost=64MiB, parallelism=4 — RESEARCH-v0.4 §2.1).
|
||||
Single operator, low-frequency logins → hashing latency < 1s is
|
||||
acceptable (R-AUTH-02).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from argon2 import PasswordHasher
|
||||
from argon2.exceptions import VerifyMismatchError
|
||||
|
||||
_ph = PasswordHasher()
|
||||
|
||||
|
||||
def hash_password(plain: str) -> str:
|
||||
"""Hash a plaintext password with argon2id. Returns the encoded hash string."""
|
||||
return _ph.hash(plain)
|
||||
|
||||
|
||||
def verify_password(stored_hash: str, plain: str) -> bool:
|
||||
"""Verify a plaintext password against a stored argon2id hash.
|
||||
|
||||
Returns False on mismatch (no exception) so the login flow can apply a
|
||||
uniform 401 + rate-limit-increment path on any auth failure.
|
||||
"""
|
||||
try:
|
||||
_ph.verify(stored_hash, plain)
|
||||
return True
|
||||
except VerifyMismatchError:
|
||||
return False
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def needs_rehash(stored_hash: str) -> bool:
|
||||
"""True if the stored hash was produced with weaker params than the
|
||||
current PasswordHasher defaults. The login flow rehashes + updates the
|
||||
store when this returns True (param upgrades without forcing a reset)."""
|
||||
return _ph.check_needs_rehash(stored_hash)
|
||||
|
||||
|
||||
__all__ = ["hash_password", "verify_password", "needs_rehash"]
|
||||
@@ -0,0 +1,34 @@
|
||||
"""Login rate limiting (TASK-03-03, D-041).
|
||||
|
||||
slowapi Limiter with an in-memory backend (single-instance — D-041).
|
||||
5 login attempts per minute per client IP. On exceed → 429 + Retry-After.
|
||||
|
||||
R-AUTH-03 (in-memory counter lost on restart) is an accepted pilot risk
|
||||
(RESEARCH-v0.4 §2.5) — a restart at most resets the counter, which slightly
|
||||
widens the brute-force window but does not enable it (argon2id + 5/min is
|
||||
still the binding control). A hand-rolled counter is the documented
|
||||
fallback if slowapi is ever removed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from slowapi import Limiter
|
||||
from slowapi.util import get_remote_address
|
||||
|
||||
limiter = Limiter(key_func=get_remote_address, storage_uri="memory://")
|
||||
|
||||
|
||||
def reset_login_rate_limit() -> None:
|
||||
"""Clear the in-memory rate-limit counters (test helper + restart-safe)."""
|
||||
try:
|
||||
limiter.reset()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def rate_limit_login():
|
||||
"""Decorator factory: 5 login attempts per minute per IP (D-041)."""
|
||||
return limiter.limit("5/minute")
|
||||
|
||||
|
||||
__all__ = ["limiter", "rate_limit_login", "reset_login_rate_limit"]
|
||||
@@ -0,0 +1,118 @@
|
||||
"""Auth route handlers — login, logout, me (TASK-03-05, D-041, D-056, D-057).
|
||||
|
||||
APIRouter(prefix="/api/operator") with:
|
||||
POST /login — rate-limited 5/min (TASK-03-03), NOT auth-gated.
|
||||
POST /logout — auth-gated (Depends(current_operator)).
|
||||
GET /me — auth-gated (React route guard — D-057).
|
||||
|
||||
Stateless cookies (D-056): logout clears the server-side session; the
|
||||
client also clears its cookie. No sessions table.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
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.auth.passwords import hash_password, needs_rehash, verify_password
|
||||
from server.auth.rate_limit import rate_limit_login
|
||||
|
||||
router = APIRouter(prefix="/api/operator", tags=["operator-auth"])
|
||||
|
||||
|
||||
class LoginBody(BaseModel):
|
||||
username: str
|
||||
password: str
|
||||
|
||||
|
||||
class OperatorOut(BaseModel):
|
||||
id: str
|
||||
username: str
|
||||
display_name: str | None
|
||||
role: str = "operator"
|
||||
|
||||
|
||||
class LoginResponse(BaseModel):
|
||||
operator: OperatorOut
|
||||
|
||||
|
||||
class MeResponse(BaseModel):
|
||||
operator: OperatorOut
|
||||
|
||||
|
||||
class OkResponse(BaseModel):
|
||||
ok: bool = True
|
||||
|
||||
|
||||
def _operator_out(op: Operator) -> OperatorOut:
|
||||
return OperatorOut(
|
||||
id=op.id,
|
||||
username=op.username,
|
||||
display_name=op.display_name,
|
||||
role=op.role,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/login", response_model=LoginResponse)
|
||||
@rate_limit_login()
|
||||
async def login(body: LoginBody, request: Request) -> LoginResponse:
|
||||
"""Rate-limited login (5/min per IP — D-041).
|
||||
|
||||
On success: sets `request.session["operator_id"]` (signed cookie via
|
||||
SessionMiddleware) + updates last_login_at. On needs_rehash → rehash +
|
||||
update the store. On failure → 401 (no cookie set).
|
||||
"""
|
||||
pg_store = getattr(request.app.state, "pg_store", None)
|
||||
if pg_store is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="operator tier unavailable (no Postgres)",
|
||||
)
|
||||
row = await pg_store.get_operator_by_username(body.username)
|
||||
if row is None or not row.get("is_active"):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="invalid credentials",
|
||||
)
|
||||
if not verify_password(row["password_hash"], body.password):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="invalid credentials",
|
||||
)
|
||||
op_id = str(row["id"])
|
||||
request.session["operator_id"] = op_id
|
||||
await pg_store.update_last_login(op_id)
|
||||
if needs_rehash(row["password_hash"]):
|
||||
new_hash = hash_password(body.password)
|
||||
async with pg_store.pool.acquire() as conn:
|
||||
await conn.execute(
|
||||
"UPDATE operators SET password_hash = $1 WHERE id = $2",
|
||||
new_hash, op_id,
|
||||
)
|
||||
return LoginResponse(
|
||||
operator=OperatorOut(
|
||||
id=op_id,
|
||||
username=row["username"],
|
||||
display_name=row.get("display_name"),
|
||||
role=row.get("role", "operator"),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@router.post("/logout", response_model=OkResponse)
|
||||
async def logout(request: Request, op: Operator = Depends(current_operator)) -> OkResponse:
|
||||
# Stateless (D-056): clearing the server session invalidates the signed
|
||||
# cookie's payload; the client also clears its cookie.
|
||||
request.session.clear()
|
||||
return OkResponse(ok=True)
|
||||
|
||||
|
||||
@router.get("/me", response_model=MeResponse)
|
||||
async def me(op: Operator = Depends(current_operator)) -> MeResponse:
|
||||
"""React route guard endpoint (D-057). 200 → render; 401 → redirect."""
|
||||
return MeResponse(operator=_operator_out(op))
|
||||
|
||||
|
||||
__all__ = ["router"]
|
||||
@@ -0,0 +1,230 @@
|
||||
"""Cohort aggregation logic + k-anonymity suppression (TASK-07-01, D-034, D-045).
|
||||
|
||||
Computes k-anonymized aggregates for the affected (path, metric, window_start)
|
||||
bins and upserts them to cohort_aggregates via PgStore. Suppression is at
|
||||
write time (auditable — RESEARCH-v0.4 §3.1): COUNT(DISTINCT learner_ref) < 10
|
||||
=> cell_suppressed=TRUE, value=NULL.
|
||||
|
||||
Metrics computed (per 7-day rolling window, per path):
|
||||
sessions_count, active_learners_count, gate_open_rate,
|
||||
median_mastery_score, failure_mode_frequency,
|
||||
rubric_criterion_means, week_distribution.
|
||||
|
||||
The session_outcome dict contains: learner_ref (opaque — D-031), path,
|
||||
scenario_id, outcome (pass/fail), rubric_scores, failure_mode, branch_path,
|
||||
timestamp.
|
||||
|
||||
No raw learner PII in Postgres (D-031): only aggregates + opaque learner_ref
|
||||
for distinct counting.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
import logging
|
||||
import statistics
|
||||
from typing import Any
|
||||
|
||||
from db.pg_store import PgStore
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
K_ANON_THRESHOLD = 10
|
||||
|
||||
|
||||
def _rolling_window(now: _dt.datetime | None = None) -> tuple[_dt.date, _dt.date]:
|
||||
"""Return the 7-day rolling window (start, end) for `now`.
|
||||
|
||||
window_start = today - 6 days, window_end = today (inclusive 7-day span).
|
||||
"""
|
||||
today = (now or _dt.datetime.now(_dt.timezone.utc)).date()
|
||||
return today - _dt.timedelta(days=6), today
|
||||
|
||||
|
||||
def _distinct_learners(sessions: list[dict[str, Any]]) -> int:
|
||||
return len({s["learner_ref"] for s in sessions if s.get("learner_ref")})
|
||||
|
||||
|
||||
async def aggregate_session(pg_store: PgStore, session_outcome: dict[str, Any]) -> None:
|
||||
"""Compute + upsert k-anonymized aggregates for one session outcome.
|
||||
|
||||
Reads the affected path's recent session set (from cohort_aggregates or
|
||||
an in-memory accumulator), recomputes the metric cells for the 7-day
|
||||
window, applies k-anon suppression, and upserts each cell idempotently.
|
||||
|
||||
Idempotent (ON CONFLICT upsert) — re-running with the same outcome
|
||||
produces the same aggregate. The caller (hook.py) passes one session at
|
||||
a time; the nightly job (nightly.py) recomputes the full window.
|
||||
"""
|
||||
path = session_outcome.get("path") or session_outcome.get("path_id") or "unknown"
|
||||
learner_ref = session_outcome.get("learner_ref") or "unknown"
|
||||
outcome = session_outcome.get("outcome", "fail")
|
||||
rubric_scores = session_outcome.get("rubric_scores") or []
|
||||
failure_mode = session_outcome.get("failure_mode")
|
||||
branch_path = session_outcome.get("branch_path") or []
|
||||
scenario_id = session_outcome.get("scenario_id")
|
||||
ts = session_outcome.get("timestamp")
|
||||
|
||||
window_start, window_end = _rolling_window(
|
||||
_dt.datetime.fromisoformat(ts) if isinstance(ts, str) else None
|
||||
)
|
||||
|
||||
# Distinct-learner count for k-anon: this session's learner + any others
|
||||
# already recorded for the same (path, window). For the per-session hook
|
||||
# we accumulate by appending to a sessions_count cell + tracking distinct
|
||||
# learner_refs via active_learners_count. The nightly job recomputes from
|
||||
# the mastery_gate_events + session log (full reconciliation).
|
||||
#
|
||||
# For the on-session-end hook we cannot cheaply know all distinct learners
|
||||
# without a raw-events table (which we deliberately do not maintain for PII
|
||||
# reasons — D-031). We instead maintain a single active_learners_count
|
||||
# counter per (path, window) and the nightly job reconciles the true
|
||||
# distinct count from mastery_gate_events. The hook uses the running
|
||||
# counter; if it is < K_ANON_THRESHOLD we suppress.
|
||||
active_count = await _bump_active_learners(pg_store, path, window_start, learner_ref)
|
||||
sessions_count = await _bump_counter(pg_store, path, "sessions_count", window_start, window_end)
|
||||
|
||||
suppressed = active_count < K_ANON_THRESHOLD
|
||||
|
||||
await _upsert_cell(pg_store, path, "sessions_count", window_start, window_end,
|
||||
float(sessions_count) if not suppressed else None,
|
||||
active_count, suppressed)
|
||||
|
||||
await _upsert_cell(pg_store, path, "active_learners_count", window_start, window_end,
|
||||
float(active_count) if not suppressed else None,
|
||||
active_count, suppressed)
|
||||
|
||||
# gate_open_rate: 1.0 if this session passed, 0.0 otherwise (running mean
|
||||
# reconciled by nightly). Stored as the fraction of pass outcomes seen.
|
||||
passed = 1.0 if outcome == "pass" else 0.0
|
||||
gate_open_rate = await _running_mean(pg_store, path, "gate_open_rate",
|
||||
window_start, window_end, passed, active_count)
|
||||
await _upsert_cell(pg_store, path, "gate_open_rate", window_start, window_end,
|
||||
gate_open_rate if not suppressed else None,
|
||||
active_count, suppressed)
|
||||
|
||||
# median_mastery_score (from rubric scores) — running median reconciled nightly
|
||||
if rubric_scores:
|
||||
scores = [float(r.get("score", r.get("weighted_mean", 0.0))) for r in rubric_scores]
|
||||
scenario_mean = statistics.mean(scores) if scores else 0.0
|
||||
median_val = await _running_mean(pg_store, path, "median_mastery_score",
|
||||
window_start, window_end, scenario_mean, active_count)
|
||||
await _upsert_cell(pg_store, path, "median_mastery_score", window_start, window_end,
|
||||
median_val if not suppressed else None,
|
||||
active_count, suppressed)
|
||||
|
||||
# rubric_criterion_means — one cell per criterion id
|
||||
for r in rubric_scores:
|
||||
cid = r.get("criterion_id") or r.get("id") or "unknown"
|
||||
score = float(r.get("score", 0.0))
|
||||
mean_val = await _running_mean(pg_store, path, f"rubric_criterion_mean:{cid}",
|
||||
window_start, window_end, score, active_count)
|
||||
await _upsert_cell(pg_store, path, f"rubric_criterion_mean:{cid}",
|
||||
window_start, window_end,
|
||||
mean_val if not suppressed else None,
|
||||
active_count, suppressed)
|
||||
|
||||
# failure_mode_frequency — one cell per observed mode
|
||||
if failure_mode:
|
||||
freq = await _bump_mode_counter(pg_store, path, f"failure_mode:{failure_mode}",
|
||||
window_start, window_end)
|
||||
await _upsert_cell(pg_store, path, f"failure_mode:{failure_mode}",
|
||||
window_start, window_end,
|
||||
float(freq) if not suppressed else None,
|
||||
active_count, suppressed)
|
||||
|
||||
# week_distribution — branch_path captures the path-week; record one cell
|
||||
# per branch outcome seen.
|
||||
if branch_path:
|
||||
last_branch = branch_path[-1] if isinstance(branch_path, list) else str(branch_path)
|
||||
freq = await _bump_mode_counter(pg_store, path, f"branch:{last_branch}",
|
||||
window_start, window_end)
|
||||
await _upsert_cell(pg_store, path, f"branch:{last_branch}",
|
||||
window_start, window_end,
|
||||
float(freq) if not suppressed else None,
|
||||
active_count, suppressed)
|
||||
|
||||
log.debug(
|
||||
"aggregate_session path=%s learner=%s outcome=%s window=%s..%s "
|
||||
"active=%d suppressed=%s",
|
||||
path, learner_ref, outcome, window_start, window_end,
|
||||
active_count, suppressed,
|
||||
)
|
||||
|
||||
|
||||
# ── Internal cell upsert + counter helpers ──────────────────────────────────
|
||||
# The PgStore.upsert_cohort_aggregate is idempotent (ON CONFLICT). We use a
|
||||
# small in-memory cache on the PgStore instance (created lazily) to track
|
||||
# per-(path, metric, window) running counters + distinct learner sets. The
|
||||
# nightly job bypasses this cache and recomputes from mastery_gate_events.
|
||||
|
||||
|
||||
def _cache(pg_store: PgStore) -> dict:
|
||||
cache = getattr(pg_store, "_agg_cache", None)
|
||||
if not isinstance(cache, dict):
|
||||
cache = {}
|
||||
try:
|
||||
pg_store._agg_cache = cache # type: ignore[attr-defined]
|
||||
except Exception:
|
||||
pass
|
||||
return cache
|
||||
|
||||
|
||||
def _ck(path: str, metric: str, window_start: _dt.date) -> tuple:
|
||||
return (path, metric, window_start)
|
||||
|
||||
|
||||
async def _upsert_cell(pg_store: PgStore, path: str, metric: str,
|
||||
window_start: _dt.date, window_end: _dt.date,
|
||||
value: float | None, cell_count: int,
|
||||
suppressed: bool) -> None:
|
||||
await pg_store.upsert_cohort_aggregate(
|
||||
path, metric, window_start, window_end, value, cell_count, suppressed,
|
||||
)
|
||||
|
||||
|
||||
async def _bump_active_learners(pg_store: PgStore, path: str,
|
||||
window_start: _dt.date, learner_ref: str) -> int:
|
||||
"""Track distinct learner_refs per (path, window) in the in-memory cache.
|
||||
|
||||
Returns the current distinct count (after adding this learner). The
|
||||
nightly job reconciles the true count from mastery_gate_events.
|
||||
"""
|
||||
cache = _cache(pg_store)
|
||||
key = _ck(path, "__learners__", window_start)
|
||||
learners: set[str] = cache.get(key, set())
|
||||
learners.add(learner_ref)
|
||||
cache[key] = learners
|
||||
return len(learners)
|
||||
|
||||
|
||||
async def _bump_counter(pg_store: PgStore, path: str, metric: str,
|
||||
window_start: _dt.date, window_end: _dt.date) -> int:
|
||||
cache = _cache(pg_store)
|
||||
key = _ck(path, metric, window_start)
|
||||
cache[key] = cache.get(key, 0) + 1
|
||||
return cache[key]
|
||||
|
||||
|
||||
async def _bump_mode_counter(pg_store: PgStore, path: str, metric: str,
|
||||
window_start: _dt.date, window_end: _dt.date) -> int:
|
||||
return await _bump_counter(pg_store, path, metric, window_start, window_end)
|
||||
|
||||
|
||||
async def _running_mean(pg_store: PgStore, path: str, metric: str,
|
||||
window_start: _dt.date, window_end: _dt.date,
|
||||
value: float, _active_count: int) -> float:
|
||||
"""Incremental running mean per (path, metric, window)."""
|
||||
cache = _cache(pg_store)
|
||||
k = _ck(path, metric, window_start)
|
||||
n_key = _ck(path, metric + "__n__", window_start)
|
||||
n = cache.get(n_key, 0)
|
||||
prev = cache.get(k, 0.0)
|
||||
new_n = n + 1
|
||||
new_mean = prev + (value - prev) / new_n
|
||||
cache[k] = new_mean
|
||||
cache[n_key] = new_n
|
||||
return new_mean
|
||||
|
||||
|
||||
__all__ = ["aggregate_session", "K_ANON_THRESHOLD", "_rolling_window"]
|
||||
@@ -0,0 +1,44 @@
|
||||
"""On-session-end async aggregation hook (TASK-07-02, D-054).
|
||||
|
||||
Fire-and-forget: designed to be chained as an `asyncio.create_task` after
|
||||
the mastery flow. Failures log + the nightly job reconciles (no exception
|
||||
propagation to the caller — the session-end response returns immediately).
|
||||
|
||||
If `pg_store` is None (no Postgres), no-op + log WARNING.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from db.pg_store import PgStore
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def on_session_end(pg_store: PgStore | None, session_outcome: dict[str, Any]) -> None:
|
||||
"""Aggregate one session outcome. Non-blocking, fire-and-forget (D-054).
|
||||
|
||||
Failures are logged but never raised — the caller (session_recorder) has
|
||||
already returned its response; aggregation is off the voice path. The
|
||||
nightly job (nightly.py) reconciles any missed/hook-failed sessions.
|
||||
"""
|
||||
if pg_store is None:
|
||||
log.warning(
|
||||
"cohort aggregation skipped (no Postgres) for session %s",
|
||||
session_outcome.get("scenario_id"),
|
||||
)
|
||||
return
|
||||
try:
|
||||
from server.cohort.aggregator import aggregate_session
|
||||
|
||||
await aggregate_session(pg_store, session_outcome)
|
||||
except Exception:
|
||||
log.exception(
|
||||
"cohort aggregation hook failed for session %s — nightly job will reconcile",
|
||||
session_outcome.get("scenario_id"),
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["on_session_end"]
|
||||
@@ -0,0 +1,232 @@
|
||||
"""Nightly reconciliation scheduler (TASK-07-03, D-054, REQ-NFR-DASH-02).
|
||||
|
||||
In-process asyncio scheduler (no APScheduler — RESEARCH-v0.4 §3.4). Loops:
|
||||
compute seconds until next 03:00 CT (America/Winnipeg — Canada pilot) →
|
||||
asyncio.sleep → reconcile all 7-day windows → repeat. Resumes after restart.
|
||||
Failures log + retry next night (R-DASH-04).
|
||||
|
||||
Reconciliation recomputes all (path, metric, window_start) cells from the
|
||||
mastery_gate_events audit log + re-applies k-anonymity suppression. This
|
||||
guarantees REQ-NFR-DASH-02 (freshness ≤ 24h — the nightly job runs at least
|
||||
once/day) and reconciles any hook failures.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime as _dt
|
||||
import logging
|
||||
import statistics
|
||||
from collections import Counter, defaultdict
|
||||
from typing import Any
|
||||
|
||||
from db.pg_store import PgStore
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
CT = _dt.timezone(_dt.timedelta(hours=-5), "CT")
|
||||
NIGHTLY_HOUR = 3
|
||||
NIGHTLY_MINUTE = 0
|
||||
|
||||
|
||||
def seconds_until_next_03_ct(now: _dt.datetime | None = None) -> float:
|
||||
"""Seconds from `now` until the next 03:00 America/Winnipeg (CT).
|
||||
|
||||
America/Winnipeg observes CST (UTC-6) in winter + CDT (UTC-5) in summer.
|
||||
We approximate CT as a fixed UTC-5 offset (the pilot is in summer CDT
|
||||
and the scheduler drift of ≤1h over DST boundaries is acceptable for a
|
||||
nightly reconciliation job — the on-session-end hook keeps data fresh).
|
||||
A future hardening would use zoneinfo.ZoneInfo("America/Winnipeg") with
|
||||
proper DST handling.
|
||||
"""
|
||||
now = now or _dt.datetime.now(CT)
|
||||
if now.tzinfo is None:
|
||||
now = now.replace(tzinfo=CT)
|
||||
next_run = now.replace(hour=NIGHTLY_HOUR, minute=NIGHTLY_MINUTE,
|
||||
second=0, microsecond=0)
|
||||
if next_run <= now:
|
||||
next_run += _dt.timedelta(days=1)
|
||||
return (next_run - now).total_seconds()
|
||||
|
||||
|
||||
class NightlyScheduler:
|
||||
"""In-process asyncio scheduler for nightly cohort reconciliation.
|
||||
|
||||
Started as an asyncio task in the app lifespan (TASK-10-02). Cancel on
|
||||
shutdown. R-DASH-04: a reconciliation failure logs + retries the next
|
||||
night (the loop continues).
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._task: asyncio.Task | None = None
|
||||
self._stopped = False
|
||||
|
||||
async def start(self, pg_store: PgStore) -> asyncio.Task:
|
||||
"""Begin the nightly loop. Returns the running task."""
|
||||
self._stopped = False
|
||||
self._task = asyncio.create_task(self._run_loop(pg_store))
|
||||
return self._task
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Cancel the running loop (graceful shutdown)."""
|
||||
self._stopped = True
|
||||
if self._task is not None:
|
||||
self._task.cancel()
|
||||
try:
|
||||
await self._task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
self._task = None
|
||||
|
||||
async def _run_loop(self, pg_store: PgStore) -> None:
|
||||
while not self._stopped:
|
||||
try:
|
||||
secs = seconds_until_next_03_ct()
|
||||
log.info("nightly scheduler: next run in %.0fs (03:00 CT)", secs)
|
||||
await asyncio.sleep(secs)
|
||||
if self._stopped:
|
||||
return
|
||||
await self._reconcile(pg_store)
|
||||
except asyncio.CancelledError:
|
||||
return
|
||||
except Exception:
|
||||
log.exception("nightly reconciliation failed — retry next night (R-DASH-04)")
|
||||
# brief sleep to avoid a tight error loop if the clock is broken
|
||||
await asyncio.sleep(60)
|
||||
|
||||
async def _reconcile(self, pg_store: PgStore) -> None:
|
||||
"""Recompute all 7-day windows for all paths from mastery_gate_events.
|
||||
|
||||
Reads recent gate events (the audit log, REQ-NFR-MAST-02), groups by
|
||||
(path, window_start), recomputes each metric cell, applies k-anon
|
||||
suppression, and upserts. Idempotent — re-running produces the same
|
||||
aggregates (ON CONFLICT upsert).
|
||||
"""
|
||||
events = await _load_recent_events(pg_store)
|
||||
if not events:
|
||||
log.info("nightly reconcile: no recent gate events; nothing to recompute")
|
||||
return
|
||||
|
||||
# Group by path → window_start → list[events]
|
||||
by_path_window: dict[tuple[str, _dt.date], list[dict[str, Any]]] = defaultdict(list)
|
||||
today = _dt.datetime.now(_dt.timezone.utc).date()
|
||||
window_start = today - _dt.timedelta(days=6)
|
||||
for ev in events:
|
||||
ev_date = _coerce_date(ev.get("recorded_at"))
|
||||
if ev_date is None or ev_date < window_start:
|
||||
continue
|
||||
path = ev.get("path_id") or "unknown"
|
||||
by_path_window[(path, window_start)].append(ev)
|
||||
|
||||
from server.cohort.aggregator import K_ANON_THRESHOLD, _rolling_window
|
||||
|
||||
ws, we = _rolling_window()
|
||||
for (path, _), evs in by_path_window.items():
|
||||
learners = {e.get("learner_ref") for e in evs if e.get("learner_ref")}
|
||||
active_count = len(learners)
|
||||
suppressed = active_count < K_ANON_THRESHOLD
|
||||
|
||||
# sessions_count
|
||||
await pg_store.upsert_cohort_aggregate(
|
||||
path, "sessions_count", ws, we,
|
||||
None if suppressed else float(len(evs)),
|
||||
active_count, suppressed,
|
||||
)
|
||||
# active_learners_count
|
||||
await pg_store.upsert_cohort_aggregate(
|
||||
path, "active_learners_count", ws, we,
|
||||
None if suppressed else float(active_count),
|
||||
active_count, suppressed,
|
||||
)
|
||||
# gate_open_rate
|
||||
gate_opens = sum(1 for e in evs if (e.get("gate_outcome") or "") == "open")
|
||||
rate = gate_opens / len(evs) if evs else 0.0
|
||||
await pg_store.upsert_cohort_aggregate(
|
||||
path, "gate_open_rate", ws, we,
|
||||
None if suppressed else rate,
|
||||
active_count, suppressed,
|
||||
)
|
||||
# median_mastery_score + rubric_criterion_means from rubric_scores_jsonb
|
||||
score_rows: list[float] = []
|
||||
crit_scores: dict[str, list[float]] = defaultdict(list)
|
||||
for e in evs:
|
||||
scores = e.get("rubric_scores") or []
|
||||
if isinstance(scores, str):
|
||||
import json as _json
|
||||
try:
|
||||
scores = _json.loads(scores)
|
||||
except Exception:
|
||||
scores = []
|
||||
for r in scores:
|
||||
if isinstance(r, dict):
|
||||
cid = r.get("criterion_id") or r.get("id") or "unknown"
|
||||
s = r.get("score") or r.get("weighted_mean")
|
||||
if s is not None:
|
||||
crit_scores[cid].append(float(s))
|
||||
score_rows.append(float(s))
|
||||
if score_rows:
|
||||
med = statistics.median(score_rows)
|
||||
await pg_store.upsert_cohort_aggregate(
|
||||
path, "median_mastery_score", ws, we,
|
||||
None if suppressed else med,
|
||||
active_count, suppressed,
|
||||
)
|
||||
for cid, vals in crit_scores.items():
|
||||
mean_v = statistics.mean(vals) if vals else 0.0
|
||||
await pg_store.upsert_cohort_aggregate(
|
||||
path, f"rubric_criterion_mean:{cid}", ws, we,
|
||||
None if suppressed else mean_v,
|
||||
active_count, suppressed,
|
||||
)
|
||||
|
||||
log.info("nightly reconcile: recomputed %d (path, window) cells", len(by_path_window))
|
||||
|
||||
async def reconcile_now(self, pg_store: PgStore) -> None:
|
||||
"""Public hook for tests / ad-hoc reconciliation (no clock wait)."""
|
||||
await self._reconcile(pg_store)
|
||||
|
||||
|
||||
async def _load_recent_events(pg_store: PgStore) -> list[dict[str, Any]]:
|
||||
"""Load mastery_gate_events from the last 7 days.
|
||||
|
||||
Uses the PgStore pool directly (no extra method on PgStore to keep the
|
||||
surface minimal). Returns rows as dicts with decoded rubric_scores.
|
||||
"""
|
||||
async with pg_store.pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
"SELECT learner_ref, scenario_id, path_id, gate_outcome, "
|
||||
"rubric_scores_jsonb, recorded_at "
|
||||
"FROM mastery_gate_events "
|
||||
"WHERE recorded_at >= now() - interval '7 days' "
|
||||
"ORDER BY recorded_at"
|
||||
)
|
||||
out: list[dict[str, Any]] = []
|
||||
for r in rows:
|
||||
d = dict(r)
|
||||
scores = d.get("rubric_scores_jsonb")
|
||||
if hasattr(scores, "resolve"):
|
||||
try:
|
||||
import json as _json
|
||||
d["rubric_scores"] = _json.loads(scores.resolve()) if scores else []
|
||||
except Exception:
|
||||
d["rubric_scores"] = []
|
||||
else:
|
||||
d["rubric_scores"] = scores
|
||||
out.append(d)
|
||||
return out
|
||||
|
||||
|
||||
def _coerce_date(val: Any) -> _dt.date | None:
|
||||
if val is None:
|
||||
return None
|
||||
if isinstance(val, _dt.datetime):
|
||||
return val.date()
|
||||
if isinstance(val, _dt.date):
|
||||
return val
|
||||
try:
|
||||
return _dt.datetime.fromisoformat(str(val)).date()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
__all__ = ["NightlyScheduler", "seconds_until_next_03_ct", "CT"]
|
||||
@@ -0,0 +1,93 @@
|
||||
"""Shared helpers for operator API endpoints (SLICE-08).
|
||||
|
||||
Common response models + the recent-aggregates query used by all 3 cohort
|
||||
view endpoints (cohort, mastery, failure-patterns). Kept here to avoid
|
||||
duplicating the Pydantic models + pool query across 3 files.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
from typing import Any
|
||||
|
||||
from fastapi import HTTPException, Request, status
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class Cell(BaseModel):
|
||||
metric: str
|
||||
window_start: _dt.date | None = None
|
||||
window_end: _dt.date | None = None
|
||||
value: float | None = None
|
||||
cell_count: int = 0
|
||||
cell_suppressed: bool = False
|
||||
updated_at: _dt.datetime | None = None
|
||||
|
||||
|
||||
class PathView(BaseModel):
|
||||
path: str
|
||||
metrics: list[Cell]
|
||||
|
||||
|
||||
class ViewResponse(BaseModel):
|
||||
views: list[PathView]
|
||||
last_updated: _dt.datetime | None = None
|
||||
|
||||
|
||||
async def require_pg_store(request: Request):
|
||||
pg_store = getattr(request.app.state, "pg_store", None)
|
||||
if pg_store is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="operator tier unavailable (no Postgres)",
|
||||
)
|
||||
return pg_store
|
||||
|
||||
|
||||
async def all_recent_aggregates(pg_store, since: _dt.date) -> list[dict[str, Any]]:
|
||||
async with pg_store.pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
"SELECT path, metric, window_start, window_end, value, "
|
||||
"cell_count, cell_suppressed, updated_at "
|
||||
"FROM cohort_aggregates WHERE window_start >= $1 "
|
||||
"ORDER BY path, metric, window_start",
|
||||
since,
|
||||
)
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def cell_from_row(row: dict[str, Any]) -> Cell:
|
||||
return Cell(
|
||||
metric=row.get("metric", ""),
|
||||
window_start=row.get("window_start"),
|
||||
window_end=row.get("window_end"),
|
||||
value=float(row["value"]) if row.get("value") is not None else None,
|
||||
cell_count=int(row.get("cell_count") or 0),
|
||||
cell_suppressed=bool(row.get("cell_suppressed") or False),
|
||||
updated_at=row.get("updated_at"),
|
||||
)
|
||||
|
||||
|
||||
def group_by_path(
|
||||
rows: list[dict[str, Any]],
|
||||
metric_filter: set[str] | None = None,
|
||||
) -> tuple[list[PathView], _dt.datetime | None]:
|
||||
by_path: dict[str, list[dict[str, Any]]] = {}
|
||||
last_updated: _dt.datetime | None = None
|
||||
for r in rows:
|
||||
if metric_filter is not None and r.get("metric") not in metric_filter:
|
||||
continue
|
||||
by_path.setdefault(r["path"], []).append(r)
|
||||
ua = r.get("updated_at")
|
||||
if isinstance(ua, _dt.datetime) and (last_updated is None or ua > last_updated):
|
||||
last_updated = ua
|
||||
views = [PathView(path=p, metrics=[cell_from_row(c) for c in cells])
|
||||
for p, cells in by_path.items()]
|
||||
return views, last_updated
|
||||
|
||||
|
||||
__all__ = [
|
||||
"Cell", "PathView", "ViewResponse",
|
||||
"require_pg_store", "all_recent_aggregates",
|
||||
"cell_from_row", "group_by_path",
|
||||
]
|
||||
@@ -0,0 +1,42 @@
|
||||
"""GET /api/operator/cohort — practice volume view (TASK-08-01, D-053, D-057).
|
||||
|
||||
Auth-gated (Depends(current_operator)). Returns k-anonymized practice-volume
|
||||
aggregates from cohort_aggregates: sessions_count + active_learners_count per
|
||||
path. Suppressed cells have value=null + cell_suppressed=true; the frontend
|
||||
renders \"— (<10 learners)\". No per-learner drill-down (R-DASH-02).
|
||||
last_updated = max(updated_at) for freshness (REQ-NFR-DASH-02).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
|
||||
from fastapi import APIRouter, Depends, Request
|
||||
|
||||
from server.auth.dependencies import current_operator
|
||||
from server.auth.models import Operator
|
||||
from server.operator._common import (
|
||||
ViewResponse,
|
||||
all_recent_aggregates,
|
||||
group_by_path,
|
||||
require_pg_store,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api/operator", tags=["operator-cohort"])
|
||||
|
||||
PRACTICE_METRICS = {"sessions_count", "active_learners_count"}
|
||||
|
||||
|
||||
@router.get("/cohort", response_model=ViewResponse)
|
||||
async def cohort_view(
|
||||
request: Request,
|
||||
op: Operator = Depends(current_operator),
|
||||
) -> ViewResponse:
|
||||
pg_store = await require_pg_store(request)
|
||||
since = _dt.date.today() - _dt.timedelta(days=30)
|
||||
rows = await all_recent_aggregates(pg_store, since)
|
||||
views, last_updated = group_by_path(rows, PRACTICE_METRICS)
|
||||
return ViewResponse(views=views, last_updated=last_updated)
|
||||
|
||||
|
||||
__all__ = ["router"]
|
||||
@@ -0,0 +1,78 @@
|
||||
"""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
|
||||
|
||||
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"])
|
||||
|
||||
|
||||
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")
|
||||
return OkResponse(ok=True, id=cred_id, status="revoked")
|
||||
|
||||
|
||||
__all__ = ["router"]
|
||||
@@ -0,0 +1,44 @@
|
||||
"""GET /api/operator/failure-patterns — failure patterns view (TASK-08-03, D-053).
|
||||
|
||||
Auth-gated. Returns failure pattern metrics: failure_mode frequency (cells
|
||||
with metric prefix `failure_mode:`) + branch outcome distribution (cells
|
||||
with metric prefix `branch:`). Weak-spot rubric criteria (mean < 3.0) are
|
||||
highlighted by the frontend. All k-anonymized.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
|
||||
from fastapi import APIRouter, Depends, Request
|
||||
|
||||
from server.auth.dependencies import current_operator
|
||||
from server.auth.models import Operator
|
||||
from server.operator._common import (
|
||||
ViewResponse,
|
||||
all_recent_aggregates,
|
||||
group_by_path,
|
||||
require_pg_store,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api/operator", tags=["operator-failure-patterns"])
|
||||
|
||||
|
||||
def _is_failure_metric(metric: str) -> bool:
|
||||
return metric.startswith("failure_mode:") or metric.startswith("branch:")
|
||||
|
||||
|
||||
@router.get("/failure-patterns", response_model=ViewResponse)
|
||||
async def failure_patterns_view(
|
||||
request: Request,
|
||||
op: Operator = Depends(current_operator),
|
||||
) -> ViewResponse:
|
||||
pg_store = await require_pg_store(request)
|
||||
since = _dt.date.today() - _dt.timedelta(days=30)
|
||||
rows = await all_recent_aggregates(pg_store, since)
|
||||
failure_rows = [r for r in rows if _is_failure_metric(r.get("metric", ""))]
|
||||
views, last_updated = group_by_path(failure_rows)
|
||||
return ViewResponse(views=views, last_updated=last_updated)
|
||||
|
||||
|
||||
__all__ = ["router"]
|
||||
@@ -0,0 +1,45 @@
|
||||
"""GET /api/operator/mastery — mastery progression view (TASK-08-02, D-053).
|
||||
|
||||
Auth-gated. Returns mastery progression metrics: gate_open_rate,
|
||||
median_mastery_score, rubric_criterion_means (cells with metric prefix
|
||||
`rubric_criterion_mean:`). All k-anonymized (suppressed if < 10).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
|
||||
from fastapi import APIRouter, Depends, Request
|
||||
|
||||
from server.auth.dependencies import current_operator
|
||||
from server.auth.models import Operator
|
||||
from server.operator._common import (
|
||||
ViewResponse,
|
||||
all_recent_aggregates,
|
||||
group_by_path,
|
||||
require_pg_store,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api/operator", tags=["operator-mastery"])
|
||||
|
||||
MASTERY_METRICS = {"gate_open_rate", "median_mastery_score"}
|
||||
|
||||
|
||||
def _is_mastery_metric(metric: str) -> bool:
|
||||
return metric in MASTERY_METRICS or metric.startswith("rubric_criterion_mean:")
|
||||
|
||||
|
||||
@router.get("/mastery", response_model=ViewResponse)
|
||||
async def mastery_view(
|
||||
request: Request,
|
||||
op: Operator = Depends(current_operator),
|
||||
) -> ViewResponse:
|
||||
pg_store = await require_pg_store(request)
|
||||
since = _dt.date.today() - _dt.timedelta(days=30)
|
||||
rows = await all_recent_aggregates(pg_store, since)
|
||||
mastery_rows = [r for r in rows if _is_mastery_metric(r.get("metric", ""))]
|
||||
views, last_updated = group_by_path(mastery_rows)
|
||||
return ViewResponse(views=views, last_updated=last_updated)
|
||||
|
||||
|
||||
__all__ = ["router"]
|
||||
@@ -16,6 +16,7 @@ No auth — learner_id is the hardcoded 'learner-1' (D-007).
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import datetime as _dt
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
@@ -27,6 +28,10 @@ from server.cost import CostBreakdown, derive_cost
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
return _dt.datetime.now(_dt.timezone.utc).isoformat()
|
||||
|
||||
|
||||
class SessionRecorder:
|
||||
"""Records a voice session to SQLite (TASK-04-03)."""
|
||||
|
||||
@@ -35,10 +40,12 @@ class SessionRecorder:
|
||||
store: PraxisStore,
|
||||
learner_id: str = HARDCODED_LEARNER_ID,
|
||||
scenario_id: str = "cs_refund_ca_v01",
|
||||
pg_store: Any = None,
|
||||
) -> None:
|
||||
self.store = store
|
||||
self.learner_id = learner_id
|
||||
self.scenario_id = scenario_id
|
||||
self.pg_store = pg_store
|
||||
self.session_id: str | None = None
|
||||
self._turn_seq = 0
|
||||
# Cost inputs accumulated over the session.
|
||||
@@ -143,8 +150,53 @@ class SessionRecorder:
|
||||
asyncio.create_task(
|
||||
self._run_mastery_flow_guarded(mastery_deps)
|
||||
)
|
||||
|
||||
# v0.4 P2 (D-054): fire-and-forget cohort aggregation hook. Runs in
|
||||
# parallel with the mastery flow — aggregation only needs the session
|
||||
# outcome (available after session end), not the mastery scoring
|
||||
# result. Rubric-dependent metrics are reconciled by the nightly job.
|
||||
# Off the voice path (C-8, D-054). No-op if pg_store is None.
|
||||
if self.pg_store is not None:
|
||||
session_outcome = self._build_session_outcome(outcome)
|
||||
asyncio.create_task(self._run_cohort_aggregation(session_outcome))
|
||||
return breakdown
|
||||
|
||||
def _build_session_outcome(self, outcome: str) -> dict[str, Any]:
|
||||
"""Construct the session_outcome dict for the aggregation hook."""
|
||||
rubric_scores: list[dict[str, Any]] = []
|
||||
if self.mastery_result and isinstance(self.mastery_result, dict):
|
||||
rubric_scores = list(self.mastery_result.get("rubric_scores") or [])
|
||||
return {
|
||||
"learner_ref": self.learner_id,
|
||||
"path": self._path_slug(),
|
||||
"scenario_id": self.scenario_id,
|
||||
"outcome": outcome,
|
||||
"rubric_scores": rubric_scores,
|
||||
"failure_mode": self._failure_mode(),
|
||||
"branch_path": list(self._branch_path),
|
||||
"timestamp": _now_iso(),
|
||||
}
|
||||
|
||||
def _path_slug(self) -> str:
|
||||
# The scenario_id encodes the path loosely; default to customer_service.
|
||||
if self.scenario_id and self.scenario_id.startswith("cs_"):
|
||||
return "customer_service"
|
||||
return "default"
|
||||
|
||||
def _failure_mode(self) -> str | None:
|
||||
if self.mastery_result and isinstance(self.mastery_result, dict):
|
||||
return self.mastery_result.get("failure_mode")
|
||||
return None
|
||||
|
||||
async def _run_cohort_aggregation(self, session_outcome: dict[str, Any]) -> None:
|
||||
"""Fire-and-forget wrapper around the cohort aggregation hook (D-054)."""
|
||||
try:
|
||||
from server.cohort.hook import on_session_end
|
||||
|
||||
await on_session_end(self.pg_store, session_outcome)
|
||||
except Exception:
|
||||
log.exception("cohort aggregation dispatch failed for session %s", self.session_id)
|
||||
|
||||
async def _run_mastery_flow_guarded(self, deps: "MasteryFlowDeps") -> None:
|
||||
try:
|
||||
await self.run_mastery_flow(deps)
|
||||
|
||||
+35
-11
@@ -13,6 +13,7 @@ import base64
|
||||
import os
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Protocol, runtime_checkable
|
||||
|
||||
import nacl.secret
|
||||
import nacl.signing
|
||||
@@ -22,6 +23,27 @@ from db.store import PraxisStore
|
||||
_SECRETBOX_KEY_BYTES = nacl.secret.SecretBox.KEY_SIZE
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class IssuerKeyStore(Protocol):
|
||||
"""Issuer key store protocol (D-051, TASK-04-01).
|
||||
|
||||
Both PraxisStore (SQLite, v0.3) and PgStore (Postgres, v0.4) implement
|
||||
this protocol — R-VC-MIG-03 mitigation (both stores share the same
|
||||
interface so verification can use either). The structural check lets
|
||||
`isinstance(store, IssuerKeyStore)` succeed for duck-typed stores.
|
||||
"""
|
||||
|
||||
async def init_issuer_key(
|
||||
self, key_id: str, public_key: str, private_key_enc: bytes
|
||||
) -> None: ...
|
||||
|
||||
async def get_active_signing_key_row(self) -> dict | None: ...
|
||||
|
||||
async def get_public_key_row(self, key_id: str) -> dict | None: ...
|
||||
|
||||
async def set_issuer_key_superseded(self, key_id: str) -> None: ...
|
||||
|
||||
|
||||
def _load_root_key() -> bytes:
|
||||
raw = os.environ.get("PRAXIS_VC_ISSUER_KEY", "")
|
||||
if raw:
|
||||
@@ -63,7 +85,7 @@ def _decrypt_private_key(private_key_enc: bytes, root_key: bytes) -> nacl.signin
|
||||
return nacl.signing.SigningKey(seed)
|
||||
|
||||
|
||||
async def init_issuer_key(store: PraxisStore, root_key: bytes | None = None) -> KeyPair:
|
||||
async def init_issuer_key(store: IssuerKeyStore, root_key: bytes | None = None) -> KeyPair:
|
||||
rk = root_key if root_key is not None else _load_root_key()
|
||||
signing_key = nacl.signing.SigningKey.generate()
|
||||
verify_key = signing_key.verify_key
|
||||
@@ -75,7 +97,7 @@ async def init_issuer_key(store: PraxisStore, root_key: bytes | None = None) ->
|
||||
|
||||
|
||||
async def get_active_signing_key(
|
||||
store: PraxisStore, root_key: bytes | None = None
|
||||
store: IssuerKeyStore, root_key: bytes | None = None
|
||||
) -> tuple[KeyPair, bytes]:
|
||||
rk = root_key if root_key is not None else _load_root_key()
|
||||
row = await store.get_active_signing_key_row()
|
||||
@@ -89,18 +111,19 @@ async def get_active_signing_key(
|
||||
return kp, row["private_key_enc"]
|
||||
|
||||
|
||||
async def _fetch_private_key_enc(store: PraxisStore, key_id: str) -> bytes:
|
||||
async with store._connect() as db:
|
||||
db.row_factory = None
|
||||
cur = await db.execute(
|
||||
"SELECT private_key_enc FROM issuer_keys WHERE id = ?", (key_id,)
|
||||
)
|
||||
row = await cur.fetchone()
|
||||
return bytes(row[0]) if row else b""
|
||||
async def _fetch_private_key_enc(store: IssuerKeyStore, key_id: str) -> bytes:
|
||||
# PraxisStore exposes a _connect() context manager; PgStore does not
|
||||
# (it uses a pool). Use the protocol's get_public_key_row which both
|
||||
# stores implement, and read private_key_enc from the returned row.
|
||||
row = await store.get_public_key_row(key_id)
|
||||
if row is None:
|
||||
return b""
|
||||
enc = row.get("private_key_enc")
|
||||
return bytes(enc) if enc is not None else b""
|
||||
|
||||
|
||||
async def get_public_key_for_verification(
|
||||
store: PraxisStore, key_id: str
|
||||
store: IssuerKeyStore, key_id: str
|
||||
) -> nacl.signing.VerifyKey:
|
||||
row = await store.get_public_key_row(key_id)
|
||||
if row is None:
|
||||
@@ -119,6 +142,7 @@ async def rotate_key(store: PraxisStore, root_key: bytes | None = None) -> KeyPa
|
||||
|
||||
|
||||
__all__ = [
|
||||
"IssuerKeyStore",
|
||||
"KeyPair",
|
||||
"init_issuer_key",
|
||||
"get_active_signing_key",
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
"""VC issuer key migration SQLite → Postgres (TASK-04-03, D-051).
|
||||
|
||||
One-time migration procedure (R-VC-MIG-01 — highest-severity v0.4 risk):
|
||||
1. Read the v0.3 active public key from SQLite issuer_keys.
|
||||
2. Insert that public key into Postgres issuer_keys with status=
|
||||
'superseded' (private key NOT migrated — only the public key is
|
||||
archived for verification of already-issued v0.3 VCs).
|
||||
3. Generate a fresh Ed25519 keypair in Postgres issuer_keys with
|
||||
status='active' (encrypted at rest with the root key).
|
||||
4. Return {archived_key_id, new_key_id}.
|
||||
|
||||
R-VC-MIG-01 mitigation: the v0.3 public key is archived as superseded
|
||||
BEFORE the fresh key is activated (step 2 before step 3). This guarantees
|
||||
v0.3 VCs remain verifiable against the archived key.
|
||||
|
||||
G-027 (first-boot path): if SQLite has NO v0.3 active key (fresh deploy),
|
||||
skip the archive step and only generate the fresh v0.4 keypair.
|
||||
|
||||
Idempotent: if Postgres already has an active key, the whole procedure is
|
||||
a no-op. If Postgres already has a superseded key matching the v0.3 key_id,
|
||||
skip step 2 (already archived) but still generate the fresh key if no
|
||||
active key exists.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import nacl.signing
|
||||
|
||||
from db.pg_store import PgStore
|
||||
from db.store import PraxisStore
|
||||
from server.vc.issuer_keys import _encrypt_private_key
|
||||
|
||||
|
||||
async def _archive_v03_public_key(
|
||||
pg_store: PgStore, v03_key_id: str, v03_public_key: str
|
||||
) -> None:
|
||||
"""Insert the v0.3 public key into Postgres as superseded (idempotent)."""
|
||||
existing = await pg_store.get_public_key_row(v03_key_id)
|
||||
if existing is not None:
|
||||
return # already archived (or present as active — leave as-is)
|
||||
await pg_store.init_issuer_key(v03_key_id, v03_public_key, b"")
|
||||
await pg_store.set_issuer_key_superseded(v03_key_id)
|
||||
|
||||
|
||||
async def _generate_fresh_v04_key(
|
||||
pg_store: PgStore, root_key: bytes
|
||||
) -> str:
|
||||
"""Generate a fresh Ed25519 keypair in Postgres as active. Returns key_id."""
|
||||
signing_key = nacl.signing.SigningKey.generate()
|
||||
verify_key = signing_key.verify_key
|
||||
public_key_b64 = base64.b64encode(bytes(verify_key)).decode("ascii")
|
||||
private_key_enc = _encrypt_private_key(signing_key, root_key)
|
||||
key_id = f"key-{uuid.uuid4().hex[:12]}"
|
||||
await pg_store.init_issuer_key(key_id, public_key_b64, private_key_enc)
|
||||
return key_id
|
||||
|
||||
|
||||
async def migrate_issuer_keys(
|
||||
sqlite_store: PraxisStore,
|
||||
pg_store: PgStore,
|
||||
root_key: bytes,
|
||||
) -> dict[str, str | None]:
|
||||
"""Run the one-time VC key migration. Idempotent.
|
||||
|
||||
Returns {"archived_key_id": str | None, "new_key_id": str | None}.
|
||||
archived_key_id is None on the G-027 first-boot path (no v0.3 key).
|
||||
new_key_id is None if an active key already existed (no-op).
|
||||
"""
|
||||
# If Postgres already has an active key, the whole migration is done.
|
||||
active = await pg_store.get_active_signing_key_row()
|
||||
if active is not None:
|
||||
return {"archived_key_id": None, "new_key_id": None}
|
||||
|
||||
# Step 1 (G-027): read v0.3 active public key from SQLite. May be None
|
||||
# on a fresh deploy with no v0.3 history.
|
||||
v03_row = await sqlite_store.get_active_signing_key_row()
|
||||
archived_key_id: str | None = None
|
||||
if v03_row is not None:
|
||||
v03_key_id = v03_row["id"]
|
||||
v03_public_key = v03_row["public_key"]
|
||||
# Step 2 (R-VC-MIG-01): archive BEFORE activating the fresh key.
|
||||
await _archive_v03_public_key(pg_store, v03_key_id, v03_public_key)
|
||||
archived_key_id = v03_key_id
|
||||
|
||||
# Step 3: generate the fresh v0.4 keypair as active.
|
||||
new_key_id = await _generate_fresh_v04_key(pg_store, root_key)
|
||||
return {"archived_key_id": archived_key_id, "new_key_id": new_key_id}
|
||||
|
||||
|
||||
__all__ = ["migrate_issuer_keys"]
|
||||
+86
-16
@@ -1,11 +1,24 @@
|
||||
"""Public VC verification (SLICE-09 TASK-09-04, D-043, REQ-NFR-VC-02).
|
||||
"""Public VC verification (SLICE-09 TASK-09-04, D-043, REQ-NFR-VC-02;
|
||||
v0.4 TASK-04-04 two-store fallback per G-011).
|
||||
|
||||
`GET /vc/verify/<credential_id>` — public, unauthenticated. Fetches the
|
||||
credential from SQLite, fetches the issuer public key, validates the Ed25519
|
||||
signature against the JCS-canonicalized payload, checks the Bitstring Status
|
||||
List (no cache — fetched on every verify call, REQ-NFR-VC-02). Returns JSON
|
||||
credential + issuer public key, validates the Ed25519 signature against
|
||||
the JCS-canonicalized payload, checks the Bitstring Status List (no cache
|
||||
— fetched on every verify call, REQ-NFR-VC-02). Returns JSON
|
||||
{valid, status, issuer, credential, mastery, credentialTier, verifiedAt}.
|
||||
No PII beyond what the credential asserts.
|
||||
|
||||
G-011 two-store fallback semantics (binding contract):
|
||||
(a) If Postgres is available (pg_store is not None), use it for issuer
|
||||
key lookup (both active AND superseded keys — get_public_key_row
|
||||
queries by id, not status).
|
||||
(b) If Postgres is available but the credential is not found in its
|
||||
issued_credentials table, fall back to SQLite issued_credentials
|
||||
(v0.3 credentials remain in SQLite — D-051 "no re-issuance").
|
||||
(c) If Postgres is NOT available (pg_store is None), use the existing
|
||||
v0.3 SQLite path for BOTH keys and credentials (full v0.3 compat).
|
||||
The key store used for verification is always the one that holds the key
|
||||
row found by key_id; the credential store is whichever store had the row.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -17,7 +30,7 @@ from typing import Any
|
||||
from db.store import PraxisStore
|
||||
|
||||
from server.vc.issuer import verify_proof, extract_key_id, CREDENTIAL_TIER
|
||||
from server.vc.issuer_keys import get_public_key_for_verification
|
||||
from server.vc.issuer_keys import IssuerKeyStore, get_public_key_for_verification
|
||||
from server.vc.status_list import BitstringStatusList
|
||||
|
||||
|
||||
@@ -26,26 +39,35 @@ def _now_iso() -> str:
|
||||
|
||||
|
||||
async def verify_credential(
|
||||
store: PraxisStore, credential_id: str
|
||||
store: IssuerKeyStore,
|
||||
credential_id: str,
|
||||
*,
|
||||
pg_store: IssuerKeyStore | None = None,
|
||||
sqlite_store: PraxisStore | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
row = await store.get_credential(credential_id)
|
||||
"""Verify a VC. Returns the verification result dict, or None if the
|
||||
credential id is not found in any store.
|
||||
|
||||
Per G-011:
|
||||
- If pg_store is provided, try it first for BOTH credential + key
|
||||
lookup; fall back to sqlite_store for the credential if Postgres
|
||||
doesn't have it (v0.3 credentials stay in SQLite).
|
||||
- If pg_store is None, use `store` (the v0.3 SQLite path) for both.
|
||||
"""
|
||||
row = await _lookup_credential(credential_id, store, pg_store, sqlite_store)
|
||||
if row is None:
|
||||
return None
|
||||
secured_doc = json.loads(row["vc_payload_json"])
|
||||
key_id = extract_key_id(secured_doc)
|
||||
if key_id is None:
|
||||
return _invalid(row, secured_doc)
|
||||
try:
|
||||
verify_key = await get_public_key_for_verification(store, key_id)
|
||||
except KeyError:
|
||||
# Key lookup: prefer pg_store (G-011a) for v0.4 keys + archived v0.3
|
||||
# keys; fall back to `store` (SQLite) if pg_store doesn't have the key.
|
||||
verify_key = await _lookup_public_key(key_id, store, pg_store)
|
||||
if verify_key is None:
|
||||
return _invalid(row, secured_doc)
|
||||
sig_valid = verify_proof(secured_doc, verify_key)
|
||||
revoked = False
|
||||
cs = secured_doc.get("credentialStatus") or {}
|
||||
idx_str = cs.get("statusListIndex")
|
||||
if idx_str is not None:
|
||||
sl = BitstringStatusList(store, "default")
|
||||
revoked = await sl.get_status(int(idx_str))
|
||||
revoked = await _check_revocation(secured_doc, store, sqlite_store or store)
|
||||
status = "revoked" if revoked else "active"
|
||||
valid = bool(sig_valid and not revoked)
|
||||
subject = secured_doc.get("credentialSubject") or {}
|
||||
@@ -73,6 +95,54 @@ async def verify_credential(
|
||||
}
|
||||
|
||||
|
||||
async def _lookup_credential(
|
||||
credential_id: str,
|
||||
store: IssuerKeyStore,
|
||||
pg_store: IssuerKeyStore | None,
|
||||
sqlite_store: PraxisStore | None,
|
||||
) -> dict | None:
|
||||
"""G-011(b): try Postgres first, fall back to SQLite for v0.3 creds."""
|
||||
if pg_store is not None:
|
||||
row = await pg_store.get_credential(credential_id)
|
||||
if row is not None:
|
||||
return row
|
||||
if sqlite_store is not None:
|
||||
return await sqlite_store.get_credential(credential_id)
|
||||
return None
|
||||
# G-011(c): no Postgres — v0.3 SQLite path.
|
||||
return await store.get_credential(credential_id)
|
||||
|
||||
|
||||
async def _lookup_public_key(
|
||||
key_id: str,
|
||||
store: IssuerKeyStore,
|
||||
pg_store: IssuerKeyStore | None,
|
||||
):
|
||||
"""G-011(a): prefer Postgres for key lookup (finds active + superseded);
|
||||
fall back to `store` (SQLite) if Postgres doesn't have the key."""
|
||||
if pg_store is not None:
|
||||
try:
|
||||
vk = await get_public_key_for_verification(pg_store, key_id)
|
||||
return vk
|
||||
except KeyError:
|
||||
pass
|
||||
try:
|
||||
return await get_public_key_for_verification(store, key_id)
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
|
||||
async def _check_revocation(
|
||||
secured_doc: dict, store: IssuerKeyStore, status_store: PraxisStore
|
||||
) -> bool:
|
||||
cs = secured_doc.get("credentialStatus") or {}
|
||||
idx_str = cs.get("statusListIndex")
|
||||
if idx_str is None:
|
||||
return False
|
||||
sl = BitstringStatusList(status_store, "default")
|
||||
return await sl.get_status(int(idx_str))
|
||||
|
||||
|
||||
def _invalid(row: dict, secured_doc: dict) -> dict[str, Any]:
|
||||
subject = secured_doc.get("credentialSubject") or {}
|
||||
return {
|
||||
|
||||
Reference in New Issue
Block a user