fix(rawcli): render ANSI, stop mangling commands, fuller Type 2 catalog
Three fixes from live hardware use:
- Colors/ANSI now render. Output routes through prompt_toolkit's
print_formatted_text(ANSI(...)) instead of print(), which showed the
escape codes literally under patch_stdout. Formatters force is_tty so
the printer decides color-vs-plain. dispatch()/handlers take an `out`.
- Auto-byte-spacing no longer mangles non-hex input. byte_space() only
regroups a *pure* hex/binary run; command words ("help transponder",
"identify", "close") and 0x/0b-prefixed tokens are left intact (a-f in
words used to trigger regrouping -> failed hex parse).
- Type 2 catalog gains PWD_AUTH (0x1B), COMPAT_WRITE (0xA0), HALT — the
NTAG213 set was missing password auth.
- identify trims the firmware buffer padding off the GET_VERSION response
(was dumping ~40 trailing 00 bytes).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -58,103 +58,105 @@ def _raw_exchange(device, payload: bytes, protocol: str = "hf14a") -> bytes | No
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def dispatch(state: RawSession, line: str) -> bool:
|
def dispatch(state: RawSession, line: str, out=print) -> bool:
|
||||||
"""Handle one input line. Returns False to exit the REPL, True to keep going. Testable
|
"""Handle one input line. ``out`` prints a (possibly ANSI-colored) string. Returns False to
|
||||||
without a live prompt."""
|
exit the REPL, True to keep going. Testable without a live prompt (default ``out=print``)."""
|
||||||
try:
|
try:
|
||||||
cmd = parse_line(line, state.entry_mode)
|
cmd = parse_line(line, state.entry_mode)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
print(f"rawcli: {exc}")
|
out(f"rawcli: {exc}")
|
||||||
return True
|
return True
|
||||||
|
|
||||||
if cmd.kind == "control":
|
if cmd.kind == "control":
|
||||||
if cmd.name in ("quit", "exit", "q"):
|
if cmd.name in ("quit", "exit", "q"):
|
||||||
return False
|
return False
|
||||||
if cmd.name == "help":
|
if cmd.name == "help":
|
||||||
_handle_help(state, cmd.args)
|
_handle_help(state, cmd.args, out)
|
||||||
elif cmd.name == "identify":
|
elif cmd.name == "identify":
|
||||||
result = identify(state, emit=print) # stream the probe exchanges to the trace
|
result = identify(state, emit=out) # stream the probe exchanges to the trace
|
||||||
print(format_summary(result))
|
out(format_summary(result))
|
||||||
elif cmd.name == "transponder":
|
elif cmd.name == "transponder":
|
||||||
print(state.transponder or "no transponder identified — run 'identify'")
|
out(state.transponder or "no transponder identified — run 'identify'")
|
||||||
elif cmd.name == "close":
|
elif cmd.name == "close":
|
||||||
clear(state)
|
clear(state)
|
||||||
print("connection closed")
|
out("connection closed")
|
||||||
elif cmd.name == "tlv":
|
elif cmd.name == "tlv":
|
||||||
_handle_tlv(state, cmd.args)
|
_handle_tlv(state, cmd.args, out)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
if cmd.kind == "call":
|
if cmd.kind == "call":
|
||||||
_handle_call(state, cmd)
|
_handle_call(state, cmd, out)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
# raw payload
|
|
||||||
if not cmd.payload:
|
if not cmd.payload:
|
||||||
return True
|
return True
|
||||||
response = _raw_exchange(state.device, cmd.payload, protocol=state.protocol)
|
response = _raw_exchange(state.device, cmd.payload, protocol=state.protocol)
|
||||||
print(render_exchange(cmd.payload, response, protocol=state.protocol,
|
out(render_exchange(cmd.payload, response, protocol=state.protocol, is_tty=True))
|
||||||
is_tty=sys.stdout.isatty()))
|
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
def _handle_call(state: RawSession, cmd) -> None:
|
def _handle_call(state: RawSession, cmd, out=print) -> None:
|
||||||
catalog = catalog_for(state)
|
catalog = catalog_for(state)
|
||||||
if catalog is None:
|
if catalog is None:
|
||||||
print("no transponder identified — run 'identify' first")
|
out("no transponder identified — run 'identify' first")
|
||||||
return
|
return
|
||||||
tc = catalog.get(cmd.name)
|
tc = catalog.get(cmd.name)
|
||||||
if tc is None:
|
if tc is None:
|
||||||
print(f"unknown command {cmd.name!r} for {catalog.name}; try 'help'")
|
out(f"unknown command {cmd.name!r} for {catalog.name}; try 'help'")
|
||||||
return
|
return
|
||||||
try:
|
try:
|
||||||
payload = tc.build(*cmd.args)
|
payload = tc.build(*cmd.args)
|
||||||
except (TypeError, ValueError) as exc:
|
except (TypeError, ValueError) as exc:
|
||||||
print(f"usage: {tc.usage()} ({exc})")
|
out(f"usage: {tc.usage()} ({exc})")
|
||||||
return
|
return
|
||||||
response = _raw_exchange(state.device, payload, protocol=catalog.protocol)
|
response = _raw_exchange(state.device, payload, protocol=catalog.protocol)
|
||||||
print(render_exchange(payload, response, protocol=catalog.protocol,
|
out(render_exchange(payload, response, protocol=catalog.protocol, is_tty=True))
|
||||||
is_tty=sys.stdout.isatty()))
|
|
||||||
|
|
||||||
|
|
||||||
def _handle_tlv(state: RawSession, args) -> None:
|
def _handle_tlv(state: RawSession, args, out=print) -> None:
|
||||||
from .tlv import format_tlv, prompt_tlv
|
from .tlv import format_tlv, prompt_tlv
|
||||||
from .parser import parse_bytes
|
from .parser import parse_bytes
|
||||||
if args:
|
if args:
|
||||||
try:
|
try:
|
||||||
data = parse_bytes(" ".join(args), state.entry_mode)
|
data = parse_bytes(" ".join(args), state.entry_mode)
|
||||||
except ValueError as exc:
|
except ValueError as exc:
|
||||||
print(f"rawcli: {exc}")
|
out(f"rawcli: {exc}")
|
||||||
return
|
return
|
||||||
else:
|
else:
|
||||||
data = prompt_tlv(state) # interactive multi-line editor
|
data = prompt_tlv(state) # interactive multi-line editor
|
||||||
if not data:
|
if not data:
|
||||||
return
|
return
|
||||||
print(format_tlv(data))
|
out(format_tlv(data))
|
||||||
|
|
||||||
|
|
||||||
def _handle_help(state: RawSession, args) -> None:
|
def _handle_help(state: RawSession, args, out=print) -> None:
|
||||||
catalog = catalog_for(state)
|
catalog = catalog_for(state)
|
||||||
if args and catalog is not None:
|
if args and catalog is not None:
|
||||||
tc = catalog.get(args[0])
|
tc = catalog.get(args[0])
|
||||||
if tc is not None:
|
if tc is not None:
|
||||||
print(f"{tc.usage()} — {tc.help}")
|
out(f"{tc.usage()} — {tc.help}")
|
||||||
return
|
return
|
||||||
print(_HELP)
|
lines = [_HELP]
|
||||||
if catalog is not None:
|
if catalog is not None:
|
||||||
print(f"\n{catalog.name} commands:")
|
lines.append(f"\n{catalog.name} commands:")
|
||||||
for name in catalog.names():
|
lines += [f" {catalog.get(n).usage():<28}{catalog.get(n).help}" for n in catalog.names()]
|
||||||
tc = catalog.get(name)
|
out("\n".join(lines))
|
||||||
print(f" {tc.usage():<28}{tc.help}")
|
|
||||||
|
|
||||||
|
|
||||||
def run(port=None) -> int:
|
def run(port=None) -> int:
|
||||||
from prompt_toolkit import PromptSession
|
from prompt_toolkit import PromptSession, print_formatted_text
|
||||||
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
|
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
|
||||||
|
from prompt_toolkit.formatted_text import ANSI
|
||||||
from prompt_toolkit.key_binding import KeyBindings
|
from prompt_toolkit.key_binding import KeyBindings
|
||||||
from prompt_toolkit.patch_stdout import patch_stdout
|
from prompt_toolkit.patch_stdout import patch_stdout
|
||||||
|
|
||||||
from .completer import RawCompleter
|
from .completer import RawCompleter
|
||||||
|
|
||||||
|
def out(text):
|
||||||
|
# parse the ANSI our formatters emit and render it through prompt_toolkit (renders
|
||||||
|
# color on a terminal, strips it when piped) — plain print() shows the codes literally
|
||||||
|
print_formatted_text(ANSI(str(text)))
|
||||||
|
|
||||||
state = RawSession(device=_connect(port))
|
state = RawSession(device=_connect(port))
|
||||||
kb = KeyBindings()
|
kb = KeyBindings()
|
||||||
install_key_bindings(kb, state)
|
install_key_bindings(kb, state)
|
||||||
@@ -176,7 +178,7 @@ def run(port=None) -> int:
|
|||||||
break
|
break
|
||||||
if not line.strip():
|
if not line.strip():
|
||||||
continue
|
continue
|
||||||
if not dispatch(state, line):
|
if not dispatch(state, line, out):
|
||||||
break
|
break
|
||||||
print("bye")
|
out("bye")
|
||||||
return 0
|
return 0
|
||||||
|
|||||||
@@ -62,9 +62,15 @@ TYPE2 = _catalog(
|
|||||||
_c("READ_SIG", [], lambda: bytes([0x3C, 0x00]), "read the ECC originality signature (0x3C)"),
|
_c("READ_SIG", [], lambda: bytes([0x3C, 0x00]), "read the ECC originality signature (0x3C)"),
|
||||||
_c("READ_CNT", ["counter"], lambda counter="0": bytes([0x39, _int(counter)]),
|
_c("READ_CNT", ["counter"], lambda counter="0": bytes([0x39, _int(counter)]),
|
||||||
"read the 24-bit NFC counter (0x39)"),
|
"read the 24-bit NFC counter (0x39)"),
|
||||||
|
_c("PWD_AUTH", ["password"],
|
||||||
|
lambda password: bytes([0x1B]) + _hex(password).ljust(4, b"\x00")[:4],
|
||||||
|
"authenticate with the 4-byte password → PACK (0x1B)"),
|
||||||
_c("WRITE", ["page", "data"],
|
_c("WRITE", ["page", "data"],
|
||||||
lambda page, data: bytes([0xA2, _int(page)]) + _hex(data).ljust(4, b"\x00")[:4],
|
lambda page, data: bytes([0xA2, _int(page)]) + _hex(data).ljust(4, b"\x00")[:4],
|
||||||
"write 4 bytes <data> to <page> (0xA2)"),
|
"write 4 bytes <data> to <page> (0xA2)"),
|
||||||
|
_c("COMPAT_WRITE", ["page"], lambda page: bytes([0xA0, _int(page)]),
|
||||||
|
"compatibility write, two-phase (0xA0)"),
|
||||||
|
_c("HALT", [], lambda: bytes([0x50, 0x00]), "halt the tag (0x50 00)"),
|
||||||
)
|
)
|
||||||
|
|
||||||
# ---- MIFARE Classic (block ops; auth is the reader's job before these) ----
|
# ---- MIFARE Classic (block ops; auth is the reader's job before these) ----
|
||||||
|
|||||||
@@ -17,14 +17,18 @@ def group_size(mode: str) -> int:
|
|||||||
|
|
||||||
|
|
||||||
def byte_space(text: str, mode: str = HEX) -> str:
|
def byte_space(text: str, mode: str = HEX) -> str:
|
||||||
"""Re-group a run of hex/binary digits into space-separated byte groups.
|
"""Re-group a **pure** run of hex/binary digits into space-separated byte groups.
|
||||||
|
|
||||||
Leaves a token that carries an explicit ``0x``/``0b`` prefix untouched (so intermixed input
|
Anything that isn't purely digits of the current mode is returned untouched — command words
|
||||||
isn't mangled); only a bare digit run is regrouped."""
|
("help", "transponder"), function calls, and explicit ``0x``/``0b`` prefixed tokens are never
|
||||||
if "x" in text.lower() or "b" in text.lower():
|
regrouped. This is what keeps auto-spacing from mangling non-hex input."""
|
||||||
return text
|
low = text.strip().lower()
|
||||||
n = group_size(mode)
|
if low.startswith("0x") or low.startswith("0b"):
|
||||||
|
return text # explicit prefix — leave intact
|
||||||
compact = text.replace(" ", "")
|
compact = text.replace(" ", "")
|
||||||
|
if not compact or any(ch not in _DIGITS[mode] for ch in compact):
|
||||||
|
return text # not a pure numeric run (e.g. a command)
|
||||||
|
n = group_size(mode)
|
||||||
return " ".join(compact[i:i + n] for i in range(0, len(compact), n))
|
return " ".join(compact[i:i + n] for i in range(0, len(compact), n))
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -147,7 +147,8 @@ def identify(session, emit=None) -> dict:
|
|||||||
scan = _try(lambda: dev.hf.iso14a.scan(), {"found": False})
|
scan = _try(lambda: dev.hf.iso14a.scan(), {"found": False})
|
||||||
if scan.get("found"):
|
if scan.get("found"):
|
||||||
session.field, session.protocol = "hf", "hf14a"
|
session.field, session.protocol = "hf", "hf14a"
|
||||||
fmt = TraceFormatter(mode="reader", decoder=_ident_decode, crc_len=0)
|
# always emit ANSI; the app's printer renders it (and strips it off a non-terminal)
|
||||||
|
fmt = TraceFormatter(mode="reader", decoder=_ident_decode, crc_len=0, is_tty=True)
|
||||||
for direction, data in _activation_frames(scan):
|
for direction, data in _activation_frames(scan):
|
||||||
emit(fmt.format(direction, data).lstrip("\n"))
|
emit(fmt.format(direction, data).lstrip("\n"))
|
||||||
model, version = None, None
|
model, version = None, None
|
||||||
@@ -184,8 +185,9 @@ def _probe_version(dev, fmt, emit) -> tuple[str | None, bytes | None]:
|
|||||||
raw = resp.get("raw") if isinstance(resp, dict) else None
|
raw = resp.get("raw") if isinstance(resp, dict) else None
|
||||||
if not raw:
|
if not raw:
|
||||||
return None, None
|
return None, None
|
||||||
emit(fmt.format(1, bytes(raw)).lstrip("\n"))
|
|
||||||
version = bytes(raw[:8])
|
version = bytes(raw[:8])
|
||||||
|
display = bytes(raw).rstrip(b"\x00") or version # drop the firmware buffer padding
|
||||||
|
emit(fmt.format(1, display).lstrip("\n"))
|
||||||
model = decode_version(version)
|
model = decode_version(version)
|
||||||
if model:
|
if model:
|
||||||
emit(f" → {model}")
|
emit(f" → {model}")
|
||||||
|
|||||||
@@ -127,6 +127,17 @@ class TestByteSpace:
|
|||||||
|
|
||||||
def test_prefixed_left_alone(self):
|
def test_prefixed_left_alone(self):
|
||||||
assert byte_space("0x3004", "hex") == "0x3004" # explicit prefix not mangled
|
assert byte_space("0x3004", "hex") == "0x3004" # explicit prefix not mangled
|
||||||
|
assert byte_space("0b0011", "hex") == "0b0011"
|
||||||
|
|
||||||
|
def test_leaves_command_words_alone(self):
|
||||||
|
# regression: auto-spacing must not mangle non-hex input (a-f in words used to break it)
|
||||||
|
assert byte_space("help transponder", "hex") == "help transponder"
|
||||||
|
assert byte_space("identify", "hex") == "identify"
|
||||||
|
assert byte_space("close", "hex") == "close" # c,e are hex digits but l,o,s aren't
|
||||||
|
assert byte_space("READ(4)", "hex") == "READ(4)"
|
||||||
|
|
||||||
|
def test_spaces_pure_hex(self):
|
||||||
|
assert byte_space("deadbeef", "hex") == "de ad be ef"
|
||||||
|
|
||||||
|
|
||||||
class TestDispatch:
|
class TestDispatch:
|
||||||
@@ -253,6 +264,7 @@ class TestCatalog:
|
|||||||
assert TYPE2.get("FAST_READ").build("0", "3") == b"\x3A\x00\x03"
|
assert TYPE2.get("FAST_READ").build("0", "3") == b"\x3A\x00\x03"
|
||||||
assert TYPE2.get("GET_VERSION").build() == b"\x60"
|
assert TYPE2.get("GET_VERSION").build() == b"\x60"
|
||||||
assert TYPE2.get("WRITE").build("4", "01020304") == b"\xA2\x04\x01\x02\x03\x04"
|
assert TYPE2.get("WRITE").build("4", "01020304") == b"\xA2\x04\x01\x02\x03\x04"
|
||||||
|
assert TYPE2.get("PWD_AUTH").build("FFFFFFFF") == b"\x1B\xFF\xFF\xFF\xFF"
|
||||||
|
|
||||||
def test_iso15_builds(self):
|
def test_iso15_builds(self):
|
||||||
assert ISO15.get("READ_BLOCK").build("4") == b"\x02\x20\x04"
|
assert ISO15.get("READ_BLOCK").build("4") == b"\x02\x20\x04"
|
||||||
|
|||||||
Reference in New Issue
Block a user