pm3py/pm3py/cli/rawcli/app.py

"""rawcli interactive loop.

Phase 1 skeleton: connect (best-effort), a live breadcrumb prompt, ``help``/``quit``, and a raw
hex payload → annotated exchange. Entry-mode toggling, identify/connection, the command catalog,
and TLV editing arrive in later phases (see the plan). The interactive loop is intentionally thin;
the testable logic lives in :mod:`session` / :mod:`trace_view` / (later) ``parser``.
"""
from __future__ import annotations

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 .catalog import catalog_for
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"; 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.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, 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
        if not isinstance(resp, dict):
            return None
        data = resp.get("raw")               # bytes; "data" is the hex string form
        if data is None:
            hexstr = resp.get("data")
            data = bytes.fromhex(hexstr) if isinstance(hexstr, str) else hexstr
        return data
    except Exception as exc:
        print(f"rawcli: exchange failed ({type(exc).__name__}: {exc})", file=sys.stderr)
        return None


def dispatch(state: RawSession, line: str, out=print) -> bool:
    """Handle one input line. ``out`` prints a (possibly ANSI-colored) string. Returns False to
    exit the REPL, True to keep going. Testable without a live prompt (default ``out=print``)."""
    try:
        cmd = parse_line(line, state.entry_mode)
    except ValueError as exc:
        out(f"rawcli: {exc}")
        return True

    if cmd.kind == "control":
        if cmd.name in ("quit", "exit", "q"):
            return False
        if cmd.name == "help":
            _handle_help(state, cmd.args, out)
        elif cmd.name == "identify":
            result = identify(state, emit=out)       # stream the probe exchanges to the trace
            out(format_summary(result))
        elif cmd.name == "transponder":
            out(state.transponder or "no transponder identified — run 'identify'")
        elif cmd.name == "close":
            clear(state)
            out("connection closed")
        elif cmd.name == "tlv":
            _handle_tlv(state, cmd.args, out)
        return True

    if cmd.kind == "call":
        _handle_call(state, cmd, out)
        return True

    if not cmd.payload:
        return True
    response = _raw_exchange(state.device, cmd.payload, protocol=state.protocol)
    out(render_exchange(cmd.payload, response, protocol=state.protocol, is_tty=True))
    return True


def _handle_call(state: RawSession, cmd, out=print) -> None:
    catalog = catalog_for(state)
    if catalog is None:
        out("no transponder identified — run 'identify' first")
        return
    tc = catalog.get(cmd.name)
    if tc is None:
        out(f"unknown command {cmd.name!r} for {catalog.name}; try 'help'")
        return
    try:
        payload = tc.build(*cmd.args)
    except (TypeError, ValueError) as exc:
        out(f"usage: {tc.usage()}  ({exc})")
        return
    response = _raw_exchange(state.device, payload, protocol=catalog.protocol)
    out(render_exchange(payload, response, protocol=catalog.protocol, is_tty=True))


def _handle_tlv(state: RawSession, args, out=print) -> None:
    from .tlv import format_tlv, prompt_tlv
    from .parser import parse_bytes
    if args:
        try:
            data = parse_bytes(" ".join(args), state.entry_mode)
        except ValueError as exc:
            out(f"rawcli: {exc}")
            return
    else:
        data = prompt_tlv(state)          # interactive multi-line editor
        if not data:
            return
    out(format_tlv(data))


def _handle_help(state: RawSession, args, out=print) -> None:
    catalog = catalog_for(state)
    if args and catalog is not None:
        tc = catalog.get(args[0])
        if tc is not None:
            out(f"{tc.usage()}  —  {tc.help}")
            return
    lines = [_HELP]
    if catalog is not None:
        lines.append(f"\n{catalog.name} commands:")
        lines += [f"  {catalog.get(n).usage():<28}{catalog.get(n).help}" for n in catalog.names()]
    out("\n".join(lines))


def run(port=None) -> int:
    from prompt_toolkit import PromptSession, print_formatted_text
    from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
    from prompt_toolkit.formatted_text import ANSI
    from prompt_toolkit.key_binding import KeyBindings
    from prompt_toolkit.patch_stdout import patch_stdout

    from .completer import RawCompleter

    from prompt_toolkit.application import get_app
    from .memory import input_hint

    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))

    def bottom_toolbar():
        # live relevance for the current input: what the command does, and where a page lives
        try:
            text = get_app().current_buffer.text
        except Exception:
            text = ""
        hint = input_hint(state, text)
        return hint if hint else f"entry {state.entry_prefix}  ·  Ctrl-/ hex/binary  ·  'help'"

    kb = KeyBindings()
    install_key_bindings(kb, state)
    session = PromptSession(
        key_bindings=kb,
        completer=RawCompleter(state),
        complete_while_typing=True,               # ipython-style live menu
        auto_suggest=AutoSuggestFromHistory(),    # inline suggestion hints
        bottom_toolbar=bottom_toolbar,            # live relevance hint
    )

    def message():
        return state.breadcrumb() + " "

    with patch_stdout():
        while True:
            try:
                line = session.prompt(message)
            except (EOFError, KeyboardInterrupt):
                break
            if not line.strip():
                continue
            if not dispatch(state, line, out):
                break
    out("bye")
    return 0