pm3py/tests/test_sim_pm3medium.py

"""Tests for PM3Medium — hardware-backed Medium using Proxmark3."""
import asyncio
import struct
import pytest
from unittest.mock import AsyncMock, MagicMock

from pm3py.sim.frame import RFFrame
from pm3py.sim.pm3medium import PM3ReaderMedium
from pm3py.core.transport import PM3Response
from pm3py.core.protocol import Cmd, PM3Status


def run(coro):
    return asyncio.get_event_loop().run_until_complete(coro)


def _mock_transport():
    t = AsyncMock()
    return t


def _14a_scan_response(uid=b"\x01\x02\x03\x04", atqa=b"\x04\x00", sak=0x08):
    """Build a mock iso14a_card_select_t response."""
    # iso14a_card_select_t: uid[10] + uidlen[1] + atqa[2] + sak[1] + ats_len[1] + ats[256]
    data = uid + b"\x00" * (10 - len(uid))
    data += bytes([len(uid)])
    data += atqa
    data += bytes([sak])
    data += bytes([0])  # ats_len
    data += b"\x00" * 256  # ats
    # Select status comes from oldarg[0]: 1=OK+ATS, 2=OK no ATS
    return PM3Response(cmd=Cmd.HF_ISO14443A_READER, status=PM3Status.SUCCESS,
                       reason=0, ng=False, data=data,
                       oldarg=(2, 0, 0))


def _14a_raw_response(data=b"\x90\x00"):
    """Build a mock raw APDU response."""
    return PM3Response(cmd=Cmd.HF_ISO14443A_READER, status=PM3Status.SUCCESS,
                       reason=0, ng=False, data=data,
                       oldarg=(len(data), 0, 0))


def _15693_response(data=b"\x00"):
    """Build a mock ISO 15693 response."""
    return PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=PM3Status.SUCCESS,
                       reason=0, ng=True, data=data, oldarg=None)


# ---------------------------------------------------------------------------
# PM3ReaderMedium — 14443-A reader
# ---------------------------------------------------------------------------

class TestPM3ReaderMedium14443A:
    def test_transmit_receive_14a_scan(self):
        t = _mock_transport()
        t.send_mix.return_value = _14a_scan_response()
        medium = PM3ReaderMedium(t, protocol="14443a")

        # Transmit REQA (triggers scan)
        run(medium.transmit_reader(RFFrame.from_hex("26")))
        resp = run(medium.receive_reader())

        assert resp is not None
        assert t.send_mix.called

    def test_transmit_receive_14a_raw(self):
        t = _mock_transport()
        t.send_mix.return_value = _14a_raw_response(b"\x90\x00")
        medium = PM3ReaderMedium(t, protocol="14443a")

        # Send a raw APDU
        apdu = b"\x00\xA4\x04\x00"
        run(medium.transmit_reader(RFFrame.from_bytes(apdu)))
        resp = run(medium.receive_reader())

        assert resp is not None
        assert resp.data == b"\x90\x00"

    def test_dropfield(self):
        t = _mock_transport()
        t.send_ng_no_response = AsyncMock()
        medium = PM3ReaderMedium(t, protocol="14443a")

        run(medium.dropfield())
        t.send_ng_no_response.assert_called_once()


# ---------------------------------------------------------------------------
# PM3ReaderMedium — 15693 reader
# ---------------------------------------------------------------------------

class TestPM3ReaderMedium15693:
    def test_transmit_receive_15693_inventory(self):
        t = _mock_transport()
        uid_lsb = b"\x03\x1B\x6A\x5C\x50\x01\x04\xE0"
        resp_data = b"\x00\x00" + uid_lsb  # flags + dsfid + uid
        t.send_ng.return_value = _15693_response(resp_data)
        medium = PM3ReaderMedium(t, protocol="15693")

        inv = bytes([0x26, 0x01, 0x00])
        run(medium.transmit_reader(RFFrame.from_bytes(inv)))
        resp = run(medium.receive_reader())

        assert resp is not None
        assert len(resp.data) >= 10

    def test_transmit_receive_15693_read_block(self):
        t = _mock_transport()
        block_data = b"\x00\xDE\xAD\xBE\xEF"  # flags + data
        t.send_ng.return_value = _15693_response(block_data)
        medium = PM3ReaderMedium(t, protocol="15693")

        read_cmd = bytes([0x22, 0x20]) + b"\x01" * 8 + bytes([0x00])
        run(medium.transmit_reader(RFFrame.from_bytes(read_cmd)))
        resp = run(medium.receive_reader())

        assert resp is not None
        assert resp.data[1:5] == b"\xDE\xAD\xBE\xEF"

    def test_timeout_returns_none(self):
        t = _mock_transport()
        t.send_ng.side_effect = TimeoutError()
        medium = PM3ReaderMedium(t, protocol="15693")

        inv = bytes([0x26, 0x01, 0x00])
        run(medium.transmit_reader(RFFrame.from_bytes(inv)))
        resp = run(medium.receive_reader())

        assert resp is None


# ---------------------------------------------------------------------------
# PM3ReaderMedium — attach/detach (not applicable for hardware)
# ---------------------------------------------------------------------------

class TestPM3ReaderMediumLimitations:
    def test_attach_raises(self):
        """Hardware medium doesn't support attaching software transponders."""
        t = _mock_transport()
        medium = PM3ReaderMedium(t, protocol="14443a")
        with pytest.raises(NotImplementedError):
            run(medium.attach(None))

    def test_detach_raises(self):
        t = _mock_transport()
        medium = PM3ReaderMedium(t, protocol="14443a")
        with pytest.raises(NotImplementedError):
            run(medium.detach(0))