#!/usr/bin/env python3
"""Hub/spoke WireGuard enrollment over clear WLAN (keys only — tip still via update.sh).

Spoke generates its own WG keypair (private never leaves the device). Hub validates
bank + fleet credential + allowlist (or open-enrollment window), then:
  wg set wg0 peer <pub> allowed-ips 10.99.<bank>.<spoke>/32
  creates deterministic gretap eg-sNN
  persists peer set for reboot restore

Separate UDP listener from progressives_http.py (31234).
"""

from __future__ import annotations

import json
import os
import socket
import subprocess
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable

from pog_edge_bank_wlan import load_fleet_hmac_key, verify_envelope, _sign_body
from pog_edge_identity import latch_cabinet_identity

DEFAULT_ENROLL_PORT = 41235
DEFAULT_STATE_DIR = "/etc/ethergold/hub"
DEFAULT_SPOKE_KEYS_DIR = "/etc/ethergold/spoke"
DEFAULT_OPEN_WINDOW_S = 600  # 10 minutes
# WG keepalive is 25s. A dead Pi (or swapped board) must fall off fanout
# without a hub reboot. 180s ≈ 7 missed keepalives — WiFi blips of 1–2 min stay.
DEFAULT_SPOKE_STALE_S = 180
REAP_INTERVAL_S = 5.0
# User-writable fallthrough when /etc/ethergold is root-only (no sudo required).
_FALLBACK_STATE_DIRS = (
    lambda: Path.home() / ".local/share/ethergold/hub",
    lambda: Path("/var/tmp/ethergold/hub"),
)
_FALLBACK_SPOKE_KEYS_DIRS = (
    lambda: Path.home() / ".local/share/ethergold/spoke",
    lambda: Path("/var/tmp/ethergold/spoke"),
)


def resolve_enroll_port() -> int:
    try:
        return int(os.environ.get("EDGE_ENROLL_PORT", str(DEFAULT_ENROLL_PORT)))
    except ValueError:
        return DEFAULT_ENROLL_PORT


def _dir_is_writable(path: Path) -> bool:
    try:
        path.mkdir(parents=True, exist_ok=True)
        probe = path / ".eg_write_probe"
        probe.write_text("ok\n", encoding="utf-8")
        probe.unlink(missing_ok=True)
        return True
    except OSError:
        return False


def ensure_hub_state_dir(
    preferred: Path | None = None,
    *,
    emit: Callable[..., None] | None = None,
) -> Path:
    """Return a hub state directory the bridge process can actually write.

    Prefers EDGE_HUB_STATE_DIR / /etc/ethergold/hub; if root-owned or missing
    write permission, automatically falls through to ``~/.local/share/ethergold/hub``
    (no manual chmod, no sudo). Sets EDGE_HUB_STATE_DIR for this process.
    """
    candidates: list[Path] = []
    env = (os.environ.get("EDGE_HUB_STATE_DIR") or "").strip()
    if preferred is not None:
        candidates.append(Path(preferred))
    if env:
        candidates.append(Path(env))
    candidates.append(Path(DEFAULT_STATE_DIR))
    for factory in _FALLBACK_STATE_DIRS:
        try:
            candidates.append(factory())
        except Exception:
            pass
    # De-dupe while preserving order
    seen: set[str] = set()
    ordered: list[Path] = []
    for c in candidates:
        key = str(c)
        if key in seen:
            continue
        seen.add(key)
        ordered.append(c)
    chosen: Path | None = None
    for cand in ordered:
        if _dir_is_writable(cand):
            chosen = cand
            break
    if chosen is None:
        chosen = Path.home() / ".local/share/ethergold/hub"
        chosen.mkdir(parents=True, exist_ok=True)
    os.environ["EDGE_HUB_STATE_DIR"] = str(chosen)
    if emit is not None and str(chosen) != DEFAULT_STATE_DIR:
        try:
            emit(
                "hub_state_dir",
                path=str(chosen),
                preferred=DEFAULT_STATE_DIR,
                reason="auto_writable_fallback"
                if str(chosen) != (env or DEFAULT_STATE_DIR)
                else "configured",
            )
        except Exception:
            pass
    return chosen


def resolve_state_dir() -> Path:
    return ensure_hub_state_dir()


def overlay_ip(bank: int, spoke_id: int) -> str:
    """Bank-scoped overlay: hub .1, spokes .2..254. Bank 0 = legacy 10.99.0.x."""
    b = max(0, min(255, int(bank)))
    s = max(1, min(254, int(spoke_id)))
    return f"10.99.{b}.{s}"


def gretap_name_for_spoke(spoke_id: int) -> str:
    return f"eg-s{int(spoke_id):02d}"


def spoke_stale_s() -> float:
    raw = os.environ.get("EDGE_SPOKE_STALE_S", str(DEFAULT_SPOKE_STALE_S))
    try:
        return float(raw)
    except (TypeError, ValueError):
        return float(DEFAULT_SPOKE_STALE_S)


def parse_wg_dump(text: str) -> dict[str, dict[str, Any]]:
    """Parse ``wg show <if> dump`` (tab-separated). Skip the interface header line."""
    out: dict[str, dict[str, Any]] = {}
    for i, line in enumerate((text or "").splitlines()):
        if i == 0 or not line.strip():
            continue
        parts = line.split("\t")
        if len(parts) < 5:
            continue
        try:
            handshake = int(parts[4] or 0)
        except ValueError:
            handshake = 0
        out[parts[0]] = {
            "endpoint": parts[2],
            "allowed_ips": parts[3],
            "handshake": handshake,
        }
    return out


def overlay_spoke_id(allowed_ips: str) -> int | None:
    for tok in (allowed_ips or "").split(","):
        host = tok.strip().split("/")[0]
        parts = host.split(".")
        if len(parts) == 4 and parts[0] == "10" and parts[1] == "99":
            try:
                sid = int(parts[3])
            except ValueError:
                continue
            if 2 <= sid <= 254:
                return sid
    return None


def peer_handshake_stale(
    handshake: int,
    *,
    now: float,
    stale_s: float,
    grace_until: float,
) -> bool:
    """True when the hub should drop this WG peer.

    ``handshake`` is unix seconds from ``wg show dump`` (0 = never).
    ``grace_until`` is unix time: do not reap never-connected peers before this
    (hub just started / just restored; spokes need time to handshake).
    """
    if stale_s <= 0:
        return False
    if now < grace_until:
        return False
    if handshake <= 0:
        return True
    return (now - float(handshake)) > stale_s


def drop_hub_peer(
    *,
    wg_if: str,
    pub: str,
    rec: dict[str, Any] | None = None,
    emit: Callable[..., None] | None = None,
    reason: str = "stale",
) -> None:
    """Remove a live WG peer + its gretap. Missing iface/peer is not fatal."""
    spoke_id = 0
    name = ""
    if rec:
        try:
            spoke_id = int(rec.get("spoke_id") or 0)
        except (TypeError, ValueError):
            spoke_id = 0
        name = str(rec.get("gretap") or "")
    if not name and spoke_id:
        name = gretap_name_for_spoke(spoke_id)
    _run(["wg", "set", wg_if, "peer", pub, "remove"], check=False)
    if name:
        _run(["ip", "link", "del", name], check=False)
    if emit:
        emit(
            "hub_peer_dropped",
            spoke_id=spoke_id or None,
            gretap=name or None,
            pub=pub[:16],
            reason=reason,
        )


def _run(cmd: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
    return subprocess.run(cmd, check=check, text=True, capture_output=True)


def wg_gen_keypair() -> tuple[str, str]:
    """Return (private, public) WireGuard keys as strings."""
    priv = _run(["wg", "genkey"]).stdout.strip()
    proc = subprocess.run(
        ["wg", "pubkey"],
        input=priv + "\n",
        check=True,
        text=True,
        capture_output=True,
    )
    return priv, proc.stdout.strip()


def _keys_dir_ready_for_spoke(keys_dir: Path) -> bool:
    """True when we can use keys_dir for enroll (read existing or generate).

    EG 3.0.2.9: /etc/ethergold/spoke can look writable (create probe) yet hold
    root:600 ``wg_private.key`` — write then raises Permission denied and enroll
    loops forever (pos5 soak 1953). Prefer the dir only if key files are absent
    or RW for this UID, and the directory itself is writable for minting keys.
    """
    if not _dir_is_writable(keys_dir):
        return False
    priv_path = keys_dir / "wg_private.key"
    pub_path = keys_dir / "wg_public.key"
    if priv_path.exists() or pub_path.exists():
        if priv_path.exists() and not (
            os.access(priv_path, os.R_OK) and os.access(priv_path, os.W_OK)
        ):
            return False
        if pub_path.exists() and not os.access(pub_path, os.R_OK):
            return False
    return True


def ensure_spoke_keys_dir(
    preferred: Path | None = None,
    *,
    emit: Callable[..., None] | None = None,
) -> Path:
    """Return a spoke key directory the bridge process can read/write.

    Prefers EDGE_SPOKE_KEYS_DIR / /etc/ethergold/spoke. When root-owned keys or
    missing write permission (pos5: Permission denied on wg_private.key), fall
    through to ``~/.local/share/ethergold/spoke`` — same pattern as
    ``ensure_hub_state_dir`` (EG 3.0.1.30 / 3.0.1.47 / 3.0.2.9).
    """
    candidates: list[Path] = []
    env = (os.environ.get("EDGE_SPOKE_KEYS_DIR") or "").strip()
    if preferred is not None:
        candidates.append(Path(preferred))
    if env:
        candidates.append(Path(env))
    candidates.append(Path(DEFAULT_SPOKE_KEYS_DIR))
    for factory in _FALLBACK_SPOKE_KEYS_DIRS:
        try:
            candidates.append(factory())
        except Exception:
            pass
    seen: set[str] = set()
    ordered: list[Path] = []
    for c in candidates:
        key = str(c)
        if key in seen:
            continue
        seen.add(key)
        ordered.append(c)
    chosen: Path | None = None
    for cand in ordered:
        if _keys_dir_ready_for_spoke(cand):
            chosen = cand
            break
    if chosen is None:
        chosen = Path.home() / ".local/share/ethergold/spoke"
        chosen.mkdir(parents=True, exist_ok=True)
    os.environ["EDGE_SPOKE_KEYS_DIR"] = str(chosen)
    if emit is not None and str(chosen) != DEFAULT_SPOKE_KEYS_DIR:
        try:
            emit(
                "spoke_keys_dir",
                path=str(chosen),
                preferred=DEFAULT_SPOKE_KEYS_DIR,
                reason="auto_writable_fallback",
            )
        except Exception:
            pass
    return chosen


def ensure_spoke_keypair(
    path: Path | None = None,
    *,
    emit: Callable[..., None] | None = None,
) -> tuple[str, str]:
    """Generate once; private key never leaves the device.

    Tries preferred / EDGE_SPOKE_KEYS_DIR / /etc/ethergold/spoke, then user
    fallthrough. If a system path has readable keys but is not writable for
    generation, still prefer those keys when both files exist and are readable.
    Root-owned unreadable/unwritable files under /etc must not block ~/.local.
    """
    candidates: list[Path] = []
    if path is not None:
        candidates.append(Path(path))
    env = (os.environ.get("EDGE_SPOKE_KEYS_DIR") or "").strip()
    if env:
        candidates.append(Path(env))
    candidates.append(Path(DEFAULT_SPOKE_KEYS_DIR))
    for factory in _FALLBACK_SPOKE_KEYS_DIRS:
        try:
            candidates.append(factory())
        except Exception:
            pass
    seen: set[str] = set()
    ordered: list[Path] = []
    for c in candidates:
        key = str(c)
        if key in seen:
            continue
        seen.add(key)
        ordered.append(c)

    last_err: OSError | None = None
    # Pass 1: read existing keypairs from any candidate (prefer system path order).
    for keys_dir in ordered:
        priv_path = keys_dir / "wg_private.key"
        pub_path = keys_dir / "wg_public.key"
        if not (priv_path.is_file() and pub_path.is_file()):
            continue
        try:
            priv = priv_path.read_text(encoding="utf-8").strip()
            pub = pub_path.read_text(encoding="utf-8").strip()
            if priv and pub:
                os.environ["EDGE_SPOKE_KEYS_DIR"] = str(keys_dir)
                return priv, pub
        except OSError as exc:
            last_err = exc
            continue

    # Pass 2: generate into first usable directory (skip locked root key residue).
    for keys_dir in ordered:
        try:
            if not _keys_dir_ready_for_spoke(keys_dir):
                continue
            priv_path = keys_dir / "wg_private.key"
            pub_path = keys_dir / "wg_public.key"
            # Existing root-owned key: do not clobber — try next candidate.
            if priv_path.exists() and not os.access(priv_path, os.W_OK):
                last_err = OSError(13, f"Permission denied: {priv_path}")
                continue
            priv, pub = wg_gen_keypair()
            priv_path.write_text(priv + "\n", encoding="utf-8")
            pub_path.write_text(pub + "\n", encoding="utf-8")
            try:
                os.chmod(priv_path, 0o600)
                os.chmod(pub_path, 0o644)
            except OSError:
                pass
            os.environ["EDGE_SPOKE_KEYS_DIR"] = str(keys_dir)
            if emit is not None and str(keys_dir) != DEFAULT_SPOKE_KEYS_DIR:
                try:
                    emit(
                        "spoke_keys_dir",
                        path=str(keys_dir),
                        preferred=DEFAULT_SPOKE_KEYS_DIR,
                        reason="auto_writable_fallback",
                    )
                except Exception:
                    pass
            return priv, pub
        except OSError as exc:
            last_err = exc
            continue

    # Last resort through writable ensure (may create ~/.local/...).
    keys_dir = ensure_spoke_keys_dir(path, emit=emit)
    priv_path = keys_dir / "wg_private.key"
    pub_path = keys_dir / "wg_public.key"
    try:
        priv, pub = wg_gen_keypair()
        priv_path.write_text(priv + "\n", encoding="utf-8")
        pub_path.write_text(pub + "\n", encoding="utf-8")
        try:
            os.chmod(priv_path, 0o600)
            os.chmod(pub_path, 0o644)
        except OSError:
            pass
        return priv, pub
    except OSError as exc:
        if last_err is not None:
            raise RuntimeError(
                f"spoke keypair unavailable under {ordered!r}: {exc}"
            ) from last_err
        raise


@dataclass
class HubPeerStore:
    path: Path
    peers: dict[str, dict[str, Any]] = field(default_factory=dict)  # pub -> record

    @classmethod
    def load(
        cls,
        state_dir: Path | None = None,
        *,
        emit: Callable[..., None] | None = None,
    ) -> "HubPeerStore":
        preferred = Path(
            state_dir
            if state_dir is not None
            else (os.environ.get("EDGE_HUB_STATE_DIR") or DEFAULT_STATE_DIR)
        )
        write_root = ensure_hub_state_dir(preferred, emit=emit)
        peers: dict[str, dict[str, Any]] = {}
        # Read merged roster: system path first (often root-provisioned), then
        # user-writable path (always the save target).
        for root in (preferred, write_root):
            path = Path(root) / "peers.json"
            if not path.is_file():
                continue
            try:
                data = json.loads(path.read_text(encoding="utf-8"))
                if isinstance(data, dict):
                    for k, v in data.items():
                        if isinstance(v, dict):
                            peers[str(k)] = v
            except (OSError, json.JSONDecodeError):
                continue
        return cls(path=write_root / "peers.json", peers=peers)

    def save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(self.peers, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        os.replace(tmp, self.path)
        try:
            os.chmod(self.path, 0o640)
        except OSError:
            pass  # non-fatal on some FS / ownership layouts

    def next_spoke_id(self) -> int:
        used = {int(p.get("spoke_id", 0)) for p in self.peers.values()}
        for cand in range(2, 255):
            if cand not in used:
                return cand
        raise RuntimeError("no free spoke ids (2..254)")

    def allowlist_pubs(self) -> set[str]:
        path = self.path.parent / "allowlist.txt"
        if not path.is_file():
            return set()
        out: set[str] = set()
        for line in path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            out.add(line.split()[0])
        return out


def apply_hub_peer(
    *,
    wg_if: str,
    pub: str,
    overlay: str,
    spoke_id: int,
    hub_overlay: str,
) -> str:
    """Live-add WG peer + deterministic gretap. Returns gretap name."""
    _run(
        [
            "wg",
            "set",
            wg_if,
            "peer",
            pub,
            "allowed-ips",
            f"{overlay}/32",
            "persistent-keepalive",
            "25",
        ]
    )
    name = gretap_name_for_spoke(spoke_id)
    # Do not delete sibling gretaps — create only if missing.
    check = subprocess.run(
        ["ip", "link", "show", name],
        capture_output=True,
        text=True,
    )
    if check.returncode != 0:
        _run(
            [
                "ip",
                "link",
                "add",
                name,
                "type",
                "gretap",
                "local",
                hub_overlay,
                "remote",
                overlay,
                "dev",
                wg_if,
            ]
        )
    _run(["ip", "link", "set", name, "up"], check=False)
    return name


def restore_hub_peers_from_disk(
    *,
    wg_if: str = "wg0",
    bank: int | None = None,
    emit: Callable[..., None] | None = None,
) -> list[str]:
    """On hub reboot: re-apply persisted peers + gretaps."""
    store = HubPeerStore.load()
    b = int(bank if bank is not None else os.environ.get("EDGE_BANK", "0") or 0)
    hub_overlay = overlay_ip(b, 1)
    names: list[str] = []
    for pub, rec in store.peers.items():
        spoke_id = int(rec.get("spoke_id", 0))
        overlay = str(rec.get("overlay") or overlay_ip(b, spoke_id))
        try:
            name = apply_hub_peer(
                wg_if=wg_if,
                pub=pub,
                overlay=overlay,
                spoke_id=spoke_id,
                hub_overlay=hub_overlay,
            )
            names.append(name)
            if emit:
                emit("hub_peer_restored", spoke_id=spoke_id, overlay=overlay, gretap=name)
        except Exception as exc:
            if emit:
                emit("hub_peer_restore_error", pub=pub[:16], message=str(exc))
    return names


def eth_ports_from_spokes(
    game_if: str | None,
    spoke_ids: list[int],
) -> str:
    """Build ETH_PORTS=game,eg-s02,eg-s03,... from live spoke set."""
    parts: list[str] = []
    if game_if:
        parts.append(game_if)
    for sid in sorted(spoke_ids):
        parts.append(gretap_name_for_spoke(sid))
    return ",".join(parts)


@dataclass
class EnrollmentHub:
    bank: int
    emit: Callable[..., None] | None = None
    """Optional: called after peer is on WG+gretap (enroll or restore)."""
    on_peer_ready: Callable[..., None] | None = None
    """Optional: called after a stale/replaced peer is dropped from WG+gretap."""
    on_peer_gone: Callable[..., None] | None = None
    port: int = field(default_factory=resolve_enroll_port)
    open_until: float = 0.0
    store: HubPeerStore | None = None
    key: bytes = field(default_factory=load_fleet_hmac_key)
    wg_if: str = "wg0"
    _stop: threading.Event = field(default_factory=threading.Event)
    _thread: threading.Thread | None = None
    _started_unix: float = 0.0
    _last_reap_mono: float = 0.0

    def __post_init__(self) -> None:
        if self.store is None:
            self.store = HubPeerStore.load(emit=self.emit)

    def _notify_peer_ready(self, rec: dict[str, Any], *, reason: str) -> None:
        if self.on_peer_ready is None:
            return
        try:
            self.on_peer_ready(rec, reason=reason)
        except Exception as exc:
            if self.emit:
                self.emit(
                    "hub_peer_ready_cb_error",
                    message=str(exc),
                    reason=reason,
                    gretap=rec.get("gretap"),
                )

    def _notify_peer_gone(self, rec: dict[str, Any], *, reason: str) -> None:
        if self.on_peer_gone is None:
            return
        try:
            self.on_peer_gone(rec, reason=reason)
        except Exception as exc:
            if self.emit:
                self.emit(
                    "hub_peer_gone_cb_error",
                    message=str(exc),
                    reason=reason,
                    gretap=rec.get("gretap"),
                )

    def _drop_pub(self, pub: str, rec: dict[str, Any] | None, *, reason: str) -> None:
        drop_hub_peer(
            wg_if=self.wg_if,
            pub=pub,
            rec=rec,
            emit=self.emit,
            reason=reason,
        )
        if self.store is not None and pub in self.store.peers:
            del self.store.peers[pub]
            self.store.save()
        if rec:
            self._notify_peer_gone(rec, reason=reason)

    def _maybe_reap(self) -> None:
        stale_s = spoke_stale_s()
        if stale_s <= 0:
            return
        now_mono = time.monotonic()
        if now_mono - self._last_reap_mono < REAP_INTERVAL_S:
            return
        self._last_reap_mono = now_mono
        now = time.time()
        grace_until = (self._started_unix or now) + stale_s
        try:
            dump = _run(["wg", "show", self.wg_if, "dump"], check=False).stdout
        except OSError:
            return
        live = parse_wg_dump(dump)
        store_peers = dict(self.store.peers) if self.store is not None else {}
        seen: set[str] = set()
        for pub, info in live.items():
            seen.add(pub)
            rec = store_peers.get(pub)
            if rec is None:
                sid = overlay_spoke_id(str(info.get("allowed_ips") or ""))
                rec = {
                    "spoke_id": sid or 0,
                    "gretap": gretap_name_for_spoke(sid) if sid else "",
                    "overlay": f"10.99.{self.bank}.{sid}" if sid else "",
                }
            if peer_handshake_stale(
                int(info.get("handshake") or 0),
                now=now,
                stale_s=stale_s,
                grace_until=grace_until,
            ):
                self._drop_pub(pub, rec, reason="stale_handshake")
        # Disk-only leftovers with no live WG row: drop after grace so a restored
        # peer that never comes back does not stay in ETH_PORTS forever.
        if now >= grace_until and self.store is not None:
            for pub, rec in list(self.store.peers.items()):
                if pub not in seen:
                    self._drop_pub(pub, rec, reason="missing_wg")

    def open_enrollment(self, window_s: float | None = None) -> float:
        secs = float(
            window_s
            if window_s is not None
            else os.environ.get("EDGE_OPEN_ENROLL_S", DEFAULT_OPEN_WINDOW_S)
        )
        self.open_until = time.time() + max(0.0, secs)
        if self.emit:
            self.emit("enroll_window_open", until=self.open_until, window_s=secs)
        return self.open_until

    def enrollment_open(self) -> bool:
        return time.time() < self.open_until

    def start(self) -> None:
        if self._thread is not None and self._thread.is_alive():
            return
        restore_hub_peers_from_disk(wg_if=self.wg_if, bank=self.bank, emit=self.emit)
        # Hot-attach every restored gretap into the live eth router (field reboot
        # already has eth-ports at process start; this covers restore+enroll gaps).
        for _pub, rec in list(self.store.peers.items()):
            self._notify_peer_ready(rec, reason="restore")
        if _env_open_at_start():
            self.open_enrollment()
        self._started_unix = time.time()
        self._last_reap_mono = 0.0
        self._stop.clear()
        self._thread = threading.Thread(target=self._run, name="edge-enroll-hub", daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=2.0)

    def _run(self) -> None:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(("", self.port))
        sock.settimeout(0.5)
        if self.emit:
            self.emit("enroll_listen", port=self.port, bank=self.bank)
        while not self._stop.is_set():
            try:
                data, addr = sock.recvfrom(65535)
            except socket.timeout:
                self._maybe_reap()
                continue
            except OSError:
                break
            self._maybe_reap()
            try:
                msg = json.loads(data.decode("utf-8"))
            except (UnicodeDecodeError, json.JSONDecodeError):
                continue
            if str(msg.get("type") or "") != "enroll_request":
                continue
            if not verify_envelope(msg, self.key):
                # Silent on crypto failure — probes get no "service exists" clue.
                if self.emit:
                    self.emit("enroll_bad_mac", from_ip=addr[0])
                continue
            # Valid fleet HMAC: reply with accept *or* signed diagnostic deny
            # (wrong bank, window closed, etc.) so commissioning seats know why.
            resp = self._handle_request(msg, addr[0])
            if resp is None:
                continue
            try:
                sock.sendto(json.dumps(resp, separators=(",", ":")).encode("utf-8"), addr)
            except OSError:
                pass
        try:
            sock.close()
        except OSError:
            pass

    def _handle_request(self, msg: dict[str, Any], from_ip: str) -> dict[str, Any] | None:
        pub = str(msg.get("wg_pub") or "").strip()
        bank = int(msg.get("bank", -1))
        if bank != self.bank:
            if self.emit:
                self.emit("enroll_denied", reason="bank_mismatch", from_ip=from_ip)
            return self._reject("bank_mismatch", msg)
        if not pub:
            if self.emit:
                self.emit("enroll_denied", reason="missing_pub", from_ip=from_ip)
            return self._reject("missing_pub", msg)
        allow = self.store.allowlist_pubs()
        open_ok = self.enrollment_open()
        if pub in self.store.peers:
            rec = self.store.peers[pub]
            # Re-apply kernel peer/gretap and hot-add eth in case Mid-session
            # gretap was torn down or bridge ports lagged.
            try:
                spoke_id = int(rec.get("spoke_id") or 0)
                overlay = str(rec.get("overlay") or overlay_ip(self.bank, spoke_id))
                hub_overlay = overlay_ip(self.bank, 1)
                gretap = apply_hub_peer(
                    wg_if=self.wg_if,
                    pub=pub,
                    overlay=overlay,
                    spoke_id=spoke_id,
                    hub_overlay=hub_overlay,
                )
                rec["gretap"] = gretap
                rec["overlay"] = overlay
                self.store.peers[pub] = rec
                self.store.save()
            except Exception as exc:
                if self.emit:
                    self.emit("enroll_apply_error", message=str(exc), from_ip=from_ip, reused=True)
            self._notify_peer_ready(rec, reason="reused")
            return self._accept_response(rec, msg, reused=True)
        if allow and pub not in allow and not open_ok:
            if self.emit:
                self.emit("enroll_denied", reason="not_allowlisted", from_ip=from_ip, pub=pub[:16])
            return self._reject("not_allowlisted", msg)
        if not allow and not open_ok:
            if self.emit:
                self.emit("enroll_denied", reason="window_closed", from_ip=from_ip)
            return self._reject("window_closed", msg)
        try:
            spoke_id = self._allocate_spoke_id(msg)
            overlay = overlay_ip(self.bank, spoke_id)
            hub_overlay = overlay_ip(self.bank, 1)
            gretap = apply_hub_peer(
                wg_if=self.wg_if,
                pub=pub,
                overlay=overlay,
                spoke_id=spoke_id,
                hub_overlay=hub_overlay,
            )
        except Exception as exc:
            if self.emit:
                self.emit("enroll_apply_error", message=str(exc), from_ip=from_ip)
            return self._reject("apply_failed", msg)
        rec = {
            "spoke_id": spoke_id,
            "overlay": overlay,
            "gretap": gretap,
            "term_number": msg.get("term_number"),
            "device_id": msg.get("device_id"),
            "enrolled_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "from_ip": from_ip,
        }
        self.store.peers[pub] = rec
        self.store.save()
        if self.emit:
            self.emit(
                "enroll_accepted",
                spoke_id=spoke_id,
                overlay=overlay,
                gretap=gretap,
                from_ip=from_ip,
            )
        self._notify_peer_ready(rec, reason="enroll")
        return self._accept_response(rec, msg, reused=False)

    def _allocate_spoke_id(self, msg: dict[str, Any]) -> int:
        """Prefer request/EDGE preferred id when free; else next free 2..254."""
        used = {int(p.get("spoke_id", 0)) for p in self.store.peers.values()}
        raw = msg.get("spoke_id")
        if raw is None or raw == "":
            raw = msg.get("preferred_spoke_id")
        if raw is not None and str(raw).strip() != "":
            try:
                want = int(raw)
            except (TypeError, ValueError):
                want = 0
            if 2 <= want <= 254 and want not in used:
                return want
            if 2 <= want <= 254 and self.store is not None and spoke_stale_s() > 0:
                occupant_pub = None
                occupant: dict[str, Any] | None = None
                for pub, rec in list(self.store.peers.items()):
                    if int(rec.get("spoke_id") or 0) == want:
                        occupant_pub, occupant = pub, rec
                        break
                if occupant_pub is not None:
                    hs = 0
                    try:
                        dump = _run(
                            ["wg", "show", self.wg_if, "dump"], check=False
                        ).stdout
                        hs = int(
                            parse_wg_dump(dump).get(occupant_pub, {}).get("handshake") or 0
                        )
                    except OSError:
                        hs = 0
                    if peer_handshake_stale(
                        hs,
                        now=time.time(),
                        stale_s=spoke_stale_s(),
                        grace_until=0.0,
                    ):
                        self._drop_pub(occupant_pub, occupant, reason="replaced")
                        return want
        return self.store.next_spoke_id()

    def _accept_response(
        self, rec: dict[str, Any], req: dict[str, Any], *, reused: bool
    ) -> dict[str, Any]:
        body: dict[str, Any] = {
            "v": 1,
            "type": "enroll_response",
            "ok": True,
            "reused": reused,
            "bank": self.bank,
            "spoke_id": rec["spoke_id"],
            "overlay": rec["overlay"],
            "hub_overlay": overlay_ip(self.bank, 1),
            "gretap": rec.get("gretap"),
            "counter": int(req.get("counter") or 0),
            "ts": int(time.time()),
        }
        body["mac"] = _sign_body({k: v for k, v in body.items() if k != "mac"}, self.key)
        return body

    def _reject(self, reason: str, req: dict[str, Any]) -> dict[str, Any]:
        """Signed diagnostic deny — only reached after fleet HMAC verified."""
        body: dict[str, Any] = {
            "v": 1,
            "type": "enroll_response",
            "ok": False,
            "reason": reason,
            "bank": self.bank,
            "counter": int(req.get("counter") or 0),
            "ts": int(time.time()),
        }
        body["mac"] = _sign_body({k: v for k, v in body.items() if k != "mac"}, self.key)
        return body


def _env_open_at_start() -> bool:
    return os.environ.get("EDGE_OPEN_ENROLL", "0").strip().lower() in ("1", "true", "yes")


def spoke_enroll_once(
    *,
    hub_ip: str,
    bank: int | None = None,
    beacon_counter: int = 0,
    device_id: str | None = None,
) -> dict[str, Any]:
    """Spoke posts enrollment request to hub; returns signed response dict."""
    ident = latch_cabinet_identity()
    b = int(bank if bank is not None else ident.bank_number)
    _priv, pub = ensure_spoke_keypair()
    key = load_fleet_hmac_key()
    body: dict[str, Any] = {
        "v": 1,
        "type": "enroll_request",
        "wg_pub": pub,
        "bank": b,
        "term_number": ident.term_number,
        "machine_role": ident.machine_role,
        "device_id": device_id or os.environ.get("EDGE_DEVICE_ID", socket.gethostname()),
        "beacon_counter": int(beacon_counter),
        "counter": int(time.time()),
        "ts": int(time.time()),
    }
    # Prefer stable field spoke id (EDGE_SPOKE_ID / LAN octet) when free on hub.
    pref = os.environ.get("EDGE_SPOKE_ID", "").strip()
    if pref.isdigit():
        body["preferred_spoke_id"] = int(pref)
        body["spoke_id"] = int(pref)
    body["mac"] = _sign_body({k: v for k, v in body.items() if k != "mac"}, key)
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(5.0)
    port = resolve_enroll_port()
    sock.sendto(json.dumps(body, separators=(",", ":")).encode("utf-8"), (hub_ip, port))
    data, _addr = sock.recvfrom(65535)
    sock.close()
    resp = json.loads(data.decode("utf-8"))
    if not verify_envelope(resp, key):
        raise RuntimeError("enroll response bad mac")
    return resp


def write_thumb_drive_bundle(
    out_dir: str | Path,
    *,
    bank: int,
    hub_overlay: str,
    hub_wg_pub: str,
    hub_endpoint: str,
    spoke_id: int,
    spoke_priv: str,
    spoke_pub: str,
) -> Path:
    """Offline enrollment fallback for seats that cannot reach the hub on WLAN."""
    root = Path(out_dir)
    root.mkdir(parents=True, exist_ok=True)
    overlay = overlay_ip(bank, spoke_id)
    conf = f"""# EtherGold hub-spoke thumb-drive enrollment — bank {bank} spoke {spoke_id}
# Tip/software still lands via Dropbox update.sh or SCP — this file is keys only.
[Interface]
PrivateKey = {spoke_priv}
ListenPort = 51820
Table = off

[Peer]
PublicKey = {hub_wg_pub}
AllowedIPs = {hub_overlay}/32
Endpoint = {hub_endpoint}
PersistentKeepalive = 25
"""
    conf_path = root / f"wg0.conf.spoke{spoke_id:02d}"
    conf_path.write_text(conf, encoding="utf-8")
    meta = {
        "bank": bank,
        "spoke_id": spoke_id,
        "overlay": overlay,
        "hub_overlay": hub_overlay,
        "gretap": gretap_name_for_spoke(spoke_id),
        "spoke_pub": spoke_pub,
        "note": "Install wg conf; on hub add peer + create gretap eg-sNN; tip via update.sh separately.",
    }
    (root / "enroll_meta.json").write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8")
    return conf_path


_hub: EnrollmentHub | None = None


def resolve_hub_uplink_ip() -> str:
    """Best-effort hub clear-WLAN / underlay IP for spoke enroll (no manual EDGE_HUB_IP).

    Order: explicit env → bank-wlan discovery learn → EDGE_UPLINK_IP → wg0 Endpoint.
    """
    for key in (
        "EDGE_HUB_IP",
        "EDGE_UPLINK_IP_LEARNED",
        "EDGE_UPLINK_IP",
    ):
        raw = (os.environ.get(key) or "").strip()
        if raw:
            return raw.split(":")[0].strip()
    endpoint = (os.environ.get("EDGE_HUB_ENDPOINT") or "").strip()
    if endpoint:
        return endpoint.split(":")[0].strip()
    conf = Path(os.environ.get("POG_EDGE_WG_CONF", "/etc/wireguard/wg0.conf"))
    try:
        if conf.is_file():
            for line in conf.read_text(encoding="utf-8", errors="replace").splitlines():
                if "Endpoint" in line and "=" in line:
                    part = line.split("=", 1)[-1].strip()
                    host = part.split(":")[0].strip()
                    if host:
                        return host
    except OSError:
        pass
    return ""


def start_enrollment_hub_for_bridge(
    bridge: Any,
    stop: threading.Event,
    router: Any | None = None,
) -> EnrollmentHub | None:
    """Explicit hub only (EDGE_TOPOLOGY=hub).

    GAMESTATS MACHINE_ROLE=M must NOT auto-start this listener — copper Masters
    (EDGE_MODE=0) and single-hop EDGE seats stay exactly as before.

    When *router* is a live DirectionalEthRouter, accepted enrolls (and restore)
    hot-add ``eg-sNN`` AF_PACKET ports so Master fanout works without restart.

    Hub peer store auto-falls back to a user-writable path when
    ``/etc/ethergold/hub`` is not writable (EG 3.0.1.30) — no manual chmod.
    """
    global _hub
    topo = os.environ.get("EDGE_TOPOLOGY", "").strip().lower()
    if topo != "hub":
        return None
    if os.environ.get("EDGE_ENROLL", "1").strip().lower() in ("0", "false", "no"):
        return None

    def _on_peer(rec: dict[str, Any], *, reason: str) -> None:
        gretap = str(rec.get("gretap") or "")
        spoke_id = rec.get("spoke_id")
        if not gretap or router is None:
            return
        try:
            ok = bool(router.add_port(gretap))
        except Exception as exc:
            bridge.event_log.emit(
                "hub_eth_port_add_error",
                gretap=gretap,
                spoke_id=spoke_id,
                reason=reason,
                message=str(exc),
            )
            return
        bridge.event_log.emit(
            "hub_eth_port_added" if ok else "hub_eth_port_add_failed",
            gretap=gretap,
            spoke_id=spoke_id,
            reason=reason,
            ports=list(router.port_names()) if hasattr(router, "port_names") else None,
        )
        # Keep launcher restart list in sync with live peers.
        try:
            root = Path(
                os.environ.get(
                    "POG_EDGE_ROOT",
                    os.path.expanduser("~/EtherGold/dev/pog-edge"),
                )
            )
            act = root / ".active_spoke_gretaps"
            if ok and act.parent.is_dir():
                lines: list[str] = []
                if act.is_file():
                    lines = [
                        ln.strip()
                        for ln in act.read_text(encoding="utf-8").splitlines()
                        if ln.strip()
                    ]
                if gretap not in lines:
                    lines.append(gretap)
                    act.write_text("\n".join(lines) + "\n", encoding="utf-8")
        except OSError:
            pass

    def _on_gone(rec: dict[str, Any], *, reason: str) -> None:
        gretap = str(rec.get("gretap") or "")
        spoke_id = rec.get("spoke_id")
        if gretap and router is not None and hasattr(router, "remove_port"):
            try:
                router.remove_port(gretap)
            except Exception as exc:
                bridge.event_log.emit(
                    "hub_eth_port_remove_error",
                    gretap=gretap,
                    spoke_id=spoke_id,
                    reason=reason,
                    message=str(exc),
                )
        try:
            ports = list(router.port_names()) if router is not None and hasattr(router, "port_names") else None
        except Exception:
            ports = None
        bridge.event_log.emit(
            "hub_eth_port_removed",
            gretap=gretap or None,
            spoke_id=spoke_id,
            reason=reason,
            ports=ports,
        )
        try:
            root = Path(
                os.environ.get(
                    "POG_EDGE_ROOT",
                    os.path.expanduser("~/EtherGold/dev/pog-edge"),
                )
            )
            act = root / ".active_spoke_gretaps"
            if gretap and act.is_file():
                lines = [
                    ln.strip()
                    for ln in act.read_text(encoding="utf-8").splitlines()
                    if ln.strip() and ln.strip() != gretap
                ]
                act.write_text(("\n".join(lines) + "\n") if lines else "", encoding="utf-8")
        except OSError:
            pass

    try:
        # Ensure writable state before EnrollmentHub constructs peer store.
        ensure_hub_state_dir(emit=bridge.event_log.emit)
        bank = int(os.environ.get("EDGE_BANK", "0") or latch_cabinet_identity().bank_number)
        _hub = EnrollmentHub(
            bank=bank,
            emit=bridge.event_log.emit,
            on_peer_ready=_on_peer,
            on_peer_gone=_on_gone,
        )
        store = _hub.store
        if store is not None:
            bridge.event_log.emit(
                "hub_peer_store_ready",
                path=str(store.path),
                peer_count=len(store.peers),
            )
        _hub.start()
    except Exception as exc:
        # Never take the bridge down over provisioning (e.g. /etc/ethergold not writable).
        bridge.event_log.emit("enroll_hub_start_error", message=str(exc))
        _hub = None
        return None

    def _watch() -> None:
        stop.wait()
        if _hub is not None:
            _hub.stop()

    threading.Thread(target=_watch, name="enroll-stop", daemon=True).start()
    return _hub


def start_spoke_enroll_for_bridge(bridge: Any, stop: threading.Event) -> None:
    """Spoke: UDP enroll retries so a running hub learns this seat live.

    EG 3.0.1.30: never require manual EDGE_HUB_IP. Poll bank-wlan discovery
    (EDGE_UPLINK_IP_LEARNED) and env until the hub appears; keep retrying
    while the bridge runs (hub may start later than spokes).
    """
    topo = os.environ.get("EDGE_TOPOLOGY", "").strip().lower()
    if topo != "spoke":
        return
    if os.environ.get("EDGE_ENROLL", "1").strip().lower() in ("0", "false", "no"):
        return

    def _worker() -> None:
        attempt = 0
        last_wait_log = 0.0
        delays = (2.0, 5.0, 10.0, 15.0, 30.0)
        while not stop.is_set():
            hub_ip = resolve_hub_uplink_ip()
            if not hub_ip:
                now = time.time()
                if now - last_wait_log >= 30.0:
                    last_wait_log = now
                    bridge.event_log.emit(
                        "spoke_enroll_wait",
                        reason="no_hub_ip",
                        note="waiting_bank_wlan_or_env",
                    )
                if stop.wait(5.0):
                    return
                continue
            attempt += 1
            try:
                # Ensure user-writable keys dir before enroll (Permission denied
                # on /etc/ethergold/spoke is handled inside ensure_spoke_keypair).
                ensure_spoke_keys_dir(emit=bridge.event_log.emit)
                resp = spoke_enroll_once(hub_ip=hub_ip)
            except Exception as exc:
                bridge.event_log.emit(
                    "spoke_enroll_error",
                    hub_ip=hub_ip,
                    attempt=attempt,
                    message=str(exc),
                    spoke_keys_dir=os.environ.get("EDGE_SPOKE_KEYS_DIR"),
                )
                delay = delays[min(attempt - 1, len(delays) - 1)]
                if stop.wait(delay):
                    return
                continue
            ok = bool(resp.get("ok"))
            bridge.event_log.emit(
                "spoke_enroll_response",
                hub_ip=hub_ip,
                ok=ok,
                reason=resp.get("reason"),
                spoke_id=resp.get("spoke_id"),
                overlay=resp.get("overlay"),
                gretap=resp.get("gretap"),
                reused=resp.get("reused"),
                attempt=attempt,
            )
            if ok:
                return
            delay = delays[min(attempt - 1, len(delays) - 1)]
            if stop.wait(delay):
                return

    try:
        threading.Thread(target=_worker, name="edge-enroll-spoke", daemon=True).start()
    except Exception as exc:
        bridge.event_log.emit("spoke_enroll_start_error", message=str(exc))


if __name__ == "__main__":
    import argparse

    ap = argparse.ArgumentParser(description="EtherGold hub-spoke enrollment helpers")
    sub = ap.add_subparsers(dest="cmd", required=True)
    p_open = sub.add_parser("open-window", help="Print open-enrollment seconds (hub operator)")
    p_open.add_argument("--seconds", type=float, default=DEFAULT_OPEN_WINDOW_S)
    p_spoke = sub.add_parser("spoke-request", help="Send one enrollment request")
    p_spoke.add_argument("--hub-ip", required=True)
    p_thumb = sub.add_parser("thumb-bundle", help="Write offline thumb-drive enrollment bundle")
    p_thumb.add_argument("--out", required=True)
    p_thumb.add_argument("--bank", type=int, default=0)
    p_thumb.add_argument("--spoke-id", type=int, required=True)
    p_thumb.add_argument("--hub-overlay", default="10.99.0.1")
    p_thumb.add_argument("--hub-wg-pub", required=True)
    p_thumb.add_argument("--hub-endpoint", required=True)
    args = ap.parse_args()
    if args.cmd == "open-window":
        print(f"EDGE_OPEN_ENROLL=1 EDGE_OPEN_ENROLL_S={int(args.seconds)}")
    elif args.cmd == "spoke-request":
        print(json.dumps(spoke_enroll_once(hub_ip=args.hub_ip), indent=2))
    elif args.cmd == "thumb-bundle":
        priv, pub = ensure_spoke_keypair()
        path = write_thumb_drive_bundle(
            args.out,
            bank=args.bank,
            hub_overlay=args.hub_overlay,
            hub_wg_pub=args.hub_wg_pub,
            hub_endpoint=args.hub_endpoint,
            spoke_id=args.spoke_id,
            spoke_priv=priv,
            spoke_pub=pub,
        )
        print(path)