Experimental/keyrelay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
56aa97d60d96de481fe076608c1cb63d1d673ef7
keyrelay/tools/relayctl.py
T
ducoterra ece0313d02 feat: relative mouse on the HID interface, and the verifier to prove it 
The board now enumerates as a boot keyboard plus a relative mouse on the same
HID interface. The pointer is captured, never grabbed: Pointer Lock in the
page, so Esc (the browser's own, never forwarded) always frees it, and every
frame restates the whole button byte so a frame lost in the WiFi cannot leave
a button held. Losing the lock, the tab or the link runs the same release
funnel as a stuck key, and boot protocol or `mouse off` refuses mouse frames
with E11 rather than sending bytes a boot interface would misparse as keys.

src/mouse.h owns the report, src/output.h the queues and the funnel; the page,
relayctl.py and watch_target.py carry the button bits and the evdev decoding.
The verifier's own tables are pinned now — REL_WHEEL is 8, not 2, and the
first version guessed from the report's field order, so a working wheel looked
dead while the summary said no scroll happened.

LLM-Generated: true
2026-11-08 23:27:42 -05:00

893 lines
42 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""relayctl -- talk to a keyrelay board from a terminal.
Stdlib only (socket, 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.
relayctl.py --host keyrelay.local status
relayctl.py --host 192.168.4.1 arm
relayctl.py --host 10.2.1.99 key Enter # one key, press and release
relayctl.py --host 10.2.1.99 mouse move 120 -40 # relative movement, split to fit
relayctl.py --host 10.2.1.99 mouse click right # press and release
relayctl.py --host 10.2.1.99 mouse drag 200 0 # press, move, release, in order
relayctl.py --host 10.2.1.99 cad # Ctrl+Alt+Del, with --yes
relayctl.py --host keyrelay.local listen # print every status heartbeat
relayctl.py --host keyrelay.local disarm
wl-paste | relayctl.py paste --yes # type what is on the clipboard
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).
`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.
"""
from __future__ import annotations
import argparse
import base64
import hashlib
import json
import os
import socket
import struct
import sys
import time
DEFAULT_PORT = 80
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
# 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=<btn>:<dx>:<dy>:<wheel>:<pan>` -- 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=<ms>:<hex>` 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 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: int = DEFAULT_PORT, token: str = "",
timeout: float = 5.0) -> None:
self.host = host
self.timeout = timeout
self.token = token
self.sock = socket.create_connection((host, port), timeout)
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 + " ":
detail = "refused"
if "401" in status_line:
detail = "passphrase required (this endpoint does not take one)"
elif "503" in status_line:
detail = "another controller already holds the bridge"
raise RelayError(f"handshake failed: {status_line} ({detail})")
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'}")
# -- plumbing -----------------------------------------------------------
def _send(self, text: str) -> None:
self.sock.sendall(encode_frame(OP_TEXT, text.encode()))
def _poll(self, want_status: bool, timeout: float):
deadline = time.time() + timeout
while time.time() < deadline:
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:
return json.loads(text[1:])
elif text == "OK":
if not want_status:
return text
elif text.startswith("E"):
raise RelayError(f"board said {text!r}")
else:
print(f"board: {text}", file=sys.stderr)
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 main() -> int:
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=DEFAULT_PORT)
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)
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)")
args = ap.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
try:
relay = Relay(args.host, args.port, args.token, args.timeout)
except (OSError, RelayError) as exc:
print(f"cannot use the board at {args.host}: {exc}", file=sys.stderr)
return 1
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())
Reference in New Issue View Git Blame Copy Permalink