diff --git a/tools/station/server.py b/tools/station/server.py index 3111748..a1439d2 100755 --- a/tools/station/server.py +++ b/tools/station/server.py @@ -78,6 +78,13 @@ class Station: self.last_seen: dict | None = None self._manifest_mtime: float | None = None self._manifest_version = "unbekannt" + # Wird True, wenn write_control() zuletzt fehlgeschlagen ist (z.B. volles oder + # schreibgeschuetztes /run) - claim_delivery() liest das, um "armed" nicht + # dauerhaft scharf zu lassen, nur weil das Verbrauchen der Freigabe nicht auf + # Platte geschrieben werden konnte (Fixrunde 1, I3). control_error landet in + # status.json, damit ein Bedienfehler nicht nur auf stdout verschwindet. + self._control_write_broken = False + self.control_error: str | None = None # Cumulative bytes already delivered to each client while --stall-after is active, # keyed by client IP. A lamp whose download stalls re-opens a fresh connection and # resumes with Range right where it left off, so a per-request counter would hand @@ -150,7 +157,51 @@ class Station: os.unlink(tmp) raise except OSError as e: - print(f"warning: cannot write {self.control_path}: {e}", flush=True) + self._control_write_broken = True + self.control_error = f"cannot write {self.control_path}: {e}" + print(f"warning: {self.control_error}", flush=True) + else: + self._control_write_broken = False + self.control_error = None + + def claim_delivery(self, target: str) -> tuple[str, str | None, bool]: + """Entscheidet unter dem Lock, ob eine Anfrage fuer `target` gerade ausgeliefert + werden darf, und verbraucht - falls die Freigabe dafuer verantwortlich ist - diese + sofort per Compare-and-Set (lesen und auf "listen" setzen, ohne das Lock + dazwischen loszulassen). Ohne das koennten zwei gleichzeitige Anfragen (zwei + Lampen, oder Erst- und Range-Resume-Verbindung derselben Lampe) beide phase == + "armed" sehen und beide ausgeliefert bekommen (Fixrunde 1, C1) - genau das Szenario, + fuer das die Freigabephase laut Spec 3.3 da ist. + + Liefert (phase, armed_target) wie sie VOR dem Verbrauch galten, plus ob die + aktuelle Anfrage ausliefern darf. Bricht die Auslieferung anschliessend ab + (status == "aborted") oder findet der Aufrufer kein Image, muss er bei + phase == "armed" restore_armed(target) aufrufen, damit die Freigabe fuer den + naechsten Bootversuch stehen bleibt. + + Schlaegt das Schreiben von "listen" fehl (I3), bekommt nur die eine Anfrage, die + den Fehler zuerst ausloest, noch ihre Auslieferung - jede weitere schlaegt hier + erneut fehl (_control_write_broken bleibt True), bis ein write_control()-Versuch + wieder klappt. Damit bleibt "armed" nicht dauerhaft scharf, nur weil /run gerade + nicht beschreibbar ist, heilt sich aber von selbst, sobald es das wieder ist.""" + with self.lock: + phase, armed_target = self.read_control() + if phase == "listen" or (phase == "armed" and armed_target != target): + return phase, armed_target, False + if phase == "armed": + was_broken = self._control_write_broken + self.write_control("listen", None) + if self._control_write_broken: + return phase, armed_target, not was_broken + return phase, armed_target, True + + def restore_armed(self, target: str) -> None: + """Setzt eine per claim_delivery() verbrauchte Freigabe zurueck, wenn die + Auslieferung danach doch nicht (vollstaendig) stattgefunden hat - ein Abbruch darf + die Freigabe nicht verbrauchen, sonst muesste am Geraet erneut von Hand gearmt + werden, obwohl die Lampe es beim naechsten Bootversuch selbst nochmal versuchen + wuerde (Spec 3.3, Brief-Schritt 3.6).""" + self.write_control("armed", target) def reserve_stall_bytes(self, ip: str, n: int) -> int: """How many of the next `n` bytes this client may still receive under --stall-after. @@ -205,6 +256,7 @@ class Station: payload["phase"] = phase payload["armed_target"] = armed_target payload["last_seen"] = self.last_seen + payload["control_error"] = self.control_error try: self.state_path.parent.mkdir(parents=True, exist_ok=True) fd, tmp = tempfile.mkstemp(dir=self.state_path.parent, prefix=".status-") @@ -322,9 +374,12 @@ class Handler(BaseHTTPRequestHandler): target, variant, name = product.target, product.variant, product.name version = self.station.manifest_version() - phase, armed_target = self.station.read_control() + # claim_delivery() liest die Phase UND verbraucht eine armed-Freigabe atomar unter + # station.lock (Fixrunde 1, C1) - zwischen dem Lesen und dem Verbrauchen darf keine + # zweite gleichzeitige Anfrage hindurchschluepfen, siehe die Methode selbst. + phase, armed_target, allowed = self.station.claim_delivery(target) - if phase == "listen" or (phase == "armed" and armed_target != target): + if not allowed: # last_seen wird trotzdem gesetzt - das ist der ganze Zweck der Horchphase: # die Anzeige soll zeigen, welches Modell da ist, auch wenn nichts rausgeht. self.station.record_migration(ip, target, variant, pid, lamp_version, name, @@ -335,6 +390,10 @@ class Handler(BaseHTTPRequestHandler): path = self.station.migrate_image_path(target, variant) if not path.is_file(): + if phase == "armed": + # claim_delivery() hat die Freigabe schon verbraucht, bevor wir wussten, + # dass es gar kein Image gibt - das darf nicht auf Kosten der Freigabe gehen. + self.station.restore_armed(target) self.station.record_migration(ip, target, variant, pid, lamp_version, name, version, 0, "notfound") self.reject(404, "no image for this device") @@ -342,12 +401,12 @@ class Handler(BaseHTTPRequestHandler): return sent, status = self._serve_file(path, started) - if phase == "armed" and status == "ok": - # Erst nach dem vollstaendigen Senden zurueck auf "listen" - _serve_file() ist - # an dieser Stelle bereits fertig. Bricht der Transfer ab (status == "aborted"), - # bleibt die Freigabe stehen, damit die Lampe es beim naechsten Boot noch - # einmal versuchen kann. - self.station.write_control("listen", None) + if phase == "armed" and status == "aborted": + # Bricht der Transfer ab, bleibt die Freigabe stehen, damit die Lampe es beim + # naechsten Boot noch einmal versuchen kann (Spec 3.3). claim_delivery() hat sie + # bereits auf "listen" gesetzt, bevor _serve_file() ueberhaupt anfing zu senden + # (das ist der Punkt von C1) - bei einem Abbruch wird das hier zurueckgenommen. + self.station.restore_armed(target) self.station.record_migration(ip, target, variant, pid, lamp_version, name, version, sent, status) @@ -403,6 +462,15 @@ class Handler(BaseHTTPRequestHandler): raise ConnectionResetError("peer closed during stall") chunk = f.read(read_size) if not chunk: + # Die Datei ist kuerzer als das eingangs per stat() ermittelte + # Content-Length (z.B. von deploy.sh waehrend einer laufenden + # Auslieferung durch eine kleinere ersetzt) - sent < remaining an + # dieser Stelle immer, denn die Schleifenbedingung oben haette sonst + # schon abgebrochen. Das ist ein abgebrochener, kein erfolgreicher + # Transfer (Fixrunde 1, I2): ohne dies wuerde eine im Migrationsmodus + # laufende "armed"-Freigabe faelschlich als verbraucht gelten, obwohl + # die Lampe nie das vollstaendige Image bekommen hat. + status = "aborted" break self.wfile.write(chunk) sent += len(chunk) diff --git a/tools/station/tests/test_server.py b/tools/station/tests/test_server.py index 1d6008c..8f5c777 100644 --- a/tools/station/tests/test_server.py +++ b/tools/station/tests/test_server.py @@ -472,20 +472,28 @@ class _Response: class MigrationTestServer: """Startet server.py mit --mode/--control und bietet die vom Brief geforderten - Hilfsmethoden. server.py schreibt status.json/control.json fuer eine Anfrage im - selben Handler-Thread, unmittelbar nachdem der Antwortkoerper vollstaendig gesendet - wurde - aber http.client's read() kann zurueckkehren, sobald Content-Length Bytes - angekommen sind, was diesem serverseitigen Schreiben knapp zuvorkommen kann. get() - legt deshalb eine kurze Pause ein, bevor es zurueckkehrt (derselbe Grund, aus dem - Station.read_state() oben mit einem Prädikat auf status.json wartet - hier gibt es - kein generisches Prädikat, das jeder Aufrufer kennt, darum die feste kleine Pause statt - einer Wiederholschleife).""" + Hilfsmethoden. + + status()/read_control() folgen demselben Muster wie Station.read_state() oben + (Zeilen 68-90): sie pollen bis zu einigen Sekunden auf ein optionales Praedikat, statt + sich auf eine feste Pause nach get() zu verlassen. Grund ist derselbe wie dort - + server.py schreibt status.json/control.json fuer eine Anfrage erst, nachdem der + Antwortkoerper vollstaendig gesendet ist, aber http.client's read() kann zurueckkehren, + sobald Content-Length Bytes angekommen sind, was diesem Schreiben knapp zuvorkommen + kann. Jeder Aufrufer kennt sein eigenes Praedikat (das ohnehin schon als Assertion + dasteht), daher kein pauschales Warten hier.""" def __init__(self, tmp_path, mode): self.image_dir = tmp_path / "images" self.image_dir.mkdir() self.state_path = tmp_path / "status.json" - self.control_path = tmp_path / "control.json" + # Eigenes Unterverzeichnis fuer die Steuerdatei, getrennt von status.json: ein Test + # fuer I3 (Fixrunde 1) macht genau dieses Verzeichnis schreibgeschuetzt, um einen + # fehlschlagenden write_control() zu erzwingen, und darf dabei nicht auch das + # Schreiben von status.json lahmlegen, ueber das derselbe Test den Fehler beobachtet. + control_dir = tmp_path / "control" + control_dir.mkdir() + self.control_path = control_dir / "control.json" self.port = free_port() self.proc = subprocess.Popen( [sys.executable, str(SERVER), "--image-dir", str(self.image_dir), @@ -511,30 +519,34 @@ class MigrationTestServer: conn.request("GET", path, headers=headers or {}) r = conn.getresponse() body = r.read() - resp = _Response(r.status, dict(r.getheaders()), body) + return _Response(r.status, dict(r.getheaders()), body) finally: conn.close() - # See the class docstring: a short pause so the server's own status.json/ - # control.json write for this request has landed before we return. - time.sleep(0.05) - return resp - def _read_json(self, path, timeout=2.0): - deadline = time.time() + timeout + def _poll_json(self, path, predicate=None, timeout=5.0): + """Wie Station.read_state() oben: liest `path` wiederholt (bis zu `timeout` + Sekunden), bis er sich parsen laesst und - falls angegeben - `predicate` erfuellt. + Ohne predicate reicht ein einziges erfolgreiches Parsen, genau wie dort.""" + last = None last_err = None + deadline = time.time() + timeout while time.time() < deadline: try: - return json.loads(path.read_text()) + last = json.loads(path.read_text()) + if predicate is None or predicate(last): + return last except (OSError, json.JSONDecodeError) as e: last_err = e - time.sleep(0.02) - raise AssertionError(f"could not read {path}: {last_err}") + time.sleep(0.02) + if last is None: + raise AssertionError(f"could not read {path}: {last_err}") + raise AssertionError(f"{path} never satisfied predicate, last seen: {last!r}") - def status(self): - return self._read_json(self.state_path) + def status(self, predicate=None): + return self._poll_json(self.state_path, predicate) - def read_control(self): - return self._read_json(self.control_path) + def read_control(self, predicate=None): + return self._poll_json(self.control_path, predicate) def set_control(self, phase, armed_target=None): tmp = self.control_path.with_name(self.control_path.name + ".tmp") @@ -577,8 +589,9 @@ def test_migration_auto_serves_known_product(migration_server): r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1.1.1.1800-dev_ota_light.img") assert r.status == 200 assert r.body == b"IMG" * 1000 - assert srv.status()["last_seen"]["target"] == "g95_cw" - assert srv.status()["last_seen"]["variant"] == "dev" + st = srv.status(lambda s: s.get("last_seen") is not None) + assert st["last_seen"]["target"] == "g95_cw" + assert st["last_seen"]["variant"] == "dev" def test_migration_listen_records_but_serves_nothing(migration_server): @@ -588,10 +601,10 @@ def test_migration_listen_records_but_serves_nothing(migration_server): srv.set_control(phase="listen") r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img") assert r.status == 404 - st = srv.status() + st = srv.status(lambda s: len(s["recent"]) >= 1) assert st["last_seen"]["target"] == "g95_cw" # erkannt assert st["served"] == 0 # aber nichts ausgeliefert - assert st["recent"][0]["status"] == "listen" + assert st["recent"][-1]["status"] == "listen" def test_arm_serves_exactly_once_then_falls_back_to_listen(migration_server): @@ -601,7 +614,7 @@ def test_arm_serves_exactly_once_then_falls_back_to_listen(migration_server): srv.set_control(phase="armed", armed_target="g95_cw") p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img" assert srv.get(p).status == 200 - assert srv.read_control()["phase"] == "listen" + assert srv.read_control(lambda c: c.get("phase") == "listen")["phase"] == "listen" assert srv.get(p).status == 404 @@ -619,14 +632,16 @@ def test_unknown_product_id_is_404_and_logged(migration_server): srv, _ = migration_server r = srv.get("/qcx_firmware_-acme.bulb_V1_ota_light.img") assert r.status == 404 - assert srv.status()["recent"][0]["status"] == "unknown" + st = srv.status(lambda s: len(s["recent"]) >= 1) + assert st["recent"][-1]["status"] == "unknown" def test_missing_migrate_image_is_404(migration_server): srv, _ = migration_server r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.e14_V1_ota_light.img") assert r.status == 404 - assert srv.status()["recent"][0]["status"] == "notfound" + st = srv.status(lambda s: len(s["recent"]) >= 1) + assert st["recent"][-1]["status"] == "notfound" def test_recovery_path_is_404_in_migration_mode(migration_server): @@ -678,4 +693,146 @@ def test_version_comes_from_manifest(migration_server): "ninlight_version": "0.3.9", "images": [{"target": "g95_cw", "variant": "dev", "file": "g95_cw_dev.img"}]})) srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img") - assert srv.status()["recent"][0]["version"] == "0.3.9" + st = srv.status(lambda s: len(s["recent"]) >= 1) + assert st["recent"][-1]["version"] == "0.3.9" + + +def test_control_file_is_world_readable_even_under_a_restrictive_umask(tmp_path): + """Dieselbe 0600-Falle wie bei status.json (siehe + test_state_file_is_world_readable_even_under_a_restrictive_umask oben), aber fuer + control.json: das CLI aus Task 4 muss die Datei lesen UND schreiben koennen.""" + old_umask = os.umask(0o077) + try: + srv = MigrationTestServer(tmp_path, "migration") + finally: + os.umask(old_umask) + try: + (srv.image_dir / "migrate").mkdir(exist_ok=True) + (srv.image_dir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG") + srv.set_control(phase="armed", armed_target="g95_cw") + + def world_readable(): + mode = stat.S_IMODE(srv.control_path.stat().st_mode) + return bool(mode & stat.S_IROTH) + + # set_control() oben schreibt direkt aus dem Test, nicht ueber den Server - das + # will hier nicht geprueft werden. Erst die Anfrage unten bringt den Server selbst + # dazu, control.json zu schreiben (armed -> listen), also denselben Codepfad + # (Station.write_control()) wie write_state() fuer status.json. + assert srv.get( + "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img").status == 200 + srv.read_control(lambda _: world_readable()) + assert world_readable(), "control.json ist nach dem Schreiben durch den Server " \ + "nicht welt-lesbar" + finally: + srv.stop() + + +def test_armed_serves_exactly_once_under_concurrency(migration_server): + """C1 (Fixrunde 1): read_control()/Phasenpruefung/Auslieferung/write_control("listen") + liefen ohne gemeinsame Sperre ab - zwei gleichzeitige Anfragen (zwei Lampen, oder Erst- + und Range-Resume-Verbindung derselben Lampe) konnten beide phase == "armed" sehen und + beide die volle Auslieferung bekommen. station.claim_delivery() muss das Lesen und das + Verbrauchen der Freigabe atomar unter einer Sperre machen, damit von n gleichzeitigen + Anfragen gegen dieselbe Freigabe hoechstens eine durchkommt.""" + srv, imgdir = migration_server + (imgdir / "migrate").mkdir(exist_ok=True) + (imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG" * 20000) + srv.set_control(phase="armed", armed_target="g95_cw") + p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img" + + n_workers = 20 + results = [] + results_lock = threading.Lock() + + def worker(): + r = srv.get(p) + with results_lock: + results.append(r.status) + + threads = [threading.Thread(target=worker) for _ in range(n_workers)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + + assert len(results) == n_workers, "nicht alle Worker sind fertig geworden" + assert results.count(200) == 1, f"erwartet genau eine 200, bekommen: {results}" + assert results.count(404) == n_workers - 1 + assert srv.read_control(lambda c: c.get("phase") == "listen")["phase"] == "listen" + + +def test_control_write_failure_blocks_further_armed_deliveries(migration_server): + """I3 (Fixrunde 1): schlaegt das Schreiben von control.json fehl (z.B. volles oder + schreibgeschuetztes /run), darf die Freigabe nicht dauerhaft auf "armed" stehen bleiben + - sonst wuerde jede weitere Lampe des Ziels ebenfalls bedient, genau die Luecke aus C1, + nur diesmal durch einen Schreibfehler statt durch Nebenlaeufigkeit ausgeloest. Die eine + Anfrage, die den Fehler ausloest, bekommt noch ihre Auslieferung (wie eine normale + Freigabe); jede weitere schlaegt fehl, bis das Schreiben wieder klappt. Der Fehler muss + ausserdem in status.json sichtbar sein, sonst wundert sich niemand, warum "arm" nicht + mehr wirkt.""" + srv, imgdir = migration_server + (imgdir / "migrate").mkdir(exist_ok=True) + (imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG") + srv.set_control(phase="armed", armed_target="g95_cw") + p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img" + + control_dir = srv.control_path.parent + old_mode = control_dir.stat().st_mode + os.chmod(control_dir, 0o500) # r-x fuer den Besitzer, kein Schreiben mehr moeglich + try: + r1 = srv.get(p) + assert r1.status == 200, "die eine vorgesehene Auslieferung muss noch rausgehen" + st = srv.status(lambda s: s.get("control_error")) + assert st["control_error"], "Schreibfehler muss in status.json sichtbar sein" + + r2 = srv.get(p) + assert r2.status == 404, \ + "ein Schreibfehler darf die Freigabe nicht dauerhaft offen lassen" + finally: + os.chmod(control_dir, old_mode) + + +def _load_server_module(): + """Laedt server.py wie test_log_error_is_not_silenced_by_log_message oben - als + eigenstaendiges Modulobjekt, ohne es in sys.modules einzutragen, damit mehrfache + Aufrufe (einer je resolve_mode-Test) sich nicht in die Quere kommen.""" + import importlib.util + + spec = importlib.util.spec_from_file_location("ninlight_station_server_rm", SERVER) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_resolve_mode_explicit_mode_wins_over_file(tmp_path): + """I5 (Fixrunde 1): --mode schlaegt --mode-file.""" + module = _load_server_module() + mode_file = tmp_path / "mode" + mode_file.write_text("migration") + assert module.resolve_mode("recovery", mode_file) == "recovery" + + +def test_resolve_mode_reads_the_mode_file(tmp_path): + module = _load_server_module() + mode_file = tmp_path / "mode" + mode_file.write_text("migration") + assert module.resolve_mode(None, mode_file) == "migration" + + +def test_resolve_mode_missing_file_means_recovery(tmp_path): + module = _load_server_module() + mode_file = tmp_path / "mode" # bewusst nicht angelegt + assert module.resolve_mode(None, mode_file) == "recovery" + + +def test_resolve_mode_garbage_file_means_recovery(tmp_path): + module = _load_server_module() + mode_file = tmp_path / "mode" + mode_file.write_text("garbage\n") + assert module.resolve_mode(None, mode_file) == "recovery" + + +def test_resolve_mode_default_without_mode_file_is_recovery(): + module = _load_server_module() + assert module.resolve_mode(None, None) == "recovery"