warpedpinball.origin

  1"""Origin message framing: authenticated UDP from a board to a listener.
  2
  3A Vector board pushes live game events (game state, end of game, reset) as
  4UDP datagrams to port 6809. Historically those went out as plain JSON to the
  5broadcast address; every board on the network shouted at every listener, which
  6is both noisy enough to jam the board's WiFi chip and trivially spoofable by
  7anything else on the LAN.
  8
  9Now a listener *registers* itself with the board over authenticated HTTP
 10(:meth:`warpedpinball.Machine.set_origin_target`), handing over a shared
 11secret. From then on the board unicasts only to that one address, and signs
 12every datagram with the secret.
 13
 14Frame layout::
 15
 16    +--------------------------+-------------------------------+
 17    | 16 ASCII hex chars (MAC) | UTF-8 JSON body               |
 18    +--------------------------+-------------------------------+
 19
 20The MAC is the first 8 bytes of ``HMAC-SHA256(secret, body)``, hex-encoded.
 21Truncation is deliberate: the board is a 150 MHz microcontroller sending one
 22of these several times a second, and 64 bits of tag is far past what a LAN
 23attacker will brute-force in the lifetime of a session secret.
 24
 25The body is a JSON object::
 26
 27    {"machine_id": "a1b2c3d4", "type": "game_state", "data": {...}, "n": 41}
 28
 29``n`` is a counter that increments with every datagram the board sends and
 30resets to zero when a listener re-registers (which also rotates the secret).
 31Receivers should drop any datagram whose ``n`` is not greater than the last
 32one accepted from that board, which is what makes a captured packet useless
 33to replay.
 34"""
 35
 36from __future__ import annotations
 37
 38import hmac
 39import json
 40import secrets
 41from hashlib import sha256
 42from typing import Any, Dict
 43
 44from .exceptions import VectorError
 45
 46#: UDP port a board sends Origin messages to.
 47ORIGIN_UDP_PORT = 6809
 48#: Length of the hex-encoded MAC prefix on every datagram.
 49MAC_LEN = 16
 50#: Bytes of secret handed to the board (as hex, so 32 characters on the wire).
 51SECRET_BYTES = 16
 52
 53
 54class OriginAuthError(VectorError):
 55    """A datagram failed authentication: bad MAC, or a malformed frame."""
 56
 57
 58def new_secret() -> str:
 59    """Generate a fresh registration secret (32 hex characters)."""
 60    return secrets.token_hex(SECRET_BYTES)
 61
 62
 63def _mac(secret: str, body: bytes) -> str:
 64    return hmac.new(secret.encode("utf-8"), body, sha256).hexdigest()[:MAC_LEN]
 65
 66
 67def pack(secret: str, body: Dict[str, Any]) -> bytes:
 68    """Build a signed datagram carrying ``body``.
 69
 70    Mirrors what the firmware sends; used by tests and simulators.
 71    """
 72    encoded = json.dumps(body).encode("utf-8")
 73    return _mac(secret, encoded).encode("ascii") + encoded
 74
 75
 76def _split(packet: bytes) -> tuple[bytes, bytes]:
 77    """Signature and body halves of a frame, or raise on a malformed one."""
 78    if len(packet) <= MAC_LEN:
 79        raise OriginAuthError("Origin datagram too short to carry a signature")
 80    return packet[:MAC_LEN], packet[MAC_LEN:]
 81
 82
 83def _decode(body: bytes) -> Dict[str, Any]:
 84    try:
 85        decoded = json.loads(body)
 86    except ValueError as exc:
 87        raise OriginAuthError(f"Origin datagram body is not valid JSON: {exc}") from None
 88    if not isinstance(decoded, dict):
 89        raise OriginAuthError("Origin datagram body is not a JSON object")
 90    return decoded
 91
 92
 93def peek(packet: bytes) -> Dict[str, Any]:
 94    """Decode a datagram's body **without verifying its signature**.
 95
 96    A listener holding secrets for many boards has a chicken-and-egg problem:
 97    it must know which board is claiming to have sent a datagram before it can
 98    choose the secret to check it against. This answers that question, and
 99    nothing else -- everything it returns is unverified and, on an open
100    network, attacker-controlled.
101
102    Use it only to route::
103
104        machine_id = origin.peek(packet).get("machine_id")
105        body = origin.unpack(secrets[machine_id], packet)   # now it's trusted
106
107    Nothing from :func:`peek` should reach storage, a decision, or a log line
108    that implies it is true.
109    """
110    return _decode(_split(packet)[1])
111
112
113def unpack(secret: str, packet: bytes) -> Dict[str, Any]:
114    """Verify and decode a datagram; raises :class:`OriginAuthError` if it
115    was not signed with ``secret`` or is not a well-formed frame."""
116    received, body = _split(packet)
117    if not hmac.compare_digest(received.decode("ascii", "replace"), _mac(secret, body)):
118        raise OriginAuthError("Origin datagram signature does not match")
119    return _decode(body)