feat(mdns): hap-bridge daemon — unicast probe to multicast replay [#619]

netinfra/mdns/hap-bridge.py

@@ -0,0 +1,193 @@
+#!/usr/bin/env python3
+"""mdns-hap-bridge: unicast-probe HomeKit devices, replay records via multicast.
+
+Why: WiFi-side HomeKit accessories (ecobee, locks) are invisible to multicast
+browsers on the wired segment (AP-side multicast filtering, see #619). They DO
+answer unicast mDNS queries. This daemon probes configured devices and
+re-announces their records on the LAN so Home Assistant's zeroconf/HomeKit
+Controller discovers them normally.
+
+Design notes:
+- Runs on the pfv-bms HAOS host (docker --restart=always), host network.
+- Config: /usr/local/etc/hap-bridge.conf — {"devices": ["192.168.1.212", ...]}
+- Record fidelity: PTR/SRV names are captured as text and re-encoded WITHOUT
+  compression pointers (a raw replay would carry dangling pointers into the
+  original packet); TXT replayed verbatim; A/AAAA raw.
+- SO_REUSEADDR + multicast group membership: coexists with host avahi.
+- Answers multicast queries for _hap._tcp.local and announces periodically.
+- Stdlib only. [#619]
+"""
+
+import json
+import logging
+import socket
+import struct
+import threading
+import time
+
+CONFIG_PATH = "/usr/local/etc/hap-bridge.conf"
+MDNS_GROUP = "224.0.0.251"
+MDNS_PORT = 5353
+REFRESH_INTERVAL = 60
+REPLAY_INTERVAL = 30
+SERVICE = "_hap._tcp.local"
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger("hap-bridge")
+
+
+def read_name(data, off):
+    parts = []
+    while True:
+        length = data[off]
+        if length == 0:
+            off += 1
+            break
+        if length & 0xC0:
+            ptr = struct.unpack(">H", data[off:off + 2])[0] & 0x3FFF
+            parts.append(read_name(data, ptr)[0])
+            off += 2
+            break
+        parts.append(data[off + 1:off + 1 + length].decode("latin1"))
+        off += 1 + length
+    return ".".join(parts), off
+
+
+def encode_name(name):
+    out = b""
+    for part in name.split("."):
+        out += bytes([len(part)]) + part.encode("latin1")
+    return out + b"\x00"
+
+
+def probe_device(ip, timeout=2.0):
+    """Return normalized records from a device's mDNS response."""
+    qname = encode_name(SERVICE)
+    packet = struct.pack(">HHHHHH", 0x4242, 0, 3, 0, 0, 0)
+    packet += b"".join(qname + struct.pack(">HH", t, 1) for t in (12, 33, 16))
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    s.settimeout(timeout)
+    try:
+        s.sendto(packet, (ip, MDNS_PORT))
+        while True:
+            data, _ = s.recvfrom(16384)
+            if not (data[2] & 0x80):
+                continue
+            qd = struct.unpack(">H", data[4:6])[0]
+            an = struct.unpack(">H", data[6:8])[0]
+            ns = struct.unpack(">H", data[8:10])[0]
+            off = 12
+            for _ in range(qd):
+                _, off = read_name(data, off)
+                off += 4
+            records = []
+            for _ in range(an + ns):
+                name, name_end = read_name(data, off)
+                rtype, _rclass, ttl, rlen = struct.unpack(">HHIH", data[name_end:name_end + 10])
+                rec = {"name": name, "type": rtype, "ttl": min(ttl, 4500)}
+                if rtype == 12:  # PTR
+                    rec["ptr_target"] = read_name(data, name_end + 10)[0]
+                elif rtype == 33:  # SRV
+                    pri, weight, port = struct.unpack(">HHH", data[name_end + 10:name_end + 16])
+                    rec.update({
+                        "srv_prio": pri,
+                        "srv_weight": weight,
+                        "srv_port": port,
+                        "srv_target": read_name(data, name_end + 16)[0],
+                    })
+                else:  # TXT / A / AAAA / anything self-contained
+                    rec["rdata_hex"] = data[name_end + 10:name_end + 10 + rlen].hex()
+                records.append(rec)
+                off = name_end + 10 + rlen
+            return records
+    except (socket.timeout, OSError):
+        return []
+    finally:
+        s.close()
+
+
+def encode_record(rec):
+    if rec["type"] == 12:
+        rdata = encode_name(rec["ptr_target"])
+    elif rec["type"] == 33:
+        rdata = struct.pack(">HHH", rec["srv_prio"], rec["srv_weight"], rec["srv_port"])
+        rdata += encode_name(rec["srv_target"])
+    else:
+        rdata = bytes.fromhex(rec["rdata_hex"])
+    wire = encode_name(rec["name"])
+    wire += struct.pack(">HHIH", rec["type"], 1, rec["ttl"], len(rdata))
+    return wire + rdata
+
+
+def build_packet(records):
+    header = struct.pack(">HHHHHH", 0, 0x8400, 0, len(records), 0, 0)
+    return header + b"".join(encode_record(r) for r in records)
+
+
+def load_config():
+    with open(CONFIG_PATH) as f:
+        return json.load(f).get("devices", [])
+
+
+def main():
+    devices = load_config()
+    log.info("starting; devices: %s", devices)
+
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind(("", MDNS_PORT))
+    mreq = socket.inet_aton(MDNS_GROUP) + socket.inet_aton("0.0.0.0")
+    sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
+
+    state = {}
+    lock = threading.Lock()
+
+    def refresh():
+        for ip in devices:
+            recs = probe_device(ip)
+            with lock:
+                if recs:
+                    state[ip] = recs
+            log.info("probe %s: %s", ip, f"{len(recs)} records" if recs else "no response")
+
+    def replay():
+        with lock:
+            snapshot = list(state.values())
+        for recs in snapshot:
+            try:
+                sock.sendto(build_packet(recs), (MDNS_GROUP, MDNS_PORT))
+            except OSError as exc:
+                log.error("replay failed: %s", exc)
+        log.info("replayed %d device(s)", len(snapshot))
+
+    def responder():
+        while True:
+            try:
+                data, _addr = sock.recvfrom(4096)
+            except OSError:
+                continue
+            if data[2] & 0x80:
+                continue
+            qd = struct.unpack(">H", data[4:6])[0]
+            off = 12
+            wants = False
+            try:
+                for _ in range(qd):
+                    name, off = read_name(data, off)
+                    if SERVICE in name:
+                        wants = True
+                    off += 4
+            except (IndexError, struct.error):
+                continue
+            if wants:
+                replay()
+
+    threading.Thread(target=responder, daemon=True).start()
+    while True:
+        refresh()
+        replay()
+        time.sleep(REPLAY_INTERVAL)
+
+
+if __name__ == "__main__":
+    main()

tests/unit/hap-bridge.sh

@@ -0,0 +1,84 @@
+#!/bin/bash
+# Unit tests for netinfra/mdns/hap-bridge.py [#619]
+# Verifies record capture normalization + packet encode/decode round-trip.
+set -euo pipefail
+
+PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+SCRIPT="$PROJECT_ROOT/netinfra/mdns/hap-bridge.py"
+
+TMP="$(mktemp -d)"
+trap 'rm -rf "$TMP"' EXIT
+
+cat > "$TMP/test_bridge.py" <<'PY'
+import importlib.util
+import struct
+import sys
+
+spec = importlib.util.spec_from_file_location("hap_bridge", sys.argv[1])
+hb = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(hb)
+
+failed = 0
+
+def check(name, cond):
+    global failed
+    print(("PASS " if cond else "FAIL ") + name)
+    if not cond:
+        failed += 1
+
+# --- encode_name / read_name round trip
+enc = hb.encode_name("Main Floor._hap._tcp.local")
+back, off = hb.read_name(enc, 0)
+check("name round-trip", back == "Main Floor._hap._tcp.local" and off == len(enc))
+
+# --- encode_record + build_packet round trip for PTR/SRV/TXT
+recs = [
+    {"name": "_hap._tcp.local", "type": 12, "ttl": 10, "ptr_target": "Main Floor._hap._tcp.local"},
+    {"name": "Main Floor._hap._tcp.local", "type": 33, "ttl": 10,
+     "srv_prio": 0, "srv_weight": 0, "srv_port": 58528, "srv_target": "Main-Floor.local"},
+    {"name": "Main Floor._hap._tcp.local", "type": 16, "ttl": 10, "rdata_hex": "0268310568656c6c6f"},
+    {"name": "Main-Floor.local", "type": 1, "ttl": 10, "rdata_hex": "c0a801d4"},
+]
+packet = hb.build_packet(recs)
+check("packet header ancount", struct.unpack(">H", packet[6:8])[0] == 4)
+
+# parse the packet back
+qd = struct.unpack(">H", packet[4:6])[0]
+an = struct.unpack(">H", packet[6:8])[0]
+off = 12
+parsed = []
+for _ in range(qd):
+    _, off = hb.read_name(packet, off)
+    off += 4
+for _ in range(an):
+    name, name_end = hb.read_name(packet, off)
+    rtype, _rc, ttl, rlen = struct.unpack(">HHIH", packet[name_end:name_end + 10])
+    rdata_start = name_end + 10
+    rec = {"name": name, "type": rtype, "ttl": ttl}
+    if rtype == 12:
+        rec["ptr_target"] = hb.read_name(packet, rdata_start)[0]
+    elif rtype == 33:
+        pri, w, port = struct.unpack(">HHH", packet[rdata_start:rdata_start + 6])
+        rec["srv_target"] = hb.read_name(packet, rdata_start + 6)[0]
+        rec["srv_port"] = port
+    else:
+        rec["rdata_hex"] = packet[rdata_start:rdata_start + rlen].hex()
+    parsed.append(rec)
+    off = rdata_start + rlen
+
+check("ptr round-trip", parsed[0]["ptr_target"] == "Main Floor._hap._tcp.local")
+check("srv target round-trip", parsed[1]["srv_target"] == "Main-Floor.local")
+check("srv port round-trip", parsed[1]["srv_port"] == 58528)
+check("txt round-trip", parsed[2]["rdata_hex"] == "0268310568656c6c6f")
+check("a round-trip", parsed[3]["rdata_hex"] == "c0a801d4")
+check("consumed whole packet", off == len(packet))
+
+sys.exit(1 if failed else 0)
+PY
+
+if timeout 60 docker run --rm -v "$SCRIPT:/b.py" -v "$TMP/test_bridge.py:/t.py" python:3.12-alpine python /t.py /b.py; then
+    echo "✅ hap-bridge packet round-trip tests passed"
+else
+    echo "❌ hap-bridge tests failed"
+    exit 1
+fi