praxis/tests/test_vc_migration.py

"""VC migration unit tests (TASK-04-05) — mocked stores.

Covers:
  - Migration script: v0.3 key archived as superseded, fresh key active,
    idempotent re-run.
  - G-027 first-boot path: no v0.3 active key in SQLite → skip archive,
    generate fresh key only.
  - Verification with PgStore: v0.4 VC (active key) verifies ✓; v0.3 VC
    (superseded key) verifies ✓ (R-VC-MIG-01 — the critical test).
  - get_public_key_row finds superseded key by id (verification fallback).
"""

from __future__ import annotations

import base64
import json
import uuid
from unittest.mock import AsyncMock, MagicMock

import nacl.signing
import pytest

from server.vc.issuer import build_vc_payload, sign, extract_key_id, verify_proof
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


def _b64(b: bytes) -> str:
    return base64.b64encode(b).decode("ascii")


# ── Migration script ────────────────────────────────────────────────────────


def _make_sqlite_store_with_v03_key(key_id="v03-key-aaa", public_key="pub-v03-b64"):
    store = MagicMock()
    store.get_active_signing_key_row = AsyncMock(
        return_value={"id": key_id, "public_key": public_key, "private_key_enc": b"\x01"}
    )
    return store


def _make_pg_store():
    store = MagicMock()
    store._rows = {}  # key_id -> row dict
    store._active = None

    async def init_issuer_key(key_id, public_key, private_key_enc):
        status = "active"
        if key_id in store._rows:
            # ON CONFLICT DO NOTHING — don't overwrite
            return
        store._rows[key_id] = {
            "id": key_id, "public_key": public_key,
            "private_key_enc": private_key_enc, "status": status,
        }
        store._active = key_id

    async def get_active_signing_key_row():
        if store._active is None:
            return None
        return dict(store._rows[store._active])

    async def get_public_key_row(key_id):
        r = store._rows.get(key_id)
        return dict(r) if r else None

    async def set_issuer_key_superseded(key_id):
        if key_id in store._rows:
            store._rows[key_id]["status"] = "superseded"
            if store._active == key_id:
                store._active = None

    store.init_issuer_key = init_issuer_key
    store.get_active_signing_key_row = get_active_signing_key_row
    store.get_public_key_row = get_public_key_row
    store.set_issuer_key_superseded = set_issuer_key_superseded
    return store


@pytest.mark.asyncio
async def test_migration_archives_v03_and_generates_fresh():
    sqlite = _make_sqlite_store_with_v03_key()
    pg = _make_pg_store()
    root = _load_root_key()
    result = await migrate_issuer_keys(sqlite, pg, root)
    assert result["archived_key_id"] == "v03-key-aaa"
    assert result["new_key_id"] is not None
    # v0.3 key is superseded in Postgres
    archived = await pg.get_public_key_row("v03-key-aaa")
    assert archived["status"] == "superseded"
    assert archived["public_key"] == "pub-v03-b64"
    # fresh key is active
    active = await pg.get_active_signing_key_row()
    assert active is not None
    assert active["id"] == result["new_key_id"]
    assert active["status"] == "active"


@pytest.mark.asyncio
async def test_migration_idempotent_when_active_key_exists():
    sqlite = _make_sqlite_store_with_v03_key()
    pg = _make_pg_store()
    root = _load_root_key()
    await migrate_issuer_keys(sqlite, pg, root)
    # second run — no-op
    result = await migrate_issuer_keys(sqlite, pg, root)
    assert result["archived_key_id"] is None
    assert result["new_key_id"] is None


@pytest.mark.asyncio
async def test_migration_g027_first_boot_no_v03_key():
    # G-027: no v0.3 active key in SQLite → skip archive, fresh key only.
    sqlite = MagicMock()
    sqlite.get_active_signing_key_row = AsyncMock(return_value=None)
    pg = _make_pg_store()
    root = _load_root_key()
    result = await migrate_issuer_keys(sqlite, pg, root)
    assert result["archived_key_id"] is None  # nothing archived
    assert result["new_key_id"] is not None   # fresh key generated
    active = await pg.get_active_signing_key_row()
    assert active is not None
    assert active["id"] == result["new_key_id"]


@pytest.mark.asyncio
async def test_migration_archives_before_activating_r_vc_mig_01():
    # R-VC-MIG-01: the v0.3 public key MUST be archived BEFORE the fresh
    # key is activated. We verify the ordering by checking that at no point
    # is there an active v0.4 key without the v0.3 key being present (as
    # superseded) in Postgres.
    sqlite = _make_sqlite_store_with_v03_key()
    pg = _make_pg_store()
    # Instrument init_issuer_key to assert the archive happened first.
    order = []
    orig_init = pg.init_issuer_key

    async def tracing_init(key_id, public_key, private_key_enc):
        order.append(key_id)
        await orig_init(key_id, public_key, private_key_enc)

    pg.init_issuer_key = tracing_init
    orig_super = pg.set_issuer_key_superseded

    async def tracing_super(key_id):
        order.append(f"supersede:{key_id}")
        await orig_super(key_id)

    pg.set_issuer_key_superseded = tracing_super
    root = _load_root_key()
    await migrate_issuer_keys(sqlite, pg, root)
    # The v0.3 key (v03-key-aaa) is init'd then superseded BEFORE the fresh
    # key is init'd (active).
    v03_idx = order.index("v03-key-aaa")
    sup_idx = order.index("supersede:v03-key-aaa")
    fresh_idx = [i for i, k in enumerate(order) if k not in ("v03-key-aaa",) and not k.startswith("supersede:")][0]
    assert v03_idx < sup_idx < fresh_idx


# ── Verification with PgStore (R-VC-MIG-01 critical test) ───────────────────


def _make_credential_store(rows: dict[str, dict]):
    store = MagicMock()
    async def get_credential(cid):
        return rows.get(cid)
    store.get_credential = get_credential
    # status list store (SQLite) — empty
    sl_store = MagicMock()
    sl_store.get_status_list = AsyncMock(return_value=None)
    sl_store.upsert_status_list = AsyncMock()
    return store, sl_store


@pytest.mark.asyncio
async def test_v03_vc_verifies_against_superseded_key_in_pg():
    """R-VC-MIG-01 critical: a v0.3 VC verifies against a Postgres store
    with the v0.3 public key archived as superseded."""
    # Generate a v0.3 keypair + credential.
    sk_v03 = nacl.signing.SigningKey.generate()
    vk_v03 = sk_v03.verify_key
    pub_v03_b64 = _b64(bytes(vk_v03))
    v03_key_id = "v03-key-real"
    payload = build_vc_payload(
        learner_ref="learner-1", path="cs-refund",
        scenarios_passed=["sc-1"], rubric_score=4.0, completed_weeks=6,
        evidence=[], credential_id="vc-v03-real", status_list_index=None,
    )
    secured, sig_b64 = sign(payload, sk_v03, v03_key_id)
    cred_row = {
        "id": "vc-v03-real", "learner_ref": "learner-1",
        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
        "signature_b64": sig_b64, "status": "active",
    }
    # Postgres store has the v0.3 key as superseded + the credential.
    pg = _make_pg_store()
    await pg.init_issuer_key(v03_key_id, pub_v03_b64, b"")
    await pg.set_issuer_key_superseded(v03_key_id)
    pg._rows[v03_key_id]["public_key"] = pub_v03_b64
    # add credential to pg via a separate mock get_credential
    async def get_cred(cid):
        if cid == "vc-v03-real":
            return cred_row
        return None
    pg.get_credential = get_cred
    # SQLite status-list store (empty → not revoked)
    sqlite_sl = MagicMock()
    sqlite_sl.get_status_list = AsyncMock(return_value=None)
    sqlite_sl.upsert_status_list = AsyncMock()
    result = await verify_credential(
        pg, "vc-v03-real", pg_store=pg, sqlite_store=sqlite_sl
    )
    assert result is not None
    assert result["valid"] is True, "v0.3 VC must verify against archived superseded key (R-VC-MIG-01)"


@pytest.mark.asyncio
async def test_v04_vc_verifies_against_active_key_in_pg():
    sk_v04 = nacl.signing.SigningKey.generate()
    vk_v04 = sk_v04.verify_key
    pub_v04_b64 = _b64(bytes(vk_v04))
    v04_key_id = "v04-key-fresh"
    payload = build_vc_payload(
        learner_ref="learner-2", path="cs-refund",
        scenarios_passed=["sc-1", "sc-2"], rubric_score=4.5, completed_weeks=6,
        evidence=[], credential_id="vc-v04-fresh", status_list_index=None,
    )
    secured, sig_b64 = sign(payload, sk_v04, v04_key_id)
    cred_row = {
        "id": "vc-v04-fresh", "learner_ref": "learner-2",
        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
        "signature_b64": sig_b64, "status": "active",
    }
    pg = _make_pg_store()
    await pg.init_issuer_key(v04_key_id, pub_v04_b64, b"\x09")
    pg._rows[v04_key_id]["public_key"] = pub_v04_b64
    async def get_cred(cid):
        return cred_row if cid == "vc-v04-fresh" else None
    pg.get_credential = get_cred
    sqlite_sl = MagicMock()
    sqlite_sl.get_status_list = AsyncMock(return_value=None)
    result = await verify_credential(
        pg, "vc-v04-fresh", pg_store=pg, sqlite_store=sqlite_sl
    )
    assert result is not None
    assert result["valid"] is True


@pytest.mark.asyncio
async def test_tampered_v03_vc_fails_verification():
    sk = nacl.signing.SigningKey.generate()
    vk = sk.verify_key
    pub_b64 = _b64(bytes(vk))
    key_id = "key-tamper"
    payload = build_vc_payload(
        learner_ref="learner-t", path="cs-refund",
        scenarios_passed=["sc-1"], rubric_score=4.0, completed_weeks=6,
        evidence=[], credential_id="vc-tamper", status_list_index=None,
    )
    secured, sig_b64 = sign(payload, sk, key_id)
    # Tamper: change the rubricScore after signing.
    secured["credentialSubject"]["rubricScore"] = 1.0
    cred_row = {
        "id": "vc-tamper", "learner_ref": "learner-t",
        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
        "signature_b64": sig_b64, "status": "active",
    }
    pg = _make_pg_store()
    await pg.init_issuer_key(key_id, pub_b64, b"")
    pg._rows[key_id]["public_key"] = pub_b64
    async def get_cred(cid):
        return cred_row if cid == "vc-tamper" else None
    pg.get_credential = get_cred
    sqlite_sl = MagicMock()
    sqlite_sl.get_status_list = AsyncMock(return_value=None)
    result = await verify_credential(
        pg, "vc-tamper", pg_store=pg, sqlite_store=sqlite_sl
    )
    assert result is not None
    assert result["valid"] is False


@pytest.mark.asyncio
async def test_verification_fallback_sqlite_when_pg_missing_credential():
    """G-011(b): credential not in Postgres → fall back to SQLite."""
    sk = nacl.signing.SigningKey.generate()
    vk = sk.verify_key
    pub_b64 = _b64(bytes(vk))
    key_id = "key-fallback"
    payload = build_vc_payload(
        learner_ref="learner-fb", path="cs-refund",
        scenarios_passed=["sc-1"], rubric_score=4.0, completed_weeks=6,
        evidence=[], credential_id="vc-fallback", status_list_index=None,
    )
    secured, sig_b64 = sign(payload, sk, key_id)
    sqlite_cred_row = {
        "id": "vc-fallback", "learner_ref": "learner-fb",
        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
        "signature_b64": sig_b64, "status": "active",
    }
    # Postgres has the key but NOT the credential.
    pg = _make_pg_store()
    await pg.init_issuer_key(key_id, pub_b64, b"")
    pg._rows[key_id]["public_key"] = pub_b64
    async def pg_get_cred(cid):
        return None  # not in Postgres
    pg.get_credential = pg_get_cred
    # SQLite has the credential + the key (v0.3 path).
    sqlite = MagicMock()
    async def sqlite_get_cred(cid):
        return sqlite_cred_row if cid == "vc-fallback" else None
    sqlite.get_credential = sqlite_get_cred
    sqlite.get_public_key_row = AsyncMock(return_value={
        "id": key_id, "public_key": pub_b64, "status": "active"
    })
    sqlite.get_status_list = AsyncMock(return_value=None)
    result = await verify_credential(
        sqlite, "vc-fallback", pg_store=pg, sqlite_store=sqlite
    )
    assert result is not None
    assert result["valid"] is True


@pytest.mark.asyncio
async def test_verification_sqlite_only_when_no_pg():
    """G-011(c): no Postgres → full v0.3 SQLite path."""
    sk = nacl.signing.SigningKey.generate()
    vk = sk.verify_key
    pub_b64 = _b64(bytes(vk))
    key_id = "key-sqlite-only"
    payload = build_vc_payload(
        learner_ref="learner-so", path="cs-refund",
        scenarios_passed=["sc-1"], rubric_score=4.0, completed_weeks=6,
        evidence=[], credential_id="vc-so", status_list_index=None,
    )
    secured, sig_b64 = sign(payload, sk, key_id)
    cred_row = {
        "id": "vc-so", "learner_ref": "learner-so",
        "vc_payload_json": json.dumps(secured, sort_keys=True, separators=(",", ":")),
        "signature_b64": sig_b64, "status": "active",
    }
    sqlite = MagicMock()
    async def get_cred(cid):
        return cred_row if cid == "vc-so" else None
    sqlite.get_credential = get_cred
    sqlite.get_public_key_row = AsyncMock(return_value={
        "id": key_id, "public_key": pub_b64, "status": "active"
    })
    sqlite.get_status_list = AsyncMock(return_value=None)
    result = await verify_credential(sqlite, "vc-so", pg_store=None, sqlite_store=sqlite)
    assert result is not None
    assert result["valid"] is True