"""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/` — public, unauthenticated. Fetches the 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 import datetime as _dt import json 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 IssuerKeyStore, get_public_key_for_verification from server.vc.status_list import BitstringStatusList def _now_iso() -> str: return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") async def verify_credential( store: IssuerKeyStore, credential_id: str, *, pg_store: IssuerKeyStore | None = None, sqlite_store: PraxisStore | None = None, ) -> dict[str, Any] | None: """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) # 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 = 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 {} issuer = secured_doc.get("issuer") return { "valid": valid, "status": status, "issuer": issuer, "credential": { "id": secured_doc.get("id"), "type": secured_doc.get("type"), "validFrom": secured_doc.get("validFrom"), "validUntil": secured_doc.get("validUntil"), }, "mastery": { "skill": subject.get("skill"), "level": subject.get("level"), "path": subject.get("path"), "rubricScore": subject.get("rubricScore"), "scenariosPassed": subject.get("scenariosPassed", []), "completedWeeks": subject.get("completedWeeks"), }, "credentialTier": subject.get("credentialTier", CREDENTIAL_TIER), "verifiedAt": _now_iso(), } 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 { "valid": False, "status": row.get("status", "active"), "issuer": secured_doc.get("issuer"), "credential": { "id": secured_doc.get("id"), "type": secured_doc.get("type"), "validFrom": secured_doc.get("validFrom"), "validUntil": secured_doc.get("validUntil"), }, "mastery": { "skill": subject.get("skill"), "level": subject.get("level"), "path": subject.get("path"), "rubricScore": subject.get("rubricScore"), "scenariosPassed": subject.get("scenariosPassed", []), "completedWeeks": subject.get("completedWeeks"), }, "credentialTier": subject.get("credentialTier", CREDENTIAL_TIER), "verifiedAt": _now_iso(), } async def revoke_credential(store: PraxisStore, credential_id: str) -> bool: row = await store.get_credential(credential_id) if row is None: return False secured_doc = json.loads(row["vc_payload_json"]) cs = secured_doc.get("credentialStatus") or {} idx_str = cs.get("statusListIndex") if idx_str is None: await store.set_credential_status(credential_id, "revoked") return True sl = BitstringStatusList(store, "default") await sl.set_status(int(idx_str), True) await store.set_credential_status(credential_id, "revoked") return True __all__ = ["verify_credential", "revoke_credential"]