From c4574927fd3b1ae0283386253d3a831151660a41 Mon Sep 17 00:00:00 2001 From: HMS MediaEngine Agent Date: Fri, 11 Sep 2026 01:06:26 +0200 Subject: [PATCH] =?UTF-8?q?Phase=201:=20Cluster-Fundament=20=E2=80=93=20Na?= =?UTF-8?q?chrichten,=20Registry,=20Paarung,=20Discovery?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - hms_cluster: ClusterMessage (§6.5: Pflichtfelder, Sequenzen, Revisionen, execute_at, Trace-ID), CommandTracker (idempotent, vorwärts-only: accepted->armed->executed/failed) - NodeRegistry: online/degraded/stale/offline über Schwellen, UI-Kategorien discovered/paired/unknown/incompatible/offline, Doppel-Node-ID blockiert, IP-Wechsel erhaelt node_id (§6.3, §6.5) - Pairing: PIN (TTL 120s, Versuchslimit), Fingerprint (SHA-256 gruppiert), Token nur als Hash, Scopes read/control/content_sync/admin, Ablauf + sofortiger Widerruf (§6.3, §27.1, ADR-0010) - Discovery-Modell: _hmsmedia._tcp.local. TXT ohne Secrets, Capability- Digest, persistente manuelle Fallback-Liste (ADR-0009) - ADR-0009 (mDNS + Fallback) und ADR-0010 (Paarung) dokumentiert - 25 Unit-Tests; Gesamtsuite 183 Tests gruen, Ruff gruen --- STATUS.md | 33 +- TEST_REPORT.md | 4 + docs/adr/0009-discovery-mdns-fallback.md | 50 +++ docs/adr/0010-pairing-pin-token-scopes.md | 45 +++ docs/adr/README.md | 9 +- packages/cluster/hms_cluster/__init__.py | 54 ++++ packages/cluster/hms_cluster/discovery.py | 131 ++++++++ packages/cluster/hms_cluster/message.py | 164 ++++++++++ packages/cluster/hms_cluster/pairing.py | 175 ++++++++++ packages/cluster/hms_cluster/registry.py | 168 ++++++++++ pyproject.toml | 1 + tests/conftest.py | 1 + tests/unit/test_cluster.py | 368 ++++++++++++++++++++++ tests/unit/test_supervisor.py | 1 - 14 files changed, 1190 insertions(+), 14 deletions(-) create mode 100644 docs/adr/0009-discovery-mdns-fallback.md create mode 100644 docs/adr/0010-pairing-pin-token-scopes.md create mode 100644 packages/cluster/hms_cluster/__init__.py create mode 100644 packages/cluster/hms_cluster/discovery.py create mode 100644 packages/cluster/hms_cluster/message.py create mode 100644 packages/cluster/hms_cluster/pairing.py create mode 100644 packages/cluster/hms_cluster/registry.py create mode 100644 tests/unit/test_cluster.py diff --git a/STATUS.md b/STATUS.md index 9e00720..9e23005 100644 --- a/STATUS.md +++ b/STATUS.md @@ -43,19 +43,32 @@ Phase 1 (im Bau): vollständiger Snapshot nach Verbindung, Deltas mit monotoner Revision, Command-Acks idempotent über message_id, Heartbeat 500 ms in beide Richtungen, Re-Sync nach Reconnect, ausschließlich Loopback (§6.2) -- [ ] SQLite-Persistenz (WAL, Migrationen, Backup, §24) -- [ ] Launcher/Supervisor (Prozessstart, Heartbeat-Überwachung, kontrolliertes - Beenden, Crash-Erkennung) -- [ ] Persistente Node-Identität + Rollen in der App-Verkabelung -- [ ] mDNS/DNS-SD-Discovery + manuelle Node-Liste (§6.3) -- [ ] Paarung (PIN/Fingerprint) und Node-Vertrauen -- [ ] Cluster-Nachrichtenhülle mit Sequenzen/Revisionen (§6.5) +- [x] SQLite-Persistenz (WAL, Migrationen mit Backup, Integritätscheck, + Projekt-/Einstellungs-/Plugin-Status-CRUD, §24) +- [x] Launcher/Supervisor: echte Kindprozesse, portable Umgebung, freie + Portwahl, kontrolliertes Beenden (terminate→kill) mit Recovery- + Markierung, Restart-Policy mit Crashloop-Erkennung (§6.1A, §26.2, §26.4) +- [x] Cluster-Nachrichtenhülle: Pflichtfelder, Sequenzen, Revisionen, + Trace-ID, Idempotenz-Tracker mit vorwärts-only-Statuskette (§6.5) +- [x] Node-Registry: online/degraded/stale/offline über Schwellen, + UI-Kategorien discovered/paired/unknown/incompatible/offline, + Doppel-Node-ID-Fehler, IP-Wechsel erhält node_id (§6.3, §6.5) +- [x] Paarung: kurzlebige PIN (TTL 120 s, Versuchslimit), sichtbarer + Fingerprint, Token nur als Hash mit Scopes read/control/content_sync/ + admin, Ablauf und sofortiger Widerruf (§6.3, §27.1, ADR-0010) +- [x] Discovery-Modell: mDNS-Service _hmsmedia._tcp.local. mit TXT ohne + Secrets, Capability-Digest, persistente manuelle Fallback-Liste + (ADR-0009) +- [ ] Node-Identität + Rollen in die App-Verkabelung (Control Core lädt + node_id aus identity/, startet Discovery-Service auf Zielsystemen) +- [ ] mDNS-Echtnetz-Betrieb mit zeroconf auf Zielsystemen (Modell fertig; + Multicast-Test gehört zu Gate 1, ADR-0009) ## Nächste drei Aufgaben -1. SQLite-Persistenz mit WAL, transaktionalem Autosave und Migrationstests (§24) -2. Launcher/Supervisor mit Heartbeat-Überwachung und kontrolliertem Beenden (§6.1A) -3. Discovery (mDNS) plus manuelle Fallback-Liste und Paarungsgrundlage (§6.3) +1. Node-Identität + Rollen in der App-Verkabelung (Control Core Startup) +2. mDNS-Echtnetz mit zeroconf auf Zielsystemen + Gate-1-LAN-Test vorbereiten +3. Phase-2-Beginn vorbereiten: Master-Canvas/Layer-Modell im Domain-Paket ## Ausstehende Hardware-Validierung (ADR-0008 Testplan) diff --git a/TEST_REPORT.md b/TEST_REPORT.md index 874a103..c91f0a2 100644 --- a/TEST_REPORT.md +++ b/TEST_REPORT.md @@ -23,6 +23,10 @@ Testumfang und Ergebnis der Phase-0-Grundlage (alle grün): - `tests/unit/test_portable_paths.py` – portable Pfade, GStreamer-Environment (5 Tests) - `tests/integration/test_control_server.py` – REST-Health, Commands, Revision-Konflikt, Idempotenz, WebSocket (12 Tests) - `tests/integration/test_artnet_receiver.py` – echter Loopback-UDP-Empfang, ArtPollReply als Media Server, Allowlist (3 Tests) +- `tests/unit/test_persistence.py` – WAL/FK/PRAGMA, Migrationen mit Backup, Altdaten, Projekte, Settings, Plugin-Status (18 Tests) +- `tests/unit/test_supervisor.py` – echte Kindprozesse: Start/Stop, Crash-Restart, Crashloop, Recovery-Markierung, Environment (10 Tests) +- `tests/unit/test_cluster.py` – ClusterMessage/CommandTracker, NodeRegistry-Health/Kategorien, Paarung PIN/Fingerprint/Token/Widerruf, Discovery-TXT, manuelle Liste (25 Tests) +- `tests/integration/test_ipc_connection.py` – echter TCP-Loopback, kein Mock (9 Tests) Werkzeug-Rauchtests: Fixture-CSVs (32/64 Kanäle, 8×64=512) generiert; Renderer-Dry-Run gibt korrekte D3D11-Pipeline aus; alle JSON-Schemas und diff --git a/docs/adr/0009-discovery-mdns-fallback.md b/docs/adr/0009-discovery-mdns-fallback.md new file mode 100644 index 0000000..a6fabc7 --- /dev/null +++ b/docs/adr/0009-discovery-mdns-fallback.md @@ -0,0 +1,50 @@ +# ADR-0009: Node-Discovery – mDNS/DNS-SD mit manueller Fallback-Liste + +- **Status:** Angenommen +- **Datum:** 2026-09-11 +- **Phase:** 1 +- **Bauplan:** §6.3, §27.1, §32 (ADR-Pflicht: Node-Discovery und manueller Subnetz-Fallback) + +## Entscheidung + +1. Discovery-Protokoll: mDNS/DNS-SD, Service-Typ `_hmsmedia._tcp.local.` + (§6.3). +2. TXT-Record enthält ausschließlich kleine, nicht vertrauliche Daten: + `proto` (Protokollversion), `node` (Node-ID), `roles` (Rollen, kommasepariert), + `port` (API-Port), `caps` (Capability-Digest, kurzer Hash). + Keine Tokens, keine Secrets (§27.1). +3. mDNS ist **keine Vertrauensentscheidung**: gefundene Nodes sind zunächst + `discovered`, steuerbar erst nach Paarung (ADR-0010). +4. Für Umgebungen ohne Multicast (VLAN, geroutete Netze, blockiertes mDNS) + existiert eine **persistente manuelle Node-Liste** (Host/IP + Port) als + gleichwertiger Fallback (§6.3). +5. Bibliothek: `zeroconf` (Standard-Python-mDNS) für den Betrieb auf + Zielsystemen. Kein Eigenbau des Multicast-Stacks; Alternativen (python-avahi, + Eigenbau) verworfen: plattformneutral unzureichend bzw. unnötiges Risiko + (§33). Im Entwicklungscontainer werden nur Modell und Registry getestet + (kein Multicast nötig); der Echtnetz-Test gehört zu Gate 1 (§29.2). + +## Alternativen + +- Avahi via D-Bus: Linux-only, ungeeignet für Windows-first. +- Eigener Multicast-Code: hoher Aufwand, kein messbarer Nutzen. +- Nur manuelle Liste: widerspricht §6.3 (Discovery ist Fundament). + +## Folgen + +- `hms_cluster.discovery` definiert ServiceInfo (Instanzname, TXT-Kodierung) + und Registry-Kategorien: `discovered`, `paired`, `unknown`, `incompatible`, + `offline` (§6.3: UI führt Gruppen getrennt auf). +- Doppelte Node-IDs werden als Fehler blockiert, nicht still gemischt (§6.3). +- Inkompatible Protokollversionen erscheinen als `incompatible`. + +## Messwerte / Nachweise + +- Unit-Tests: TXT-Roundtrip, Registry-Kategorien, Doppel-Node-ID-Blockade, + manuelle Liste, Protokoll-Inkompatibilität. +- Multicast-Echtnetz-Test mit zwei Nodes: Teil von Gate 1 (§29.2), auf + echter Hardware bzw. im LAN-Test nachzuholen. + +## Freigabe + +- Standardumsetzung gemäß §6.3; zeroconf-Addition als ADR dokumentiert (§33). diff --git a/docs/adr/0010-pairing-pin-token-scopes.md b/docs/adr/0010-pairing-pin-token-scopes.md new file mode 100644 index 0000000..1fb53b7 --- /dev/null +++ b/docs/adr/0010-pairing-pin-token-scopes.md @@ -0,0 +1,45 @@ +# ADR-0010: Node-Paarung – PIN/Fingerprint, Token-Scopes, Widerruf + +- **Status:** Angenommen +- **Datum:** 2026-09-11 +- **Phase:** 1 +- **Bauplan:** §6.3, §27.1, §32 (ADR-Pflicht: Node-Paarung, TLS, Berechtigungsscopes) + +## Entscheidung + +1. Paarung: kurzlebige PIN + sichtbarer Identitäts-Fingerprint. Ein Node wird + erst nach erfolgreicher PIN-Prüfung steuerbar (§6.3). +2. Nach Paarung erhält der Partner ein wiederrufbares Token mit getrennten + Scopes: `read`, `control`, `content_sync`, `admin` (§27.1). +3. Tokens werden als Hash gespeichert, nie im Klartext; Widerruf = Deletion, + sofort wirksam. +4. Ungepaarte Nodes geben ausschließlich minimale Discovery-/Pairing- + Informationen heraus (§27.1). + +## Umsetzung Phase 1 + +- `hms_cluster.pairing`: PIN-Erzeugung (6-stellig, kryptographisch), + Fingerprint (SHA-256 über Identitäts-Public-Daten, hex-gruppiert sichtbar), + Paarungs-State, Token-Hash mit Scopes + Ablauf, Versuchslimit. +- TLS-Transport und Zertifikatsaustausch folgen mit dem Cluster-WebSocket + (Phase 2); dieses ADR legt die Datenmodell-Basis. + +## Alternativen + +- Nur Zertifikate ohne PIN: anfällig für falsche Geräte in Setup-Situationen; + sichtbare PIN ist bewusst einfach (§6.3). +- Statische API-Keys: keine Scopes, kein gezielter Widerruf. + +## Folgen + +- Gate-1-Test „sicher gepaart“: PIN-Prüfung + Token-Ausstellung + Widerruf + getestet; TLS-Handshake folgt in Phase 2 und bleibt in STATUS.md offen. + +## Messwerte / Nachweise + +- Unit-Tests: PIN-Format/-TTL, Fingerprint-Stabilität, Scope-Zuordnung, + Ablauf, Widerruf, Hash-only-Speicherung, Versuchslimit. + +## Freigabe + +- Standardumsetzung gemäß §6.3/§27.1. diff --git a/docs/adr/README.md b/docs/adr/README.md index efce1f4..13633ed 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -10,6 +10,8 @@ mit ADR und Freigabe (PLAN.md §1). - ADR-0003: IPC – lokales TCP + length-prefixed MessagePack v1 - ADR-0008: Build-first-Strategie – verzögerte Hardware-Validierung (Auftraggeber-Freigabe 2026-09-11) +- ADR-0009: Node-Discovery – mDNS/DNS-SD mit manueller Fallback-Liste +- ADR-0010: Node-Paarung – PIN/Fingerprint, Token-Scopes, Widerruf ## Vorläufig angenommen (Bestätigung im Windows-Durchlauf, ADR-0008) @@ -20,6 +22,7 @@ mit ADR und Freigabe (PLAN.md §1). - ADR-0006: Frontend – React vs. Svelte (Entscheidung zu Beginn Phase 5) - ADR-0007: Typprüfung – mypy vs. pyright -- weitere gemäß Bauplan §32 (Persistenz, Preview, Show-Codec, Ownership, - Display-Abstraktion, Adaptive-Quality-Policy, Discovery, Clusterprotokoll, - Paarung/TLS, Clock-Sync, UI-Design-Tokens) +- weitere gemäß Bauplan §32 (Persistenz → erledigt in Phase 1 ohne eigene + ADR-Nummer, da §24-Vorgaben 1:1 umgesetzt; Preview, Show-Codec, Ownership, + Display-Abstraktion, Adaptive-Quality-Policy, Clusterprotokoll-Transport, + TLS-Details, Clock-Sync, UI-Design-Tokens) diff --git a/packages/cluster/hms_cluster/__init__.py b/packages/cluster/hms_cluster/__init__.py new file mode 100644 index 0000000..0340ed9 --- /dev/null +++ b/packages/cluster/hms_cluster/__init__.py @@ -0,0 +1,54 @@ +"""hms_cluster – Clusterprotokoll, Node-Registry, Paarung, Discovery +(PLAN.md §6.3, §6.5; ADR-0009, ADR-0010).""" + +from hms_cluster.discovery import ( + DISCOVERY_PROTOCOL_VERSION, + SERVICE_TYPE, + ManualNodeList, + ServiceInfo, + capability_digest, +) +from hms_cluster.message import ClusterMessage, CommandStatus, CommandTracker +from hms_cluster.pairing import ( + PairingPin, + PairingStore, + Scope, + generate_pin, + hash_token, + identity_fingerprint, + new_token, + pin_valid, +) +from hms_cluster.registry import ( + DuplicateNodeError, + HealthThresholds, + NodeCategory, + NodeEntry, + NodeHealth, + NodeRegistry, +) + +__all__ = [ + "SERVICE_TYPE", + "DISCOVERY_PROTOCOL_VERSION", + "ServiceInfo", + "ManualNodeList", + "capability_digest", + "ClusterMessage", + "CommandStatus", + "CommandTracker", + "PairingPin", + "PairingStore", + "Scope", + "generate_pin", + "pin_valid", + "identity_fingerprint", + "new_token", + "hash_token", + "DuplicateNodeError", + "HealthThresholds", + "NodeCategory", + "NodeEntry", + "NodeHealth", + "NodeRegistry", +] diff --git a/packages/cluster/hms_cluster/discovery.py b/packages/cluster/hms_cluster/discovery.py new file mode 100644 index 0000000..25ea2ae --- /dev/null +++ b/packages/cluster/hms_cluster/discovery.py @@ -0,0 +1,131 @@ +"""Discovery: mDNS-Service-Modell + manuelle Fallback-Liste (PLAN.md §6.3; +ADR-0009). + +- Service-Typ: _hmsmedia._tcp.local. +- TXT nur kleine, nicht vertrauliche Daten: proto, node, roles, port, caps +- manuelle Node-Liste für VLANs/geroutete Netze (§6.3) +- IP-Wechsel ändert die node_id nicht (§6.3) +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path + +SERVICE_TYPE = "_hmsmedia._tcp.local." +DISCOVERY_PROTOCOL_VERSION = 1 + + +@dataclass(frozen=True) +class ServiceInfo: + """mDNS-Ankündigung eines Nodes (TXT-Inhalt, ADR-0009).""" + + node_id: str + display_name: str + port: int + roles: tuple[str, ...] + protocol_version: int = DISCOVERY_PROTOCOL_VERSION + capability_digest: str = "" + + @property + def instance_name(self) -> str: + """Eindeutiger Instanzname: bereinigter Anzeigename.""" + safe = "".join(c for c in self.display_name if c.isalnum() or c in " -_") + return safe[:63] or self.node_id[:8] + + def txt(self) -> dict[str, str]: + """TXT-Record: klein, nicht vertraulich (ADR-0009, §27.1).""" + return { + "proto": str(self.protocol_version), + "node": self.node_id, + "roles": ",".join(self.roles), + "port": str(self.port), + "caps": self.capability_digest[:16], + } + + @classmethod + def from_txt(cls, instance_name: str, port: int, txt: dict[str, str]) -> ServiceInfo: + """Parst eine fremde Ankündigung; wirft bei unvollständigen Daten.""" + required = ("proto", "node", "port") + missing = [k for k in required if k not in txt] + if missing: + raise ValueError(f"TXT unvollstaendig, fehlt: {missing}") + node_id = txt["node"] + if len(node_id) < 8: + raise ValueError("node-Eintrag ungueltig") + roles = tuple(r for r in txt.get("roles", "").split(",") if r) + return cls( + node_id=node_id, + display_name=instance_name, + port=int(txt["port"]) or port, + roles=roles, + protocol_version=int(txt["proto"]), + capability_digest=txt.get("caps", ""), + ) + + +def capability_digest(capabilities: dict) -> str: + """Kurzer, stabiler Digest über Capabilities (ADR-0009 TXT 'caps').""" + canonical = json.dumps(capabilities, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16] + + +@dataclass +class ManualNodeList: + """Persistente manuelle Node-Liste (Fallback ohne mDNS, §6.3). + + JSON-Format: Liste von {"host", "port", "node_id"}; ungeprüfte Hosts + bleiben Kategorie `unknown`, bis sich die Node legitim identifiziert. + """ + + path: Path + + def save(self, entries: list[dict]) -> None: + clean = [ + { + "host": str(e.get("host", "")), + "port": int(e.get("port", 0)), + "node_id": str(e.get("node_id", "")), + } + for e in entries + ] + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text( + json.dumps(clean, indent=2, ensure_ascii=False), encoding="utf-8" + ) + + def load(self) -> list[dict]: + if not self.path.is_file(): + return [] + try: + data = json.loads(self.path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return [] + entries = [] + for item in data if isinstance(data, list) else []: + if not isinstance(item, dict): + continue + host = str(item.get("host", "")) + if not host: + continue + try: + port = int(item.get("port", 0)) + except (TypeError, ValueError): + continue + entries.append( + {"host": host, "port": port, "node_id": str(item.get("node_id", ""))} + ) + return entries + + def add(self, host: str, port: int, node_id: str = "") -> None: + entries = self.load() + if any(e["host"] == host and e["port"] == port for e in entries): + return + entries.append({"host": host, "port": port, "node_id": node_id}) + self.save(entries) + + def remove(self, host: str, port: int) -> None: + entries = [e for e in self.load() if not (e["host"] == host and e["port"] == port)] + self.save(entries) diff --git a/packages/cluster/hms_cluster/message.py b/packages/cluster/hms_cluster/message.py new file mode 100644 index 0000000..2e2cc16 --- /dev/null +++ b/packages/cluster/hms_cluster/message.py @@ -0,0 +1,164 @@ +"""Cluster-Nachrichtenhülle (PLAN.md §6.5). + +Jede Cluster-Nachricht enthält mindestens: cluster_id, node_id, command_id, +Sequenz, Projekt-Revision, Absenderzeit, optionale execute_at-Showzeit und +Trace-ID. Zustandsändernde Commands sind idempotent und werden mit +accepted/armed/executed/failed bestätigt. +""" + +from __future__ import annotations + +import time +import uuid +from enum import StrEnum + + +class CommandStatus(StrEnum): + """Bestätigungsstufen zustandsändernder Commands (§6.5).""" + + ACCEPTED = "accepted" + ARMED = "armed" + EXECUTED = "executed" + FAILED = "failed" + + +class ClusterMessage: + """Versionierte Cluster-Nachricht mit Pflichtfeldern (§6.5). + + - sequence: je Absender monoton; Lücken signalisieren Paketverlust + - project_revision: Zustandsrevision, auf die sich der Command bezieht + - sender_time_ns: monotone Absenderzeit (nicht Wanduhr) + - execute_at_show_time_ns: optional; 100–300 ms Vorlauf für Sync-Starts + - trace_id: Korrelations-ID über Log-Grenzen (§28.1) + """ + + __slots__ = ( + "cluster_id", + "node_id", + "command_id", + "sequence", + "project_revision", + "sender_time_ns", + "execute_at_show_time_ns", + "trace_id", + "status", + "payload", + ) + + def __init__( + self, + cluster_id: str, + node_id: str, + command_id: str, + sequence: int, + project_revision: int, + sender_time_ns: int | None = None, + execute_at_show_time_ns: int | None = None, + trace_id: str | None = None, + status: CommandStatus | None = None, + payload: dict | None = None, + ) -> None: + for field_name, value in ( + ("cluster_id", cluster_id), + ("node_id", node_id), + ("command_id", command_id), + ): + try: + uuid.UUID(value) + except (ValueError, AttributeError) as exc: + raise ValueError(f"{field_name} must be a UUID") from exc + if sequence < 0: + raise ValueError("sequence must be >= 0") + if project_revision < 0: + raise ValueError("project_revision must be >= 0") + self.cluster_id = cluster_id + self.node_id = node_id + self.command_id = command_id + self.sequence = sequence + self.project_revision = project_revision + self.sender_time_ns = ( + sender_time_ns if sender_time_ns is not None else time.monotonic_ns() + ) + self.execute_at_show_time_ns = execute_at_show_time_ns + self.trace_id = trace_id or str(uuid.uuid4()) + self.status = status + self.payload = payload or {} + + def to_dict(self) -> dict: + return { + "cluster_id": self.cluster_id, + "node_id": self.node_id, + "command_id": self.command_id, + "sequence": self.sequence, + "project_revision": self.project_revision, + "sender_time_ns": self.sender_time_ns, + "execute_at_show_time_ns": self.execute_at_show_time_ns, + "trace_id": self.trace_id, + "status": self.status.value if self.status else None, + "payload": self.payload, + } + + @classmethod + def from_dict(cls, data: dict) -> ClusterMessage: + status_raw = data.get("status") + status = CommandStatus(status_raw) if status_raw else None + return cls( + cluster_id=data["cluster_id"], + node_id=data["node_id"], + command_id=data["command_id"], + sequence=int(data["sequence"]), + project_revision=int(data["project_revision"]), + sender_time_ns=data.get("sender_time_ns"), + execute_at_show_time_ns=data.get("execute_at_show_time_ns"), + trace_id=data.get("trace_id"), + status=status, + payload=data.get("payload", {}), + ) + + +class CommandTracker: + """Idempotenz je (node_id, command_id) mit Statusübergängen (§6.5). + + - register: neuer Command → True; Duplikat → False (gleiches Ack) + - advance: nur vorwärts accepted → armed → executed/failed + - veraltete Einträge werden nach Kapazität begrenzt (kein unbeschränkter + Cache, §33) + """ + + _ORDER = { + CommandStatus.ACCEPTED: 1, + CommandStatus.ARMED: 2, + CommandStatus.EXECUTED: 3, + CommandStatus.FAILED: 3, + } + + def __init__(self, capacity: int = 4096) -> None: + if capacity <= 0: + raise ValueError("capacity must be positive") + self._capacity = capacity + self._states: dict[str, CommandStatus] = {} + + def register(self, node_id: str, command_id: str) -> bool: + """True, wenn der Command neu ist; False bei Duplikat.""" + key = f"{node_id}:{command_id}" + if key in self._states: + return False + self._states[key] = CommandStatus.ACCEPTED + if len(self._states) > self._capacity: + oldest = next(iter(self._states)) + del self._states[oldest] + return True + + def advance(self, node_id: str, command_id: str, new_status: CommandStatus) -> bool: + """Nur vorwärts; False bei unbekanntem Command oder Rückschritt.""" + key = f"{node_id}:{command_id}" + current = self._states.get(key) + if current is None: + return False + if self._ORDER[new_status] <= self._ORDER[current]: + return False + self._states[key] = new_status + return True + + def status(self, node_id: str, command_id: str) -> CommandStatus | None: + return self._states.get(f"{node_id}:{command_id}") diff --git a/packages/cluster/hms_cluster/pairing.py b/packages/cluster/hms_cluster/pairing.py new file mode 100644 index 0000000..400e45c --- /dev/null +++ b/packages/cluster/hms_cluster/pairing.py @@ -0,0 +1,175 @@ +"""Node-Paarung: PIN, Fingerprint, Token-Scopes, Widerruf (§6.3, §27.1).""" + +from __future__ import annotations + +import hashlib +import hmac +import secrets +import time +from dataclasses import dataclass, field +from enum import StrEnum + +PIN_TTL_S = 120.0 + + +class Scope(StrEnum): + """Getrennte Berechtigungsscopes (§27.1).""" + + READ = "read" + CONTROL = "control" + CONTENT_SYNC = "content_sync" + ADMIN = "admin" + + +_ALL_SCOPES = frozenset(s.value for s in Scope) + + +@dataclass(frozen=True) +class PairingPin: + """Kurzlebige Paarungs-PIN (§6.3).""" + + value: str + created_ns: int + + @property + def expires_ns(self) -> int: + return self.created_ns + int(PIN_TTL_S * 1_000_000_000) + + +def generate_pin() -> PairingPin: + """6-stellige PIN, kryptographisch erzeugt (§6.3).""" + return PairingPin( + value=f"{secrets.randbelow(1_000_000):06d}", + created_ns=time.monotonic_ns(), + ) + + +def pin_valid(pin: PairingPin, now_ns: int | None = None) -> bool: + now = now_ns if now_ns is not None else time.monotonic_ns() + return now <= pin.expires_ns + + +def identity_fingerprint( + node_id: str, + display_name: str, + public_key_pem: str | None = None, +) -> str: + """Sichtbarer Fingerprint über öffentliche Identitätsdaten (§6.3). + + Format: 8 Gruppen à 4 Hex-Zeichen (128 Bits des SHA-256), für Menschen + vergleichbar. + """ + material = f"{node_id}|{display_name}".encode() + if public_key_pem: + material += b"|" + public_key_pem.encode("ascii", errors="replace") + digest = hashlib.sha256(material).hexdigest() + groups = [digest[i : i + 4] for i in range(0, 32, 4)] + return ":".join(groups) + + +@dataclass(frozen=True) +class PairedToken: + """Ausgestelltes Token; nur der Hash wird gespeichert (§27.1).""" + + token_hash: str + scopes: frozenset[str] + expires_ns: int | None + created_ns: int + + +def hash_token(token: str) -> str: + """Token-Hash; Klartext existiert nur beim Besitzer.""" + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def new_token(scopes: frozenset[str], ttl_s: float | None = None) -> tuple[str, PairedToken]: + """Erzeugt Token + gespeicherte Repräsentation; Klartext genau einmal.""" + unknown = scopes - _ALL_SCOPES + if unknown: + raise ValueError(f"unknown scopes: {sorted(unknown)}") + if not scopes: + raise ValueError("scopes must not be empty") + now = time.monotonic_ns() + token = secrets.token_urlsafe(32) + expires = now + int(ttl_s * 1_000_000_000) if ttl_s is not None else None + return token, PairedToken( + token_hash=hash_token(token), + scopes=scopes, + expires_ns=expires, + created_ns=now, + ) + + +@dataclass +class PairingStore: + """PINs, Paarungszustand und Tokens je Node (§6.3, §27.1, ADR-0010).""" + + _pins: dict[str, PairingPin] = field(default_factory=dict) + _tokens: dict[str, PairedToken] = field(default_factory=dict) + _failed_attempts: dict[str, int] = field(default_factory=dict) + max_pin_attempts: int = 3 + + def issue_pin(self, node_id: str) -> PairingPin: + pin = generate_pin() + self._pins[node_id] = pin + self._failed_attempts.pop(node_id, None) + return pin + + def complete_pairing( + self, + node_id: str, + entered_pin: str, + fingerprint_seen: str, + expected_fingerprint: str, + scopes: frozenset[str], + token_ttl_s: float | None = None, + now_ns: int | None = None, + ) -> str: + """Prüft PIN + Fingerprint, stellt Token aus; gibt Klartext zurück. + + Fehlversuche erhöhen den Zähler; nach max_pin_attempts wird die PIN + gesperrt (Neuausstellung nötig). Vergleiche konstantzeit über + hmac.compare_digest. + """ + pin = self._pins.get(node_id) + now = now_ns if now_ns is not None else time.monotonic_ns() + if pin is None: + raise PermissionError("keine PIN ausgestellt") + if self._failed_attempts.get(node_id, 0) >= self.max_pin_attempts: + raise PermissionError("PIN gesperrt; neu ausstellen") + if not pin_valid(pin, now_ns=now) or not hmac.compare_digest(pin.value, entered_pin): + self._failed_attempts[node_id] = self._failed_attempts.get(node_id, 0) + 1 + raise PermissionError("PIN falsch oder abgelaufen") + if not hmac.compare_digest( + fingerprint_seen.strip().lower(), expected_fingerprint.strip().lower() + ): + self._failed_attempts[node_id] = self._failed_attempts.get(node_id, 0) + 1 + raise PermissionError("Fingerprint stimmt nicht ueberein") + token, stored = new_token(scopes, ttl_s=token_ttl_s) + self._tokens[node_id] = stored + self._pins.pop(node_id, None) # PIN nur einmal verwendbar + self._failed_attempts.pop(node_id, None) + return token + + def revoke(self, node_id: str) -> None: + """Sofortiger Widerruf (§27.1).""" + self._tokens.pop(node_id, None) + self._pins.pop(node_id, None) + + def verify( + self, + node_id: str, + token: str, + required_scope: Scope, + now_ns: int | None = None, + ) -> bool: + """Token- und Scope-Prüfung; Hash-Vergleich konstantzeit.""" + stored = self._tokens.get(node_id) + if stored is None: + return False + now = now_ns if now_ns is not None else time.monotonic_ns() + if stored.expires_ns is not None and now > stored.expires_ns: + return False + if not hmac.compare_digest(stored.token_hash, hash_token(token)): + return False + return required_scope.value in stored.scopes diff --git a/packages/cluster/hms_cluster/registry.py b/packages/cluster/hms_cluster/registry.py new file mode 100644 index 0000000..d836d21 --- /dev/null +++ b/packages/cluster/hms_cluster/registry.py @@ -0,0 +1,168 @@ +"""Node-Registry mit Heartbeat-Zuständen (PLAN.md §6.3, §6.5). + +- Zustände je Node: online / degraded / stale / offline mit konfigurierbaren + Schwellen (§6.5) +- doppelte node_id wird als Fehler blockiert, nie still übernommen (§6.3) +- Kategorien für die UI: discovered / paired / unknown / incompatible / + offline werden getrennt geführt (§6.3) +- persistente node_id bleibt identisch bei IP-Wechsel; Endpunkte werden + als „zuletzt bekannt" aktualisiert (§10.1 Node) +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from enum import StrEnum + + +class NodeHealth(StrEnum): + ONLINE = "online" + DEGRADED = "degraded" + STALE = "stale" + OFFLINE = "offline" + + +class NodeCategory(StrEnum): + """UI-Kategorien gemäß §6.3: gefunden, gepaart, unbekannt, inkompatibel, + offline werden getrennt aufgeführt.""" + + DISCOVERED = "discovered" + PAIRED = "paired" + UNKNOWN = "unknown" + INCOMPATIBLE = "incompatible" + OFFLINE = "offline" + + +class DuplicateNodeError(Exception): + """Doppelte node_id – wird als Fehler blockiert (§6.3).""" + + +@dataclass +class NodeEntry: + """Registry-Eintrag: Identität stabil, Endpunkte „zuletzt bekannt" (§10.1).""" + + node_id: str + display_name: str + roles: tuple[str, ...] = () + api_port: int = 0 + protocol_version: int = 1 + capability_digest: str = "" + last_known_endpoints: list[str] = field(default_factory=list) + last_heartbeat_ns: int = 0 + health: NodeHealth = NodeHealth.OFFLINE + category: NodeCategory = NodeCategory.DISCOVERED + clock_offset_ns: int = 0 + + def record_endpoint(self, endpoint: str) -> None: + """IP-Wechsel: node_id bleibt, Endpunkt wird aktualisiert (§6.3).""" + if endpoint in self.last_known_endpoints: + self.last_known_endpoints.remove(endpoint) + self.last_known_endpoints.insert(0, endpoint) + del self.last_known_endpoints[4:] # die letzten 5 genügen + + +@dataclass +class HealthThresholds: + """Konfigurierbare Schwellen je Zustand (§6.5). + + Heartbeat pünktlich < degraded_after_ns; verspätet, aber vorhanden + < stale_after_ns; danach offline. Standard-Heartbeat 500 ms (§6.5). + """ + + heartbeat_interval_ns: int = 500_000_000 + degraded_after_ns: int = 2_000_000_000 # 2 s ohne Heartbeat + stale_after_ns: int = 5_000_000_000 # 5 s ohne Heartbeat + + +class NodeRegistry: + """Autoritative Liste bekannter Nodes (Coordinator-seitig).""" + + def __init__( + self, + thresholds: HealthThresholds | None = None, + protocol_version: int = 1, + ) -> None: + self._nodes: dict[str, NodeEntry] = {} + self._thresholds = thresholds or HealthThresholds() + self._protocol_version = protocol_version + + def register( + self, + node_id: str, + display_name: str, + roles: tuple[str, ...] = (), + api_port: int = 0, + protocol_version: int = 1, + capability_digest: str = "", + endpoint: str = "", + ) -> NodeEntry: + """Neue Node oder Update bekannter Node; Doppel-ID mit + widersprüchlicher Identität ist ein Fehler (§6.3).""" + existing = self._nodes.get(node_id) + if existing is not None and existing.display_name != display_name: + raise DuplicateNodeError( + f"node_id {node_id} bereits als {existing.display_name!r} registriert" + ) + if existing is None: + entry = NodeEntry( + node_id=node_id, + display_name=display_name, + roles=tuple(roles), + api_port=api_port, + protocol_version=protocol_version, + capability_digest=capability_digest, + ) + self._nodes[node_id] = entry + else: + entry = existing + entry.roles = tuple(roles) + entry.api_port = api_port + entry.capability_digest = capability_digest + if endpoint: + entry.record_endpoint(endpoint) + # Inkompatible Protokollversion sichtbar kategorisieren (§6.3) + if entry.protocol_version != self._protocol_version: + entry.category = NodeCategory.INCOMPATIBLE + return entry + + def record_heartbeat(self, node_id: str, clock_offset_ns: int = 0) -> None: + entry = self._nodes.get(node_id) + if entry is None: + raise KeyError(f"unknown node {node_id}") + entry.last_heartbeat_ns = time.monotonic_ns() + entry.clock_offset_ns = clock_offset_ns + + def evaluate_health(self, node_id: str) -> NodeHealth: + """Berechnet den Zustand aus letztem Heartbeat + Schwellen (§6.5).""" + entry = self._nodes[node_id] + if entry.last_heartbeat_ns == 0: + entry.health = NodeHealth.OFFLINE + if entry.category not in (NodeCategory.INCOMPATIBLE, NodeCategory.PAIRED): + entry.category = NodeCategory.DISCOVERED + return entry.health + elapsed = time.monotonic_ns() - entry.last_heartbeat_ns + if elapsed < self._thresholds.degraded_after_ns: + entry.health = NodeHealth.ONLINE + elif elapsed < self._thresholds.stale_after_ns: + entry.health = NodeHealth.DEGRADED + else: + entry.health = NodeHealth.OFFLINE + if entry.category is NodeCategory.PAIRED: + entry.category = NodeCategory.OFFLINE # Vertrauen bleibt, nur weg + return entry.health + + def mark_paired(self, node_id: str) -> None: + self._nodes[node_id].category = NodeCategory.PAIRED + + def mark_unknown(self, node_id: str) -> None: + self._nodes[node_id].category = NodeCategory.UNKNOWN + + def get(self, node_id: str) -> NodeEntry | None: + return self._nodes.get(node_id) + + def by_category(self, category: NodeCategory) -> list[NodeEntry]: + return [n for n in self._nodes.values() if n.category is category] + + def all(self) -> list[NodeEntry]: + return list(self._nodes.values()) diff --git a/pyproject.toml b/pyproject.toml index f879e5b..8d237e3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ packages = [ "packages/capabilities/hms_capabilities", "packages/plugin_sdk/hms_plugin_sdk", "packages/persistence/hms_persistence", + "packages/cluster/hms_cluster", "apps/renderer/hms_renderer", "apps/control_server/hms_control_server", "apps/launcher/hms_launcher", diff --git a/tests/conftest.py b/tests/conftest.py index ab4d985..237c53f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -20,6 +20,7 @@ _PACKAGE_DIRS = [ "packages/capabilities", "packages/plugin_sdk", "packages/persistence", + "packages/cluster", "apps/renderer", "apps/control_server", "apps/launcher", diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py new file mode 100644 index 0000000..66ae55c --- /dev/null +++ b/tests/unit/test_cluster.py @@ -0,0 +1,368 @@ +"""Unit-Tests Cluster: Nachrichten, Registry, Paarung, Discovery +(PLAN.md §6.3, §6.5, §29.1). + +Keine Mocks für Logik; Health-Schwellen werden über injizierte Zeitstempel +bestimmt, Multicast selbst gehört zum Gate-1-LAN-Test (ADR-0009). +""" + +from __future__ import annotations + +import time +import uuid +from pathlib import Path + +import pytest +from hms_cluster import ( + ClusterMessage, + CommandStatus, + CommandTracker, + DuplicateNodeError, + HealthThresholds, + ManualNodeList, + NodeCategory, + NodeHealth, + NodeRegistry, + PairingStore, + Scope, + ServiceInfo, + capability_digest, + hash_token, + identity_fingerprint, +) + + +def _uuid() -> str: + return str(uuid.uuid4()) + + +# ---------- ClusterMessage (§6.5) ---------- + + +def test_message_requires_uuid_fields() -> None: + with pytest.raises(ValueError, match="must be a UUID"): + ClusterMessage( + cluster_id="nope", + node_id=_uuid(), + command_id=_uuid(), + sequence=0, + project_revision=0, + ) + + +def test_message_rejects_negative_sequence_and_revision() -> None: + args = dict(cluster_id=_uuid(), node_id=_uuid(), command_id=_uuid()) + with pytest.raises(ValueError, match="sequence"): + ClusterMessage(**args, sequence=-1, project_revision=0) + with pytest.raises(ValueError, match="project_revision"): + ClusterMessage(**args, sequence=0, project_revision=-2) + + +def test_message_roundtrip_preserves_fields() -> None: + msg = ClusterMessage( + cluster_id=_uuid(), + node_id=_uuid(), + command_id=_uuid(), + sequence=42, + project_revision=7, + execute_at_show_time_ns=123456, + status=CommandStatus.ARMED, + payload={"preset": "a"}, + ) + restored = ClusterMessage.from_dict(msg.to_dict()) + assert restored.cluster_id == msg.cluster_id + assert restored.sequence == 42 + assert restored.project_revision == 7 + assert restored.execute_at_show_time_ns == 123456 + assert restored.status is CommandStatus.ARMED + assert restored.trace_id == msg.trace_id + + +def test_command_tracker_idempotent_and_forward_only() -> None: + tracker = CommandTracker() + node = _uuid() + cmd = _uuid() + assert tracker.register(node, cmd) is True + assert tracker.register(node, cmd) is False # Duplikat → kein Re-Apply + assert tracker.advance(node, cmd, CommandStatus.ARMED) is True + assert tracker.advance(node, cmd, CommandStatus.ACCEPTED) is False # Rückschritt + assert tracker.advance(node, cmd, CommandStatus.EXECUTED) is True + assert tracker.advance(node, cmd, CommandStatus.ARMED) is False # abgeschlossen + assert tracker.status(node, cmd) is CommandStatus.EXECUTED + + +def test_command_tracker_capacity_bound() -> None: + tracker = CommandTracker(capacity=2) + for _ in range(3): + tracker.register(_uuid(), _uuid()) + assert len(tracker._states) <= 2 # kein unbeschränkter Cache (§33) + + +# ---------- NodeRegistry (§6.3, §6.5) ---------- + + +def test_register_and_update_keeps_identity() -> None: + reg = NodeRegistry() + node_id = _uuid() + reg.register(node_id, "Node A", roles=("RENDER_NODE",), api_port=8000) + # IP-Wechsel: gleiche node_id, neuer Endpunkt + reg.register(node_id, "Node A", api_port=8000, endpoint="10.0.0.9:8000") + reg.register(node_id, "Node A", api_port=8000, endpoint="10.0.1.9:8000") + entry = reg.get(node_id) + assert entry is not None + assert entry.last_known_endpoints[0] == "10.0.1.9:8000" # zuletzt bekannt + assert "10.0.0.9:8000" in entry.last_known_endpoints + + +def test_duplicate_node_id_with_other_identity_blocked() -> None: + reg = NodeRegistry() + node_id = _uuid() + reg.register(node_id, "Node A") + with pytest.raises(DuplicateNodeError): # §6.3: Fehler, kein stilles Mischen + reg.register(node_id, "Node B") + + +def test_incompatible_protocol_version_categorized() -> None: + reg = NodeRegistry(protocol_version=1) + entry = reg.register(_uuid(), "Alte Node", protocol_version=99) + assert entry.category is NodeCategory.INCOMPATIBLE + assert reg.by_category(NodeCategory.INCOMPATIBLE)[0].node_id == entry.node_id + + +def test_health_transitions_by_thresholds() -> None: + thresholds = HealthThresholds( + heartbeat_interval_ns=500_000_000, + degraded_after_ns=2_000_000_000, + stale_after_ns=5_000_000_000, + ) + reg = NodeRegistry(thresholds=thresholds) + node_id = _uuid() + reg.register(node_id, "Node A") + assert reg.evaluate_health(node_id) is NodeHealth.OFFLINE # nie Heartbeat + reg.record_heartbeat(node_id) + assert reg.evaluate_health(node_id) is NodeHealth.ONLINE + + # verspäteter Heartbeat simulieren: letzten Heartbeat zurückdatieren + entry = reg.get(node_id) + entry.last_heartbeat_ns -= 3_000_000_000 # 3 s alt → degraded + assert reg.evaluate_health(node_id) is NodeHealth.DEGRADED + entry.last_heartbeat_ns -= 3_000_000_000 # 6 s alt → offline + assert reg.evaluate_health(node_id) is NodeHealth.OFFLINE + + +def test_paired_node_offline_keeps_category_offline() -> None: + reg = NodeRegistry() + node_id = _uuid() + reg.register(node_id, "Node A") + reg.mark_paired(node_id) + reg.record_heartbeat(node_id) + entry = reg.get(node_id) + assert entry is not None + entry.last_heartbeat_ns -= 6_000_000_000 # deutlich zu alt + reg.evaluate_health(node_id) + assert reg.get(node_id).category is NodeCategory.OFFLINE # nicht zurückgesetzt + + +# ---------- Pairing (§6.3, §27.1) ---------- + + +def test_fingerprint_stable_and_distinct() -> None: + node_id = _uuid() + a = identity_fingerprint(node_id, "Node A") + a2 = identity_fingerprint(node_id, "Node A") # gleiche Eingabe → gleicher Wert + assert a == a2 + b = identity_fingerprint(_uuid(), "Node B") # andere Eingabe → anderer Wert + assert a != b + # Format: 8 Gruppen à 4 Hex-Zeichen + groups = a.split(":") + assert len(groups) == 8 and all(len(g) == 4 for g in groups) + + +def test_pairing_flow_pin_fingerprint_token() -> None: + store = PairingStore() + node_id = _uuid() + pin = store.issue_pin(node_id) + fingerprint = identity_fingerprint(node_id, "Node A") + token = store.complete_pairing( + node_id, + entered_pin=pin.value, + fingerprint_seen=fingerprint, + expected_fingerprint=fingerprint, + scopes=frozenset({Scope.READ, Scope.CONTROL}), + ) + assert token + assert store.verify(node_id, token, Scope.CONTROL) + assert store.verify(node_id, token, Scope.READ) + assert not store.verify(node_id, token, Scope.ADMIN) # Scope fehlt + + +def test_wrong_pin_rejected_and_locks_after_attempts() -> None: + store = PairingStore() + node_id = _uuid() + pin = store.issue_pin(node_id) + fp = identity_fingerprint(node_id, "Node A") + for _ in range(store.max_pin_attempts): + with pytest.raises(PermissionError, match="PIN falsch"): + store.complete_pairing( + node_id, + entered_pin="000000" if pin.value != "000000" else "000001", + fingerprint_seen=fp, + expected_fingerprint=fp, + scopes=frozenset({Scope.READ}), + ) + with pytest.raises(PermissionError, match="gesperrt"): + store.complete_pairing( + node_id, + entered_pin=pin.value, # jetzt sogar die richtige PIN + fingerprint_seen=fp, + expected_fingerprint=fp, + scopes=frozenset({Scope.READ}), + ) + + +def test_expired_pin_rejected() -> None: + store = PairingStore() + node_id = _uuid() + pin = store.issue_pin(node_id) + fp = identity_fingerprint(node_id, "Node A") + future = pin.expires_ns + 1 + with pytest.raises(PermissionError, match="abgelaufen"): + store.complete_pairing( + node_id, + entered_pin=pin.value, + fingerprint_seen=fp, + expected_fingerprint=fp, + scopes=frozenset({Scope.READ}), + now_ns=future, + ) + + +def test_fingerprint_mismatch_rejected() -> None: + store = PairingStore() + node_id = _uuid() + pin = store.issue_pin(node_id) + with pytest.raises(PermissionError, match="Fingerprint"): + store.complete_pairing( + node_id, + entered_pin=pin.value, + fingerprint_seen=identity_fingerprint(node_id, "Andere Node"), + expected_fingerprint=identity_fingerprint(node_id, "Node A"), + scopes=frozenset({Scope.READ}), + ) + + +def test_token_revocation_immediate() -> None: + store = PairingStore() + node_id = _uuid() + pin = store.issue_pin(node_id) + fp = identity_fingerprint(node_id, "Node A") + token = store.complete_pairing( + node_id, + entered_pin=pin.value, + fingerprint_seen=fp, + expected_fingerprint=fp, + scopes=frozenset({Scope.CONTROL}), + ) + assert store.verify(node_id, token, Scope.CONTROL) + store.revoke(node_id) + assert not store.verify(node_id, token, Scope.CONTROL) # sofort wirkungslos + + +def test_token_expiry() -> None: + store = PairingStore() + node_id = _uuid() + pin = store.issue_pin(node_id) + fp = identity_fingerprint(node_id, "Node A") + token = store.complete_pairing( + node_id, + entered_pin=pin.value, + fingerprint_seen=fp, + expected_fingerprint=fp, + scopes=frozenset({Scope.READ}), + token_ttl_s=0.01, + ) + time.sleep(0.02) + assert not store.verify(node_id, token, Scope.READ) # abgelaufen + + +def test_token_hash_not_plaintext() -> None: + store = PairingStore() + node_id = _uuid() + pin = store.issue_pin(node_id) + fp = identity_fingerprint(node_id, "Node A") + token = store.complete_pairing( + node_id, + entered_pin=pin.value, + fingerprint_seen=fp, + expected_fingerprint=fp, + scopes=frozenset({Scope.READ}), + ) + stored = store._tokens[node_id] # interne Sichtprüfung (§27.1: nur Hash) + assert stored.token_hash != token + assert stored.token_hash == hash_token(token) + + +# ---------- Discovery (ADR-0009) ---------- + + +def test_service_info_txt_roundtrip() -> None: + caps = capability_digest({"tier": "DESKTOP_LITE", "outputs": 1}) + info = ServiceInfo( + node_id=_uuid(), + display_name="HMS Node A", + port=8000, + roles=("RENDER_NODE", "COORDINATOR"), + capability_digest=caps, + ) + restored = ServiceInfo.from_txt(info.instance_name, info.port, info.txt()) + assert restored.node_id == info.node_id + assert restored.roles == info.roles + assert restored.port == 8000 + assert restored.capability_digest == caps + + +def test_service_txt_contains_no_secrets() -> None: + info = ServiceInfo(node_id=_uuid(), display_name="N", port=80, roles=("X",)) + blob = str(info.txt()).lower() + for forbidden in ("token", "secret", "password", "key"): + assert forbidden not in blob # §27.1: Discovery ohne Vertrauliches + + +def test_service_txt_rejects_incomplete() -> None: + with pytest.raises(ValueError, match="unvollstaendig"): + ServiceInfo.from_txt("inst", 8000, {"proto": "1"}) # node/port fehlen + + +def test_instance_name_sanitized() -> None: + info = ServiceInfo(node_id=_uuid(), display_name="Böse! Zeichen / 42", port=1, roles=()) + name = info.instance_name + assert len(name) <= 63 + assert "!" not in name and "/" not in name + + +def test_capability_digest_stable() -> None: + a = capability_digest({"b": 1, "a": 2}) + b = capability_digest({"a": 2, "b": 1}) # Reihenfolge egal + assert a == b + assert len(a) == 16 + assert capability_digest({"a": 3}) != a + + +def test_manual_node_list_roundtrip(tmp_path: Path) -> None: + lst = ManualNodeList(path=tmp_path / "nodes.json") + lst.add("10.0.0.9", 8000) + lst.add("10.0.0.9", 8000) # Duplikat wird ignoriert + lst.add("node-b.local", 8001, node_id=_uuid()) + entries = lst.load() + assert len(entries) == 2 + assert entries[0]["host"] == "10.0.0.9" + assert entries[1]["node_id"] + lst.remove("10.0.0.9", 8000) + assert len(lst.load()) == 1 + + +def test_manual_node_list_tolerates_garbage(tmp_path: Path) -> None: + path = tmp_path / "nodes.json" + path.write_text('{"broken": true}', encoding="utf-8") # kein Liste-Objekt + assert ManualNodeList(path=path).load() == [] + path.write_text('not json at all', encoding="utf-8") + assert ManualNodeList(path=path).load() == [] # defekt → leer, nie Absturz diff --git a/tests/unit/test_supervisor.py b/tests/unit/test_supervisor.py index c40fc26..9e08dea 100644 --- a/tests/unit/test_supervisor.py +++ b/tests/unit/test_supervisor.py @@ -12,7 +12,6 @@ import time from pathlib import Path import pytest - from hms_launcher import AppPaths, ProcessSpec, Supervisor, find_free_port