feat(rawcli): identify + persistent connection (Phase 3)

- identify.py: probe HF (14a -> 15693) then LF; set session field /
  protocol / transponder. The 14a scan uses NO_DISCONNECT so the tag
  stays selected (the persistent connection). SAK -> coarse family name
  for the breadcrumb.
- session: add `protocol` (drives raw-exchange routing)
- app: connect via Proxmark3.sync() (scan/raw become plain calls); wire
  identify / transponder / close control commands; protocol-aware
  _raw_exchange; breadcrumb fills in field + transponder after identify

8 new tests (identify 14a/15693/none, clear, control commands). Device
probes are mocked; live identify is the user's hardware test.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
michael
2026-07-14 12:10:51 -07:00
parent 54e52f029f
commit e024294ec6
4 changed files with 174 additions and 11 deletions

View File

@@ -12,35 +12,45 @@ import sys
from .session import RawSession
from .parser import parse_line
from .entry import install_key_bindings
from .identify import identify, clear, format_summary
from .trace_view import render_exchange
_HELP = """\
rawcli — raw command interface
help show this help
identify probe the field; sets HF/LF + transponder, keeps the connection
transponder show the identified transponder
close drop the tag connection
quit | exit | q leave rawcli
<hex ...> send raw bytes to the tag; the annotated exchange prints above
Type hex as "30 04" or "3004". Entry-mode toggle (0x/0b), identify, and function-style
transponder commands (READ(4), GET_DATA) land in later phases."""
Type hex as "30 04" or "3004"; Ctrl-/ toggles hex/binary entry (0x/0b). Function-style
transponder commands (READ(4), GET_DATA) land in the next phase."""
def _connect(port):
"""Best-effort device connection; returns the client or None (offline)."""
try:
from pm3py import Proxmark3
return Proxmark3(port)
return Proxmark3.sync(port) # sync proxy: scan()/raw() are plain calls
except Exception as exc: # no device / connect failure — run offline
print(f"rawcli: no device ({type(exc).__name__}: {exc}); running offline", file=sys.stderr)
return None
def _raw_exchange(device, payload: bytes) -> bytes | None:
"""Issue one raw 14a exchange, returning the response bytes (or None)."""
def _raw_exchange(device, payload: bytes, protocol: str = "hf14a") -> bytes | None:
"""Issue one raw exchange over the identified protocol, returning the response bytes."""
if device is None:
return None
try:
if protocol == "hf14a":
resp = device.hf.iso14a.raw(payload)
elif protocol == "hf15":
raw = getattr(device.hf.iso15, "raw", None)
resp = raw(payload) if raw else None
else:
return None # lf raw exchange lands in a later phase
return resp.get("data") if isinstance(resp, dict) else None
except Exception as exc:
print(f"rawcli: exchange failed ({type(exc).__name__}: {exc})", file=sys.stderr)
@@ -61,9 +71,13 @@ def dispatch(state: RawSession, line: str) -> bool:
return False
if cmd.name == "help":
print(_HELP)
else:
# identify / transponder / close land in Phase 3
print(f"rawcli: '{cmd.name}' is not wired yet (coming in a later phase)")
elif cmd.name == "identify":
print(format_summary(identify(state)))
elif cmd.name == "transponder":
print(state.transponder or "no transponder identified — run 'identify'")
elif cmd.name == "close":
clear(state)
print("connection closed")
return True
if cmd.kind == "call":
@@ -74,8 +88,8 @@ def dispatch(state: RawSession, line: str) -> bool:
# raw payload
if not cmd.payload:
return True
response = _raw_exchange(state.device, cmd.payload)
print(render_exchange(cmd.payload, response, protocol="hf14a",
response = _raw_exchange(state.device, cmd.payload, protocol=state.protocol)
print(render_exchange(cmd.payload, response, protocol=state.protocol,
is_tty=sys.stdout.isatty()))
return True

View File

@@ -0,0 +1,86 @@
"""Tag identification for rawcli.
``identify`` probes HF (ISO 14443-A, then 15693) and then LF, records the RF field, the wire
protocol, and a display name on the session, and — because the 14a scan uses NO_DISCONNECT — the
tag stays selected (the persistent connection). Device methods are called synchronously (the app
connects via ``Proxmark3.sync()``).
"""
from __future__ import annotations
from typing import Any
# Coarse SAK → 14a family, for the breadcrumb name (GET_VERSION-level precision is a later add).
_SAK_NAMES = {
0x00: "MIFARE UL/NTAG",
0x08: "MIFARE Classic 1K",
0x09: "MIFARE Mini",
0x18: "MIFARE Classic 4K",
0x10: "MIFARE Plus 2K",
0x11: "MIFARE Plus 4K",
0x20: "ISO14443-4",
0x28: "JCOP",
}
def name_14a(scan: dict) -> str:
sak = scan.get("sak")
if isinstance(sak, int):
return _SAK_NAMES.get(sak, f"ISO14443-A (SAK {sak:02X})")
return "ISO14443-A"
def _try(fn, default):
try:
return fn()
except Exception:
return default
def identify(session) -> dict:
"""Probe the field and update ``session`` (field / protocol / transponder). Returns a
summary dict (``found`` + details)."""
dev = session.device
if dev is None:
return {"found": False, "error": "no device connected"}
scan = _try(lambda: dev.hf.iso14a.scan(), {"found": False})
if scan.get("found"):
session.field, session.protocol = "hf", "hf14a"
session.transponder = name_14a(scan)
return {"found": True, "field": "hf", "protocol": "14a",
"transponder": session.transponder, "uid": scan.get("uid"), "raw": scan}
scan15 = _try(lambda: dev.hf.iso15.scan(), {"found": False})
if scan15.get("found") or scan15.get("uid"):
session.field, session.protocol = "hf", "hf15"
session.transponder = "ISO15693"
return {"found": True, "field": "hf", "protocol": "15693",
"transponder": "ISO15693", "uid": scan15.get("uid"), "raw": scan15}
lf = _try(lambda: dev.lf.search(), None)
if lf:
session.field, session.protocol = "lf", "lf"
name = getattr(lf, "tag_type", None) or getattr(lf, "type", None) or "LF tag"
session.transponder = str(name)
return {"found": True, "field": "lf", "protocol": "lf",
"transponder": session.transponder, "raw": lf}
return {"found": False}
def clear(session) -> None:
"""Forget the identified tag (drop the logical connection); the device stays open."""
session.field = None
session.transponder = None
session.protocol = "hf14a"
dev = session.device
if dev is not None:
_try(lambda: dev.hf.dropfield(), None)
def format_summary(result: dict) -> str:
if not result.get("found"):
return f"no tag found{': ' + result['error'] if result.get('error') else ''}"
uid = result.get("uid")
uid_hex = uid.hex().upper() if isinstance(uid, (bytes, bytearray)) else (uid or "?")
return f"{result['transponder']} ({result['field'].upper()} / {result['protocol']}) UID {uid_hex}"

View File

@@ -21,6 +21,7 @@ class RawSession:
self.device = device
self.field: str | None = None
self.transponder: str | None = None
self.protocol: str = "hf14a" # wire protocol for raw exchanges (set by identify)
self.entry_mode: str = HEX
self.connected: bool = device is not None

View File

@@ -1,6 +1,8 @@
"""rawcli Phase 1 — session state, breadcrumb, trace rendering, CLI dispatch. Hardware-free."""
import re
from unittest.mock import MagicMock
import pytest
from pm3py.cli.main import build_parser, main
@@ -8,6 +10,7 @@ from pm3py.cli.rawcli.session import RawSession
from pm3py.cli.rawcli.trace_view import render_exchange
from pm3py.cli.rawcli.parser import parse_token, parse_bytes, parse_line
from pm3py.cli.rawcli.entry import byte_space
from pm3py.cli.rawcli.identify import identify, clear, name_14a
from pm3py.cli.rawcli.app import dispatch
_ANSI = re.compile(r"\x1b\[[0-9;]*m")
@@ -138,3 +141,62 @@ class TestDispatch:
def test_bad_hex_reports(self, capsys):
assert dispatch(RawSession(), "0b0011") is True
assert "whole number of bytes" in capsys.readouterr().out
def _mock_device(scan14=None, scan15=None):
dev = MagicMock()
dev.hf.iso14a.scan.return_value = scan14 or {"found": False}
dev.hf.iso15.scan.return_value = scan15 or {"found": False}
dev.lf.search.return_value = None
return dev
class TestIdentify:
def test_no_device(self):
s = RawSession()
r = identify(s)
assert r["found"] is False and s.field is None
def test_14a_sets_field_and_name(self):
s = RawSession(device=_mock_device(
scan14={"found": True, "sak": 0x08, "uid": b"\x01\x02\x03\x04"}))
r = identify(s)
assert r["found"] and s.field == "hf" and s.protocol == "hf14a"
assert s.transponder == "MIFARE Classic 1K"
assert r["uid"] == b"\x01\x02\x03\x04"
def test_15693_fallback(self):
s = RawSession(device=_mock_device(scan15={"found": True, "uid": b"\xE0\x04"}))
r = identify(s)
assert r["found"] and s.field == "hf" and s.protocol == "hf15"
assert s.transponder == "ISO15693"
def test_name_unknown_sak(self):
assert name_14a({"sak": 0x99}) == "ISO14443-A (SAK 99)"
def test_clear_forgets_tag(self):
s = RawSession(device=_mock_device(scan14={"found": True, "sak": 0x08}))
identify(s)
clear(s)
assert s.field is None and s.transponder is None and s.protocol == "hf14a"
class TestControlCommands:
def test_identify_command(self, capsys):
s = RawSession(device=_mock_device(
scan14={"found": True, "sak": 0x00, "uid": b"\x04\x11\x22\x33"}))
dispatch(s, "identify")
out = capsys.readouterr().out
assert "MIFARE UL/NTAG" in out and "04112233" in out
assert s.breadcrumb() == "[raw / hf / MIFARE UL/NTAG / 0x]"
def test_transponder_when_none(self, capsys):
dispatch(RawSession(), "transponder")
assert "run 'identify'" in capsys.readouterr().out
def test_close(self, capsys):
s = RawSession(device=_mock_device(scan14={"found": True, "sak": 0x08}))
dispatch(s, "identify")
dispatch(s, "close")
assert "closed" in capsys.readouterr().out
assert s.transponder is None