Files
ninlight-w600/tools/station/server.py
T
Michael Sauer 1cb5359754 fix(station): Manifest-Merge robust gegen Fehlform, Version pro Eintrag
Fixrunde 2 zu "Migrationsimages beim Deploy signieren":

- N1 (Important): das Merge-Skript im SSH-Heredoc von deploy.sh fing beim
  Bestandsmanifest nur Lesefehler und kaputtes JSON ab (OSError/ValueError).
  War die Datei gueltiges JSON mit falscher Form ("[]", "null", ein String,
  oder ein images-Eintrag, der selbst kein Objekt ist), krachte .get() mit
  AttributeError - und das VOR der Installation von
  server.py/products.py/CLI/Unit und vor dem Neustart, liess einen Pi also
  mit neuen Images, aber altem Stationscode zurueck. Jetzt wird alte wie
  neue Manifest-Struktur konsequent auf isinstance(dict)/isinstance(list)
  geprueft, jede Fehlform faellt auf ein leeres Manifest zurueck statt
  abzustuerzen. Neues Szenario 21 deckt alle vier Faelle aus der
  Review-Tabelle ab (kaputt, leer, falsch geformtes JSON, kaputter Eintrag).

- N2 (Important): das zusammengefuehrte Manifest hat "ninlight_version"
  bisher blind vom neuen Manifest uebernommen, obwohl uebernommene
  Alt-Eintraege (deren Images unveraendert liegen bleiben, siehe Fixrunde 1
  I1) eine andere Firmware-Version tragen koennen - eine konkrete falsche
  Version beim Protokollieren ist schlechter als das vorherige "unbekannt".
  Die Version wandert jetzt PRO EINTRAG ins Manifest (deploy.sh erzeugt sie
  dort, der Merge behaelt sie); das manifestweite Feld bleibt nur als
  Bestandskompatibilitaet und wird nur gesetzt, wenn wirklich alle Eintraege
  uebereinstimmen. server.py liest beim Ausliefern jetzt die Version DES
  AUSGELIEFERTEN EINTRAGS (Station.image_version(target, variant) ersetzt
  manifest_version()) und faellt nur auf das manifestweite Feld zurueck,
  wenn der Eintrag selbst keine eigene traegt. test_server.py bekommt einen
  neuen Test fuer den Vorrang des Eintrags vor dem manifestweiten Feld,
  Szenario 14 in test_deploy.sh prueft jetzt zusaetzlich unterschiedliche
  Versionen zwischen zusammengefuehrten Eintraegen und dass das
  manifestweite Feld dann fehlt statt falsch zu sein. Die Spec
  (docs/superpowers/specs/.../design.md, Abschnitt 4, lokal/nicht
  versioniert) ist nachgezogen. Die dadurch ueberfluessig gewordene
  "sorted(versions)[0]"-Stelle aus Fixrunde 1 ist entfallen.

- Kleinigkeit: die Remote-Meldung, wenn migrate/ von diesem Rechner leer
  ist, behauptet nicht mehr "kein Schluesselverzeichnis" (das stimmt im
  Fall "Verzeichnis vorhanden, aber leer" nicht) - sie nennt jetzt keinen
  bestimmten Grund mehr, nur die Konsequenz.
- Kleinigkeit: --exclude=manifest.json ist jetzt --exclude=/manifest.json,
  konsistent mit dem daneben verankerten --exclude=/migrate.

Alle drei Aenderungen (N1, N2, beide Kleinigkeiten) sind vor dem Fix per
Rot-Probe an einer Kopie von deploy.sh verifiziert: mit den urspruenglichen
Fehlern schlagen Szenario 21 (N1) bzw. Szenario 14 (N2) zuverlaessig fehl.

Bewusst nicht angefasst (Review-Vorgabe): der fehlende HOME-Testfall, und
dass der Merge kein os.path.exists() prueft.
2026-09-13 20:40:03 +02:00

575 lines
29 KiB
Python
Executable File

#!/usr/bin/env python3
"""ninlight recovery/migration station.
Serves firmware images to lamps that look for a rescue station while booting (mode
"recovery": ninlight_<target>_ota.img) or to old Klyqa lamps looking for their original
update server (mode "migration": qcx_firmware_<product-id>_<version>_ota_light.img).
Deliberately minimal: only files matching one of these two exact patterns ever leave the
machine, and only the pattern for the active mode is served.
Standard library only - the Pi gets no extra packages.
"""
from __future__ import annotations
import argparse
import contextlib
import json
import os
import re
import select
import socket
import sys
import tempfile
import threading
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
# products.py lives next to this file. sys.path[0] already covers that when server.py is
# run the normal way (as a script or via `python3 server.py ...`), but a test that loads
# this module through importlib.util.spec_from_file_location (see test_server.py's
# test_log_error_is_not_silenced_by_log_message) does not get that for free.
sys.path.insert(0, str(Path(__file__).resolve().parent))
import products
TARGETS = ("evk", "e27_rgb", "e14_rgb", "gu10_rgb", "e27_cw", "e14_cw", "gu10_cw", "g95_cw")
PATH_RE = re.compile(r"^/ninlight_(" + "|".join(TARGETS) + r")_ota\.img$")
TRAILER_SIZE = 512
TRAILER_MAGIC = b"NINL"
VERSION_OFF = 21
VERSION_LEN = 16
RECENT_MAX = 8
CHUNK = 32 * 1024
def now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def trailer_version(path: Path) -> str:
"""Firmware version from the image trailer, or '?' if the file carries none."""
try:
with path.open("rb") as f:
if path.stat().st_size < TRAILER_SIZE:
return "?"
f.seek(-TRAILER_SIZE, os.SEEK_END)
tail = f.read(TRAILER_SIZE)
except OSError:
return "?"
if tail[0:4] != TRAILER_MAGIC:
return "?"
return tail[VERSION_OFF:VERSION_OFF + VERSION_LEN].rstrip(b"\x00").decode("ascii", "replace")
class Station:
def __init__(self, image_dir: Path, state_path: Path, stall_after: int | None,
mode: str = "recovery", control_path: Path | None = None) -> None:
self.image_dir = image_dir
self.state_path = state_path
self.stall_after = stall_after
self.mode = mode
self.control_path = control_path or Path("/run/ninlight-station/control.json")
self.since = now_iso()
self.served = 0
self.recent: list[dict] = []
# Zuletzt erkanntes Modell im Migrationsmodus, auch wenn (noch) nichts ausgeliefert
# wurde - das ist es, was die Anzeige gross darstellt (Spec 5.2).
self.last_seen: dict | None = None
self._manifest_mtime: float | None = None
# Der geparste Inhalt von manifest.json (oder {}, wenn sie fehlt/kaputt/falsch
# geformt ist) statt nur einer einzelnen Version - seit Fixrunde 2 (N2) traegt
# jeder Eintrag seine eigene Version, image_version() muss also den ganzen
# Eintragssatz vorhalten, nicht nur einen String.
self._manifest_cache: dict = {}
# 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
# every reconnect a brand-new --stall-after allowance and the stall would never
# actually stick (see reserve_stall_bytes below).
self.stall_sent: dict[str, int] = {}
# ThreadingHTTPServer runs one handler thread per connection, and several lamps
# can be downloading at once. Every mutation of served/recent/stall_sent and the
# snapshot that gets serialized to disk must happen under this same lock, or
# concurrent requests lose increments and stomp on each other's status.json writes.
self.lock = threading.RLock()
self.write_state()
def image_path(self, target: str) -> Path:
return self.image_dir / f"ninlight_{target}_ota.img"
def migrate_image_path(self, target: str, variant: str) -> Path:
return self.image_dir / "migrate" / f"{target}_{variant}.img"
def _load_manifest(self) -> dict:
"""Geparster Inhalt von <image_dir>/migrate/manifest.json, neu eingelesen bei
jeder Aenderung der Datei (mtime-Vergleich). Fehlt sie, ist sie unlesbar/kaputtes
JSON, oder ist ihr Wurzelelement kein Objekt, ist das Ergebnis {} - dieselbe
Formpruefung wie beim Merge in deploy.sh (Fixrunde 2, N1), damit eine falsch
geformte, aber syntaktisch gueltige Datei nicht mit einem AttributeError den
Server abschiesst."""
path = self.image_dir / "migrate" / "manifest.json"
try:
mtime = path.stat().st_mtime
except OSError:
self._manifest_mtime = None
self._manifest_cache = {}
return self._manifest_cache
if mtime != self._manifest_mtime:
try:
data = json.loads(path.read_text())
except (OSError, ValueError):
data = {}
if not isinstance(data, dict):
data = {}
self._manifest_cache = data
self._manifest_mtime = mtime
return self._manifest_cache
def image_version(self, target: str, variant: str) -> str:
"""Firmware-Version fuer genau das Migrationsimage, das gerade ausgeliefert
wird - die Migrationsimages selbst tragen keinen ninlight-Trailer, aus dem der
Server sie lesen koennte (Spec Abschnitt 4). Bevorzugt die Version aus dem
passenden Eintrag in manifest.json; nur wenn der Eintrag keine eigene traegt
(Bestandskompatibilitaet mit einem manifestweiten Feld), faellt es auf das
Feld auf oberster Ebene zurueck. Ein einzelnes manifestweites Feld allein
waere seit dem Zusammenfuehren mehrerer Deploys (deploy.sh, Fixrunde 1 I1)
schlicht nicht mehr wahr: uebernommene Alt-Eintraege koennen eine andere
Version tragen als die gerade neu signierten (Fixrunde 2, N2)."""
data = self._load_manifest()
images = data.get("images")
if isinstance(images, list):
for entry in images:
if not isinstance(entry, dict):
continue
if entry.get("target") == target and entry.get("variant") == variant:
version = entry.get("ninlight_version")
if isinstance(version, str) and version:
return version
break
top = data.get("ninlight_version")
return top if isinstance(top, str) and top else "unbekannt"
def read_control(self):
"""Phase und Freigabe. Fehlt die Datei oder ist sie unlesbar/kaputt, gilt "auto" -
eine fehlende Steuerdatei darf eine Migration nie blockieren, sondern hoechstens
bequemer machen als beabsichtigt. Bei jeder Anfrage frisch gelesen, damit
"ninlight-station arm" ohne Neustart des Dienstes wirkt."""
try:
d = json.loads(self.control_path.read_text())
except (OSError, ValueError):
return "auto", None
phase = d.get("phase")
if phase not in ("auto", "listen", "armed"):
return "auto", None
return phase, d.get("armed_target")
def write_control(self, phase: str, armed_target: str | None) -> None:
with self.lock:
try:
self.control_path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=self.control_path.parent, prefix=".control-")
try:
with os.fdopen(fd, "w") as f:
json.dump({"phase": phase, "armed_target": armed_target}, f)
# Dasselbe Muster wie write_state() unten, aus demselben Grund: der
# Dienst laeuft als root, mkstemp() legt 0600 an, und ohne das
# explizite chmod() koennte das CLI die Steuerdatei nicht lesen.
os.chmod(tmp, 0o644)
os.replace(tmp, self.control_path)
except BaseException:
with contextlib.suppress(OSError):
os.unlink(tmp)
raise
except OSError as e:
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.
Tracks bytes cumulatively per client IP, across requests, so a stalled connection
that the lamp abandons and resumes via Range does not get a fresh budget just
because it is a new TCP connection. Only called when stall_after is set.
"""
with self.lock:
sent_so_far = self.stall_sent.get(ip, 0)
allowed = max(0, min(n, self.stall_after - sent_so_far))
self.stall_sent[ip] = sent_so_far + allowed
return allowed
def record(self, ip: str, target: str, version: str, nbytes: int, status: str) -> None:
with self.lock:
if status == "ok":
self.served += 1
self.recent.append({"ts": now_iso(), "ip": ip, "target": target,
"version": version, "bytes": nbytes, "status": status})
del self.recent[:-RECENT_MAX]
self.write_state()
def record_migration(self, ip: str, target: str | None, variant: str | None,
product_id: str, lamp_version: str, product_name: str | None,
version: str, nbytes: int, status: str) -> None:
"""Protokolliert eine Migrationsanfrage. Anders als record() oben wird hier auch
dann ein Eintrag angelegt (Status "listen"/"unknown"/"notfound"), wenn nichts
ausgeliefert wurde - das ist der ganze Zweck der Horchphase. last_seen wird nur bei
einer erkannten Produkt-ID gesetzt (target is not None); eine unbekannte Produkt-ID
hat kein Modell, das die Anzeige zeigen koennte."""
with self.lock:
if status == "ok":
self.served += 1
entry = {"ts": now_iso(), "ip": ip, "target": target, "variant": variant,
"product_id": product_id, "lamp_version": lamp_version,
"version": version, "bytes": nbytes, "status": status}
self.recent.append(entry)
del self.recent[:-RECENT_MAX]
if target is not None:
self.last_seen = {"product_id": product_id, "target": target,
"variant": variant, "product_name": product_name,
"lamp_version": lamp_version, "ip": ip, "ts": entry["ts"]}
self.write_state()
def write_state(self) -> None:
with self.lock:
payload = {"since": self.since, "mode": self.mode, "served": self.served,
"recent": list(self.recent)}
if self.mode == "migration":
phase, armed_target = self.read_control()
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-")
try:
with os.fdopen(fd, "w") as f:
json.dump(payload, f, indent=2)
# The service runs as root and mkstemp() creates the file 0600, so
# without this an operator reading status.json as a normal user (see
# tools/station/ninlight-station) gets "permission denied" - and since
# os.replace() preserves the mode it arrives with, that wrong mode
# would otherwise survive onto the live file on every single write,
# not just the first. os.chmod() ignores the process umask (unlike the
# mode argument to mkstemp/open), so a restrictive umask on the Pi
# can't quietly bring the 0600 back. Set it before the rename so
# there's never a window where the path holds a 0600 file.
os.chmod(tmp, 0o644)
os.replace(tmp, self.state_path)
except BaseException:
with contextlib.suppress(OSError):
os.unlink(tmp)
raise
except OSError as e:
print(f"warning: cannot write {self.state_path}: {e}", flush=True)
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
server_version = "ninlight-station"
@property
def station(self) -> Station:
return self.server.station # type: ignore[attr-defined]
def log_message(self, fmt: str, *args) -> None: # noqa: D102 - quieter default log
pass
def log_error(self, fmt: str, *args) -> None:
# BaseHTTPRequestHandler.log_error() calls log_message() by default, which we
# silence above in favour of our own log() calls - but that also silences it for
# requests that never reach do_GET/do_POST (a malformed request line, a client
# timeout mid-request). The station is the only diagnostic tool available during a
# rescue, so those cases must still produce a line instead of nothing.
try:
client = self.client_address[0]
except (AttributeError, IndexError):
client = "?"
print(f"{now_iso()} {client} ERROR {fmt % args}", flush=True)
def log(self, status: int, nbytes: int, started: float) -> None:
print(f"{now_iso()} {self.client_address[0]} {self.path} {status} "
f"{nbytes}B {time.monotonic() - started:.1f}s", flush=True)
def do_POST(self) -> None:
started = time.monotonic()
self.reject(405, "method not allowed")
self.log(405, 0, started)
do_PUT = do_DELETE = do_HEAD = do_POST
def reject(self, status: int, text: str) -> None:
body = (text + "\n").encode()
self.send_response(status)
self.send_header("Content-Type", "text/plain")
self.send_header("Content-Length", str(len(body)))
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None:
started = time.monotonic()
if self.station.mode == "migration":
self._do_get_migration(started)
else:
self._do_get_recovery(started)
def _do_get_recovery(self, started: float) -> None:
m = PATH_RE.match(self.path)
if not m:
self.reject(404, "not found")
self.log(404, 0, started)
return
target = m.group(1)
path = self.station.image_path(target)
if not path.is_file():
# Record before responding: a client that sees the 404 complete may check
# status.json right away, so the entry must already be on disk by then.
self.station.record(self.client_address[0], target, "?", 0, "notfound")
self.reject(404, "no image for this device")
self.log(404, 0, started)
return
version = trailer_version(path)
sent, status = self._serve_file(path, started)
# Record only now that the transfer has actually finished (or failed): "ok" must
# never be written before the last byte is confirmed sent, or status.json can
# claim success for a device that never received the end of the image.
self.station.record(self.client_address[0], target, version, sent, status)
def _do_get_migration(self, started: float) -> None:
parsed = products.parse_request(self.path)
if parsed is None:
# Ein fremder Pfad, kein qcx_firmware_*-Muster - kein Protokolleintrag in
# status.json, das waere nur Rauschen (Brief Schritt 3.4).
self.reject(404, "not found")
self.log(404, 0, started)
return
pid, lamp_version, product = parsed
ip = self.client_address[0]
if product is None:
self.station.record_migration(ip, None, None, pid, lamp_version, None, "?", 0,
"unknown")
self.reject(404, "unknown product")
self.log(404, 0, started)
return
target, variant, name = product.target, product.variant, product.name
version = self.station.image_version(target, variant)
# 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 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,
version, 0, "listen")
self.reject(404, "listening only")
self.log(404, 0, started)
return
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")
self.log(404, 0, started)
return
sent, status = self._serve_file(path, started)
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)
def _serve_file(self, path: Path, started: float) -> tuple[int, str]:
"""Sendet `path` mit Range-Unterstuetzung; liefert (gesendete Bytes, Status).
Gemeinsam fuer Rettungs- und Migrationsroute - Uebertragungsmechanik (Range,
--stall-after, Abbruchbehandlung) ist identisch, nur was mit dem Ergebnis
geschieht, unterscheidet sich."""
size = path.stat().st_size
start = 0
rng = self.headers.get("Range", "")
if rng.startswith("bytes="):
first = rng.split("=", 1)[1].split("-", 1)[0]
if first.isdigit() and int(first) < size:
start = int(first)
self.send_response(206 if start else 200)
self.send_header("Content-Type", "application/octet-stream")
self.send_header("Content-Length", str(size - start))
if start:
self.send_header("Content-Range", f"bytes {start}-{size - 1}/{size}")
self.send_header("Accept-Ranges", "bytes")
self.send_header("Connection", "close")
self.end_headers()
sent = 0
status = "ok"
remaining = size - start
client_ip = self.client_address[0]
try:
with path.open("rb") as f:
f.seek(start)
while sent < remaining:
read_size = min(CHUNK, remaining - sent)
if self.station.stall_after is not None:
# Reserve against this client's cumulative budget, not just this
# request's byte count - see Station.reserve_stall_bytes.
read_size = self.station.reserve_stall_bytes(client_ip, read_size)
if read_size == 0:
# test aid: hold the connection open and send nothing more, so
# the firmware's stall handling can be exercised on real
# hardware. We still notice if the peer gives up and closes its
# side, so a test (or an impatient client) that hangs up is
# recorded as "aborted" instead of leaking the thread forever.
while True:
readable, _, _ = select.select([self.connection], [], [], 1.0)
if readable:
try:
peek = self.connection.recv(1, socket.MSG_PEEK)
except OSError:
peek = b""
if not peek:
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)
except (BrokenPipeError, ConnectionResetError, TimeoutError):
status = "aborted"
self.log(206 if start else 200, sent, started)
return sent, status
def resolve_mode(explicit_mode: str | None, mode_file: Path | None) -> str:
"""Bestimmt den Betriebsmodus. Vorrang: ein ausdrueckliches --mode schlaegt
--mode-file, --mode-file schlaegt die Vorgabe "recovery". Eine fehlende oder
unlesbare/unerwartete Modusdatei bedeutet ebenfalls "recovery" - Task 4 baut die
systemd-Unit darauf auf, dass der Modus ohne Neustart-Argumentaenderung aus einer
Datei kommt (/run/ninlight-station/mode), und ein fehlerhafter Zustand dieser Datei
darf den Dienst nicht zum Absturz bringen oder unbeabsichtigt scharf schalten."""
if explicit_mode is not None:
return explicit_mode
if mode_file is not None:
try:
text = mode_file.read_text().strip()
except OSError:
return "recovery"
if text in ("recovery", "migration"):
return text
return "recovery"
return "recovery"
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(description="ninlight recovery/migration station HTTP server")
p.add_argument("--image-dir", required=True, type=Path)
p.add_argument("--bind", default="192.168.8.1")
p.add_argument("--port", type=int, default=80)
p.add_argument("--state", type=Path, default=Path("/run/ninlight-station/status.json"))
p.add_argument("--mode", choices=("recovery", "migration"), default=None,
help="ausdruecklicher Modus - schlaegt --mode-file (Vorgabe: siehe dort)")
p.add_argument("--mode-file", type=Path, default=None,
help="Datei mit 'recovery' oder 'migration'; wird ohne --mode verwendet, "
"Vorgabe bei fehlender/kaputter Datei ist 'recovery'")
p.add_argument("--control", type=Path, default=Path("/run/ninlight-station/control.json"),
help="Steuerdatei fuer Phase/Freigabe im Migrationsmodus")
p.add_argument("--stall-after", type=int, default=None,
help="test aid: stop sending after N bytes but keep the connection open")
a = p.parse_args(argv)
if not a.image_dir.is_dir():
print(f"error: image directory {a.image_dir} does not exist", file=sys.stderr)
return 2
mode = resolve_mode(a.mode, a.mode_file)
station = Station(a.image_dir, a.state, a.stall_after, mode=mode, control_path=a.control)
httpd = ThreadingHTTPServer((a.bind, a.port), Handler)
httpd.daemon_threads = True
httpd.station = station # type: ignore[attr-defined]
images = sorted(p.name for p in a.image_dir.glob("ninlight_*_ota.img"))
print(f"{now_iso()} listening on {a.bind}:{a.port}, mode={mode}, {len(images)} image(s): "
f"{', '.join(images) or 'none'}", flush=True)
if a.stall_after is not None:
print(f"{now_iso()} WARNING: --stall-after {a.stall_after} is a test aid", flush=True)
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
return 0
if __name__ == "__main__":
raise SystemExit(main())