feat(profiles): export.py — resolve a profile's active modules to Debian packages

Co-Authored-By: Gerald KERMA <devel@cybermind.fr>
This commit is contained in:
CyberMind-FR 2026-07-18 11:13:12 +02:00
parent b72162b881
commit 7ebad95adb
2 changed files with 219 additions and 0 deletions

View File

@ -0,0 +1,107 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# 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-<id>).
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)

View File

@ -0,0 +1,112 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# 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-<id> 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