feat(vc): upgrade to W3C VC Data Model v2.0 (Phase 1, dual-accept)

agent/ambassador.py

@@ -8,7 +8,7 @@
 from fastapi import FastAPI
 from contextlib import asynccontextmanager
 
-from app.credentials import issue_credential
+from app.credentials import issue_credential, vc_valid_from, vc_valid_until
 
 # ---------------------------------------------------------------------------
 # Config
@@ -70,8 +70,8 @@ async def ensure_self_registered():
             """INSERT INTO credentials (subject_did, credential_type, issuer, issued_at, expires_at, proof_value, raw_vc)
             VALUES ($1, $2, $3, $4, $5, $6, $7)""",
             AMBASSADOR_DID, "AgentTrustCredential", vc["issuer"],
-            datetime.datetime.fromisoformat(vc["issuanceDate"].replace("Z", "")),
-            datetime.datetime.fromisoformat(vc["expirationDate"].replace("Z", "")),
+            datetime.datetime.fromisoformat(vc_valid_from(vc).replace("Z", "")),
+            datetime.datetime.fromisoformat(vc_valid_until(vc).replace("Z", "")),
             vc["proof"]["proofValue"],
             json.dumps(vc),
         )

app/credentials.py

@@ -1,29 +1,51 @@
-"""MolTrust Verifiable Credentials - W3C VC Data Model"""
+"""MolTrust Verifiable Credentials - W3C VC Data Model v2.0
+
+Issuance: emits W3C VC Data Model v2 only (validFrom/validUntil, v2 @context).
+Verification: dual-accept — recognises v2 (validFrom/validUntil) AND legacy
+v1 (issuanceDate/expirationDate) so previously-issued credentials still
+verify until the dataset is fully rotated.
+"""
 import os, json, datetime, hashlib
 from nacl.signing import SigningKey
 from app.crypto.kms_signer import get_decrypted_signing_key_hex
 
 ISSUER_DID = "did:web:api.moltrust.ch"
 
+VC_V2_CONTEXT = "https://www.w3.org/ns/credentials/v2"
+VC_V1_CONTEXT = "https://www.w3.org/2018/credentials/v1"
+MOLTRUST_CONTEXT = "https://api.moltrust.ch/contexts/trust/v1"
+
+
 def get_signing_key():
     hex_key = get_decrypted_signing_key_hex()
     return SigningKey(bytes.fromhex(hex_key))
 
+
+def vc_valid_from(vc: dict) -> str:
+    """Return the issuance instant — `validFrom` (v2), falling back to `issuanceDate` (v1)."""
+    return vc.get("validFrom") or vc.get("issuanceDate", "") or ""
+
+
+def vc_valid_until(vc: dict) -> str:
+    """Return the expiry instant — `validUntil` (v2), falling back to `expirationDate` (v1)."""
+    return vc.get("validUntil") or vc.get("expirationDate", "") or ""
+
+
 def issue_credential(subject_did: str, credential_type: str, claims: dict) -> dict:
     now = datetime.datetime.utcnow()
     credential = {
         "@context": [
-            "https://www.w3.org/2018/credentials/v1",
-            "https://api.moltrust.ch/contexts/trust/v1"
+            VC_V2_CONTEXT,
+            MOLTRUST_CONTEXT,
         ],
         "type": ["VerifiableCredential", credential_type],
         "issuer": ISSUER_DID,
-        "issuanceDate": now.isoformat() + "Z",
-        "expirationDate": (now + datetime.timedelta(days=365)).isoformat() + "Z",
+        "validFrom": now.isoformat() + "Z",
+        "validUntil": (now + datetime.timedelta(days=365)).isoformat() + "Z",
         "credentialSubject": {
             "id": subject_did,
-            **claims
-        }
+            **claims,
+        },
     }
 
     signing_key = get_signing_key()
@@ -35,10 +57,11 @@ def issue_credential(subject_did: str, credential_type: str, claims: dict) -> di
         "created": now.isoformat() + "Z",
         "verificationMethod": f"{ISSUER_DID}#key-1",
         "proofPurpose": "assertionMethod",
-        "proofValue": signed.signature.hex()
+        "proofValue": signed.signature.hex(),
     }
     return credential
 
+
 def verify_credential(credential: dict) -> dict:
     proof = credential.get("proof")
     if not proof:
@@ -55,7 +78,7 @@ def verify_credential(credential: dict) -> dict:
         verify_key = signing_key.verify_key
         verify_key.verify(payload, signature)
 
-        exp = credential.get("expirationDate", "")
+        exp = vc_valid_until(credential)
         if exp:
             exp_dt = datetime.datetime.fromisoformat(exp.replace("Z", ""))
             if datetime.datetime.utcnow() > exp_dt:

app/main.py

@@ -1020,8 +1020,8 @@ async def register_agent(request: Request, body: RegisterRequest, api_key: str =
                 """INSERT INTO credentials (subject_did, credential_type, issuer, issued_at, expires_at, proof_value, raw_vc)
                 VALUES ($1, $2, $3, $4, $5, $6, $7)""",
                 agent_did, "AgentTrustCredential", auto_vc["issuer"],
-                datetime.datetime.fromisoformat(auto_vc["issuanceDate"].replace("Z","")),
-                datetime.datetime.fromisoformat(auto_vc["expirationDate"].replace("Z","")),
+                datetime.datetime.fromisoformat(vc_valid_from(auto_vc).replace("Z","")),
+                datetime.datetime.fromisoformat(vc_valid_until(auto_vc).replace("Z","")),
                 auto_vc["proof"]["proofValue"],
                 json.dumps(auto_vc)
             )
@@ -2886,7 +2886,7 @@ async def register_batch(request: Request):
 
 
 # --- Verifiable Credentials ---
-from app.credentials import issue_credential, verify_credential
+from app.credentials import issue_credential, verify_credential, vc_valid_from, vc_valid_until
 from app.ipfs_publisher import publish_to_ipfs, get_ipfs_url
 
 class IssueVCRequest(BaseModel):
@@ -2953,8 +2953,8 @@ async def issue_vc(request: Request, body: IssueVCRequest, api_key: str = Depend
                 """INSERT INTO credentials (subject_did, credential_type, issuer, issued_at, expires_at, proof_value, raw_vc)
                 VALUES ($1, $2, $3, $4, $5, $6, $7)""",
                 body.subject_did, body.credential_type, vc["issuer"],
-                datetime.datetime.fromisoformat(vc["issuanceDate"].replace("Z","")),
-                datetime.datetime.fromisoformat(vc["expirationDate"].replace("Z","")),
+                datetime.datetime.fromisoformat(vc_valid_from(vc).replace("Z","")),
+                datetime.datetime.fromisoformat(vc_valid_until(vc).replace("Z","")),
                 vc["proof"]["proofValue"],
                 json.dumps(vc)
             )
@@ -5353,15 +5353,16 @@ class MusicRevokeRequest(BaseModel):
 
 def _build_music_vc(row) -> dict:
     """Build VerifiedMusicCredential from DB row."""
+    issued = row["issued_at"].isoformat() if hasattr(row["issued_at"], "isoformat") else str(row["issued_at"])
     return {
         "@context": [
-            "https://www.w3.org/2018/credentials/v1",
+            "https://www.w3.org/ns/credentials/v2",
             "https://moltrust.ch/ns/music/v1",
         ],
         "type": ["VerifiableCredential", "VerifiedMusicCredential"],
         "id": row["id"],
         "issuer": "did:moltrust:registry",
-        "issuanceDate": row["issued_at"].isoformat() if hasattr(row["issued_at"], "isoformat") else str(row["issued_at"]),
+        "validFrom": issued,
         "credentialSubject": {
             "agentDid": row["agent_did"],
             "humanName": row["human_name"],
@@ -5377,7 +5378,7 @@ def _build_music_vc(row) -> dict:
             },
             "provenance": {
                 "trackHash": row["track_hash"],
-                "issuanceDate": row["issued_at"].isoformat() if hasattr(row["issued_at"], "isoformat") else str(row["issued_at"]),
+                "issuanceDate": issued,
                 "euAiActCompliance": "Article 50(2)",
             },
         },
@@ -5445,15 +5446,16 @@ async def issue_music_credential(request: Request, body: MusicCredentialRequest,
     now = datetime.datetime.utcnow()
 
     # Build VC
+    issued_ts = now.isoformat() + "Z"
     vc = {
         "@context": [
-            "https://www.w3.org/2018/credentials/v1",
+            "https://www.w3.org/ns/credentials/v2",
             "https://moltrust.ch/ns/music/v1",
         ],
         "type": ["VerifiableCredential", "VerifiedMusicCredential"],
         "id": credential_id,
         "issuer": "did:moltrust:registry",
-        "issuanceDate": now.isoformat() + "Z",
+        "validFrom": issued_ts,
         "credentialSubject": {
             "agentDid": body.agent_did,
             "humanName": body.human_name,
@@ -5469,7 +5471,7 @@ async def issue_music_credential(request: Request, body: MusicCredentialRequest,
             },
             "provenance": {
                 "trackHash": track_hash,
-                "issuanceDate": now.isoformat() + "Z",
+                "issuanceDate": issued_ts,
                 "euAiActCompliance": "Article 50(2)",
             },
         },

app/swarm/endorsement.py

@@ -133,14 +133,14 @@ async def issue_endorsement(
     vc_id = f"urn:uuid:{uuid.uuid4()}"
     vc = {
         "@context": [
-            "https://www.w3.org/2018/credentials/v1",
-            "https://moltrust.ch/credentials/v1"
+            "https://www.w3.org/ns/credentials/v2",
+            "https://moltrust.ch/credentials/v1",
         ],
         "id": vc_id,
         "type": ["VerifiableCredential", "SkillEndorsementCredential"],
         "issuer": endorser_did,
-        "issuanceDate": now.isoformat(),
-        "expirationDate": expires_at.isoformat(),
+        "validFrom": now.isoformat(),
+        "validUntil": expires_at.isoformat(),
         "credentialSubject": {
             "id": endorsed_did,
             "skill": skill,

backfill_credentials.py

@@ -3,7 +3,7 @@
 sys.path.insert(0, os.path.expanduser('~/moltstack'))
 
 import asyncpg
-from app.credentials import issue_credential
+from app.credentials import issue_credential, vc_valid_from, vc_valid_until
 
 async def main():
     conn = await asyncpg.connect(host='localhost', database='moltstack',
@@ -39,8 +39,8 @@ async def main():
                 """INSERT INTO credentials (subject_did, credential_type, issuer, issued_at, expires_at, proof_value, raw_vc)
                 VALUES ($1, $2, $3, $4, $5, $6, $7)""",
                 did, 'AgentTrustCredential', vc['issuer'],
-                __import__('datetime').datetime.fromisoformat(vc['issuanceDate'].replace('Z','')),
-                __import__('datetime').datetime.fromisoformat(vc['expirationDate'].replace('Z','')),
+                __import__('datetime').datetime.fromisoformat(vc_valid_from(vc).replace('Z','')),
+                __import__('datetime').datetime.fromisoformat(vc_valid_until(vc).replace('Z','')),
                 vc['proof']['proofValue'],
                 json.dumps(vc)
             )

tests/test_credentials_vc_v2.py

@@ -0,0 +1,168 @@
+"""VC Data Model v2 issuance + dual-accept verification.
+
+Phase-1 contract:
+  - Newly issued credentials use v2 (`@context` v2 URL, `validFrom` / `validUntil`).
+  - `verify_credential` still accepts legacy v1 credentials (`issuanceDate` /
+    `expirationDate`) so credentials minted before this migration keep verifying.
+"""
+import datetime
+import json
+import os
+import sys
+import pytest
+
+# Resolve `from app...` from the test file's own repo root (works on Hetzner
+# server layout AND on a local checkout — conftest pins the server path which
+# is irrelevant for these pure-unit tests).
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from app.credentials import (  # noqa: E402
+    VC_V2_CONTEXT,
+    VC_V1_CONTEXT,
+    MOLTRUST_CONTEXT,
+    vc_valid_from,
+    vc_valid_until,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helper functions — dual-format read
+# ---------------------------------------------------------------------------
+
+def test_valid_from_prefers_v2():
+    vc = {"validFrom": "2026-05-25T10:00:00Z", "issuanceDate": "2020-01-01T00:00:00Z"}
+    assert vc_valid_from(vc) == "2026-05-25T10:00:00Z"
+
+
+def test_valid_from_falls_back_to_v1():
+    vc = {"issuanceDate": "2026-05-25T10:00:00Z"}
+    assert vc_valid_from(vc) == "2026-05-25T10:00:00Z"
+
+
+def test_valid_from_missing_returns_empty():
+    assert vc_valid_from({}) == ""
+
+
+def test_valid_until_prefers_v2():
+    vc = {"validUntil": "2027-05-25T10:00:00Z", "expirationDate": "2020-01-01T00:00:00Z"}
+    assert vc_valid_until(vc) == "2027-05-25T10:00:00Z"
+
+
+def test_valid_until_falls_back_to_v1():
+    vc = {"expirationDate": "2027-05-25T10:00:00Z"}
+    assert vc_valid_until(vc) == "2027-05-25T10:00:00Z"
+
+
+def test_valid_until_missing_returns_empty():
+    assert vc_valid_until({}) == ""
+
+
+# ---------------------------------------------------------------------------
+# Issue + verify roundtrip — needs a signing key in env
+# ---------------------------------------------------------------------------
+
+_TEST_SEED_HEX = "11" * 32
+
+
+def _ensure_test_signing_key():
+    """Install a deterministic test signing key.
+
+    `app.credentials.get_signing_key` reads from KMS in production; tests run
+    without that infra. We swap the function in the `app.credentials` module
+    so both `issue_credential` and `verify_credential` see the same dev key.
+    Patches the module-level binding rather than the underlying KMS module
+    because `credentials.py` imports the KMS function by name at load time.
+    """
+    from nacl.signing import SigningKey
+    from app import credentials as _cred
+    if getattr(_cred, "_TEST_KEY_PATCHED", False):
+        return
+    test_sk = SigningKey(bytes.fromhex(_TEST_SEED_HEX))
+    _cred.get_signing_key = lambda: test_sk
+    _cred._TEST_KEY_PATCHED = True
+
+
+def _test_signing_key():
+    from nacl.signing import SigningKey
+    return SigningKey(bytes.fromhex(_TEST_SEED_HEX))
+
+
+def test_issue_credential_emits_v2_shape():
+    _ensure_test_signing_key()
+    from app.credentials import issue_credential
+    vc = issue_credential("did:moltrust:test_v2_issue", "TestCredential", {"k": "v"})
+
+    assert VC_V2_CONTEXT in vc["@context"], f"v2 context missing: {vc['@context']}"
+    assert MOLTRUST_CONTEXT in vc["@context"]
+    assert "validFrom" in vc and vc["validFrom"]
+    assert "validUntil" in vc and vc["validUntil"]
+    # Legacy v1 fields MUST NOT be present on freshly-issued v2 credentials
+    assert "issuanceDate" not in vc
+    assert "expirationDate" not in vc
+
+
+def test_verify_accepts_legacy_v1_credential():
+    _ensure_test_signing_key()
+    from app.credentials import issue_credential, verify_credential, ISSUER_DID
+
+    # Mint a v2 credential normally, then re-shape into a v1 envelope that we
+    # re-sign — this is what a credential issued by the pre-migration code
+    # looks like after it was persisted to the DB.
+    vc = issue_credential("did:moltrust:test_v1_legacy", "TestCredential", {"k": "v"})
+    legacy = {
+        "@context": [VC_V1_CONTEXT, MOLTRUST_CONTEXT],
+        "type": vc["type"],
+        "issuer": vc["issuer"],
+        "issuanceDate": vc["validFrom"],
+        "expirationDate": vc["validUntil"],
+        "credentialSubject": vc["credentialSubject"],
+    }
+    # Re-sign with the same key the patched issuer uses
+    sk = _test_signing_key()
+    payload = json.dumps(legacy, sort_keys=True).encode()
+    sig = sk.sign(payload).signature.hex()
+    legacy["proof"] = {
+        "type": "Ed25519Signature2020",
+        "created": legacy["issuanceDate"],
+        "verificationMethod": f"{ISSUER_DID}#key-1",
+        "proofPurpose": "assertionMethod",
+        "proofValue": sig,
+    }
+    result = verify_credential(legacy)
+    assert result["valid"] is True, f"legacy v1 credential failed verify: {result}"
+
+
+def test_verify_v2_credential_roundtrip():
+    _ensure_test_signing_key()
+    from app.credentials import issue_credential, verify_credential
+    vc = issue_credential("did:moltrust:test_v2_roundtrip", "TestCredential", {"k": "v"})
+    result = verify_credential(vc)
+    assert result["valid"] is True, f"v2 roundtrip failed: {result}"
+
+
+def test_verify_expired_credential_v2():
+    _ensure_test_signing_key()
+    from app.credentials import verify_credential, ISSUER_DID
+
+    past = (datetime.datetime.utcnow() - datetime.timedelta(days=2)).isoformat() + "Z"
+    expired = {
+        "@context": [VC_V2_CONTEXT, MOLTRUST_CONTEXT],
+        "type": ["VerifiableCredential", "TestCredential"],
+        "issuer": ISSUER_DID,
+        "validFrom": past,
+        "validUntil": past,  # already expired
+        "credentialSubject": {"id": "did:moltrust:test_expired"},
+    }
+    sk = _test_signing_key()
+    payload = json.dumps(expired, sort_keys=True).encode()
+    sig = sk.sign(payload).signature.hex()
+    expired["proof"] = {
+        "type": "Ed25519Signature2020",
+        "created": past,
+        "verificationMethod": f"{ISSUER_DID}#key-1",
+        "proofPurpose": "assertionMethod",
+        "proofValue": sig,
+    }
+    result = verify_credential(expired)
+    assert result["valid"] is False
+    assert "expired" in result["error"].lower()