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


def dispatch(state: RawSession, line: str) -> bool:
    """Handle one input line. Returns False to exit the REPL, True to keep going. Testable
    without a live prompt."""
    try:
        cmd = parse_line(line, state.entry_mode)
    except ValueError as exc:
        print(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)
        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")
        elif cmd.name == "tlv":
            _handle_tlv(state, cmd.args)
        return True

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

    # raw payload
    if not cmd.payload:
        return True
    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


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


def _handle_tlv(state: RawSession, args) -> 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:
            print(f"rawcli: {exc}")
            return
    else:
        data = prompt_tlv(state)          # interactive multi-line editor
        if not data:
            return
    print(format_tlv(data))


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


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

    from .completer import RawCompleter

    state = RawSession(device=_connect(port))
    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
    )

    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):
                break
    print("bye")
    return 0