diff --git a/pm3py/client.py b/pm3py/client.py index 8093458..9506ac9 100644 --- a/pm3py/client.py +++ b/pm3py/client.py @@ -2,6 +2,8 @@ from .transport import PM3Transport, PM3Response, ProgressCallback from .hw import HWCommands from .lf import LFCommands +from .hf import HFCommands +from .hf_14a import HF14ACommands class Proxmark3: @@ -20,6 +22,8 @@ class Proxmark3: self._transport = PM3Transport(port, baudrate) self.hw = HWCommands(self._transport) self.lf = LFCommands(self._transport) + self.hf = HFCommands(self._transport) + self.hf.a14 = HF14ACommands(self._transport) async def connect(self) -> None: await self._transport.connect() diff --git a/pm3py/hf.py b/pm3py/hf.py new file mode 100644 index 0000000..bbcc747 --- /dev/null +++ b/pm3py/hf.py @@ -0,0 +1,75 @@ +"""High frequency commands: hf.*""" +import struct +import asyncio +from .protocol import Cmd +from .transport import PM3Transport, PM3Error, ProgressCallback + + +class HFCommands: + """High frequency commands.""" + + def __init__(self, transport: PM3Transport): + self._t = transport + # Sub-modules attached by client.py + self.a14 = None # type: ignore + self.iso15 = None # type: ignore + self.mf = None # type: ignore + + async def tune(self, on_progress: ProgressCallback = None) -> dict: + """Measure HF antenna voltage.""" + resp = await self._t.send_ng(Cmd.MEASURE_ANTENNA_TUNING_HF, timeout=10.0) + voltage = struct.unpack_from("= 4 else 0 + result = { + "voltage_mV": voltage, + "voltage_V": round(voltage / 1000.0, 3), + } + if on_progress: + ret = on_progress({"type": "hf_tune", "result": result}) + if asyncio.iscoroutine(ret): + await ret + return result + + async def search(self, on_progress: ProgressCallback = None) -> dict: + """Search for HF tags. Tries 14a, then 15.""" + results = {} + + if on_progress: + ret = on_progress({"type": "hf_search", "phase": "14a"}) + if asyncio.iscoroutine(ret): + await ret + + if self.a14: + try: + card = await self.a14.scan() + if card.get("uid"): + results["iso14443a"] = card + except PM3Error: + pass + + if on_progress: + ret = on_progress({"type": "hf_search", "phase": "15693"}) + if asyncio.iscoroutine(ret): + await ret + + if self.iso15: + try: + tag = await self.iso15.scan() + if tag.get("uid"): + results["iso15693"] = tag + except PM3Error: + pass + + results["found"] = len(results) > 0 + return results + + async def sniff(self, samples: int = 0, skip: int = 0, + on_progress: ProgressCallback = None) -> dict: + """Sniff HF traffic.""" + payload = struct.pack(" dict: + """Turn off HF field.""" + resp = await self._t.send_ng(Cmd.HF_DROPFIELD) + return {"status": resp.status} diff --git a/pm3py/hf_14a.py b/pm3py/hf_14a.py new file mode 100644 index 0000000..ad129fc --- /dev/null +++ b/pm3py/hf_14a.py @@ -0,0 +1,96 @@ +"""ISO 14443-A commands: hf.14a.*""" +import struct +from .protocol import Cmd +from .transport import PM3Transport, PM3Error + +# ISO14A command flags (from pm3_cmd.h) +ISO14A_CONNECT = 0x0001 +ISO14A_NO_DISCONNECT = 0x0002 +ISO14A_RAW = 0x0008 +ISO14A_TOPAZMODE = 0x0100 +ISO14A_NO_RATS = 0x0200 + + +class HF14ACommands: + """ISO 14443-A commands.""" + + def __init__(self, transport: PM3Transport): + self._t = transport + + async def scan(self) -> dict: + """Scan for ISO14443A card. Returns UID, ATQA, SAK, ATS.""" + resp = await self._t.send_mix( + Cmd.HF_ISO14443A_READER, + arg0=ISO14A_CONNECT | ISO14A_NO_DISCONNECT, + timeout=5.0, + ) + + select_status = resp.oldarg[0] if resp.oldarg else 0 + if select_status == 0: + return {"found": False, "uid": None} + + # Parse iso14a_card_select_t from response data + data = resp.data + if len(data) < 13: # uid(10) + uidlen(1) + atqa(2) + return {"found": False, "uid": None} + + uid_raw = data[0:10] + uid_len = data[10] + atqa = data[11:13] + sak = data[13] if len(data) > 13 else 0 + ats_len = data[14] if len(data) > 14 else 0 + ats = data[15:15 + ats_len] if ats_len > 0 and len(data) > 15 else b"" + + uid = uid_raw[:uid_len] + + return { + "found": True, + "uid": uid.hex(), + "uid_len": uid_len, + "atqa": atqa.hex(), + "sak": sak, + "ats": ats.hex() if ats else None, + "select_status": select_status, # 1=OK+ATS, 2=OK no ATS, 3=proprietary + } + + async def raw(self, data: bytes, flags: int = 0, timeout_14a: int = 0) -> dict: + """Send raw ISO14443A command.""" + resp = await self._t.send_mix( + Cmd.HF_ISO14443A_READER, + arg0=flags | ISO14A_RAW | ISO14A_NO_DISCONNECT, + arg1=len(data), + arg2=timeout_14a, + payload=data, + timeout=5.0, + ) + return { + "status": resp.status, + "data": resp.data.hex() if resp.data else None, + "raw": resp.data, + } + + async def sniff(self, param: int = 0) -> dict: + """Sniff ISO14443A traffic.""" + resp = await self._t.send_mix( + Cmd.HF_ISO14443A_SNIFF, arg0=param, timeout=30.0, + ) + return {"status": resp.status} + + async def sim(self, uid: bytes, sak: int = 0x08, atqa: bytes = b"\x04\x00", + flags: int = 0) -> dict: + """Simulate ISO14443A card.""" + if len(uid) == 4: + uid_flag = 0x0010 + elif len(uid) == 7: + uid_flag = 0x0020 + elif len(uid) == 10: + uid_flag = 0x0030 + else: + raise ValueError("UID must be 4, 7, or 10 bytes") + + flag_val = flags | uid_flag | 0x0001 # FLAG_INTERACTIVE + payload = atqa + bytes([sak]) + uid + resp = await self._t.send_mix( + Cmd.HF_ISO14443A_SIMULATE, arg0=flag_val, timeout=60.0, payload=payload, + ) + return {"status": resp.status} diff --git a/tests/test_hf.py b/tests/test_hf.py new file mode 100644 index 0000000..cfd6a62 --- /dev/null +++ b/tests/test_hf.py @@ -0,0 +1,13 @@ +import asyncio +from unittest.mock import AsyncMock +from pm3py.protocol import Cmd +from pm3py.transport import PM3Response +from pm3py.hf import HFCommands + +def test_hf_tune(): + t = AsyncMock() + hf = HFCommands(t) + t.send_ng.return_value = PM3Response(cmd=Cmd.MEASURE_ANTENNA_TUNING_HF, status=0, + reason=0, ng=True, data=b"\x00" * 4) + result = asyncio.get_event_loop().run_until_complete(hf.tune()) + assert "voltage_mV" in result diff --git a/tests/test_hf_14a.py b/tests/test_hf_14a.py new file mode 100644 index 0000000..a28594c --- /dev/null +++ b/tests/test_hf_14a.py @@ -0,0 +1,25 @@ +import struct +import asyncio +from unittest.mock import AsyncMock +from pm3py.protocol import Cmd +from pm3py.transport import PM3Response +from pm3py.hf_14a import HF14ACommands + +def test_14a_scan(): + t = AsyncMock() + hf14a = HF14ACommands(t) + # Build iso14a_card_select_t response + uid = b"\x04\x01\x02\x03\x04\x05\x06" + card = uid + b"\x00" * 3 # uid[10] + card += bytes([7]) # uidlen + card += b"\x44\x00" # atqa + card += bytes([0x08]) # sak + card += bytes([0]) # ats_len + card += b"\x00" * 256 # ats + resp = PM3Response(cmd=Cmd.ACK, status=0, reason=0, ng=False, + data=card, oldarg=[1, 0, 0]) + t.send_mix.return_value = resp + result = asyncio.get_event_loop().run_until_complete(hf14a.scan()) + assert result["uid"] == "04010203040506" + assert result["uid_len"] == 7 + assert result["sak"] == 0x08