trezor-firmware/python/tools/pybridge.py

#!/usr/bin/env python3
# ### INSTRUCTIONS FOR USE ####
#
# 1. install Python 3.7 and up
# 2. make sure you have `pip3` command available
# 3. from command line, run the following:
#        pip3 install trezor[hidapi] gevent bottle
# 4. (ONLY for TT or T1 >= 1.8.0) Make sure you have libusb available.
#    4a. on Windows, download:
#        https://github.com/libusb/libusb/releases/download/v1.0.26/libusb-1.0.26-binaries.7z
#        Extract file VS2015-x64/dll/libusb-1.0.dll and place it in your working directory.
#    4b. on MacOS, assuming you have Homebrew, run `brew install libusb`
#        Otherwise download the above, extract the file macos_<best version>/lib/libusb.1.0.0.dylib
#        and place it in your working directory.
#    4c. on Linux, use your package manager to install `libusb` or `libusb-1.0` package.
#        (but on Linux you most likely already have it)
# 4. Shut down Trezor Suite (and bridge if you are running it separately
# 5. Disconnect and then reconnect your Trezor.
# 6. Run the following command from the command line:
#        python3 pybridge.py
# 7. Start Suite again, or use any other Trezor-compatible software.
# 8. Output of pybridge goes to console and also to file `pybridge.log`
from __future__ import annotations  # type: ignore [unknown import symbol]

from gevent import monkey

monkey.patch_all()

import json
import logging
import struct
import time
import typing as t

import click
from bottle import post, request, response, run

import trezorlib.mapping
import trezorlib.models
import trezorlib.transport
from trezorlib.client import TrezorClient
from trezorlib.protobuf import format_message
from trezorlib.transport.bridge import BridgeTransport
from trezorlib.ui import TrezorClientUI

# ignore bridge. we are the bridge
BridgeTransport.ENABLED = False


logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s %(levelname)s %(message)s",
    handlers=[
        logging.FileHandler("pybridge.log"),
        logging.StreamHandler(),
    ],
)

LOG = logging.getLogger()


class SilentUI(TrezorClientUI):
    def get_pin(self, _code: t.Any) -> str:
        return ""

    def get_passphrase(self) -> str:
        return ""

    def button_request(self, _br: t.Any) -> None:
        pass


class Session:
    SESSION_COUNTER = 0
    SESSIONS: dict[str, Session] = {}

    def __init__(self, transport: Transport) -> None:
        self.id = self._new_id()
        self.transport = transport
        self.SESSIONS[self.id] = self

    def release(self) -> None:
        self.SESSIONS.pop(self.id, None)
        self.transport.release()

    @classmethod
    def find(cls, sid: str) -> Session | None:
        return cls.SESSIONS.get(sid)

    @classmethod
    def _new_id(cls) -> str:
        id = str(cls.SESSION_COUNTER)
        cls.SESSION_COUNTER += 1
        return id


class Transport:
    TRANSPORT_COUNTER = 0
    TRANSPORTS: dict[str, Transport] = {}

    def __init__(self, transport: trezorlib.transport.Transport) -> None:
        self.path = transport.get_path()
        self.session: Session | None = None
        self.transport = transport

        client = TrezorClient(transport, ui=SilentUI())
        self.model = (
            trezorlib.models.by_name(client.features.model) or trezorlib.models.TREZOR_T
        )
        client.end_session()

    def acquire(self, sid: str) -> str:
        if self.session_id() != sid:
            raise Exception("Session mismatch")
        if self.session is not None:
            self.session.release()

        self.session = Session(self)
        self.transport.begin_session()
        return self.session.id

    def release(self) -> None:
        self.transport.end_session()
        self.session = None

    def session_id(self) -> str | None:
        if self.session is not None:
            return self.session.id
        else:
            return None

    def to_json(self) -> dict:
        vid, pid = next(iter(self.model.usb_ids), (0, 0))
        return {
            "debug": False,
            "debugSession": None,
            "path": self.path,
            "product": pid,
            "vendor": vid,
            "session": self.session_id(),
        }

    def write(self, msg_id: int, data: bytes) -> None:
        self.transport.write(msg_id, data)

    def read(self) -> tuple[int, bytes]:
        return self.transport.read()

    @classmethod
    def find(cls, path: str) -> Transport | None:
        return cls.TRANSPORTS.get(path)

    @classmethod
    def enumerate(cls) -> t.Iterable[Transport]:
        transports = {t.get_path(): t for t in trezorlib.transport.enumerate_devices()}
        for path in transports:
            if path not in cls.TRANSPORTS:
                cls.TRANSPORTS[path] = Transport(transports[path])

        for path in list(cls.TRANSPORTS):
            if path not in transports:
                cls.TRANSPORTS.pop(path, None)

        return cls.TRANSPORTS.values()


FILTERS: dict[int, t.Callable[[int, bytes], tuple[int, bytes]]] = {}


def log_message(prefix: str, msg_id: int, data: bytes) -> None:
    try:
        msg = trezorlib.mapping.DEFAULT_MAPPING.decode(msg_id, data)
        LOG.info("=== %s: [%s] %s", prefix, msg_id, format_message(msg))
    except Exception:
        LOG.info("=== %s: [%s] undecoded bytes %s", prefix, msg_id, data.hex())


def decode_data(hex_data: str) -> tuple[int, bytes]:
    data = bytes.fromhex(hex_data)
    headerlen = struct.calcsize(">HL")
    msg_type, datalen = struct.unpack(">HL", data[:headerlen])
    return msg_type, data[headerlen : headerlen + datalen]


def encode_data(msg_type: int, msg_data: bytes) -> str:
    data = struct.pack(">HL", msg_type, len(msg_data)) + msg_data
    return data.hex()


def check_origin() -> None:
    response.set_header("Access-Control-Allow-Origin", "*")


@post("/")  # type: ignore [Untyped function decorator]
def index():
    check_origin()
    return {"version": "2.0.27"}


@post("/configure")  # type: ignore [Untyped function decorator]
def do_configure():
    return index()


@post("/enumerate")  # type: ignore [Untyped function decorator]
def do_enumerate():
    check_origin()
    trezor_json = [transport.to_json() for transport in Transport.enumerate()]
    return json.dumps(trezor_json)


@post("/acquire/<path>/<sid>")  # type: ignore [Untyped function decorator]
def do_acquire(path: str, sid: str):
    check_origin()
    if sid == "null":
        sid = None  # type: ignore [cannot be assigned to declared type]
    trezor = Transport.find(path)
    if trezor is None:
        response.status = 404
        return {"error": "invalid path"}

    try:
        return {"session": trezor.acquire(sid)}
    except Exception:
        response.status = 400
        return {"error": "wrong previous session"}


@post("/release/<sid>")  # type: ignore [Untyped function decorator]
def do_release(sid: str):
    check_origin()
    session = Session.find(sid)
    if session is None:
        response.status = 404
        return {"error": "invalid session"}
    session.release()
    return {"session": sid}


@post("/call/<sid>")  # type: ignore [Untyped function decorator]
def do_call(sid: str):
    check_origin()
    session = Session.find(sid)
    if session is None:
        response.status = 404
        return {"error": "invalid session"}

    msg_type, msg_data = decode_data(request.body.read().decode())

    if msg_type in FILTERS:
        msg_type, msg_data = FILTERS[msg_type](msg_type, msg_data)
    log_message("CALLING", msg_type, msg_data)

    session.transport.write(msg_type, msg_data)
    resp_type, resp_data = session.transport.read()

    if resp_type in FILTERS:
        resp_type, resp_data = FILTERS[resp_type](resp_type, resp_data)
    log_message("RESPONSE", resp_type, resp_data)

    return encode_data(resp_type, resp_data)


@post("/post/<sid>")  # type: ignore [Untyped function decorator]
def do_post(sid: str):
    check_origin()
    session = Session.find(sid)
    if session is None:
        response.status = 404
        return {"error": "invalid session"}

    msg_type, msg_data = decode_data(request.body.read().decode())
    session.transport.write(msg_type, msg_data)
    return {"session": sid}


@post("/read/<sid>")  # type: ignore [Untyped function decorator]
def do_read(sid: str):
    check_origin()
    session = Session.find(sid)
    if session is None:
        response.status = 404
        return {"error": "invalid session"}

    resp_type, resp_data = session.transport.read()
    print("=== RESPONSE:")
    msg = trezorlib.mapping.DEFAULT_MAPPING.decode(resp_type, resp_data)
    print(format_message(msg))

    return encode_data(resp_type, resp_data)


@post("/listen")  # type: ignore [Untyped function decorator]
def do_listen():
    check_origin()
    try:
        data = json.load(request.body)
    except Exception:
        response.status = 400
        return {"error": "invalid json"}

    for _ in range(10):
        trezor_json = [transport.to_json() for transport in Transport.enumerate()]
        if trezor_json != data:
            # `yield` turns the function into a generator which allows gevent to
            # run it in a greenlet, so that the time.sleep() call doesn't block
            yield json.dumps(trezor_json)
            return
        time.sleep(1)


# def example_filter(msg_id: int, data: bytes) -> tuple[int, bytes]:
#     msg = trezorlib.mapping.DEFAULT_MAPPING.decode(msg_id, data)
#     assert isinstance(msg, messages.Features)
#     msg.model = "Example"
#     return trezorlib.mapping.DEFAULT_MAPPING.encode(msg)


# FILTERS[messages.Features.MESSAGE_WIRE_TYPE] = example_filter


@click.command()
@click.argument("port", type=int, default=21325)
def main(port: int) -> None:
    run(host="127.0.0.1", port=port, server="gevent")


if __name__ == "__main__":
    main()