#!/usr/bin/env python3 """relayctl -- talk to a keyrelay board from a terminal. Stdlib only (socket, ssl, hashlib, base64, json): the board's protocol is a WebSocket of tiny text commands, and the point of a keyboard bridge is that it needs nothing installed to be useful. A TLS board (the default) needs exactly one trust argument, and refuses to run without one -- this tool will not be the component that quietly downgrades: --fingerprint pin the board certificate (the `cert` command on the board's serial console prints it; also $KEYRELAY_TLS_FP). Verification is off during the handshake, then the peer's DER is hashed and a mismatch aborts before a single frame -- including T -- is sent. Real pinning, not TOFU, and nothing to distribute. --ca verify against a CA (or a pinned self-signed cert file) with hostname checking on. --insecure no verification; prints a warning on every run. --plain plaintext ws:// on :80, for -D KR_TLS=0 rollback boards only. Opt-in, because the TLS board refuses that port anyway (301 / 426). relayctl.py --fingerprint $FP --host keyrelay.local status relayctl.py --fingerprint $FP --host 192.168.4.1 arm relayctl.py --fingerprint $FP --host 10.2.1.99 key Enter # one key, press+release relayctl.py --fingerprint $FP --host 10.2.1.99 mouse move 120 -40 # split to fit relayctl.py --fingerprint $FP --host 10.2.1.99 mouse click right # press+release relayctl.py --fingerprint $FP --host 10.2.1.99 mouse drag 200 0 # in order relayctl.py --fingerprint $FP --host 10.2.1.99 cad # Ctrl+Alt+Del, --yes relayctl.py --fingerprint $FP --host keyrelay.local listen # status heartbeats relayctl.py --fingerprint $FP --host keyrelay.local disarm wl-paste | relayctl.py --fingerprint $FP paste --yes # type what is on the clipboard relayctl.py --fingerprint $FP --take-over status # take a bridge left held, then use it The passphrase is taken from --token, then $KEYRELAY_TOKEN. It is never written to disk and never put in a URL (the /status HTTP endpoint takes it as a query parameter because curl is the point of that endpoint; this tool uses the WebSocket instead). Over TLS both the passphrase and the keystrokes are encrypted in transit; the fingerprint is what says the other end is *this* board rather than somebody's. `paste` is the one command that types more than a single key, because sending text a human copied is a thing a keyboard bridge is for. It is gated accordingly: --yes is required, the board is armed by that same deliberate invocation (arming cannot outlive a controller, so no separate `arm` command could have done it), the text is never echoed back, and the run disarms again afterwards. What is still absent is a script or macro facility: the board cannot store a sequence, cannot be told to type later, and nothing here replays one. See README "What this deliberately is not". `mouse` is the same argument with a pointer. Movement is relative -- the board reports how far, never where -- so the target's own pointer speed decides where a given `move` lands, and there is no position for this tool to be wrong about. `click` and `drag` press real buttons and need --yes for that reason; a movement does not, because moving a cursor is reversible by definition. There is no gesture recording, no auto-click and no "move to the OK button", because the moment this tool could actuate a target on its own it would be a different and worse device than a keyboard. A busy board is a refusal, never a fallback. If another controller holds the bridge, this tool authenticated into the board's one pending-claim slot and says so -- and names --take-over, which is the same one deliberate act the page's TAKE OVER button sends: `X` after the passphrase. The handover runs through the board's release funnel -- the incumbent is told (E12), every key and every mouse button is released -- and the session continues DISARMED. `--take-over` against a free board is a no-op, not an error. Neither can arm anything: the takeover cannot produce an armed board, and the arm still needs its own deliberate command, exactly as it did before anyone was displaced. What the TLS in this tool does NOT do is make anything safe by itself. An encrypted session is still only a controller that has to know the passphrase, and a session that drops is a lost link the board answers by releasing every key and every mouse button. The certificate says *which* board, and only if you asked -- which is why this tool's contribution is procedural: never pick the weaker option for the operator. No fallback from wss:// to ws://, no default-on --insecure, no "try :80 first", no quiet downgrade when a pin fails; every one of those is a decision the operator makes in writing, and the failure text names the real choice instead of blaming the passphrase for a transport problem. """ from __future__ import annotations import argparse import base64 import hashlib import json import os import socket import ssl import struct import sys import time DEFAULT_PORT = 80 # plaintext ws:// -- reachable only via --plain TLS_DEFAULT_PORT = 443 # wss:// on a KR_TLS=1 board WS_PATH = "/ws" # The whole point of `keys` is verifying the keystroke path in one armed session. # A cap keeps it that: enough for a test vector, not a way to type a paragraph. MAX_SESSION_KEYS = 16 # Pasted text. All three numbers are the board's, not this tool's -- keep them in # step with src/config.h (kTextFrameMax, kTextJobCap) or the board starts refusing # frames this tool thought were legal. TEXT_FRAME_MAX = 120 # characters per P= frame MAX_PASTE_CHARS = 2048 # characters one run holds on the board DEFAULT_TEXT_PACE_MS = 12 # what the board uses when asked for 0 # Mouse movement. Both numbers are the board's -- keep them in step with src/config.h # (kMouseDeltaMax, kMouseButtonMask) or this tool builds frames the board refuses. # One M= frame is one HID report, and a report carries one signed byte per axis, so a # larger movement is split into several frames here rather than guessed at there. MOUSE_DELTA_MAX = 127 # one signed byte, per axis, per frame MOUSE_BUTTON_MASK = 0x1f # five buttons: all the report can carry MOUSE_FRAME_MS = 8 # the pace the controller page sends at, matched here # How long to wait after `OK` for the board's adoption status. A controller gets # `OK` and a status frame back to back (the firmware's auth path sends both); a # socket the board parked as a PENDING CLAIM gets `OK` and silence, because the # status frame belongs to the controller alone. Half a second is orders of # magnitude more than one LAN round-trip, and it is the board's own behaviour -- # this tool never sends a frame to guess at an answer it was already owed. ADOPT_PROBE_S = 0.5 # Button name -> the report's bit, in the order the HID report descriptor lays them out. # NOTE the DOM's MouseEvent.button numbers are NOT this order (there 1 is middle and 2 is # right), which is why src/web_page.h carries its own mapping and both are pinned by # tests: get it wrong and a right-click arrives as a middle-click, cleanly parsed. MOUSE_BUTTONS = { "left": 0x01, "right": 0x02, "middle": 0x04, "back": 0x08, "forward": 0x10, } GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" OP_TEXT, OP_PING, OP_PONG, OP_CLOSE = 0x1, 0x9, 0xA, 0x8 # KeyboardEvent.code -> HID usage id, the same table the controller page carries # (src/web_page.h). Names here are those codes, so `key ShiftLeft` and # `key BracketRight` work the way they do in the browser. CODE_TO_USAGE = { "KeyA": 0x04, "KeyB": 0x05, "KeyC": 0x06, "KeyD": 0x07, "KeyE": 0x08, "KeyF": 0x09, "KeyG": 0x0A, "KeyH": 0x0B, "KeyI": 0x0C, "KeyJ": 0x0D, "KeyK": 0x0E, "KeyL": 0x0F, "KeyM": 0x10, "KeyN": 0x11, "KeyO": 0x12, "KeyP": 0x13, "KeyQ": 0x14, "KeyR": 0x15, "KeyS": 0x16, "KeyT": 0x17, "KeyU": 0x18, "KeyV": 0x19, "KeyW": 0x1A, "KeyX": 0x1B, "KeyY": 0x1C, "KeyZ": 0x1D, "Digit1": 0x1E, "Digit2": 0x1F, "Digit3": 0x20, "Digit4": 0x21, "Digit5": 0x22, "Digit6": 0x23, "Digit7": 0x24, "Digit8": 0x25, "Digit9": 0x26, "Digit0": 0x27, "Enter": 0x28, "Escape": 0x29, "Backspace": 0x2A, "Tab": 0x2B, "Space": 0x2C, "Minus": 0x2D, "Equal": 0x2E, "BracketLeft": 0x2F, "BracketRight": 0x30, "Backslash": 0x31, "IntlBackslash": 0x32, "Semicolon": 0x33, "Quote": 0x34, "Backquote": 0x35, "Comma": 0x36, "Period": 0x37, "Slash": 0x38, "CapsLock": 0x39, "F1": 0x3A, "F2": 0x3B, "F3": 0x3C, "F4": 0x3D, "F5": 0x3E, "F6": 0x3F, "F7": 0x40, "F8": 0x41, "F9": 0x42, "F10": 0x43, "F11": 0x44, "F12": 0x45, "PrintScreen": 0x46, "ScrollLock": 0x47, "Pause": 0x48, "Insert": 0x49, "Home": 0x4A, "PageUp": 0x4B, "Delete": 0x4C, "End": 0x4D, "PageDown": 0x4E, "ArrowRight": 0x4F, "ArrowLeft": 0x50, "ArrowDown": 0x51, "ArrowUp": 0x52, "NumLock": 0x53, "NumpadDivide": 0x54, "NumpadMultiply": 0x55, "NumpadSubtract": 0x56, "NumpadAdd": 0x57, "NumpadEnter": 0x58, "Numpad1": 0x59, "Numpad2": 0x5A, "Numpad3": 0x5B, "Numpad4": 0x5C, "Numpad5": 0x5D, "Numpad6": 0x5E, "Numpad7": 0x5F, "Numpad8": 0x60, "Numpad9": 0x61, "Numpad0": 0x62, "NumpadDecimal": 0x63, "IntlRo": 0x64, "ContextMenu": 0x65, "NumpadEqual": 0x67, "F13": 0x68, "F14": 0x69, "F15": 0x6A, "F16": 0x6B, "F17": 0x6C, "F18": 0x6D, "F19": 0x6E, "F20": 0x6F, "F21": 0x70, "F22": 0x71, "F23": 0x72, "F24": 0x73, } MOD_CODE = { "ControlLeft": 0x01, "ShiftLeft": 0x02, "AltLeft": 0x04, "MetaLeft": 0x08, "ControlRight": 0x10, "ShiftRight": 0x20, "AltRight": 0x40, "MetaRight": 0x80, } class RelayError(RuntimeError): pass # --------------------------------------------------------------------------- # Text -> keystrokes # --------------------------------------------------------------------------- def normalize_paste(text: str) -> str: """CRLF and a lone CR both become LF. The board maps \r to Enter, so an unnormalised Windows paste would type two Enters per line. The board cannot fix this itself: a CRLF pair can straddle a frame boundary, so normalising belongs to whoever holds the whole text. """ return text.replace("\r\n", "\n").replace("\r", "\n") def untypeable(text: str) -> list[str]: """Characters with no key on a US-ANSI keyboard. Tab, newline and printable ASCII are what the board's character map can reach. Anything else -- curly quotes, em dashes, UTF-8 accents -- is reported instead of guessed at, because a paste that quietly loses a character is worse than one that refuses to go. """ return [ch for ch in text if not (ord(ch) == 9 or ord(ch) == 10 or 32 <= ord(ch) <= 126)] def mouse_frame(buttons: int, dx: int = 0, dy: int = 0, wheel: int = 0, pan: int = 0) -> str: """Render `M=::::` -- one mouse report, five fields. Strict in the same direction the firmware parses: the button byte must fit the five bits the report has and every movement must fit one signed byte, or this raises. A frame the board would refuse is not sent, and nothing is clamped to make it fit -- clamp a 300 and the cursor quietly travels half as far as was asked. """ if buttons & ~MOUSE_BUTTON_MASK: raise RelayError(f"buttons {buttons:#02x} does not fit a five-button report " f"(have: {', '.join(MOUSE_BUTTONS)})") for name, v in (("dx", dx), ("dy", dy), ("wheel", wheel), ("pan", pan)): if not -MOUSE_DELTA_MAX <= v <= MOUSE_DELTA_MAX: raise RelayError(f"{name}={v} is more than one report's worth " f"(+/-{MOUSE_DELTA_MAX}); split it") return f"M={buttons:02x}:{dx}:{dy}:{wheel}:{pan}" def split_moves(dx: int, dy: int, wheel: int = 0, pan: int = 0) -> list[tuple]: """Greedy split of a movement into one-report chunks, same shape as the page's. Both axes travel together in each chunk, so a diagonal keeps its shape rather than arriving as an L, and the sums are exact -- 300 px out means 300 px on the target. """ out: list[tuple] = [] while dx or dy or wheel or pan: step = [max(-MOUSE_DELTA_MAX, min(MOUSE_DELTA_MAX, v)) for v in (dx, dy, wheel, pan)] out.append(tuple(step)) dx -= step[0] dy -= step[1] wheel -= step[2] pan -= step[3] return out or [(0, 0, 0, 0)] def mouse_frames(action: str, values: list[str], button: str = "left") -> list[tuple]: """Turn a `mouse` subcommand into the frames to send, as (buttons, dx, dy, wheel, pan). A click is two frames -- the button restated down, then restated up -- with the movement in between, because the byte on the wire is the whole button state and the order is the entire difference between a click and a drag. """ if button not in MOUSE_BUTTONS: # Checked for every action, not only the ones that use it: `--button thumb` is a # typo, and a command that quietly ignores a nonsense option is one that will be # typed again. raise RelayError(f"unknown button {button!r} (have: {', '.join(MOUSE_BUTTONS)})") if button != "left" and action not in ("click", "drag"): # `mouse move 100 0 --button right` would otherwise move the cursor and pretend the # button was involved. It is not: only the verbs that press something take a # button, and saying so beats inventing a meaning for it later. raise RelayError(f"`{action}` does not press a button, so --button has no " f"meaning here (click and drag do take one)") if action == "release": return [(0, 0, 0, 0, 0)] if action == "move": if len(values) not in (2, 3, 4): raise RelayError("usage: mouse move DX DY [WHEEL] [PAN]") nums = [int(v) for v in values] + [0] * (4 - len(values)) return [(0, *rest) for rest in split_moves(*nums)] if action == "wheel": if len(values) != 1: raise RelayError("usage: mouse wheel NOTCHES (negative scrolls up)") # split_moves returns (dx, dy, wheel, pan); the wheel belongs in the third slot. return [(0, 0, 0, part[2], part[3]) for part in split_moves(0, 0, int(values[0]))] bit = MOUSE_BUTTONS[button] if action == "click": return [(bit, 0, 0, 0, 0), (0, 0, 0, 0, 0)] if action == "drag": if len(values) not in (2, 3, 4): raise RelayError("usage: mouse drag DX DY [WHEEL] [PAN]") nums = [int(v) for v in values] + [0] * (4 - len(values)) return [(bit, 0, 0, 0, 0)] + [(bit, *rest) for rest in split_moves(*nums)] \ + [(0, 0, 0, 0, 0)] raise RelayError(f"unknown mouse action {action!r} " "(try move, click, drag, wheel or release)") def text_frame(delay_ms: int, chunk: str) -> str: """One `P=:` command, byte-identical to proto.h's formatText().""" return f"P={delay_ms}:{chunk.encode('utf-8').hex()}" # --------------------------------------------------------------------------- # WebSocket, the smallest useful slice # --------------------------------------------------------------------------- def expect_accept(key: str) -> str: """Sec-WebSocket-Accept for a Sec-WebSocket-Key (RFC 6455 1.3).""" digest = hashlib.sha1((key + GUID).encode()).digest() return base64.b64encode(digest).decode() def handshake_refusal(status_line: str) -> str: """Why the board said no, phrased so the operator can act on it. These shapes used to collapse into one "refused" and they are not the same problem: 426 the board refuses to speak plaintext at all. Its controller channel is `wss://` on :443, and a `ws://` upgrade on :80 is answered rather than quietly accepted -- invariant 5, never silently take the weaker option. The fix is a trust argument (--fingerprint from the board's `cert`), not a different port. 301 the same refusal, for a request that was not an upgrade: the port redirects. 401 the endpoint needs a passphrase this path cannot carry. 503 another controller already holds the bridge. Saying "wrong passphrase" when the board means "use TLS" is the failure this function exists to prevent: the operator then re-types a passphrase that was never the problem, five times, and locks the token out for kTokenLockMs on a board that was trying to protect them. tests/test_relayctl.py pins that 426 and 401 do not share a sentence. """ code = "" for part in status_line.split(): if len(part) == 3 and part.isdigit(): code = part break if code == "426": return ("board refuses plaintext: its controller channel is wss:// on :443. " "Re-run with a trust argument (--fingerprint from the `cert` " "command on the board's serial console, or --ca); --plain only reaches " "a board built with -D KR_TLS=0") if code == "301": return ("board redirected to https://: it no longer serves content over plaintext " "http -- use the TLS defaults (--fingerprint or --ca)") if code == "401": return "passphrase required (this endpoint does not take one)" if code == "503": return "another controller already holds the bridge" if code == "400": return "malformed request -- not a keyrelay WebSocket endpoint" return "refused" # --------------------------------------------------------------------------- # TLS: how this client decides to trust the board, or refuses to guess # --------------------------------------------------------------------------- FINGERPRINT_HEX = 64 # one SHA-256, lowercase hex def normalize_fingerprint(text: str) -> str: """Lowercase 64-hex of a SHA-256 fingerprint, whatever shape it was pasted in. Accepts what the board's `cert` prints (bare hex) and what OpenSSL prints (`AA:BB:...`); refuses anything else rather than pinning a typo. A pin that silently means nothing is worse than no pin, because it reads like one. """ raw = "".join((text or "").split()).replace(":", "") if len(raw) != FINGERPRINT_HEX or any(c not in "0123456789abcdef" for c in raw.lower()): raise RelayError( f"{text!r} is not a SHA-256 fingerprint: want {FINGERPRINT_HEX} hex " f"digits (the `cert` command on the board's serial console prints it)") return raw.lower() def select_transport(fingerprint: str, ca: str, insecure: bool, plain: bool) -> str: """One trust decision from the four switches: "pin" | "ca" | "insecure" | "plain". Refuses, in two directions. None of them: refuse to connect at all -- a tool that defaults to plaintext, or to trust-on-first-use, is the component that quietly downgrades (AGENTS.md invariant 5), so the default posture is a refusal that says what to pass. More than one: they are different trust stories (a pin, a chain, no verification, no TLS), and picking one silently is guessing. """ if plain: if fingerprint or ca or insecure: raise RelayError("--plain is the -D KR_TLS=0 rollback path; do not combine " "it with a TLS trust argument (--fingerprint/--ca/--insecure)") return "plain" chosen = [name for name, on in (("--fingerprint", fingerprint), ("--ca", ca), ("--insecure", insecure)) if on] if len(chosen) > 1: raise RelayError(f"{' and '.join(chosen)} are different trust decisions; pass " f"exactly one") if not chosen: raise RelayError( "refusing to connect without saying how to trust the board: pass " "--fingerprint (or $KEYRELAY_TLS_FP; the `cert` command on the " "board's serial console prints it), or --ca , or --insecure for an " "unverified run this tool will warn about every time; --plain only for a " "board built with -D KR_TLS=0") if fingerprint: return "pin" return "ca" if ca else "insecure" def resolve_port(mode: str, port) -> int: """Explicit --port always wins; otherwise TLS is 443 and --plain is 80. One function because "which port did this flag combination pick" must not be folklore spread across call sites -- pinning a certificate on the plaintext port would be a very confident way to verify nothing. """ if port is not None: return int(port) return DEFAULT_PORT if mode == "plain" else TLS_DEFAULT_PORT def build_tls_context(mode: str, ca: str = "") -> ssl.SSLContext: """The handshake policy for one TLS mode. Pinning and --insecure share the handshake settings deliberately: the pin does its work AFTER the handshake, on the DER the peer actually presented, so it works against a self-signed certificate with no CA ceremony -- and an empty verification path cannot be mistaken for a filled one. --ca is the opposite shape: OpenSSL verifies the chain AND the hostname, because that is the option for operators who ran their own CA on purpose. """ if mode == "ca": ctx = ssl.create_default_context(cafile=ca) ctx.check_hostname = True # the certificate is issued for keyrelay.local ctx.verify_mode = ssl.CERT_REQUIRED return ctx ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE # the pin hashes the peer cert itself; see above return ctx def peer_fingerprint(sock) -> str: """SHA-256 of the DER the peer presented -- the same value `cert` prints.""" der = sock.getpeercert(binary_form=True) if not der: raise RelayError("the TLS peer presented no certificate to fingerprint") return hashlib.sha256(der).hexdigest() def encode_frame(opcode: int, payload: bytes = b"") -> bytes: """Client frames are always masked (RFC 6455 5.1: a server must reject others).""" out = bytearray([0x80 | opcode]) n = len(payload) if n < 126: out.append(0x80 | n) elif n <= 0xFFFF: out.append(0x80 | 126) out += struct.pack(">H", n) else: out.append(0x80 | 127) out += struct.pack(">Q", n) mask = os.urandom(4) out += mask out += bytes(b ^ mask[i & 3] for i, b in enumerate(payload)) return bytes(out) def mask_payload(payload: bytes, mask: bytes) -> bytes: """XOR with the repeating 4-byte mask key. Its own inverse, as the spec says.""" return bytes(b ^ mask[i & 3] for i, b in enumerate(payload)) class Receiver: """Incremental frame decoder for the unmasked frames the board sends.""" def __init__(self) -> None: self.buf = bytearray() def feed(self, data: bytes): """Yield (opcode, payload) for every complete frame in `data`.""" self.buf.extend(data) while True: if len(self.buf) < 2: return b0, b1 = self.buf[0], self.buf[1] if b1 & 0x80: raise RelayError("board sent a masked frame (RFC 6455 forbids that)") if not b0 & 0x80: raise RelayError("board sent a fragmented frame") length = b1 & 0x7F offset = 2 if length == 126: if len(self.buf) < offset + 2: return length = struct.unpack(">H", bytes(self.buf[offset:offset + 2]))[0] offset += 2 elif length == 127: if len(self.buf) < offset + 8: return length = struct.unpack(">Q", bytes(self.buf[offset:offset + 8]))[0] offset += 8 if len(self.buf) < offset + length: return payload = bytes(self.buf[offset:offset + length]) del self.buf[:offset + length] yield b0 & 0x0F, payload class Relay: def __init__(self, host: str, port=None, token: str = "", timeout: float = 5.0, fingerprint: str = "", ca: str = "", insecure: bool = False, plain: bool = False, take_over: bool = False) -> None: self.host = host self.timeout = timeout self.token = token # A status frame that arrives while this side is waiting for some OTHER # reply is the freshest truth about the board there is -- and the adoption # probe below depends on not losing it: the firmware sends a controller's # `OK` and its status back to back, often in one batch, and abandoning the # batch at the OK would make a free board look busy for exactly the wrong # reason. Held here, read once by the probe. self._pending_status = None # The transport decision happens before a socket exists: "no trust # argument" must refuse in the terminal, not connect and find out. self.mode = select_transport(fingerprint, ca, insecure, plain) self.port = resolve_port(self.mode, port) self.pinned = "" if self.mode == "pin": self.pinned = normalize_fingerprint(fingerprint) if self.mode == "insecure": # On every run, not once per install: an opt-in that stops announcing # itself becomes a default. print("WARNING: --insecure -- not verifying the board's certificate. " "Anyone on this network can answer. Use --fingerprint (the `cert` " "command on the serial console prints it) unless you know exactly " "why you cannot.", file=sys.stderr) self.sock = socket.create_connection((host, self.port), timeout) if self.mode != "plain": try: ctx = build_tls_context(self.mode, ca=ca) self.sock = ctx.wrap_socket(self.sock, server_hostname=host) except (ssl.SSLError, OSError) as exc: self.sock.close() if self.mode == "ca": raise RelayError( f"TLS to {host} failed verification: {exc} -- a certificate " f"verify failure usually means the board regenerated its " f"identity (`nvs erase`) or you are talking to a different " f"device; compare the fingerprint with `cert`") from exc raise RelayError( f"TLS handshake with {host} failed: {exc} -- if the board answered " f"the plaintext port with a redirect or a 426, you are on the wrong " f"port or the wrong device; a TLS keyrelay is on 443") from exc if self.mode == "pin": # Abort BEFORE a single frame -- above all before T. A pin # checked after the WebSocket handshake is a pin on the passphrase. try: got = peer_fingerprint(self.sock) except RelayError: self.sock.close() raise if got != self.pinned: self.sock.close() raise RelayError( f"certificate fingerprint mismatch: {host} presented " f"{got}, you pinned {self.pinned}. Nothing was sent. Either " f"the board regenerated its identity (`nvs erase` reprints " f"the new one once) or somebody else is answering -- which " f"is exactly the attack this check exists for. Re-read the " f"fingerprint with `cert` on the board's serial console.") self.receiver = Receiver() key = base64.b64encode(os.urandom(16)).decode() request = ( f"GET {WS_PATH} HTTP/1.1\r\n" f"Host: {host}\r\n" f"Upgrade: websocket\r\n" f"Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" f"Sec-WebSocket-Version: 13\r\n\r\n" ) self.sock.sendall(request.encode()) head = b"" while b"\r\n\r\n" not in head: chunk = self.sock.recv(4096) if not chunk: raise RelayError("server closed during the handshake") head += chunk if len(head) > 16384: raise RelayError("handshake response too large") status_line = head.split(b"\r\n", 1)[0].decode(errors="replace") # Header NAMES are case-insensitive; header VALUES are not. Lowercasing the # whole block -- as the first version of this did -- lowercases the base64 # accept hash too, and base64 is case-sensitive, so every handshake looked # broken while the board was answering correctly. Parse into fields instead. fields: dict[str, str] = {} for line in head.split(b"\r\n\r\n", 1)[0].decode("latin-1").split("\r\n")[1:]: name, sep, value = line.partition(":") if sep: fields[name.strip().lower()] = value.strip() if " 101 " not in status_line + " ": raise RelayError(f"handshake failed: {status_line} " f"({handshake_refusal(status_line)})") want = expect_accept(key) got = fields.get("sec-websocket-accept", "") if got != want: raise RelayError(f"handshake failed: Sec-WebSocket-Accept is {got!r}, " f"expected {want!r}") self._send("T" + self.token) reply = self._await_reply() if reply != "OK": raise RelayError(f"authentication refused: {reply or 'no reply'}") # Did the board ADOPT us? `OK` means the passphrase was right; the frame # that follows is the answer to "is the bridge yours". A controller gets its # status immediately (auth sends OK then status in one go); a socket the # board parked in its spare slot as a pending claim -- another controller # holds the bridge -- gets `OK` and nothing else, because the status frame # belongs to the controller alone. Silence here is therefore the board # reporting `busy` in exactly the terms it chose: a link state, never an # identity. Reading is all this costs -- no frame is sent to guess at an # answer a controller was already owed. if self._pending_status is not None: # The status rode in with the OK (the usual case on a LAN): adopted, # no waiting, and no timeout being mistaken for a fact. self._pending_status = None self.busy = False else: self.busy = self._poll(want_status=True, timeout=ADOPT_PROBE_S) is None self.took_over = False if self.busy: if not take_over: # The refusal names the flag rather than letting some later command # earn an E6 nobody can interpret. Free the claim slot for the one # person who decides, promptly. self.sock.close() raise RelayError( "another controller holds this bridge: the board authenticated " "us as a pending claim and kept the keyboard. Re-run with " "--take-over to take it: the incumbent is told (E12), every key " "and every mouse button is released through the release funnel, " "and the board comes back DISARMED -- nothing types until the " "new session arms on purpose.") # The one deliberate act, and the only frame beyond T and K a claim may # send. The board tells the incumbent, runs the funnel, lands DISARMED, # and adopts us: OK, then status, like any other authenticated start. self._send("X") self.took_over = True # -- plumbing ----------------------------------------------------------- def _send(self, text: str) -> None: self.sock.sendall(encode_frame(OP_TEXT, text.encode())) def _poll(self, want_status: bool, timeout: float): # The whole batch is drained before any early return: a status frame that # rides in behind the reply this call is waiting for is kept in # _pending_status, not abandoned with the generator. Losing it here is how # the adoption probe would read a free board as busy. deadline = time.time() + timeout while time.time() < deadline: hit = None for opcode, payload in self._read_any(max(0.0, deadline - time.time())): if opcode == OP_PING: self.sock.sendall(encode_frame(OP_PONG, payload)) elif opcode == OP_CLOSE: raise RelayError("board closed the connection") elif opcode == OP_TEXT: text = payload.decode(errors="replace") if text.startswith("S{"): if want_status: hit = json.loads(text[1:]) else: self._pending_status = json.loads(text[1:]) elif text == "OK": if not want_status: hit = text elif text.startswith("E"): raise RelayError(f"board said {text!r}") else: print(f"board: {text}", file=sys.stderr) if hit is not None: return hit if not want_status: return None return None def _read_any(self, timeout: float): self.sock.settimeout(max(0.05, timeout)) try: data = self.sock.recv(4096) except socket.timeout: return [] if not data: raise RelayError("connection closed by the board") return self.receiver.feed(data) def _await_reply(self, timeout: float = 4.0): return self._poll(want_status=False, timeout=timeout) # -- commands ----------------------------------------------------------- def wait_idle(self, timeout: float) -> bool: """Poll until the board says it has nothing left to type, or the time is up. Keeps the connection talking while it waits: a controller that says nothing for kNoClientTimeoutMs is one the board disarms, and a long paste would otherwise stop itself halfway through. `typing` in the status frame is the board's own count, so this is flow control rather than a guess at timing. """ deadline = time.time() + timeout next_beat = time.time() + 2.0 while time.time() < deadline: if time.time() >= next_beat: self._send("K") next_beat = time.time() + 2.0 got = self._poll(want_status=True, timeout=0.25) if got is not None and not got.get("typing"): return True return False def type_text(self, text: str, delay_ms: int = 0, show: bool = True): """Type text, one frame per TEXT_FRAME_MAX characters, board-paced. Returns (characters confirmed typed, seconds elapsed). A frame the board refuses raises RelayError out of _poll, so a run that cannot continue does not quietly carry on with what is left of the text. """ pace = delay_ms or DEFAULT_TEXT_PACE_MS chunks = [text[i:i + TEXT_FRAME_MAX] for i in range(0, len(text), TEXT_FRAME_MAX)] started = time.time() sent = confirmed = 0 for chunk in chunks: self._send(text_frame(delay_ms, chunk)) sent += len(chunk) # The chunk's own typing time, plus a heartbeat's worth of slack. if self.wait_idle(len(chunk) * (pace + 4) / 1000.0 + 4.0): confirmed = sent if show: print(f" {confirmed}/{len(text)} characters", end="\r", flush=True) if confirmed != sent: break if show and chunks: print(" " * 34, end="\r" if confirmed == len(text) else "\n", flush=True) return confirmed, time.time() - started def _drain(self, seconds: float = 0.4) -> None: """Throw away whatever is already buffered. The board heartbeats status at 1 Hz whether or not anyone asked, and the key helpers do not read while they type, so seconds of stale frames pile up in the socket. Reading a status without draining first returns one of *those* -- which is how `disarm` looked like it had failed while printing `armed YES`: the frame shown was a heartbeat sent before the command. Pending error frames are still raised, because an E-frame is never stale. """ deadline = time.time() + seconds while time.time() < deadline: for opcode, payload in self._read_any(min(0.1, max(0.01, deadline - time.time()))): if opcode == OP_PING: # Answer it the way _poll does: an unanswered ping is a link the # board may decide is dead, and this method must not be the one # thing that breaks the keepalive (it called methods that do not # exist, which only ever showed up on a session over 15 s long). self.sock.sendall(encode_frame(OP_PONG, payload)) elif opcode == OP_CLOSE: raise RelayError("board closed the connection") elif opcode == OP_TEXT: text = payload.decode("utf-8", "replace") if text.startswith("E"): raise RelayError(f"board said {text!r}") def status(self) -> dict: self._drain() self._send("H") got = self._poll(want_status=True, timeout=self.timeout) if got is None: raise RelayError("no status reply") return got def _settle(self, want_armed: bool, timeout: float = 2.0) -> dict: """Poll until the board reports the arming we asked for.""" deadline = time.time() + timeout last: dict | None = None while time.time() < deadline: self._drain(0.0) got = self._poll(want_status=True, timeout=0.25) if got is not None: last = got if bool(got.get("armed")) == want_armed: return got return last or {} def arm(self) -> dict: self._drain() self._send("A1") state = self._settle(True) if not state.get("armed"): raise RelayError("the board did not arm: " + ", ".join(f"{k}={v}" for k, v in state.items())) return state def disarm(self) -> dict: self._drain() self._send("R") # keys up first, then disarm self._send("A0") state = self._settle(False) if state.get("armed"): raise RelayError("the board did not disarm -- use the button on the board") return state def release(self) -> None: self._send("R") def state(self, mods: int, usages) -> None: body = ",".join(f"{u:02x}" for u in usages) self._send(f"S={mods:02x}:{body}") def one_key(self, usage: int, hold_ms: int) -> None: self.state(0x00, [usage]) time.sleep(hold_ms / 1000.0) self.state(0x00, []) def chord(self, mods: int, usages) -> None: body = ",".join(f"{u:02x}" for u in usages) self._send(f"Y={mods:02x}:{body}") def mouse(self, frames, gap_ms: float = MOUSE_FRAME_MS) -> int: """Send mouse reports, paced so they arrive like a mouse and not a flood. No waiting for acknowledgement between frames -- a movement is not a state the board has to confirm, and the board's own queue is what absorbs a burst. But the gap is real: sent as fast as a socket will take them, a few hundred reports would pile into the board's queue and trip the overflow fail-safe, which releases everything and disarms. An error frame (E10 no mouse interface, E11 boot protocol) raises out of the next read rather than being swallowed. """ sent = 0 for buttons, dx, dy, wheel, pan in frames: self._send(mouse_frame(buttons, dx, dy, wheel, pan)) sent += 1 if gap_ms: time.sleep(gap_ms / 1000.0) self._drain(0.0) # pick up anything the board said about them return sent def listen(self, seconds: float) -> None: """Print heartbeats. Sends keepalives: a silent client is one the board disarms after kNoClientTimeoutMs, and watching must not change what is watched.""" deadline = time.time() + seconds next_beat = time.time() + 2.0 while time.time() < deadline: if time.time() >= next_beat: self._send("K") next_beat = time.time() + 2.0 try: for opcode, payload in self._read_any(1.0): if opcode == OP_PING: self.sock.sendall(encode_frame(OP_PONG, payload)) elif opcode == OP_TEXT: text = payload.decode(errors="replace") if text.startswith("S{"): s = json.loads(text[1:]) print(f" net={s.get('net'):4s} ip={s.get('ip'):15s} " f"usb={s.get('usb')} armed={s.get('armed')} " f"keys={s.get('keys')} chords={s.get('chords')} " f"mframes={s.get('mframes', 0)} " f"mbtn=0x{s.get('mbtn', 0):02x} " f"drop={s.get('drop')} leds=0x{s.get('leds'):02x}") else: print(f" {text}") elif opcode == OP_CLOSE: return except socket.timeout: continue def close(self) -> None: try: self.sock.sendall(encode_frame(OP_CLOSE)) except OSError: pass self.sock.close() # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def resolve_key(name: str) -> int: if name.lower().startswith("0x"): value = int(name, 16) if not 0 < value <= 0xFF: raise RelayError("a usage id is 01..ff") return value if name in CODE_TO_USAGE: return CODE_TO_USAGE[name] known = ", ".join(sorted(CODE_TO_USAGE)[:12]) raise RelayError(f"unknown key {name!r} (try 0x28, or one of: {known}, ...)") def describe(status: dict) -> str: lines = [ f" firmware {status.get('v')}", f" uptime {status.get('up', 0)} s", f" network {status.get('net')} ssid={status.get('ssid') or '-'} " f"ip={status.get('ip') or '-'}", f" target {'attached over USB' if status.get('usb') else 'NOT ATTACHED'}", f" armed {'YES -- keystrokes are being forwarded' if status.get('armed') else 'no'}", f" counters keys={status.get('keys')} chords={status.get('chords')} " f"dropped={status.get('drop')}", f" mouse {'can report' if status.get('mouse') else 'CANNOT REPORT'} " f"buttons=0x{status.get('mbtn', 0):02x} reports={status.get('mframes', 0)}", f" text {status.get('typed', 0)} characters typed" + (f", {status['typing']} still to go" if status.get('typing') else ""), f" leds caps={bool(status.get('leds', 0) & 2)} " f"num={bool(status.get('leds', 0) & 1)}", ] if status.get("dflt"): lines.append(" WARNING the board still uses the default passphrase") if status.get("usb") and not status.get("mouse"): lines.append(" NOTE no mouse reports: `mouse on` over serial and reboot, or" "\n the target chose boot protocol (it would read a mouse" "\n report as keystrokes, so the board refuses them)") if status.get("roll"): lines.append(" NOTE more than six keys are down: ErrorRollOver is out") return "\n".join(lines) def build_arg_parser() -> argparse.ArgumentParser: """The CLI surface, apart from main() so tests can parse without running.""" ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--host", default="keyrelay.local", help="board address (default keyrelay.local; on its own AP use " "192.168.4.1)") ap.add_argument("--port", type=int, default=None, help="override the port: default 443 over TLS, 80 with --plain") ap.add_argument("--token", default=os.environ.get("KEYRELAY_TOKEN", ""), help="control passphrase (or $KEYRELAY_TOKEN)") ap.add_argument("--timeout", type=float, default=5.0) ap.add_argument("--fingerprint", default=os.environ.get("KEYRELAY_TLS_FP", ""), metavar="SHA256", help="pin the board certificate: abort before sending anything " "unless the peer's DER hashes to this (or $KEYRELAY_TLS_FP); " "the serial `cert` command prints it") ap.add_argument("--ca", default="", metavar="PEM", help="verify the board certificate against this CA file (or " "pinned self-signed certificate), hostname checking on") ap.add_argument("--insecure", action="store_true", help="connect without any verification; warns on stderr on " "every run -- a one-off, not a habit") ap.add_argument("--plain", action="store_true", help="plaintext ws:// on port 80: only for boards built with " "-D KR_TLS=0 (the rollback); a TLS board refuses this") ap.add_argument("--take-over", action="store_true", help="if another controller holds the bridge, take it (X): the " "incumbent is told (E12), every key and mouse button is " "released through the release funnel, and the board comes " "back DISARMED. Without it a busy board is a refusal; " "against a free board it is a no-op. Neither path arms: " "arm is still its own deliberate command") sub = ap.add_subparsers(dest="cmd", required=True) sub.add_parser("status", help="print what the board reports") sub.add_parser("arm", help="arm the bridge (nothing types before this)") sub.add_parser("disarm", help="release every key and disarm") sub.add_parser("release", help="release every key, stay armed") p_cad = sub.add_parser("cad", help="Ctrl+Alt+Del -- needs --yes, and a thought") p_cad.add_argument("--yes", action="store_true", help="acknowledge that this reboots a Linux virtual terminal") p_key = sub.add_parser("key", help="press and release exactly one key") p_key.add_argument("name", help="key code name (Enter, KeyA, F7...) or 0x28") p_key.add_argument("--hold", type=int, default=40, metavar="MS", help="how long to hold it (default 40)") p_chord = sub.add_parser("chord", help="press one chord: mods plus usages") p_chord.add_argument("mods", help="modifier byte, e.g. 05 = LCtrl|LAlt") p_chord.add_argument("usages", nargs="*", help="usage ids, e.g. 4c") p_listen = sub.add_parser("listen", help="print status heartbeats") p_listen.add_argument("seconds", nargs="?", type=float, default=20.0) p_paste = sub.add_parser( "paste", help="type text a human is sending (e.g. `wl-paste | relayctl paste --yes`)") p_paste.add_argument("text", nargs="?", help="the text; omit it, or give '-', to read stdin") p_paste.add_argument("--delay", type=int, default=0, metavar="MS", help="milliseconds per character (0 = the board's default)") p_paste.add_argument("--strip", action="store_true", help="drop characters this keyboard has no key for, instead " "of refusing the paste") p_paste.add_argument("--trailing-enter", action="store_true", help="press Enter once the text has been typed") p_paste.add_argument("--keep-armed", action="store_true", help="leave the bridge armed afterwards") p_paste.add_argument("--yes", action="store_true", help="acknowledge that this types into the target") p_mouse = sub.add_parser( "mouse", help="move the target's pointer, or press one of its buttons") p_mouse.add_argument("action", choices=["move", "click", "drag", "wheel", "release"], help="move: relative dx dy (and optional wheel, pan); " "click: press and release one button; drag: press, move, " "release; wheel: NOTCHES (negative scrolls up); " "release: every button up") p_mouse.add_argument("values", nargs="*", metavar="N", help="numbers for move / drag (px, px [wheel] [pan]) or " "wheel (notches)") p_mouse.add_argument("--button", default="left", help="left, right, middle, back or forward (default left)") p_mouse.add_argument("--gap", type=int, default=MOUSE_FRAME_MS, metavar="MS", help=f"milliseconds between reports (default {MOUSE_FRAME_MS}, " "the rate the controller page uses)") p_mouse.add_argument("--keep-armed", action="store_true", help="stay armed afterwards") p_mouse.add_argument("--yes", action="store_true", help="required for click and drag, which can actuate something") p_keys = sub.add_parser( "keys", help="verification session: arm, press these keys, disarm again") p_keys.add_argument("names", nargs="+", metavar="KEY", help=f"key names or 0xNN, at most {MAX_SESSION_KEYS}") p_keys.add_argument("--gap", type=int, default=120, metavar="MS", help="pause between keys (default 120)") p_keys.add_argument("--hold", type=int, default=40, metavar="MS", help="how long each key is held (default 40)") p_keys.add_argument("--keep-armed", action="store_true", help="stay armed afterwards (the board still disarms on " "silence 8 s after this process exits)") return ap def main() -> int: args = build_arg_parser().parse_args() if not args.token: print("no passphrase: --token or $KEYRELAY_TOKEN", file=sys.stderr) return 2 # Checked before anything is contacted: the acknowledgement is the point, and # finding out you forgot it should not require a reachable board, a token and a # connection that then has to be torn down again. if args.cmd == "paste" and not args.yes: print("paste types a whole run of characters into the machine the board is " "plugged into.\nRe-run with --yes if a human means it.\n\n" " printf 'hello\\n' | relayctl.py --host keyrelay.local paste --yes", file=sys.stderr) return 4 # Everything about the text that can be settled locally is settled here, before a # socket is opened: a paste the board would refuse should fail in the terminal # rather than half-arm a bridge and then stop. paste_text = "" if args.cmd == "paste": # No argument, or '-', is the clipboard: `wl-paste | relayctl paste --yes`. raw = sys.stdin.read() if args.text in (None, "-") else args.text paste_text = normalize_paste(raw) if args.trailing_enter: paste_text += "\n" if not paste_text: print("nothing to type: stdin was empty and no text was given", file=sys.stderr) return 2 bad = untypeable(paste_text) if bad: if not args.strip: first = bad[0] print(f"{len(bad)} of {len(paste_text)} characters have no key on a " f"US-ANSI keyboard (first is {first!r} " f"U+{ord(first):04X}).\nRe-run with --strip to type the rest, " f"or fix the text.", file=sys.stderr) return 2 paste_text = "".join(ch for ch in paste_text if ord(ch) == 9 or ord(ch) == 10 or 32 <= ord(ch) <= 126) print(f"dropped {len(bad)} character(s) with no key here") if not paste_text: print("nothing left to type", file=sys.stderr) return 2 if len(paste_text) > MAX_PASTE_CHARS: print(f"{len(paste_text)} characters is more than one run holds " f"({MAX_PASTE_CHARS}); send it in pieces", file=sys.stderr) return 2 # Settled locally too: "you gave me no way to trust the board" is a usage # answer, not a connection failure -- and refusing here means no socket was # ever opened to find that out. try: if select_transport(args.fingerprint, args.ca, args.insecure, args.plain) == "pin": normalize_fingerprint(args.fingerprint) except RelayError as exc: print(str(exc), file=sys.stderr) return 2 try: relay = Relay(args.host, args.port, args.token, args.timeout, fingerprint=args.fingerprint, ca=args.ca, insecure=args.insecure, plain=args.plain, take_over=args.take_over) except (OSError, RelayError) as exc: print(f"cannot use the board at {args.host}: {exc}", file=sys.stderr) return 1 if args.take_over: # Say out loud which of the two no-surprises paths this took -- the # operator clicked a takeover, and "what happened to the other session" # is the question this answers before any keystroke goes out. print("took the bridge -- the incumbent was told (E12), every key and mouse " "button was released, and the board is DISARMED" if relay.took_over else "the board was free -- --take-over had nothing to take") try: if args.cmd == "status": print(describe(relay.status())) elif args.cmd == "arm": relay.arm() print("armed -- keystrokes now go to the target") print(describe(relay.status())) elif args.cmd == "disarm": relay.disarm() print("disarmed, every key released") elif args.cmd == "release": relay.release() print("every key released (still armed)") elif args.cmd == "key": usage = resolve_key(args.name) before = relay.status() if not before.get("armed"): print("the bridge is disarmed: nothing was sent. Run `arm` first.", file=sys.stderr) return 3 print(f"sending {args.name} (usage 0x{usage:02x}) for {args.hold} ms ...") relay.one_key(usage, args.hold) elif args.cmd == "chord": mods = int(args.mods, 16) usages = [int(u, 16) for u in args.usages] if not usages and not mods: print("a chord needs a modifier or a key", file=sys.stderr) return 2 print(f"sending chord Y={mods:02x}:" f"{','.join(f'{u:02x}' for u in usages)}") relay.chord(mods, usages) elif args.cmd == "cad": if not args.yes: print("Ctrl+Alt+Del: on a Linux virtual terminal this REBOOTS the " "target; on Windows it is the secure attention sequence.\n" "Re-run with --yes if that is what you meant.", file=sys.stderr) return 4 status = relay.status() if not status.get("armed"): print("the bridge is disarmed: nothing was sent.", file=sys.stderr) return 3 print("sending Ctrl+Alt+Del ...") relay.chord(0x05, [0x4C]) elif args.cmd == "mouse": # A click can actuate whatever is under the pointer, so it is gated like the # other things that have consequences beyond a character appearing. A # movement is not: it moves a cursor, which is reversible by definition. if args.action in ("click", "drag") and not args.yes: print(f"`{args.action}` presses a real mouse button on the machine the " "board is plugged into, wherever its pointer happens to be.\n" "Re-run with --yes if a human is looking at it.", file=sys.stderr) return 4 try: frames = mouse_frames(args.action, args.values, args.button) except (RelayError, ValueError) as exc: print(f"{exc}", file=sys.stderr) return 2 # Armed inside this one connection, like `keys` and `paste`: the board # disarms the moment a controller drops, so "arm in one invocation and use it # in the next" is not a thing that can work -- and the LED is red for exactly # as long as this session lasts. status = relay.status() if not status.get("mouse", 1): print("this board cannot report a mouse: `mouse on` over serial and " "reboot, or the target chose boot protocol (it would read a " "mouse report as keystrokes, so the board refuses them). Nothing " "was armed and nothing was sent.", file=sys.stderr) return 3 relay.arm() print("armed -- LED should be solid red on the board") n = relay.mouse(frames, args.gap) print(f"{n} mouse report(s) sent ({args.action}" + (f", {args.button}" if args.button != "left" else "") + ")") if not args.keep_armed: relay.disarm() print("disarmed, every key and button released") print(describe(relay.status())) elif args.cmd == "keys": # One deliberate session in one connection: the board disarms the moment # a controller drops (that is the safety property that makes the rest of # this work), so arm-and-type has to be a single client's business. It is # capped and it disarms afterwards: a session, not a script. if len(args.names) > MAX_SESSION_KEYS: print(f"at most {MAX_SESSION_KEYS} keys per session -- this is a " f"verification tool, not a typer", file=sys.stderr) return 2 usages = [resolve_key(n) for n in args.names] print("target: " + " ".join( f"{n}({u:#04x})" for n, u in zip(args.names, usages))) relay.arm() print("armed -- LED should be solid red on the board") for name, usage in zip(args.names, usages): print(f" {name} ...", end=" ", flush=True) relay.one_key(usage, args.hold) time.sleep(args.gap / 1000.0) print() if args.keep_armed: print("left armed") else: relay.disarm() print("disarmed, every key released") print(describe(relay.status())) elif args.cmd == "paste": text = paste_text # Deliberately the count and not the content: what was typed is nobody's # business, including this terminal's scrollback. print(f"about to type {len(text)} characters into " f"{args.host} at {args.delay or DEFAULT_TEXT_PACE_MS} ms each") relay.arm() print("armed -- the board's LED should be solid red") typed, elapsed = relay.type_text(text, args.delay) print(f"typed {typed} of {len(text)} characters in {elapsed:.1f}s") if args.keep_armed: print("left armed") else: relay.disarm() print("disarmed, every key released") elif args.cmd == "listen": relay.listen(args.seconds) except RelayError as exc: print(str(exc), file=sys.stderr) return 1 except KeyboardInterrupt: print() finally: relay.close() return 0 if __name__ == "__main__": sys.exit(main())