praxis/server/vc/migrate_keys.py

"""VC issuer key migration SQLite → Postgres (TASK-04-03, D-051).

One-time migration procedure (R-VC-MIG-01 — highest-severity v0.4 risk):
  1. Read the v0.3 active public key from SQLite issuer_keys.
  2. Insert that public key into Postgres issuer_keys with status=
     'superseded' (private key NOT migrated — only the public key is
     archived for verification of already-issued v0.3 VCs).
  3. Generate a fresh Ed25519 keypair in Postgres issuer_keys with
     status='active' (encrypted at rest with the root key).
  4. Return {archived_key_id, new_key_id}.

R-VC-MIG-01 mitigation: the v0.3 public key is archived as superseded
BEFORE the fresh key is activated (step 2 before step 3). This guarantees
v0.3 VCs remain verifiable against the archived key.

G-027 (first-boot path): if SQLite has NO v0.3 active key (fresh deploy),
skip the archive step and only generate the fresh v0.4 keypair.

Idempotent: if Postgres already has an active key, the whole procedure is
a no-op. If Postgres already has a superseded key matching the v0.3 key_id,
skip step 2 (already archived) but still generate the fresh key if no
active key exists.
"""

from __future__ import annotations

import base64
import uuid
from typing import Any

import nacl.signing

from db.pg_store import PgStore
from db.store import PraxisStore
from server.vc.issuer_keys import _encrypt_private_key


async def _archive_v03_public_key(
    pg_store: PgStore, v03_key_id: str, v03_public_key: str
) -> None:
    """Insert the v0.3 public key into Postgres as superseded (idempotent)."""
    existing = await pg_store.get_public_key_row(v03_key_id)
    if existing is not None:
        return  # already archived (or present as active — leave as-is)
    await pg_store.init_issuer_key(v03_key_id, v03_public_key, b"")
    await pg_store.set_issuer_key_superseded(v03_key_id)


async def _generate_fresh_v04_key(
    pg_store: PgStore, root_key: bytes
) -> str:
    """Generate a fresh Ed25519 keypair in Postgres as active. Returns key_id."""
    signing_key = nacl.signing.SigningKey.generate()
    verify_key = signing_key.verify_key
    public_key_b64 = base64.b64encode(bytes(verify_key)).decode("ascii")
    private_key_enc = _encrypt_private_key(signing_key, root_key)
    key_id = f"key-{uuid.uuid4().hex[:12]}"
    await pg_store.init_issuer_key(key_id, public_key_b64, private_key_enc)
    return key_id


async def migrate_issuer_keys(
    sqlite_store: PraxisStore,
    pg_store: PgStore,
    root_key: bytes,
) -> dict[str, str | None]:
    """Run the one-time VC key migration. Idempotent.

    Returns {"archived_key_id": str | None, "new_key_id": str | None}.
    archived_key_id is None on the G-027 first-boot path (no v0.3 key).
    new_key_id is None if an active key already existed (no-op).
    """
    # If Postgres already has an active key, the whole migration is done.
    active = await pg_store.get_active_signing_key_row()
    if active is not None:
        return {"archived_key_id": None, "new_key_id": None}

    # Step 1 (G-027): read v0.3 active public key from SQLite. May be None
    # on a fresh deploy with no v0.3 history.
    v03_row = await sqlite_store.get_active_signing_key_row()
    archived_key_id: str | None = None
    if v03_row is not None:
        v03_key_id = v03_row["id"]
        v03_public_key = v03_row["public_key"]
        # Step 2 (R-VC-MIG-01): archive BEFORE activating the fresh key.
        await _archive_v03_public_key(pg_store, v03_key_id, v03_public_key)
        archived_key_id = v03_key_id

    # Step 3: generate the fresh v0.4 keypair as active.
    new_key_id = await _generate_fresh_v04_key(pg_store, root_key)
    return {"archived_key_id": archived_key_id, "new_key_id": new_key_id}


__all__ = ["migrate_issuer_keys"]