aliro-project/harness/tests/fake_card.py

"""In-process Python port of AliroApplet's AUTH0+AUTH1 paths.

Used by test_reader_transaction_offline.py to validate the reader without
needing a real card or pyscard. Mirrors AliroApplet processAuth0 +
processAuth1 -- but only enough to satisfy a one-shot reader transaction.

Skips: provisioning, replay protection (we never call AUTH1 twice in tests),
step-up, mailbox, vendor extensions, error paths beyond happy-path returns.
"""

from dataclasses import dataclass

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import (
    decode_dss_signature,
    encode_dss_signature,
)
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

from aliro_harness.reader.auth1 import (
    build_table_812,
    build_table_813,
)
from aliro_harness.reader.crypto import derive_kdh
from aliro_harness.reader.key_derivation import (
    OFF_EXPEDITED_SK_DEVICE,
    build_salt_volatile,
    derive_expedited_session_keys,
)

# AIDs and INS values -- match the applet
EXPEDITED_AID = bytes.fromhex("A000000909ACCE5501")
INS_SELECT = 0xA4
INS_AUTH0 = 0x80
INS_AUTH1 = 0x81

# FCI bytes the real applet returns from sendExpeditedFci. Pre-computed so
# the fake card returns a byte-identical FCI.
#   6F 15                            -- FCI template, outer len = 21
#     84 09 A0 00 00 09 09 AC CE 55 01   -- DF name = EXPEDITED_AID
#     A5 08                               -- proprietary, inner len = 8
#       80 02 00 00                      -- application type = CSA (0x0000)
#       5C 02 01 00                      -- protocol version = 0x0100
EXPEDITED_FCI = bytes.fromhex(
    "6F15"
    "8409" "A000000909ACCE5501"
    "A508" "80020000" "5C020100"
)

# AUTH0 TLV tags (AliroApplet)
TAG_COMMAND_PARAMETERS = 0x41
TAG_AUTHENTICATION_POLICY = 0x42
TAG_PROTOCOL_VERSION = 0x5C
TAG_READER_EPUBK = 0x87
TAG_TRANSACTION_ID = 0x4C
TAG_READER_IDENTIFIER = 0x4D
TAG_CREDENTIAL_EPUBK = 0x86

# AUTH1 TLV tags
TAG_READER_SIGNATURE = 0x9E
TAG_READER_CERT = 0x90

PROTOCOL_VERSION_1_0 = 0x0100


@dataclass
class _FakeSession:
    """Per-transaction state captured during AUTH0, consumed by AUTH1."""

    reader_ephem_pub_uncompressed: bytes
    reader_group_id: bytes
    reader_group_sub_id: bytes
    transaction_id: bytes
    command_parameters: int
    authentication_policy: int
    credential_ephem_pub_uncompressed: bytes | None = None


@dataclass
class FakeAliroCard:
    """Holds provisioned trust state + per-transaction state.

    Mirrors just enough of AliroApplet to drive the reader through
    SELECT -> AUTH0 -> AUTH1 successfully.
    """

    # Provisioned trust state
    reader_pub: ec.EllipticCurvePublicKey
    credential_priv: ec.EllipticCurvePrivateKey
    credential_pub: ec.EllipticCurvePublicKey
    has_access_document: bool = False

    # Per-transaction state, set during processAuth0
    auth0_done: bool = False
    session: _FakeSession | None = None
    credential_ephem_priv: ec.EllipticCurvePrivateKey | None = None

    @classmethod
    def provisioned(
        cls,
        *,
        reader_pub: ec.EllipticCurvePublicKey,
        credential_priv: ec.EllipticCurvePrivateKey,
        credential_pub: ec.EllipticCurvePublicKey,
        has_access_document: bool = False,
    ) -> "FakeAliroCard":
        return cls(
            reader_pub=reader_pub,
            credential_priv=credential_priv,
            credential_pub=credential_pub,
            has_access_document=has_access_document,
        )

    # ---------------- Transport ----------------

    def transmit(self, apdu: bytes) -> tuple[bytes, int]:
        """Implements ISO 7816 SELECT + AUTH0 + AUTH1 INSes."""
        if len(apdu) < 4:
            return b"", 0x6700
        cla, ins, p1, _p2 = apdu[0], apdu[1], apdu[2], apdu[3]

        # SELECT is CLA=0x00
        if cla == 0x00 and ins == INS_SELECT and p1 == 0x04:
            if len(apdu) < 5:
                return b"", 0x6700
            lc = apdu[4]
            aid = apdu[5 : 5 + lc]
            if aid == EXPEDITED_AID:
                # Fresh SELECT clears any prior session state.
                self.auth0_done = False
                self.session = None
                self.credential_ephem_priv = None
                return EXPEDITED_FCI, 0x9000
            return b"", 0x6A82  # file not found

        if cla == 0x80 and ins == INS_AUTH0:
            return self._process_auth0(apdu)
        if cla == 0x80 and ins == INS_AUTH1:
            return self._process_auth1(apdu)
        if cla != 0x80:
            return b"", 0x6E00  # CLA not supported
        return b"", 0x6D00  # INS not supported

    # ---------------- AUTH0 ----------------

    def _process_auth0(self, apdu: bytes) -> tuple[bytes, int]:
        """Validate AUTH0 TLVs, capture session state, generate ephemeral,
        return 0x86 TLV (uncompressed 65B credential_ePubK)."""
        if len(apdu) < 5:
            return b"", 0x6700
        lc = apdu[4]
        data = apdu[5 : 5 + lc]
        if len(data) != lc:
            return b"", 0x6700

        try:
            parsed = self._parse_auth0(data)
        except _WrongData:
            return b"", 0x6A80

        # Capture session state
        self.session = _FakeSession(
            reader_ephem_pub_uncompressed=parsed["reader_ephem_pub"],
            reader_group_id=parsed["reader_group_id"],
            reader_group_sub_id=parsed["reader_group_sub_id"],
            transaction_id=parsed["transaction_id"],
            command_parameters=parsed["command_parameters"],
            authentication_policy=parsed["authentication_policy"],
        )

        # Generate fresh credential ephemeral keypair
        self.credential_ephem_priv = ec.generate_private_key(ec.SECP256R1())
        ephem_pub_uncompressed = self.credential_ephem_priv.public_key().public_bytes(
            encoding=serialization.Encoding.X962,
            format=serialization.PublicFormat.UncompressedPoint,
        )
        self.session.credential_ephem_pub_uncompressed = ephem_pub_uncompressed

        # Build response: 0x86 0x41 [65B]
        resp = bytes([TAG_CREDENTIAL_EPUBK, len(ephem_pub_uncompressed)]) + ephem_pub_uncompressed
        self.auth0_done = True
        return resp, 0x9000

    def _parse_auth0(self, data: bytes) -> dict:
        if len(data) == 0:
            raise _WrongData()
        seen = 0
        out = {}
        i = 0
        while i < len(data):
            if i + 2 > len(data):
                raise _WrongData()
            tag = data[i]
            length = data[i + 1]
            value_off = i + 2
            if value_off + length > len(data):
                raise _WrongData()
            value = data[value_off : value_off + length]

            if tag == TAG_COMMAND_PARAMETERS:
                if length != 1:
                    raise _WrongData()
                out["command_parameters"] = value[0]
                seen |= 0x01
            elif tag == TAG_AUTHENTICATION_POLICY:
                if length != 1:
                    raise _WrongData()
                out["authentication_policy"] = value[0]
                seen |= 0x02
            elif tag == TAG_PROTOCOL_VERSION:
                if length != 2:
                    raise _WrongData()
                version = (value[0] << 8) | value[1]
                if version != PROTOCOL_VERSION_1_0:
                    raise _WrongData()
                seen |= 0x04
            elif tag == TAG_READER_EPUBK:
                if length != 65 or value[0] != 0x04:
                    raise _WrongData()
                out["reader_ephem_pub"] = value
                seen |= 0x08
            elif tag == TAG_TRANSACTION_ID:
                if length != 16:
                    raise _WrongData()
                out["transaction_id"] = value
                seen |= 0x10
            elif tag == TAG_READER_IDENTIFIER:
                if length != 32:
                    raise _WrongData()
                out["reader_group_id"] = value[:16]
                out["reader_group_sub_id"] = value[16:]
                seen |= 0x20
            # else: unknown tag, silently accept (spec 1.4.3)
            i = value_off + length
        if seen != 0x3F:
            raise _WrongData()
        return out

    # ---------------- AUTH1 ----------------

    def _process_auth1(self, apdu: bytes) -> tuple[bytes, int]:
        """Verify reader sig, derive keys, sign Table 8-13 with credential
        priv, build Table 8-11 plaintext, GCM-encrypt, return."""
        if not self.auth0_done:
            return b"", 0x6985  # CONDITIONS_NOT_SATISFIED
        # Mirror applet: clear AUTH0_DONE before crypto, to prevent replay.
        self.auth0_done = False

        if len(apdu) < 5:
            return b"", 0x6700
        lc = apdu[4]
        data = apdu[5 : 5 + lc]
        if len(data) != lc:
            return b"", 0x6700

        try:
            auth1 = self._parse_auth1(data)
        except _WrongData:
            return b"", 0x6A80

        cmd_params = auth1["command_parameters"]
        reader_raw_sig = auth1["reader_signature"]

        # Reconstruct Table 8-12 and verify against captured reader pub
        reader_ephem_pub = self.session.reader_ephem_pub_uncompressed
        cred_ephem_pub = self.session.credential_ephem_pub_uncompressed
        txn_id = self.session.transaction_id
        reader_id_32 = (
            self.session.reader_group_id + self.session.reader_group_sub_id
        )

        table_812 = build_table_812(
            reader_id_32=reader_id_32,
            credential_ephem_pub_x=cred_ephem_pub[1:33],
            reader_ephem_pub_x=reader_ephem_pub[1:33],
            transaction_id=txn_id,
        )
        if not self._verify_reader_sig(table_812, reader_raw_sig):
            return b"", 0x6A80

        # Derive session keys
        sk_device = self._derive_sk_device()

        # Sign Table 8-13 with credential_priv
        table_813 = build_table_813(
            reader_id_32=reader_id_32,
            credential_ephem_pub_x=cred_ephem_pub[1:33],
            reader_ephem_pub_x=reader_ephem_pub[1:33],
            transaction_id=txn_id,
        )
        der_sig = self.credential_priv.sign(table_813, ec.ECDSA(hashes.SHA256()))
        r, s = decode_dss_signature(der_sig)
        ud_raw_sig = r.to_bytes(32, "big") + s.to_bytes(32, "big")

        # Build Table 8-11 plaintext
        plaintext = self._build_table_811_plaintext(cmd_params, ud_raw_sig)

        # Encrypt with sk_device, IV = 7x00 || 01 || 0x00000001
        iv = b"\x00" * 7 + b"\x01" + (1).to_bytes(4, "big")
        ct_and_tag = AESGCM(sk_device).encrypt(iv, plaintext, associated_data=None)
        return ct_and_tag, 0x9000

    def _parse_auth1(self, data: bytes) -> dict:
        if len(data) == 0:
            raise _WrongData()
        seen = 0
        out = {}
        i = 0
        while i < len(data):
            if i + 2 > len(data):
                raise _WrongData()
            tag = data[i]
            length = data[i + 1]
            value_off = i + 2
            if value_off + length > len(data):
                raise _WrongData()
            value = data[value_off : value_off + length]

            if tag == TAG_COMMAND_PARAMETERS:
                if length != 1:
                    raise _WrongData()
                out["command_parameters"] = value[0]
                seen |= 0x01
            elif tag == TAG_READER_SIGNATURE:
                if length != 64:
                    raise _WrongData()
                out["reader_signature"] = value
                seen |= 0x02
            elif tag == TAG_READER_CERT:
                # v1: rejected explicitly
                raise _WrongData()
            # else: unknown tag, silently accept
            i = value_off + length
        if seen != 0x03:
            raise _WrongData()
        return out

    def _verify_reader_sig(self, table_812: bytes, raw_sig_64: bytes) -> bool:
        if len(raw_sig_64) != 64:
            return False
        r = int.from_bytes(raw_sig_64[:32], "big")
        s = int.from_bytes(raw_sig_64[32:], "big")
        der = encode_dss_signature(r, s)
        try:
            self.reader_pub.verify(der, table_812, ec.ECDSA(hashes.SHA256()))
            return True
        except InvalidSignature:
            return False

    def _derive_sk_device(self) -> bytes:
        """Mirrors AliroApplet.deriveSessionKeys."""
        reader_ephem_pub = self.session.reader_ephem_pub_uncompressed
        cred_ephem_pub = self.session.credential_ephem_pub_uncompressed
        txn_id = self.session.transaction_id

        # Kdh: ECDH(credential_ephem_priv, reader_ephem_pub) -> HKDF with salt=txn_id
        kdh = derive_kdh(
            self.credential_ephem_priv,
            reader_ephem_pub,
            txn_id,
        )

        # x-coords
        reader_long_term_x = self._pub_x(self.reader_pub)
        credential_long_term_x = self._pub_x(self.credential_pub)

        salt = build_salt_volatile(
            reader_long_term_pub_x=reader_long_term_x,
            reader_group_id=self.session.reader_group_id,
            reader_group_sub_id=self.session.reader_group_sub_id,
            reader_ephem_pub_x=reader_ephem_pub[1:33],
            transaction_id=txn_id,
            command_parameters=self.session.command_parameters,
            authentication_policy=self.session.authentication_policy,
            credential_long_term_pub_x=credential_long_term_x,
        )

        info = cred_ephem_pub[1:33]
        derived = derive_expedited_session_keys(kdh, salt, info)
        return derived[OFF_EXPEDITED_SK_DEVICE : OFF_EXPEDITED_SK_DEVICE + 32]

    def _build_table_811_plaintext(self, cmd_params: int, ud_raw_sig: bytes) -> bytes:
        """Table 8-11 plaintext per AliroApplet.buildTable811Plaintext."""
        out = bytearray()
        if (cmd_params & 0x01) == 0:
            # 0x4E 0x08 [key_slot] = first 8B of SHA-1(uncompressed credential_PubK)
            cred_pub_uncomp = self.credential_pub.public_bytes(
                encoding=serialization.Encoding.X962,
                format=serialization.PublicFormat.UncompressedPoint,
            )
            sha1 = hashes.Hash(hashes.SHA1())
            sha1.update(cred_pub_uncomp)
            digest = sha1.finalize()
            out += bytes([0x4E, 0x08]) + digest[:8]
        else:
            # 0x5A 0x41 [0x04 || x || y]
            cred_pub_uncomp = self.credential_pub.public_bytes(
                encoding=serialization.Encoding.X962,
                format=serialization.PublicFormat.UncompressedPoint,
            )
            out += bytes([0x5A, 0x41]) + cred_pub_uncomp

        # 0x9E 0x40 [raw r||s UD signature]
        out += bytes([0x9E, 0x40]) + ud_raw_sig

        # 0x5E 0x02 [signaling_bitmap] (big-endian 16-bit)
        bitmap = 0
        if self.has_access_document:
            bitmap |= 0x0001  # bit 0: AD retrievable
            bitmap |= 0x0004  # bit 2: step-up AID required on NFC
        out += bytes([0x5E, 0x02, (bitmap >> 8) & 0xFF, bitmap & 0xFF])

        return bytes(out)

    @staticmethod
    def _pub_x(pub: ec.EllipticCurvePublicKey) -> bytes:
        nums = pub.public_numbers()
        return nums.x.to_bytes(32, "big")


class _WrongData(Exception):
    """Raised internally when TLV parsing fails; mapped to SW=6A80."""
    pass