feat(station): Umschalter fuer Migrationsimages mit/ohne WLAN-Daten

Die WLAN-Zugangsdaten stecken zur Bauzeit in der Firmware, es gibt also
zwei Abzuege je Ziel. Die Station haelt jetzt beide vor - images/
migrate-nocreds (aus tools/build_nocreds.sh, Lampe startet im
Onboarding-AP) und images/migrate-wlan (aus build/, Lampe haengt sofort
im eigenen WLAN) - und der Bediener waehlt per Umschalter, welcher
ausgeliefert wird: "ninlight-station bildsatz" oder der Menueeintrag am
Taster.

Der Zustand liegt in /var/lib/ninlight-station/imageset, ueberlebt
Neustarts und steht dauerhaft in der Fusszeile des Panels - in jedem
Modus, auch bei ausgeschalteter Station: man soll vor dem Einschalten
sehen, was gilt.

Vorgabe und Rueckfallwert in JEDEM Zweifelsfall ist "ohne-wlan". Im
Automatikmodus wird jede Klyqa-Lampe in Reichweite migriert, auch
fremde; ein Abzug ohne Zugangsdaten ist dann ein harmloser Fehlgriff,
einer mit Zugangsdaten traegt das eigene WLAN-Passwort in ein fremdes
Geraet.

Fehlt der gewaehlte Satz (Verzeichnis weg oder leer), wird NICHTS
ausgeliefert - kein stilles Ausweichen auf den anderen Satz. Der Server
antwortet 404 mit Status "noimageset", CLI und Panel sagen "FEHLT".

server.py und die Stations-Unit muessen mit: die Auswahl der
ausgelieferten Datei ist genau die Aufgabe des Servers, und ein
Symlink-Tausch haette "gewaehlt, aber nicht vorhanden" von "dieses
Modell fehlt" nicht mehr unterscheidbar gemacht.
This commit is contained in:
Michael Sauer
2026-09-15 02:17:36 +02:00
parent 3642672c71
commit 300edafc3e
15 changed files with 1381 additions and 200 deletions
+165 -35
View File
@@ -494,12 +494,16 @@ class MigrationTestServer:
control_dir = tmp_path / "control"
control_dir.mkdir()
self.control_path = control_dir / "control.json"
# Der Umschalter des Bildsatzes. Die Datei existiert zunaechst NICHT - das ist
# der Auslieferungszustand und muss "ohne-wlan" (migrate-nocreds/) bedeuten.
self.imageset_path = tmp_path / "imageset"
self.port = free_port()
self.proc = subprocess.Popen(
[sys.executable, str(SERVER), "--image-dir", str(self.image_dir),
"--bind", "127.0.0.1", "--port", str(self.port),
"--state", str(self.state_path), "--mode", mode,
"--control", str(self.control_path)],
"--control", str(self.control_path),
"--imageset-file", str(self.imageset_path)],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
)
deadline = time.time() + 10
@@ -563,6 +567,17 @@ class MigrationTestServer:
def write_raw_control(self, text):
self.control_path.write_text(text)
def set_image_set(self, value):
"""Stellt den Umschalter so um, wie es das CLI tut: eine Datei mit dem Namen des
Bildsatzes. `None` loescht sie wieder (= Auslieferungszustand)."""
if value is None:
try:
self.imageset_path.unlink()
except FileNotFoundError:
pass
return
self.imageset_path.write_text(value)
def stop(self):
self.proc.terminate()
self.proc.wait(timeout=5)
@@ -584,8 +599,8 @@ def migration_server(tmp_path):
def test_migration_auto_serves_known_product(migration_server):
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG" * 1000)
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG" * 1000)
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
@@ -599,9 +614,9 @@ def test_migration_dev_only_in_version_serves_the_dev_image(migration_server):
ging hier das produktionssignierte Image raus (200, und die Lampe lehnte es ab) -
jetzt muss das dev-signierte gehen, und status.json muss die Quelle mitfuehren."""
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"DEV" * 1000)
(imgdir / "migrate" / "g95_cw_prod.img").write_bytes(b"PROD" * 1000)
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"DEV" * 1000)
(imgdir / "migrate-nocreds" / "g95_cw_prod.img").write_bytes(b"PROD" * 1000)
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1.1.2357-dev_ota_light.img")
assert r.status == 200
assert r.body == b"DEV" * 1000
@@ -617,9 +632,9 @@ def test_migration_production_lamp_still_gets_the_prod_image(migration_server):
"""Die erfolgreich migrierte GU10: nirgends ein "-dev", also prod - und die Quelle
"none", damit am Panel/in status.json steht, worauf die Einstufung beruht."""
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "gu10_rgb_dev.img").write_bytes(b"DEV" * 1000)
(imgdir / "migrate" / "gu10_rgb_prod.img").write_bytes(b"PROD" * 1000)
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "gu10_rgb_dev.img").write_bytes(b"DEV" * 1000)
(imgdir / "migrate-nocreds" / "gu10_rgb_prod.img").write_bytes(b"PROD" * 1000)
r = srv.get("/qcx_firmware_-klyqa.lighting.rgb-cw-ww.gu10_V1.1.0.1138_ota_light.img")
assert r.status == 200
assert r.body == b"PROD" * 1000
@@ -630,8 +645,8 @@ def test_migration_production_lamp_still_gets_the_prod_image(migration_server):
def test_migration_listen_records_but_serves_nothing(migration_server):
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG")
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
srv.set_control(phase="listen")
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img")
assert r.status == 404
@@ -643,8 +658,8 @@ def test_migration_listen_records_but_serves_nothing(migration_server):
def test_arm_serves_exactly_once_then_falls_back_to_listen(migration_server):
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG")
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "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"
assert srv.get(p).status == 200
@@ -654,8 +669,8 @@ def test_arm_serves_exactly_once_then_falls_back_to_listen(migration_server):
def test_armed_for_other_target_does_not_serve(migration_server):
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG")
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
srv.set_control(phase="armed", armed_target="gu10_rgb")
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img")
assert r.status == 404
@@ -671,7 +686,12 @@ def test_unknown_product_id_is_404_and_logged(migration_server):
def test_missing_migrate_image_is_404(migration_server):
srv, _ = migration_server
srv, imgdir = migration_server
# Der Bildsatz ist da (er enthaelt ein anderes Modell), nur dieses eine Modell fehlt
# darin - ein anderer Zustand als ein fehlender Bildsatz (siehe
# test_a_selected_but_missing_image_set_delivers_nothing).
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.e14_V1_ota_light.img")
assert r.status == 404
st = srv.status(lambda s: len(s["recent"]) >= 1)
@@ -692,8 +712,8 @@ def test_migration_path_is_404_in_recovery_mode(server):
def test_missing_control_file_means_auto(migration_server):
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG")
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
srv.delete_control()
assert srv.get(
"/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img").status == 200
@@ -701,8 +721,8 @@ def test_missing_control_file_means_auto(migration_server):
def test_corrupt_control_file_means_auto(migration_server):
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG")
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
srv.write_raw_control("{das ist kein json")
assert srv.get(
"/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img").status == 200
@@ -710,8 +730,8 @@ def test_corrupt_control_file_means_auto(migration_server):
def test_range_resume_on_migration_image(migration_server):
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(bytes(range(256)) * 40)
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(bytes(range(256)) * 40)
p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"
r = srv.get(p, headers={"Range": "bytes=1000-"})
assert r.status == 206
@@ -724,9 +744,9 @@ def test_version_comes_from_manifest(migration_server):
Version traegt - Bestandskompatibilitaet mit einem Manifest aus der Zeit vor
Fixrunde 2 (N2)."""
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG")
(imgdir / "migrate" / "manifest.json").write_text(json.dumps({
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
(imgdir / "migrate-nocreds" / "manifest.json").write_text(json.dumps({
"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")
@@ -742,10 +762,10 @@ def test_version_prefers_the_delivered_entry_over_the_manifest_wide_field(migrat
darf dann nicht das manifestweite Feld fuer beide protokollieren, sondern muss die
Version DES AUSGELIEFERTEN Eintrags nehmen."""
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG-DEV")
(imgdir / "migrate" / "g95_cw_prod.img").write_bytes(b"IMG-PROD")
(imgdir / "migrate" / "manifest.json").write_text(json.dumps({
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG-DEV")
(imgdir / "migrate-nocreds" / "g95_cw_prod.img").write_bytes(b"IMG-PROD")
(imgdir / "migrate-nocreds" / "manifest.json").write_text(json.dumps({
# Manifestweites Feld absichtlich falsch/veraltet, damit ein Zurueckfallen
# darauf im Test auffallen wuerde.
"ninlight_version": "1.2.0",
@@ -770,8 +790,8 @@ def test_control_file_is_world_readable_even_under_a_restrictive_umask(tmp_path)
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.image_dir / "migrate-nocreds").mkdir(exist_ok=True)
(srv.image_dir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
srv.set_control(phase="armed", armed_target="g95_cw")
def world_readable():
@@ -799,8 +819,8 @@ def test_armed_serves_exactly_once_under_concurrency(migration_server):
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)
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "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"
@@ -835,8 +855,8 @@ def test_control_write_failure_blocks_further_armed_deliveries(migration_server)
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")
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "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"
@@ -899,3 +919,113 @@ def test_resolve_mode_garbage_file_means_recovery(tmp_path):
def test_resolve_mode_default_without_mode_file_is_recovery():
module = _load_server_module()
assert module.resolve_mode(None, None) == "recovery"
# --- Umschalter: mit oder ohne einkompilierte WLAN-Zugangsdaten ---------------------
#
# Zwei Bildsaetze liegen auf dem Pi (images/migrate-nocreds und images/migrate-wlan);
# welcher ausgeliefert wird, entscheidet allein der Umschalter. Geprueft wird deshalb
# immer an der TATSAECHLICH AUSGELIEFERTEN DATEI, nicht an einem Pfad oder einem
# Statusfeld - der Fehlgriff, gegen den der Umschalter da ist, ist ein falscher Inhalt
# in einer Lampe.
def _both_sets(imgdir):
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"OHNE" * 100)
(imgdir / "migrate-wlan").mkdir(exist_ok=True)
(imgdir / "migrate-wlan" / "g95_cw_dev.img").write_bytes(b"MITW" * 100)
G95_DEV = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"
def test_the_default_image_set_is_the_one_without_wifi_credentials(migration_server):
"""Ohne Umschalterdatei (Auslieferungszustand) geht der Abzug OHNE Zugangsdaten
raus. Die sichere Vorgabe ist die, bei der ein Fehlgriff niemandem das eigene
WLAN-Passwort in ein fremdes Geraet schreibt."""
srv, imgdir = migration_server
_both_sets(imgdir)
r = srv.get(G95_DEV)
assert r.status == 200
assert r.body == b"OHNE" * 100
def test_the_switch_decides_which_file_is_actually_delivered(migration_server):
srv, imgdir = migration_server
_both_sets(imgdir)
srv.set_image_set("mit-wlan")
assert srv.get(G95_DEV).body == b"MITW" * 100
# ... und wieder zurueck, ohne Dienstneustart: der Umschalter wirkt sofort.
srv.set_image_set("ohne-wlan")
assert srv.get(G95_DEV).body == b"OHNE" * 100
def test_a_broken_switch_file_falls_back_to_the_safe_set(migration_server):
"""Unbekannter Inhalt ist kein Grund, den teureren Satz zu nehmen."""
srv, imgdir = migration_server
_both_sets(imgdir)
srv.set_image_set("voellig-anderes")
assert srv.get(G95_DEV).body == b"OHNE" * 100
def test_a_selected_but_missing_image_set_delivers_nothing(migration_server):
"""Der gewaehlte Satz liegt nicht auf dem Pi: es darf NICHTS rausgehen - insbesondere
nicht die Datei des anderen Satzes, die daneben liegt und genauso heisst."""
srv, imgdir = migration_server
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"OHNE" * 100)
srv.set_image_set("mit-wlan") # migrate-wlan/ existiert nicht
r = srv.get(G95_DEV)
assert r.status == 404
assert b"OHNE" not in r.body
st = srv.status(lambda s: len(s["recent"]) >= 1)
assert st["served"] == 0
# Eigener Status, damit am Panel/in status.json "Satz fehlt" von "dieses Modell
# fehlt" unterscheidbar bleibt.
assert st["recent"][-1]["status"] == "noimageset"
def test_a_missing_image_set_does_not_burn_the_release(migration_server):
"""Wie beim fehlenden Einzelimage: eine Freigabe darf nicht daran verfallen, dass es
gar nichts auszuliefern gab."""
srv, imgdir = migration_server
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
srv.set_image_set("mit-wlan")
srv.set_control(phase="armed", armed_target="g95_cw")
assert srv.get(G95_DEV).status == 404
assert srv.read_control()["phase"] == "armed"
def test_the_version_comes_from_the_manifest_of_the_selected_set(migration_server):
"""Nach dem Umschalten darf nicht die Version des vorigen Satzes stehenbleiben - die
beiden manifest.json entstehen im selben Deploy und koennen dieselbe mtime tragen."""
srv, imgdir = migration_server
_both_sets(imgdir)
for name, version in (("migrate-nocreds", "0.4.0-onb"), ("migrate-wlan", "0.4.0-wlan")):
(imgdir / name / "manifest.json").write_text(json.dumps({
"images": [{"target": "g95_cw", "variant": "dev", "file": "g95_cw_dev.img",
"ninlight_version": version}]}))
# Beiden Manifesten dieselbe mtime geben - genau der Fall, den ein reiner
# mtime-Vergleich nicht bemerkt haette.
stamp = 1600000000
for name in ("migrate-nocreds", "migrate-wlan"):
os.utime(imgdir / name / "manifest.json", (stamp, stamp))
assert srv.get(G95_DEV).status == 200
assert srv.status(lambda s: s["recent"])["recent"][-1]["version"] == "0.4.0-onb"
srv.set_image_set("mit-wlan")
assert srv.get(G95_DEV).status == 200
assert srv.status(
lambda s: len(s["recent"]) >= 2
)["recent"][-1]["version"] == "0.4.0-wlan"
def test_read_image_set_defaults_without_a_file(tmp_path):
module = _load_server_module()
assert module.read_image_set(None) == "ohne-wlan"
assert module.read_image_set(tmp_path / "fehlt") == "ohne-wlan"
p = tmp_path / "imageset"
p.write_text("mit-wlan\n")
assert module.read_image_set(p) == "mit-wlan"