feat(lf): FSK stack + HID Prox demod (faithful lfdemod.c port)
New pm3py/lf/fsk.py — a function-for-function Python port of the Proxmark FSK stack (firmware/common/lfdemod.c): fsk_wave_demod (count samples between 0->1 transitions: short wave = fc/8, long = fc/10) + aggregate_bits (runs -> clock- resolution bits) = fskdemod, plus preamble_search, HIDdemodFSK, detectAWID hooks. Ported to track the C reference (and thus real-tag output), not reinvented. protocols.decode_hid: FSK2a + Manchester -> Wiegand. Format is told by frame structure, not parity alone — the 26-bit H10301 wire frame carries a 0x20 header at bit 37 + a sentinel 1 at bit 26 (per firmware add_HID_preamble); the 37-bit H10304 frame is header-less (37-bit Manchester count, no 0x20). Each candidate is still parity-validated by re-encoding through the shipped HIDProxTag, so a false alignment can't slip through. Added to the identify decoder chain after EM4100. Synthetic HID tests build the real add_HID_preamble framing (header+sentinel for 26-bit, bare 37-bit) and render fc/8/fc/10 square waves, so a green test means the demod inverts what a real HID card / PM3 sim emits. Full suite 1232 green. AWID (96-bit) framing decode + PSK/Indala + biphase/FDX-B queued next. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -16,6 +16,7 @@ _SAMPLE_COUNT = 15000
|
|||||||
# every emitted-credential decoder, tried in order (headered/most-specific first)
|
# every emitted-credential decoder, tried in order (headered/most-specific first)
|
||||||
_DECODERS = (
|
_DECODERS = (
|
||||||
protocols.decode_em4100,
|
protocols.decode_em4100,
|
||||||
|
protocols.decode_hid,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
207
pm3py/lf/fsk.py
Normal file
207
pm3py/lf/fsk.py
Normal file
@@ -0,0 +1,207 @@
|
|||||||
|
"""FSK demodulation — a faithful Python port of the Proxmark FSK stack.
|
||||||
|
|
||||||
|
Ported function-for-function from ``firmware/common/lfdemod.c`` (``fsk_wave_demod`` +
|
||||||
|
``aggregate_bits`` = ``fskdemod``, ``preambleSearch``, ``HIDdemodFSK``, ``detectAWID``) so it
|
||||||
|
tracks the C reference — and therefore real-tag output — rather than being a fresh reinvention.
|
||||||
|
The C thresholds against the signal mean and counts samples between 0->1 transitions: a short
|
||||||
|
wave (~fc/8) is one symbol, a long wave (~fc/10) the other; ``aggregate_bits`` then collapses
|
||||||
|
runs of like waves into clock-resolution bits. HID/AWID are FSK2a (fc/10 high, fc/8 low, RF/50).
|
||||||
|
|
||||||
|
Everything downstream (Wiegand fields, parity) is validated by the shipped ``HIDProxTag`` model.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
# lfdemod.h NOISE_AMPLITUDE_THRESHOLD — below this peak-to-mean swing the capture is "noise".
|
||||||
|
# Conservative; a real hardware capture tunes this alongside the ASK path.
|
||||||
|
NOISE_AMPLITUDE_THRESHOLD = 8
|
||||||
|
|
||||||
|
|
||||||
|
def _mean(samples) -> float:
|
||||||
|
return sum(samples) / len(samples) if samples else 0.0
|
||||||
|
|
||||||
|
|
||||||
|
def is_noise(samples) -> bool:
|
||||||
|
"""Port of signalprop.isnoise: too little amplitude above the mean to carry data."""
|
||||||
|
if len(samples) < 64:
|
||||||
|
return True
|
||||||
|
return (max(samples) - _mean(samples)) < NOISE_AMPLITUDE_THRESHOLD
|
||||||
|
|
||||||
|
|
||||||
|
def find_mod_start(src, mean: float, exp_wave: int) -> int:
|
||||||
|
"""Port of ``findModStart`` — skip leading noise/slow chip start-up to the first clean wave."""
|
||||||
|
n = len(src)
|
||||||
|
if n < 21:
|
||||||
|
return 0
|
||||||
|
i = 1
|
||||||
|
wave_cnt = 0
|
||||||
|
thresh_cnt = 0
|
||||||
|
above = src[0] >= mean
|
||||||
|
while i < n - 20:
|
||||||
|
if src[i] < mean and above:
|
||||||
|
thresh_cnt += 1
|
||||||
|
if thresh_cnt > 2 and wave_cnt < exp_wave + 1:
|
||||||
|
break
|
||||||
|
above = False
|
||||||
|
wave_cnt = 0
|
||||||
|
elif src[i] >= mean and not above:
|
||||||
|
thresh_cnt += 1
|
||||||
|
if thresh_cnt > 2 and wave_cnt < exp_wave + 1:
|
||||||
|
break
|
||||||
|
above = True
|
||||||
|
wave_cnt = 0
|
||||||
|
else:
|
||||||
|
wave_cnt += 1
|
||||||
|
if thresh_cnt > 10:
|
||||||
|
break
|
||||||
|
i += 1
|
||||||
|
return i
|
||||||
|
|
||||||
|
|
||||||
|
def fsk_wave_demod(samples, fchigh: int = 10, fclow: int = 8):
|
||||||
|
"""Port of ``fsk_wave_demod``: envelope -> one bit per FSK wave (1 = short/fc-low wave,
|
||||||
|
0 = long/fc-high wave), counting samples between consecutive 0->1 transitions. Returns
|
||||||
|
``(bits, start_idx)``."""
|
||||||
|
n = len(samples)
|
||||||
|
if n < 1024:
|
||||||
|
return [], 0
|
||||||
|
mean = _mean(samples)
|
||||||
|
thr = [0 if s < mean else 1 for s in samples]
|
||||||
|
|
||||||
|
idx = find_mod_start(samples, mean, fchigh)
|
||||||
|
last_transition = idx
|
||||||
|
idx += 1
|
||||||
|
out: list[int] = []
|
||||||
|
start_idx = 0
|
||||||
|
pre_last = last = curr = 0
|
||||||
|
|
||||||
|
while idx < n - 20:
|
||||||
|
if thr[idx - 1] < thr[idx]: # 0 -> 1 transition
|
||||||
|
pre_last, last, curr = last, curr, idx - last_transition
|
||||||
|
if curr < fclow - 2:
|
||||||
|
pass # garbage noise
|
||||||
|
elif curr < fchigh - 1: # short wave -> 1
|
||||||
|
if len(out) > 1 and last > fchigh - 2 and pre_last < fchigh - 1:
|
||||||
|
out[-1] = 1 # fix a 9 surrounded by 8s
|
||||||
|
out.append(1)
|
||||||
|
if out and start_idx == 0:
|
||||||
|
start_idx = idx - fclow
|
||||||
|
elif curr > fchigh + 1 and len(out) < 3:
|
||||||
|
out = [] # leading garbage -> reset
|
||||||
|
elif curr == fclow + 1 and last == fclow - 1: # 7 then 9 -> two 8s
|
||||||
|
out.append(1)
|
||||||
|
if out and start_idx == 0:
|
||||||
|
start_idx = idx - fclow
|
||||||
|
else: # long wave -> 0
|
||||||
|
out.append(0)
|
||||||
|
if out and start_idx == 0:
|
||||||
|
start_idx = idx - fchigh
|
||||||
|
last_transition = idx
|
||||||
|
idx += 1
|
||||||
|
return out, start_idx
|
||||||
|
|
||||||
|
|
||||||
|
def aggregate_bits(bits, clk: int, invert: int, fchigh: int, fclow: int):
|
||||||
|
"""Port of ``aggregate_bits``: collapse runs of like waves into clock-resolution data bits
|
||||||
|
(a run of n waves at field-clock fc spans n*fc/clk bits)."""
|
||||||
|
if not bits:
|
||||||
|
return []
|
||||||
|
out: list[int] = []
|
||||||
|
lastval = bits[0]
|
||||||
|
n = 1
|
||||||
|
hclk = clk // 2
|
||||||
|
i = 1
|
||||||
|
for i in range(1, len(bits)):
|
||||||
|
n += 1
|
||||||
|
if bits[i] == lastval:
|
||||||
|
continue
|
||||||
|
if bits[i - 1] == 1:
|
||||||
|
n = (n * fclow + hclk) // clk
|
||||||
|
else:
|
||||||
|
n = (n * fchigh + hclk) // clk
|
||||||
|
if n == 0:
|
||||||
|
n = 1
|
||||||
|
out.extend([bits[i - 1] ^ invert] * n)
|
||||||
|
n = 0
|
||||||
|
lastval = bits[i]
|
||||||
|
if n > clk // fchigh and len(bits) >= 2: # trailing same-frequency run
|
||||||
|
if bits[i - 1] == 1:
|
||||||
|
n = (n * fclow + hclk) // clk
|
||||||
|
else:
|
||||||
|
n = (n * fchigh + hclk) // clk
|
||||||
|
out.extend([bits[-1] ^ invert] * n)
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def fskdemod(samples, rflen: int = 50, invert: int = 1, fchigh: int = 10, fclow: int = 8):
|
||||||
|
"""Full FSK demod: envelope -> decoded 1s/0s (``fsk_wave_demod`` + ``aggregate_bits``)."""
|
||||||
|
if is_noise(samples):
|
||||||
|
return []
|
||||||
|
waves, _ = fsk_wave_demod(list(samples), fchigh, fclow)
|
||||||
|
return aggregate_bits(waves, rflen, invert, fchigh, fclow)
|
||||||
|
|
||||||
|
|
||||||
|
def preamble_search(bits, preamble):
|
||||||
|
"""Port of ``preambleSearch``: find ``preamble`` in ``bits``. Returns ``(found, start_idx,
|
||||||
|
size)`` where ``size`` is the gap to the second occurrence (the frame length) when there is
|
||||||
|
one, else the remaining length."""
|
||||||
|
plen = len(preamble)
|
||||||
|
size = len(bits)
|
||||||
|
if size <= plen:
|
||||||
|
return (False, 0, size)
|
||||||
|
found = 0
|
||||||
|
start = 0
|
||||||
|
for idx in range(size - plen):
|
||||||
|
if bits[idx:idx + plen] == preamble:
|
||||||
|
found += 1
|
||||||
|
if found == 1:
|
||||||
|
start = idx
|
||||||
|
elif found == 2:
|
||||||
|
return (True, start, idx - start)
|
||||||
|
return (found > 0, start, size)
|
||||||
|
|
||||||
|
|
||||||
|
# HID: 00011101 start-of-frame, then Manchester (10 -> 1, 01 -> 0)
|
||||||
|
HID_PREAMBLE = [0, 0, 0, 1, 1, 1, 0, 1]
|
||||||
|
# AWID: 00000001 preamble, 96-bit frame
|
||||||
|
AWID_PREAMBLE = [0, 0, 0, 0, 0, 0, 0, 1]
|
||||||
|
|
||||||
|
_M32 = 0xFFFFFFFF
|
||||||
|
|
||||||
|
|
||||||
|
def hid_demod_fsk(samples):
|
||||||
|
"""Port of ``HIDdemodFSK``: FSK2a demod + preamble + Manchester -> ``(hi2, hi, lo, nbits)``
|
||||||
|
96-bit shift register plus the number of data bits Manchester-decoded (needed to tell the
|
||||||
|
header-less 37-bit format from the 26-bit one), or ``None``. Field extraction is the caller's."""
|
||||||
|
bits = fskdemod(samples, 50, 1, 10, 8)
|
||||||
|
if len(bits) < 96 * 2:
|
||||||
|
return None
|
||||||
|
found, start, size = preamble_search(bits, HID_PREAMBLE)
|
||||||
|
if not found:
|
||||||
|
return None
|
||||||
|
num_start = start + len(HID_PREAMBLE)
|
||||||
|
hi2 = hi = lo = 0
|
||||||
|
nbits = 0
|
||||||
|
idx = num_start
|
||||||
|
while (idx - num_start) < size - len(HID_PREAMBLE) and idx + 1 < len(bits):
|
||||||
|
a, b = bits[idx], bits[idx + 1]
|
||||||
|
if a == b:
|
||||||
|
break # not Manchester -> end of frame
|
||||||
|
hi2 = ((hi2 << 1) | (hi >> 31)) & _M32
|
||||||
|
hi = ((hi << 1) | (lo >> 31)) & _M32
|
||||||
|
lo = ((lo << 1) | (1 if (a and not b) else 0)) & _M32
|
||||||
|
nbits += 1
|
||||||
|
idx += 2
|
||||||
|
if hi == 0 and lo == 0:
|
||||||
|
return None
|
||||||
|
return (hi2, hi, lo, nbits)
|
||||||
|
|
||||||
|
|
||||||
|
def awid_demod_fsk(samples):
|
||||||
|
"""Port of ``detectAWID``: FSK2a demod + 00000001 preamble -> the 96 frame bits, or ``None``."""
|
||||||
|
bits = fskdemod(samples, 50, 1, 10, 8)
|
||||||
|
if len(bits) < 96:
|
||||||
|
return None
|
||||||
|
found, start, size = preamble_search(bits, AWID_PREAMBLE)
|
||||||
|
if not found or size != 96:
|
||||||
|
return None
|
||||||
|
return bits[start:start + 96]
|
||||||
@@ -8,11 +8,12 @@ exactly one place. A decoder returns a structured dict on a clean, validated fra
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from . import dsp
|
from . import dsp
|
||||||
|
from . import fsk
|
||||||
# Import the models through the ``pm3py.sim`` package (not their deep module paths): sim's
|
# Import the models through the ``pm3py.sim`` package (not their deep module paths): sim's
|
||||||
# __init__ re-exports them, and going through it forces sim to fully initialize first, avoiding
|
# __init__ re-exports them, and going through it forces sim to fully initialize first, avoiding
|
||||||
# a circular import (em4100 -> pm3py.sim.frame -> sim.__init__ -> em4100) when pm3py.lf is the
|
# a circular import (em4100 -> pm3py.sim.frame -> sim.__init__ -> em4100) when pm3py.lf is the
|
||||||
# first thing loaded.
|
# first thing loaded.
|
||||||
from pm3py.sim import EM4100Code, T5577Config
|
from pm3py.sim import EM4100Code, T5577Config, HIDProxTag
|
||||||
|
|
||||||
# config word -> preset name (invert T5577Config._PRESETS) for "emulating ..." labelling
|
# config word -> preset name (invert T5577Config._PRESETS) for "emulating ..." labelling
|
||||||
_PRESET_BY_WORD = {word: name for name, word in T5577Config._PRESETS.items()}
|
_PRESET_BY_WORD = {word: name for name, word in T5577Config._PRESETS.items()}
|
||||||
@@ -64,6 +65,44 @@ def _find_em4100(bits) -> EM4100Code | None:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _hid_field(full: int, fmt: str, nbits: int) -> dict | None:
|
||||||
|
"""Extract the low ``nbits`` as a ``fmt`` Wiegand credential, accepting only if re-encoding
|
||||||
|
the decoded facility+card reproduces the bits exactly (proves both parities)."""
|
||||||
|
wiegand = [(full >> (nbits - 1 - i)) & 1 for i in range(nbits)]
|
||||||
|
dec = HIDProxTag.decode_wiegand(wiegand, fmt)
|
||||||
|
fc, cn = dec["facility_code"], dec["card_number"]
|
||||||
|
if HIDProxTag._encode_wiegand(fc, cn, fmt) != wiegand:
|
||||||
|
return None
|
||||||
|
return {
|
||||||
|
"protocol": "HID Prox", "format": fmt, "facility_code": fc, "card_number": cn,
|
||||||
|
"wiegand": f"{full & ((1 << nbits) - 1):0{(nbits + 3) // 4}X}",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def decode_hid(samples) -> dict | None:
|
||||||
|
"""Demodulate an HID Prox envelope (FSK2a + Manchester) to its Wiegand credential.
|
||||||
|
|
||||||
|
Format is told by frame *structure*, not parity alone (2 parity bits can't separate 26- from
|
||||||
|
37-bit): the 26-bit H10301 wire frame carries a 0x20 header at bit 37 and a sentinel 1 at
|
||||||
|
bit 26 with the Wiegand below; the 37-bit H10304 frame is header-less, so a 37-bit Manchester
|
||||||
|
count with no 0x20 marks it. Each candidate is still parity-validated by re-encoding through
|
||||||
|
the shipped ``HIDProxTag``."""
|
||||||
|
r = fsk.hid_demod_fsk(samples)
|
||||||
|
if r is None:
|
||||||
|
return None
|
||||||
|
hi2, hi, lo, nbits = r
|
||||||
|
full = (hi << 32) | lo # low 64 bits cover 26- and 37-bit formats
|
||||||
|
if (hi & 0x20) and (lo & (1 << 26)): # 26-bit standard header + sentinel
|
||||||
|
got = _hid_field(full, "H10301", 26)
|
||||||
|
if got:
|
||||||
|
return got
|
||||||
|
if nbits == 37 and not (hi & 0x20): # header-less 37-bit frame
|
||||||
|
got = _hid_field(full, "H10304", 37)
|
||||||
|
if got:
|
||||||
|
return got
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def decode_t55xx_config(samples) -> dict | None:
|
def decode_t55xx_config(samples) -> dict | None:
|
||||||
"""Recover a T55xx block-0 config word from a block-0 read response (ASK/Manchester modes).
|
"""Recover a T55xx block-0 config word from a block-0 read response (ASK/Manchester modes).
|
||||||
|
|
||||||
|
|||||||
@@ -6,9 +6,10 @@ shipped model, a green test means the demod is the true inverse of what the tag
|
|||||||
"""
|
"""
|
||||||
from unittest.mock import MagicMock
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
from pm3py.lf import dsp, demod_samples, read_config, identify_lf
|
from pm3py.lf import dsp, fsk, demod_samples, read_config, identify_lf
|
||||||
from pm3py.lf import protocols
|
from pm3py.lf import protocols
|
||||||
from pm3py.transponders.lf.em.em4100 import EM4100Tag
|
from pm3py.transponders.lf.em.em4100 import EM4100Tag
|
||||||
|
from pm3py.transponders.lf.hid.hid import HIDProxTag
|
||||||
from pm3py.transponders.lf.atmel.t5577 import T5577Config, CONFIG_EM4100
|
from pm3py.transponders.lf.atmel.t5577 import T5577Config, CONFIG_EM4100
|
||||||
|
|
||||||
|
|
||||||
@@ -38,6 +39,40 @@ def _t55xx_config_env(word, rf=64, **kw):
|
|||||||
return _manchester_env(bits, rf=rf, **kw)
|
return _manchester_env(bits, rf=rf, **kw)
|
||||||
|
|
||||||
|
|
||||||
|
def _fsk_env(symbols, high=200, low=60, repeats=3):
|
||||||
|
"""Render FSK2a final-bit symbols (0 -> fc/8 wave, 1 -> fc/10 wave) as a square-wave envelope,
|
||||||
|
fcAll-style: (fc-halfFC) low then halfFC high per wave, ~clk/fc waves filling clk=50."""
|
||||||
|
env = []
|
||||||
|
for _ in range(repeats):
|
||||||
|
for s in symbols:
|
||||||
|
fc = 10 if s else 8
|
||||||
|
half = fc >> 1
|
||||||
|
for _w in range(round(50 / fc)):
|
||||||
|
env += [low] * (fc - half) + [high] * half
|
||||||
|
return bytes(env)
|
||||||
|
|
||||||
|
|
||||||
|
def _bits(value, n):
|
||||||
|
return [(value >> (n - 1 - i)) & 1 for i in range(n)]
|
||||||
|
|
||||||
|
|
||||||
|
def _hid_env(fc, cn, fmt="H10301", repeats=5, **kw):
|
||||||
|
"""A faithful HID Prox envelope (matches firmware add_HID_preamble): SOF + Manchester(frame),
|
||||||
|
FSK2a modulated. 26-bit carries a 0x20 header (bit 37) + sentinel (bit 26); 37-bit is bare."""
|
||||||
|
w_int = 0
|
||||||
|
for b in HIDProxTag._encode_wiegand(fc, cn, fmt):
|
||||||
|
w_int = (w_int << 1) | b
|
||||||
|
if fmt == "H10301": # 26-bit: header + sentinel, 38-bit frame
|
||||||
|
full = (0x20 << 32) | (1 << 26) | w_int
|
||||||
|
frame = _bits(full, 38)
|
||||||
|
else: # 37-bit: header-less
|
||||||
|
frame = _bits(w_int, 37)
|
||||||
|
manch = []
|
||||||
|
for d in frame:
|
||||||
|
manch += [1, 0] if d else [0, 1] # data 1 -> fc10,fc8 ; 0 -> fc8,fc10
|
||||||
|
return _fsk_env(fsk.HID_PREAMBLE + manch, repeats=repeats, **kw)
|
||||||
|
|
||||||
|
|
||||||
# --- DSP primitives ---------------------------------------------------------------------------
|
# --- DSP primitives ---------------------------------------------------------------------------
|
||||||
|
|
||||||
class TestDSP:
|
class TestDSP:
|
||||||
@@ -90,6 +125,45 @@ class TestEM4100:
|
|||||||
assert r["card_number"] == 0x01020304
|
assert r["card_number"] == 0x01020304
|
||||||
|
|
||||||
|
|
||||||
|
# --- HID Prox (FSK) ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestFSK:
|
||||||
|
def test_fskdemod_recovers_preamble(self):
|
||||||
|
# a bare SOF + a few symbols demodulates to the HID preamble pattern
|
||||||
|
bits = fsk.fskdemod(_fsk_env(fsk.HID_PREAMBLE + [0, 1] * 20))
|
||||||
|
found, start, _ = fsk.preamble_search(bits, fsk.HID_PREAMBLE)
|
||||||
|
assert found
|
||||||
|
|
||||||
|
def test_noise_rejected(self):
|
||||||
|
assert fsk.is_noise(bytes([128] * 5000))
|
||||||
|
assert fsk.fskdemod(bytes([128] * 5000)) == []
|
||||||
|
|
||||||
|
|
||||||
|
class TestHID:
|
||||||
|
def test_h10301_roundtrip(self):
|
||||||
|
r = protocols.decode_hid(_hid_env(123, 45678))
|
||||||
|
assert r and r["protocol"] == "HID Prox" and r["format"] == "H10301"
|
||||||
|
assert r["facility_code"] == 123 and r["card_number"] == 45678
|
||||||
|
|
||||||
|
def test_various_h10301(self):
|
||||||
|
for fc, cn in ((0, 1), (255, 65535), (77, 12345)):
|
||||||
|
r = protocols.decode_hid(_hid_env(fc, cn))
|
||||||
|
assert r and r["facility_code"] == fc and r["card_number"] == cn, f"{fc}/{cn}"
|
||||||
|
|
||||||
|
def test_h10304_37bit(self):
|
||||||
|
r = protocols.decode_hid(_hid_env(4321, 123456, fmt="H10304")) # CN < 2^19
|
||||||
|
assert r and r["format"] == "H10304"
|
||||||
|
assert r["facility_code"] == 4321 and r["card_number"] == 123456
|
||||||
|
|
||||||
|
def test_not_hid_returns_none(self):
|
||||||
|
assert protocols.decode_hid(_em4100_env(0x0102030405)) is None
|
||||||
|
assert protocols.decode_hid(bytes([128] * 6000)) is None
|
||||||
|
|
||||||
|
def test_demod_samples_picks_hid(self):
|
||||||
|
r = demod_samples(_hid_env(200, 9999))
|
||||||
|
assert r["protocol"] == "HID Prox" and r["card_number"] == 9999
|
||||||
|
|
||||||
|
|
||||||
# --- T55xx config -----------------------------------------------------------------------------
|
# --- T55xx config -----------------------------------------------------------------------------
|
||||||
|
|
||||||
class TestT55xxConfig:
|
class TestT55xxConfig:
|
||||||
|
|||||||
Reference in New Issue
Block a user