feat(assist): bounded action catalog → scoped ctl argv (no arbitrary shell, allow-list) (ref sous-projet 2)

Co-Authored-By: Gerald KERMA <devel@cybermind.fr>
This commit is contained in:
CyberMind-FR 2026-07-25 12:49:45 +02:00
parent 29729ff711
commit 73869cc295
2 changed files with 126 additions and 0 deletions

View File

@ -0,0 +1,87 @@
# 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 :: assist.catalog the ONLY actions a center may run in a session.
Each action maps to a fixed argv (a scoped ctl or a read command). No entry
ever yields a shell string; every argument is validated against a strict
allow-list so a compromised center can never widen the surface. auth/secrets
scopes are unreachable (NON_DELEGATABLE parity).
"""
from __future__ import annotations
import re
from typing import List, Optional
NON_DELEGATABLE = {"auth", "secrets"}
_MODULE_RE = re.compile(r"^secubox-[a-z0-9][a-z0-9-]{1,40}$")
_SCOPE_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,40}$")
# Allow-listed modules a center may restart/toggle/reload. Conservative on
# purpose; extend deliberately. (No secubox-auth, no secubox-core.)
MODULE_ALLOW = frozenset({
"secubox-dns", "secubox-dpi", "secubox-crowdsec", "secubox-netdata",
"secubox-wireguard", "secubox-qos", "secubox-vhost", "secubox-nextcloud",
"secubox-mediaflow", "secubox-cdn", "secubox-nac", "secubox-netmodes",
})
class CatalogError(Exception):
"""Unknown action, disallowed target, or unsafe argument."""
def _safe(arg: str, pattern: re.Pattern) -> str:
if arg is None or not pattern.match(arg):
raise CatalogError(f"invalid argument: {arg!r}")
return arg
def _module(arg: str) -> str:
m = _safe(arg, _MODULE_RE)
if m not in MODULE_ALLOW:
raise CatalogError(f"module not allow-listed: {m}")
return m
def _scope(arg: str) -> str:
s = _safe(arg, _SCOPE_RE)
if s in NON_DELEGATABLE:
raise CatalogError(f"scope not delegatable: {s}")
return s
def resolve(action: str, arg: Optional[str]) -> List[str]:
"""Return the exact argv for a catalog action, or raise CatalogError."""
if action == "status.all":
return ["/usr/sbin/secubox-assistctl", "diag", "status"]
if action == "diag.collect":
return ["/usr/sbin/secubox-assistctl", "diag", "bundle"]
if action == "logs.tail":
unit = _module(arg) # only secubox-* units, allow-listed
return ["journalctl", "-u", unit, "-n", "200", "--no-pager"]
if action == "service.restart":
return ["sudo", "-n", "/usr/sbin/secubox-assistctl", "service", "restart", _module(arg)]
if action == "service.toggle":
# arg form "secubox-dns:on" | "secubox-dns:off"
mod, _, state = (arg or "").partition(":")
if state not in ("on", "off"):
raise CatalogError("toggle needs <module>:on|off")
return ["sudo", "-n", "/usr/sbin/secubox-assistctl", "service", "toggle", _module(mod), state]
if action == "config.reload":
return ["sudo", "-n", "/usr/sbin/secubox-assistctl", "config", "reload", _scope(arg)]
if action == "config.rollback":
return ["sudo", "-n", "/usr/sbin/secubox-assistctl", "config", "rollback", _scope(arg)]
raise CatalogError(f"unknown action: {action}")
CATALOG = {
"status.all": {"kind": "diag", "argv": ["/usr/sbin/secubox-assistctl", "diag", "status"], "needs": None},
"diag.collect": {"kind": "diag", "argv": ["/usr/sbin/secubox-assistctl", "diag", "bundle"], "needs": None},
"logs.tail": {"kind": "read", "argv": ["journalctl", "-u", "<module>", "-n", "200", "--no-pager"], "needs": "module"},
"service.restart": {"kind": "ctl", "argv": ["sudo", "-n", "/usr/sbin/secubox-assistctl", "service", "restart", "<module>"], "needs": "module"},
"service.toggle": {"kind": "ctl", "argv": ["sudo", "-n", "/usr/sbin/secubox-assistctl", "service", "toggle", "<module>", "<state>"], "needs": "module"},
"config.reload": {"kind": "ctl", "argv": ["sudo", "-n", "/usr/sbin/secubox-assistctl", "config", "reload", "<scope>"], "needs": "scope"},
"config.rollback": {"kind": "ctl", "argv": ["sudo", "-n", "/usr/sbin/secubox-assistctl", "config", "rollback", "<scope>"], "needs": "scope"},
}

View File

@ -0,0 +1,39 @@
# 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 pytest
from assist import catalog
def test_status_all_is_readonly_argv():
argv = catalog.resolve("status.all", None)
assert isinstance(argv, list) and argv # never a shell string
def test_service_restart_allowed_module():
argv = catalog.resolve("service.restart", "secubox-dns")
assert "secubox-dns" in argv
assert not any(";" in a or "&&" in a or "|" in a for a in argv)
def test_unknown_action_rejected():
with pytest.raises(catalog.CatalogError):
catalog.resolve("rm.rf", "/")
def test_module_outside_allowlist_rejected():
with pytest.raises(catalog.CatalogError):
catalog.resolve("service.restart", "sshd")
def test_secrets_scope_rejected():
with pytest.raises(catalog.CatalogError):
catalog.resolve("config.reload", "secrets")
with pytest.raises(catalog.CatalogError):
catalog.resolve("config.reload", "auth")
def test_shell_metachars_in_arg_rejected():
with pytest.raises(catalog.CatalogError):
catalog.resolve("logs.tail", "secubox-dns; rm -rf /")