PFVCluster/netinfra/mdns/hap-bridge.py

#!/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 os
import logging
import socket
import struct
import threading
import time

CONFIG_PATH = os.environ.get("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 decode_rr_header(data, off):
    """Return (rtype, rclass, ttl, rdata_len) for the RR header at offset."""
    return struct.unpack(">HHIH", data[off:off + 10])


def merge_records(prev_recs, new_recs):
    """Union record sets by (name, type); new records replace same-key old ones.

    Devices often answer a multi-question probe with partial record sets
    (ecobee returns only the PTR, #619); accumulate across probes so a thin
    response never discards previously learned SRV/TXT/A records.
    """
    merged = {(r["name"], r["type"]): r for r in prev_recs}
    for r in new_recs:
        merged[(r["name"], r["type"])] = r
    return list(merged.values())


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 = decode_rr_header(data, name_end)
                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] = merge_records(state.get(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()