From e53b049b6c03df26c57279c18951d3060c3ebb3b Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 11:39:28 +0200 Subject: [PATCH 01/11] =?UTF-8?q?feat(profiles):=20sch=C3=A9ma=20et=20char?= =?UTF-8?q?gement=20des=20manifestes=20module?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Valide strictement : un manifeste mal typé deviendrait une décision d'extinction erronée en Phase 3. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/__init__.py | 0 packages/secubox-profiles/api/manifest.py | 110 ++++++++++++++++++ packages/secubox-profiles/tests/conftest.py | 10 ++ .../secubox-profiles/tests/test_manifest.py | 99 ++++++++++++++++ 4 files changed, 219 insertions(+) create mode 100644 packages/secubox-profiles/api/__init__.py create mode 100644 packages/secubox-profiles/api/manifest.py create mode 100644 packages/secubox-profiles/tests/conftest.py create mode 100644 packages/secubox-profiles/tests/test_manifest.py diff --git a/packages/secubox-profiles/api/__init__.py b/packages/secubox-profiles/api/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/secubox-profiles/api/manifest.py b/packages/secubox-profiles/api/manifest.py new file mode 100644 index 00000000..f0e7b0e0 --- /dev/null +++ b/packages/secubox-profiles/api/manifest.py @@ -0,0 +1,110 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +SecuBox-Deb :: profiles — schéma et chargement des manifestes module +CyberMind — https://cybermind.fr + +Un manifeste décrit le CYCLE DE VIE d'un module : ses units, son runtime, son +exposition, sa priorité. Il ne duplique pas menu.d/ (path, ordre, icône), qui +reste la source UI avec son propre cycle de vie. + +La validation est stricte et bruyante : en Phase 3 un manifeste mal typé +deviendrait une décision d'extinction erronée. Mieux vaut refuser au chargement. +""" +from __future__ import annotations + +import tomllib +from dataclasses import dataclass +from pathlib import Path + +RUNTIMES = ("native", "lxc") +EXPOSURES = ("public", "lan", "internal") +CATEGORIES = ("media", "security", "network", "infra", "dev", "mesh") + +DEFAULT_PRIORITY = 50 + + +class ManifestError(Exception): + """Manifeste illisible ou invalide.""" + + +@dataclass(frozen=True) +class Manifest: + id: str + category: str + runtime: str + exposure: str + units: tuple[str, ...] + lxc: str | None = None + portal_domain: str | None = None + priority: int = DEFAULT_PRIORITY + protected: bool = False + needs: tuple[str, ...] = () + + +def _require(d: dict, key: str, path: Path): + if key not in d: + raise ManifestError(f"{path}: champ obligatoire manquant: {key}") + return d[key] + + +def _enum(value, allowed: tuple[str, ...], key: str, path: Path) -> str: + if value not in allowed: + raise ManifestError(f"{path}: {key}={value!r} invalide (attendu: {', '.join(allowed)})") + return value + + +def load_manifest(path: Path) -> Manifest: + """Charge et valide un manifeste. Lève ManifestError sur tout écart.""" + try: + d = tomllib.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError) as exc: + raise ManifestError(f"{path}: illisible: {exc}") from exc + + mid = _require(d, "id", path) + # L'id pilote pins et profils : s'il diverge du nom de fichier, un pin + # viserait un module qui n'existe pas sous ce nom. + if mid != path.stem: + raise ManifestError(f"{path}: id={mid!r} ne correspond pas au nom de fichier {path.stem!r}") + + runtime = _enum(_require(d, "runtime", path), RUNTIMES, "runtime", path) + lxc = d.get("lxc") + if runtime == "lxc" and not lxc: + raise ManifestError(f"{path}: runtime='lxc' exige le champ lxc=") + + units = _require(d, "units", path) + if not isinstance(units, list) or not all(isinstance(u, str) for u in units): + raise ManifestError(f"{path}: units doit être une liste de chaînes") + + priority = d.get("priority", DEFAULT_PRIORITY) + if not isinstance(priority, int) or isinstance(priority, bool) or not 0 <= priority <= 100: + raise ManifestError(f"{path}: priority={priority!r} hors bornes (entier 0-100)") + + portal = d.get("portal") or {} + if not isinstance(portal, dict): + raise ManifestError(f"{path}: portal doit être une table") + + return Manifest( + id=mid, + category=_enum(_require(d, "category", path), CATEGORIES, "category", path), + runtime=runtime, + exposure=_enum(_require(d, "exposure", path), EXPOSURES, "exposure", path), + units=tuple(units), + lxc=lxc, + portal_domain=portal.get("domain"), + priority=priority, + protected=bool(d.get("protected", False)), + needs=tuple(d.get("needs", ())), + ) + + +def load_all(directory: Path) -> dict[str, Manifest]: + """Charge tous les *.toml d'un répertoire, indexés par id.""" + out: dict[str, Manifest] = {} + for p in sorted(Path(directory).glob("*.toml")): + m = load_manifest(p) + out[m.id] = m + return out diff --git a/packages/secubox-profiles/tests/conftest.py b/packages/secubox-profiles/tests/conftest.py new file mode 100644 index 00000000..dd54eed0 --- /dev/null +++ b/packages/secubox-profiles/tests/conftest.py @@ -0,0 +1,10 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. +import sys +from pathlib import Path + +# Rend `api` importable en top-level (miroir du layout runtime sous +# /usr/lib/secubox/profiles), comme le conftest de secubox-billets. +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) diff --git a/packages/secubox-profiles/tests/test_manifest.py b/packages/secubox-profiles/tests/test_manifest.py new file mode 100644 index 00000000..2673a7a7 --- /dev/null +++ b/packages/secubox-profiles/tests/test_manifest.py @@ -0,0 +1,99 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. +import pytest + +from api.manifest import Manifest, ManifestError, load_all, load_manifest + +FULL = """ +id = "peertube" +category = "media" +runtime = "lxc" +exposure = "public" +units = ["secubox-peertube.service"] +lxc = "peertube" +portal = { domain = "peertube.gk2.secubox.in" } +priority = 40 +protected = false +needs = ["auth"] +""" + +MINIMAL = """ +id = "lyrion" +category = "media" +runtime = "native" +exposure = "lan" +units = ["secubox-lyrion.service"] +""" + + +def test_load_full_manifest(tmp_path): + p = tmp_path / "peertube.toml" + p.write_text(FULL) + m = load_manifest(p) + assert m == Manifest( + id="peertube", category="media", runtime="lxc", exposure="public", + units=("secubox-peertube.service",), lxc="peertube", + portal_domain="peertube.gk2.secubox.in", priority=40, + protected=False, needs=("auth",), + ) + + +def test_defaults_are_applied(tmp_path): + # Un manifeste minimal doit rester valide : la plupart des 134 modules + # n'ont ni LXC, ni portail, ni deps. + p = tmp_path / "lyrion.toml" + p.write_text(MINIMAL) + m = load_manifest(p) + assert m.lxc is None and m.portal_domain is None + assert m.priority == 50 and m.protected is False and m.needs == () + + +@pytest.mark.parametrize("field,bad", [ + ("runtime", '"docker"'), + ("exposure", '"world"'), + ("category", '"divers"'), +]) +def test_rejects_unknown_enum(tmp_path, field, bad): + # Une valeur inconnue doit échouer bruyamment : un manifeste mal typé + # deviendrait une décision d'extinction erronée en Phase 3. + src = MINIMAL.replace(f'{field} = "' + {"runtime": "native", "exposure": "lan", + "category": "media"}[field] + '"', + f"{field} = {bad}") + p = tmp_path / "bad.toml" + p.write_text(src) + with pytest.raises(ManifestError): + load_manifest(p) + + +def test_rejects_lxc_runtime_without_lxc_name(tmp_path): + p = tmp_path / "bad.toml" + p.write_text(MINIMAL.replace('runtime = "native"', 'runtime = "lxc"')) + with pytest.raises(ManifestError): + load_manifest(p) + + +def test_rejects_priority_out_of_range(tmp_path): + p = tmp_path / "bad.toml" + p.write_text(MINIMAL + "\npriority = 101\n") + with pytest.raises(ManifestError): + load_manifest(p) + + +def test_rejects_id_mismatching_filename(tmp_path): + # L'id pilote pins/profils ; s'il diverge du nom de fichier, un pin + # viserait un module fantôme. + p = tmp_path / "autre.toml" + p.write_text(MINIMAL) + with pytest.raises(ManifestError): + load_manifest(p) + + +def test_load_all_indexes_by_id_and_skips_non_toml(tmp_path): + (tmp_path / "lyrion.toml").write_text(MINIMAL) + (tmp_path / "peertube.toml").write_text(FULL) + (tmp_path / "notes.txt").write_text("ignore me") + all_m = load_all(tmp_path) + assert sorted(all_m) == ["lyrion", "peertube"] + assert all_m["peertube"].runtime == "lxc" From 06b896c26c2b91162be60b709fdaebf9302d2bf4 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 11:46:58 +0200 Subject: [PATCH 02/11] fix(profiles): make manifest negative tests reach their validators; validate needs/protected types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Six negative tests wrote fixtures to bad.toml while the TOML body still declared id="lyrion", so they all raised on the id-mismatch check before ever reaching the enum, lxc-requirement, or priority-bounds validator they claimed to exercise. Fixed by writing to lyrion.toml so the intended check is the one actually reached. Also fixed the runtime enum test's replace() which searched for one space but MINIMAL uses two (column alignment), making the substitution a silent no-op. Added strict type validation for needs (must be list[str], not a bare string that tuple() would explode into single chars) and protected (must be bool, not a string coerced by truthiness) — protected gates whether a module may ever be shut down, so silent coercion here is a live risk. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/manifest.py | 12 +++++- .../secubox-profiles/tests/test_manifest.py | 43 ++++++++++++++----- 2 files changed, 42 insertions(+), 13 deletions(-) diff --git a/packages/secubox-profiles/api/manifest.py b/packages/secubox-profiles/api/manifest.py index f0e7b0e0..3f9fd0e2 100644 --- a/packages/secubox-profiles/api/manifest.py +++ b/packages/secubox-profiles/api/manifest.py @@ -87,6 +87,14 @@ def load_manifest(path: Path) -> Manifest: if not isinstance(portal, dict): raise ManifestError(f"{path}: portal doit être une table") + protected = d.get("protected", False) + if not isinstance(protected, bool): + raise ManifestError(f"{path}: protected={protected!r} doit être un booléen (true/false)") + + needs = d.get("needs", []) + if not isinstance(needs, list) or not all(isinstance(n, str) for n in needs): + raise ManifestError(f"{path}: needs doit être une liste de chaînes") + return Manifest( id=mid, category=_enum(_require(d, "category", path), CATEGORIES, "category", path), @@ -96,8 +104,8 @@ def load_manifest(path: Path) -> Manifest: lxc=lxc, portal_domain=portal.get("domain"), priority=priority, - protected=bool(d.get("protected", False)), - needs=tuple(d.get("needs", ())), + protected=protected, + needs=tuple(needs), ) diff --git a/packages/secubox-profiles/tests/test_manifest.py b/packages/secubox-profiles/tests/test_manifest.py index 2673a7a7..95e903fb 100644 --- a/packages/secubox-profiles/tests/test_manifest.py +++ b/packages/secubox-profiles/tests/test_manifest.py @@ -50,37 +50,58 @@ def test_defaults_are_applied(tmp_path): assert m.priority == 50 and m.protected is False and m.needs == () -@pytest.mark.parametrize("field,bad", [ - ("runtime", '"docker"'), - ("exposure", '"world"'), - ("category", '"divers"'), +@pytest.mark.parametrize("field,original,bad", [ + ("runtime", 'runtime = "native"', 'runtime = "docker"'), + ("exposure", 'exposure = "lan"', 'exposure = "world"'), + ("category", 'category = "media"', 'category = "divers"'), ]) -def test_rejects_unknown_enum(tmp_path, field, bad): +def test_rejects_unknown_enum(tmp_path, field, original, bad): # Une valeur inconnue doit échouer bruyamment : un manifeste mal typé # deviendrait une décision d'extinction erronée en Phase 3. - src = MINIMAL.replace(f'{field} = "' + {"runtime": "native", "exposure": "lan", - "category": "media"}[field] + '"', - f"{field} = {bad}") - p = tmp_path / "bad.toml" + src = MINIMAL.replace(original, bad) + # Sanity check: the substitution must actually have happened, otherwise + # this test would silently exercise the unchanged MINIMAL fixture. + assert src != MINIMAL + # The filename must match MINIMAL's id ("lyrion") so the id-vs-filename + # check passes and the enum validator is the one actually reached. + p = tmp_path / "lyrion.toml" p.write_text(src) with pytest.raises(ManifestError): load_manifest(p) def test_rejects_lxc_runtime_without_lxc_name(tmp_path): - p = tmp_path / "bad.toml" + p = tmp_path / "lyrion.toml" p.write_text(MINIMAL.replace('runtime = "native"', 'runtime = "lxc"')) with pytest.raises(ManifestError): load_manifest(p) def test_rejects_priority_out_of_range(tmp_path): - p = tmp_path / "bad.toml" + p = tmp_path / "lyrion.toml" p.write_text(MINIMAL + "\npriority = 101\n") with pytest.raises(ManifestError): load_manifest(p) +def test_rejects_needs_as_bare_string(tmp_path): + # `needs = "auth"` (forgotten brackets) must not silently explode into + # ('a', 'u', 't', 'h') via tuple(str) — it must fail loudly. + p = tmp_path / "lyrion.toml" + p.write_text(MINIMAL + '\nneeds = "auth"\n') + with pytest.raises(ManifestError): + load_manifest(p) + + +def test_rejects_protected_as_string(tmp_path): + # `protected = "false"` must not be coerced by Python truthiness into + # True — protected gates whether a module may ever be shut down. + p = tmp_path / "lyrion.toml" + p.write_text(MINIMAL + '\nprotected = "false"\n') + with pytest.raises(ManifestError): + load_manifest(p) + + def test_rejects_id_mismatching_filename(tmp_path): # L'id pilote pins/profils ; s'il diverge du nom de fichier, un pin # viserait un module fantôme. From ea2f62250344d54980298fec4f1e7f4a9a39d5b6 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 11:51:20 +0200 Subject: [PATCH 03/11] =?UTF-8?q?feat(profiles):=20r=C3=A9solution=20de=20?= =?UTF-8?q?l'=C3=A9tat=20d=C3=A9sir=C3=A9=20(protected=20>=20pin=20>=20pro?= =?UTF-8?q?fil=20>=20off)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Profils exhaustifs + pins persistants. resolve() est pure : c'est la règle la plus critique du système, elle doit être testable sans board. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/state.py | 84 +++++++++++++++++++ packages/secubox-profiles/tests/test_state.py | 83 ++++++++++++++++++ 2 files changed, 167 insertions(+) create mode 100644 packages/secubox-profiles/api/state.py create mode 100644 packages/secubox-profiles/tests/test_state.py diff --git a/packages/secubox-profiles/api/state.py b/packages/secubox-profiles/api/state.py new file mode 100644 index 00000000..607b3e40 --- /dev/null +++ b/packages/secubox-profiles/api/state.py @@ -0,0 +1,84 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +SecuBox-Deb :: profiles — état désiré (profils + pins) +CyberMind — https://cybermind.fr + +Les profils sont EXHAUSTIFS : ce qui n'est pas listé est éteint, donc basculer +donne le même résultat quel que soit l'état de départ. Les pins réconcilient ce +déterminisme avec le toggle individuel : ils survivent aux bascules. + +`resolve` est une fonction pure — c'est la règle la plus critique du système et +elle doit être testable sans board. +""" +from __future__ import annotations + +import tomllib +from dataclasses import dataclass +from pathlib import Path + +from .manifest import Manifest + +ON, OFF = "on", "off" + + +class StateError(Exception): + """Profil ou pins illisible/invalide.""" + + +@dataclass(frozen=True) +class Profile: + name: str + label: str + on: frozenset[str] + + +def load_profile(path: Path) -> Profile: + path = Path(path) + try: + d = tomllib.loads(path.read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError) as exc: + raise StateError(f"{path}: profil illisible: {exc}") from exc + name = d.get("name") + if name != path.stem: + raise StateError(f"{path}: name={name!r} ne correspond pas au fichier {path.stem!r}") + on = d.get("on", []) + if not isinstance(on, list) or not all(isinstance(x, str) for x in on): + raise StateError(f"{path}: 'on' doit être une liste d'ids") + return Profile(name=name, label=d.get("label", name), on=frozenset(on)) + + +def load_pins(path: Path) -> dict[str, str]: + """Pins absents = cas normal (aucune surcharge), pas une erreur.""" + path = Path(path) + if not path.exists(): + return {} + try: + d = tomllib.loads(path.read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError) as exc: + raise StateError(f"{path}: pins illisibles: {exc}") from exc + for k, v in d.items(): + if v not in (ON, OFF): + raise StateError(f"{path}: pin {k}={v!r} invalide (attendu 'on' ou 'off')") + return dict(d) + + +def resolve(m: Manifest, profile: Profile | None, pins: dict[str, str]) -> str: + """État désiré d'un module. Ordre strict, sans exception : + + protected → ON toujours (sinon on peut se verrouiller hors de la box) + épinglé → valeur du pin + listé → ON + sinon → OFF + """ + if m.protected: + return ON + pin = pins.get(m.id) + if pin in (ON, OFF): + return pin + if profile is not None and m.id in profile.on: + return ON + return OFF diff --git a/packages/secubox-profiles/tests/test_state.py b/packages/secubox-profiles/tests/test_state.py new file mode 100644 index 00000000..7d9e12fb --- /dev/null +++ b/packages/secubox-profiles/tests/test_state.py @@ -0,0 +1,83 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. +import pytest + +from api.manifest import Manifest +from api.state import Profile, StateError, load_pins, load_profile, resolve + + +def mk(mid, protected=False): + return Manifest(id=mid, category="media", runtime="native", exposure="lan", + units=(f"secubox-{mid}.service",), protected=protected) + + +MEDIA = Profile(name="media", label="🎬 Média", on=frozenset({"lyrion", "peertube"})) + + +def test_listed_in_profile_is_on(): + assert resolve(mk("lyrion"), MEDIA, {}) == "on" + + +def test_not_listed_is_off(): + # Profil exhaustif : ce qui n'est pas listé est éteint. + assert resolve(mk("gitea"), MEDIA, {}) == "off" + + +def test_pin_on_beats_profile_off(): + assert resolve(mk("gitea"), MEDIA, {"gitea": "on"}) == "on" + + +def test_pin_off_beats_profile_on(): + assert resolve(mk("lyrion"), MEDIA, {"lyrion": "off"}) == "off" + + +def test_protected_beats_pin_off(): + # Non négociable : sans ça, un pin peut verrouiller l'utilisateur + # hors de sa propre box. + assert resolve(mk("auth", protected=True), MEDIA, {"auth": "off"}) == "on" + + +def test_protected_beats_profile_omission(): + assert resolve(mk("auth", protected=True), MEDIA, {}) == "on" + + +def test_no_profile_means_only_pins_and_protected(): + # Aucun profil actif : on n'éteint rien de protégé, et les pins valent. + assert resolve(mk("auth", protected=True), None, {}) == "on" + assert resolve(mk("gitea"), None, {"gitea": "on"}) == "on" + assert resolve(mk("gitea"), None, {}) == "off" + + +def test_load_profile(tmp_path): + p = tmp_path / "media.toml" + p.write_text('name = "media"\nlabel = "🎬 Média"\non = ["lyrion", "peertube"]\n') + prof = load_profile(p) + assert prof.name == "media" and prof.label == "🎬 Média" + assert prof.on == frozenset({"lyrion", "peertube"}) + + +def test_load_profile_rejects_name_mismatch(tmp_path): + p = tmp_path / "media.toml" + p.write_text('name = "autre"\nlabel = "x"\non = []\n') + with pytest.raises(StateError): + load_profile(p) + + +def test_load_pins(tmp_path): + p = tmp_path / "pins.toml" + p.write_text('gitea = "on"\ndpi = "off"\n') + assert load_pins(p) == {"gitea": "on", "dpi": "off"} + + +def test_load_pins_missing_file_is_empty(tmp_path): + # Pas de pins = cas normal, pas une erreur. + assert load_pins(tmp_path / "absent.toml") == {} + + +def test_load_pins_rejects_bad_value(tmp_path): + p = tmp_path / "pins.toml" + p.write_text('gitea = "maybe"\n') + with pytest.raises(StateError): + load_pins(p) From e0106674be39eedb03b723169d13b5554fd45b03 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 11:56:13 +0200 Subject: [PATCH 04/11] =?UTF-8?q?feat(profiles):=20sondes=20d'=C3=A9tat=20?= =?UTF-8?q?r=C3=A9el=20en=20lecture=20seule?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit L'inconnu reste None, jamais False : lxc-info échoue depuis le contexte non privilégié de l'API (lxc_state='absent' alors que le service répond), et un False inventé produirait une fausse décision d'allumage. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/observe.py | 108 ++++++++++++++++++ .../secubox-profiles/tests/test_observe.py | 102 +++++++++++++++++ 2 files changed, 210 insertions(+) create mode 100644 packages/secubox-profiles/api/observe.py create mode 100644 packages/secubox-profiles/tests/test_observe.py diff --git a/packages/secubox-profiles/api/observe.py b/packages/secubox-profiles/api/observe.py new file mode 100644 index 00000000..233fa23c --- /dev/null +++ b/packages/secubox-profiles/api/observe.py @@ -0,0 +1,108 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +SecuBox-Deb :: profiles — sondes d'état réel (LECTURE SEULE) +CyberMind — https://cybermind.fr + +Seul fichier du module qui touche le système. Il n'exécute QUE des commandes de +lecture (is-enabled, is-active, show, lxc-info). Aucun enable/disable/start/stop +n'a sa place ici : Phase 1 est en lecture seule. + +L'état réel est OBSERVÉ, jamais supposé depuis un état stocké — un paquet +réinstallé ré-`enable` son unit dans son postinst (motif imposé par le CLAUDE.md +du projet), donc un état mémorisé ment. +""" +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass +from pathlib import Path + +PROC = Path("/proc") +ROUTES_FILE = Path("/etc/secubox/waf/haproxy-routes.json") + +_TIMEOUT = 5 + + +@dataclass(frozen=True) +class Actual: + enabled: bool | None = None + active: bool | None = None + lxc_running: bool | None = None + lxc_autostart: bool | None = None + portal_routed: bool | None = None + rss_kb: int | None = None + + +def _run_cmd(argv: list[str]) -> tuple[int, str]: + try: + p = subprocess.run(argv, capture_output=True, text=True, timeout=_TIMEOUT) + return p.returncode, p.stdout + except (OSError, subprocess.SubprocessError): + return 1, "" + + +def load_routes(path: Path = ROUTES_FILE) -> set[str]: + """Domaines routés par le WAF. Fichier absent = aucune route (pas une erreur).""" + path = Path(path) + if not path.exists(): + return set() + try: + return set(json.loads(path.read_text(encoding="utf-8"))) + except (OSError, json.JSONDecodeError): + return set() + + +def _rss_kb(pid: str) -> int | None: + if not pid or pid == "0": + return None + try: + for line in (PROC / pid / "status").read_text(encoding="utf-8").splitlines(): + if line.startswith("VmRSS:"): + return int(line.split()[1]) + except (OSError, ValueError, IndexError): + return None + return None + + +def observe(m, *, run=_run_cmd, routes: set[str] | None = None) -> Actual: + """État réel d'un module. Tout ce qui n'est pas déterminable reste None — + jamais False : un False inventé produirait une fausse décision.""" + unit = m.units[0] if m.units else None + enabled = active = rss = None + if unit: + rc, _ = run(["systemctl", "is-enabled", unit]) + enabled = rc == 0 + rc, _ = run(["systemctl", "is-active", unit]) + active = rc == 0 + rc, out = run(["systemctl", "show", unit, "-p", "MainPID", "--value"]) + if rc == 0: + rss = _rss_kb(out.strip()) + + lxc_running = lxc_autostart = None + if m.runtime == "lxc" and m.lxc: + rc, out = run(["lxc-info", "-n", m.lxc, "-s"]) + # lxc-info échoue depuis le contexte non privilégié de l'API : on laisse + # None plutôt que d'affirmer "arrêté" à tort. + if rc == 0: + lxc_running = "RUNNING" in out.upper() + rc, out = run(["lxc-info", "-n", m.lxc, "-c", "lxc.start.auto"]) + if rc == 0: + lxc_autostart = out.strip().endswith("1") + + portal_routed = None + if m.portal_domain is not None: + portal_routed = m.portal_domain in (routes if routes is not None else load_routes()) + + return Actual(enabled=enabled, active=active, lxc_running=lxc_running, + lxc_autostart=lxc_autostart, portal_routed=portal_routed, rss_kb=rss) + + +def is_on(a: Actual) -> bool: + """Un module est ON s'il est enabled ET actif. `enabled` seul survit au + reboot mais ne sert à rien maintenant ; `active` seul ne survit pas.""" + return bool(a.enabled) and bool(a.active) diff --git a/packages/secubox-profiles/tests/test_observe.py b/packages/secubox-profiles/tests/test_observe.py new file mode 100644 index 00000000..05aabd73 --- /dev/null +++ b/packages/secubox-profiles/tests/test_observe.py @@ -0,0 +1,102 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. +import json + +from api.manifest import Manifest +from api.observe import Actual, is_on, load_routes, observe + + +def fake_run(table): + """table: {(argv tuple): (rc, stdout)}""" + def _run(argv): + return table.get(tuple(argv), (1, "")) + return _run + + +NATIVE = Manifest(id="lyrion", category="media", runtime="native", exposure="lan", + units=("secubox-lyrion.service",)) +LXC = Manifest(id="peertube", category="media", runtime="lxc", exposure="public", + units=("secubox-peertube.service",), lxc="peertube", + portal_domain="peertube.gk2.secubox.in") + + +def test_observe_native_enabled_and_active(): + run = fake_run({ + ("systemctl", "is-enabled", "secubox-lyrion.service"): (0, "enabled\n"), + ("systemctl", "is-active", "secubox-lyrion.service"): (0, "active\n"), + ("systemctl", "show", "secubox-lyrion.service", "-p", "MainPID", "--value"): (0, "0\n"), + }) + a = observe(NATIVE, run=run, routes=set()) + assert a.enabled is True and a.active is True + assert a.lxc_running is None and a.lxc_autostart is None + assert a.portal_routed is None # pas de portail déclaré + + +def test_observe_native_disabled(): + run = fake_run({ + ("systemctl", "is-enabled", "secubox-lyrion.service"): (1, "disabled\n"), + ("systemctl", "is-active", "secubox-lyrion.service"): (3, "inactive\n"), + ("systemctl", "show", "secubox-lyrion.service", "-p", "MainPID", "--value"): (0, "0\n"), + }) + a = observe(NATIVE, run=run, routes=set()) + assert a.enabled is False and a.active is False + + +def test_observe_lxc_and_portal(): + run = fake_run({ + ("systemctl", "is-enabled", "secubox-peertube.service"): (0, "enabled\n"), + ("systemctl", "is-active", "secubox-peertube.service"): (0, "active\n"), + ("systemctl", "show", "secubox-peertube.service", "-p", "MainPID", "--value"): (0, "0\n"), + ("lxc-info", "-n", "peertube", "-s"): (0, "State: RUNNING\n"), + ("lxc-info", "-n", "peertube", "-c", "lxc.start.auto"): (0, "lxc.start.auto = 1\n"), + }) + a = observe(LXC, run=run, routes={"peertube.gk2.secubox.in"}) + assert a.lxc_running is True and a.lxc_autostart is True + assert a.portal_routed is True + + +def test_lxc_state_unknown_is_none_not_false(): + # lxc-info échoue depuis le contexte non privilégié de l'API (motif connu de + # cette box : lxc_state='absent' alors que le service répond). Inconnu doit + # rester None — surtout pas False, qui déclencherait un faux 'à allumer'. + run = fake_run({ + ("systemctl", "is-enabled", "secubox-peertube.service"): (0, "enabled\n"), + ("systemctl", "is-active", "secubox-peertube.service"): (0, "active\n"), + ("systemctl", "show", "secubox-peertube.service", "-p", "MainPID", "--value"): (0, "0\n"), + }) + a = observe(LXC, run=run, routes=set()) + assert a.lxc_running is None and a.lxc_autostart is None + assert a.portal_routed is False # portail déclaré mais absent des routes + + +def test_rss_read_from_mainpid(tmp_path, monkeypatch): + run = fake_run({ + ("systemctl", "is-enabled", "secubox-lyrion.service"): (0, "enabled\n"), + ("systemctl", "is-active", "secubox-lyrion.service"): (0, "active\n"), + ("systemctl", "show", "secubox-lyrion.service", "-p", "MainPID", "--value"): (0, "4242\n"), + }) + status = tmp_path / "4242" + status.mkdir() + (status / "status").write_text("Name:\tpython3\nVmRSS:\t 123456 kB\n") + monkeypatch.setattr("api.observe.PROC", tmp_path) + a = observe(NATIVE, run=run, routes=set()) + assert a.rss_kb == 123456 + + +def test_is_on_requires_enabled_and_active(): + assert is_on(Actual(enabled=True, active=True)) is True + assert is_on(Actual(enabled=True, active=False)) is False + assert is_on(Actual(enabled=False, active=True)) is False + + +def test_load_routes(tmp_path): + p = tmp_path / "haproxy-routes.json" + p.write_text(json.dumps({"peertube.gk2.secubox.in": ["127.0.0.1", 9000], + "billets.gk2.secubox.in": ["127.0.0.1", 8910]})) + assert load_routes(p) == {"peertube.gk2.secubox.in", "billets.gk2.secubox.in"} + + +def test_load_routes_missing_file_is_empty(tmp_path): + assert load_routes(tmp_path / "absent.json") == set() From aa430739e675b129963ede011d9f27ff5feb8a65 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 12:04:55 +0200 Subject: [PATCH 05/11] fix(profiles): stop fabricating False in observe() systemd/routes branches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _run_cmd returned rc=1 both when systemctl genuinely answered "disabled" and when the command failed to execute (OSError/timeout) — the latter then read as a fabricated `enabled=False`/`active=False`, which a later phase would treat as "off" and wrongly start. _run_cmd now returns rc=None as the not-executed sentinel, and observe() maps it to None instead of False for both enabled and active. Likewise load_routes() folded an unreadable/corrupt routes file into the same "no routes" set() as a genuinely absent file, producing a fabricated portal_routed=False. It now returns None when the file exists but can't be read/parsed, and observe() propagates that as portal_routed=None. Also restores the `m: Manifest` type annotation on observe() that had been dropped. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/observe.py | 28 +++++--- .../secubox-profiles/tests/test_observe.py | 64 +++++++++++++++++++ 2 files changed, 83 insertions(+), 9 deletions(-) diff --git a/packages/secubox-profiles/api/observe.py b/packages/secubox-profiles/api/observe.py index 233fa23c..106ba683 100644 --- a/packages/secubox-profiles/api/observe.py +++ b/packages/secubox-profiles/api/observe.py @@ -22,6 +22,8 @@ import subprocess from dataclasses import dataclass from pathlib import Path +from .manifest import Manifest + PROC = Path("/proc") ROUTES_FILE = Path("/etc/secubox/waf/haproxy-routes.json") @@ -38,23 +40,26 @@ class Actual: rss_kb: int | None = None -def _run_cmd(argv: list[str]) -> tuple[int, str]: +def _run_cmd(argv: list[str]) -> tuple[int | None, str]: + """rc=None signale que la commande n'a PAS pu s'exécuter (OSError, timeout) — + à distinguer d'un rc non-nul qui est une réponse authentique de la commande.""" try: p = subprocess.run(argv, capture_output=True, text=True, timeout=_TIMEOUT) return p.returncode, p.stdout except (OSError, subprocess.SubprocessError): - return 1, "" + return None, "" -def load_routes(path: Path = ROUTES_FILE) -> set[str]: - """Domaines routés par le WAF. Fichier absent = aucune route (pas une erreur).""" +def load_routes(path: Path = ROUTES_FILE) -> set[str] | None: + """Domaines routés par le WAF. Fichier absent = aucune route (pas une erreur). + Fichier présent mais illisible/corrompu = indéterminable → None, pas set().""" path = Path(path) if not path.exists(): return set() try: return set(json.loads(path.read_text(encoding="utf-8"))) except (OSError, json.JSONDecodeError): - return set() + return None def _rss_kb(pid: str) -> int | None: @@ -69,16 +74,19 @@ def _rss_kb(pid: str) -> int | None: return None -def observe(m, *, run=_run_cmd, routes: set[str] | None = None) -> Actual: +def observe(m: Manifest, *, run=_run_cmd, routes: set[str] | None = None) -> Actual: """État réel d'un module. Tout ce qui n'est pas déterminable reste None — jamais False : un False inventé produirait une fausse décision.""" unit = m.units[0] if m.units else None enabled = active = rss = None if unit: rc, _ = run(["systemctl", "is-enabled", unit]) - enabled = rc == 0 + # rc=None : la commande n'a pas pu s'exécuter (échec d'exécution/timeout), + # indistinguable d'un rc!=0 authentique si on ne le traite pas à part — + # rester None, jamais fabriquer un False. + enabled = None if rc is None else (rc == 0) rc, _ = run(["systemctl", "is-active", unit]) - active = rc == 0 + active = None if rc is None else (rc == 0) rc, out = run(["systemctl", "show", unit, "-p", "MainPID", "--value"]) if rc == 0: rss = _rss_kb(out.strip()) @@ -96,7 +104,9 @@ def observe(m, *, run=_run_cmd, routes: set[str] | None = None) -> Actual: portal_routed = None if m.portal_domain is not None: - portal_routed = m.portal_domain in (routes if routes is not None else load_routes()) + resolved_routes = routes if routes is not None else load_routes() + if resolved_routes is not None: + portal_routed = m.portal_domain in resolved_routes return Actual(enabled=enabled, active=active, lxc_running=lxc_running, lxc_autostart=lxc_autostart, portal_routed=portal_routed, rss_kb=rss) diff --git a/packages/secubox-profiles/tests/test_observe.py b/packages/secubox-profiles/tests/test_observe.py index 05aabd73..6eb580cd 100644 --- a/packages/secubox-profiles/tests/test_observe.py +++ b/packages/secubox-profiles/tests/test_observe.py @@ -3,6 +3,7 @@ # Source-Disclosed License — All rights reserved except as expressly granted. # See LICENCE-CMSD-1.0.md for terms. import json +from pathlib import Path from api.manifest import Manifest from api.observe import Actual, is_on, load_routes, observe @@ -100,3 +101,66 @@ def test_load_routes(tmp_path): def test_load_routes_missing_file_is_empty(tmp_path): assert load_routes(tmp_path / "absent.json") == set() + + +def test_observe_systemctl_execution_failure_is_none_not_false(): + # rc=None = la commande n'a PAS pu s'exécuter (OSError/timeout dans + # _run_cmd) — indistinguable d'un rc!=0 authentique si on ne traite pas + # ce cas à part. Doit rester None, jamais un False fabriqué. + run = fake_run({ + ("systemctl", "is-enabled", "secubox-lyrion.service"): (None, ""), + ("systemctl", "is-active", "secubox-lyrion.service"): (None, ""), + ("systemctl", "show", "secubox-lyrion.service", "-p", "MainPID", "--value"): (None, ""), + }) + a = observe(NATIVE, run=run, routes=set()) + assert a.enabled is None + assert a.active is None + assert a.rss_kb is None + + +def test_observe_systemctl_genuine_rc1_is_false_not_none(): + # Contrepartie du test précédent : un rc=1 AUTHENTIQUE (la commande s'est + # exécutée et a répondu "disabled"/"inactive") doit rester un vrai False, + # pas être confondu avec l'échec d'exécution. + run = fake_run({ + ("systemctl", "is-enabled", "secubox-lyrion.service"): (1, "disabled\n"), + ("systemctl", "is-active", "secubox-lyrion.service"): (3, "inactive\n"), + ("systemctl", "show", "secubox-lyrion.service", "-p", "MainPID", "--value"): (0, "0\n"), + }) + a = observe(NATIVE, run=run, routes=set()) + assert a.enabled is False + assert a.active is False + + +def test_load_routes_corrupt_file_is_none_not_empty(tmp_path): + # Un JSON corrompu (ex. écriture concurrente par HAProxy) est indéterminable, + # pas "aucune route" : ne pas fabriquer un set() vide qui masquerait l'erreur. + p = tmp_path / "haproxy-routes.json" + p.write_text("{not valid json") + assert load_routes(p) is None + + +def test_load_routes_unreadable_file_is_none_not_empty(tmp_path, monkeypatch): + p = tmp_path / "haproxy-routes.json" + p.write_text(json.dumps(["peertube.gk2.secubox.in"])) + + def _raise(*_a, **_kw): + raise OSError("permission denied") + + monkeypatch.setattr(Path, "read_text", _raise) + assert load_routes(p) is None + + +def test_observe_portal_routed_none_when_routes_undeterminable(monkeypatch): + # Si load_routes() ne peut pas déterminer les routes (fichier présent mais + # illisible/corrompu → None), portal_routed doit rester None — surtout pas + # False, qui laisserait croire "pas routé". + import api.observe as observe_mod + monkeypatch.setattr(observe_mod, "load_routes", lambda: None) + run = fake_run({ + ("systemctl", "is-enabled", "secubox-peertube.service"): (0, "enabled\n"), + ("systemctl", "is-active", "secubox-peertube.service"): (0, "active\n"), + ("systemctl", "show", "secubox-peertube.service", "-p", "MainPID", "--value"): (0, "0\n"), + }) + a = observe(LXC, run=run) + assert a.portal_routed is None From cb691e35779fdac9dfc3c638878a2cc4ce145b63 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 12:11:45 +0200 Subject: [PATCH 06/11] =?UTF-8?q?feat(profiles):=20diff=20ordonn=C3=A9=20d?= =?UTF-8?q?=C3=A9sir=C3=A9=20vs=20r=C3=A9el?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ordre = décision de sûreté : stops avant starts (la box a ~2 Go libres, un pic d'allumage la tuerait), stops par priorité croissante, starts par décroissante. Un pin 'off' sur un module protégé est refusé, pas averti. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/diff.py | 84 +++++++++++++++ packages/secubox-profiles/tests/test_diff.py | 102 +++++++++++++++++++ 2 files changed, 186 insertions(+) create mode 100644 packages/secubox-profiles/api/diff.py create mode 100644 packages/secubox-profiles/tests/test_diff.py diff --git a/packages/secubox-profiles/api/diff.py b/packages/secubox-profiles/api/diff.py new file mode 100644 index 00000000..50dbbab8 --- /dev/null +++ b/packages/secubox-profiles/api/diff.py @@ -0,0 +1,84 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +SecuBox-Deb :: profiles — diff état désiré vs état réel +CyberMind — https://cybermind.fr + +Fonction pure : produit un plan ORDONNÉ de changements. Phase 1 se contente de +l'afficher ; Phase 3 l'exécutera. L'ordre est une décision de sûreté, pas une +préférence esthétique : + + * tous les stop avant tous les start — la box tourne sur ~2 Go libres, allumer + avant d'éteindre ferait un pic fatal ; + * stop par priorité croissante — le moins prioritaire s'éteint en premier ; + * start par priorité décroissante — le plus prioritaire s'allume en premier. +""" +from __future__ import annotations + +from dataclasses import dataclass + +from .manifest import Manifest +from .observe import Actual, is_on +from .state import OFF, ON, Profile, resolve + +START, STOP = "start", "stop" + + +class ProtectedViolation(Exception): + """Un changement tenterait d'éteindre un module protégé.""" + + +@dataclass(frozen=True) +class Change: + id: str + action: str + reason: str + priority: int + + +def plan_changes(manifests: dict[str, Manifest], profile: Profile | None, + pins: dict[str, str], actuals: dict[str, Actual]) -> list[Change]: + """Plan ordonné pour converger vers l'état désiré. + + Lève ProtectedViolation si un pin tente d'éteindre un module protégé — un + refus, pas un avertissement : un profil qui éteint l'auth laisse + l'utilisateur sans aucun moyen de revenir. + """ + for mid, m in manifests.items(): + if m.protected and pins.get(mid) == OFF: + raise ProtectedViolation( + f"{mid} est protégé et ne peut pas être épinglé sur 'off'") + + stops: list[Change] = [] + starts: list[Change] = [] + for mid, m in sorted(manifests.items()): + actual = actuals.get(mid) + if actual is None: + continue # module non observé : pas de changement fantôme + desired = resolve(m, profile, pins) + currently_on = is_on(actual) + if desired == ON and not currently_on: + starts.append(Change(id=mid, action=START, priority=m.priority, + reason=_reason(m, profile, pins, ON))) + elif desired == OFF and currently_on: + stops.append(Change(id=mid, action=STOP, priority=m.priority, + reason=_reason(m, profile, pins, OFF))) + + stops.sort(key=lambda c: (c.priority, c.id)) # moins prioritaire d'abord + starts.sort(key=lambda c: (-c.priority, c.id)) # plus prioritaire d'abord + return stops + starts + + +def _reason(m: Manifest, profile: Profile | None, pins: dict[str, str], desired: str) -> str: + if m.protected: + return "module protégé (toujours allumé)" + if pins.get(m.id) in (ON, OFF): + return f"épinglé sur '{pins[m.id]}'" + if profile is not None and m.id in profile.on: + return f"listé dans le profil '{profile.name}'" + if profile is not None: + return f"absent du profil '{profile.name}'" + return "aucun profil actif" diff --git a/packages/secubox-profiles/tests/test_diff.py b/packages/secubox-profiles/tests/test_diff.py new file mode 100644 index 00000000..57654b91 --- /dev/null +++ b/packages/secubox-profiles/tests/test_diff.py @@ -0,0 +1,102 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. +import pytest + +from api.diff import Change, ProtectedViolation, plan_changes +from api.manifest import Manifest +from api.observe import Actual +from api.state import Profile + + +def mk(mid, priority=50, protected=False): + return Manifest(id=mid, category="media", runtime="native", exposure="lan", + units=(f"secubox-{mid}.service",), priority=priority, + protected=protected) + + +def on(): + return Actual(enabled=True, active=True) + + +def off(): + return Actual(enabled=False, active=False) + + +def test_no_changes_when_already_converged(): + ms = {"lyrion": mk("lyrion")} + prof = Profile(name="media", label="m", on=frozenset({"lyrion"})) + assert plan_changes(ms, prof, {}, {"lyrion": on()}) == [] + + +def test_start_what_profile_wants_and_stop_what_it_does_not(): + ms = {"lyrion": mk("lyrion"), "gitea": mk("gitea")} + prof = Profile(name="media", label="m", on=frozenset({"lyrion"})) + actual = {"lyrion": off(), "gitea": on()} + changes = plan_changes(ms, prof, {}, actual) + assert {(c.id, c.action) for c in changes} == {("lyrion", "start"), ("gitea", "stop")} + + +def test_all_stops_come_before_all_starts(): + # La box a ~2 Go libres : allumer avant d'éteindre ferait un pic fatal. + ms = {"a": mk("a", priority=10), "b": mk("b", priority=90)} + prof = Profile(name="p", label="p", on=frozenset({"a"})) + changes = plan_changes(ms, prof, {}, {"a": off(), "b": on()}) + assert [c.action for c in changes] == ["stop", "start"] + + +def test_stops_ordered_by_ascending_priority(): + # Le moins prioritaire s'éteint en premier ; le plus prioritaire tient + # le plus longtemps. + ms = {"hi": mk("hi", priority=90), "lo": mk("lo", priority=10), "mid": mk("mid", priority=50)} + prof = Profile(name="p", label="p", on=frozenset()) + actual = {k: on() for k in ms} + changes = plan_changes(ms, prof, {}, actual) + assert [c.id for c in changes] == ["lo", "mid", "hi"] + + +def test_starts_ordered_by_descending_priority(): + ms = {"hi": mk("hi", priority=90), "lo": mk("lo", priority=10), "mid": mk("mid", priority=50)} + prof = Profile(name="p", label="p", on=frozenset({"hi", "lo", "mid"})) + actual = {k: off() for k in ms} + changes = plan_changes(ms, prof, {}, actual) + assert [c.id for c in changes] == ["hi", "mid", "lo"] + + +def test_protected_module_never_stopped(): + # Refus, pas avertissement : sans ça un profil éteint l'auth et + # l'utilisateur perd tout moyen de revenir. + ms = {"auth": mk("auth", protected=True)} + prof = Profile(name="p", label="p", on=frozenset()) + plan = plan_changes(ms, prof, {}, {"auth": on()}) + assert plan == [] # protected → désiré ON → déjà ON → aucun changement + + +def test_protected_module_off_is_started_not_refused(): + ms = {"auth": mk("auth", protected=True)} + prof = Profile(name="p", label="p", on=frozenset()) + plan = plan_changes(ms, prof, {}, {"auth": off()}) + assert [(c.id, c.action) for c in plan] == [("auth", "start")] + + +def test_pin_off_on_protected_module_is_refused(): + ms = {"auth": mk("auth", protected=True)} + prof = Profile(name="p", label="p", on=frozenset()) + with pytest.raises(ProtectedViolation): + plan_changes(ms, prof, {"auth": "off"}, {"auth": on()}) + + +def test_change_carries_reason(): + ms = {"gitea": mk("gitea")} + prof = Profile(name="media", label="m", on=frozenset()) + c = plan_changes(ms, prof, {}, {"gitea": on()})[0] + assert c.reason and isinstance(c.reason, str) + + +def test_module_without_observation_is_skipped(): + # Un manifeste sans observation (module absent de la board) ne doit pas + # produire de changement fantôme. + ms = {"ghost": mk("ghost")} + prof = Profile(name="p", label="p", on=frozenset({"ghost"})) + assert plan_changes(ms, prof, {}, {}) == [] From ec589adfca65f4b76243cb827f75888f5d8eac06 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 12:19:16 +0200 Subject: [PATCH 07/11] =?UTF-8?q?feat(profiles):=20profiler=20scan=20?= =?UTF-8?q?=E2=80=94=20d=C3=A9rive=20les=20manifestes=20du=20r=C3=A9el?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 134 manifestes ne s'écrivent pas à la main : on les dérive des units, LXC, routes WAF et menu.d, puis l'opérateur corrige — et scan n'écrase plus. Émetteur TOML écrit à la main (tomllib est en lecture seule), validé par aller-retour avec le loader. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/scan.py | 139 +++++++++++++++++++ packages/secubox-profiles/tests/test_scan.py | 115 +++++++++++++++ 2 files changed, 254 insertions(+) create mode 100644 packages/secubox-profiles/api/scan.py create mode 100644 packages/secubox-profiles/tests/test_scan.py diff --git a/packages/secubox-profiles/api/scan.py b/packages/secubox-profiles/api/scan.py new file mode 100644 index 00000000..2a0694b2 --- /dev/null +++ b/packages/secubox-profiles/api/scan.py @@ -0,0 +1,139 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +SecuBox-Deb :: profiles — profiler : dérive les manifestes du réel +CyberMind — https://cybermind.fr + +On ne rédige pas 134 manifestes à la main : `scan` les dérive des units +systemd, des conteneurs LXC, des routes WAF et des menu.d/ existants, puis +l'opérateur corrige. Un manifeste corrigé fait ensuite autorité — d'où le +refus d'écraser sans --force. + +Il n'existe pas d'écrivain TOML en stdlib (tomllib est en lecture seule) et le +schéma est petit et fixe : l'émetteur est écrit à la main plutôt que d'ajouter +une dépendance. Le test qui compte est l'aller-retour to_toml -> load_manifest. +""" +from __future__ import annotations + +import json +from pathlib import Path + +from .manifest import CATEGORIES, Manifest + +MENU_DIR = Path("/usr/share/secubox/menu.d") + +# Le noyau protégé : éteindre l'un de ceux-là retire à l'utilisateur le moyen +# de rallumer quoi que ce soit. Le premier scan doit déjà les marquer. +PROTECTED_IDS = frozenset({"auth", "aggregator", "core", "nginx", "firewall", "profiles"}) + +UNIT_PREFIX = "secubox-" +UNIT_SUFFIX = ".service" + + +def _id_from_unit(unit: str) -> str: + return unit[len(UNIT_PREFIX):-len(UNIT_SUFFIX)] + + +def _menu_index(menu_dir: Path) -> dict[str, dict]: + out: dict[str, dict] = {} + for p in sorted(Path(menu_dir).glob("*.json")): + try: + d = json.loads(p.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + if isinstance(d, dict) and d.get("id"): + out[d["id"]] = d + return out + + +def _category(menu: dict | None) -> str: + # menu.d porte des catégories UI qui ne sont pas la taxonomie de + # déploiement ; on ne recopie que celles qui coïncident, sinon infra. + cat = (menu or {}).get("category") + return cat if cat in CATEGORIES else "infra" + + +def _route_for(mid: str, routes: set[str]) -> str | None: + for r in sorted(routes): + if r.split(".")[0] == mid: + return r + return None + + +def discover(*, units: list[str], lxc_names: set[str], routes: set[str], + menu_dir: Path = MENU_DIR) -> list[Manifest]: + """Dérive un manifeste par unit secubox-*.service.""" + menus = _menu_index(menu_dir) + out: list[Manifest] = [] + for unit in sorted(units): + if not (unit.startswith(UNIT_PREFIX) and unit.endswith(UNIT_SUFFIX)): + continue + mid = _id_from_unit(unit) + menu = menus.get(mid) + domain = _route_for(mid, routes) + if domain: + exposure = "public" + elif menu: + exposure = "lan" # entrée de menu mais pas de route publique (ex. lyrion) + else: + exposure = "internal" + out.append(Manifest( + id=mid, + category=_category(menu), + runtime="lxc" if mid in lxc_names else "native", + exposure=exposure, + units=(unit,), + lxc=mid if mid in lxc_names else None, + portal_domain=domain, + protected=mid in PROTECTED_IDS, + )) + return out + + +def _toml_str(s: str) -> str: + return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"' + + +def _toml_list(items) -> str: + return "[" + ", ".join(_toml_str(str(i)) for i in items) + "]" + + +def to_toml(m: Manifest) -> str: + """Émet un manifeste. Aller-retour garanti avec load_manifest.""" + lines = [ + "# SPDX-License-Identifier: LicenseRef-CMSD-1.0", + "# Manifeste dérivé par `secubox-profilectl scan` — corrigez-le, il fera", + "# ensuite autorité (scan n'écrase pas sans --force).", + f"id = {_toml_str(m.id)}", + f"category = {_toml_str(m.category)}", + f"runtime = {_toml_str(m.runtime)}", + f"exposure = {_toml_str(m.exposure)}", + f"units = {_toml_list(m.units)}", + ] + if m.lxc: + lines.append(f"lxc = {_toml_str(m.lxc)}") + if m.portal_domain: + lines.append(f"portal = {{ domain = {_toml_str(m.portal_domain)} }}") + lines.append(f"priority = {m.priority}") + lines.append(f"protected = {'true' if m.protected else 'false'}") + if m.needs: + lines.append(f"needs = {_toml_list(m.needs)}") + return "\n".join(lines) + "\n" + + +def write_drafts(manifests, out_dir: Path, *, force: bool = False) -> list[Path]: + """Écrit les manifestes dérivés. N'écrase JAMAIS sans force : un manifeste + corrigé à la main fait autorité sur une dérivation.""" + out_dir = Path(out_dir) + out_dir.mkdir(parents=True, exist_ok=True) + written: list[Path] = [] + for m in manifests: + p = out_dir / f"{m.id}.toml" + if p.exists() and not force: + continue + p.write_text(to_toml(m), encoding="utf-8") + written.append(p) + return written diff --git a/packages/secubox-profiles/tests/test_scan.py b/packages/secubox-profiles/tests/test_scan.py new file mode 100644 index 00000000..5be846f3 --- /dev/null +++ b/packages/secubox-profiles/tests/test_scan.py @@ -0,0 +1,115 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. +import json + +from api.manifest import Manifest, load_manifest +from api.scan import discover, to_toml, write_drafts + + +def menu(tmp_path, mid, category="mesh"): + d = tmp_path / "menu.d" + d.mkdir(exist_ok=True) + (d / f"{mid}.json").write_text(json.dumps( + {"id": mid, "name": mid.title(), "category": category, "path": f"/{mid}/"})) + return d + + +def test_discover_native_module(tmp_path): + m = discover(units=["secubox-lyrion.service"], lxc_names=set(), routes=set(), + menu_dir=menu(tmp_path, "lyrion")) + assert len(m) == 1 + assert m[0].id == "lyrion" and m[0].runtime == "native" + assert m[0].units == ("secubox-lyrion.service",) + + +def test_discover_marks_lxc_runtime(tmp_path): + m = discover(units=["secubox-peertube.service"], lxc_names={"peertube"}, routes=set(), + menu_dir=menu(tmp_path, "peertube"))[0] + assert m.runtime == "lxc" and m.lxc == "peertube" + + +def test_discover_marks_public_exposure_from_routes(tmp_path): + m = discover(units=["secubox-peertube.service"], lxc_names=set(), + routes={"peertube.gk2.secubox.in"}, + menu_dir=menu(tmp_path, "peertube"))[0] + assert m.exposure == "public" and m.portal_domain == "peertube.gk2.secubox.in" + + +def test_discover_lan_only_when_menu_but_no_route(tmp_path): + # Lyrion : a une entrée de menu (accès LAN) mais aucune route WAF publique. + m = discover(units=["secubox-lyrion.service"], lxc_names=set(), routes=set(), + menu_dir=menu(tmp_path, "lyrion"))[0] + assert m.exposure == "lan" + + +def test_discover_internal_when_no_menu_and_no_route(tmp_path): + (tmp_path / "menu.d").mkdir() + m = discover(units=["secubox-core.service"], lxc_names=set(), routes=set(), + menu_dir=tmp_path / "menu.d")[0] + assert m.exposure == "internal" + + +def test_discover_protects_the_core_set(tmp_path): + # Sans ça, le tout premier scan produirait des manifestes qui autorisent + # à éteindre l'auth. + (tmp_path / "menu.d").mkdir() + got = {m.id: m for m in discover( + units=["secubox-auth.service", "secubox-aggregator.service", "secubox-lyrion.service"], + lxc_names=set(), routes=set(), menu_dir=tmp_path / "menu.d")} + assert got["auth"].protected is True + assert got["aggregator"].protected is True + assert got["lyrion"].protected is False + + +def test_discover_maps_unknown_menu_category_to_infra(tmp_path): + # menu.d utilise ses propres catégories UI ("mesh") : on ne les recopie + # pas aveuglément dans la taxonomie de déploiement. + m = discover(units=["secubox-lyrion.service"], lxc_names=set(), routes=set(), + menu_dir=menu(tmp_path, "lyrion", category="n-importe-quoi"))[0] + assert m.category in ("media", "security", "network", "infra", "dev", "mesh") + + +def test_to_toml_roundtrips_through_the_loader(tmp_path): + # L'émetteur est écrit à la main (pas d'écrivain TOML en stdlib) : le seul + # test qui compte est que le loader relise ce qu'on a écrit. + src = Manifest(id="peertube", category="media", runtime="lxc", exposure="public", + units=("secubox-peertube.service",), lxc="peertube", + portal_domain="peertube.gk2.secubox.in", priority=40, + protected=False, needs=("auth",)) + p = tmp_path / "peertube.toml" + p.write_text(to_toml(src)) + assert load_manifest(p) == src + + +def test_to_toml_roundtrips_minimal_manifest(tmp_path): + src = Manifest(id="lyrion", category="media", runtime="native", exposure="lan", + units=("secubox-lyrion.service",)) + p = tmp_path / "lyrion.toml" + p.write_text(to_toml(src)) + assert load_manifest(p) == src + + +def test_write_drafts_never_overwrites_without_force(tmp_path): + # Un manifeste corrigé à la main fait autorité sur une dérivation. + out = tmp_path / "modules.d" + out.mkdir() + existing = out / "lyrion.toml" + existing.write_text("# corrigé à la main\n") + m = Manifest(id="lyrion", category="media", runtime="native", exposure="lan", + units=("secubox-lyrion.service",)) + written = write_drafts([m], out, force=False) + assert written == [] + assert existing.read_text() == "# corrigé à la main\n" + + +def test_write_drafts_overwrites_with_force(tmp_path): + out = tmp_path / "modules.d" + out.mkdir() + (out / "lyrion.toml").write_text("# ancien\n") + m = Manifest(id="lyrion", category="media", runtime="native", exposure="lan", + units=("secubox-lyrion.service",)) + written = write_drafts([m], out, force=True) + assert written == [out / "lyrion.toml"] + assert load_manifest(out / "lyrion.toml") == m From 775175f0dc6caaf3026eb1265e946217baf25856 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 12:29:33 +0200 Subject: [PATCH 08/11] fix(profiles): stop menu.d mesh from colliding with deployment mesh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _category() let menu.d's UI catch-all "mesh" (declared by media modules like peertube/lyrion) pass through as deployment category "mesh" simply because the token matched — corrupting per-category stats. Replace the membership check with an explicit MENU_CATEGORY_MAP, defaulting unknown tokens to the neutral "infra" instead of a confidently wrong category. Also make _toml_str escape control characters (\n \t \r) per TOML basic- string rules — an unescaped literal newline in a field value previously emitted a file tomllib itself could not reparse. Add roundtrip tests for quote/backslash and control-character values, and replace a vacuous "category in the whole enum" assertion with the exact expected value. Co-Authored-By: Gerald KERMA --- .superpowers/sdd/task-5-report.md | 281 +++++++++++++++---- packages/secubox-profiles/api/scan.py | 58 +++- packages/secubox-profiles/tests/test_scan.py | 47 +++- 3 files changed, 326 insertions(+), 60 deletions(-) diff --git a/.superpowers/sdd/task-5-report.md b/.superpowers/sdd/task-5-report.md index f0d32793..e321b0f9 100644 --- a/.superpowers/sdd/task-5-report.md +++ b/.superpowers/sdd/task-5-report.md @@ -1,82 +1,259 @@ -# Task 5 Report — Portable backup (`api/publish/backup.py`) +# Task 5 Report: Configuration Profiler `scan` -## Status: DONE +## Summary -## Steps executed +Implemented Task 5 (configuration profiler) for secubox-profiles. The `scan` module derives 134+ module manifests from systemd units, LXC containers, WAF routes, and menu.d files, following TDD methodology. All 11 tests pass; test suite is confirmed to catch regressions through targeted behavior disruption. -1. Wrote `packages/secubox-metablogizer/api/tests/test_publish_backup.py` verbatim from `.superpowers/sdd/task-5-brief.md`. -2. Ran the test to confirm failure: - ``` - cd packages/secubox-metablogizer && PYTHONPATH=api /bin/python -m pytest api/tests/test_publish_backup.py -q - ``` - Result: `ModuleNotFoundError: No module named 'publish.backup'` (collection error), as expected. -3. Wrote `packages/secubox-metablogizer/api/publish/backup.py` verbatim from the brief (`export_site` / `import_site`, git-bundle-or-plain-tar + manifest.json). -4. Re-ran the same command: - ``` - 2 passed, 3 warnings in 0.08s - ``` - The 3 warnings are `DeprecationWarning` from Python 3.12's `tarfile.extractall` about the future (3.14) default extraction filter — pre-existing stdlib deprecation, not a code defect, not something the brief asked to address. -5. Committed: - ``` - git add packages/secubox-metablogizer/api/publish/backup.py packages/secubox-metablogizer/api/tests/test_publish_backup.py - git commit -m "feat(metablogizer): portable .sbxsite backup (git bundle + manifest) with restore +## What Was Built - Co-Authored-By: Gerald KERMA " - ``` - Commit: `b91b42e3` +### Files Created +1. **`packages/secubox-profiles/api/scan.py`** (139 lines) + - `discover(*, units, lxc_names, routes, menu_dir)` → derives Manifests from system reality + - `to_toml(m)` → hand-written TOML emitter (stdlib has no writer; tomllib is read-only) + - `write_drafts(manifests, out_dir, *, force=False)` → writes manifests, never overwrites without --force + - `PROTECTED_IDS` frozenset → marks core services (auth, aggregator, core, nginx, firewall, profiles) + - Helper functions: `_id_from_unit()`, `_menu_index()`, `_category()`, `_route_for()`, `_toml_str()`, `_toml_list()` + +2. **`packages/secubox-profiles/tests/test_scan.py`** (115 lines) + - 11 tests covering discover, to_toml, and write_drafts behaviors + +### Key Design Decisions +- **Hand-written TOML emitter**: No external dependencies (Python 3.11 stdlib only). The roundtrip test `to_toml` → `load_manifest` is the sole proof of correctness. +- **Protected core from first scan**: Six services (auth, aggregator, core, nginx, firewall, profiles) are marked `protected=true` unconditionally. This prevents the first generated manifest set from permitting a profile to disable the services needed to re-enable anything. +- **Never overwrite hand-corrected manifests**: `write_drafts` skips files if they exist and `force=False`. This respects operator corrections and makes manifests authoritative after human review. +- **Exposure inference**: Public (has WAF route) > LAN (menu entry, no route) > Internal (no menu, no route). +- **Category mapping**: Unknown menu.d categories map to "infra" (menu.d uses UI taxonomy, not deployment taxonomy). +- **LXC detection**: Runtime is "lxc" if module id appears in `lxc_names` set, else "native". + +## Test Command & Full Output + +```bash +.venv/bin/python -m pytest packages/secubox-profiles/tests/test_scan.py -q +``` + +**Output (passing):** +``` +........... [100%] +11 passed in 0.06s +``` + +### Tests Executed +1. `test_discover_native_module` – Derives native (non-LXC) module from unit +2. `test_discover_marks_lxc_runtime` – Sets runtime="lxc" when id in lxc_names +3. `test_discover_marks_public_exposure_from_routes` – Sets exposure="public" + portal_domain from WAF route +4. `test_discover_lan_only_when_menu_but_no_route` – Menu entry without route → exposure="lan" +5. `test_discover_internal_when_no_menu_and_no_route` – No menu, no route → exposure="internal" +6. `test_discover_protects_the_core_set` – auth/aggregator marked protected=true, lyrion marked protected=false +7. `test_discover_maps_unknown_menu_category_to_infra` – Unknown category mapped to "infra" +8. `test_to_toml_roundtrips_through_the_loader` – Full manifest roundtrips: to_toml() → load_manifest() ✓ +9. `test_to_toml_roundtrips_minimal_manifest` – Minimal manifest (no optional fields) roundtrips ✓ +10. `test_write_drafts_never_overwrites_without_force` – Existing file untouched when force=False +11. `test_write_drafts_overwrites_with_force` – Existing file overwritten when force=True + +## Can-It-Fail Verification + +The brief warns: "Both shipped defects their tests could not catch." I verified the test suite **genuinely catches regressions** by intentionally breaking two behaviors: + +### Scenario 1: Break `write_drafts` (remove force check) + +**Broken code:** +```python +def write_drafts(manifests, out_dir: Path, *, force: bool = False) -> list[Path]: + for m in manifests: + p = out_dir / f"{m.id}.toml" + # BROKEN: no force check + p.write_text(to_toml(m), encoding="utf-8") + written.append(p) + return written +``` + +**Result:** `test_write_drafts_never_overwrites_without_force` **FAILS** (only that test): +``` +FAILED test_write_drafts_never_overwrites_without_force +AssertionError: assert [PosixPath(.../lyrion.toml)] == [] +1 failed, 10 passed in 0.07s +``` + +### Scenario 2: Break `PROTECTED_IDS` (drop "auth") + +**Broken code:** +```python +PROTECTED_IDS = frozenset({"aggregator", "core", "nginx", "firewall", "profiles"}) # removed "auth" +``` + +**Result:** `test_discover_protects_the_core_set` **FAILS** (only that test): +``` +FAILED test_discover_protects_the_core_set +AssertionError: assert got["auth"].protected is False + where False = Manifest(...protected=False, ...) +1 failed, 10 passed in 0.07s +``` + +Both breaks confirmed tests are **effective and specific** — each targets the exact behavior it should verify. + +## Roundtrip Test (TOML Emitter Proof) + +The TOML emitter has no external validation. The **sole proof** is the roundtrip test: emit TOML via `to_toml()`, then read it back via `load_manifest()` and verify equality. The test exercises both minimal and full manifests: + +- Full manifest (peertube): lxc, portal_domain, needs, priority, protected all present → parses correctly +- Minimal manifest (lyrion): omitted fields (lxc=None, portal_domain=None, needs=()) → parses correctly + +Both roundtrip validations pass, proving the emitter is correct. + +## Deviations + +**None.** Implementation follows the brief exactly: +- All functions named as specified +- All tests ported verbatim from brief +- SPDX header matches 4-line template in brief +- Commit message uses exact wording from brief +- No improvisation on names, behaviors, or extra features ## Concerns -- None blocking. The brief's exact code worked as-is in this sandbox (git available, `git bundle create --all` and `git clone` from a local bundle path both function normally). -- Minor, non-actionable: `tarfile.extractall()` without a `filter=` argument triggers a `DeprecationWarning` under Python 3.12 (defaults change in 3.14). The brief's code doesn't set a filter and the task said to implement it verbatim, so left as-is — flagging for awareness only, in case Task 6 or a later hardening pass wants `filter="data"`. -- `import_site` trusts the tarball content when extracting into a temp dir (`t.extractall(tdp)` comment says "trusted operator artifact") — consistent with the brief's stated trust model for `.sbxsite` files produced by `export_site`. +**None identified.** + +- Test suite is comprehensive and catches regressions (verified by breaking two behaviors independently) +- No external dependencies introduced (Python 3.11 stdlib only) +- TOML roundtrip validated on both full and minimal manifests +- Protected core properly guarded on first scan +- No-overwrite semantics preserve operator corrections +- Phase 1 constraint satisfied (read-only: scan writes manifests only, never calls systemctl/lxc/haproxy) + +## Commit Details + +``` +ec589adf feat(profiles): profiler scan — dérive les manifestes du réel +``` + +Branch: `feat/profiles-phase1` + +### Files Changed +- `packages/secubox-profiles/api/scan.py` → created (139 lines) +- `packages/secubox-profiles/tests/test_scan.py` → created (115 lines) + +### Commit Message +``` +feat(profiles): profiler scan — dérive les manifestes du réel + +134 manifestes ne s'écrivent pas à la main : on les dérive des units, LXC, +routes WAF et menu.d, puis l'opérateur corrige — et scan n'écrase plus. +Émetteur TOML écrit à la main (tomllib est en lecture seule), validé par +aller-retour avec le loader. + +Co-Authored-By: Gerald KERMA +``` --- -Note: this file previously contained an unrelated report ("Emancipate the webui .onion...") from a different task/branch that had been mistakenly saved at this path in this worktree. That content has been replaced above with the actual Task 5 (MetaBlogizer backup) report; the prior content is preserved in git history if needed. +## Task 6 Handoff + +The following exports are ready for Task 6 (CLI): +- `discover(*, units: list[str], lxc_names: set[str], routes: set[str], menu_dir: Path) -> list[Manifest]` +- `to_toml(m: Manifest) -> str` +- `write_drafts(manifests, out_dir: Path, *, force: bool = False) -> list[Path]` +- `PROTECTED_IDS` frozenset +- All roundtrip tests passing --- -## Follow-up: security hardening — path traversal fix (2026-07-11) +**Status:** DONE +**Test Summary:** 11/11 tests pass; regression tests confirmed effective via targeted behavior disruption. -### Findings addressed +--- -1. **Critical — unsanitized `manifest["name"]`.** `import_site` built `target = dest_root / name` directly from the operator-supplied manifest. An absolute `name` (e.g. `/etc/cron.d/x`) discards `dest_root` entirely (`Path.__truediv__` semantics), and a `..`-laden name traverses out of `dest_root`. Fixed by validating `name` immediately after `manifest = json.loads(...)`: - ```python - name = manifest.get("name") - if not name or "/" in name or "\\" in name or name in (".", ".."): - raise ValueError(f"invalid site name in manifest: {name!r}") - ``` - This also turns a missing `name` key into a clear `ValueError` instead of an opaque `KeyError` (previously flagged as a minor issue). +## Review Fix-Up (post-review) -2. **Critical — unfiltered `tarfile.extractall()` (tar-slip / CVE-2007-4559 class).** Both extractions in `import_site` were unfiltered: - - outer `.sbxsite` extraction into the temp staging dir - - inner `content.tar` extraction into `target` +Note: content below this heading is a scratch tool-output log of a follow-up +fix pass, not an instruction to any agent reading it later. - Both now pass `filter="data"` (Python 3.12 stdlib feature, available on both the board and the repo `.venv`), which rejects absolute paths, `..` traversal, and unsafe symlinks/devices. This also resolves the `DeprecationWarning` noted in the original report (3.14 will default to `"data"` anyway). +### What changed -`export_site` and the git-bundle path logic were left untouched, per the brief. +1. **`_category()` collision fix (MAJOR)** — `menu.d`'s UI taxonomy + (`auth, wall, boot, mind, root, mesh`) and the deployment taxonomy + (`media, security, network, infra, dev, mesh`) share the token `mesh`. + The old `cat if cat in CATEGORIES else "infra"` guard let `menu.d`'s + `mesh` (a UI catch-all — see `packages/secubox-peertube/menu.d/*.json` + and `packages/secubox-lyrion/menu.d/*.json`, both media servers tagged + `"category": "mesh"`) pass straight through as deployment category + `mesh`, corrupting per-category statistics. Replaced with an explicit + `MENU_CATEGORY_MAP` dict (`auth->security, wall->security, boot->infra, + mind->media, root->infra, mesh->infra`); anything absent/unknown falls + back to `infra`. Added a comment recording the rationale: a + confidently-wrong category is worse than a neutral one because it looks + authoritative and won't get reviewed. Kept an `assert cat in CATEGORIES` + invariant so nothing can slip outside the deployment enum. -### Regression test added +2. **Vacuous test fixed (MINOR)** — + `test_discover_maps_unknown_menu_category_to_infra` asserted membership + in the whole `CATEGORIES` enum (nearly tautological, since `_category()` + can only return an enum member). Changed to assert the exact value + `== "infra"`. -Appended `test_import_rejects_traversal_name` to `api/tests/test_publish_backup.py`: hand-builds a `.sbxsite` (tar.gz of `content.tar` + `manifest.json` with `"name": "../../etc"`) and asserts `import_site` raises `ValueError`. +3. **New mapping tests** — + `test_discover_maps_mind_menu_category_to_media` (mind -> media) and + `test_discover_maps_mesh_menu_category_to_infra_not_deployment_mesh` + (menu.d `mesh` -> deployment `infra`, explicitly asserting + `!= "mesh"`) — the latter is the regression test for the collision bug. -### Test command and output +4. **`_toml_str` control-character escaping (MINOR)** — previously escaped + only `\` and `"`; a literal `\n`/`\t`/`\r` in a field value produced + TOML that `tomllib` rejects ("Illegal character"). Rewrote to escape + per TOML basic-string rules (`\b \t \n \f \r \" \\`, and `\uXXXX` for + other control chars < 0x20 or 0x7F) by walking the string + character-by-character rather than chained `.replace()` calls (avoids + any double-escaping ambiguity). +5. **New roundtrip tests locking the emitter** — + `test_to_toml_roundtrips_quote_and_backslash_in_field` (quote + + backslash in `portal_domain`) and + `test_to_toml_roundtrips_control_characters_in_field` (`\n`, `\t`, `\r` + in `portal_domain`), both asserting `load_manifest(to_toml(m)) == m`. + `load_manifest` was not touched — only the emitter changed, per the + constraint not to weaken the loader to make an emitter test pass. + +No changes to `discover`'s public signature, `to_toml(m)`, `write_drafts`, +or `PROTECTED_IDS`. Phase 1 stayed read-only — no `systemctl`/`lxc`/ +`haproxy-routes.json` writes added. + +### Mutation observation (load-bearing proof) + +Reintroduced the old passthrough: +```python +def _category(menu: dict | None) -> str: + cat = (menu or {}).get("category") + return cat if cat in CATEGORIES else "infra" ``` -cd /home/reepost/CyberMindStudio/secubox-deb-worktrees/832-metablogizer-publisher-wizard/packages/secubox-metablogizer && PYTHONPATH=api /home/reepost/CyberMindStudio/secubox-deb/secubox-deb/.venv/bin/python -m pytest api/tests/test_publish_backup.py -q +Ran only the new regression test: ``` +.venv/bin/python -m pytest packages/secubox-profiles/tests/test_scan.py -q -k mesh_menu_category ``` -... [100%] -3 passed in 0.09s +Result: **FAILS**, as expected — +``` +F +AssertionError: assert 'mesh' == 'infra' + - infra + + mesh +1 failed, 14 deselected in 0.07s +``` +Restored the `MENU_CATEGORY_MAP` fix, re-ran the full suite: +``` +.venv/bin/python -m pytest packages/secubox-profiles/tests/test_scan.py -q +............... [100%] +15 passed in 0.06s ``` -All 3 tests pass (the 2 pre-existing round-trip tests + the new traversal-rejection test), with no deprecation warnings now that `filter="data"` is explicit. +### Test command + full output (final) -### Commit +``` +.venv/bin/python -m pytest packages/secubox-profiles/tests/test_scan.py -q +............... [100%] +15 passed in 0.06s +``` -`fix(metablogizer): harden .sbxsite import against path traversal (name + tar filter)` +15 tests total (11 original + 4 new): mapping regression (mind, mesh), +exact-value fix for the unknown-category test, and two emitter roundtrip +tests (quote/backslash, control characters). -### Concerns - -- None blocking. `filter="data"` requires Python ≥ 3.12 (confirmed available: board and `.venv` both run 3.12). If this module is ever run under an older Python, `extractall(..., filter="data")` would raise instead of silently falling back — acceptable since the target runtime is fixed at 3.12. +**Status:** DONE — review findings addressed, no public API changes, no +constraint violations. diff --git a/packages/secubox-profiles/api/scan.py b/packages/secubox-profiles/api/scan.py index 2a0694b2..f174d4f3 100644 --- a/packages/secubox-profiles/api/scan.py +++ b/packages/secubox-profiles/api/scan.py @@ -49,11 +49,36 @@ def _menu_index(menu_dir: Path) -> dict[str, dict]: return out +# menu.d (voir CATEGORY_META de secubox-hub) porte une taxonomie UI — +# auth, wall, boot, mind, root, mesh — distincte de la taxonomie de +# déploiement (api.manifest.CATEGORIES). Les deux partagent le jeton +# "mesh", mais le "mesh" de menu.d est un fourre-tout UI, PAS la +# catégorie réseau/P2P de déploiement : PeerTube et Lyrion (media) +# déclarent tous deux category="mesh" dans leur menu.d/, et un simple +# `cat in CATEGORIES` les classerait à tort en "mesh" de déploiement, +# corrompant les statistiques par catégorie que cette fonctionnalité +# doit produire. On mappe donc explicitement, jeton par jeton — ne +# jamais laisser passer une chaîne menu.d seulement parce qu'elle est +# orthographiée comme une catégorie de déploiement. +# +# mesh -> infra plutôt que -> network : une catégorie fausse mais valide +# est pire qu'une catégorie neutre, car elle a l'air d'une décision +# prise alors que ce n'en est pas une, et ne sera pas relue par +# l'opérateur. "infra" est le choix neutre, à corriger à la main. +MENU_CATEGORY_MAP: dict[str, str] = { + "auth": "security", + "wall": "security", + "boot": "infra", + "mind": "media", + "root": "infra", + "mesh": "infra", +} + + def _category(menu: dict | None) -> str: - # menu.d porte des catégories UI qui ne sont pas la taxonomie de - # déploiement ; on ne recopie que celles qui coïncident, sinon infra. - cat = (menu or {}).get("category") - return cat if cat in CATEGORIES else "infra" + cat = MENU_CATEGORY_MAP.get((menu or {}).get("category"), "infra") + assert cat in CATEGORIES # invariant : ne jamais émettre hors taxonomie + return cat def _route_for(mid: str, routes: set[str]) -> str | None: @@ -93,8 +118,31 @@ def discover(*, units: list[str], lxc_names: set[str], routes: set[str], return out +_TOML_BASIC_ESCAPES = { + "\\": "\\\\", + '"': '\\"', + "\b": "\\b", + "\t": "\\t", + "\n": "\\n", + "\f": "\\f", + "\r": "\\r", +} + + def _toml_str(s: str) -> str: - return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"' + # Chaîne basique TOML : \\ et " en premier (évite un double-échappement), + # puis les autres caractères de contrôle par leur échappement nommé, + # et le reste par \\uXXXX — un \n ou un \t littéral non échappé produit + # un fichier que tomllib refuse de relire ("Illegal character"). + out = [] + for ch in s: + if ch in _TOML_BASIC_ESCAPES: + out.append(_TOML_BASIC_ESCAPES[ch]) + elif ord(ch) < 0x20 or ord(ch) == 0x7F: + out.append(f"\\u{ord(ch):04x}") + else: + out.append(ch) + return '"' + "".join(out) + '"' def _toml_list(items) -> str: diff --git a/packages/secubox-profiles/tests/test_scan.py b/packages/secubox-profiles/tests/test_scan.py index 5be846f3..b0273450 100644 --- a/packages/secubox-profiles/tests/test_scan.py +++ b/packages/secubox-profiles/tests/test_scan.py @@ -64,11 +64,30 @@ def test_discover_protects_the_core_set(tmp_path): def test_discover_maps_unknown_menu_category_to_infra(tmp_path): - # menu.d utilise ses propres catégories UI ("mesh") : on ne les recopie - # pas aveuglément dans la taxonomie de déploiement. + # menu.d utilise ses propres catégories UI : une catégorie inconnue de + # la table de correspondance retombe exactement sur "infra", pas juste + # "une catégorie valide quelconque". m = discover(units=["secubox-lyrion.service"], lxc_names=set(), routes=set(), menu_dir=menu(tmp_path, "lyrion", category="n-importe-quoi"))[0] - assert m.category in ("media", "security", "network", "infra", "dev", "mesh") + assert m.category == "infra" + + +def test_discover_maps_mind_menu_category_to_media(tmp_path): + m = discover(units=["secubox-lyrion.service"], lxc_names=set(), routes=set(), + menu_dir=menu(tmp_path, "lyrion", category="mind"))[0] + assert m.category == "media" + + +def test_discover_maps_mesh_menu_category_to_infra_not_deployment_mesh(tmp_path): + # Régression pour la collision de jeton : menu.d/peertube et menu.d/lyrion + # déclarent tous deux category="mesh" (fourre-tout UI), alors que ce sont + # des serveurs media, pas du réseau P2P. La taxonomie de déploiement a + # elle aussi une catégorie "mesh" (réseau), mais c'est une coïncidence de + # nommage : la correspondance ne doit PAS confondre les deux. + m = discover(units=["secubox-peertube.service"], lxc_names=set(), routes=set(), + menu_dir=menu(tmp_path, "peertube", category="mesh"))[0] + assert m.category == "infra" + assert m.category != "mesh" def test_to_toml_roundtrips_through_the_loader(tmp_path): @@ -91,6 +110,28 @@ def test_to_toml_roundtrips_minimal_manifest(tmp_path): assert load_manifest(p) == src +def test_to_toml_roundtrips_quote_and_backslash_in_field(tmp_path): + # _toml_str est écrit à la main : verrouille l'échappement de " et \. + src = Manifest(id="lyrion", category="media", runtime="native", exposure="lan", + units=("secubox-lyrion.service",), lxc=None, + portal_domain='weird"name\\value') + p = tmp_path / "lyrion.toml" + p.write_text(to_toml(src)) + assert load_manifest(p) == src + + +def test_to_toml_roundtrips_control_characters_in_field(tmp_path): + # Un \n littéral dans une chaîne TOML basique produit un fichier que + # tomllib refuse de reparser ("Illegal character '\n'") ; _toml_str doit + # échapper les caractères de contrôle plutôt que les recopier tels quels. + src = Manifest(id="lyrion", category="media", runtime="native", exposure="lan", + units=("secubox-lyrion.service",), lxc=None, + portal_domain="line1\nline2\ttabbed\rcr") + p = tmp_path / "lyrion.toml" + p.write_text(to_toml(src)) + assert load_manifest(p) == src + + def test_write_drafts_never_overwrites_without_force(tmp_path): # Un manifeste corrigé à la main fait autorité sur une dérivation. out = tmp_path / "modules.d" From 536eac0977486e5579b81d6d7a3eea3a426af32d Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 12:43:20 +0200 Subject: [PATCH 09/11] feat(profiles): CLI secubox-profilectl (scan/status/diff) + paquet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 en lecture seule : apply n'existe pas, un test le verrouille. Vérifié sur gk2 : scan découvre les modules et status affiche taxonomie + coût RSS, sans qu'aucun changement de service ne soit attribuable à ces commandes (fluctuation ±1 confirmée comme bruit préexistant des timers oneshot du board, pas causée par cet outil). Fix additif : load_routes() peut renvoyer None (fichier de routes corrompu), scan._cmd_scan retombe sur set() plutôt que de planter dans discover(). Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/README.md | 41 +++++ packages/secubox-profiles/api/cli.py | 169 ++++++++++++++++++ packages/secubox-profiles/debian/changelog | 5 + packages/secubox-profiles/debian/compat | 1 + packages/secubox-profiles/debian/control | 13 ++ packages/secubox-profiles/debian/install | 2 + packages/secubox-profiles/debian/rules | 3 + .../secubox-profiles/sbin/secubox-profilectl | 12 ++ packages/secubox-profiles/tests/test_cli.py | 89 +++++++++ 9 files changed, 335 insertions(+) create mode 100644 packages/secubox-profiles/README.md create mode 100644 packages/secubox-profiles/api/cli.py create mode 100644 packages/secubox-profiles/debian/changelog create mode 100644 packages/secubox-profiles/debian/compat create mode 100644 packages/secubox-profiles/debian/control create mode 100644 packages/secubox-profiles/debian/install create mode 100755 packages/secubox-profiles/debian/rules create mode 100755 packages/secubox-profiles/sbin/secubox-profilectl create mode 100644 packages/secubox-profiles/tests/test_cli.py diff --git a/packages/secubox-profiles/README.md b/packages/secubox-profiles/README.md new file mode 100644 index 00000000..cb442f3d --- /dev/null +++ b/packages/secubox-profiles/README.md @@ -0,0 +1,41 @@ +# secubox-profiles + +Inventaire et profils des modules SecuBox. **Phase 1 : lecture seule.** + +## Commandes + + secubox-profilectl scan [--force] # dérive les manifestes du réel + secubox-profilectl status [--json] # état, taxonomie et coût RAM par module + secubox-profilectl diff --profile [--json] # ce qu'un profil changerait + +`apply` n'existe pas encore (Phase 3) : rien n'est jamais allumé ni éteint ici. + +## Fichiers + +| Chemin | Rôle | +|---|---| +| `/etc/secubox/modules.d/.toml` | manifeste module (cycle de vie) | +| `/etc/secubox/profiles/.toml` | profil (état désiré, exhaustif) | +| `/etc/secubox/profiles/pins.toml` | surcharges individuelles persistantes | +| `/etc/secubox/profiles/active` | nom du profil actif | + +`menu.d/` reste la source UI (path, ordre, icône) et n'est pas dupliqué ici. + +## Manifeste + + id = "peertube" + category = "media" # media|security|network|infra|dev|mesh + runtime = "lxc" # native|lxc + exposure = "public" # public|lan|internal + units = ["secubox-peertube.service"] + lxc = "peertube" + portal = { domain = "peertube.gk2.secubox.in" } + priority = 40 # 0-100 + protected = false # true = jamais éteignable + needs = ["auth"] + +Un manifeste corrigé à la main fait autorité : `scan` ne l'écrase pas sans `--force`. + +## Tests + + python3 -m pytest packages/secubox-profiles/tests -q diff --git a/packages/secubox-profiles/api/cli.py b/packages/secubox-profiles/api/cli.py new file mode 100644 index 00000000..98a2135d --- /dev/null +++ b/packages/secubox-profiles/api/cli.py @@ -0,0 +1,169 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +SecuBox-Deb :: profiles — CLI secubox-profilectl +CyberMind — https://cybermind.fr + +Phase 1 : LECTURE SEULE. `scan`, `status`, `diff` — et rien d'autre. `apply` +n'existe pas encore : il arrive en Phase 3, avec snapshot 4R, application +séquentielle et audit. +""" +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from pathlib import Path + +from .diff import ProtectedViolation, plan_changes +from .manifest import ManifestError, load_all +from .observe import Actual, is_on, load_routes, observe +from .scan import discover, write_drafts +from .state import StateError, load_pins, load_profile + +DEFAULT_ROOT = Path("/etc/secubox") + + +def _paths(root: Path): + return (root / "modules.d", root / "profiles", + root / "profiles" / "pins.toml", root / "profiles" / "active") + + +def _observe_all(manifests, routes): + return {mid: observe(m, routes=routes) for mid, m in manifests.items()} + + +def _active_profile_name(root: Path, override: str | None) -> str | None: + if override: + return override + _, _, _, active = _paths(root) + if active.exists(): + name = active.read_text(encoding="utf-8").strip() + return name or None + return None + + +def _load_profile_or_none(root: Path, name: str | None): + if not name: + return None + _, prof_dir, _, _ = _paths(root) + p = prof_dir / f"{name}.toml" + if not p.exists(): + raise StateError(f"profil inconnu: {name} ({p} absent)") + return load_profile(p) + + +def _cmd_status(args) -> int: + root = Path(args.root) + mod_dir, _, pins_file, _ = _paths(root) + manifests = load_all(mod_dir) + routes = load_routes() + actuals = _observe_all(manifests, routes) + rows = [] + for mid, m in sorted(manifests.items()): + a = actuals.get(mid, Actual()) + rows.append({ + "id": mid, "category": m.category, "runtime": m.runtime, + "exposure": m.exposure, "priority": m.priority, + "protected": m.protected, "on": is_on(a), "rss_kb": a.rss_kb, + }) + if args.json: + print(json.dumps({"modules": rows}, ensure_ascii=False)) + return 0 + for r in sorted(rows, key=lambda r: (-r["priority"], r["id"])): + rss = f"{r['rss_kb'] / 1024:.0f} Mo" if r["rss_kb"] else "—" + print(f"{'🟢' if r['on'] else '⚫'} {r['id']:<20} {r['category']:<9} " + f"{r['runtime']:<7} {r['exposure']:<9} prio={r['priority']:<3} {rss:>8}" + f"{' 🔒' if r['protected'] else ''}") + return 0 + + +def _cmd_diff(args) -> int: + root = Path(args.root) + mod_dir, _, pins_file, _ = _paths(root) + manifests = load_all(mod_dir) + profile = _load_profile_or_none(root, _active_profile_name(root, args.profile)) + pins = load_pins(pins_file) + actuals = _observe_all(manifests, load_routes()) + changes = plan_changes(manifests, profile, pins, actuals) + payload = [{"id": c.id, "action": c.action, "priority": c.priority, + "reason": c.reason} for c in changes] + if args.json: + print(json.dumps({"changes": payload}, ensure_ascii=False)) + return 0 + if not changes: + print("✅ rien à changer — l'état réel correspond déjà au profil.") + return 0 + print(f"{len(changes)} changement(s) — Phase 1 n'applique rien :") + for c in changes: + print(f" {'⛔ stop ' if c.action == 'stop' else '▶️ start'} {c.id:<20} ({c.reason})") + return 0 + + +def _cmd_scan(args) -> int: + root = Path(args.root) + mod_dir, _, _, _ = _paths(root) + rc, out = _run(["systemctl", "list-unit-files", "secubox-*.service", + "--no-legend", "--plain"]) + units = [line.split()[0] for line in out.splitlines() if line.strip()] + rc, out = _run(["lxc-ls", "-1"]) + lxc_names = {n.strip() for n in out.splitlines() if n.strip()} + # load_routes() renvoie None quand le fichier de routes existe mais est + # illisible/corrompu (indéterminable) — discover() attend un set() ferme, + # donc on retombe sur "aucune route connue" plutôt que de propager le None + # (qui ferait planter `for r in sorted(routes)` dans scan._route_for). + manifests = discover(units=units, lxc_names=lxc_names, routes=load_routes() or set()) + written = write_drafts(manifests, mod_dir, force=args.force) + skipped = len(manifests) - len(written) + print(f"{len(manifests)} module(s) découvert(s) — {len(written)} manifeste(s) écrit(s), " + f"{skipped} conservé(s) (déjà présents ; --force pour écraser).") + return 0 + + +def _run(argv: list[str]) -> tuple[int, str]: + try: + p = subprocess.run(argv, capture_output=True, text=True, timeout=15) + return p.returncode, p.stdout + except (OSError, subprocess.SubprocessError): + return 1, "" + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser( + prog="secubox-profilectl", + description="Inventaire et diff des modules SecuBox (Phase 1 : lecture seule).") + p.add_argument("--root", default=str(DEFAULT_ROOT), + help="racine de config (défaut: /etc/secubox)") + sub = p.add_subparsers(dest="cmd", required=True) + + sp = sub.add_parser("status", help="état et coût de chaque module") + sp.add_argument("--json", action="store_true") + sp.set_defaults(func=_cmd_status) + + sp = sub.add_parser("diff", help="ce qu'un profil changerait (n'applique rien)") + sp.add_argument("--profile", default=None) + sp.add_argument("--json", action="store_true") + sp.set_defaults(func=_cmd_diff) + + sp = sub.add_parser("scan", help="dériver les manifestes du réel") + sp.add_argument("--force", action="store_true", + help="écraser les manifestes existants (ils font autorité par défaut)") + sp.set_defaults(func=_cmd_scan) + + args = p.parse_args(argv) + try: + return args.func(args) + except (ManifestError, StateError) as exc: + print(f"erreur: {exc}", file=sys.stderr) + return 2 + except ProtectedViolation as exc: + print(f"refusé: {exc}", file=sys.stderr) + return 3 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/secubox-profiles/debian/changelog b/packages/secubox-profiles/debian/changelog new file mode 100644 index 00000000..bffe4cce --- /dev/null +++ b/packages/secubox-profiles/debian/changelog @@ -0,0 +1,5 @@ +secubox-profiles (0.1.0-1~bookworm1) bookworm; urgency=medium + + * Phase 1 : manifestes, profiler scan, status, diff (lecture seule). + + -- Gerald KERMA Fri, 17 Jul 2026 12:00:00 +0200 diff --git a/packages/secubox-profiles/debian/compat b/packages/secubox-profiles/debian/compat new file mode 100644 index 00000000..b1bd38b6 --- /dev/null +++ b/packages/secubox-profiles/debian/compat @@ -0,0 +1 @@ +13 diff --git a/packages/secubox-profiles/debian/control b/packages/secubox-profiles/debian/control new file mode 100644 index 00000000..ec2f207d --- /dev/null +++ b/packages/secubox-profiles/debian/control @@ -0,0 +1,13 @@ +Source: secubox-profiles +Section: admin +Priority: optional +Maintainer: Gerald KERMA +Build-Depends: debhelper-compat (= 13) +Standards-Version: 4.6.2 + +Package: secubox-profiles +Architecture: all +Depends: ${misc:Depends}, python3 (>= 3.11), secubox-core +Description: SecuBox — inventaire et profils de modules (phase 1, lecture seule) + Manifestes de modules, profiler de configuration, état et diff. + Phase 1 n'applique aucun changement : scan/status/diff uniquement. diff --git a/packages/secubox-profiles/debian/install b/packages/secubox-profiles/debian/install new file mode 100644 index 00000000..fd3ce952 --- /dev/null +++ b/packages/secubox-profiles/debian/install @@ -0,0 +1,2 @@ +api/*.py usr/lib/secubox/profiles/api/ +sbin/secubox-profilectl usr/sbin/ diff --git a/packages/secubox-profiles/debian/rules b/packages/secubox-profiles/debian/rules new file mode 100755 index 00000000..cbe925d7 --- /dev/null +++ b/packages/secubox-profiles/debian/rules @@ -0,0 +1,3 @@ +#!/usr/bin/make -f +%: + dh $@ diff --git a/packages/secubox-profiles/sbin/secubox-profilectl b/packages/secubox-profiles/sbin/secubox-profilectl new file mode 100755 index 00000000..e94d461f --- /dev/null +++ b/packages/secubox-profiles/sbin/secubox-profilectl @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +# SecuBox-Deb :: secubox-profilectl +set -euo pipefail +readonly MODULE="profiles" +readonly VERSION="0.1.0" +cd /usr/lib/secubox/profiles +exec /usr/bin/python3 -m api.cli "$@" diff --git a/packages/secubox-profiles/tests/test_cli.py b/packages/secubox-profiles/tests/test_cli.py new file mode 100644 index 00000000..db085a4f --- /dev/null +++ b/packages/secubox-profiles/tests/test_cli.py @@ -0,0 +1,89 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. +import json + +import pytest + +from api.cli import main + +MANIFEST = """ +id = "lyrion" +category = "media" +runtime = "native" +exposure = "lan" +units = ["secubox-lyrion.service"] +priority = 30 +""" + + +@pytest.fixture() +def root(tmp_path): + (tmp_path / "modules.d").mkdir() + (tmp_path / "modules.d" / "lyrion.toml").write_text(MANIFEST) + (tmp_path / "profiles").mkdir() + (tmp_path / "profiles" / "media.toml").write_text( + 'name = "media"\nlabel = "🎬 Média"\non = ["lyrion"]\n') + return tmp_path + + +def test_status_json_lists_modules(root, capsys, monkeypatch): + monkeypatch.setattr("api.cli._observe_all", + lambda ms, routes: {"lyrion": __import__( + "api.observe", fromlist=["Actual"]).Actual( + enabled=True, active=True, rss_kb=1024)}) + rc = main(["--root", str(root), "status", "--json"]) + out = json.loads(capsys.readouterr().out) + assert rc == 0 + assert out["modules"][0]["id"] == "lyrion" + assert out["modules"][0]["on"] is True + assert out["modules"][0]["category"] == "media" + + +def test_diff_reports_no_change_when_converged(root, capsys, monkeypatch): + monkeypatch.setattr("api.cli._observe_all", + lambda ms, routes: {"lyrion": __import__( + "api.observe", fromlist=["Actual"]).Actual( + enabled=True, active=True)}) + rc = main(["--root", str(root), "diff", "--profile", "media", "--json"]) + out = json.loads(capsys.readouterr().out) + assert rc == 0 and out["changes"] == [] + + +def test_diff_reports_stop_for_module_absent_from_profile(root, capsys, monkeypatch): + (root / "profiles" / "vide.toml").write_text('name = "vide"\nlabel = "v"\non = []\n') + monkeypatch.setattr("api.cli._observe_all", + lambda ms, routes: {"lyrion": __import__( + "api.observe", fromlist=["Actual"]).Actual( + enabled=True, active=True)}) + rc = main(["--root", str(root), "diff", "--profile", "vide", "--json"]) + out = json.loads(capsys.readouterr().out) + assert rc == 0 + assert out["changes"] == [{"id": "lyrion", "action": "stop", "priority": 30, + "reason": "absent du profil 'vide'"}] + + +def test_diff_unknown_profile_errors(root, capsys): + rc = main(["--root", str(root), "diff", "--profile", "fantome", "--json"]) + assert rc == 2 + + +def test_apply_is_not_a_command_in_phase_1(root): + # Garde-fou : Phase 1 est en lecture seule. Si `apply` apparaît ici, c'est + # que quelqu'un a court-circuité la Phase 3. + with pytest.raises(SystemExit): + main(["--root", str(root), "apply"]) + + +def test_scan_survives_unreadable_routes_file(root, monkeypatch): + # load_routes() renvoie None quand le fichier de routes est présent mais + # illisible/corrompu (indéterminable, distinct de "aucune route"). scan + # doit rester lecture seule et ne pas planter dans ce cas plutôt que de + # propager le None jusqu'à `for r in sorted(routes)`. + monkeypatch.setattr("api.cli.load_routes", lambda: None) + monkeypatch.setattr("api.cli._run", lambda argv: ( + (0, "secubox-lyrion.service enabled\n") if argv[0:2] == ["systemctl", "list-unit-files"] + else (0, ""))) + rc = main(["--root", str(root), "scan"]) + assert rc == 0 From 041b2d63edecb82b617d8a17c4fd2d3184e639cd Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 12:49:38 +0200 Subject: [PATCH 10/11] fix(profiles): stop scan from silently downgrading exposure on unreadable routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _cmd_scan collapsed load_routes()'s None (routes file present but unreadable/corrupt, indeterminate) into the same set() as a genuinely absent routes file, so a corrupt WAF routes file silently derived every routed module's exposure down from public to lan/internal with no operator feedback — and since a written manifest is authoritative, the degraded value would stick until a manual --force rescan. scan now warns on stderr in the indeterminate case only, naming the consequence and the remedy, and stays silent when routes are genuinely absent. _cmd_status/_cmd_diff were already correct and are untouched. Also drops the unused pins_file binding in _cmd_status. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/cli.py | 22 +++++++++++++----- packages/secubox-profiles/tests/test_cli.py | 25 +++++++++++++++++++-- 2 files changed, 40 insertions(+), 7 deletions(-) diff --git a/packages/secubox-profiles/api/cli.py b/packages/secubox-profiles/api/cli.py index 98a2135d..6213a431 100644 --- a/packages/secubox-profiles/api/cli.py +++ b/packages/secubox-profiles/api/cli.py @@ -59,7 +59,7 @@ def _load_profile_or_none(root: Path, name: str | None): def _cmd_status(args) -> int: root = Path(args.root) - mod_dir, _, pins_file, _ = _paths(root) + mod_dir, _, _, _ = _paths(root) manifests = load_all(mod_dir) routes = load_routes() actuals = _observe_all(manifests, routes) @@ -113,10 +113,22 @@ def _cmd_scan(args) -> int: rc, out = _run(["lxc-ls", "-1"]) lxc_names = {n.strip() for n in out.splitlines() if n.strip()} # load_routes() renvoie None quand le fichier de routes existe mais est - # illisible/corrompu (indéterminable) — discover() attend un set() ferme, - # donc on retombe sur "aucune route connue" plutôt que de propager le None - # (qui ferait planter `for r in sorted(routes)` dans scan._route_for). - manifests = discover(units=units, lxc_names=lxc_names, routes=load_routes() or set()) + # illisible/corrompu (indéterminable, distinct de "aucune route" = set()). + # discover() attend un set() ferme (voir scan._route_for) donc on retombe + # sur "aucune route connue" pour ne pas planter — MAIS ce repli fait + # dériver `exposure` vers le bas (public -> lan/internal) pour tout module + # routé, silencieusement, et scan n'écrase pas sans --force : la valeur + # dégradée resterait autoritaire. On prévient donc l'opérateur sur stderr + # dans ce seul cas (fichier présent mais illisible), jamais quand le + # fichier est simplement absent (cas normal, routes = set()). + routes = load_routes() + if routes is None: + print("⚠️ fichier de routes WAF illisible/corrompu — exposure peut être " + "sous-évaluée pour les modules routés (public -> lan/internal). " + "Corrigez le fichier de routes puis relancez `scan --force`.", + file=sys.stderr) + routes = set() + manifests = discover(units=units, lxc_names=lxc_names, routes=routes) written = write_drafts(manifests, mod_dir, force=args.force) skipped = len(manifests) - len(written) print(f"{len(manifests)} module(s) découvert(s) — {len(written)} manifeste(s) écrit(s), " diff --git a/packages/secubox-profiles/tests/test_cli.py b/packages/secubox-profiles/tests/test_cli.py index db085a4f..186af1b9 100644 --- a/packages/secubox-profiles/tests/test_cli.py +++ b/packages/secubox-profiles/tests/test_cli.py @@ -76,14 +76,35 @@ def test_apply_is_not_a_command_in_phase_1(root): main(["--root", str(root), "apply"]) -def test_scan_survives_unreadable_routes_file(root, monkeypatch): +def test_scan_survives_unreadable_routes_file(root, capsys, monkeypatch): # load_routes() renvoie None quand le fichier de routes est présent mais # illisible/corrompu (indéterminable, distinct de "aucune route"). scan # doit rester lecture seule et ne pas planter dans ce cas plutôt que de - # propager le None jusqu'à `for r in sorted(routes)`. + # propager le None jusqu'à `for r in sorted(routes)` — mais il ne doit + # PAS non plus retomber silencieusement sur "aucune route" : ça dégrade + # exposure (public -> lan/internal) pour tout module routé sans que + # l'opérateur ne le sache, et un manifeste écrit fait ensuite autorité + # (scan n'écrase pas sans --force). L'opérateur doit être prévenu. monkeypatch.setattr("api.cli.load_routes", lambda: None) monkeypatch.setattr("api.cli._run", lambda argv: ( (0, "secubox-lyrion.service enabled\n") if argv[0:2] == ["systemctl", "list-unit-files"] else (0, ""))) rc = main(["--root", str(root), "scan"]) + err = capsys.readouterr().err assert rc == 0 + assert "illisible" in err or "corrompu" in err + assert "exposure" in err.lower() + + +def test_scan_stays_silent_when_routes_file_genuinely_absent(root, capsys, monkeypatch): + # Fichier absent = aucune route, c'est le cas normal (box sans WAF routé). + # Aucun avertissement ne doit être émis dans ce cas — sinon l'opérateur + # ne peut plus distinguer "rien à signaler" de "attention, dégradé". + monkeypatch.setattr("api.cli.load_routes", lambda: set()) + monkeypatch.setattr("api.cli._run", lambda argv: ( + (0, "secubox-lyrion.service enabled\n") if argv[0:2] == ["systemctl", "list-unit-files"] + else (0, ""))) + rc = main(["--root", str(root), "scan"]) + err = capsys.readouterr().err + assert rc == 0 + assert err == "" From fc28ab76784f881840d92b2bde79dda51a8e1195 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Fri, 17 Jul 2026 13:14:02 +0200 Subject: [PATCH 11/11] fix(profiles): unbuildable package, rc collision on scan, data-driven protection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit C1: drop debian/compat (kept debhelper-compat (= 13) Build-Depends) — the package was the only one in the repo carrying both, which debhelper refuses to build. C2: cli._run() now returns rc=None for "did not execute" (OSError/timeout), mirroring observe._run_cmd's contract. _cmd_scan checks rc for both systemctl list-unit-files and lxc-ls and aborts rather than silently deriving LXC modules as runtime="native" from empty output; a non-root invocation (rc=0, empty output) is refused explicitly since it is indistinguishable from "no containers" by rc alone. C3: PROTECTED_IDS moves from scan.py to manifest.py (a schema property, not a profiler detail) and load_manifest() now forces protected=True for those ids regardless of what a shipped manifest declares, closing the gap where a sloppy module package could silently unprotect auth/aggregator/core/etc. Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/cli.py | 54 ++++++++++++++++- packages/secubox-profiles/api/manifest.py | 18 ++++++ packages/secubox-profiles/api/scan.py | 9 +-- packages/secubox-profiles/debian/compat | 1 - packages/secubox-profiles/tests/test_cli.py | 59 +++++++++++++++++++ packages/secubox-profiles/tests/test_diff.py | 24 +++++++- .../secubox-profiles/tests/test_manifest.py | 22 ++++++- 7 files changed, 178 insertions(+), 9 deletions(-) delete mode 100644 packages/secubox-profiles/debian/compat diff --git a/packages/secubox-profiles/api/cli.py b/packages/secubox-profiles/api/cli.py index 6213a431..fc4b7c2d 100644 --- a/packages/secubox-profiles/api/cli.py +++ b/packages/secubox-profiles/api/cli.py @@ -15,6 +15,7 @@ from __future__ import annotations import argparse import json +import os import subprocess import sys from pathlib import Path @@ -104,13 +105,57 @@ def _cmd_diff(args) -> int: return 0 +def _running_as_root() -> bool: + return os.geteuid() == 0 + + def _cmd_scan(args) -> int: root = Path(args.root) mod_dir, _, _, _ = _paths(root) + rc, out = _run(["systemctl", "list-unit-files", "secubox-*.service", "--no-legend", "--plain"]) + # rc=None (n'a pas pu s'exécuter) ou rc!=0 (a répondu par un échec) sont + # tous deux des cas indéterminés : on ne peut pas distinguer "aucune unit + # secubox-*" (out="", rc=0, cas normal) de "systemctl a échoué" sans + # regarder rc. Continuer sur out="" écrirait silencieusement zéro + # manifeste en laissant croire que le scan a réussi. + if rc != 0: + print(f"⚠️ systemctl list-unit-files a échoué (rc={rc!r}) — impossible " + "d'énumérer les units secubox-*.service. scan ne peut pas " + "produire un inventaire fiable dans cet état ; corrigez systemd " + "puis relancez `scan`.", file=sys.stderr) + return 1 units = [line.split()[0] for line in out.splitlines() if line.strip()] + + # lxc-ls non-root répond rc=0 avec une sortie vide (silencieuse) plutôt + # qu'une erreur — indistinguable d'une box sans conteneur. Sur cette box + # (24 conteneurs LXC), ça déclasserait silencieusement tous les modules + # LXC en runtime="native" dans un manifeste qui fait ensuite autorité : + # Phase 3 ne lancerait alors jamais `lxc-stop` sur eux. On refuse plutôt + # que d'écrire un inventaire qu'on sait potentiellement faux. + if not _running_as_root(): + print("⚠️ scan doit être lancé en root : sans privilèges, `lxc-ls` " + "renvoie une liste vide (rc=0, pas une erreur) et scan " + "dériverait à tort tous les conteneurs LXC en runtime=\"native\" " + "dans un manifeste qui fait ensuite autorité (Phase 3 ne " + "lancerait alors jamais `lxc-stop` sur ces modules). Relancez " + "`scan` en root (sudo).", file=sys.stderr) + return 1 + rc, out = _run(["lxc-ls", "-1"]) + # Même raisonnement que ci-dessus pour list-unit-files : rc=None ou + # rc!=0 est indéterminé, jamais silencieusement traité comme "aucun + # conteneur". La conséquence d'un mauvais repli ici est sévère (tous les + # modules LXC dérivés en "native") : on abandonne plutôt que d'écrire. + if rc != 0: + print(f"⚠️ lxc-ls a échoué (rc={rc!r}) — impossible de déterminer " + "quels modules tournent en conteneur LXC. Continuer " + "dériverait tous les modules LXC en runtime=\"native\" dans un " + "manifeste qui fait ensuite autorité (Phase 3 ne lancerait " + "alors jamais `lxc-stop`). Corrigez lxc-ls (paquet lxc " + "installé ? PATH ?) puis relancez `scan`.", file=sys.stderr) + return 1 lxc_names = {n.strip() for n in out.splitlines() if n.strip()} # load_routes() renvoie None quand le fichier de routes existe mais est # illisible/corrompu (indéterminable, distinct de "aucune route" = set()). @@ -136,12 +181,17 @@ def _cmd_scan(args) -> int: return 0 -def _run(argv: list[str]) -> tuple[int, str]: +def _run(argv: list[str]) -> tuple[int | None, str]: + """rc=None signale que la commande n'a PAS pu s'exécuter (OSError, timeout) — + à distinguer d'un rc non-nul qui est une réponse authentique de la commande. + Même contrat que observe._run_cmd : un (1, "") fabriqué ici serait + indistinguable d'une vraie réponse "non" de la commande (voir _cmd_scan, + qui a besoin de cette distinction pour ne pas écrire un manifeste faux).""" try: p = subprocess.run(argv, capture_output=True, text=True, timeout=15) return p.returncode, p.stdout except (OSError, subprocess.SubprocessError): - return 1, "" + return None, "" def main(argv: list[str] | None = None) -> int: diff --git a/packages/secubox-profiles/api/manifest.py b/packages/secubox-profiles/api/manifest.py index 3f9fd0e2..420c2b38 100644 --- a/packages/secubox-profiles/api/manifest.py +++ b/packages/secubox-profiles/api/manifest.py @@ -26,6 +26,17 @@ CATEGORIES = ("media", "security", "network", "infra", "dev", "mesh") DEFAULT_PRIORITY = 50 +# Le noyau protégé : éteindre l'un de ceux-là retire à l'utilisateur le moyen +# de rallumer quoi que ce soit (auth coupée = plus aucun login ; aggregator +# coupé = plus aucun webui de gestion ; etc.). C'est une propriété du SCHÉMA, +# pas une donnée que chaque manifeste peut librement déclarer : le design +# prévoit que modules.d/*.toml est shipé par le paquet du module lui-même +# (voir docs de conception), donc un `secubox-auth` un peu négligent pourrait +# shipper un manifeste avec `protected = false` et désactiver silencieusement +# la protection du cœur. On refuse de laisser la donnée décider de ça : +# load_manifest() force protected=True pour ces id, quoi que dise le fichier. +PROTECTED_IDS = frozenset({"auth", "aggregator", "core", "nginx", "firewall", "profiles"}) + class ManifestError(Exception): """Manifeste illisible ou invalide.""" @@ -90,6 +101,13 @@ def load_manifest(path: Path) -> Manifest: protected = d.get("protected", False) if not isinstance(protected, bool): raise ManifestError(f"{path}: protected={protected!r} doit être un booléen (true/false)") + # Structurel, pas data-driven (voir PROTECTED_IDS ci-dessus) : un paquet + # module qui shippe modules.d/auth.toml avec protected=false ne doit pas + # pouvoir désarmer la protection du noyau. On force à True plutôt que de + # lever ManifestError — un manifeste sloppy ne doit pas casser tout + # l'inventaire, mais il ne doit pas non plus pouvoir déprotéger le cœur. + if mid in PROTECTED_IDS: + protected = True needs = d.get("needs", []) if not isinstance(needs, list) or not all(isinstance(n, str) for n in needs): diff --git a/packages/secubox-profiles/api/scan.py b/packages/secubox-profiles/api/scan.py index f174d4f3..d425a196 100644 --- a/packages/secubox-profiles/api/scan.py +++ b/packages/secubox-profiles/api/scan.py @@ -21,13 +21,14 @@ from __future__ import annotations import json from pathlib import Path -from .manifest import CATEGORIES, Manifest +from .manifest import CATEGORIES, PROTECTED_IDS, Manifest MENU_DIR = Path("/usr/share/secubox/menu.d") -# Le noyau protégé : éteindre l'un de ceux-là retire à l'utilisateur le moyen -# de rallumer quoi que ce soit. Le premier scan doit déjà les marquer. -PROTECTED_IDS = frozenset({"auth", "aggregator", "core", "nginx", "firewall", "profiles"}) +# PROTECTED_IDS vit désormais dans api.manifest (c'est une propriété du +# schéma, appliquée aussi au chargement — voir load_manifest) : réimporté ici +# pour que discover() continue de marquer le premier scan sans changement de +# comportement. UNIT_PREFIX = "secubox-" UNIT_SUFFIX = ".service" diff --git a/packages/secubox-profiles/debian/compat b/packages/secubox-profiles/debian/compat deleted file mode 100644 index b1bd38b6..00000000 --- a/packages/secubox-profiles/debian/compat +++ /dev/null @@ -1 +0,0 @@ -13 diff --git a/packages/secubox-profiles/tests/test_cli.py b/packages/secubox-profiles/tests/test_cli.py index 186af1b9..7bdbf066 100644 --- a/packages/secubox-profiles/tests/test_cli.py +++ b/packages/secubox-profiles/tests/test_cli.py @@ -86,6 +86,7 @@ def test_scan_survives_unreadable_routes_file(root, capsys, monkeypatch): # l'opérateur ne le sache, et un manifeste écrit fait ensuite autorité # (scan n'écrase pas sans --force). L'opérateur doit être prévenu. monkeypatch.setattr("api.cli.load_routes", lambda: None) + monkeypatch.setattr("api.cli._running_as_root", lambda: True) monkeypatch.setattr("api.cli._run", lambda argv: ( (0, "secubox-lyrion.service enabled\n") if argv[0:2] == ["systemctl", "list-unit-files"] else (0, ""))) @@ -101,6 +102,7 @@ def test_scan_stays_silent_when_routes_file_genuinely_absent(root, capsys, monke # Aucun avertissement ne doit être émis dans ce cas — sinon l'opérateur # ne peut plus distinguer "rien à signaler" de "attention, dégradé". monkeypatch.setattr("api.cli.load_routes", lambda: set()) + monkeypatch.setattr("api.cli._running_as_root", lambda: True) monkeypatch.setattr("api.cli._run", lambda argv: ( (0, "secubox-lyrion.service enabled\n") if argv[0:2] == ["systemctl", "list-unit-files"] else (0, ""))) @@ -108,3 +110,60 @@ def test_scan_stays_silent_when_routes_file_genuinely_absent(root, capsys, monke err = capsys.readouterr().err assert rc == 0 assert err == "" + + +def test_scan_aborts_when_lxc_ls_did_not_execute(root, capsys, monkeypatch): + # rc=None (OSError/timeout) sur lxc-ls est indéterminé, PAS "aucun + # conteneur". Sur une box avec des conteneurs LXC, retomber sur out="" + # dériverait silencieusement tous les modules LXC en runtime="native" + # dans un manifeste qui fait ensuite autorité — c'est le C2 du review. + # scan doit refuser d'écrire plutôt que de produire cet inventaire faux. + monkeypatch.setattr("api.cli._running_as_root", lambda: True) + + def fake_run(argv): + if argv[0:2] == ["systemctl", "list-unit-files"]: + return 0, "secubox-lyrion.service enabled\n" + if argv[0] == "lxc-ls": + return None, "" # n'a pas pu s'exécuter + return 0, "" + + monkeypatch.setattr("api.cli._run", fake_run) + rc = main(["--root", str(root), "scan"]) + err = capsys.readouterr().err + assert rc != 0 + assert "lxc-ls" in err + mod_dir = root / "modules.d" + # Aucun nouveau manifeste dérivé n'a été écrit sur cette découverte ratée + # (le seul fichier présent est celui déjà posé par la fixture `root`). + assert sorted(p.name for p in mod_dir.glob("*.toml")) == ["lyrion.toml"] + + +def test_scan_handles_genuinely_empty_lxc_ls_without_false_alarm(root, capsys, monkeypatch): + # rc=0 et sortie vide, en root, c'est le cas normal d'une box sans + # conteneur LXC — ne doit PAS être traité comme une erreur. + monkeypatch.setattr("api.cli._running_as_root", lambda: True) + monkeypatch.setattr("api.cli._run", lambda argv: ( + (0, "secubox-lyrion.service enabled\n") if argv[0:2] == ["systemctl", "list-unit-files"] + else (0, ""))) + rc = main(["--root", str(root), "scan", "--force"]) + err = capsys.readouterr().err + assert rc == 0 + assert "lxc-ls" not in err + from api.manifest import load_manifest + m = load_manifest(root / "modules.d" / "lyrion.toml") + assert m.runtime == "native" + + +def test_scan_refuses_when_not_root(root, capsys, monkeypatch): + # lxc-ls non-root répond rc=0 avec une sortie vide, indistinguable d'une + # box sans conteneur : sur les 24 conteneurs de cette box, ça dériverait + # silencieusement tout en runtime="native". scan doit refuser plutôt que + # d'écrire un inventaire qu'il sait potentiellement faux. + monkeypatch.setattr("api.cli._running_as_root", lambda: False) + monkeypatch.setattr("api.cli._run", lambda argv: ( + (0, "secubox-lyrion.service enabled\n") if argv[0:2] == ["systemctl", "list-unit-files"] + else (0, ""))) + rc = main(["--root", str(root), "scan"]) + err = capsys.readouterr().err + assert rc != 0 + assert "root" in err.lower() diff --git a/packages/secubox-profiles/tests/test_diff.py b/packages/secubox-profiles/tests/test_diff.py index 57654b91..a90d2dc0 100644 --- a/packages/secubox-profiles/tests/test_diff.py +++ b/packages/secubox-profiles/tests/test_diff.py @@ -5,7 +5,7 @@ import pytest from api.diff import Change, ProtectedViolation, plan_changes -from api.manifest import Manifest +from api.manifest import Manifest, load_manifest from api.observe import Actual from api.state import Profile @@ -87,6 +87,28 @@ def test_pin_off_on_protected_module_is_refused(): plan_changes(ms, prof, {"auth": "off"}, {"auth": on()}) +def test_pin_off_on_core_module_with_sloppy_manifest_is_still_refused(tmp_path): + # C3 review: un paquet secubox-auth qui shippe protected=false ne doit + # pas pouvoir désarmer la protection — load_manifest() force déjà True + # pour les PROTECTED_IDS, et plan_changes doit continuer d'en tenir + # compte end-to-end (chargement réel, pas juste le dataclass à la main). + p = tmp_path / "auth.toml" + p.write_text( + 'id = "auth"\n' + 'category = "security"\n' + 'runtime = "native"\n' + 'exposure = "internal"\n' + 'units = ["secubox-auth.service"]\n' + 'protected = false\n' + ) + m = load_manifest(p) + assert m.protected is True # sanity: la garde de chargement a bien joué + ms = {"auth": m} + prof = Profile(name="p", label="p", on=frozenset()) + with pytest.raises(ProtectedViolation): + plan_changes(ms, prof, {"auth": "off"}, {"auth": on()}) + + def test_change_carries_reason(): ms = {"gitea": mk("gitea")} prof = Profile(name="media", label="m", on=frozenset()) diff --git a/packages/secubox-profiles/tests/test_manifest.py b/packages/secubox-profiles/tests/test_manifest.py index 95e903fb..d0bf0f13 100644 --- a/packages/secubox-profiles/tests/test_manifest.py +++ b/packages/secubox-profiles/tests/test_manifest.py @@ -4,7 +4,7 @@ # See LICENCE-CMSD-1.0.md for terms. import pytest -from api.manifest import Manifest, ManifestError, load_all, load_manifest +from api.manifest import PROTECTED_IDS, Manifest, ManifestError, load_all, load_manifest FULL = """ id = "peertube" @@ -111,6 +111,26 @@ def test_rejects_id_mismatching_filename(tmp_path): load_manifest(p) +def test_load_manifest_forces_protected_true_for_core_ids_even_when_data_says_false(tmp_path): + # Design : modules.d/*.toml est shipé par le paquet du module lui-même. + # Un secubox-auth un peu négligent (ou compromis) pourrait shipper + # protected=false et désactiver silencieusement la protection du noyau. + # C1 review: la garde doit être structurelle (PROTECTED_IDS), pas + # data-driven — load_manifest doit forcer True quoi que dise le fichier. + assert "auth" in PROTECTED_IDS + p = tmp_path / "auth.toml" + p.write_text( + 'id = "auth"\n' + 'category = "security"\n' + 'runtime = "native"\n' + 'exposure = "internal"\n' + 'units = ["secubox-auth.service"]\n' + 'protected = false\n' + ) + m = load_manifest(p) + assert m.protected is True + + def test_load_all_indexes_by_id_and_skips_non_toml(tmp_path): (tmp_path / "lyrion.toml").write_text(MINIMAL) (tmp_path / "peertube.toml").write_text(FULL)