aliro-project/harness/tests/test_personalizer_orchestrator.py

"""Orchestrator tests — verifies the full APDU sequence is sent in the
right order, with the right shapes, and that a non-9000 response halts."""

import pytest
from cryptography.hazmat.primitives.asymmetric import ec

from aliro_harness.issuer.access_document import build_access_document
from aliro_harness.personalizer.apdus import (
    INS_COMMIT,
    INS_FINALIZE_ACCESS_DOC,
    INS_SET_CRED_ISSUER_PUBK,
    INS_SET_CRED_PRIV,
    INS_SET_CRED_PUBK,
    INS_SET_READER_PUBK,
    INS_WRITE_ACCESS_DOC,
)
from aliro_harness.personalizer.credentials import (
    extract_priv_scalar,
    extract_pub_xy,
)
from aliro_harness.personalizer.orchestrator import (
    PersonalizationError,
    TrustArtifacts,
    personalize_card,
)


class FakeTransport:
    """Records each APDU sent and returns 0x9000 unless preconfigured."""

    def __init__(self, sw_overrides: dict[int, int] | None = None) -> None:
        self.sent: list[bytes] = []
        self._sw_overrides = sw_overrides or {}

    def __call__(self, apdu: bytes) -> tuple[bytes, int]:
        idx = len(self.sent)
        self.sent.append(apdu)
        return b"", self._sw_overrides.get(idx, 0x9000)


@pytest.fixture
def artifacts() -> TrustArtifacts:
    cred = ec.generate_private_key(ec.SECP256R1())
    reader = ec.generate_private_key(ec.SECP256R1())
    issuer = ec.generate_private_key(ec.SECP256R1())
    return TrustArtifacts(
        credential_priv=extract_priv_scalar(cred),
        credential_pubk_xy=extract_pub_xy(cred.public_key()),
        reader_pubk_xy=extract_pub_xy(reader.public_key()),
        credential_issuer_pub=extract_pub_xy(issuer.public_key()),
        access_document=build_access_document(
            issuer_private_key=issuer,
            access_credential_public_key=cred.public_key(),
        ),
    )


def test_full_sequence_in_expected_order(artifacts):
    transport = FakeTransport()
    personalize_card(transport, artifacts)

    insns = [apdu[1] for apdu in transport.sent]
    # SELECT (0xA4) → SET_PRIV → SET_PUBK → SET_READER_PUBK →
    #   SET_CRED_ISSUER_PUBK → WRITE_AD+ → FINALIZE_AD → COMMIT.
    # SET_CRED_ISSUER_PUBK must land BEFORE WRITE_AD chunks so that by
    # the time FINALIZE runs, the issuer trust anchor is staged.
    assert insns[0] == 0xA4
    assert insns[1] == INS_SET_CRED_PRIV
    assert insns[2] == INS_SET_CRED_PUBK
    assert insns[3] == INS_SET_READER_PUBK
    assert insns[4] == INS_SET_CRED_ISSUER_PUBK
    # AD writes (variable count) then a single finalize, then commit.
    assert all(i == INS_WRITE_ACCESS_DOC for i in insns[5:-2])
    assert insns[-2] == INS_FINALIZE_ACCESS_DOC
    assert insns[-1] == INS_COMMIT


def test_credential_issuer_pubk_apdu_carries_exact_bytes(artifacts):
    transport = FakeTransport()
    personalize_card(transport, artifacts)
    # Find the INS_SET_CRED_ISSUER_PUBK APDU.
    issuer_apdu = next(a for a in transport.sent if a[1] == INS_SET_CRED_ISSUER_PUBK)
    # CLA INS P1 P2 Lc data
    assert issuer_apdu[0] == 0x80
    assert issuer_apdu[2] == 0x00
    assert issuer_apdu[3] == 0x00
    assert issuer_apdu[4] == 0x40  # Lc = 64
    assert issuer_apdu[5:] == artifacts.credential_issuer_pub
    assert len(artifacts.credential_issuer_pub) == 64


def test_select_apdu_carries_provisioning_aid(artifacts):
    transport = FakeTransport()
    personalize_card(transport, artifacts)
    select = transport.sent[0]
    assert select.hex().upper().endswith("A000000909ACCE559901")


def test_credential_priv_apdu_carries_exact_bytes(artifacts):
    transport = FakeTransport()
    personalize_card(transport, artifacts)
    apdu = transport.sent[1]  # SET_CRED_PRIV right after SELECT
    assert apdu[5:] == artifacts.credential_priv


def test_failure_at_set_credential_priv_aborts_sequence(artifacts):
    # SELECT (idx 0) returns 9000, SET_CRED_PRIV (idx 1) returns 6985.
    transport = FakeTransport(sw_overrides={1: 0x6985})
    with pytest.raises(PersonalizationError) as exc_info:
        personalize_card(transport, artifacts)
    assert exc_info.value.sw == 0x6985
    assert "SET credential_PrivK" in str(exc_info.value)
    # No further APDUs should have been sent past the failure.
    assert len(transport.sent) == 2


def test_access_document_chunks_concatenate_to_original(artifacts):
    transport = FakeTransport()
    personalize_card(transport, artifacts)
    write_apdus = [a for a in transport.sent if a[1] == INS_WRITE_ACCESS_DOC]
    # Each WRITE APDU: CLA INS P1 P2 Lc <chunk>
    reconstructed = b"".join(a[5:] for a in write_apdus)
    assert reconstructed == artifacts.access_document


def test_finalize_carries_total_ad_length_in_p1p2(artifacts):
    transport = FakeTransport()
    personalize_card(transport, artifacts)
    finalize = next(a for a in transport.sent if a[1] == INS_FINALIZE_ACCESS_DOC)
    total = (finalize[2] << 8) | finalize[3]
    assert total == len(artifacts.access_document)