From 7aa6320a72469b186db9e294624ae8334ec9c2dc Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 02:11:12 +0000 Subject: [PATCH] Add Origin message framing, formats wrappers, and origin target registration Boards have historically pushed live game events to the broadcast address as plain JSON: every board shouting at every listener, noisy enough to jam the board's WiFi chip and trivially spoofable by anything else on the LAN. This adds the client half of a targeted, authenticated replacement: - `warpedpinball.origin` defines the datagram frame -- a truncated HMAC-SHA256 tag over a JSON body carrying a per-send counter -- with `new_secret()`, `pack()`, and `unpack()`. - `Machine.set_origin_target()` / `clear_origin_target()` register a listener over authenticated HTTP. Omitting `ip` lets the board use the address the request arrived from, which is what a listener behind NAT needs. - `Machine.formats()`, `active_format()`, and `set_format()` wrap the formats routes, so callers no longer need the escape hatch for them. `set_format()` sends the options block capitalized, which is the casing the firmware actually reads. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01VmtVSp58rDkDBRPzL9hJkt --- docs/machine.md | 33 +++++++++++++- tests/test_machine.py | 27 ++++++++++++ tests/test_origin.py | 53 +++++++++++++++++++++++ warpedpinball/__init__.py | 5 ++- warpedpinball/machine.py | 56 ++++++++++++++++++++++++ warpedpinball/origin.py | 90 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 262 insertions(+), 2 deletions(-) create mode 100644 tests/test_origin.py create mode 100644 warpedpinball/origin.py diff --git a/docs/machine.md b/docs/machine.md index 844a926..e02dd70 100644 --- a/docs/machine.md +++ b/docs/machine.md @@ -116,10 +116,41 @@ m.wait_until_reachable(timeout=120) Other wrappers include `machine_id()`, `game_name()`, `active_config()`, `wifi_status()`, `faults()`, `peers()`, `tournament()`, `claimable_scores()` / `claim_score()`, `export_scores()` / `import_scores()`, -`reset_leaderboard()` / `reset_tournament()`, `date()` / `set_date()`, and the +`reset_leaderboard()` / `reset_tournament()`, `date()` / `set_date()`, the +formats family (`formats()`, `active_format()`, `set_format()`), and the adjustments family (`adjustments()`, `capture_adjustments()`, `restore_adjustments()`, `name_adjustment()`). +## Live game events over UDP + +`watch_game()` polls, which is fine for one machine and wasteful for twenty. +A board can instead *push* its game events to you, as signed UDP datagrams on +port 6809 -- what Origin uses to follow a room full of machines at once. + +Register yourself as the board's target, and it sends only to you: + +```python +from warpedpinball import origin + +secret = origin.new_secret() +m.set_origin_target(secret) # board sends to wherever this call came from +m.set_origin_target(secret, ip="192.168.1.5") # ...or to somewhere else + +# Then, on a socket bound to port 6809: +event = origin.unpack(secret, datagram) # raises OriginAuthError if not ours +print(event["type"], event["data"]) + +m.clear_origin_target() # stop the stream +``` + +A board with no registered target sends nothing at all, and a board with one +never broadcasts -- one address, one listener. Every datagram is signed with +the secret and carries a counter `n` that increases with each send, so drop +anything whose `n` is not greater than the last you accepted from that board. +Re-registering rotates the secret and resets the counter, so a listener that +restarts just registers again. See +[`warpedpinball.origin`](api-reference.md) for the frame layout. + ## The raw escape hatch Every firmware route is reachable even without a wrapper: diff --git a/tests/test_machine.py b/tests/test_machine.py index 47db633..dc87386 100644 --- a/tests/test_machine.py +++ b/tests/test_machine.py @@ -54,6 +54,11 @@ def make_machine(responses=None, streams=None, password="pw", requires_password= "/api/memory/toggle-broadcast", True, ), + (lambda m: m.formats(), "/api/formats/available", False), + (lambda m: m.active_format(), "/api/formats/active", False), + (lambda m: m.set_format(2), "/api/formats/set", True), + (lambda m: m.set_origin_target("abc"), "/api/origin/target", True), + (lambda m: m.clear_origin_target(), "/api/origin/target", True), ] @@ -199,6 +204,28 @@ def test_read_decodes_int(): assert machine.read(0x01, 2, byteorder="little") == 0x0201 +def test_set_format_bodies(): + machine, transport = make_machine() + machine.set_format(2) + machine.set_format(2, {"GetPlayerID": {"Value": True}}) + bodies = [body for _, body, _ in transport.calls] + assert bodies[0] == {"format_id": 2} + # The firmware reads the options block capitalized. + assert bodies[1]["Options"] == {"GetPlayerID": {"Value": True}} + + +def test_origin_target_bodies(): + machine, transport = make_machine() + machine.set_origin_target("s3cret") + machine.set_origin_target("s3cret", ip="192.168.1.20") + machine.clear_origin_target() + bodies = [body for _, body, _ in transport.calls] + # No ip means "wherever this request came from" -- the board fills it in. + assert bodies[0] == {"enable": True, "secret": "s3cret"} + assert bodies[1]["ip"] == "192.168.1.20" + assert bodies[2] == {"enable": False} + + def test_set_memory_broadcast_bodies(): machine, transport = make_machine() machine.set_memory_broadcast(True, frequency_ms=250) diff --git a/tests/test_origin.py b/tests/test_origin.py new file mode 100644 index 0000000..b125e88 --- /dev/null +++ b/tests/test_origin.py @@ -0,0 +1,53 @@ +"""Origin datagram framing: signing, verification, and tamper rejection.""" + +import json + +import pytest + +from warpedpinball import origin + + +def test_pack_unpack_round_trip(): + secret = origin.new_secret() + body = {"machine_id": "a1b2c3d4", "type": "game_state", "data": {"ball": 2}, "n": 7} + assert origin.unpack(secret, origin.pack(secret, body)) == body + + +def test_new_secret_is_random_hex(): + assert len(origin.new_secret()) == origin.SECRET_BYTES * 2 + assert origin.new_secret() != origin.new_secret() + + +def test_wrong_secret_is_rejected(): + packet = origin.pack("right", {"type": "reset", "n": 1}) + with pytest.raises(origin.OriginAuthError): + origin.unpack("wrong", packet) + + +def test_tampered_body_is_rejected(): + secret = origin.new_secret() + packet = origin.pack(secret, {"type": "end_of_game", "n": 1}) + with pytest.raises(origin.OriginAuthError): + origin.unpack(secret, packet[: origin.MAC_LEN] + b'{"type":"reset","n":1}') + + +@pytest.mark.parametrize( + "packet", + [ + b"", + b"short", + b"0" * origin.MAC_LEN, # signature but no body + b"not-hex-at-all!!" + b'{"n":1}', + ], +) +def test_malformed_frames_are_rejected(packet): + with pytest.raises(origin.OriginAuthError): + origin.unpack("secret", packet) + + +def test_non_json_and_non_object_bodies_are_rejected(): + secret = origin.new_secret() + for body in (b"not json", json.dumps([1, 2]).encode()): + packet = origin._mac(secret, body).encode() + body + with pytest.raises(origin.OriginAuthError): + origin.unpack(secret, packet) diff --git a/warpedpinball/__init__.py b/warpedpinball/__init__.py index f2efd58..392eda2 100644 --- a/warpedpinball/__init__.py +++ b/warpedpinball/__init__.py @@ -30,9 +30,10 @@ VectorServerError, ) from .machine import GameEvent, Machine +from .origin import OriginAuthError from .transports.http import HttpTransport -__version__ = "0.2.2" +__version__ = "0.3.0" __all__ = [ "connect", @@ -43,6 +44,8 @@ "GameEvent", "DiscoveredMachine", "HttpTransport", + "origin", + "OriginAuthError", "VectorError", "TransportError", "DeviceUnreachableError", diff --git a/warpedpinball/machine.py b/warpedpinball/machine.py index 8c113a0..110cb58 100644 --- a/warpedpinball/machine.py +++ b/warpedpinball/machine.py @@ -459,6 +459,62 @@ def diff_snapshots(a: bytes, b: bytes) -> List[Tuple[int, int, int]]: changes.append((i, va, vb)) return changes + # -- formats ----------------------------------------------------------------------- + + def formats(self) -> Any: + """Available game formats from ``/api/formats/available``.""" + return self.call("/api/formats/available") + + def active_format(self) -> Any: + """The format the machine is currently playing, from ``/api/formats/active``.""" + return self.call("/api/formats/active") + + def set_format(self, format_id: Any, options: Optional[Dict[str, Any]] = None) -> Any: + """Activate a game format via ``/api/formats/set`` (authenticated). + + ``format_id`` is the numeric id (or the name) of one of the formats + :meth:`formats` returned. ``options`` are that format's configurable + settings, shaped like the ``Options`` block in the format's metadata. + """ + body: Dict[str, Any] = {"format_id": format_id} + if options: + # The firmware reads this key capitalized, matching the casing it + # uses when it hands the options back out of /api/formats/available. + body["Options"] = options + return self._call_gated("/api/formats/set", body=body, authenticated=True) + + # -- origin messages --------------------------------------------------------------- + + def set_origin_target(self, secret: str, ip: Optional[str] = None) -> Any: + """Register where the board unicasts its Origin messages. Authenticated. + + The board pushes live game events (game state, end of game, reset) as + UDP datagrams to port 6809. Until a target is registered it sends + nothing at all; once registered it sends *only* to this one address, + signing every datagram with ``secret`` (see :mod:`warpedpinball.origin` + for the frame layout and :func:`warpedpinball.origin.new_secret` for + generating one). + + ``ip`` is the IPv4 address to send to; when omitted the board uses the + address this request arrived from, which is what you want whenever the + listener runs where this code runs -- including behind NAT, where the + board sees the translated address and the listener could not have + named it. Over USB there is no requester address, so pass ``ip``. + + Registering again rotates the secret and resets the board's datagram + counter, so a listener that restarts simply re-registers. + """ + body: Dict[str, Any] = {"enable": True, "secret": secret} + if ip is not None: + body["ip"] = ip + return self._call_gated("/api/origin/target", body=body, authenticated=True) + + def clear_origin_target(self) -> Any: + """Stop the board sending Origin messages anywhere. Authenticated.""" + return self._call_gated( + "/api/origin/target", body={"enable": False}, authenticated=True + ) + # -- polling ----------------------------------------------------------------------- def watch_game(self, interval: float = 1.0) -> Iterator[GameEvent]: diff --git a/warpedpinball/origin.py b/warpedpinball/origin.py new file mode 100644 index 0000000..9701fad --- /dev/null +++ b/warpedpinball/origin.py @@ -0,0 +1,90 @@ +"""Origin message framing: authenticated UDP from a board to a listener. + +A Vector board pushes live game events (game state, end of game, reset) as +UDP datagrams to port 6809. Historically those went out as plain JSON to the +broadcast address; every board on the network shouted at every listener, which +is both noisy enough to jam the board's WiFi chip and trivially spoofable by +anything else on the LAN. + +Now a listener *registers* itself with the board over authenticated HTTP +(:meth:`warpedpinball.Machine.set_origin_target`), handing over a shared +secret. From then on the board unicasts only to that one address, and signs +every datagram with the secret. + +Frame layout:: + + +--------------------------+-------------------------------+ + | 16 ASCII hex chars (MAC) | UTF-8 JSON body | + +--------------------------+-------------------------------+ + +The MAC is the first 8 bytes of ``HMAC-SHA256(secret, body)``, hex-encoded. +Truncation is deliberate: the board is a 150 MHz microcontroller sending one +of these several times a second, and 64 bits of tag is far past what a LAN +attacker will brute-force in the lifetime of a session secret. + +The body is a JSON object:: + + {"machine_id": "a1b2c3d4", "type": "game_state", "data": {...}, "n": 41} + +``n`` is a counter that increments with every datagram the board sends and +resets to zero when a listener re-registers (which also rotates the secret). +Receivers should drop any datagram whose ``n`` is not greater than the last +one accepted from that board, which is what makes a captured packet useless +to replay. +""" + +from __future__ import annotations + +import hmac +import json +import secrets +from hashlib import sha256 +from typing import Any, Dict + +from .exceptions import VectorError + +#: UDP port a board sends Origin messages to. +ORIGIN_UDP_PORT = 6809 +#: Length of the hex-encoded MAC prefix on every datagram. +MAC_LEN = 16 +#: Bytes of secret handed to the board (as hex, so 32 characters on the wire). +SECRET_BYTES = 16 + + +class OriginAuthError(VectorError): + """A datagram failed authentication: bad MAC, or a malformed frame.""" + + +def new_secret() -> str: + """Generate a fresh registration secret (32 hex characters).""" + return secrets.token_hex(SECRET_BYTES) + + +def _mac(secret: str, body: bytes) -> str: + return hmac.new(secret.encode("utf-8"), body, sha256).hexdigest()[:MAC_LEN] + + +def pack(secret: str, body: Dict[str, Any]) -> bytes: + """Build a signed datagram carrying ``body``. + + Mirrors what the firmware sends; used by tests and simulators. + """ + encoded = json.dumps(body).encode("utf-8") + return _mac(secret, encoded).encode("ascii") + encoded + + +def unpack(secret: str, packet: bytes) -> Dict[str, Any]: + """Verify and decode a datagram; raises :class:`OriginAuthError` if it + was not signed with ``secret`` or is not a well-formed frame.""" + if len(packet) <= MAC_LEN: + raise OriginAuthError("Origin datagram too short to carry a signature") + received, body = packet[:MAC_LEN], packet[MAC_LEN:] + if not hmac.compare_digest(received.decode("ascii", "replace"), _mac(secret, body)): + raise OriginAuthError("Origin datagram signature does not match") + try: + decoded = json.loads(body) + except ValueError as exc: + raise OriginAuthError(f"Origin datagram body is not valid JSON: {exc}") from None + if not isinstance(decoded, dict): + raise OriginAuthError("Origin datagram body is not a JSON object") + return decoded