From 7ebad95adb5198e62d771f4b67d2fdb82583f95d Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Sat, 18 Jul 2026 11:13:12 +0200 Subject: [PATCH] =?UTF-8?q?feat(profiles):=20export.py=20=E2=80=94=20resol?= =?UTF-8?q?ve=20a=20profile's=20active=20modules=20to=20Debian=20packages?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Gerald KERMA --- packages/secubox-profiles/api/export.py | 107 +++++++++++++++++ .../secubox-profiles/tests/test_export.py | 112 ++++++++++++++++++ 2 files changed, 219 insertions(+) create mode 100644 packages/secubox-profiles/api/export.py create mode 100644 packages/secubox-profiles/tests/test_export.py diff --git a/packages/secubox-profiles/api/export.py b/packages/secubox-profiles/api/export.py new file mode 100644 index 00000000..1fef9358 --- /dev/null +++ b/packages/secubox-profiles/api/export.py @@ -0,0 +1,107 @@ +# 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 — export installateur (lecture seule) +CyberMind — https://cybermind.fr + +Résout l'ensemble désiré ON d'un profil (via state.resolve, pur) et le mappe +sur les paquets Debian propriétaires (units -> dpkg -S, repli secubox-). +Un module dont le paquet reste introuvable part dans `unresolved` — JAMAIS un +paquet fabriqué : un installateur qui omet un module en silence casse l'image. +""" +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass +from pathlib import Path + +from .manifest import Manifest +from .state import ON, Profile, resolve + +_UNIT_DIRS = ("/usr/lib/systemd/system/", "/lib/systemd/system/", + "/etc/systemd/system/") + + +@dataclass(frozen=True) +class ExportResult: + profile: str + on_ids: list[str] + packages: list[str] + unresolved: list[str] + rss_estimate_mo: int + + +def _run(argv: list[str]) -> tuple[int | None, str]: + """rc=None => la commande n'a PAS pu s'exécuter (à distinguer d'un rc!=0, + réponse authentique). Même contrat que cli._run / observe._run_cmd.""" + try: + p = subprocess.run(argv, capture_output=True, text=True, timeout=15) + return p.returncode, p.stdout + except (OSError, subprocess.SubprocessError): + return None, "" + + +def _package_for_unit(unit: str, run) -> str | None: + for base in _UNIT_DIRS: + rc, out = run(["dpkg", "-S", base + unit]) + if rc == 0 and ":" in out: + return out.splitlines()[0].split(":", 1)[0].strip() + rc, out = run(["dpkg", "-S", unit]) # bare name (dpkg matches the path) + if rc == 0 and ":" in out: + return out.splitlines()[0].split(":", 1)[0].strip() + return None + + +def _installed(pkg: str, run) -> bool: + rc, out = run(["dpkg-query", "-W", "-f=${Status}", pkg]) + return rc == 0 and "install ok installed" in out + + +def resolve_packages(manifests: dict[str, Manifest], profile: Profile | None, + pins: dict[str, str], *, run=_run, + rss_kb: dict[str, int] | None = None) -> ExportResult: + rss_kb = rss_kb or {} + on = [m for _, m in sorted(manifests.items()) if resolve(m, profile, pins) == ON] + packages: set[str] = set() + unresolved: list[str] = [] + for m in on: + pkg = None + for u in m.units: + pkg = _package_for_unit(u, run) + if pkg: + break + if not pkg: + candidate = f"secubox-{m.id}" + if _installed(candidate, run): + pkg = candidate + if pkg: + packages.add(pkg) + else: + unresolved.append(m.id) + rss_mo = int(sum(rss_kb.get(m.id, 0) for m in on) / 1024) + return ExportResult( + profile=profile.name if profile else "", + on_ids=[m.id for m in on], + packages=sorted(packages), + unresolved=sorted(unresolved), + rss_estimate_mo=rss_mo, + ) + + +def format_pkglist(r: ExportResult) -> str: + return "\n".join(r.packages) + + +def format_apt(r: ExportResult) -> str: + return "apt-get install -y" + ("" if not r.packages else " " + " ".join(r.packages)) + + +def format_json(r: ExportResult) -> str: + return json.dumps({ + "profile": r.profile, "on_ids": r.on_ids, "packages": r.packages, + "unresolved": r.unresolved, "rss_estimate_mo": r.rss_estimate_mo, + }, ensure_ascii=False, indent=2) diff --git a/packages/secubox-profiles/tests/test_export.py b/packages/secubox-profiles/tests/test_export.py new file mode 100644 index 00000000..95632dab --- /dev/null +++ b/packages/secubox-profiles/tests/test_export.py @@ -0,0 +1,112 @@ +# 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.export import (ExportResult, format_apt, format_json, format_pkglist, + resolve_packages) +from api.manifest import Manifest +from api.state import Profile + + +def _m(mid, units=(), protected=False): + return Manifest(id=mid, category="infra", runtime="native", + exposure="lan", units=tuple(units), protected=protected) + + +def _fake_run(mapping): + """mapping: unit-or-path substring -> package name (dpkg -S hit). + dpkg-query -W install-check succeeds for packages in `mapping.values()`.""" + pkgs = set(mapping.values()) + + def run(argv): + if argv[:2] == ["dpkg", "-S"]: + target = argv[2] + for key, pkg in mapping.items(): + if key in target: + return 0, f"{pkg}: {target}\n" + return 1, "" + if argv[:1] == ["dpkg-query"]: + pkg = argv[-1] + return (0, "install ok installed") if pkg in pkgs else (1, "") + return None, "" + return run + + +def test_resolve_maps_units_to_packages_deduped_sorted(): + manifests = { + "auth": _m("auth", ["secubox-auth.service"], protected=True), + "waf": _m("waf", ["secubox-waf.service"]), + "off1": _m("off1", ["secubox-off1.service"]), + } + profile = Profile(name="p", label="p", on=frozenset({"waf"})) # off1 not listed + run = _fake_run({"secubox-auth.service": "secubox-auth", + "secubox-waf.service": "secubox-waf"}) + r = resolve_packages(manifests, profile, {}, run=run) + assert r.on_ids == ["auth", "waf"] # protected auth + listed waf, off1 excluded + assert r.packages == ["secubox-auth", "secubox-waf"] + assert r.unresolved == [] + + +def test_fallback_to_secubox_id_when_no_unit(): + # core is protected (always ON) and has no units → must fall back to the + # secubox- package name, verified installed via dpkg-query. + manifests = {"core": _m("core", units=(), protected=True)} + + def run(argv): + if argv[:1] == ["dpkg-query"]: + return (0, "install ok installed") if argv[-1] == "secubox-core" else (1, "") + if argv[:2] == ["dpkg", "-S"]: + return 1, "" # no unit → no dpkg -S hit + return None, "" + r = resolve_packages(manifests, None, {}, run=run) + assert r.packages == ["secubox-core"] + assert r.unresolved == [] + + +def test_truly_unknown_module_is_unresolved_never_fabricated(): + manifests = {"ghost": _m("ghost", ["secubox-ghost.service"])} + profile = Profile(name="p", label="p", on=frozenset({"ghost"})) + + def run(argv): + if argv[:2] == ["dpkg", "-S"]: + return 1, "" # dpkg -S: not found + if argv[:1] == ["dpkg-query"]: + return 1, "" # secubox-ghost NOT installed + return None, "" + r = resolve_packages(manifests, profile, {}, run=run) + assert r.packages == [] + assert r.unresolved == ["ghost"] + + +def test_dpkg_cannot_run_degrades_to_unresolved_not_a_package(): + manifests = {"waf": _m("waf", ["secubox-waf.service"])} + profile = Profile(name="p", label="p", on=frozenset({"waf"})) + + def run(argv): + return None, "" # rc=None everywhere: command could not run + r = resolve_packages(manifests, profile, {}, run=run) + assert r.packages == [] + assert r.unresolved == ["waf"] + + +def test_rss_estimate_summed_over_on_set(): + manifests = {"a": _m("a", ["secubox-a.service"]), + "b": _m("b", ["secubox-b.service"], protected=True)} + profile = Profile(name="p", label="p", on=frozenset({"a"})) + run = _fake_run({"secubox-a.service": "secubox-a", "secubox-b.service": "secubox-b"}) + r = resolve_packages(manifests, profile, {}, run=run, + rss_kb={"a": 20480, "b": 10240}) # 20 + 10 Mo + assert r.rss_estimate_mo == 30 + + +def test_formatters(): + r = ExportResult(profile="lite", on_ids=["auth", "waf"], + packages=["secubox-auth", "secubox-waf"], + unresolved=["ghost"], rss_estimate_mo=42) + assert format_pkglist(r) == "secubox-auth\nsecubox-waf" + assert format_apt(r) == "apt-get install -y secubox-auth secubox-waf" + doc = json.loads(format_json(r)) + assert doc["profile"] == "lite" and doc["packages"] == ["secubox-auth", "secubox-waf"] + assert doc["unresolved"] == ["ghost"] and doc["rss_estimate_mo"] == 42