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/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/__init__.py b/packages/secubox-profiles/api/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/secubox-profiles/api/cli.py b/packages/secubox-profiles/api/cli.py new file mode 100644 index 00000000..fc4b7c2d --- /dev/null +++ b/packages/secubox-profiles/api/cli.py @@ -0,0 +1,231 @@ +# 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 os +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, _, _, _ = _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 _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()). + # 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), " + f"{skipped} conservé(s) (déjà présents ; --force pour écraser).") + return 0 + + +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 None, "" + + +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/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/api/manifest.py b/packages/secubox-profiles/api/manifest.py new file mode 100644 index 00000000..420c2b38 --- /dev/null +++ b/packages/secubox-profiles/api/manifest.py @@ -0,0 +1,136 @@ +# 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 + +# 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.""" + + +@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") + + 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): + 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), + runtime=runtime, + exposure=_enum(_require(d, "exposure", path), EXPOSURES, "exposure", path), + units=tuple(units), + lxc=lxc, + portal_domain=portal.get("domain"), + priority=priority, + protected=protected, + needs=tuple(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/api/observe.py b/packages/secubox-profiles/api/observe.py new file mode 100644 index 00000000..106ba683 --- /dev/null +++ b/packages/secubox-profiles/api/observe.py @@ -0,0 +1,118 @@ +# 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 + +from .manifest import Manifest + +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 | 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 None, "" + + +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 None + + +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: 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]) + # 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 = 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()) + + 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: + 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) + + +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/api/scan.py b/packages/secubox-profiles/api/scan.py new file mode 100644 index 00000000..d425a196 --- /dev/null +++ b/packages/secubox-profiles/api/scan.py @@ -0,0 +1,188 @@ +# 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, PROTECTED_IDS, Manifest + +MENU_DIR = Path("/usr/share/secubox/menu.d") + +# 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" + + +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 + + +# 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: + 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: + 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 + + +_TOML_BASIC_ESCAPES = { + "\\": "\\\\", + '"': '\\"', + "\b": "\\b", + "\t": "\\t", + "\n": "\\n", + "\f": "\\f", + "\r": "\\r", +} + + +def _toml_str(s: str) -> str: + # 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: + 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/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/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/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/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_cli.py b/packages/secubox-profiles/tests/test_cli.py new file mode 100644 index 00000000..7bdbf066 --- /dev/null +++ b/packages/secubox-profiles/tests/test_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. +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, 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)` — 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._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"]) + 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._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"]) + 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 new file mode 100644 index 00000000..a90d2dc0 --- /dev/null +++ b/packages/secubox-profiles/tests/test_diff.py @@ -0,0 +1,124 @@ +# 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, load_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_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()) + 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, {}, {}) == [] diff --git a/packages/secubox-profiles/tests/test_manifest.py b/packages/secubox-profiles/tests/test_manifest.py new file mode 100644 index 00000000..d0bf0f13 --- /dev/null +++ b/packages/secubox-profiles/tests/test_manifest.py @@ -0,0 +1,140 @@ +# 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 PROTECTED_IDS, 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,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, 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(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 / "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 / "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. + p = tmp_path / "autre.toml" + p.write_text(MINIMAL) + with pytest.raises(ManifestError): + 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) + (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" diff --git a/packages/secubox-profiles/tests/test_observe.py b/packages/secubox-profiles/tests/test_observe.py new file mode 100644 index 00000000..6eb580cd --- /dev/null +++ b/packages/secubox-profiles/tests/test_observe.py @@ -0,0 +1,166 @@ +# 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 pathlib import Path + +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() + + +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 diff --git a/packages/secubox-profiles/tests/test_scan.py b/packages/secubox-profiles/tests/test_scan.py new file mode 100644 index 00000000..b0273450 --- /dev/null +++ b/packages/secubox-profiles/tests/test_scan.py @@ -0,0 +1,156 @@ +# 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 : 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 == "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): + # 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_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" + 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 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)