kent/announcer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Mesh: sonene melder seg selv; mDNS på LAN, NetBird-provisjonering

Browse source 
Sonene trenger ikke stå i [zones] på sentralen. En sone sender hjerteslag
hvert 30. s (POST /api/mesh/register med navn, egen adresse og
mesh-nøkkelen), dukker opp i lista med en gang, og tas ut etter 90 s uten
slag. Riktig nøkkel er godkjenningen. Statisk [zones] vinner over et navn
som melder seg (409) – en skrivefeil i config skal ikke kunne overstyres av
en boks på nettet.

Hvorfor push og ikke pull fra NetBird: netbird status --json gir navn, IP og
status per peer, men ikke grupper. Pull ville krevd en service-token mot
management-serveren og virket bare over NetBird. Push virker likt over
NetBird og LAN, uten token, og gir liveness gratis.

Sonen finner sentralen lagvis: ANNOUNCER_SENTRAL hvis satt (NetBird-DNS-navn,
konvensjon announcer-<navn>.skynet.corp), ellers mDNS (_announcer._tcp,
discovery.py). Egen adresse regnes ut som ruten mot sentralen – over NetBird
blir det wt0-adressen, på LAN LAN-adressen – med ANNOUNCER_URL som
overstyring. Hjerteslag hvert 5. s til første OK, så oppstart ikke tar et
halvt minutt.

Containeren bruker nå vertens nett. mDNS-multicast når ikke LAN-et fra
bridge-nettet, så det var ingen vei utenom – og det ga tilbake ufw: det
gamle Docker-smutthullet er borte, portregler virker igjen, og BIND_IP er
appens eget --host. Appen lytter alltid også på 127.0.0.1 for helsesjekken.

install.sh tar NETBIRD_SETUP_KEY: installerer NetBird, melder maskina inn
som announcer-<navn>, venter på wt0, binder til NetBird-adressen.
Management-URL leses fra /etc/netbird/config.json når den finnes.

Rollen avgjøres av config (Config.role): sentral hvis [zones] eller
ANNOUNCER_ROLE=sentral, sone hvis bare nøkkelen er satt, ellers som før.
Dispatcher og ZonePoller slår opp soner ved hver bruk, siden de kommer og
går. register følger, som play, aldri med «alt åpent». Flask sorterte
JSON-nøkler alfabetisk, så sonene kom i feil rekkefølge i /api/status –
app.json.sort_keys = False.

25 nye tester, 173 totalt. Verifisert i compose på host-nett: to soner
meldte seg innen 4 s uten noe [zones], fordeling til en av dem havnet i kø,
avahi-browse på verten ser sentralen, en stoppet sone ble tatt ut etter 90 s
og var tilbake 2 s etter start. Underveis fant live-testen at testkonfigen
fortsatt hadde statiske a/b – og sentralen avviste hjerteslagene med 409,
som den skal.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Kent 2026-10-09 02:50:25 +02:00
parent c4a15799dd
commit 7c521f8d1f
21 changed files with 1106 additions and 137 deletions
Download patch file Download diff file Expand all files Collapse all files

240
mesh.py
View file

@ -25,6 +25,7 @@ import logging
import queue
import re
import secrets
import socket
import threading
import time
import urllib.error
@ -144,9 +145,12 @@ class Dispatcher:
send(zone, path, dispatch) -> (status, item_id, error)
"""
def __init__(self, zones: dict[str, Zone], *, synthesize: Callable, play_local: Callable,
send: Callable, timeout: float = 5.0, on_finished: Callable | None = None) -> None:
self.zones = zones
def __init__(self, zones: "dict[str, Zone] | Callable[[], dict[str, Zone]]", *,
synthesize: Callable, play_local: Callable, send: Callable,
timeout: float = 5.0, on_finished: Callable | None = None) -> None:
# Soner kommer og går (hjerteslag), så de slås opp ved hver bruk.
# Et fast dict godtas også, for tester og statisk config.
self._zones = zones if callable(zones) else (lambda: zones)
self.synthesize = synthesize
self.play_local = play_local
self.send = send
@ -158,13 +162,18 @@ class Dispatcher:
self._thread: threading.Thread | None = None
# -- oppslag -----------------------------------------------------------
@property
def zones(self) -> dict[str, Zone]:
return self._zones()
def names(self) -> list[str]:
return list(self.zones)
def resolve(self, requested: Any) -> list[str]:
"""Tom/None = alle. Ukjent navn kaster ValueError før noe sendes."""
zones = self.zones
if not requested:
return self.names()
return list(zones)
if isinstance(requested, str):
requested = [requested]
if not isinstance(requested, (list, tuple)):
@ -172,7 +181,7 @@ class Dispatcher:
out: list[str] = []
for name in requested:
name = str(name).strip()
if name not in self.zones:
if name not in zones:
raise ValueError(f"Ukjent sone: {name}")
if name not in out:
out.append(name)
@ -227,8 +236,13 @@ class Dispatcher:
return d
threads: list[threading.Thread] = []
zones = self.zones
for name, result in d.zones.items():
zone = self.zones[name]
zone = zones.get(name)
if zone is None:
# Meldt av mellom innlevering og fordeling.
result.status, result.error = "unreachable", "sonen er borte"
continue
if zone.local:
try:
result.item_id = self.play_local(d, path)
@ -357,19 +371,21 @@ class ZonePoller:
eget anlegg aldri er to sekunder på etterskudd.
"""
def __init__(self, zones: dict[str, Zone], *, fetch: Callable, interval: float = 2.0,
def __init__(self, zones: "dict[str, Zone] | Callable[[], dict[str, Zone]]", *,
fetch: Callable, interval: float = 2.0,
local_status: Callable[[], dict] | None = None) -> None:
self.zones = zones
self._zones = zones if callable(zones) else (lambda: zones)
self.fetch = fetch
self.interval = float(interval)
self.local_status = local_status
self.lock = threading.Lock()
self.cache: dict[str, dict] = {
z.name: {"reachable": False, "error": "ikke spurt ennå"}
for z in zones.values() if not z.local
}
self.cache: dict[str, dict] = {}
self._thread: threading.Thread | None = None
@property
def zones(self) -> dict[str, Zone]:
return self._zones()
def poll_once(self) -> None:
remote = [z for z in self.zones.values() if not z.local]
results: dict[str, dict] = {}
@ -389,6 +405,9 @@ class ZonePoller:
for z in remote:
self.cache[z.name] = results.get(z.name, {"reachable": False, "error": "svarte ikke"})
self.cache[z.name]["checked"] = time.time()
# Soner som er meldt av skal ikke bli liggende i cachen.
for gone in set(self.cache) - {z.name for z in remote}:
del self.cache[gone]
def snapshot(self) -> dict[str, dict]:
out: dict[str, dict] = {}
@ -397,7 +416,7 @@ class ZonePoller:
out[z.name] = {"reachable": True, **(self.local_status() if self.local_status else {})}
else:
with self.lock:
out[z.name] = dict(self.cache.get(z.name, {"reachable": False}))
out[z.name] = dict(self.cache.get(z.name, {"reachable": False, "error": "ikke spurt ennå"}))
return out
def start(self) -> None:
@ -412,3 +431,198 @@ class ZonePoller:
except Exception: # noqa: BLE001
log.exception("sonepolleren feilet")
time.sleep(self.interval)
# --------------------------------------------------------------------------
# Register: sonene sentralen kjenner akkurat nå
# --------------------------------------------------------------------------
@dataclass
class Registered:
zone: Zone
first_seen: float
last_seen: float
version: str = ""
class Registry:
"""Sonene en sentral kjenner: eget anlegg, statisk [zones], og de som har
meldt seg med hjerteslag. En som slutter å slå, tas ut etter stale_seconds.
Statisk config vinner alltid over et navn som melder seg – en skrivefeil
i config skal ikke kunne overstyres av en boks på nettet.
"""
def __init__(self, static: dict[str, Zone], *, local: Zone | None = None,
stale_seconds: float = 90.0) -> None:
self.local = local
self.static = dict(static)
self.stale = float(stale_seconds)
self.registered: dict[str, Registered] = {}
self.lock = threading.Lock()
def register(self, name: str, url: str, version: str = "") -> Zone:
name = str(name).strip()
if not ZONE_NAME.fullmatch(name):
raise ValueError(f"ugyldig sonenavn {name!r}")
if self.local and name == self.local.name:
raise ValueError(f"{name} er sentralens eget navn")
if name in self.static:
raise ValueError(f"{name} er satt i [zones] på sentralen")
url = str(url).strip().rstrip("/")
if not url.startswith(("http://", "https://")):
raise ValueError("url må være http(s)://…")
zone = Zone(name, url)
now = time.time()
with self.lock:
prev = self.registered.get(name)
if prev and prev.zone == zone:
prev.last_seen, prev.version = now, version
else:
if prev:
log.info("sone %s byttet adresse: %s -> %s", name, prev.zone.url, url)
else:
log.info("ny sone meldte seg: %s på %s", name, url)
self.registered[name] = Registered(zone, now, now, version)
return zone
def prune(self, now: float | None = None) -> list[str]:
now = time.time() if now is None else now
with self.lock:
gone = [n for n, r in self.registered.items() if now - r.last_seen > self.stale]
for n in gone:
del self.registered[n]
for n in gone:
log.warning("sone %s har ikke meldt seg på %d s – tatt ut", n, int(self.stale))
return gone
def zones(self) -> dict[str, Zone]:
out: dict[str, Zone] = {}
if self.local:
out[self.local.name] = self.local
out.update(self.static)
with self.lock:
for n in sorted(self.registered):
out[n] = self.registered[n].zone
return out
def info(self) -> dict[str, dict]:
"""Til UI-et: hvor sonen kommer fra, og når den sist sa noe."""
out: dict[str, dict] = {}
if self.local:
out[self.local.name] = {"source": "local", "url": ""}
for n, z in self.static.items():
out[n] = {"source": "static", "url": z.url}
with self.lock:
for n in sorted(self.registered):
r = self.registered[n]
out[n] = {"source": "registered", "url": r.zone.url,
"last_seen": r.last_seen, "version": r.version}
return out
# --------------------------------------------------------------------------
# Hjerteslag: sonens side
# --------------------------------------------------------------------------
def local_ip_towards(host: str, port: int = 8090) -> str:
"""IP-en denne maskina bruker for å nå host – altså den sentralen må
svare til. Ingen pakke sendes; connect() på en UDP-socket velger bare
rute. Over NetBird blir det wt0-adressen, på LAN LAN-adressen."""
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.connect((host, port))
return sock.getsockname()[0]
def post_register(sentral_url: str, *, name: str, url: str, secret: str,
version: str = "", timeout: float = 5.0) -> tuple[bool, str]:
body = json.dumps({"name": name, "url": url, "version": version}).encode()
req = urllib.request.Request(
f"{sentral_url.rstrip('/')}/api/mesh/register",
data=body,
headers={"Content-Type": "application/json", "X-Mesh-Key": secret, "Accept": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=timeout) as r: # noqa: S310
r.read()
return True, ""
except urllib.error.HTTPError as e:
try:
msg = json.loads(e.read().decode()).get("error", "")
except Exception: # noqa: BLE001
msg = ""
return False, f"{e.code} {msg or e.reason}".strip()
except (urllib.error.URLError, TimeoutError, OSError) as e:
return False, str(getattr(e, "reason", e))[:120]
class Heartbeat:
"""Melder denne sonen hos sentralen med jevne mellomrom.
sentral() gir sentralens URL akkurat nå – fra config, eller fra mDNS –
eller None når ingen er kjent. Da venter vi og prøver igjen; en sone skal
kunne starte før sentralen finnes.
"""
def __init__(self, *, name: str, port: int, secret: str, sentral: Callable[[], str | None],
url: str = "", interval: float = 30.0, version: str = "",
post: Callable = post_register) -> None:
self.name = name
self.port = int(port)
self.secret = secret
self.sentral = sentral
self.fixed_url = url.strip().rstrip("/")
self.interval = float(interval)
self.version = version
self.post = post
self.last_ok: float | None = None
self.last_error = ""
self.last_sentral = ""
self._thread: threading.Thread | None = None
def my_url(self, sentral_url: str) -> str:
if self.fixed_url:
return self.fixed_url
host = urllib.request.urlparse(sentral_url).hostname or "127.0.0.1"
return f"http://{local_ip_towards(host, self.port)}:{self.port}"
def beat_once(self) -> bool:
sentral = self.sentral()
if not sentral:
self.last_error = "ingen sentral kjent"
return False
try:
url = self.my_url(sentral)
except OSError as e:
self.last_error = f"finner ikke egen adresse mot {sentral}: {e}"
return False
ok, err = self.post(sentral, name=self.name, url=url, secret=self.secret, version=self.version)
if ok:
if self.last_sentral != sentral:
log.info("meldt hos sentralen %s som %s (%s)", sentral, self.name, url)
self.last_ok, self.last_error, self.last_sentral = time.time(), "", sentral
else:
if err != self.last_error:
log.warning("fikk ikke meldt meg hos %s: %s", sentral, err)
self.last_error = err
return ok
def status(self) -> dict:
return {"sentral": self.last_sentral or None, "last_ok": self.last_ok, "error": self.last_error}
def start(self) -> None:
if self._thread is None:
self._thread = threading.Thread(target=self._loop, daemon=True, name="heartbeat")
self._thread.start()
def _loop(self) -> None:
while True:
try:
self.beat_once()
except Exception: # noqa: BLE001
log.exception("hjerteslaget feilet")
# Raskere når vi ikke har fått kontakt ennå, så oppstart ikke tar et halvt minutt.
time.sleep(self.interval if self.last_ok else min(self.interval, 5.0))