Created
July 5, 2026 08:47
-
-
Save maxweisspoker/d5b2174b1ad325fa947bfbf5ea9b309e to your computer and use it in GitHub Desktop.
working trezor-bridge for Qubes
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| """ | |
| Trezor HTTP bridge for Qubes OS. | |
| Makes Trezor hardware wallets work with Trezor Suite and MetaMask (and | |
| anything else that uses trezord or Trezor Connect) on Qubes OS, where the | |
| Trezor is attached to an AppVM via USB-over-IP (qvm-usb). | |
| The problem: Trezor Suite and MetaMask use WebUSB, which requires device | |
| resets and continuous re-enumeration that break under Qubes USBIP. Simpler | |
| tools like Electrum and Monero use plain libusb and work fine. | |
| The solution: this script emulates the trezord-go HTTP API that Trezor | |
| Connect talks to, backed by trezorlib's plain libusb transport (the same | |
| path Electrum uses). Applications think they're talking to the official | |
| Trezor Bridge; the bridge talks to the device over simple, reliable libusb. | |
| Tested flows: account discovery, ETH/ERC-20 transactions, personal_sign, | |
| EIP-712 typed data signing, passphrase entry, address verification, device | |
| settings changes. Works with Trezor Model T; should work with any WebUSB | |
| Trezor (Safe 3, Safe 5). Does NOT work with Trezor Model One (HID device, | |
| different transport). | |
| QUICK START | |
| =========== | |
| Dependencies (install in the Qubes template, not the AppVM): | |
| sudo dnf install python3-trezor python3-libusb1 | |
| 1. Attach your Trezor to the AppVM: | |
| qvm-usb attach <appvm> sys-usb:<device> | |
| Or use the Qubes Devices widget. Verify: lsusb | grep -i trezor | |
| 2. Start this bridge BEFORE opening your browser/Suite: | |
| python3 trezor-bridge.py | |
| Leave it running in its own terminal. | |
| 3a. For Trezor Suite (desktop or https://suite.trezor.io/web): | |
| Just open Suite. It finds the bridge automatically. | |
| 3b. For MetaMask or other browser extensions: | |
| Launch Chrome with this flag (required — without it, Chrome blocks | |
| the browser-to-bridge connection silently): | |
| google-chrome --disable-features=LocalNetworkAccessChecks | |
| Then use MetaMask's "Connect hardware wallet" → Trezor normally. | |
| 4. When done, Ctrl+C the bridge to release the device for other apps | |
| (Electrum, Monero, trezorctl, etc.). | |
| NOTES | |
| ===== | |
| - Only one app can own the Trezor at a time. The bridge holds the USB | |
| interface for its entire lifetime. Stop the bridge to hand the device | |
| back to Electrum/Monero/trezorctl. | |
| - If MetaMask popups appear as tiny/invisible windows, check your | |
| taskbar — Chrome sometimes opens MetaMask notification windows at | |
| near-zero size. Maximize them from the taskbar. | |
| - After a crash or unclean exit, replug the Trezor (qvm-usb detach + | |
| attach) before restarting the bridge to clear stale USB data. | |
| - Do NOT use this for firmware upgrades. Firmware flashing reboots the | |
| device into bootloader mode, which re-enumerates USB — the one thing | |
| USBIP handles worst. Use a direct-USB environment for firmware. | |
| - This is an unofficial reimplementation of trezord's HTTP API, tested | |
| against Trezor Suite 26.6.x and Trezor Connect v9. Future protocol | |
| changes could require updates. | |
| ARCHITECTURE (for developers) | |
| ============================= | |
| - Per-session reader thread reassembles V1-framed messages (64-byte USB | |
| chunks with ?## headers) into a queue. /read long-polls the queue; | |
| /abort flips an event without touching USB. | |
| - No USB enumeration while any session is active (static descriptor | |
| cache) — prevents control-transfer interference over USBIP. | |
| - Open transports pooled and reused across session re-acquires (no | |
| open/close churn over USBIP). | |
| - Per-request protocol framing: responses are framed to match the | |
| client's declared protocol ("v1" or "bridge"). | |
| - Dual-port: 21328 (Suite desktop) + 21325 (browsers). | |
| """ | |
| import collections | |
| import json | |
| import struct | |
| import sys | |
| import threading | |
| import time | |
| from http.server import HTTPServer, BaseHTTPRequestHandler | |
| from socketserver import ThreadingMixIn | |
| try: | |
| from trezorlib.transport.webusb import WebUsbTransport | |
| import usb1 | |
| except ImportError: | |
| print("trezorlib and usb1 required: pip install trezor libusb1") | |
| sys.exit(1) | |
| LISTEN_PORTS = [21328, 21325] # 21328: Suite desktop, 21325: browsers/web Suite | |
| CHUNK = 64 | |
| USB_TIMEOUT_MS = 300 # matches trezorlib's USB_COMM_TIMEOUT_MS | |
| # v1 hack from before the response-framing fix: Suite's first Cancel got a | |
| # Failure(Cancelled) reply that Suite could not DECODE (we framed it wrong), | |
| # which looked like Suite choking on the Cancel itself. With per-request | |
| # protocol framing in place this should be unnecessary — flag kept for one | |
| # falsification run, delete after web Suite confirms. | |
| CONVERT_FIRST_CANCEL = False | |
| MSG_CANCEL = 20 | |
| MSG_INITIALIZE = 0 | |
| def log(msg): | |
| ts = time.strftime("%H:%M:%S") + f".{int(time.time() * 1000) % 1000:03d}" | |
| print(f"[{ts}] {msg}", file=sys.stderr, flush=True) | |
| # --------------------------------------------------------------------------- | |
| # global state | |
| # --------------------------------------------------------------------------- | |
| state_lock = threading.Lock() # guards the tables below | |
| usb_lock = threading.Lock() # serializes libusb enumeration | |
| sessions_by_id = {} # session_id -> DeviceSession | |
| path_sessions = {} # dev_path -> session_id | |
| path_owners = {} # dev_path -> sessionOwner string (from acquire) | |
| open_transports = {} # dev_path -> WebUsbTransport (held open) | |
| broken_paths = set() # dev_paths whose open transport hit a USB error | |
| session_counter = 0 | |
| cached_descriptors = None # last enumeration result (session field unset) | |
| cache_time = 0.0 | |
| def enumerate_transports(): | |
| try: | |
| return WebUsbTransport.enumerate() | |
| except Exception as e: | |
| log(f"enumerate failed: {e}") | |
| return [] | |
| def any_active_session(): | |
| with state_lock: | |
| return bool(sessions_by_id) | |
| def get_descriptors(force=False): | |
| """Descriptor list for /enumerate and /listen. | |
| NEVER touches USB while a session is active — enumeration sends live | |
| control transfers (getProduct) that race the session's interrupt I/O. | |
| """ | |
| global cached_descriptors, cache_time | |
| active = any_active_session() | |
| now = time.monotonic() | |
| if not active and (force or cached_descriptors is None or now - cache_time > 2): | |
| with usb_lock: | |
| devs = enumerate_transports() | |
| cached_descriptors = [{ | |
| "path": d.get_path(), | |
| "vendor": d.device.getVendorID(), | |
| "product": d.device.getProductID(), | |
| } for d in devs] | |
| cache_time = time.monotonic() | |
| base = cached_descriptors or [] | |
| with state_lock: | |
| return [{ | |
| "path": d["path"], | |
| "vendor": d["vendor"], | |
| "product": d["product"], | |
| "session": path_sessions.get(d["path"]), | |
| "sessionOwner": path_owners.get(d["path"]), | |
| "debug": False, | |
| "debugSession": None, | |
| } for d in base] | |
| # --------------------------------------------------------------------------- | |
| # message queue with put-back (so a failed HTTP write doesn't lose a message) | |
| # --------------------------------------------------------------------------- | |
| class MsgQueue: | |
| def __init__(self): | |
| self._items = collections.deque() | |
| self._cond = threading.Condition() | |
| def put(self, item): | |
| with self._cond: | |
| self._items.append(item) | |
| self._cond.notify() | |
| def put_front(self, item): | |
| with self._cond: | |
| self._items.appendleft(item) | |
| self._cond.notify() | |
| def get(self, timeout): | |
| with self._cond: | |
| if not self._items: | |
| self._cond.wait(timeout) | |
| if self._items: | |
| return self._items.popleft() | |
| return None | |
| # --------------------------------------------------------------------------- | |
| # per-session reader thread | |
| # --------------------------------------------------------------------------- | |
| class DeviceSession: | |
| """Owns the IN endpoint for one acquired session. | |
| The reader thread reassembles complete v1 messages (?## framing, 64-byte | |
| chunks) into self.queue. Each interruptRead uses a 300ms timeout so the | |
| thread can notice self.stop quickly — the same URB cadence trezorlib | |
| itself uses, so no new USBIP churn is introduced. | |
| """ | |
| def __init__(self, session_id, dev_path, transport): | |
| self.id = session_id | |
| self.path = dev_path | |
| self.transport = transport | |
| self.queue = MsgQueue() | |
| self.stop = threading.Event() | |
| self.abort = threading.Event() | |
| self.dead = False | |
| self.first_message_seen = False | |
| self.write_lock = threading.Lock() | |
| self.reader = threading.Thread( | |
| target=self._read_loop, name=f"reader-{session_id}", daemon=True) | |
| self.reader.start() | |
| # -- reading ------------------------------------------------------------ | |
| def _read_chunk(self, usb_handle, endpoint): | |
| """One 64-byte chunk or None on timeout/stop. Raises on USB error. | |
| Trezor V1 always uses padded 64-byte packets, so a short read is stale | |
| or misaligned data (e.g. left in the pipe after an unclean bridge exit). | |
| Discard it and keep reading rather than killing the reader thread — the | |
| stream resyncs on the next ?##-aligned packet. This is what prevents the | |
| "reader died: unexpected chunk size N" failure after a hard shutdown. | |
| """ | |
| while not self.stop.is_set(): | |
| try: | |
| chunk = usb_handle.interruptRead(endpoint, CHUNK, USB_TIMEOUT_MS) | |
| except usb1.USBErrorTimeout: | |
| return None | |
| if chunk: | |
| if len(chunk) != CHUNK: | |
| log(f"session {self.id}: discarding {len(chunk)}-byte " | |
| f"stale/misaligned chunk {bytes(chunk[:12]).hex()}") | |
| time.sleep(0.002) | |
| continue | |
| return chunk | |
| time.sleep(0.001) | |
| return None | |
| def _read_loop(self): | |
| handle_obj = self.transport.handle # WebUsbHandle | |
| usb_handle = handle_obj.handle # usb1.USBDeviceHandle | |
| endpoint = 0x80 | handle_obj.endpoint | |
| log(f"session {self.id}: reader started ({self.path})") | |
| try: | |
| chunk_debug = True # full hex logging for diagnostics | |
| while not self.stop.is_set(): | |
| first = self._read_chunk(usb_handle, endpoint) | |
| if first is None: | |
| continue | |
| if chunk_debug: | |
| log(f"session {self.id}: RAW header candidate " | |
| f"({len(first)}B): {bytes(first).hex()}") | |
| if first[:3] != b"?##": | |
| log(f"session {self.id}: dropping non-header chunk " | |
| f"(magic={bytes(first[:3]).hex()}, " | |
| f"expected=3f2323)") | |
| continue | |
| msg_type, datalen = struct.unpack(">HL", first[3:9]) | |
| buf = bytearray(first[9:]) | |
| complete = True | |
| while len(buf) < datalen: | |
| nxt = self._read_chunk(usb_handle, endpoint) | |
| if nxt is None: | |
| if self.stop.is_set(): | |
| complete = False | |
| break | |
| continue # mid-message timeout: keep waiting | |
| if chunk_debug: | |
| log(f"session {self.id}: RAW continuation " | |
| f"({len(nxt)}B): {bytes(nxt[:16]).hex()}...") | |
| if nxt[:1] != b"?": | |
| log(f"session {self.id}: bad continuation magic, " | |
| f"abandoning message type={msg_type}") | |
| complete = False | |
| break | |
| buf.extend(nxt[1:]) | |
| if complete: | |
| log(f"session {self.id}: msg type={msg_type} len={datalen}") | |
| self.queue.put(("msg", msg_type, bytes(buf[:datalen]))) | |
| except Exception as e: | |
| if not self.stop.is_set(): | |
| log(f"session {self.id}: reader died: {e}") | |
| self.dead = True | |
| with state_lock: | |
| broken_paths.add(self.path) | |
| self.queue.put(("error", f"device read failed: {e}")) | |
| log(f"session {self.id}: reader exiting") | |
| # -- writing ------------------------------------------------------------ | |
| def write(self, msg_type, payload): | |
| with self.write_lock: | |
| self.transport.write(msg_type, payload) | |
| # -- lifecycle ---------------------------------------------------------- | |
| def shutdown(self): | |
| """Stop the reader thread. Does NOT close the USB handle.""" | |
| self.stop.set() | |
| self.abort.set() | |
| self.reader.join(timeout=2.0) | |
| if self.reader.is_alive(): | |
| log(f"session {self.id}: WARNING reader did not stop in 2s") | |
| # --------------------------------------------------------------------------- | |
| # helpers | |
| # --------------------------------------------------------------------------- | |
| def decode_v1(data): | |
| if len(data) >= 3 and data[0:3] == b"?##": | |
| data = data[3:] | |
| msg_type = struct.unpack(">H", data[0:2])[0] | |
| msg_len = struct.unpack(">I", data[2:6])[0] | |
| return msg_type, data[6:6 + msg_len] | |
| def encode_response(msg_type, payload, protocol): | |
| """Frame a device message per the client-requested protocol. | |
| Suite declares the wire protocol per request ({"protocol": "v1"|"bridge"}) | |
| and DECODES the response with that same protocol — reply in the wrong | |
| framing and @trezor/protocol throws "Malformed protocol format". | |
| v1: 3f2323 ("?##") + type(u16BE) + len(u32BE) + payload | |
| bridge: type(u16BE) + len(u32BE) + payload | |
| """ | |
| frame = struct.pack(">H", msg_type) + struct.pack(">I", len(payload)) + payload | |
| if protocol == "v1": | |
| frame = b"?##" + frame | |
| return frame.hex() | |
| def parse_body(body): | |
| """Extract (data_hex, protocol) from a request body. | |
| Bodies are JSON {"data": hex, "protocol": "v1"|"bridge"} from current | |
| Suite; bare hex from legacy clients (treated as protocol "bridge"). | |
| """ | |
| raw = body.decode().strip() | |
| data_hex, protocol = raw, "bridge" | |
| try: | |
| msg = json.loads(raw) | |
| if isinstance(msg, dict): | |
| data_hex = msg.get("data") or "" | |
| protocol = msg.get("protocol") or "bridge" | |
| except (json.JSONDecodeError, AttributeError): | |
| pass | |
| return data_hex, protocol | |
| def get_session(session_id): | |
| with state_lock: | |
| return sessions_by_id.get(session_id) | |
| # --------------------------------------------------------------------------- | |
| # HTTP handler | |
| # --------------------------------------------------------------------------- | |
| class BridgeHandler(BaseHTTPRequestHandler): | |
| def log_message(self, fmt, *args): | |
| pass | |
| def _cors(self): | |
| origin = self.headers.get("Origin", "") | |
| if origin: | |
| self.send_header("Access-Control-Allow-Origin", origin) | |
| self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS") | |
| self.send_header("Access-Control-Allow-Headers", "Content-Type") | |
| # Chromium Private Network Access: https page -> 127.0.0.1 requires | |
| # this on the preflight or every request is silently blocked | |
| self.send_header("Access-Control-Allow-Private-Network", "true") | |
| def _respond(self, code, data): | |
| log(f" <<< [{code}] {str(data)[:120]}") | |
| body = json.dumps(data).encode() | |
| self.send_response(code) | |
| self._cors() | |
| self.send_header("Content-Length", str(len(body))) | |
| self.end_headers() | |
| self.wfile.write(body) | |
| def _read_body(self): | |
| length = int(self.headers.get("Content-Length", 0)) | |
| return self.rfile.read(length) if length > 0 else b"" | |
| def do_OPTIONS(self): | |
| self.send_response(200) | |
| self._cors() | |
| self.end_headers() | |
| def do_GET(self): | |
| log(f">>> GET {self.path}") | |
| if self.path.rstrip("/") == "": | |
| return self._respond(200, { | |
| "version": "3.2.1", | |
| "protocolMessages": True, | |
| "githash": "qubes-bridge-v2", | |
| }) | |
| self.send_response(404) | |
| self._cors() | |
| self.end_headers() | |
| def do_POST(self): | |
| try: | |
| self._dispatch() | |
| except (BrokenPipeError, ConnectionResetError): | |
| # client dropped the connection (Suite does this alongside /abort) | |
| log(f" client disconnected during {self.path}") | |
| def _dispatch(self): | |
| log(f">>> POST {self.path}") | |
| path = self.path.rstrip("/") | |
| if path == "" or path == "/": | |
| return self._respond(200, { | |
| "version": "3.2.1", | |
| "protocolMessages": True, | |
| "githash": "qubes-bridge-v2", | |
| }) | |
| if path == "/configure": | |
| return self._respond(200, {"version": "3.2.1", "protocolMessages": True}) | |
| if path == "/enumerate": | |
| return self._respond(200, get_descriptors(force=True)) | |
| if path == "/listen": | |
| return self._handle_listen() | |
| if path.startswith("/acquire/"): | |
| return self._handle_acquire(path) | |
| if path.startswith("/release/"): | |
| return self._handle_release(path) | |
| if path.startswith("/call/"): | |
| return self._handle_call(path) | |
| if path.startswith("/post/"): | |
| return self._handle_post(path) | |
| if path.startswith("/read/"): | |
| return self._handle_read(path) | |
| if path.startswith("/abort/"): | |
| return self._handle_abort(path) | |
| self.send_response(404) | |
| self._cors() | |
| self.end_headers() | |
| # -- discovery ---------------------------------------------------------- | |
| def _handle_listen(self): | |
| body = self._read_body() | |
| try: | |
| prev = json.loads(body) if body else [] | |
| except json.JSONDecodeError: | |
| prev = [] | |
| prev_normalized = [{ | |
| "path": d.get("path"), | |
| "vendor": d.get("vendor"), | |
| "product": d.get("product"), | |
| "session": d.get("session"), | |
| "sessionOwner": d.get("sessionOwner"), | |
| "debug": d.get("debug", False), | |
| "debugSession": d.get("debugSession"), | |
| } for d in prev] | |
| prev_json = json.dumps(prev_normalized, sort_keys=True) | |
| current = get_descriptors() | |
| for _ in range(30): | |
| current = get_descriptors() | |
| if json.dumps(current, sort_keys=True) != prev_json: | |
| break | |
| time.sleep(1) | |
| self._respond(200, current) | |
| # -- session lifecycle ---------------------------------------------------- | |
| def _handle_acquire(self, path): | |
| global session_counter | |
| parts = path.split("/") | |
| if len(parts) < 3: | |
| return self._respond(400, {"error": "invalid path"}) | |
| dev_path = parts[2] | |
| previous = parts[3] if len(parts) > 3 else "null" | |
| owner = None | |
| try: | |
| body = json.loads(self._read_body() or b"{}") | |
| if isinstance(body, dict): | |
| owner = body.get("sessionOwner") | |
| except json.JSONDecodeError: | |
| pass | |
| # retire any existing session on this device (lenient about `previous` | |
| # mismatches, like v1 — Suite normally passes the right one) | |
| with state_lock: | |
| cur_sid = path_sessions.pop(dev_path, None) | |
| path_owners.pop(dev_path, None) | |
| old = sessions_by_id.pop(cur_sid, None) if cur_sid else None | |
| if cur_sid and previous not in (cur_sid, "null"): | |
| log(f"acquire: previous={previous} but current={cur_sid} " | |
| f"(proceeding anyway)") | |
| if old: | |
| old.shutdown() # join outside state_lock | |
| with state_lock: | |
| transport = open_transports.get(dev_path) | |
| path_broken = dev_path in broken_paths | |
| if transport is not None and path_broken: | |
| log(f"acquire: {dev_path} marked broken, discarding old transport") | |
| try: | |
| transport.end_session() | |
| except Exception: | |
| pass | |
| with state_lock: | |
| open_transports.pop(dev_path, None) | |
| broken_paths.discard(dev_path) | |
| transport = None | |
| if transport is None: | |
| with usb_lock: | |
| transport = next( | |
| (d for d in enumerate_transports() if d.get_path() == dev_path), | |
| None) | |
| if transport is None: | |
| return self._respond(400, {"error": "device not found"}) | |
| try: | |
| transport.begin_session() # opens handle, claims interface | |
| except Exception as e: | |
| return self._respond(400, {"error": f"open failed: {e}"}) | |
| log(f"acquire: opened {dev_path} (handle will be reused across " | |
| f"re-acquires; Ctrl+C the bridge to free the device)") | |
| else: | |
| log(f"acquire: reusing open handle for {dev_path}") | |
| with state_lock: | |
| session_counter += 1 | |
| sid = str(session_counter) | |
| sess = DeviceSession(sid, dev_path, transport) | |
| sessions_by_id[sid] = sess | |
| path_sessions[dev_path] = sid | |
| path_owners[dev_path] = owner | |
| open_transports[dev_path] = transport | |
| self._respond(200, {"session": sid}) | |
| def _handle_release(self, path): | |
| parts = path.split("/") | |
| if len(parts) < 3: | |
| return self._respond(400, {"error": "invalid path"}) | |
| sid = parts[2] | |
| with state_lock: | |
| sess = sessions_by_id.pop(sid, None) | |
| if sess and path_sessions.get(sess.path) == sid: | |
| path_sessions.pop(sess.path, None) | |
| path_owners.pop(sess.path, None) | |
| if sess is None: | |
| return self._respond(400, {"error": "session not found"}) | |
| sess.shutdown() | |
| # transport intentionally stays open in open_transports for reuse | |
| self._respond(200, {"session": sid}) | |
| def _handle_abort(self, path): | |
| parts = path.split("/") | |
| sid = parts[2] if len(parts) > 2 else None | |
| sess = get_session(sid) if sid else None | |
| if sess: | |
| log(f"session {sid}: abort requested (HTTP-level only)") | |
| sess.abort.set() | |
| return self._respond(200, {}) | |
| # -- I/O ------------------------------------------------------------------ | |
| def _write_decoded(self, sess, body): | |
| """Write the client message to the device; returns declared protocol.""" | |
| data_hex, protocol = parse_body(body) | |
| data = bytes.fromhex(data_hex) | |
| msg_type, msg_payload = decode_v1(data) | |
| if (CONVERT_FIRST_CANCEL and msg_type == MSG_CANCEL | |
| and not sess.first_message_seen): | |
| log(f"session {sess.id}: first message is Cancel — " | |
| f"converting to Initialize") | |
| msg_type, msg_payload = MSG_INITIALIZE, b"" | |
| sess.first_message_seen = True | |
| log(f"session {sess.id}: write type={msg_type} len={len(msg_payload)} " | |
| f"proto={protocol}") | |
| sess.write(msg_type, msg_payload) | |
| return protocol | |
| def _pop_message(self, sess, overall_timeout=None): | |
| """Long-poll sess.queue. Returns ('msg', t, p) / ('error', e) / None. | |
| None means aborted or timed out. Blocks indefinitely by default, like | |
| real trezord — /abort or a dead session are the exits. | |
| """ | |
| sess.abort.clear() | |
| deadline = (time.monotonic() + overall_timeout) if overall_timeout else None | |
| while True: | |
| item = sess.queue.get(timeout=0.25) | |
| if item is not None: | |
| return item | |
| if sess.abort.is_set(): | |
| return None | |
| if sess.dead: | |
| return ("error", "device disconnected") | |
| if deadline and time.monotonic() > deadline: | |
| return None | |
| def _deliver(self, sess, item, protocol): | |
| """Send a popped message; on client-gone, put it back for the next /read.""" | |
| if item is None: | |
| # canonical trezord error string — the client matches on it | |
| return self._respond(400, {"error": "Aborted by signal"}) | |
| if item[0] == "error": | |
| return self._respond(400, {"error": item[1]}) | |
| _, msg_type, payload = item | |
| try: | |
| self._respond(200, { | |
| "data": encode_response(msg_type, payload, protocol), | |
| "protocol": protocol, | |
| }) | |
| except (BrokenPipeError, ConnectionResetError): | |
| log(f"session {sess.id}: client vanished mid-response, " | |
| f"re-queuing message type={msg_type}") | |
| sess.queue.put_front(item) | |
| def _handle_post(self, path): | |
| parts = path.split("/") | |
| if len(parts) < 3: | |
| return self._respond(400, {"error": "invalid path"}) | |
| sess = get_session(parts[2]) | |
| if sess is None: | |
| return self._respond(400, {"error": "session not found"}) | |
| if sess.dead: | |
| return self._respond(400, {"error": "device disconnected"}) | |
| try: | |
| protocol = self._write_decoded(sess, self._read_body()) | |
| self._respond(200, {"data": "", "protocol": protocol}) | |
| except Exception as e: | |
| log(f"session {sess.id}: /post ERROR: {e}") | |
| self._respond(400, {"error": str(e)}) | |
| def _handle_read(self, path): | |
| parts = path.split("/") | |
| if len(parts) < 3: | |
| return self._respond(400, {"error": "invalid path"}) | |
| sess = get_session(parts[2]) | |
| if sess is None: | |
| return self._respond(400, {"error": "session not found"}) | |
| _, protocol = parse_body(self._read_body()) | |
| self._deliver(sess, self._pop_message(sess), protocol) | |
| def _handle_call(self, path): | |
| parts = path.split("/") | |
| if len(parts) < 3: | |
| return self._respond(400, {"error": "invalid path"}) | |
| sess = get_session(parts[2]) | |
| if sess is None: | |
| return self._respond(400, {"error": "session not found"}) | |
| if sess.dead: | |
| return self._respond(400, {"error": "device disconnected"}) | |
| try: | |
| protocol = self._write_decoded(sess, self._read_body()) | |
| except Exception as e: | |
| log(f"session {sess.id}: /call write ERROR: {e}") | |
| return self._respond(400, {"error": str(e)}) | |
| self._deliver(sess, self._pop_message(sess, overall_timeout=60), protocol) | |
| # --------------------------------------------------------------------------- | |
| # main | |
| # --------------------------------------------------------------------------- | |
| def main(): | |
| devs = enumerate_transports() | |
| if devs: | |
| print(f"Trezor found: {devs[0].get_path()}") | |
| else: | |
| print("No Trezor detected yet (attach via qvm-usb)") | |
| class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): | |
| daemon_threads = True | |
| servers = [] | |
| for port in LISTEN_PORTS: | |
| server = ThreadingHTTPServer(("127.0.0.1", port), BridgeHandler) | |
| servers.append(server) | |
| print(f"Bridge v2 listening on http://127.0.0.1:{port}") | |
| print("Start Trezor Suite (bridge must start BEFORE Suite).") | |
| print("Press Ctrl+C to stop.\n") | |
| threads = [threading.Thread(target=s.serve_forever, daemon=True) | |
| for s in servers[1:]] | |
| for t in threads: | |
| t.start() | |
| try: | |
| servers[0].serve_forever() | |
| except KeyboardInterrupt: | |
| print("\nShutting down...") | |
| with state_lock: | |
| live = list(sessions_by_id.values()) | |
| transports = list(open_transports.values()) | |
| sessions_by_id.clear() | |
| path_sessions.clear() | |
| open_transports.clear() | |
| for s in live: | |
| s.shutdown() | |
| for t in transports: | |
| try: | |
| t.end_session() | |
| except Exception: | |
| pass | |
| for s in servers: | |
| s.server_close() | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment