ducoterra
e0780f58e0
**Verified / implemented (phase 02 final pass)** - All 6 tasks already in `complete/` → verification pass. Device measurement **found a real defect (README bug 11)**: the cert carried its SAN *value* under the OID written from the ASCII text `"2.5.29.17"` instead of DER content bytes `55 1D 11` (`openssl x509 -text` showed `1.10.46.53…`, hex `30 2f 06 09 32 2e 35…`), i.e. **no SAN** → hostname validation impossible while every host check stayed green (they walked the value, never the label). - Fixed in the pure module: `kSanOidTlv`/`sanOidDer()`, `sanExtensionFullDer()` (pinned to the 43 bytes OpenSSL writes for these names), `derCarriesSan()` (OID *and* value); `tls.cpp` uses them + 3 `static_assert`s; `cert` now warns and names the remedy. 33 new checks in `test_tls_pem.cpp`, incl. the pre-fix extension byte-for-byte (`derContains` yes / `derCarriesSan` no). - Corrected a false finding: the earlier "fingerprint rotated once" was an **OpenSSL `-ext` artifact** (`C8:76…` vs the real `61:54:19:30…`, same DER). Re-measured: serial `cert` == live handshake DER hash; page 51 626 B, `sha256 dd16e195…f04e9` == locally assembled page. **Tests / lint (100 % green)** - `bash .agents/validate.sh` → **GATE_EXIT=0**: `RESULT: all suites passed` in the plain *and* `SANITIZE=1` run, **4826 checks / 0 failures** each; `bridge`/`console`/`rollback` **SUCCESS**; `page bytes: 51626`; `README/AGENTS claims: OK`; `repo hygiene: OK (71 tracked files)`. - Footprint: bridge **RAM 30.4 % (99 616 B) / Flash 39.6 % (1 246 145 B)** vs baseline 26.7 %/87 592 & 36.1 %/1 135 293 → **+12 024 B RAM / +110 852 B flash**; TLS alone +6 160 B / +100 008 B (rollback image byte-identical); session heap 187 936 → ~145.5 k (1 session) → ~107.9 k (2). - `ruff check --select F,E9 tools/ tests/` → All checks passed; `bash -n` clean (gate, runner, probe); `-Wall -Wextra` warning-free. **Completion criteria**: 1 ✓ (gate green + RAM/Flash + session heap quoted); 2 ✓ as recorded (boot prints no fingerprint on the NVS-load path, `cert` prints on demand, identity survived reflashes, nothing about the key is printed anywhere — `nvs erase`→first-boot leg **not re-run**); 3 ✓ re-measured (51 626 == gate; `:80` → 301, no page bytes); 4 partial — 426/401/404 + pinning handshake measured, but arming/typing needs the operator's passphrase (`dflt:0`) and the target is this machine, so no input was sent; 5 ✓ (no `.pem/.key/.crt` tracked or on disk; hygiene block ran); 6 ✓ (`[env:rollback]` built every run + README section); 7 ✓. **Documentation**: README — check count 4793→4826 (intro + Quick start), `test_tls_pem` 829→862 with new pins, footprint/TLS-cost rows re-measured, "Ten bugs"→"Eleven bugs" + new entry + both anchors, `cert` row, Troubleshooting names-warning row, `tls_pem.h` architecture entry, nine/ten pure-module wording. AGENTS.md (staleness only): "Ten bugs"→"Eleven bugs" ×2, 4826, and the certificate-names row extended with the DER-OID rule. `.agents/HW_VERIFICATION.md` — 426 body/headers re-measured, fingerprint row correction + `-ext` rule, new SAN row (`failed → fixed in tree, device leg needs nvs erase`) with closing commands; phase report written to `.agents/reports/02_tls_control_channel/02_tls_control_channel__00_phase.a1.md`. **Deviations / live-infra**: made **no** reflash and **no** `nvs erase` (erase also destroys the token and Wi-Fi credentials; flashing puts a live HID keyboard into ROM download mode on this machine). Live steps were read-only `curl`/`openssl` plus one capped `console.py cert` (opening the CDC resets the board — permitted, recorded). No phase/PLAN files touched; everything left in the working tree for the harness to commit. **Next pending phase**: `03_session_takeover` — `01_pending_claim_socket.md`.
989 lines
50 KiB
Python
989 lines
50 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for the Python side: relayctl's WebSocket codec, and that its key table
|
|
has not drifted from the one in the controller page.
|
|
|
|
python3 tests/test_relayctl.py
|
|
|
|
No third-party dependencies, so it runs anywhere the tools do.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
import pathlib
|
|
import re
|
|
import struct
|
|
import sys
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "tools"))
|
|
|
|
import relayctl # noqa: E402
|
|
|
|
failures = 0
|
|
checks = 0
|
|
|
|
|
|
def check(condition, label):
|
|
global checks, failures
|
|
checks += 1
|
|
if not condition:
|
|
failures += 1
|
|
print(f" FAIL {label}")
|
|
|
|
|
|
def check_eq(actual, expected, label):
|
|
global checks, failures
|
|
checks += 1
|
|
if actual != expected:
|
|
failures += 1
|
|
print(f" FAIL {label}: got {actual!r}, want {expected!r}")
|
|
|
|
|
|
print("handshake")
|
|
# RFC 6455 1.3's worked example -- the same vector the C++ suite asserts, so the
|
|
# two implementations are pinned to the same truth.
|
|
check_eq(relayctl.expect_accept("dGhlIHNhbXBsZSBub25jZQ=="),
|
|
"s3pPLMBiTxaQ9kYGzzhZRbK+xOo=", "Sec-WebSocket-Accept of the RFC example")
|
|
check(relayctl.expect_accept("differentKey==") != relayctl.expect_accept("dGhlIQ=="),
|
|
"different keys give different accepts")
|
|
|
|
print("frame encoding")
|
|
frame = relayctl.encode_frame(relayctl.OP_TEXT, b"A1")
|
|
check_eq(len(frame), 2 + 4 + 2, "short masked frame length")
|
|
check_eq(frame[0], 0x81, "FIN + text opcode")
|
|
check_eq(frame[1] & 0x80, 0x80, "mask bit set (clients must mask)")
|
|
check_eq(frame[1] & 0x7F, 2, "payload length in the 7-bit field")
|
|
mask = frame[2:6]
|
|
check_eq(bytes(b ^ mask[i & 3] for i, b in enumerate(frame[6:])), b"A1",
|
|
"payload round-trips through the mask")
|
|
|
|
long_payload = b"x" * 300
|
|
frame = relayctl.encode_frame(relayctl.OP_TEXT, long_payload)
|
|
check_eq(frame[1] & 0x7F, 126, "126 signals the 16-bit length")
|
|
check_eq(struct.unpack(">H", frame[2:4])[0], 300, "16-bit length value")
|
|
mask = frame[4:8]
|
|
check_eq(relayctl.mask_payload(frame[8:], mask), long_payload,
|
|
"mask_payload is its own inverse")
|
|
check_eq(len(relayctl.encode_frame(relayctl.OP_TEXT, b"")), 6,
|
|
"empty frame still carries a mask key")
|
|
|
|
print("frame decoding")
|
|
receiver = relayctl.Receiver()
|
|
|
|
|
|
def server_frame(opcode, payload):
|
|
"""The unmasked frames the board sends."""
|
|
out = bytearray([0x80 | opcode, len(payload)])
|
|
out += payload
|
|
return bytes(out)
|
|
|
|
|
|
status = server_frame(relayctl.OP_TEXT, b'S{"armed":1,"usb":1}')
|
|
# Deliver it one byte at a time: the decoder must not invent a frame early.
|
|
emitted = []
|
|
for i in range(len(status)):
|
|
emitted += list(receiver.feed(status[i:i + 1]))
|
|
check_eq(len(emitted), 1, "exactly one frame from byte-at-a-time delivery")
|
|
check_eq(emitted[0][0], relayctl.OP_TEXT, "decoded opcode")
|
|
check_eq(emitted[0][1], b'S{"armed":1,"usb":1}',
|
|
"decoded payload, including the board's 'S' status prefix")
|
|
|
|
# Two frames in one read, plus a split across the boundary.
|
|
two = server_frame(relayctl.OP_TEXT, b"OK") + server_frame(relayctl.OP_PING, b"hi")
|
|
receiver = relayctl.Receiver()
|
|
got = list(receiver.feed(two))
|
|
check_eq(len(got), 2, "both frames decoded from one read")
|
|
check_eq(got[0][1], b"OK", "first frame payload")
|
|
check_eq(got[1][0], relayctl.OP_PING, "second frame opcode")
|
|
|
|
receiver = relayctl.Receiver()
|
|
got = list(receiver.feed(two[:5])) + list(receiver.feed(two[5:]))
|
|
check_eq(len(got), 2, "a frame split mid-payload is still decoded once")
|
|
|
|
# 16-bit length path, and the protocol violations the client refuses.
|
|
receiver = relayctl.Receiver()
|
|
big = bytearray([0x81, 126]) + struct.pack(">H", 130) + b"y" * 130
|
|
check_eq(list(receiver.feed(bytes(big)))[0][1], b"y" * 130, "16-bit length decoded")
|
|
|
|
receiver = relayctl.Receiver()
|
|
try:
|
|
list(receiver.feed(bytes([0x81, 0x82]) + b"\x00" * 6))
|
|
check(False, "a masked server frame must be rejected")
|
|
except relayctl.RelayError:
|
|
check(True, "masked server frame rejected")
|
|
|
|
receiver = relayctl.Receiver()
|
|
try:
|
|
list(receiver.feed(bytes([0x01, 0x01, b"a"[0]])))
|
|
check(False, "a fragmented frame must be rejected")
|
|
except relayctl.RelayError:
|
|
check(True, "fragmented server frame rejected")
|
|
|
|
print("key table")
|
|
check_eq(relayctl.resolve_key("Enter"), 0x28, "Enter is usage 0x28")
|
|
check_eq(relayctl.resolve_key("0x28"), 0x28, "numeric usage accepted")
|
|
check_eq(relayctl.resolve_key("Delete"), 0x4C, "Delete is Delete-Forward 0x4C")
|
|
for bad, why in [("0x00", "usage 0 is not a key"), ("0x1ff", "usage must fit a byte"),
|
|
("NotAKey", "unknown name")]:
|
|
try:
|
|
relayctl.resolve_key(bad)
|
|
check(False, why)
|
|
except (relayctl.RelayError, ValueError):
|
|
check(True, why)
|
|
|
|
# The page and this tool must agree, or `relayctl key Enter` and pressing Enter in
|
|
# the browser would be different keys. Compare the JS object in web_page.h.
|
|
page = (pathlib.Path(__file__).resolve().parent.parent / "src" / "web_page.h").read_text()
|
|
match = re.search(r"const CODE_TO_USAGE = \{(.*?)\};", page, re.S)
|
|
check(match is not None, "CODE_TO_USAGE found in src/web_page.h")
|
|
if match:
|
|
page_map = {}
|
|
for name, usage in re.findall(r"([A-Za-z][A-Za-z0-9]*)\s*:\s*(0x[0-9a-fA-F]+)",
|
|
match.group(1)):
|
|
page_map[name] = int(usage, 16)
|
|
check(len(page_map) > 80, f"page table parsed ({len(page_map)} entries)")
|
|
only_page = sorted(set(page_map) - set(relayctl.CODE_TO_USAGE))
|
|
only_cli = sorted(set(relayctl.CODE_TO_USAGE) - set(page_map))
|
|
check_eq(only_page, [], "keys the page knows but relayctl does not")
|
|
check_eq(only_cli, [], "keys relayctl knows but the page does not")
|
|
mismatched = sorted(k for k in set(page_map) & set(relayctl.CODE_TO_USAGE)
|
|
if page_map[k] != relayctl.CODE_TO_USAGE[k])
|
|
check_eq(mismatched, [], "same name, different usage id")
|
|
check_eq(page_map.get("Enter"), 0x28, "Enter agrees on 0x28")
|
|
check_eq(page_map.get("KeyA"), 0x04, "KeyA agrees on 0x04")
|
|
|
|
print("pasted text")
|
|
root = pathlib.Path(__file__).resolve().parent.parent
|
|
relay_src = (root / "tools" / "relayctl.py").read_text()
|
|
proto_src = (root / "src" / "proto.h").read_text()
|
|
text_src = (root / "src" / "text.h").read_text()
|
|
|
|
check_eq(relayctl.normalize_paste("a\r\nb"), "a\nb", "CRLF becomes one Enter")
|
|
check_eq(relayctl.normalize_paste("a\rb"), "a\nb", "a lone CR becomes Enter")
|
|
check_eq(relayctl.normalize_paste("a\nb"), "a\nb", "LF is left alone")
|
|
check_eq(relayctl.text_frame(0, "hi"), "P=0:6869", "the frame relayctl sends")
|
|
check_eq(relayctl.text_frame(250, "hi"), "P=250:6869", "the pace travels in decimal")
|
|
check_eq(relayctl.text_frame(0, "a"), "P=0:61", "one character")
|
|
|
|
# What the board's character map can type, read straight out of src/text.h: the
|
|
# switch spells out the punctuation and the ranges spell out the alphanumerics.
|
|
firmware = set()
|
|
for literal in re.findall(r"case '(\\.|[^'\\])':", text_src):
|
|
firmware.add(ord(literal.encode("latin-1").decode("unicode_escape")))
|
|
check("ch >= 'a' && ch <= 'z'" in text_src, "the lowercase range is still a range")
|
|
check("ch >= 'A' && ch <= 'Z'" in text_src, "the uppercase range is still a range")
|
|
check("ch >= '0' && ch <= '9'" in text_src, "the digit range is still a range")
|
|
firmware |= set(range(ord("a"), ord("z") + 1))
|
|
firmware |= set(range(ord("A"), ord("Z") + 1))
|
|
firmware |= set(range(ord("0"), ord("9") + 1))
|
|
|
|
expected = {9, 10, 13} | set(range(32, 127))
|
|
check_eq(sorted(firmware), sorted(expected),
|
|
"the board types exactly tab, newline, CR and printable ASCII")
|
|
|
|
# ... and the tool's validator must agree with it on every ASCII character, once the
|
|
# line endings are normalised. A character the tool waves through that the board then
|
|
# refuses would fail the whole paste at the board instead of at the terminal.
|
|
for code in range(128):
|
|
ch = chr(code)
|
|
tool_says_ok = not relayctl.untypeable(relayctl.normalize_paste(ch))
|
|
if tool_says_ok != (code in firmware):
|
|
check(False, f"code {code:#04x}: tool={tool_says_ok} board={code in firmware}")
|
|
check(True, "tool and board agree on all 128 ASCII codes")
|
|
|
|
for bad in ["\u2019", "\u2014", "\u00e9", "\u00a0", "\x00", "\x07", "\x7f"]:
|
|
check_eq(relayctl.untypeable("ok" + bad), [bad], f"refuses {bad!r}")
|
|
check_eq(relayctl.untypeable("line1\nline2\tcol\r"), ["\r"],
|
|
"CR is flagged before normalisation, and never sent")
|
|
|
|
# The three numbers have to be the board's numbers, or a paste dies on E8 halfway
|
|
# through because the tool split the text somewhere the firmware will not accept.
|
|
config_src = (root / "src" / "config.h").read_text()
|
|
|
|
|
|
def config_value(name):
|
|
# Decimal or hex: kMouseDeltaMax is 127 and kMouseButtonMask is 0x1F, and both are
|
|
# numbers a client has to agree with the firmware about.
|
|
found = re.search(rf"\b{name}\s*=\s*(0[xX][0-9a-fA-F]+|\d+)\s*;", config_src)
|
|
return int(found.group(1), 0) if found else None
|
|
|
|
|
|
check_eq(relayctl.TEXT_FRAME_MAX, config_value("kTextFrameMax"),
|
|
"relayctl TEXT_FRAME_MAX == kTextFrameMax")
|
|
check_eq(relayctl.MAX_PASTE_CHARS, config_value("kTextJobCap"),
|
|
"relayctl MAX_PASTE_CHARS == kTextJobCap")
|
|
check_eq(relayctl.DEFAULT_TEXT_PACE_MS, config_value("kTextPeriodMsDefault"),
|
|
"relayctl DEFAULT_TEXT_PACE_MS == kTextPeriodMsDefault")
|
|
page_src = (root / "src" / "web_page.h").read_text()
|
|
check_eq(int(re.search(r"const TEXT_FRAME_MAX = (\d+);", page_src).group(1)),
|
|
config_value("kTextFrameMax"), "page TEXT_FRAME_MAX == kTextFrameMax")
|
|
check_eq(int(re.search(r"const TEXT_JOB_MAX = (\d+);", page_src).group(1)),
|
|
config_value("kTextJobCap"), "page TEXT_JOB_MAX == kTextJobCap")
|
|
check(config_value("kTextFrameMax") * 2 + 7 <= config_value("kWsMessageCap"),
|
|
"the largest text frame fits one WebSocket message")
|
|
|
|
# _drain answered the board's ping with two methods that do not exist, which only
|
|
# surfaced on a session longer than the 15 s ping interval. A paste is such a session.
|
|
check("_send_frame" not in relay_src, "_drain no longer calls the missing _send_frame")
|
|
check("_check_error" not in relay_src, "_drain no longer calls the missing _check_error")
|
|
drained = relayctl.Relay._drain
|
|
src_lines = inspect.getsource(drained)
|
|
check("OP_PONG" in src_lines, "_drain answers a ping")
|
|
check("raise RelayError" in src_lines, "_drain still surfaces an error frame")
|
|
|
|
print("mouse")
|
|
check_eq(relayctl.mouse_frame(0x01, 30, -4), "M=01:30:-4:0:0",
|
|
"a mouse frame is the button byte then four movements")
|
|
check_eq(relayctl.mouse_frame(0), "M=00:0:0:0:0",
|
|
"and all five fields are always written -- the firmware parses exactly five")
|
|
check_eq(relayctl.mouse_frame(0x1f, -127, 127, -127, 127), "M=1f:-127:127:-127:127",
|
|
"the full signed range")
|
|
|
|
for bad, why in (((0x20, 0, 0, 0, 0), "a sixth button bit has nowhere to go"),
|
|
((0, 128, 0, 0, 0), "128 is more than one report"),
|
|
((0, -128, 0, 0, 0), "and so is -128"),
|
|
((0, 0, 0, 128, 0), "the wheel is one byte too")):
|
|
try:
|
|
relayctl.mouse_frame(*bad)
|
|
check(False, f"mouse_frame should refuse: {why}")
|
|
except relayctl.RelayError:
|
|
check(True, f"mouse_frame refuses: {why}")
|
|
|
|
# A movement bigger than one report is split here, exactly as the page splits it, and the
|
|
# split is exact -- clamping it would move the cursor half as far as was asked and say
|
|
# nothing about it.
|
|
for dx, dy in ((300, 0), (0, -260), (300, -260), (127, 127), (1, -1), (-500, 500)):
|
|
parts = relayctl.split_moves(dx, dy)
|
|
check_eq(sum(p[0] for p in parts), dx, f"the X split of {dx},{dy} adds up exactly")
|
|
check_eq(sum(p[1] for p in parts), dy, f"and the Y of {dx},{dy}")
|
|
check(all(abs(p[0]) <= relayctl.MOUSE_DELTA_MAX and abs(p[1]) <= relayctl.MOUSE_DELTA_MAX
|
|
for p in parts), f"and no part of {dx},{dy} exceeds one report")
|
|
|
|
check_eq(relayctl.mouse_frames("click", [], "left"),
|
|
[(0x01, 0, 0, 0, 0), (0, 0, 0, 0, 0)],
|
|
"a click is the button restated down then restated up")
|
|
check_eq(relayctl.mouse_frames("release", []), [(0, 0, 0, 0, 0)],
|
|
"release restates the whole button byte as nothing")
|
|
drag = relayctl.mouse_frames("drag", ["300", "0"], "right")
|
|
check_eq(drag[0], (0x02, 0, 0, 0, 0), "a drag presses first")
|
|
check(drag[-1] == (0, 0, 0, 0, 0), "and releases last")
|
|
check(all(f[0] == 0x02 for f in drag[1:-1]),
|
|
"and every movement in between carries the button, so it is a drag and not a move")
|
|
check_eq(sum(f[1] for f in drag), 300, "and the drag travels the whole distance")
|
|
check_eq(relayctl.mouse_frames("wheel", ["-3"]), [(0, 0, 0, -3, 0)],
|
|
"wheel notches go where the report's wheel byte is")
|
|
for bad_args, why in ((("move", ["1"], "left"), "move needs dx and dy"),
|
|
(("wheel", [], "left"), "wheel needs a count"),
|
|
(("move", ["1", "2"], "thumb"), "an unknown button is refused"),
|
|
(("wiggle", [], "left"), "an unknown action is refused")):
|
|
try:
|
|
relayctl.mouse_frames(*bad_args)
|
|
check(False, f"mouse_frames should refuse: {why}")
|
|
except (relayctl.RelayError, ValueError):
|
|
check(True, f"mouse_frames refuses: {why}")
|
|
|
|
# Every frame this tool can build must be one the firmware parses. The shape is asserted
|
|
# rather than a reimplementation of the parser, because the parser itself is tested
|
|
# against the same frames in tests/test_proto.cpp.
|
|
mframe = re.compile(r"^M=[0-9a-f]{2}:-?\d{1,3}:-?\d{1,3}:-?\d{1,3}:-?\d{1,3}$")
|
|
built = [relayctl.mouse_frame(*f)
|
|
for f in relayctl.mouse_frames("drag", ["500", "-500", "4", "-4"], "middle")]
|
|
check(all(mframe.match(f) for f in built), f"every built frame parses ({len(built)})")
|
|
check(all(len(f) <= config_value("kWsMessageCap") for f in built),
|
|
"and every one fits in one WebSocket message")
|
|
|
|
# The numbers are the board's. If they drift, the board starts answering E6 to frames
|
|
# this tool believed were legal.
|
|
check_eq(relayctl.MOUSE_DELTA_MAX, config_value("kMouseDeltaMax"),
|
|
"relayctl MOUSE_DELTA_MAX == kMouseDeltaMax")
|
|
check_eq(relayctl.MOUSE_BUTTON_MASK, config_value("kMouseButtonMask"),
|
|
"relayctl MOUSE_BUTTON_MASK == kMouseButtonMask")
|
|
check_eq(int(re.search(r"const MOUSE_DELTA_MAX = (\d+);", page_src).group(1)),
|
|
config_value("kMouseDeltaMax"), "page MOUSE_DELTA_MAX == kMouseDeltaMax")
|
|
check_eq(int(re.search(r"const MOUSE_BUTTON_MASK = (0x[0-9a-f]+);", page_src).group(1), 16),
|
|
config_value("kMouseButtonMask"), "page MOUSE_BUTTON_MASK == kMouseButtonMask")
|
|
check_eq(int(re.search(r"const MOUSE_FRAME_MS = (\d+);", page_src).group(1)),
|
|
relayctl.MOUSE_FRAME_MS, "the page and the CLI pace mouse frames identically")
|
|
|
|
# Button bits, agreed between the page and this tool even though the page keys them by
|
|
# DOM number and this keys them by name. Getting one wrong is a clean frame that is the
|
|
# wrong button -- the Alt+Tab failure again, one layer down.
|
|
check_eq({name: bits for name, bits in relayctl.MOUSE_BUTTONS.items()},
|
|
{"left": 0x01, "right": 0x02, "middle": 0x04, "back": 0x08, "forward": 0x10},
|
|
"the button bits are the report descriptor's order")
|
|
check("0: 0x01, 2: 0x02, 1: 0x04, 3: 0x08, 4: 0x10" in page_src,
|
|
"the page maps DOM button numbers to those same bits (DOM 1 is middle, 2 is right)")
|
|
|
|
print("protocol strings")
|
|
|
|
check_eq("S=%02x:%s" % (0x0A, "04,28"), "S=0a:04,28", "state frame format")
|
|
check_eq("Y=%02x:%s" % (0x05, "4c"), "Y=05:4c", "chord frame format")
|
|
check_eq("T" + "hunter2", "Thunter2", "auth frame format")
|
|
|
|
# The single-character commands relayctl emits must be ones proto.h accepts, and
|
|
# proto.h's own documented list must still mention them: a change on either side
|
|
# has to be a deliberate edit in two places.
|
|
for command in ("A1", "A0", "R", "H", "K"):
|
|
check(f'"{command}"' in relay_src, f'relayctl sends "{command}"')
|
|
for command in ("'A'", "'R'", "'H'", "'K'", "'T'", "'S'", "'Y'", "'C'", "'P'", "'M'"):
|
|
check(command in proto_src, f"proto.h parses {command}")
|
|
check("case 'M':" in proto_src, "proto.h routes M= to the mouse parser")
|
|
|
|
# The two mouse error codes have to be in the README's list. A client that cannot look up
|
|
# what E11 means cannot tell the operator why the pointer is not moving, and this suite is
|
|
# the only place that notices when the code and the document stop agreeing.
|
|
readme_src = (root / "README.md").read_text()
|
|
for code, where in (("10 the mouse interface is off", "proto.h"),
|
|
("11 the target asked for boot protocol", "proto.h")):
|
|
check(code in proto_src, f"proto.h's error list documents {code}")
|
|
check(code.split()[0] in readme_src,
|
|
f"README's error list has code {code.split()[0]}")
|
|
check("mouse interface is off" in readme_src and "boot protocol" in readme_src,
|
|
"README explains what both mouse refusals mean, not just their numbers")
|
|
check("`M=<btn>`" in readme_src or "M=<btn>" in readme_src or "`M=`" in readme_src,
|
|
"README's wire protocol table lists the M= command")
|
|
check("if (c == 'Y' && n == 0 && out.mods == 0)" in proto_src,
|
|
"proto.h still requires a chord to press something")
|
|
|
|
print("the fail-safe wiring")
|
|
# Source pins, and deliberately honest about being source pins: control_server.cpp and
|
|
# main.cpp need a socket and a board, so nothing here runs them. What they were caught
|
|
# doing is invisible to a compiler and to every behavioural test on the host, and each of
|
|
# these three is a sentence that a later edit can quietly undo.
|
|
main_src = (root / "src" / "main.cpp").read_text()
|
|
server_src = (root / "src" / "control_server.cpp").read_text()
|
|
|
|
# The backstop must not ask whether a controller is linked: a controller that has gone
|
|
# away is exactly the case where linked() is false, which made the silence timer
|
|
# uncallable in the one situation it exists for -- the board sat armed with a key held
|
|
# forever, and read-only /status traffic kept it that way.
|
|
check("if (armed && control.controllerSeen() &&" in main_src,
|
|
"loop()'s silence disarm runs off controllerSeen(), not linked()")
|
|
check("if (armed && control.linked()" not in main_src,
|
|
"nothing gates an armed-and-silent disarm on the link still being there")
|
|
|
|
# Accepting a TCP connection is not the operator being alive.
|
|
accept = server_src[server_src.index("void ControlServer::acceptNew"):
|
|
server_src.index("size_t ControlServer::readSome")]
|
|
# Code lines only -- the function carries a comment explaining why it does NOT stamp the
|
|
# clock, and a check that flagged the comment would be worse than no check at all.
|
|
accept_code = "\n".join(l for l in accept.splitlines()
|
|
if not l.strip().startswith(("//", "*", "/*")))
|
|
check("_lastInboundMs" not in accept_code,
|
|
"acceptNew() does not refresh the controller-silence clock")
|
|
|
|
# And closing a read-only /status request must not refresh it either -- the first fix
|
|
# stamped it on every close, which held a dead controller's armed state open exactly as
|
|
# effectively as the original bug did.
|
|
close_fn = server_src[server_src.index("void ControlServer::closeLink"):
|
|
server_src.index("}", server_src.index("_onLinkChange(_ctx, false)"))]
|
|
check("if (wasLinked) _lastInboundMs = millis();" in close_fn,
|
|
"closeLink() stamps the silence clock only when a controller link is what closed")
|
|
check("const bool wasLinked = _phase == Phase::WebSocket;" in close_fn,
|
|
"closeLink() treats anything that reached the WebSocket phase as able to leave "
|
|
"keys down, not only what is still marked authenticated")
|
|
|
|
print("one routing path, two transports")
|
|
# Phase 02 task 04 moved the TLS session underneath the routing instead of beside it. The
|
|
# claims are structural, so they are pinned as source sentences: what they defend is a
|
|
# second path that can disagree, and nothing that runs on a host can observe two paths
|
|
# agreeing until one of them is wrong.
|
|
conn_src = (root / "src" / "conn.h").read_text()
|
|
ws_src = (root / "src" / "ws_codec.h").read_text()
|
|
|
|
# readSome()'s rule, carried over from the TLS task and now the only clock-stamping read:
|
|
# bytes count as "the operator is alive" only from an authenticated controller on the
|
|
# WebSocket. Over TLS a handshake, a /status poll and a viewer refresh all look identical
|
|
# to the transport, so if this condition ever loosens to "bytes arrived", anything on the
|
|
# network can hold a dead controller's armed state open -- bug 10's shape again.
|
|
read_fn = server_src[server_src.index("size_t ControlServer::readSome"):
|
|
server_src.index("\n}", server_src.index("size_t ControlServer::readSome"))]
|
|
check("if (_phase == Phase::WebSocket && _authed) _lastInboundMs = millis();" in read_fn,
|
|
"readSome() stamps the silence clock only for an authenticated controller on the "
|
|
"WebSocket, whatever transport the bytes came over")
|
|
check("_lastInboundMs" in read_fn, "readSome() is still the code that notices inbound bytes")
|
|
# ... and it is the ONLY inbound path that may. Three writers of that clock would mean a
|
|
# route that can keep the board armed: readSome, and closeLink stamping on the way out.
|
|
stamps = [l.strip() for l in server_src.splitlines() if "_lastInboundMs = millis()" in l]
|
|
check_eq(len(stamps), 3,
|
|
"exactly three writers of the controller-silence clock: auth, readSome, closeLink")
|
|
check(any("the controller exists from here" in l for l in stamps),
|
|
"one of them is the authentication itself")
|
|
# The spare slot must not be able to reach that clock even by accident: it reads the Conn
|
|
# directly, never through readSome().
|
|
viewer_fn = server_src[server_src.index("void ControlServer::serveViewer"):
|
|
server_src.index("void ControlServer::stopViewer")]
|
|
# Code lines only: the function carries a comment explaining why it does NOT go through
|
|
# readSome(), and a check that flagged the comment would be worse than no check at all.
|
|
viewer_code = "\n".join(l for l in viewer_fn.splitlines()
|
|
if not l.strip().startswith(("//", "*", "/*")))
|
|
check("readSome" not in viewer_code,
|
|
"the read-only slot reads its Conn directly, so it cannot refresh the controller clock")
|
|
|
|
# One HTTP response writer. The task's completion criterion is a grep that comes up empty,
|
|
# so this is that grep, as a check: exactly one formatter in src/, and the only literal
|
|
# response status line outside it is the 101 protocol switch (which is not a response).
|
|
writers = sorted(p.name for p in (root / "src").glob("*")
|
|
if '"HTTP/1.1 ' in p.read_text(errors="replace"))
|
|
check_eq(writers, ["conn.h", "control_server.cpp"],
|
|
"only conn.h's formatter and the 101 upgrade line write an HTTP status line")
|
|
check_eq(server_src.count('"HTTP/1.1 '), 1,
|
|
"control_server.cpp holds exactly one, the WebSocket 101 -- not a second response writer")
|
|
check("respondAndClose(rsp)" in server_src and "respondViewer(rsp)" in server_src,
|
|
"both slots queue a built response through the same pair of writers")
|
|
# serveReadonly is the single read-only answer, reachable from both listeners/slots.
|
|
check_eq(server_src.count("serveReadonly(req, rsp,"), 2,
|
|
"the page/status routing is reached from the controller slot and the spare, once each")
|
|
check("void ControlServer::serveReadonly" in server_src,
|
|
"and it is defined once")
|
|
check_eq(server_src.count("plaintextRoute("), 1,
|
|
"the :80 answer table is consulted in exactly one place, inside serveReadonly")
|
|
# Both slots go through the transport view; nothing in the routing consults the port.
|
|
check("_ctl.read(" in conn_src or "_ctl.read(" in server_src or "_plain->read" in conn_src,
|
|
"the controller slot reads through kr::Conn")
|
|
check("_view.read(" in server_src, "the spare slot reads through kr::Conn too")
|
|
check("_server.hasClient()" in server_src and "::accept(_tlsListen" in server_src,
|
|
"both listeners still feed one accept decision")
|
|
|
|
# The :80 policy itself is pure (kr::plaintextRoute, walked in tests/test_conn.cpp); these
|
|
# pins keep the firmware from re-deciding it locally.
|
|
check("PlainRoute::UpgradeRequired" in conn_src and "PlainRoute::Redirect" in conn_src,
|
|
"conn.h spells out the redirect and the 426 as data, not as scattered ifs")
|
|
check("namespace kr" in conn_src, "conn.h is in namespace kr, like every other module")
|
|
check("token=" in conn_src and "The value is never copied" in conn_src,
|
|
"the token-in-URL test looks for the key without copying the passphrase")
|
|
# The Host header had to be parsed to build a Location that keeps the operator's address.
|
|
check('ciEquals(ls, "host:", 5)' in ws_src, "ws_codec parses Host: for the redirect")
|
|
check("char host[" in ws_src, "and carries it on Request")
|
|
|
|
# What the terminal client says when the board wants TLS. A refusal that reads like a
|
|
# passphrase problem sends the operator re-typing a correct secret until the board locks it.
|
|
check("board refuses plaintext" in relayctl.handshake_refusal("HTTP/1.1 426 Upgrade Required"),
|
|
"426 is named as a TLS refusal")
|
|
check("passphrase" not in relayctl.handshake_refusal("HTTP/1.1 426 Upgrade Required").lower(),
|
|
"and the 426 wording does not blame the passphrase")
|
|
check("passphrase" in relayctl.handshake_refusal("HTTP/1.1 401 Unauthorized"),
|
|
"401 still says what it always said")
|
|
check(relayctl.handshake_refusal("HTTP/1.1 503 Service Unavailable")
|
|
!= relayctl.handshake_refusal("HTTP/1.1 426 Upgrade Required"),
|
|
"'busy' and 'this board wants TLS' are never the same sentence")
|
|
check("https" in relayctl.handshake_refusal("HTTP/1.1 301 Moved Permanently"),
|
|
"a redirect on :80 is reported as the port policy it is")
|
|
for line, want in (("HTTP/1.1 418 Teapot", "refused"), ("garbage", "refused")):
|
|
check_eq(relayctl.handshake_refusal(line), want, f"unknown status {line!r}")
|
|
|
|
# The page the transport sends: the byte count tests/test_conn.cpp drains through the
|
|
# chunk arithmetic has to be the page that is actually in the tree. The page is a
|
|
# chain of raw-string chunks joined by one #if KR_TLS build marker (task 05), so
|
|
# "the page" is the chunks plus exactly one marker branch -- and both branches are
|
|
# asserted to be the same length, which is what keeps every pinned count (this one,
|
|
# the gate's, the device's `wc -c`) build-independent. That is a test, not a hope.
|
|
def served_page(src, tls=True):
|
|
"""The bytes one build serves: the raw-string chunks concatenated around the
|
|
marker branch that build takes. None when the layout is not chunks joined by
|
|
markers, so a restructured page fails loudly instead of counting half of it."""
|
|
chunks = re.findall(r'R"HTML\((.*?)\)HTML"', src, re.S)
|
|
marks = re.findall(r'#if KR_TLS\n\s*"([^"]*)"\n#else\n\s*"([^"]*)"\n#endif', src)
|
|
if len(chunks) < 1 or len(marks) != len(chunks) - 1:
|
|
return None
|
|
out = ""
|
|
for i, chunk in enumerate(chunks):
|
|
out += chunk
|
|
if i < len(marks):
|
|
out += marks[i][0] if tls else marks[i][1]
|
|
return out
|
|
|
|
|
|
page_tls = served_page(page_src, True)
|
|
page_plain = served_page(page_src, False)
|
|
check(page_tls is not None and page_plain is not None,
|
|
"every join between the page's raw-string chunks is one #if KR_TLS marker")
|
|
check(all(len(a) == len(b) for a, b in
|
|
re.findall(r'#if KR_TLS\n\s*"([^"]*)"\n#else\n\s*"([^"]*)"\n#endif',
|
|
page_src)),
|
|
"both branches of every build marker are the same length, so the served page "
|
|
"is the same size whichever build flashes")
|
|
check(page_tls is not None and "var KR_TLS_BUILD=1;" in page_tls,
|
|
"the TLS build's page carries its marker")
|
|
check("var KR_TLS_BUILD=0;" in page_plain,
|
|
"and the -D KR_TLS=0 rollback page carries its own")
|
|
check_eq(len(page_tls.encode("utf-8")), len(page_plain.encode("utf-8")),
|
|
"the two builds serve byte-count-equal pages")
|
|
pinned_page_bytes = int(re.search(r"kServedPageBytes = (\d+)",
|
|
(root / "tests" / "test_conn.cpp").read_text()).group(1))
|
|
check_eq(len(page_tls.encode("utf-8")), pinned_page_bytes,
|
|
"the page the chunk arithmetic is tested against is the page in web_page.h")
|
|
|
|
# The page's only scheme choice is its own origin; a TLS board never has an http: origin
|
|
# because :80 answers a redirect (pinned above, walked in test_conn).
|
|
check(page_src.count('"wss://"') == 1 and page_src.count('"ws://"') == 1,
|
|
"the page derives the socket scheme from location.protocol, in exactly one place")
|
|
check('location.protocol === "https:" ? "wss://" : "ws://"' in page_src,
|
|
"https: origin gives wss://, and ws:// is reachable only on a KR_TLS=0 build")
|
|
# The guard the other way round (task 05): a TLS build's page on an http: origin
|
|
# refuses to connect and refuses the passphrase. The behaviour is walked in
|
|
# tests/test_page.js; these pins keep the refusal from being edited into a warning.
|
|
check("function servedInClear()" in page_src and "if (servedInClear())" in page_src,
|
|
"the page asks whether it arrived in the clear, in connect() and at load")
|
|
check("this page arrived in the clear; reload over https" in page_src,
|
|
"and it says why, in words the operator can act on")
|
|
check('$("token").disabled = true' in page_src,
|
|
"a clear-served page will not even take passphrase input")
|
|
check("paintChannel(s)" in page_src and 'id="certFp"' in page_src
|
|
and 'id="pTls"' in page_src,
|
|
"the status frame's tls/fp pair is rendered: a pill and a certificate line")
|
|
|
|
print("the client side of pinning")
|
|
# The half of phase 02 the operator types. None of this may guess: no TLS argument
|
|
# means no connection, a wrong fingerprint means nothing was sent, and the only
|
|
# path to the plaintext port is the flag that says it knows what it is doing.
|
|
import contextlib # noqa: E402
|
|
import hashlib # noqa: E402
|
|
import io # noqa: E402
|
|
import os # noqa: E402
|
|
import socket as socketmod # noqa: E402
|
|
|
|
# --- the decision functions, walked directly ---------------------------------
|
|
check_eq(relayctl.select_transport("a" * 64, "", False, False), "pin",
|
|
"a fingerprint pins")
|
|
check_eq(relayctl.select_transport("", "/tmp/ca.pem", False, False), "ca",
|
|
"a CA file verifies a chain")
|
|
check_eq(relayctl.select_transport("", "", True, False), "insecure",
|
|
"--insecure is a decision, not a default")
|
|
check_eq(relayctl.select_transport("", "", False, True), "plain",
|
|
"--plain is the rollback path")
|
|
for bad, why in ((("", "", False, False), "no trust argument at all"),
|
|
(("a" * 64, "/tmp/ca.pem", False, False), "a pin and a CA at once"),
|
|
(("", "", True, True), "--insecure together with --plain"),
|
|
(("a" * 64, "", False, True), "a pin on the rollback path")):
|
|
try:
|
|
relayctl.select_transport(*bad)
|
|
check(False, f"select_transport should refuse: {why}")
|
|
except relayctl.RelayError as exc:
|
|
check(True, f"select_transport refuses: {why}")
|
|
if why == "no trust argument at all":
|
|
for option in ("--fingerprint", "--ca", "--insecure", "--plain"):
|
|
check(option in str(exc),
|
|
f"the refusal names {option} as a way forward")
|
|
|
|
check_eq(relayctl.resolve_port("pin", None), 443, "a pinned run defaults to :443")
|
|
check_eq(relayctl.resolve_port("ca", None), 443, "a verified run defaults to :443")
|
|
check_eq(relayctl.resolve_port("insecure", None), 443,
|
|
"even an unverified TLS run defaults to the TLS port")
|
|
check_eq(relayctl.resolve_port("plain", None), 80,
|
|
"--plain is the only path to :80")
|
|
check_eq(relayctl.resolve_port("pin", 8443), 8443, "an explicit --port wins")
|
|
|
|
check_eq(relayctl.normalize_fingerprint("A" * 64), "a" * 64,
|
|
"a bare hex fingerprint normalises to lowercase")
|
|
check_eq(relayctl.normalize_fingerprint(":".join("a1" for _ in range(32))), "a1" * 32,
|
|
"OpenSSL's colon pairs are accepted")
|
|
check_eq(relayctl.normalize_fingerprint(" " + "ab" * 32 + "\n"), "ab" * 32,
|
|
"and so is the whitespace a serial console wraps in")
|
|
for bad in ("", "deadbeef", "zz" * 32, "ab" * 31, "ab" * 32 + "0"):
|
|
try:
|
|
relayctl.normalize_fingerprint(bad)
|
|
check(False, f"a malformed fingerprint is refused: {bad[:12]!r}")
|
|
except relayctl.RelayError:
|
|
check(True, f"malformed fingerprint refused: {bad[:12]!r}")
|
|
|
|
# --- the connection itself, against a fake board ------------------------------
|
|
FAKE_DER = b"-----BEGIN FAKE CERTIFICATE----- der-bytes pinning never sees"
|
|
|
|
|
|
class FakeBoardSock:
|
|
"""Records every byte this side sends; answers the upgrade and `T<token>`.
|
|
|
|
The recorded frame list is the point: "the pin aborted before T<token>" can
|
|
only be shown by what did NOT go out.
|
|
"""
|
|
|
|
def __init__(self, host, port, close_early=False, bad_accept=False, flood=False):
|
|
self.host, self.port = host, port
|
|
self.bad_accept = bad_accept # answer the upgrade with a wrong accept hash
|
|
self.flood = flood # answer with headers that never terminate
|
|
self.sent = [] # every sendall payload, in order
|
|
self.inbox = bytearray()
|
|
self.upgraded = False
|
|
self.post = bytearray() # bytes after the 101 (client WS frames)
|
|
self.closed = False
|
|
self.timeout = None
|
|
self.close_early = close_early # answer the upgrade with a clean EOF
|
|
|
|
def settimeout(self, t):
|
|
self.timeout = t
|
|
|
|
def sendall(self, data):
|
|
data = bytes(data)
|
|
self.sent.append(data)
|
|
if self.close_early:
|
|
return # received, but the board answers nothing
|
|
if not self.upgraded:
|
|
blob = b"".join(self.sent)
|
|
if b"\r\n\r\n" in blob:
|
|
m = re.search(rb"Sec-WebSocket-Key: (\S+)", blob)
|
|
if self.flood:
|
|
self.inbox += b"HTTP/1.1 101 " + b"x" * 20000
|
|
self.upgraded = True
|
|
return
|
|
accept = (b"not-the-accept" if self.bad_accept
|
|
else relayctl.expect_accept(m.group(1).decode()).encode())
|
|
self.inbox += (b"HTTP/1.1 101 Switching Protocols\r\n"
|
|
b"Sec-WebSocket-Accept: " + accept + b"\r\n\r\n")
|
|
self.upgraded = True
|
|
return
|
|
self.post += data
|
|
while len(self.post) >= 2:
|
|
length = self.post[1] & 0x7F
|
|
key = self.post[2:6]
|
|
payload = bytes(b ^ key[i & 3] for i, b in enumerate(self.post[6:6 + length]))
|
|
del self.post[:6 + length]
|
|
if payload.startswith(b"T"):
|
|
self.inbox += relayctl_server_frame(relayctl.OP_TEXT, b"OK")
|
|
|
|
def recv(self, n):
|
|
if self.inbox:
|
|
data = bytes(self.inbox[:n])
|
|
del self.inbox[:n]
|
|
return data
|
|
if self.close_early:
|
|
return b"" # clean EOF: the board hung up mid-handshake
|
|
raise socketmod.timeout("fake board: no more bytes right now")
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def relayctl_server_frame(opcode, payload):
|
|
out = bytearray([0x80 | opcode, len(payload)])
|
|
out += payload
|
|
return bytes(out)
|
|
|
|
|
|
class FakeSsl:
|
|
"""Stand-in for the ssl module: real policy code runs against these knobs, and
|
|
wrap_socket returns the same socket with a getpeercert of our choosing."""
|
|
|
|
PROTOCOL_TLS_CLIENT = "tls-client"
|
|
CERT_NONE = 0
|
|
CERT_REQUIRED = 2
|
|
|
|
class SSLError(Exception):
|
|
pass
|
|
|
|
def __init__(self, der=FAKE_DER, wrap_error=None, empty_der=False):
|
|
self.der = der
|
|
self.wrap_error = wrap_error # raised by wrap_socket, simulating a failed TLS
|
|
self.empty_der = empty_der # peer completes TLS but shows no certificate
|
|
self.contexts = [] # (kind, ctx) for every context built
|
|
|
|
def SSLContext(self, protocol):
|
|
ctx = FakeSsl.Ctx(self, protocol)
|
|
self.contexts.append(("manual", ctx))
|
|
return ctx
|
|
|
|
def create_default_context(self, cafile=None):
|
|
ctx = FakeSsl.Ctx(self, "default")
|
|
ctx.cafile = cafile
|
|
self.contexts.append(("default", ctx))
|
|
return ctx
|
|
|
|
class Ctx:
|
|
def __init__(self, owner, protocol):
|
|
self.owner, self.protocol = owner, protocol
|
|
self.check_hostname = None
|
|
self.verify_mode = None
|
|
self.cafile = None
|
|
self.server_hostname = None
|
|
|
|
def wrap_socket(self, sock, server_hostname=None):
|
|
self.server_hostname = server_hostname
|
|
if self.owner.wrap_error is not None:
|
|
raise self.owner.wrap_error
|
|
der = b"" if self.owner.empty_der else self.owner.der
|
|
sock.getpeercert = lambda binary_form=False: der if binary_form else {}
|
|
return sock
|
|
|
|
|
|
def connect_with(fingerprint="", ca="", insecure=False, plain=False, port=None,
|
|
der=FAKE_DER, token="sekrit", wrap_error=None, close_early=False,
|
|
empty_der=False, bad_accept=False, flood=False,
|
|
real_create=socketmod.create_connection, real_ssl=relayctl.ssl):
|
|
fake = FakeSsl(der, wrap_error=wrap_error, empty_der=empty_der)
|
|
socks = []
|
|
|
|
def fake_create(addr, timeout):
|
|
s = FakeBoardSock(addr[0], addr[1], close_early=close_early,
|
|
bad_accept=bad_accept, flood=flood)
|
|
socks.append(s)
|
|
return s
|
|
|
|
relayctl.ssl = fake
|
|
relayctl.socket.create_connection = fake_create
|
|
stderr = io.StringIO()
|
|
relay, err = None, None
|
|
try:
|
|
with contextlib.redirect_stderr(stderr):
|
|
relay = relayctl.Relay("keyrelay.local", port, token, 1.0,
|
|
fingerprint=fingerprint, ca=ca,
|
|
insecure=insecure, plain=plain)
|
|
except relayctl.RelayError as exc:
|
|
err = exc
|
|
finally:
|
|
relayctl.ssl = real_ssl
|
|
relayctl.socket.create_connection = real_create
|
|
return relay, err, socks, fake, stderr.getvalue()
|
|
|
|
|
|
def client_frame_payloads(sock):
|
|
"""Decode the masked client frames this side sent, skipping the HTTP upgrade."""
|
|
blob = b"".join(sock.sent)
|
|
tail = blob.split(b"\r\n\r\n", 1)[1] if b"\r\n\r\n" in blob else b""
|
|
out = []
|
|
while len(tail) >= 2:
|
|
length = tail[1] & 0x7F
|
|
key = tail[2:6]
|
|
out.append(bytes(b ^ key[i & 3] for i, b in enumerate(tail[6:6 + length])))
|
|
tail = tail[6 + length:]
|
|
return out
|
|
|
|
# A wrong fingerprint aborts BEFORE a single frame -- above all before T<token>.
|
|
_wrong = "0" * 64
|
|
relay, err, socks, fake, warns = connect_with(fingerprint=_wrong)
|
|
check(isinstance(err, relayctl.RelayError), "a wrong fingerprint raises")
|
|
check(err is not None and "fingerprint mismatch" in str(err),
|
|
"and says exactly what mismatched")
|
|
check(socks and socks[0].sent == [],
|
|
"nothing at all went out -- not the upgrade, and certainly not T<token>")
|
|
|
|
# The right fingerprint (in any sane spelling) connects, and the passphrase frame
|
|
# is the first thing on the encrypted wire.
|
|
_right = hashlib.sha256(FAKE_DER).hexdigest()
|
|
relay, err, socks, fake, warns = connect_with(fingerprint=_right)
|
|
check(err is None and relay is not None, f"the pinned fingerprint connects ({err})")
|
|
check(relay is not None and relay.mode == "pin", "the mode is the pin")
|
|
if socks:
|
|
frames = client_frame_payloads(socks[0])
|
|
check(frames and frames[0] == b"Tsekrit",
|
|
"the first frame on a pinned session is the authentication")
|
|
check(b"Tsekrit" not in b"".join(socks[0].sent),
|
|
"the token never travels as plaintext bytes -- it is only in masked frames")
|
|
check(socks and socks[0].port == 443, "pinning dials :443")
|
|
relay2, err2, socks2, _, _ = connect_with(
|
|
fingerprint=":".join(_right[i:i + 2] for i in range(0, 64, 2)).upper())
|
|
check(err2 is None, "a colon-separated uppercase pin is the same pin")
|
|
|
|
# TLS failures, in the two shapes the operator can actually act on.
|
|
relay, err, socks, fake, warns = connect_with(
|
|
fingerprint=_right, wrap_error=FakeSsl.SSLError("handshake refused"))
|
|
check(err is not None and "TLS handshake with keyrelay.local failed" in str(err),
|
|
"a failed TLS handshake is named as that")
|
|
check(err is not None and "426" in str(err),
|
|
"and it names the plaintext-port mistake without ever proposing to make it")
|
|
check(socks and socks[0].closed, "and the socket is closed on the way out")
|
|
|
|
relay, err, socks, fake, warns = connect_with(
|
|
ca="/tmp/ca.pem", wrap_error=FakeSsl.SSLError("certificate verify failed"))
|
|
check(err is not None and "failed verification" in str(err),
|
|
"a --ca verification failure is not collapsed into 'handshake failed'")
|
|
check(err is not None and "nvs erase" in str(err),
|
|
"and it says the two things a verify failure can mean")
|
|
|
|
relay, err, socks, fake, warns = connect_with(fingerprint=_right, empty_der=True)
|
|
check(err is not None and "no certificate" in str(err),
|
|
"a TLS peer that shows no certificate is refused, not hashed as empty")
|
|
check(socks and socks[0].closed, "and that socket is closed too")
|
|
|
|
# --plain against a board that hangs up instead of speaking WebSocket.
|
|
relay, err, socks, fake, warns = connect_with(plain=True, close_early=True)
|
|
check(err is not None and "server closed during the handshake" in str(err),
|
|
"a closed handshake is reported for what it is")
|
|
|
|
relay, err, socks, fake, warns = connect_with(plain=True, bad_accept=True)
|
|
check(err is not None and "Sec-WebSocket-Accept" in str(err),
|
|
"a wrong accept hash refuses -- the RFC 6455 check is not decoration")
|
|
|
|
relay, err, socks, fake, warns = connect_with(plain=True, flood=True)
|
|
check(err is not None and "handshake response too large" in str(err),
|
|
"and a header block that never ends is bounded, not buffered forever")
|
|
|
|
# --ca verifies with OpenSSL's own machinery: hostname checking ON, chain REQUIRED.
|
|
relay, err, socks, fake, warns = connect_with(ca="/tmp/board-ca.pem")
|
|
check(err is None, f"--ca connects ({err})")
|
|
kinds = [k for k, _ in fake.contexts]
|
|
check(kinds == ["default"], "--ca builds its context with create_default_context")
|
|
ctx = fake.contexts[0][1]
|
|
check_eq(ctx.cafile, "/tmp/board-ca.pem", "and loads the file the operator named")
|
|
check_eq(ctx.check_hostname, True, "--ca keeps hostname checking on")
|
|
check_eq(ctx.verify_mode, 2, "and requires the chain")
|
|
check(ctx.server_hostname == "keyrelay.local",
|
|
"the hostname travels -- SNI and the check share one value")
|
|
|
|
# --insecure: connects, warns, and warns on EVERY run (it is stderr, not a memory).
|
|
relay, err, socks, fake, warns = connect_with(insecure=True)
|
|
check(err is None, "--insecure connects")
|
|
check("--insecure" in warns and "not verifying" in warns,
|
|
"--insecure is loud about what it skipped")
|
|
check_eq(socks[0].port, 443, "and still uses the TLS port")
|
|
_, _, _, fake_p, warns_p = connect_with(plain=True)
|
|
check(fake_p.contexts == [], "--plain builds no TLS context at all")
|
|
check(warns_p == "", "--plain warns about nothing it did not skip")
|
|
|
|
# --plain is the only way to :80, and it is explicit every time.
|
|
relay, err, socks, fake, warns = connect_with(plain=True)
|
|
check(err is None and socks[0].port == 80, "--plain reaches :80")
|
|
frames = client_frame_payloads(socks[0])
|
|
check(frames and frames[0] == b"Tsekrit",
|
|
"the rollback session authenticates the same way")
|
|
|
|
# --- main(): the refusals that need no board -----------------------------------
|
|
|
|
|
|
def run_main(argv, fp_env=None, token_env="hunter2secret"):
|
|
"""Run main() with the transport faked: create_connection records the address
|
|
instead of dialing, and ssl is the fake from above. Any credential that
|
|
reaches the returned text is a test failure, so the caller never has to
|
|
trust this helper to scrub it."""
|
|
calls = []
|
|
|
|
def record_create(addr, timeout):
|
|
calls.append(addr)
|
|
return FakeBoardSock(addr[0], addr[1])
|
|
|
|
old_argv = sys.argv
|
|
old_env = os.environ.get("KEYRELAY_TLS_FP"), os.environ.get("KEYRELAY_TOKEN")
|
|
sys.argv = ["relayctl.py"] + argv
|
|
if fp_env is None:
|
|
os.environ.pop("KEYRELAY_TLS_FP", None)
|
|
else:
|
|
os.environ["KEYRELAY_TLS_FP"] = fp_env
|
|
if token_env is None:
|
|
os.environ.pop("KEYRELAY_TOKEN", None)
|
|
else:
|
|
os.environ["KEYRELAY_TOKEN"] = token_env
|
|
real_ssl_main = relayctl.ssl
|
|
relayctl.ssl = FakeSsl()
|
|
relayctl.socket.create_connection = record_create
|
|
out, err_stream = io.StringIO(), io.StringIO()
|
|
code, exc = None, None
|
|
try:
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err_stream):
|
|
code = relayctl.main()
|
|
except BaseException as e: # noqa: BLE001 -- reported, then re-raised below
|
|
exc = e
|
|
finally:
|
|
sys.argv = old_argv
|
|
for name, value in zip(("KEYRELAY_TLS_FP", "KEYRELAY_TOKEN"), old_env):
|
|
if value is None:
|
|
os.environ.pop(name, None)
|
|
else:
|
|
os.environ[name] = value
|
|
relayctl.socket.create_connection = real_create_main
|
|
relayctl.ssl = real_ssl_main
|
|
if exc is not None:
|
|
raise exc
|
|
return code, out.getvalue(), err_stream.getvalue(), calls
|
|
|
|
|
|
real_create_main = socketmod.create_connection
|
|
|
|
code, out, err_text, calls = run_main(["--host", "keyrelay.local", "status"])
|
|
check_eq(code, 2, "no trust argument: main refuses with a usage error")
|
|
check(calls == [], "and refuses without opening a socket")
|
|
for option in ("--fingerprint", "--ca", "--insecure", "--plain"):
|
|
check(option in err_text, f"the refusal names {option}")
|
|
check("hunter2secret" not in err_text and "hunter2secret" not in out,
|
|
"the refusal prints no credential")
|
|
|
|
code, out, err_text, calls = run_main(["--fingerprint", "nothex", "status"])
|
|
check_eq(code, 2, "a malformed pin is refused before the network too")
|
|
check(calls == [] and "SHA-256" in err_text, "with the shape of what a pin must be")
|
|
|
|
# $KEYRELAY_TLS_FP is honoured like $KEYRELAY_TOKEN: the gate passes and the
|
|
# socket that follows is the board's. The fake's DER does not match the pin,
|
|
# so the run ends at the mismatch -- proof the pin is wired through main, and
|
|
# the token (never echoed) stays in the terminal.
|
|
code, out, err_text, calls = run_main(["--timeout", "0.1", "status"],
|
|
fp_env="dead" * 16)
|
|
check(calls == [("keyrelay.local", 443)],
|
|
"$KEYRELAY_TLS_FP satisfied the gate and selected :443")
|
|
check(code == 1 and "fingerprint mismatch" in err_text,
|
|
"and a pin that does not match still aborts the run, through main()")
|
|
check("hunter2secret" not in err_text and "hunter2secret" not in out,
|
|
"the passphrase is not printed on the abort path")
|
|
|
|
code, out, err_text, calls = run_main(["paste"], fp_env="ab" * 32, token_env=None)
|
|
check_eq(code, 2, "the token pre-flight still runs before anything else")
|
|
check("passphrase" in err_text and calls == [],
|
|
"no token: refused in the terminal, exactly as before TLS")
|
|
code, out, err_text, calls = run_main(["paste"], fp_env="ab" * 32)
|
|
check_eq(code, 4, "paste still needs --yes before any connection")
|
|
check(calls == [], "and that check still does not require a reachable board")
|
|
|
|
print("the target-side verifier")
|
|
# Twice now the verifier itself was the bug -- mislabelled LEDs, unnamed key codes that
|
|
# looked like lost keystrokes -- so the thing that decides what "it arrived" means gets
|
|
# tested too, and the mouse half of it especially: it is new, and a watcher that misread
|
|
# REL_X would either cry wolf or miss a mouse that worked.
|
|
import watch_target as wt # noqa: E402
|
|
|
|
# EV_REL is what makes a node a pointing device. Nothing here trusts a device name.
|
|
check_eq(wt.classify(1 << wt.EV_REL | 1 << wt.EV_KEY), "mouse", "EV_REL means mouse")
|
|
check_eq(wt.classify(1 << wt.EV_KEY | 1 << wt.EV_LED), "keyboard",
|
|
"EV_KEY without EV_REL means keyboard")
|
|
check_eq(wt.classify(1 << wt.EV_ABS), None,
|
|
"an absolute-only device is neither, and is not watched as a mouse")
|
|
check_eq(wt.capability_bits("/nonexistent/does/not/exist"), 0,
|
|
"an unreadable sysfs file is 0 rather than an exception")
|
|
|
|
m = wt.MouseWatch(False)
|
|
check(m.event(0, 0, 20) is None and m.event(0, 1, -6) is None,
|
|
"a raw axis event is not a line -- one report is one line")
|
|
check(m.event(0, 0, 15) is None, "and more movement in the same report accumulates")
|
|
line = m.flush(0.1)
|
|
check(line is not None and "+35" in line and "-6" in line,
|
|
f"the report comes out as the sum of its axes ({line!r})")
|
|
check(m.flush(0.2) is None, "a report with no movement prints nothing")
|
|
check_eq(m.x, 35, "net X is the exact sum of what the target received")
|
|
check_eq(m.y, -6, "and so is net Y")
|
|
check_eq(m.travel, 41, "and the path length counts both, not the diagonal")
|
|
|
|
# REL_WHEEL_HI_RES is 1/240 of a notch. Counting it as a notch would report 240x the
|
|
# scroll that happened, which is the same class of error as the mislabelled LEDs.
|
|
# The axis numbers, pinned to the kernel's, not to the HID report's byte order. Getting
|
|
# this wrong made a working wheel look dead -- see the comment on REL_NAMES.
|
|
check_eq((wt.REL_X, wt.REL_Y, wt.REL_WHEEL, wt.REL_HWHEEL), (0, 1, 8, 6),
|
|
"axis codes are the kernel's: REL_WHEEL is 8, not 2")
|
|
check_eq((wt.REL_WHEEL_HI_RES, wt.REL_HWHEEL_HI_RES), (11, 12),
|
|
"and the hi-res wheel codes are 11/12, not 6/7")
|
|
|
|
m2 = wt.MouseWatch(False)
|
|
check(m2.event(0, wt.REL_WHEEL, -3) is not None,
|
|
"a wheel notch is printed as it happens, not coalesced with movement")
|
|
m2.event(0, wt.REL_WHEEL_HI_RES, 720) # the same notches again, finer
|
|
check_eq(m2.wheel, -3, "hi-res wheel events are not counted as extra notches")
|
|
check("REL_WHEEL" in "\n".join(m2.report()),
|
|
"and the summary names the axis it saw, so a wheel that arrived is visible")
|
|
m3 = wt.MouseWatch(False)
|
|
m3.event(0, 40, 5) # an axis this tool does not know
|
|
check(any("rel40" in l for l in m3.report()),
|
|
"an unnamed axis code is counted and shown, never dropped in silence")
|
|
m2.event(0, 272, 1)
|
|
check("still held" in "\n".join(m2.report()),
|
|
"a button the target still believes is down is called out, in those words")
|
|
m2.event(0, 272, 0)
|
|
check("still held" not in "\n".join(m2.report()),
|
|
"and the warning goes away when it is released")
|
|
|
|
k = wt.KeyboardWatch(False)
|
|
k.event(0, 30, 1)
|
|
k.event(0, 30, 0)
|
|
check("a" in "".join(k.typed), "the keyboard watcher still reassembles typed text")
|
|
kc = wt.KeyboardWatch(False)
|
|
for code in (29, 56, 111):
|
|
kc.event(0, code, 1)
|
|
check(any("Ctrl+Alt+Del" in l for l in kc.report()),
|
|
"and still names Ctrl+Alt+Del as one combination, not three keys")
|
|
|
|
print(f"relayctl {checks} checks, {failures} failures")
|
|
sys.exit(1 if failures else 0)
|