213 lines
7.9 KiB
Python
213 lines
7.9 KiB
Python
"""Hardware commands: hw.*"""
|
|
import struct
|
|
import asyncio
|
|
from .protocol import Cmd, PM3Status
|
|
from .transport import PM3Transport, PM3Response, PM3Error, ProgressCallback
|
|
|
|
|
|
class HWCommands:
|
|
"""Proxmark3 hardware commands."""
|
|
|
|
def __init__(self, transport: PM3Transport):
|
|
self._t = transport
|
|
|
|
async def ping(self, length: int = 32) -> dict:
|
|
"""Send ping and verify echo. Returns {success, length, data}."""
|
|
data = bytes(i & 0xFF for i in range(length))
|
|
resp = await self._t.send_ng(Cmd.PING, data)
|
|
match = resp.data == data
|
|
return {"success": match, "length": len(resp.data), "data": resp.data.hex()}
|
|
|
|
async def version(self) -> dict:
|
|
"""Get firmware version. Returns {chip_id, section_size, version_string}."""
|
|
resp = await self._t.send_ng(Cmd.VERSION, timeout=3.0)
|
|
if len(resp.data) < 12:
|
|
raise PM3Error("Version response too short")
|
|
chip_id, section_size, vstr_len = struct.unpack_from("<III", resp.data, 0)
|
|
vstr = resp.data[12:12 + vstr_len].decode("utf-8", errors="replace").rstrip("\x00")
|
|
return {
|
|
"chip_id": f"0x{chip_id:08X}",
|
|
"section_size": section_size,
|
|
"version_string": vstr,
|
|
}
|
|
|
|
async def status(self, speed_test_timeout: int = 0) -> dict:
|
|
"""Get device status. Returns {status}."""
|
|
if speed_test_timeout > 0:
|
|
payload = struct.pack("<i", speed_test_timeout)
|
|
else:
|
|
payload = b""
|
|
resp = await self._t.send_ng(Cmd.STATUS, payload, timeout=2.0 + speed_test_timeout / 1000)
|
|
return {"status": resp.status}
|
|
|
|
async def led(self, led: str | int = "all", *,
|
|
on: bool = False, off: bool = False, toggle: bool = False,
|
|
brightness: int = -1, pulse: bool = False, fade: bool = False,
|
|
blink: bool = False, speed: int = 500, count: int = 5) -> dict:
|
|
"""Control LEDs on the Proxmark3.
|
|
|
|
Args:
|
|
led: Color name(s), letter(s), or bitmask int.
|
|
Names: "green", "red", "orange", "blue", "all"
|
|
Letters: "a", "b", "c", "d"
|
|
Comma-separated: "green,red"
|
|
Bitmask: 1=A, 2=B, 4=C, 8=D
|
|
|
|
Action kwargs (pick one):
|
|
on: Turn LED on
|
|
off: Turn LED off
|
|
toggle: Toggle LED state
|
|
brightness: Set PWM brightness 0-100 (PWM LEDs only)
|
|
pulse: Pulse effect (PWM LEDs only)
|
|
fade: Fade effect (PWM LEDs only)
|
|
blink: Blink effect (all LEDs)
|
|
|
|
Effect params:
|
|
speed: Cycle time in ms (default 500)
|
|
count: Repeat count, 0=infinite (default 5)
|
|
"""
|
|
# Resolve LED mask
|
|
if isinstance(led, int):
|
|
led_mask = led & 0x0F
|
|
else:
|
|
led_mask = self._parse_led_str(led)
|
|
|
|
# Resolve action
|
|
if brightness >= 0:
|
|
action_code = 3
|
|
brightness_val = min(max(brightness, 0), 100)
|
|
elif off:
|
|
action_code = 0
|
|
brightness_val = 0
|
|
elif toggle:
|
|
action_code = 2
|
|
brightness_val = 0
|
|
elif pulse:
|
|
action_code = 4
|
|
brightness_val = 0
|
|
elif fade:
|
|
action_code = 5
|
|
brightness_val = 0
|
|
elif blink:
|
|
action_code = 6
|
|
brightness_val = 0
|
|
elif on or True: # default to "on"
|
|
action_code = 1
|
|
brightness_val = 0
|
|
|
|
payload = struct.pack("<BBBHH", led_mask, action_code, brightness_val, speed, count)
|
|
resp = await self._t.send_ng(Cmd.LED_CONTROL, payload)
|
|
return {"status": resp.status}
|
|
|
|
@staticmethod
|
|
def _parse_led_str(led: str) -> int:
|
|
"""Parse LED string into bitmask. Accepts names, letters, or 'all'."""
|
|
LED_MAP = {
|
|
"a": 0x01, "b": 0x02, "c": 0x04, "d": 0x08,
|
|
"green": 0x02, "orange": 0x01, "red": 0x04, "blue": 0x08,
|
|
"all": 0x0F,
|
|
}
|
|
mask = 0
|
|
for token in led.lower().replace(" ", "").split(","):
|
|
if token in LED_MAP:
|
|
mask |= LED_MAP[token]
|
|
else:
|
|
raise ValueError(f"Unknown LED: {token!r}. Use: {', '.join(LED_MAP.keys())}")
|
|
return mask
|
|
|
|
async def dbg(self, level: int | None = None) -> dict:
|
|
"""Get or set debug level. None=get, 0-4=set."""
|
|
if level is None:
|
|
resp = await self._t.send_ng(Cmd.GET_DBGMODE)
|
|
if len(resp.data) >= 1:
|
|
return {"level": resp.data[0]}
|
|
return {"level": 0}
|
|
else:
|
|
payload = struct.pack("<B", level)
|
|
resp = await self._t.send_ng(Cmd.SET_DBGMODE, payload)
|
|
return {"level": level, "status": resp.status}
|
|
|
|
async def tearoff(self, delay_us: int = 0, on: bool = True) -> dict:
|
|
"""Set tearoff parameters."""
|
|
payload = struct.pack("<HbBB", delay_us, 0, int(on), int(not on))
|
|
resp = await self._t.send_ng(Cmd.SET_TEAROFF, payload)
|
|
return {"status": resp.status}
|
|
|
|
async def break_loop(self) -> None:
|
|
"""Send break to stop long-running operations."""
|
|
await self._t.send_ng_no_response(Cmd.BREAK_LOOP)
|
|
|
|
async def standalone(self) -> dict:
|
|
"""Start standalone mode."""
|
|
resp = await self._t.send_ng(Cmd.STANDALONE, timeout=5.0)
|
|
return {"status": resp.status}
|
|
|
|
async def reset(self) -> None:
|
|
"""Reset the device."""
|
|
await self._t.send_ng_no_response(Cmd.HARDWARE_RESET)
|
|
|
|
async def capabilities(self) -> dict:
|
|
"""Get device capabilities."""
|
|
resp = await self._t.send_ng(Cmd.CAPABILITIES)
|
|
if len(resp.data) < 9:
|
|
raise PM3Error("Capabilities response too short")
|
|
version = resp.data[0]
|
|
baudrate = struct.unpack_from("<I", resp.data, 1)[0]
|
|
bigbuf_size = struct.unpack_from("<I", resp.data, 5)[0]
|
|
# Bit flags start at byte 9
|
|
flags = 0
|
|
if len(resp.data) > 9:
|
|
flags = struct.unpack_from("<I", resp.data, 9)[0] if len(resp.data) >= 13 else resp.data[9]
|
|
return {
|
|
"version": version,
|
|
"baudrate": baudrate,
|
|
"bigbuf_size": bigbuf_size,
|
|
"raw_flags": flags,
|
|
}
|
|
|
|
async def tune(self, on_progress: ProgressCallback = None) -> dict:
|
|
"""Run full antenna tuning. Returns LF/HF voltages, peak, Q factor."""
|
|
resp = await self._t.send_ng(Cmd.MEASURE_ANTENNA_TUNING, timeout=20.0)
|
|
if resp.status != 0:
|
|
raise PM3Error("Antenna tuning failed", status=resp.status)
|
|
if len(resp.data) < 28:
|
|
raise PM3Error("Tune response too short")
|
|
|
|
v_lf134, v_lf125, v_lfconf, v_hf, peak_v, peak_f, divisor = struct.unpack_from("<IIIIIIi", resp.data, 0)
|
|
results = list(resp.data[28:28 + 256]) if len(resp.data) >= 284 else []
|
|
|
|
ANTENNA_ERROR = 1.09
|
|
LF_DIVISOR_125 = 95
|
|
LF_DIVISOR_134 = 88
|
|
|
|
def div2freq(d):
|
|
return 12000.0 / (d + 1) if d > 0 else 0
|
|
|
|
result = {
|
|
"lf": {
|
|
"125kHz_mV": v_lf125,
|
|
"125kHz_V": round(v_lf125 * ANTENNA_ERROR / 1000, 2),
|
|
"134kHz_mV": v_lf134,
|
|
"134kHz_V": round(v_lf134 * ANTENNA_ERROR / 1000, 2),
|
|
"conf_mV": v_lfconf,
|
|
"peak_mV": peak_v,
|
|
"peak_V": round(peak_v * ANTENNA_ERROR / 1000, 2),
|
|
"peak_freq_kHz": round(div2freq(peak_f), 2),
|
|
"peak_divisor": peak_f,
|
|
"divisor": divisor,
|
|
},
|
|
"hf": {
|
|
"13.56MHz_mV": v_hf,
|
|
"13.56MHz_V": round(v_hf * ANTENNA_ERROR / 1000, 2),
|
|
},
|
|
"scan_results": results,
|
|
}
|
|
|
|
if on_progress:
|
|
event = {"type": "tune_complete", "result": result}
|
|
ret = on_progress(event)
|
|
if asyncio.iscoroutine(ret):
|
|
await ret
|
|
|
|
return result
|