"""W3C VC 2.0 issuance — Ed25519 + JCS + eddsa-jcs-2022 proof (SLICE-09 TASK-09-02). Builds a Verifiable Credential per VC-DM 2.0, secures it with a Data Integrity `eddsa-jcs-2022` proof (JCS canonicalization, Ed25519 signature), and persists it to SQLite. The `issue_credential` coroutine is the entry point wired into SessionRecorder.run_mastery_flow (grill Axis 8 MUST). Credential tier is `formative` (grill Axis 4 MUST #1) — the v0.3 credential is a formative mastery signal, not a high-stakes summative credential. """ from __future__ import annotations import base64 import datetime as _dt import hashlib import json import os import uuid from typing import Any import canonicaljson import nacl.signing from db.store import PraxisStore from server.vc.issuer_keys import KeyPair, get_active_signing_key from server.vc.status_list import BitstringStatusList ISSUER_URL_DEFAULT = "https://praxis.example/issuers/v0.3" CONTEXTS = [ "https://www.w3.org/ns/credentials/v2", "https://praxis.example/contexts/mastery/v1", ] CREDENTIAL_TIER = "formative" def _issuer_url() -> str: return os.environ.get("PRAXIS_ISSUER_URL", ISSUER_URL_DEFAULT).rstrip("/") def _now_iso() -> str: return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _valid_until(issuance_iso: str, years: int = 3) -> str: dt = _dt.datetime.strptime(issuance_iso, "%Y-%m-%dT%H:%M:%SZ").replace( tzinfo=_dt.timezone.utc ) return (dt + _dt.timedelta(days=365 * years)).strftime("%Y-%m-%dT%H:%M:%SZ") def build_vc_payload( learner_ref: str, path: str, scenarios_passed: list[str], rubric_score: float, completed_weeks: int, evidence: list[dict[str, Any]] | None, credential_id: str | None = None, status_list_index: int | None = None, ) -> dict[str, Any]: issuance = _now_iso() issuer = _issuer_url() cid = credential_id or f"vc-{uuid.uuid4().hex[:16]}" payload: dict[str, Any] = { "@context": list(CONTEXTS), "id": f"{issuer}/vc/{cid}", "type": ["VerifiableCredential", "MasteryCredential"], "issuer": issuer, "validFrom": issuance, "validUntil": _valid_until(issuance, 3), "name": f"Mastery of {path.replace('-', ' ').title()}", "description": ( "Praxis v0.3 formative mastery credential — the holder demonstrated " "competency across varied scenarios, scored against a 5-level rubric." ), "credentialTier": CREDENTIAL_TIER, "credentialSubject": { "id": f"urn:uuid:{learner_ref}", "type": "Person", "skill": path, "level": "mastery", "path": path, "completedWeeks": completed_weeks, "rubricScore": round(float(rubric_score), 3), "rubricMax": 5.0, "rubricThreshold": 3.5, "scenariosPassed": list(scenarios_passed), "credentialTier": CREDENTIAL_TIER, "evidence": evidence or [], }, } if status_list_index is not None: payload["credentialStatus"] = { "type": "BitstringStatusListEntry", "statusPurpose": "revocation", "statusListIndex": str(status_list_index), "statusListCredential": f"{issuer}/status/default", } return payload def canonicalize(payload: dict[str, Any]) -> bytes: return canonicaljson.encode_canonical_json(payload) def _build_proof_config(key_id: str) -> dict[str, Any]: issuer = _issuer_url() return { "type": "DataIntegrityProof", "cryptosuite": "eddsa-jcs-2022", "created": _now_iso(), "verificationMethod": f"{issuer}/keys/{key_id}", "proofPurpose": "assertionMethod", } def _compute_hash_data( unsecured_doc: dict[str, Any], proof_options: dict[str, Any] ) -> bytes: canonical_doc = canonicalize(unsecured_doc) canonical_proof = canonicalize(proof_options) return hashlib.sha256(canonical_proof).digest() + hashlib.sha256( canonical_doc ).digest() def sign(payload: dict[str, Any], signing_key: nacl.signing.SigningKey, key_id: str) -> tuple[dict[str, Any], str]: proof_options = _build_proof_config(key_id) hash_data = _compute_hash_data(payload, proof_options) signed = signing_key.sign(hash_data) signature_bytes = signed.signature signature_b64 = base64.b64encode(signature_bytes).decode("ascii") proof = dict(proof_options) proof["proofValue"] = signature_b64 secured = dict(payload) secured["proof"] = proof return secured, signature_b64 def verify_proof( secured_doc: dict[str, Any], verify_key: nacl.signing.VerifyKey, ) -> bool: if "proof" not in secured_doc: return False proof = secured_doc["proof"] proof_value_b64 = proof.get("proofValue") if not proof_value_b64: return False proof_options = {k: v for k, v in proof.items() if k != "proofValue"} unsecured = {k: v for k, v in secured_doc.items() if k != "proof"} hash_data = _compute_hash_data(unsecured, proof_options) try: sig = base64.b64decode(proof_value_b64) verify_key.verify(hash_data, sig) return True except Exception: return False def extract_key_id(secured_doc: dict[str, Any]) -> str | None: proof = secured_doc.get("proof") or {} vm = proof.get("verificationMethod") or "" if "/" in vm: return vm.rsplit("/", 1)[-1] return None async def issue_credential( store: PraxisStore, signing_key: nacl.signing.SigningKey | None = None, learner_id: str = "", path: str = "", scenarios_passed: list[str] | None = None, rubric_score: float = 0.0, completed_weeks: int = 6, evidence: list[dict[str, Any]] | None = None, key_id: str | None = None, ) -> str: if signing_key is None or key_id is None: kp, _enc = await get_active_signing_key(store) signing_key = kp.signing_key key_id = kp.key_id scenarios = list(scenarios_passed or []) ev = list(evidence or []) status_list = BitstringStatusList(store, "default") slot = await status_list.allocate_slot() cred_id = f"vc-{uuid.uuid4().hex[:16]}" payload = build_vc_payload( learner_ref=learner_id, path=path, scenarios_passed=scenarios, rubric_score=rubric_score, completed_weeks=completed_weeks, evidence=ev, credential_id=cred_id, status_list_index=slot, ) secured, signature_b64 = sign(payload, signing_key, key_id) payload_json = json.dumps(secured, sort_keys=True, separators=(",", ":")) await store.insert_credential(cred_id, learner_id, payload_json, signature_b64) return cred_id __all__ = [ "build_vc_payload", "canonicalize", "sign", "verify_proof", "extract_key_id", "issue_credential", "CREDENTIAL_TIER", ]