feat(pm3py): ISO15693 + MIFARE Classic commands
This commit is contained in:
@@ -4,6 +4,8 @@ from .hw import HWCommands
|
||||
from .lf import LFCommands
|
||||
from .hf import HFCommands
|
||||
from .hf_14a import HF14ACommands
|
||||
from .hf_15 import HF15Commands
|
||||
from .hf_mf import HFMFCommands
|
||||
|
||||
|
||||
class Proxmark3:
|
||||
@@ -24,6 +26,8 @@ class Proxmark3:
|
||||
self.lf = LFCommands(self._transport)
|
||||
self.hf = HFCommands(self._transport)
|
||||
self.hf.a14 = HF14ACommands(self._transport)
|
||||
self.hf.iso15 = HF15Commands(self._transport)
|
||||
self.hf.mf = HFMFCommands(self._transport)
|
||||
|
||||
async def connect(self) -> None:
|
||||
await self._transport.connect()
|
||||
|
||||
93
pm3py/hf_15.py
Normal file
93
pm3py/hf_15.py
Normal file
@@ -0,0 +1,93 @@
|
||||
"""ISO 15693 commands: hf.15.*"""
|
||||
import struct
|
||||
from .protocol import Cmd
|
||||
from .transport import PM3Transport, PM3Error
|
||||
|
||||
# ISO15 PM3 flags
|
||||
ISO15_CONNECT = 0x01
|
||||
ISO15_NO_DISCONNECT = 0x02
|
||||
ISO15_RAW = 0x04
|
||||
ISO15_APPEND_CRC = 0x08
|
||||
ISO15_READ_RESPONSE = 0x10
|
||||
|
||||
|
||||
class HF15Commands:
|
||||
"""ISO 15693 commands."""
|
||||
|
||||
def __init__(self, transport: PM3Transport):
|
||||
self._t = transport
|
||||
|
||||
async def scan(self) -> dict:
|
||||
"""Scan for ISO15693 tags using inventory command."""
|
||||
# ISO15693 inventory: flags=0x06 (high data rate + inventory), cmd=0x01
|
||||
iso_cmd = bytes([0x06, 0x01, 0x00]) # flags, inventory cmd, mask_len=0
|
||||
flags = ISO15_CONNECT | ISO15_READ_RESPONSE | ISO15_APPEND_CRC
|
||||
payload = struct.pack("<BH", flags, len(iso_cmd)) + iso_cmd
|
||||
resp = await self._t.send_ng(Cmd.HF_ISO15693_COMMAND, payload, timeout=5.0)
|
||||
|
||||
if resp.status != 0 or len(resp.data) < 10:
|
||||
return {"found": False, "uid": None}
|
||||
|
||||
resp_flags = resp.data[0]
|
||||
dsfid = resp.data[1]
|
||||
uid = resp.data[2:10]
|
||||
|
||||
return {
|
||||
"found": True,
|
||||
"uid": uid[::-1].hex(), # ISO15693 UIDs are reversed
|
||||
"dsfid": dsfid,
|
||||
"response_flags": resp_flags,
|
||||
}
|
||||
|
||||
async def rdbl(self, block: int, uid: str | None = None) -> dict:
|
||||
"""Read a single block."""
|
||||
if uid:
|
||||
uid_bytes = bytes.fromhex(uid)[::-1] # reverse for wire
|
||||
iso_cmd = bytes([0x22, 0x20]) + uid_bytes + bytes([block])
|
||||
else:
|
||||
iso_cmd = bytes([0x02, 0x20, block]) # addressed without UID
|
||||
|
||||
flags = ISO15_CONNECT | ISO15_READ_RESPONSE | ISO15_APPEND_CRC
|
||||
payload = struct.pack("<BH", flags, len(iso_cmd)) + iso_cmd
|
||||
resp = await self._t.send_ng(Cmd.HF_ISO15693_COMMAND, payload, timeout=5.0)
|
||||
|
||||
if resp.status != 0 or len(resp.data) < 2:
|
||||
return {"success": False, "error": resp.status}
|
||||
|
||||
resp_flags = resp.data[0]
|
||||
if resp_flags != 0:
|
||||
return {"success": False, "error_flag": resp_flags}
|
||||
|
||||
lock_status = resp.data[1]
|
||||
block_data = resp.data[2:]
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"block": block,
|
||||
"locked": bool(lock_status),
|
||||
"data": block_data.hex(),
|
||||
"raw": block_data,
|
||||
}
|
||||
|
||||
async def wrbl(self, block: int, data: bytes | str, uid: str | None = None) -> dict:
|
||||
"""Write a single block."""
|
||||
if isinstance(data, str):
|
||||
data = bytes.fromhex(data)
|
||||
|
||||
if uid:
|
||||
uid_bytes = bytes.fromhex(uid)[::-1]
|
||||
iso_cmd = bytes([0x22, 0x21]) + uid_bytes + bytes([block]) + data
|
||||
else:
|
||||
iso_cmd = bytes([0x02, 0x21, block]) + data
|
||||
|
||||
flags = ISO15_CONNECT | ISO15_READ_RESPONSE | ISO15_APPEND_CRC
|
||||
payload = struct.pack("<BH", flags, len(iso_cmd)) + iso_cmd
|
||||
resp = await self._t.send_ng(Cmd.HF_ISO15693_COMMAND, payload, timeout=5.0)
|
||||
|
||||
success = resp.status == 0 and len(resp.data) > 0 and resp.data[0] == 0
|
||||
return {"success": success, "block": block}
|
||||
|
||||
async def sniff(self) -> dict:
|
||||
"""Sniff ISO15693 traffic."""
|
||||
resp = await self._t.send_ng(Cmd.HF_ISO15693_SNIFF, timeout=30.0)
|
||||
return {"status": resp.status}
|
||||
138
pm3py/hf_mf.py
Normal file
138
pm3py/hf_mf.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""MIFARE Classic commands: hf.mf.*"""
|
||||
import struct
|
||||
from .protocol import Cmd
|
||||
from .transport import PM3Transport, PM3Error
|
||||
|
||||
MF_KEY_A = 0
|
||||
MF_KEY_B = 1
|
||||
MFBLOCK_SIZE = 16
|
||||
|
||||
|
||||
class HFMFCommands:
|
||||
"""MIFARE Classic commands."""
|
||||
|
||||
def __init__(self, transport: PM3Transport):
|
||||
self._t = transport
|
||||
|
||||
def _parse_key(self, key: str | bytes) -> bytes:
|
||||
if isinstance(key, str):
|
||||
key = bytes.fromhex(key)
|
||||
if len(key) != 6:
|
||||
raise ValueError("Key must be 6 bytes")
|
||||
return key
|
||||
|
||||
async def rdbl(self, block: int, key: str | bytes = "FFFFFFFFFFFF",
|
||||
key_type: int = MF_KEY_A) -> dict:
|
||||
"""Read a MIFARE Classic block."""
|
||||
key_bytes = self._parse_key(key)
|
||||
payload = struct.pack("<BB", block, key_type) + key_bytes
|
||||
resp = await self._t.send_ng(Cmd.HF_MIFARE_READBL, payload, timeout=5.0)
|
||||
|
||||
if resp.status != 0:
|
||||
return {"success": False, "error": resp.status}
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"block": block,
|
||||
"data": resp.data[:MFBLOCK_SIZE].hex(),
|
||||
"raw": resp.data[:MFBLOCK_SIZE],
|
||||
}
|
||||
|
||||
async def wrbl(self, block: int, data: str | bytes,
|
||||
key: str | bytes = "FFFFFFFFFFFF",
|
||||
key_type: int = MF_KEY_A) -> dict:
|
||||
"""Write a MIFARE Classic block."""
|
||||
key_bytes = self._parse_key(key)
|
||||
if isinstance(data, str):
|
||||
data = bytes.fromhex(data)
|
||||
if len(data) != MFBLOCK_SIZE:
|
||||
raise ValueError(f"Block data must be {MFBLOCK_SIZE} bytes")
|
||||
|
||||
# MIX format: arg0=blockno, arg1=keytype
|
||||
# payload = key(6) + reserved(10) + data(16) = 32 bytes
|
||||
payload = key_bytes + b"\x00" * 10 + data
|
||||
resp = await self._t.send_mix(
|
||||
Cmd.HF_MIFARE_WRITEBL, arg0=block, arg1=key_type,
|
||||
payload=payload, timeout=5.0,
|
||||
)
|
||||
|
||||
success = resp.oldarg[0] > 0 if resp.oldarg else resp.status == 0
|
||||
return {"success": success, "block": block}
|
||||
|
||||
async def rdsc(self, sector: int, key: str | bytes = "FFFFFFFFFFFF",
|
||||
key_type: int = MF_KEY_A) -> dict:
|
||||
"""Read an entire MIFARE Classic sector."""
|
||||
# Sectors 0-31 have 4 blocks, sectors 32-39 have 16 blocks
|
||||
if sector < 32:
|
||||
first_block = sector * 4
|
||||
num_blocks = 4
|
||||
else:
|
||||
first_block = 128 + (sector - 32) * 16
|
||||
num_blocks = 16
|
||||
|
||||
blocks = {}
|
||||
for i in range(num_blocks):
|
||||
block_num = first_block + i
|
||||
result = await self.rdbl(block_num, key=key, key_type=key_type)
|
||||
blocks[block_num] = result
|
||||
|
||||
return {
|
||||
"sector": sector,
|
||||
"first_block": first_block,
|
||||
"num_blocks": num_blocks,
|
||||
"blocks": blocks,
|
||||
}
|
||||
|
||||
async def chk(self, block: int, keys: list[str | bytes],
|
||||
key_type: int = MF_KEY_A) -> dict:
|
||||
"""Check which key works for a block."""
|
||||
for key in keys:
|
||||
try:
|
||||
result = await self.rdbl(block, key=key, key_type=key_type)
|
||||
if result.get("success"):
|
||||
key_hex = key if isinstance(key, str) else key.hex()
|
||||
return {"found": True, "key": key_hex, "key_type": key_type}
|
||||
except PM3Error:
|
||||
continue
|
||||
return {"found": False}
|
||||
|
||||
async def sniff(self) -> dict:
|
||||
"""Sniff MIFARE traffic."""
|
||||
resp = await self._t.send_ng(Cmd.HF_MIFARE_SNIFF, timeout=30.0)
|
||||
return {"status": resp.status}
|
||||
|
||||
async def sim(self, uid: str | bytes, size: str = "1k") -> dict:
|
||||
"""Simulate MIFARE Classic card.
|
||||
size: "mini", "1k", "2k", "4k"
|
||||
"""
|
||||
if isinstance(uid, str):
|
||||
uid = bytes.fromhex(uid)
|
||||
|
||||
size_flags = {"mini": 0x0000, "1k": 0x0040, "2k": 0x0080, "4k": 0x00C0}
|
||||
if size not in size_flags:
|
||||
raise ValueError(f"Unknown size: {size}")
|
||||
|
||||
uid_flag = {4: 0x0010, 7: 0x0020, 10: 0x0030}.get(len(uid))
|
||||
if uid_flag is None:
|
||||
raise ValueError("UID must be 4, 7, or 10 bytes")
|
||||
|
||||
flags = 0x0001 | uid_flag | size_flags[size] # FLAG_INTERACTIVE
|
||||
resp = await self._t.send_mix(
|
||||
Cmd.HF_MIFARE_SIMULATE, arg0=flags, payload=uid, timeout=60.0,
|
||||
)
|
||||
return {"status": resp.status}
|
||||
|
||||
async def nested(self, block: int, key: str | bytes = "FFFFFFFFFFFF",
|
||||
key_type: int = MF_KEY_A,
|
||||
target_block: int = 0, target_key_type: int = MF_KEY_A) -> dict:
|
||||
"""Run nested attack to recover keys."""
|
||||
key_bytes = self._parse_key(key)
|
||||
payload = struct.pack("<BB", block, key_type) + key_bytes
|
||||
payload += struct.pack("<BB", target_block, target_key_type)
|
||||
resp = await self._t.send_ng(Cmd.HF_MIFARE_NESTED, payload, timeout=30.0)
|
||||
return {"status": resp.status, "data": resp.data.hex()}
|
||||
|
||||
async def cident(self) -> dict:
|
||||
"""Identify magic (gen1/gen2) card type."""
|
||||
resp = await self._t.send_ng(Cmd.HF_MIFARE_CIDENT, timeout=5.0)
|
||||
return {"status": resp.status, "data": resp.data.hex() if resp.data else None}
|
||||
15
tests/test_hf_15.py
Normal file
15
tests/test_hf_15.py
Normal file
@@ -0,0 +1,15 @@
|
||||
import asyncio
|
||||
from unittest.mock import AsyncMock
|
||||
from pm3py.protocol import Cmd
|
||||
from pm3py.transport import PM3Response
|
||||
from pm3py.hf_15 import HF15Commands
|
||||
|
||||
def test_15_scan():
|
||||
t = AsyncMock()
|
||||
iso15 = HF15Commands(t)
|
||||
# Fake response: flags(1) + DSFID(1) + UID(8)
|
||||
resp_data = bytes([0x00, 0x00]) + bytes([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0xE0])
|
||||
t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0,
|
||||
reason=0, ng=True, data=resp_data)
|
||||
result = asyncio.get_event_loop().run_until_complete(iso15.scan())
|
||||
assert "uid" in result
|
||||
26
tests/test_hf_mf.py
Normal file
26
tests/test_hf_mf.py
Normal file
@@ -0,0 +1,26 @@
|
||||
import struct
|
||||
import asyncio
|
||||
from unittest.mock import AsyncMock
|
||||
from pm3py.protocol import Cmd
|
||||
from pm3py.transport import PM3Response
|
||||
from pm3py.hf_mf import HFMFCommands
|
||||
|
||||
def test_mf_rdbl():
|
||||
t = AsyncMock()
|
||||
mf = HFMFCommands(t)
|
||||
block_data = bytes(range(16))
|
||||
t.send_ng.return_value = PM3Response(cmd=Cmd.HF_MIFARE_READBL, status=0,
|
||||
reason=0, ng=True, data=block_data)
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
mf.rdbl(block=0, key="FFFFFFFFFFFF"))
|
||||
assert result["data"] == block_data.hex()
|
||||
assert len(result["raw"]) == 16
|
||||
|
||||
def test_mf_wrbl():
|
||||
t = AsyncMock()
|
||||
mf = HFMFCommands(t)
|
||||
t.send_mix.return_value = PM3Response(cmd=Cmd.ACK, status=0, reason=0,
|
||||
ng=False, data=b"", oldarg=[1, 0, 0])
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
mf.wrbl(block=4, key="FFFFFFFFFFFF", data="00" * 16))
|
||||
assert result["success"] is True
|
||||
Reference in New Issue
Block a user