praxis/tests/test_vc_issuer.py

"""VC issuer unit tests (SLICE-09 TASK-09-05).

Covers: key generation, sign/verify round-trip, tamper detection (flip a byte
in payload → verify fails), JCS canonicalization determinism (same dict → same
bytes, run twice), status list set/get, revocation invalidates verification.
"""

from __future__ import annotations

import asyncio
import base64
import json
from pathlib import Path

import nacl.signing
import pytest

from db.migrate import apply_migrations
from db.store import PraxisStore
from server.vc import issuer, issuer_keys
from server.vc.status_list import BitstringStatusList


def _await(coro):
    return asyncio.run(coro)


@pytest.fixture
def tmp_db(tmp_path: Path) -> Path:
    return tmp_path / "test_vc.db"


def _make_store(db_path: Path) -> PraxisStore:
    apply_migrations(db_path)
    return PraxisStore(db_path)


def test_init_issuer_key_generates_ed25519_keypair(tmp_db: Path):
    store = _make_store(tmp_db)
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    assert kp.key_id.startswith("key-")
    assert len(kp.public_key_b64) > 0
    pk_bytes = base64.b64decode(kp.public_key_b64)
    assert len(pk_bytes) == 32
    assert bytes(kp.verify_key) == pk_bytes


def test_sign_verify_round_trip(tmp_db: Path):
    store = _make_store(tmp_db)
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    payload = issuer.build_vc_payload(
        learner_ref="learner-1",
        path="customer-service",
        scenarios_passed=["s1", "s2", "s3"],
        rubric_score=4.1,
        completed_weeks=6,
        evidence=[{"type": "Evidence", "rubricMean": 4.1}],
        status_list_index=0,
    )
    secured, sig_b64 = issuer.sign(payload, kp.signing_key, kp.key_id)
    assert issuer.verify_proof(secured, kp.verify_key) is True
    sig = base64.b64decode(sig_b64)
    assert len(sig) == 64


def test_tamper_detection_flipped_byte_fails(tmp_db: Path):
    store = _make_store(tmp_db)
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    payload = issuer.build_vc_payload(
        learner_ref="learner-1",
        path="customer-service",
        scenarios_passed=["s1"],
        rubric_score=3.8,
        completed_weeks=6,
        evidence=[],
        status_list_index=0,
    )
    secured, _ = issuer.sign(payload, kp.signing_key, kp.key_id)
    secured["credentialSubject"]["rubricScore"] = 1.1
    assert issuer.verify_proof(secured, kp.verify_key) is False


def test_tamper_proof_value_fails(tmp_db: Path):
    store = _make_store(tmp_db)
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    payload = issuer.build_vc_payload(
        learner_ref="learner-1",
        path="customer-service",
        scenarios_passed=["s1"],
        rubric_score=3.8,
        completed_weeks=6,
        evidence=[],
        status_list_index=0,
    )
    secured, sig_b64 = issuer.sign(payload, kp.signing_key, kp.key_id)
    flipped = bytearray(base64.b64decode(sig_b64))
    flipped[0] ^= 0x01
    secured["proof"]["proofValue"] = base64.b64encode(bytes(flipped)).decode("ascii")
    assert issuer.verify_proof(secured, kp.verify_key) is False


def test_jcs_canonicalization_determinism():
    d = {
        "b": 2,
        "a": 1,
        "nested": {"z": [3, 2, 1], "y": "hello"},
    }
    c1 = issuer.canonicalize(d)
    c2 = issuer.canonicalize(d)
    assert c1 == c2
    parsed = json.loads(c1.decode("utf-8"))
    assert parsed == {"a": 1, "b": 2, "nested": {"y": "hello", "z": [3, 2, 1]}}


def test_jcs_key_ordering_is_sorted():
    d = {"zeta": 1, "alpha": 2, "mid": 3}
    c = issuer.canonicalize(d)
    text = c.decode("utf-8")
    assert text.index('"alpha"') < text.index('"mid"') < text.index('"zeta"')


def test_status_list_set_get_round_trip(tmp_db: Path):
    store = _make_store(tmp_db)
    sl = BitstringStatusList(store, "default")
    _await(sl.set_status(5, True))
    assert _await(sl.get_status(5)) is True
    assert _await(sl.get_status(6)) is False
    _await(sl.set_status(5, False))
    assert _await(sl.get_status(5)) is False


def test_status_list_allocate_slot_returns_free_index(tmp_db: Path):
    store = _make_store(tmp_db)
    sl = BitstringStatusList(store, "default")
    s1 = _await(sl.allocate_slot())
    s2 = _await(sl.allocate_slot())
    assert s1 == 0
    assert s2 == 1


def test_revocation_invalidates_verification(tmp_db: Path):
    store = _make_store(tmp_db)
    root = b"k" * 32
    kp = _await(issuer_keys.init_issuer_key(store, root))
    cred_id = _await(
        issuer.issue_credential(
            store=store,
            signing_key=kp.signing_key,
            key_id=kp.key_id,
            learner_id="learner-1",
            path="customer-service",
            scenarios_passed=["s1", "s2", "s3"],
            rubric_score=4.1,
            completed_weeks=6,
            evidence=[{"type": "Evidence", "rubricMean": 4.1}],
        )
    )
    row = _await(store.get_credential(cred_id))
    assert row is not None
    secured = json.loads(row["vc_payload_json"])
    assert issuer.verify_proof(secured, kp.verify_key) is True
    cs = secured["credentialStatus"]
    idx = int(cs["statusListIndex"])
    sl = BitstringStatusList(store, "default")
    _await(sl.set_status(idx, True))
    _await(store.set_credential_status(cred_id, "revoked"))
    revoked = _await(sl.get_status(idx))
    assert revoked is True


def test_credential_tier_is_formative_in_payload():
    payload = issuer.build_vc_payload(
        learner_ref="learner-1",
        path="customer-service",
        scenarios_passed=["s1"],
        rubric_score=4.0,
        completed_weeks=6,
        evidence=[],
        status_list_index=0,
    )
    assert payload["credentialTier"] == "formative"
    assert payload["credentialSubject"]["credentialTier"] == "formative"