diff --git a/README.en.md b/README.en.md index 3408835..46d069a 100644 --- a/README.en.md +++ b/README.en.md @@ -83,6 +83,10 @@ everything from a **password-protected web UI** — no CLI or hand-edited config - **REST API:** a JSON API (`/api/v1`) for automation — mapping CRUD, start/stop/restart, status and ports; **bearer-token** auth (with a role — `viewer` is read-only); OpenAPI 3.0 (`/api/v1/openapi.json`). The token is generated in Settings. +- **LAN cluster (fleet view):** nodes on the same network **auto-discover** each other + (signed UDP broadcast — no mDNS) and one node shows **every node's mappings in a single + read-only table**, each row tagged with its host (name + IP). Set the **same shared key** + on every node so they trust each other. Off by default; enable under Settings → LAN cluster. - **Deployment:** official **Docker** image + `docker-compose`; **systemd** unit; Linux+Windows × Python 3.10–3.13 **CI** (GitHub Actions). - **MQTT publishing (optional):** publish each mapping's serial lines to an MQTT diff --git a/README.md b/README.md index 70def72..7778f5a 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,11 @@ düzenlemeye gerek yok. - **REST API:** otomasyon için JSON API (`/api/v1`) — eşleme CRUD, başlat/durdur/yeniden, durum ve portlar; **bearer-token** kimlik doğrulama (rol seçilebilir — `viewer` salt-okunur); OpenAPI 3.0 (`/api/v1/openapi.json`). Token, Ayarlar'dan üretilir. +- **LAN cluster (filo görünümü):** aynı ağdaki düğümler birbirini **otomatik bulur** + (imzalı UDP broadcast — mDNS yok) ve bir düğüm **tüm düğümlerin eşlemelerini tek + salt-okunur tabloda** gösterir; her satır hangi bilgisayara ait olduğunu (ad + IP) + belirtir. Düğümlerin birbirine güvenmesi için her birine **aynı paylaşılan anahtar** + girilir. Varsayılan kapalı; Ayarlar → LAN cluster'dan açılır. - **Dağıtım:** resmi **Docker** imajı + `docker-compose`; **systemd** birimi; Linux+Windows × Python 3.10–3.13 **CI** (GitHub Actions). - **MQTT yayınlama (opsiyonel):** seri satırlarını eşleme başına bir MQTT broker'ına diff --git a/ROADMAP.md b/ROADMAP.md index 8aa7b79..e9cacaf 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -82,7 +82,11 @@ - **Sparkplug B** edge payloads (Modbus register + MQTT plumbing already in place) - Modbus: write support (FC 5/6/15/16), per-point MQTT→register control, RTU inter-frame tuning; RS-485 hardware auto-RTS (`TIOCSRS485`) UI -- Multi-host **fleet dashboard** (manage several instances; subscription tier) +- ~~LAN **cluster dashboard**: nodes auto-discover each other (signed UDP broadcast, + no mDNS) and one node shows every node's mappings in a single read-only table, each + row tagged with its host (name + IP)~~ ✅ shipped (v2.5a) +- Cluster: remote control (start/stop/edit a peer's mappings from one screen) + per-node + health/uptime in the unified view - classic `ser2net.yaml` import for migration ### Icebox / conditional diff --git a/app/config.py b/app/config.py index d741064..ba8cbf1 100644 --- a/app/config.py +++ b/app/config.py @@ -526,6 +526,44 @@ def validate(self) -> None: raise ConfigError(f"OIDC default role must be one of {ROLES} or blank.") +@dataclass +class ClusterSettings: + """LAN cluster: instances discover each other via signed UDP broadcast beacons + and one node aggregates every node's mappings into a single read-only view. + Opt-in: disabled until `enabled` is set AND a shared `key` is configured. The + key both signs beacons (so only same-key nodes trust each other) and guards the + peer-facing status endpoint. `advertise_ip` overrides the auto-detected LAN IP + that peers use to reach this node's web UI (blank => auto).""" + enabled: bool = False + key: str = "" # shared secret (PSK); empty => cluster off + discovery_port: int = 41750 # UDP port for broadcast beacons + advertise_ip: str = "" # override advertised web-UI IP (blank => auto-detect) + + @staticmethod + def from_dict(d: dict[str, Any]) -> "ClusterSettings": + d = dict(d or {}) + known = {f.name for f in dataclasses.fields(ClusterSettings)} + return ClusterSettings(**{k: v for k, v in d.items() if k in known}) + + @property + def active(self) -> bool: + return bool(self.enabled and self.key.strip()) + + def validate(self) -> None: + if not self.enabled: + return + if not self.key.strip(): + raise ConfigError("Cluster is enabled but no shared key is set.") + if not (1 <= int(self.discovery_port) <= 65535): + raise ConfigError("Cluster discovery port must be 1..65535.") + if self.advertise_ip.strip(): + try: + ipaddress.ip_address(self.advertise_ip.strip()) + except ValueError: + raise ConfigError( + f"Cluster advertise IP is not a valid address: {self.advertise_ip}") from None + + @dataclass class User: """A web-UI account. `role` gates what the user may do; `pwd_version` is bumped @@ -563,6 +601,9 @@ class AppConfig: users: list[User] = field(default_factory=list) ldap: LdapSettings = field(default_factory=LdapSettings) oidc: OidcSettings = field(default_factory=OidcSettings) + cluster: ClusterSettings = field(default_factory=ClusterSettings) + # Stable per-node identity for the LAN cluster (generated once, persisted). + instance_id: str = field(default_factory=lambda: secrets.token_hex(8)) defaults: dict[str, Any] = field(default_factory=dict) # serial defaults mappings: list[MappingConfig] = field(default_factory=list) @@ -587,6 +628,8 @@ def from_dict(d: dict[str, Any]) -> "AppConfig": users=users, ldap=LdapSettings.from_dict(d.get("ldap", {})), oidc=OidcSettings.from_dict(d.get("oidc", {})), + cluster=ClusterSettings.from_dict(d.get("cluster", {})), + instance_id=d.get("instance_id") or secrets.token_hex(8), defaults=d.get("defaults", {}) or {}, mappings=[MappingConfig.from_dict(m) for m in d.get("mappings", [])], ) @@ -603,6 +646,8 @@ def to_dict(self) -> dict[str, Any]: "users": [asdict(u) for u in self.users], "ldap": asdict(self.ldap), "oidc": asdict(self.oidc), + "cluster": asdict(self.cluster), + "instance_id": self.instance_id, "defaults": self.defaults, "mappings": [m.to_dict() for m in self.mappings], } @@ -646,6 +691,7 @@ def validate(self) -> None: self.ldap.validate() self.oidc.validate() + self.cluster.validate() seen: dict[tuple[str, str, int], str] = {} # (proto, bind_ip, port) -> name serial_owner: dict[str, tuple[str, str]] = {} # device -> (mapping_id, name) diff --git a/app/engine/cluster.py b/app/engine/cluster.py new file mode 100644 index 0000000..c39e0da --- /dev/null +++ b/app/engine/cluster.py @@ -0,0 +1,217 @@ +"""LAN cluster discovery — nodes find each other and share a read-only view. + +Each ser2net instance periodically broadcasts a small UDP *beacon* (its node id, +hostname, and the IP:port + scheme of its web UI) signed with HMAC-SHA256 over +the shared cluster key. Every instance listens on the same UDP port and records +beacons whose signature verifies with its own key — so only nodes configured with +the *same* key trust each other. Peers not heard from within ``PEER_TTL`` seconds +expire. + +Deliberately **not** mDNS/zeroconf and no extra dependencies: just an asyncio +DatagramProtocol + a periodic broadcast task. This module only does *discovery* +(who is out there and how to reach their web UI). Aggregating a peer's mappings +is an HTTP GET to that peer's ``/api/cluster/local`` (guarded by the shared key); +``fetch_peer`` performs it off the event loop. The browser only ever talks to the +node it logged into — that node fans out to its peers server-side. +""" +from __future__ import annotations + +import asyncio +import contextlib +import hashlib +import hmac +import json +import socket +import ssl +import time +import urllib.request + +from . import netinfo + +BEACON_INTERVAL = 5.0 # seconds between outbound beacons +PEER_TTL = 20.0 # a peer is dropped if not heard from within this window +_LOOPBACK = ("0.0.0.0", "::", "127.0.0.1", "localhost", "") + + +def _sign(key: str, body: str) -> str: + return hmac.new(key.encode("utf-8"), body.encode("utf-8"), hashlib.sha256).hexdigest() + + +class _BeaconProtocol(asyncio.DatagramProtocol): + def __init__(self, on_datagram): + self._on_datagram = on_datagram + + def datagram_received(self, data, addr): + self._on_datagram(data, addr) + + def error_received(self, exc): # ICMP port-unreachable etc. — ignore + pass + + +class ClusterDiscovery: + """Owns the UDP discovery socket + beacon loop + live peer table. Reads the + cluster config live from AppConfig so a settings change is picked up on the + next restart (start/stop are driven by AppState's engine lifecycle).""" + + def __init__(self, config, logger): + self.config = config # AppConfig (cluster + admin_ui + instance_id, read live) + self.log = logger + self._transport = None + self._beacon_task: asyncio.Task | None = None + self._peers: dict[str, dict] = {} # id -> {id,name,ip,port,scheme,last_seen} + self._running = False + + # ----- config helpers ----- + @property + def cluster(self): + return self.config.cluster + + def advertised(self) -> tuple[str, int, str]: + """(ip, port, scheme) other nodes use to reach THIS node's web UI.""" + c = self.cluster + ui = self.config.admin_ui + ip = c.advertise_ip.strip() + if not ip: + bind = (ui.bind_ip or "").strip() + if bind not in _LOOPBACK: + ip = bind + else: + ip = netinfo.primary_lan_ip() or "127.0.0.1" + scheme = "https" if ui.tls_enabled else "http" + return ip, int(ui.port), scheme + + # ----- beacon (de)serialization ----- + def _make_beacon(self) -> bytes: + ip, port, scheme = self.advertised() + payload = {"v": 1, "id": self.config.instance_id, "name": socket.gethostname(), + "ip": ip, "port": port, "scheme": scheme, "t": int(time.time())} + body = json.dumps(payload, separators=(",", ":"), sort_keys=True) + return json.dumps({"d": body, "s": _sign(self.cluster.key, body)}).encode("utf-8") + + def handle_datagram(self, data: bytes, addr) -> None: + """Verify a received beacon and record/refresh the peer. Pure (no I/O) so it + can be called directly from tests to inject a peer.""" + key = self.cluster.key + if not key: + return + try: + env = json.loads(data.decode("utf-8")) + body, sig = env["d"], env["s"] + except (ValueError, KeyError, AttributeError, TypeError): + return + if not hmac.compare_digest(_sign(key, body), str(sig)): + return # different key (or tampered) — not part of our cluster + try: + p = json.loads(body) + except ValueError: + return + pid = str(p.get("id", "")) + if not pid or pid == self.config.instance_id: + return # ignore our own beacon + ip = str(p.get("ip", "")).strip() + if ip in _LOOPBACK and addr: + ip = addr[0] # advertised IP unusable — fall back to the packet's source + scheme = p.get("scheme") if p.get("scheme") in ("http", "https") else "http" + self._peers[pid] = { + "id": pid, "name": str(p.get("name", "?")), "ip": ip, + "port": int(p.get("port", 0) or 0), "scheme": scheme, "last_seen": time.time(), + } + + def peers(self) -> list[dict]: + """Live peers (TTL-pruned), sorted by name then ip.""" + now = time.time() + live = [p for p in self._peers.values() if now - p["last_seen"] <= PEER_TTL] + self._peers = {p["id"]: p for p in live} + return sorted(live, key=lambda p: (p["name"].lower(), p["ip"])) + + # ----- lifecycle ----- + async def start(self) -> None: + if self._running or not self.cluster.active: + return + loop = asyncio.get_running_loop() + port = int(self.cluster.discovery_port) + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + with contextlib.suppress(AttributeError, OSError): + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + try: + sock.bind(("", port)) + except OSError as e: + sock.close() + self.log(f"cluster: could not bind UDP {port}: {e} — discovery disabled") + return + self._transport, _ = await loop.create_datagram_endpoint( + lambda: _BeaconProtocol(self.handle_datagram), sock=sock) + self._running = True + self._beacon_task = asyncio.create_task(self._beacon_loop(), name="cluster-beacon") + self.log(f"cluster: discovery on UDP {port} as '{socket.gethostname()}' " + f"({self.config.instance_id[:8]})") + + async def stop(self) -> None: + self._running = False + if self._beacon_task is not None: + self._beacon_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._beacon_task + self._beacon_task = None + if self._transport is not None: + with contextlib.suppress(Exception): + self._transport.close() + self._transport = None + self._peers.clear() + + async def _beacon_loop(self) -> None: + while True: + try: + self._broadcast() + except Exception as e: # never let the loop die on a transient send error + self.log(f"cluster: beacon send failed: {e}") + await asyncio.sleep(BEACON_INTERVAL) + + def _broadcast(self) -> None: + if self._transport is None: + return + msg = self._make_beacon() + port = int(self.cluster.discovery_port) + for addr in self._broadcast_addrs(): + with contextlib.suppress(OSError): + self._transport.sendto(msg, (addr, port)) + + @staticmethod + def _broadcast_addrs() -> list[str]: + """Global broadcast plus each NIC's directed broadcast (some networks only + deliver the subnet-directed form). Best-effort; psutil is optional.""" + addrs = ["255.255.255.255"] + try: + import psutil + for nic in psutil.net_if_addrs().values(): + for a in nic: + if getattr(a, "family", None) == socket.AF_INET and getattr(a, "broadcast", None): + addrs.append(a.broadcast) + except Exception: + pass + seen, out = set(), [] + for a in addrs: + if a and a not in seen: + seen.add(a) + out.append(a) + return out + + # ----- peer aggregation (HTTP) ----- + async def fetch_peer(self, peer: dict, timeout: float = 2.5) -> dict | None: + url = f"{peer['scheme']}://{peer['ip']}:{peer['port']}/api/cluster/local" + return await asyncio.to_thread(self._http_get_json, url, timeout) + + def _http_get_json(self, url: str, timeout: float) -> dict | None: + req = urllib.request.Request(url, headers={"X-Cluster-Key": self.cluster.key}) + ctx = None + if url.startswith("https"): + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE # cluster trust is the shared key, not the cert + try: + with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r: + return json.loads(r.read().decode("utf-8")) + except Exception: + return None diff --git a/app/state.py b/app/state.py index 5b7ba36..80071a5 100644 --- a/app/state.py +++ b/app/state.py @@ -21,6 +21,7 @@ from logging.handlers import RotatingFileHandler from .config import AppConfig, ConfigStore, lock_down_dir +from .engine.cluster import ClusterDiscovery from .engine.portlist import PortWatcher from .engine.supervisor import Supervisor from .web.auth import LoginRateLimiter @@ -99,6 +100,7 @@ def __init__(self, store: ConfigStore, config: AppConfig): self._mapping_loggers: dict[str, logging.Logger] = {} self.supervisor = Supervisor(logger=self.log, logger_factory=self.make_mapping_logger) self.ports = PortWatcher(interval=2.0) + self.cluster = ClusterDiscovery(self.config, self.log) self.rate_limiter = LoginRateLimiter() self.config_lock = asyncio.Lock() self.started_at = time.time() @@ -284,6 +286,7 @@ async def start_engine(self) -> None: self.prune_mapping_logs() # drop logs for mappings that no longer exist await self.ports.start() await self.supervisor.start_all(self.config) + await self.cluster.start() # no-op unless the LAN cluster is enabled + keyed # run an immediate pass, then hourly, to enforce log retention self._maint_task = asyncio.create_task(self._log_maintenance_loop(), name="log-maintenance") @@ -293,5 +296,6 @@ async def stop_engine(self) -> None: with _suppress(asyncio.CancelledError): await self._maint_task self._maint_task = None + await self.cluster.stop() await self.supervisor.stop_all() await self.ports.stop() diff --git a/app/web/routes.py b/app/web/routes.py index a4d03f4..33c795e 100644 --- a/app/web/routes.py +++ b/app/web/routes.py @@ -10,10 +10,12 @@ import asyncio import contextlib +import hmac import json import os import re import secrets +import socket import subprocess from urllib.parse import urlparse @@ -24,7 +26,8 @@ from starlette.staticfiles import StaticFiles from ..config import ( - ROLE_RANK, ROLES, ConfigError, LdapSettings, MappingConfig, OidcSettings, User, + ROLE_RANK, ROLES, ClusterSettings, ConfigError, LdapSettings, MappingConfig, + OidcSettings, User, ) from ..engine import netinfo from . import auth, ldap_auth, oidc_auth @@ -36,6 +39,20 @@ _TRIGGER = {"HX-Trigger": "refreshMappings"} +def _mapping_kind_label(m) -> str: + if m.kind == "serialbridge": + return "serial↔serial" + return f"{m.network.mode}/{m.network.protocol}" + + +def _mapping_endpoint(m) -> str: + if m.kind == "serialbridge": + return f"{m.serial.port} ↔ {m.serial_b.port}" + if m.network.mode == "client": + return f"→ {m.network.remote_host}:{m.network.remote_port}" + return f"{m.network.bind_ip}:{m.network.port}" + + def build_routes(templates, state, static_dir): def render(request, name, status_code=200, headers=None, **ctx): return templates.TemplateResponse(request, name, ctx, @@ -222,6 +239,7 @@ async def dashboard(request): mappings=state.config.mappings, statuses=state.supervisor.all_status(), admin=state.config.admin_ui, + cluster_on=state.config.cluster.active, me=current_user(request), can_edit=can_edit(request), ) @@ -234,6 +252,7 @@ async def settings_get(request, ok=None, error=None, new_api_token=None): me=current_user(request), users=state.config.users, roles=ROLES, ldap=state.config.ldap, ldap_lib=_module_present("ldap3"), oidc=state.config.oidc, oidc_lib=_module_present("authlib"), + cluster=state.config.cluster, instance_id=state.config.instance_id, uptime=int(state.started_at)) async def settings_password_post(request): @@ -607,6 +626,93 @@ async def api_status(request): statuses=state.supervisor.all_status(), can_edit=can_edit(request)) + # ---------------- LAN cluster (federated read-only view) ---------------- + def local_node_payload() -> dict: + """This node's identity + a status row per mapping — the unit other nodes + aggregate. No secrets: just what the unified dashboard table shows.""" + cfg = state.config + ip, port, scheme = state.cluster.advertised() + statuses = state.supervisor.all_status() + maps = [] + for m in cfg.mappings: + st = statuses.get(m.id) or {} + maps.append({ + "id": m.id, "name": m.name, "enabled": m.enabled, + "kind": _mapping_kind_label(m), "endpoint": _mapping_endpoint(m), + "serial": m.serial.port, + "state": st.get("state", "stopped"), + "client_count": st.get("client_count", 0), + "bytes_in": st.get("bytes_in", 0), "bytes_out": st.get("bytes_out", 0), + }) + return {"id": cfg.instance_id, "name": socket.gethostname(), + "ip": ip, "port": port, "scheme": scheme, "mappings": maps} + + async def cluster_local(request): + """Peer-facing: returns this node's mappings to another cluster node that + presents the matching shared key. Guarded by the key, not a user session + (added to PUBLIC_PATHS) — read-only, exposes no credentials.""" + cfg = state.config + key = request.headers.get("x-cluster-key", "") + if not (cfg.cluster.active and key and hmac.compare_digest(key, cfg.cluster.key)): + return JSONResponse({"error": "forbidden"}, status_code=403) + return JSONResponse(local_node_payload()) + + async def cluster_status(request): + """Aggregated view for the browser: local node + every discovered peer, + fetched server-side with the shared key. Renders the unified table.""" + if not state.config.cluster.active: + return render(request, "_cluster_body.html", nodes=[]) + local = local_node_payload() + nodes = [{**local, "self": True, "online": True}] + peers = state.cluster.peers() + results = await asyncio.gather(*[state.cluster.fetch_peer(p) for p in peers]) \ + if peers else [] + for p, data in zip(peers, results, strict=False): + if data: + nodes.append({ + "id": data.get("id", p["id"]), "name": data.get("name", p["name"]), + "ip": data.get("ip", p["ip"]), "port": data.get("port", p["port"]), + "scheme": data.get("scheme", p["scheme"]), + "self": False, "online": True, "mappings": data.get("mappings", []), + }) + else: + nodes.append({"id": p["id"], "name": p["name"], "ip": p["ip"], + "port": p["port"], "scheme": p["scheme"], + "self": False, "online": False, "mappings": []}) + return render(request, "_cluster_body.html", nodes=nodes) + + async def settings_cluster_post(request): + if deny := require_role(request, "admin"): + return deny + form = await request.form() + if not auth.csrf_token_matches(request, form.get("_csrf")): + return PlainTextResponse("CSRF validation failed. Reload the page.", status_code=403) + try: + port = int(form.get("discovery_port") or 41750) + except ValueError: + return await settings_get(request, error="Cluster discovery port must be a number.") + new = ClusterSettings( + enabled=form.get("enabled") == "on", + key=(form.get("key") or "").strip(), + discovery_port=port, + advertise_ip=(form.get("advertise_ip") or "").strip(), + ) + async with state.config_lock: + old = state.config.cluster + state.config.cluster = new + try: + await state.asave() + except ConfigError as e: + state.config.cluster = old + return await settings_get(request, error=str(e)) + # apply live: restart discovery so a key/port/enable change takes effect now + await state.cluster.stop() + await state.cluster.start() + state.log(f"cluster settings updated (enabled={new.enabled})") + state.audit(client_ip(request), "cluster_settings", + "enabled" if new.active else "disabled") + return await settings_get(request, ok="Cluster settings saved.") + async def api_ports_json(request): return JSONResponse(state.ports.get()) @@ -751,6 +857,7 @@ async def mapping_action(request): Route("/settings/users/{username}/delete", settings_users_delete, methods=["POST"]), Route("/settings/ldap", settings_ldap_post, methods=["POST"]), Route("/settings/oidc", settings_oidc_post, methods=["POST"]), + Route("/settings/cluster", settings_cluster_post, methods=["POST"]), Route("/healthz", healthz), Route("/metrics", metrics), Route("/settings/config/export", config_export), @@ -760,6 +867,8 @@ async def mapping_action(request): Route("/api/mappings/{mid}/duplicate", mapping_duplicate, methods=["POST"]), WebSocketRoute("/api/mappings/{mid}/console", console_ws), Route("/api/status", api_status), + Route("/api/cluster/status", cluster_status), + Route("/api/cluster/local", cluster_local), Route("/api/ports.json", api_ports_json), Route("/api/ports/refresh", api_ports_refresh, methods=["POST"]), Route("/api/ports/table", api_ports_table), diff --git a/app/web/server.py b/app/web/server.py index 4054074..f63d0e1 100644 --- a/app/web/server.py +++ b/app/web/server.py @@ -23,11 +23,14 @@ STATIC_DIR = os.path.join(os.path.dirname(__file__), "static") PUBLIC_PREFIXES = ("/static", "/auth/") # /auth/oidc/* is the unauthenticated SSO flow -PUBLIC_PATHS = {"/login", "/setup", "/healthz", "/favicon.ico"} +# /api/cluster/local is authenticated by the shared cluster KEY (checked in the +# handler), not a user session — so it must skip the session gate here. +PUBLIC_PATHS = {"/login", "/setup", "/healthz", "/favicon.ico", "/api/cluster/local"} # High-frequency automatic UI refreshes — not logged to all.log to avoid flooding # the audit trail (the user actions that drive them ARE logged). -QUIET_PATHS = {"/api/status", "/api/ports/table", "/api/ports.json", "/healthz", "/favicon.ico"} +QUIET_PATHS = {"/api/status", "/api/cluster/status", "/api/cluster/local", + "/api/ports/table", "/api/ports.json", "/healthz", "/favicon.ico"} def _wants_json(request) -> bool: diff --git a/app/web/templates/_cluster_body.html b/app/web/templates/_cluster_body.html new file mode 100644 index 0000000..60da862 --- /dev/null +++ b/app/web/templates/_cluster_body.html @@ -0,0 +1,46 @@ + + + + + + + + + {% for n in nodes %} + {% set hosturl = n.scheme ~ '://' ~ n.ip ~ ':' ~ n.port %} + {% set hostcell %} + {{ n.name }}{% if n.self %} this node{% endif %} +
{{ n.ip }}:{{ n.port }}
+ {% endset %} + {% if not n.online %} + + + + + {% elif not n.mappings %} + + + + + {% else %} + {% for m in n.mappings %} + + + + + + + + + + + + {% endfor %} + {% endif %} + {% else %} + + {% endfor %} + +
HostMappingTypeEndpointSerialStateClientsIn / Out
{{ hostcell|safe }}offline / unreachable
{{ hostcell|safe }}no mappings on this node
{{ hostcell|safe }}{{ m.name }}{% if not m.enabled %} disabled{% endif %}{{ m.kind }}{{ m.endpoint }}{{ m.serial }}{{ m.state }}{{ m.client_count }}{{ m.bytes_in }} / {{ m.bytes_out }} + {% if not n.self %}Open{% endif %} +
No cluster nodes discovered yet. Beacons are sent every few seconds.
diff --git a/app/web/templates/dashboard.html b/app/web/templates/dashboard.html index d29b7be..49ad3b5 100644 --- a/app/web/templates/dashboard.html +++ b/app/web/templates/dashboard.html @@ -15,6 +15,18 @@

Serial → Network mappings

+{% if cluster_on %} +
+
+

Cluster — all nodes on the LAN

+ read-only · auto-discovered · each row shows its host (name + IP) +
+
+

Discovering nodes…

+
+
+{% endif %} +

Detected serial ports

diff --git a/app/web/templates/settings.html b/app/web/templates/settings.html index bba6ad0..d024356 100644 --- a/app/web/templates/settings.html +++ b/app/web/templates/settings.html @@ -156,6 +156,25 @@
+
+

LAN cluster

+

Discover other ser2net nodes on the local network (signed UDP + broadcast — no mDNS) and show every node's mappings in one read-only table on the + dashboard, each row tagged with its host (name + IP). Set the same shared key + on every node so they trust each other. This node's ID: + {{ instance_id[:8] }}.

+
+ + + +
+ + +
+ +
+
+

Admin TLS (HTTPS)

Encrypts this management UI. Changes take effect after a restart.

diff --git a/ser2net.py b/ser2net.py index efbfba3..7e74fbe 100644 --- a/ser2net.py +++ b/ser2net.py @@ -17,7 +17,7 @@ import os import sys -__version__ = "2.4.0" +__version__ = "2.5.0" ROOT = os.path.dirname(os.path.abspath(__file__)) FROZEN = getattr(sys, "frozen", False) # True inside a PyInstaller build diff --git a/tests/run_all.py b/tests/run_all.py index fabf5f6..70f9ed5 100644 --- a/tests/run_all.py +++ b/tests/run_all.py @@ -16,6 +16,7 @@ import time HERE = os.path.dirname(os.path.abspath(__file__)) +TEST_TIMEOUT = 240 # seconds; longest real tests finish in ~7s, so this only traps hangs # Confirmed cross-platform (stdlib + ./lib; no socat, no real serial device). PORTABLE = [ @@ -40,6 +41,7 @@ "test_modbus_poll.py", "test_ldap.py", "test_oidc.py", + "test_cluster.py", ] # Data-path tests that need socat-backed PTYs (Linux). Best-effort: skipped if the @@ -64,7 +66,19 @@ def run_one(name: str) -> bool: print(f"SKIP {name} (missing)") return True t0 = time.time() - proc = subprocess.run([sys.executable, path], capture_output=True, text=True) + # Per-test wall-clock cap: a wedged test must fail loudly (with its partial + # output) rather than hang the whole CI job until the 6-hour runner limit. + try: + proc = subprocess.run([sys.executable, path], capture_output=True, + text=True, timeout=TEST_TIMEOUT) + except subprocess.TimeoutExpired as e: + dt = time.time() - t0 + print(f"FAIL {name} ({dt:.1f}s, TIMED OUT after {TEST_TIMEOUT}s)") + out = (e.stdout or "") if isinstance(e.stdout, str) else (e.stdout or b"").decode("utf-8", "replace") + err = (e.stderr or "") if isinstance(e.stderr, str) else (e.stderr or b"").decode("utf-8", "replace") + sys.stdout.write(out[-2000:]) + sys.stderr.write(err[-2000:]) + return False dt = time.time() - t0 ok = proc.returncode == 0 print(f"{'PASS' if ok else 'FAIL'} {name} ({dt:.1f}s)") diff --git a/tests/test_cluster.py b/tests/test_cluster.py new file mode 100644 index 0000000..c27d456 --- /dev/null +++ b/tests/test_cluster.py @@ -0,0 +1,200 @@ +"""LAN cluster: discovery beacons + peer table + key-guarded endpoints (Phase 2). + +Two parts: + 1) in-process unit of ClusterDiscovery — beacon sign/verify, peer registration, + wrong-key rejection, self-ignore, TTL expiry (no real UDP needed). + 2) the real server with the cluster enabled — /api/cluster/local is guarded by + the shared key (403 without / 200 with), reflects this node's mappings, and + /api/cluster/status (session-authed) renders the unified table with the host. + +Run: python3 tests/test_cluster.py +""" +import http.cookiejar +import json +import os +import shutil +import socket +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.parse +import urllib.request + +ROOT = os.path.join(os.path.dirname(__file__), "..") +sys.path.insert(0, os.path.join(ROOT, "lib")) +sys.path.insert(0, ROOT) + +from app.config import AppConfig, ClusterSettings # noqa: E402 +from app.engine.cluster import ClusterDiscovery, PEER_TTL, _sign # noqa: E402 + + +def free_port(): + s = socket.socket() + s.bind(("127.0.0.1", 0)) + p = s.getsockname()[1] + s.close() + return p + + +def wait_port(port, timeout=15): + end = time.time() + timeout + while time.time() < end: + with socket.socket() as s: + s.settimeout(0.3) + if s.connect_ex(("127.0.0.1", port)) == 0: + return True + time.sleep(0.2) + return False + + +def _node(instance_id, key, port=41750): + cfg = AppConfig() + cfg.instance_id = instance_id + cfg.cluster = ClusterSettings(enabled=True, key=key, discovery_port=port) + return ClusterDiscovery(cfg, lambda m: None) + + +def test_discovery_unit(): + a = _node("aaaa1111", "shared") + b = _node("bbbb2222", "shared") + + # A's signed beacon registers as a peer on B (same key) + beacon = a._make_beacon() + b.handle_datagram(beacon, ("10.0.0.5", 41750)) + peers = b.peers() + assert len(peers) == 1 and peers[0]["id"] == "aaaa1111", peers + assert peers[0]["ip"] and peers[0]["port"] == 8080, peers + print("beacon sign + verify + peer registration OK") + + # when a node advertises a loopback IP (LAN IP undetectable), the peer's IP + # falls back to the packet's source address + body = json.dumps({"v": 1, "id": "dddd4444", "name": "edge", "ip": "127.0.0.1", + "port": 8080, "scheme": "http", "t": 0}, + separators=(",", ":"), sort_keys=True) + loop_beacon = json.dumps({"d": body, "s": _sign("shared", body)}).encode() + b.handle_datagram(loop_beacon, ("10.0.0.7", 41750)) + dd = next(p for p in b.peers() if p["id"] == "dddd4444") + assert dd["ip"] == "10.0.0.7", dd + b._peers.pop("dddd4444") # keep the rest of the test's peer count clean + print("loopback advertise -> packet-source IP fallback OK") + + # a node configured with a different key must ignore the beacon + c = _node("cccc3333", "other-key") + c.handle_datagram(beacon, ("10.0.0.5", 41750)) + assert c.peers() == [], "beacon signed with a different key must be ignored" + print("beacon with wrong key rejected OK") + + # a node ignores its own beacon + a.handle_datagram(a._make_beacon(), ("127.0.0.1", 41750)) + assert a.peers() == [], "node must ignore its own beacon" + print("self-beacon ignored OK") + + # garbage / unsigned data is dropped, doesn't crash + b.handle_datagram(b"not-json", ("10.0.0.9", 41750)) + b.handle_datagram(json.dumps({"d": "{}", "s": "deadbeef"}).encode(), ("10.0.0.9", 41750)) + assert len(b.peers()) == 1, "garbage beacons must not register peers" + print("garbage/forged beacons dropped OK") + + # stale peers expire after PEER_TTL + b._peers["aaaa1111"]["last_seen"] = time.time() - PEER_TTL - 5 + assert b.peers() == [], "stale peer should expire" + print("stale peer expiry OK") + + +def raw_request(ui, path, headers=None, method="GET"): + req = urllib.request.Request(f"http://127.0.0.1:{ui}{path}", method=method) + for k, v in (headers or {}).items(): + req.add_header(k, v) + try: + r = urllib.request.urlopen(req, timeout=10) + return r.status, r.read().decode("utf-8") + except urllib.error.HTTPError as e: + return e.code, e.read().decode("utf-8") + + +def test_endpoints(): + tmp = tempfile.mkdtemp(prefix="ser2net_cluster_") + cfg = os.path.join(tmp, "config.json") + ui = free_port() + key = "test-cluster-key-123" + with open(cfg, "w") as fh: + json.dump({"admin_ui": {"bind_ip": "127.0.0.1", "port": ui}, + "cluster": {"enabled": True, "key": key, "discovery_port": free_port()}}, fh) + srv = subprocess.Popen([sys.executable, "ser2net.py", "--no-bootstrap", "--config", cfg], + cwd=ROOT, stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + try: + assert wait_port(ui), "server did not start" + + # become admin first (before setup, the first-run gate redirects everything + # to /setup, so the key guard can't be observed) + jar = http.cookiejar.CookieJar() + op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) + + def csrf(): + return next(c.value for c in jar if c.name == "ser2net_csrf") + + def post(path, fields, header=False): + req = urllib.request.Request(f"http://127.0.0.1:{ui}{path}", + data=urllib.parse.urlencode(fields).encode(), method="POST") + req.add_header("Content-Type", "application/x-www-form-urlencoded") + if header: + req.add_header("X-CSRF-Token", csrf()) + return op.open(req, timeout=10) + + op.open(f"http://127.0.0.1:{ui}/setup", timeout=10) + post("/setup", {"username": "admin", "password": "adminpass123", + "password2": "adminpass123", "_csrf": csrf()}) + + # peer-facing endpoint is guarded by the shared key (no user session) + code, _ = raw_request(ui, "/api/cluster/local") + assert code == 403, f"no key must be 403, got {code}" + code, _ = raw_request(ui, "/api/cluster/local", {"X-Cluster-Key": "wrong"}) + assert code == 403, f"wrong key must be 403, got {code}" + code, body = raw_request(ui, "/api/cluster/local", {"X-Cluster-Key": key}) + assert code == 200, f"correct key must be 200, got {code}" + data = json.loads(body) + assert data["name"] == socket.gethostname() and data["id"], data + assert data["mappings"] == [], "no mappings yet" + print("cluster/local: key-guarded (403/403/200), returns node identity OK") + + port = free_port() + post("/api/mappings/save", { + "_csrf": csrf(), "name": "EDGE1", "enabled": "on", "kind": "net", + "serial_port": "/dev/ttyTEST", "serial_baudrate": "9600", + "network_mode": "server", "network_protocol": "raw", + "network_bind_ip": "127.0.0.1", "network_port": str(port)}, header=True) + + # the mapping now shows up in this node's cluster payload + _, body = raw_request(ui, "/api/cluster/local", {"X-Cluster-Key": key}) + data = json.loads(body) + names = [m["name"] for m in data["mappings"]] + assert "EDGE1" in names, names + assert any(m["endpoint"].endswith(str(port)) for m in data["mappings"]), data["mappings"] + print("cluster/local: created mapping appears with its endpoint OK") + + # the aggregated table (session-authed) renders this node + mapping + html = op.open(f"http://127.0.0.1:{ui}/api/cluster/status", timeout=10).read().decode() + assert socket.gethostname() in html and "this node" in html, html[:400] + assert "EDGE1" in html, "mapping should appear in the unified table" + print("cluster/status: unified table shows host (name+IP) + mapping OK") + + print("\nPASS: LAN cluster discovery + key-guarded aggregated view") + finally: + srv.terminate() + try: + srv.wait(timeout=5) + except subprocess.TimeoutExpired: + srv.kill() + shutil.rmtree(tmp, ignore_errors=True) + + +def main(): + test_discovery_unit() + test_endpoints() + + +if __name__ == "__main__": + main() diff --git a/tests/test_modbus_gateway.py b/tests/test_modbus_gateway.py index dc1a3b1..b188d05 100644 --- a/tests/test_modbus_gateway.py +++ b/tests/test_modbus_gateway.py @@ -120,16 +120,19 @@ async def main(): mwriter.close() with suppress(Exception): - await mwriter.wait_closed() + await asyncio.wait_for(mwriter.wait_closed(), timeout=2) print("\nPASS: Modbus RTU<->TCP gateway") finally: + # Python 3.12+ changed Server.wait_closed() to block until every active + # connection finishes — bound every wait_closed() so a lingering handler + # can't hang teardown forever (the assertions above already passed). runner._server.close() with suppress(Exception): - await runner._server.wait_closed() + await asyncio.wait_for(runner._server.wait_closed(), timeout=2) sw.close() slave_srv.close() with suppress(Exception): - await slave_srv.wait_closed() + await asyncio.wait_for(slave_srv.wait_closed(), timeout=2) if __name__ == "__main__": diff --git a/tests/test_modbus_poll.py b/tests/test_modbus_poll.py index 9152df4..3a4f6cc 100644 --- a/tests/test_modbus_poll.py +++ b/tests/test_modbus_poll.py @@ -109,11 +109,13 @@ async def main(): runner._stop.set() poll.cancel() with suppress(Exception): - await poll + await asyncio.wait_for(poll, timeout=2) sw.close() slave_srv.close() + # Python 3.12+ Server.wait_closed() blocks until active connections finish; + # bound it so a lingering slave handler can't hang teardown forever. with suppress(Exception): - await slave_srv.wait_closed() + await asyncio.wait_for(slave_srv.wait_closed(), timeout=2) if __name__ == "__main__":