290 lines
12 KiB
Python
290 lines
12 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."""
|
|
|
|
# PWM-capable LEDs by platform (bitmask)
|
|
# PM3 Easy: A (green) and B (red) support PWM
|
|
# RDV4: A (orange) and D (red2) support PWM
|
|
_PWM_LEDS_EASY = 0x03 # A=0x01 | B=0x02
|
|
_PWM_LEDS_RDV4 = 0x09 # A=0x01 | D=0x08
|
|
|
|
def __init__(self, transport: PM3Transport):
|
|
self._t = transport
|
|
self._is_rdv4: bool | None = None # cached after first capabilities() call
|
|
|
|
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)
|
|
"""
|
|
# Fetch capabilities if not cached (needed for color mapping + PWM validation)
|
|
if self._is_rdv4 is None:
|
|
try:
|
|
await self.capabilities()
|
|
except PM3Error:
|
|
pass # unknown platform, default to Easy mapping
|
|
|
|
# Resolve LED mask
|
|
if isinstance(led, int):
|
|
led_mask = led & 0x0F
|
|
else:
|
|
led_mask = self._parse_led_str(led)
|
|
|
|
# Resolve action — explicit False means "off"
|
|
if on is False and not any([off, toggle, pulse, fade, blink, brightness >= 0]):
|
|
action_code = 0
|
|
brightness_val = 0
|
|
elif 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
|
|
else: # default to "on"
|
|
action_code = 1
|
|
brightness_val = 0
|
|
|
|
ACTION_NAMES = {0: "off", 1: "on", 2: "toggle", 3: "pwm", 4: "pulse", 5: "fade", 6: "blink"}
|
|
|
|
# Validate PWM capability for PWM actions (brightness, pulse, fade)
|
|
if action_code in (3, 4, 5):
|
|
await self._check_pwm_capable(led_mask, ACTION_NAMES[action_code])
|
|
|
|
payload = struct.pack("<BBBHH", led_mask, action_code, brightness_val, speed, count)
|
|
await self._t.send_ng(Cmd.LED_CONTROL, payload)
|
|
return {
|
|
"led": led_mask,
|
|
"action": ACTION_NAMES[action_code],
|
|
"brightness": brightness_val if action_code == 3 else None,
|
|
"speed": speed if action_code >= 4 else None,
|
|
"count": count if action_code in (4, 6) else None,
|
|
}
|
|
|
|
def _parse_led_str(self, led: str) -> int:
|
|
"""Parse LED string into bitmask. Color mapping depends on platform.
|
|
|
|
Physical layout:
|
|
RDV4: A=orange, B=green, C=red, D=red2
|
|
PM3 Easy: A=green, B=red, C=orange, D=blue
|
|
"""
|
|
rdv4 = self._is_rdv4 # None if unknown, assume Easy
|
|
# Letters always map directly
|
|
LETTER_MAP = {"a": 0x01, "b": 0x02, "c": 0x04, "d": 0x08, "all": 0x0F}
|
|
# Color names differ by platform
|
|
COLOR_MAP_RDV4 = {"green": 0x02, "orange": 0x01, "red": 0x04, "red2": 0x08,
|
|
"g": 0x02, "o": 0x01, "r": 0x04, "r2": 0x08}
|
|
COLOR_MAP_EASY = {"green": 0x01, "orange": 0x04, "red": 0x02, "blue": 0x08,
|
|
"g": 0x01, "o": 0x04, "r": 0x02, "b": 0x08}
|
|
color_map = COLOR_MAP_RDV4 if rdv4 else COLOR_MAP_EASY
|
|
|
|
mask = 0
|
|
for token in led.lower().replace(" ", "").split(","):
|
|
if token in LETTER_MAP:
|
|
mask |= LETTER_MAP[token]
|
|
elif token in color_map:
|
|
mask |= color_map[token]
|
|
elif token == "blue" and rdv4:
|
|
raise ValueError("RDV4 has no blue LED")
|
|
elif token in ("red2", "r2") and not rdv4:
|
|
raise ValueError("PM3 Easy has no red2 LED, use 'blue' instead")
|
|
else:
|
|
raise ValueError(f"Unknown LED: {token!r}. Use color names or letters a-d")
|
|
return mask
|
|
|
|
async def _check_pwm_capable(self, led_mask: int, action: str) -> None:
|
|
"""Raise if any LEDs in mask don't support PWM."""
|
|
if self._is_rdv4 is None:
|
|
return # can't validate, let it through
|
|
|
|
pwm_mask = self._PWM_LEDS_RDV4 if self._is_rdv4 else self._PWM_LEDS_EASY
|
|
non_pwm = led_mask & ~pwm_mask
|
|
if non_pwm:
|
|
platform = "RDV4" if self._is_rdv4 else "PM3 Easy"
|
|
LED_NAMES = {0x01: "A", 0x02: "B", 0x04: "C", 0x08: "D"}
|
|
bad = [LED_NAMES[b] for b in (0x01, 0x02, 0x04, 0x08) if non_pwm & b]
|
|
good = [LED_NAMES[b] for b in (0x01, 0x02, 0x04, 0x08) if pwm_mask & b]
|
|
raise PM3Error(
|
|
f"LED(s) {','.join(bad)} do not support {action} on {platform}. "
|
|
f"PWM-capable LEDs: {','.join(good)}"
|
|
)
|
|
|
|
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]
|
|
|
|
# Parse bit flags (bytes 9+)
|
|
is_rdv4 = False
|
|
compiled_with_lf = False
|
|
compiled_with_iso14443a = False
|
|
compiled_with_iso15693 = False
|
|
if len(resp.data) > 12:
|
|
is_rdv4 = bool(resp.data[12] & 0x02) # byte 12, bit 1
|
|
if len(resp.data) > 9:
|
|
compiled_with_lf = bool(resp.data[9] & 0x80) # byte 9, bit 7
|
|
if len(resp.data) > 10:
|
|
compiled_with_iso14443a = bool(resp.data[10] & 0x40) # byte 10, bit 6
|
|
if len(resp.data) > 11:
|
|
compiled_with_iso15693 = bool(resp.data[11] & 0x01) # byte 11, bit 0
|
|
|
|
self._is_rdv4 = is_rdv4
|
|
|
|
return {
|
|
"version": version,
|
|
"baudrate": baudrate,
|
|
"bigbuf_size": bigbuf_size,
|
|
"is_rdv4": is_rdv4,
|
|
"compiled_with_lf": compiled_with_lf,
|
|
"compiled_with_iso14443a": compiled_with_iso14443a,
|
|
"compiled_with_iso15693": compiled_with_iso15693,
|
|
}
|
|
|
|
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
|