Add Origin message framing, formats wrappers, and origin target registration

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:

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)

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)

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",

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]:

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