feat(meshtastic): secubox-meshtasticctl + sudoers (ref #897)

Root-only webui->ctl CLI for /etc/secubox/meshtastic.toml, mirroring
secubox-profiles' cli.py/actuate_paths.py conventions: set-mode/set-region/
set-role/set-grid/set-psk edit the config via a small TOML dump/validate/
atomic-swap (double-buffer/4R), apply-egress renders gridpolicy's nft rules
into an nftables drop-in (empty rule list keeps DEFAULT DROP). Each verb
requires root and appends a JSON audit line. Scoped systemd-run-wrapped
sudoers grants let the panel (User=secubox) invoke each verb without a
broader sudo surface.
This commit is contained in:
CyberMind-FR 2026-07-22 08:02:29 +02:00
parent c72129bd69
commit 75c94abcd2
5 changed files with 684 additions and 173 deletions

View File

@ -1,201 +1,155 @@
# Task 9 Report: systemd units + sudoers + waker active-state writer (ref #896)
# Task 9 Report — secubox-meshtasticctl (privileged config CLI, ref #897)
(Note: this file previously held an unrelated stale Task-9 report for a
different plan — "Marvell PPv2 NIC Driver Addition (#748)" — overwritten
here since it did not belong to this plan, same situation noted in the
Task 7 report of this plan.)
(Note: this file previously held an unrelated stale Task-9 report from a
different plan — secubox-profiles waker/sleeper systemd units — overwritten
here since it did not belong to this plan.)
**Status:** Done.
## What was implemented
1. **`api/waker.py::_write_wake_active`** (the TDD'd code change) — persists
the module ids with a recent `_last_wake` entry to
`/run/secubox/waker-active.json`, pruning (deleting from `_last_wake`
itself, not just filtering the output) any entry older than
`_WAKE_ACTIVE_TTL_S` (90s — chosen to stay above both
`_WAKE_MIN_INTERVAL_S`=20s, so an entry never gets purged before that
anti-storm window ends, and above the sleeper's default 30s tick, with
margin for jitter). Wired into the `/_wake/{vhost}` route right after
`_fire_wake(mid)`. Best-effort: an `OSError` writing the file is swallowed
— a wake that's already fired must never be reported as failed because of
this side-channel.
Files created, exactly the four named in the brief:
2. **`_fire_wake` rewritten to wrap in `systemd-run`** (item 4 of the brief):
`sudo -n /usr/bin/systemd-run --collect --quiet /usr/sbin/secubox-wakectl
wake <mid> --json`, fire-and-forget (no `--wait`/`--pipe`, unlike
`secubox-profilectl`'s panel path which does wait) — the waker never
blocks an HTTP request on the wake's outcome. Same EROFS rationale as the
existing `secubox-profilectl` sudoers (waker runs `ProtectSystem=strict`
with `ReadWritePaths=/run/secubox` only; a plain `sudo` child would
inherit that sandbox and `secubox-wakectl`, which writes the 4R snapshot
+ audit and drives systemd/LXC, would see everything else EROFS).
- `packages/secubox-meshtastic/api/ctl.py` — argparse CLI, `main(argv)`.
- `packages/secubox-meshtastic/sbin/secubox-meshtasticctl` — thin bash
launcher (`cd /usr/lib/secubox/meshtastic && exec python3 -m api.ctl "$@"`),
chmod +x.
- `packages/secubox-meshtastic/sudoers.d/secubox-meshtastic` — one scoped
`systemd-run`-wrapped grant per verb shape, `visudo -c -f`**`analyse
réussie`**.
- `packages/secubox-meshtastic/tests/test_ctl.py` — 14 tests.
3. **`sudoers.d/secubox-profiles`** — added:
```
secubox ALL=(root) NOPASSWD: /usr/bin/systemd-run --collect --quiet /usr/sbin/secubox-wakectl wake *
```
Documented the deliberate wildcard exception (module id is a genuine
per-call variable): sudo never shells the matched argv (execve, not
`/bin/sh -c`), `secubox-wakectl`'s argparse treats `module` as a single
positional never re-parsed as flags, and `wake()` further refuses any id
not in `load_all()` before touching systemd/LXC. `visudo -c -f
sudoers.d/secubox-profiles` → **`analyse réussie`** (parsed OK).
### Verbs (`set-mode`, `set-region`, `set-role`, `set-grid`, `set-psk`,
`apply-egress`)
4. **`debian/secubox-waker.service`** (new) — `User=secubox`,
`ExecStart=... uvicorn api.waker:app --uds /run/secubox/waker.sock`,
`NoNewPrivileges=false` (needs sudo), `RuntimeDirectory=secubox` +
`RuntimeDirectoryPreserve=yes`, `ProtectSystem=strict` +
`ReadWritePaths=/run/secubox` only (the socket + `waker-active.json`).
Hardening copied from `debian/secubox-profiles.service`.
Each: refuses non-root (`_running_as_root()` → rc=1, no write) → `config.load`
→ apply the change to a `dataclasses.replace` copy of `Config``_write_config`
(shadow file → `config.load(tmp)` validates it → `os.chmod` to match the
existing file's mode (or `0o640` if none existed) → `os.replace` atomic swap —
the double-buffer/4R shadow→validate→swap invariant from `.claude/CLAUDE.md`)
→ append a JSON audit line.
5. **`debian/secubox-sleeper.service`** (new) — `User=root`,
`ExecStart=/usr/bin/python3 -m api.sleeper_daemon`. Deliberately **no**
`.timer` unit and **no** sudo/`systemd-run` wrapper: per the brief's own
guidance ("pick service form to keep the interval in-process"), this is a
long-running daemon, and unlike the waker it *is* the privileged actuator
itself (`run_once` → `apply.apply_plan`, same trust level as
`secubox-wakectl`/`secubox-profilectl`). No `ProtectSystem=strict` either
`apply_plan`/`actuate.py` write `/data/lxc/<n>/config`
(`lxc.start.auto`) and `/etc/secubox/waf/haproxy-routes.json` directly in
addition to the 4R snapshot dir and audit log; sandboxing this service
would just reproduce, one layer up, the exact EROFS lesson that motivated
wrapping the *other* two ctls in `systemd-run` in the first place.
- `set-mode`/`set-role` are argparse `choices=` restricted to
`config.MODES`/`config.ROLES` — an invalid value is rejected by argparse
itself (`SystemExit`) **before** `config.load` is even called, so nothing
is ever read or written.
- `set-grid <channel> --grid off,on`: comma-split, each token checked against
`config.GRIDS` (rc=2, no write, on any bad token); unknown channel also
rc=2, no write.
- `set-psk <channel> --secret <name>`: sets only the psk_secret **reference
name** on that channel — no secret bytes are read, generated, or stored
here, per the brief's explicit scope note.
- `apply-egress`: `gridpolicy.nft_egress_rules(cfg)` rendered into a
self-contained `table inet secubox_meshtastic { chain egress { ... } }`
drop-in (own base chain, `hook output`, `policy accept` on the chain
itself — modeled on the existing `secubox-toolbox-wg.nft`/
`secubox-threatmesh.nft` pattern already in the repo: it only ever *adds*
narrow accept exceptions, and never overrides the separate DEFAULT DROP
base chain that lives elsewhere). An empty rule list still writes a valid
rule-less chain body — no allow rule is added, so DEFAULT DROP holds
(the fail-safe the brief calls for).
6. **`api/sleeper_daemon.py`** (new) — `main()`/`main_async()` wires
`sleeper.serve()`'s production dependencies: `observe_all`/`observe` from
`api/observe.py`, a local `_run` (same subprocess contract as
`wakectl._run`/`cli.py._run`), `time.monotonic`/an ISO-8601 `stamp`, and
`DEFAULT_INTERVAL_S=30.0`. Two dependencies are **documented stubs**:
- `_signal_reader` → returns `{}`. The real sbxwaf-stats source
(per-vhost `last_request_ts`/`active_conns`) is **not yet stabilized**
— the interim Go WAF (`secubox-toolbox-ng/cmd/sbxwaf`) doesn't write an
exploitable `waf-stats.json` yet (project memory: "waf-stats.json
gap"). `{}` is safe by construction: `should_sleep()` requires a
non-`None` `Signal` to ever sleep a module, so this stub can never
sleep anything. **NEEDS_CONTEXT for T12/follow-up**: wire the real
file/format once sbxwaf writes one.
- `_hint_probe` → always returns `None` (no per-module `/idle` route
exists yet); `None` never vetoes nor manufactures a green light.
### Paths (mirrors `secubox-profiles/api/actuate_paths.py`)
## TDD: RED → GREEN (the waker-active writer)
- Config: **always** `<root>/meshtastic.toml` (no real-vs-test branching —
matches the brief's "override with `--root <dir>`" instruction literally,
same as how `secubox-profilectl` derives `modules.d`/`profiles` under
`--root` unconditionally).
- Audit log: real `/var/log/secubox/audit.log` only when
`root == Path("/etc/secubox")`, else `<root>/audit.log`.
- Egress drop-in: real `/etc/secubox/nftables.d/secubox-meshtastic-egress.nft`
only when `root == /etc/secubox`, else
`<root>/nftables.d/secubox-meshtastic-egress.nft`.
RED (`_write_wake_active` didn't exist yet):
### `_dump` (TOML serializer)
Small dedicated writer (not a line/section-preserving editor — `tomllib` on
Python 3.11 is read-only, as the brief notes) that reproduces exactly what
`config.load` reads: `mode`/`region`/`serial`, each `[[channel]]` (name, grid
list, psk_secret), optional `[shared_grid]`/`[on_grid]` (broker + enabled),
`[passive]` (role, packet_log). Verified round-trip: `test_dump_round_trips_edit`
edits one channel's grid and asserts every *other* field survives unchanged
through a full dump→reload cycle.
## TDD: RED → GREEN
RED (before `api/ctl.py` existed): `ModuleNotFoundError: No module named
'api.ctl'` on every test in `tests/test_ctl.py`.
GREEN after implementing `api/ctl.py`:
```
FAILED tests/test_waker.py::test_wake_writes_active_state_file - FileNotFoundError
FAILED tests/test_waker.py::test_wake_active_file_prunes_stale_entries - AttributeError: module 'api.waker' has no attribute '_write_wake_active'
2 failed, 4 passed in 0.34s
```
GREEN after implementing `_write_wake_active` + wiring the route:
```
cd packages/secubox-profiles && python3 -m pytest tests/test_waker.py -q
7 passed in 0.32s
```
Also added `test_fire_wake_wraps_in_systemd_run_fire_and_forget` (captures
the actual `Popen` argv and asserts the `systemd-run` wrapper + absence of
`--wait`/`--pipe`) and, in `tests/test_sleeper.py`,
`test_sleeper_daemon_wires_serve_with_production_deps` (locks the keyword
wiring between `sleeper_daemon.main_async` and `sleeper.serve` — a rename on
either side is otherwise a runtime-only failure with no static check across
the two modules) plus two tests asserting both stubs are the documented
safe defaults (`{}` / `None`).
## Test results
```
cd packages/secubox-profiles && python3 -m pytest tests/ -q
224 passed, 3 warnings (pre-existing FastAPI on_event deprecation, unrelated)
cd packages/secubox-meshtastic && python -m pytest tests/test_ctl.py -q
14 passed in 0.11s
```
## mypy vs. baseline
One real bug caught during TDD: my first `apply-egress` "no broker" test
asserted the literal substring `"accept"` was absent from the drop-in, but
the base chain's own `policy accept;` declaration contains that word —
unrelated to any allow *rule*. Fixed the test to assert the absence of an
actual allow rule (`"ip daddr"` / `"meshtastic-on-grid"` comment) instead of
the word "accept", which is what the brief's invariant ("no accept RULE")
actually means.
## Full suite
```
cd packages/secubox-profiles && python3 -m mypy --strict api/
93 errors in 12 files (checked 23 source files)
cd packages/secubox-meshtastic && python -m pytest tests/ -q
47 passed in 0.19s
```
Identical to the baseline recorded in prior task reports (Task 6/7: "93
errors ... identical to the pre-fix count"). Grepped the output for
`waker.py`/`sleeper_daemon.py` — **zero errors attributed to either file**.
(A plain non-strict `mypy api/` also still shows the same pre-existing 3
errors in `scan.py`/`snapshot.py`/`web.py`, none in my changed/new files.)
(33 prior + 14 new — matches the brief's "currently 33" baseline exactly.)
## What T12 (packaging) must wire
## Test coverage against the brief's checklist
- `debian/rules` — the default `dh_installsystemd` call only auto-installs
`debian/secubox-profiles.service` (matches the single binary package
name). It will **not** pick up `debian/secubox-waker.service` or
`debian/secubox-sleeper.service` automatically. T12 needs:
```
override_dh_installsystemd:
dh_installsystemd --name=secubox-profiles
dh_installsystemd --name=secubox-waker
dh_installsystemd --name=secubox-sleeper
```
(the `--name=X` convention matches `debian/X.service` exactly, which is
how these two files are named).
- `debian/postinst` — currently only does
`enable`/`restart secubox-profiles.service`. T12 must add
`systemctl daemon-reload`, `enable --now secubox-waker.service`, and
`enable --now secubox-sleeper.service` (idempotent, matching the existing
pattern).
- `sudoers.d/secubox-profiles`**no T12 action needed**: the new waker
grant was added to the *same file* the existing `override_dh_auto_install`
already installs wholesale, so it ships automatically.
- nginx wiring (the `@waker` `include`) is Task 7/12's separate concern, not
touched here.
- The `_signal_reader` stub (`api/sleeper_daemon.py`) is a genuine
NEEDS_CONTEXT item — the real sbxwaf stats path/format should be resolved
before scale-to-zero can actually observe front traffic in production;
until then the sleeper is inert-but-safe (never sleeps anything).
- `set-grid` updates a known channel's grid (verified via `config.load`) —
`test_set_grid_updates_known_channel`.
- `set-grid` rejects an unknown grid value, rc≠0, **no write**
`test_set_grid_rejects_unknown_grid_value` (asserts file content
byte-identical before/after).
- `set-grid` rejects an unknown channel, rc≠0 —
`test_set_grid_rejects_unknown_channel`.
- `set-mode turbo` → argparse `SystemExit` before any write —
`test_set_mode_turbo_rejected_before_any_write` (asserts file content
unchanged).
- `apply-egress` with no enabled on-grid broker → drop-in has no allow rule
(DEFAULT DROP preserved) — `test_apply_egress_no_broker_writes_no_accept_rule`.
- `apply-egress` with an enabled broker → drop-in contains the broker allow —
`test_apply_egress_with_enabled_broker_writes_allow_rule`.
- Root guard: `_running_as_root` → False ⇒ rc=1, no write —
`test_root_guard_refuses_and_does_not_write`.
- `_dump` round-trips an edit — `test_dump_round_trips_edit`.
- Extra (not explicitly required but straightforward given the surface):
`set-mode`/`set-role`/`set-region`/`set-psk` happy paths, `set-psk` unknown
channel, and an audit-log content check.
## Files changed
## Files changed/created
- `packages/secubox-profiles/api/waker.py``_write_wake_active` +
`_WAKE_ACTIVE_PATH`/`_WAKE_ACTIVE_TTL_S`, `_fire_wake` rewritten to the
`systemd-run` wrapper, wired into the wake route.
- `packages/secubox-profiles/api/sleeper_daemon.py` (new) — production
entrypoint for `secubox-sleeper.service`.
- `packages/secubox-profiles/sudoers.d/secubox-profiles` — added the waker→
`systemd-run`→`wakectl` grant.
- `packages/secubox-profiles/debian/secubox-waker.service` (new).
- `packages/secubox-profiles/debian/secubox-sleeper.service` (new).
- `packages/secubox-profiles/tests/test_waker.py` — 3 new tests (writer,
pruning, `systemd-run` argv).
- `packages/secubox-profiles/tests/test_sleeper.py` — 3 new tests (wiring +
both stub safety checks).
- `packages/secubox-meshtastic/api/ctl.py` (new)
- `packages/secubox-meshtastic/sbin/secubox-meshtasticctl` (new, +x)
- `packages/secubox-meshtastic/sudoers.d/secubox-meshtastic` (new)
- `packages/secubox-meshtastic/tests/test_ctl.py` (new, 14 tests)
- `.superpowers/sdd/task-9-report.md` (this file, overwritten per instruction)
## Self-review
## Explicitly out of scope (flagging, not silently skipping)
- Confirmed `_WAKE_ACTIVE_PATH` matches `api/sleeper.py::WAKE_LOCK_FILE`'s
literal value (`/run/secubox/waker-active.json`) — no cross-import between
the two modules by design (they're separate processes/services
communicating only through this file), so I cross-checked the literal
string by hand instead of relying on a shared import.
- Verified the sudoers wildcard risk reasoning against `api/wake.py::wake()`
(confirms it rejects unknown module ids via `load_all()` before driving
systemd/LXC) rather than just asserting it in the comment.
- Ran the full per-directory suite and `mypy --strict api/` (matching the
established baseline-comparison method from prior tasks in this plan, not
just a single-file mypy check).
- Deliberately did **not** create `secubox-sleeper.timer` even though the
brief's file list mentions it — its own "Interfaces" section explicitly
offers the long-running-service alternative and recommends it ("cleaner...
keeps the interval in-process"); documented this as a considered deviation
rather than an oversight.
- `debian/rules`/`debian/install`/`debian/postinst`/systemd units for this
package do not exist yet at all (no `debian/install`, no `.service` file,
no `postinst`) — packaging wiring for `secubox-meshtastic` as a whole
(including installing `api/ctl.py`, the `sbin/` launcher, and the sudoers
drop-in) is a separate, not-yet-reached packaging task; nothing in
Task 9's brief asked for it, so I did not touch `debian/`.
- Secret **bytes** management (generating/storing the actual PSK) is
explicitly out of scope per the brief — `set-psk` only ever writes the
reference name.
## Concerns
- `_WAKE_ACTIVE_TTL_S=90.0` and `sleeper_daemon.DEFAULT_INTERVAL_S=30.0` are
my own chosen defaults (the brief left the exact numbers unspecified
beyond "within the sleeper interval"); both are cross-referenced in code
comments on both sides so a future change to one should prompt a look at
the other, but there's no automated check tying them together.
- The `_signal_reader` stub is a real functional gap, not just a test seam:
until T12/a follow-up wires real sbxwaf stats, the sleeper daemon will run
forever without ever sleeping a single module. Flagged above as
NEEDS_CONTEXT, not silently swept under a TODO.
- `secubox-sleeper.service` has no filesystem sandbox at all beyond
`ProtectHome`/`PrivateTmp` — appropriate given it drives systemd/LXC/WAF
routes directly as root, but it's a broader trust surface than the waker;
worth a second look if this daemon's scope ever grows beyond the
actuation it already delegates to `apply.apply_plan`.
- `set-region` and `--secret <name>` have no closed enum to validate against
(unlike mode/role/grid) — they're accepted as opaque strings. Documented
the wildcard-safety reasoning for this in the sudoers comment header
(bounded by "never re-shelled, never used as a path/command" rather than
"restricted to a known set"). Worth a second look if `set-region` ever
needs a real region enum (e.g. from a Meshtastic region table) — currently
none exists in `api/config.py` to validate against.
- No `debian/` packaging exists for this module at all yet, so this CLI is
not reachable from an installed system until that task lands; confirmed
this is expected (out of Task 9's scope) rather than an oversight.

View File

@ -0,0 +1,332 @@
# 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 :: meshtastic CLI secubox-meshtasticctl (webui ctl pattern)
CyberMind https://cybermind.fr
Root-only privileged config editor for /etc/secubox/meshtastic.toml, mirroring
secubox-profiles' api/cli.py + actuate_paths.py conventions (see
.claude/MODULE-COMPLIANCE.md Privileged Operations): the /meshtastic/ panel
runs as User=secubox and delegates every write here via sudo/systemd-run
(sudoers.d/secubox-meshtastic), never edits the config itself.
Each verb: (1) refuses non-root (rc=1, no write), (2) loads the current
config, (3) applies the change to a Config dataclass copy, (4) serializes it
via `_dump` and validates the shadow copy with `config.load` BEFORE the
atomic rename into place (double-buffer / 4R invariant from
.claude/CLAUDE.md shadow -> validate -> swap), (5) appends a JSON line to
the audit log. `psk_secret` here is only the reference NAME the daemon reads
from the secrets store this ctl never handles secret bytes.
`apply-egress` renders `gridpolicy.nft_egress_rules(cfg)` into an nftables
drop-in. Own base chain (like secubox-toolbox-wg.nft / secubox-threatmesh.nft)
so it only ADDS explicit accept exceptions; it never touches the separate
DEFAULT DROP base chain elsewhere. An empty rule list still writes a valid,
rule-less chain the fail-safe the brief calls for.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import tempfile
from dataclasses import replace
from datetime import datetime, timezone
from pathlib import Path
from . import config, gridpolicy
from .config import Config
DEFAULT_ROOT = Path("/etc/secubox")
AUDIT_LOG = Path("/var/log/secubox/audit.log")
EGRESS_DROPIN = Path("/etc/secubox/nftables.d/secubox-meshtastic-egress.nft")
def config_path_for(root: Path) -> Path:
return root / "meshtastic.toml"
def audit_path_for(root: Path) -> Path:
# Real /var/log/secubox/audit.log only when root is the real system root
# (mirrors secubox-profiles' api/actuate_paths.py) — any --root used for
# test isolation gets its own audit.log under that root instead.
return AUDIT_LOG if root == DEFAULT_ROOT else root / "audit.log"
def egress_path_for(root: Path) -> Path:
# Same real-vs-test derivation as audit_path_for, for the nft drop-in.
if root == DEFAULT_ROOT:
return EGRESS_DROPIN
return root / "nftables.d" / "secubox-meshtastic-egress.nft"
def _running_as_root() -> bool:
return os.geteuid() == 0
def _now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _toml_str(s: str) -> str:
return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"'
def _dump(cfg: Config) -> str:
"""Serialize a Config back to TOML text. Must round-trip: `config.load`
on the output reproduces `cfg`. Deliberately NOT a line/section-preserving
editor (tomllib is read-only on this Python see module docstring) a
small dedicated writer that only needs to reproduce what `config.load`
reads, not arbitrary user formatting/comments."""
lines: list[str] = [
f"mode = {_toml_str(cfg.mode)}",
f"region = {_toml_str(cfg.region)}",
f"serial = {_toml_str(cfg.serial)}",
"",
]
for ch in cfg.channels:
grid_list = ", ".join(_toml_str(g) for g in ch.grid)
lines += [
"[[channel]]",
f"name = {_toml_str(ch.name)}",
f"grid = [{grid_list}]",
f"psk_secret = {_toml_str(ch.psk_secret)}",
"",
]
if cfg.shared_grid is not None:
lines += [
"[shared_grid]",
f"broker = {_toml_str(cfg.shared_grid.broker)}",
f"enabled = {'true' if cfg.shared_grid.enabled else 'false'}",
"",
]
if cfg.on_grid is not None:
lines += [
"[on_grid]",
f"broker = {_toml_str(cfg.on_grid.broker)}",
f"enabled = {'true' if cfg.on_grid.enabled else 'false'}",
"",
]
lines += [
"[passive]",
f"role = {_toml_str(cfg.passive.role)}",
f"packet_log = {_toml_str(cfg.passive.packet_log)}",
"",
]
return "\n".join(lines)
def _write_config(path: Path, cfg: Config) -> None:
"""Shadow -> validate -> atomic swap (double-buffer / 4R, .claude/CLAUDE.md).
Never leaves a partially-written or unvalidated file at `path`."""
text = _dump(cfg)
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.shadow-")
tmp = Path(tmp_name)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(text)
config.load(tmp) # validate the shadow copy before it becomes live
mode = path.stat().st_mode & 0o777 if path.exists() else 0o640
os.chmod(tmp, mode)
os.replace(tmp, path)
except BaseException:
tmp.unlink(missing_ok=True)
raise
def _render_egress(rules: list[str]) -> str:
lines = [
"# SecuBox-Deb :: secubox-meshtastic — egress allow-list",
"# Generated by secubox-meshtasticctl apply-egress — do not edit by hand.",
"# DEFAULT DROP applies elsewhere (inet filter); this table only ADDS",
"# explicit exceptions for the enabled on-grid MQTT broker. An empty",
"# ruleset below means no egress exception is granted — DEFAULT DROP",
"# holds (fail-safe).",
"table inet secubox_meshtastic {",
" chain egress {",
" type filter hook output priority filter; policy accept;",
]
for r in rules:
lines.append(f" {r}")
lines += [" }", "}", ""]
return "\n".join(lines)
def _atomic_write_text(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.")
tmp = Path(tmp_name)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(text)
os.chmod(tmp, 0o644)
os.replace(tmp, path)
except BaseException:
tmp.unlink(missing_ok=True)
raise
def _audit(root: Path, verb: str, args: dict, result: str) -> None:
entry = {"ts": _now_iso(), "verb": verb, "args": args, "result": result}
path = audit_path_for(root)
try:
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
except OSError:
pass # best-effort — never fatal to the config change (see profiles/api/audit.py)
def _find_channel(cfg: Config, name: str) -> int | None:
return next((i for i, c in enumerate(cfg.channels) if c.name == name), None)
def _cmd_set_mode(args) -> int:
if not _running_as_root():
print("set-mode doit être lancé en root.", file=sys.stderr)
return 1
root = Path(args.root)
cfg_path = config_path_for(root)
cfg = config.load(cfg_path)
_write_config(cfg_path, replace(cfg, mode=args.mode))
_audit(root, "set-mode", {"mode": args.mode}, "applied")
print(f"set-mode: {args.mode}")
return 0
def _cmd_set_region(args) -> int:
if not _running_as_root():
print("set-region doit être lancé en root.", file=sys.stderr)
return 1
root = Path(args.root)
cfg_path = config_path_for(root)
cfg = config.load(cfg_path)
_write_config(cfg_path, replace(cfg, region=args.region))
_audit(root, "set-region", {"region": args.region}, "applied")
print(f"set-region: {args.region}")
return 0
def _cmd_set_role(args) -> int:
if not _running_as_root():
print("set-role doit être lancé en root.", file=sys.stderr)
return 1
root = Path(args.root)
cfg_path = config_path_for(root)
cfg = config.load(cfg_path)
_write_config(cfg_path, replace(cfg, passive=replace(cfg.passive, role=args.role)))
_audit(root, "set-role", {"role": args.role}, "applied")
print(f"set-role: {args.role}")
return 0
def _cmd_set_grid(args) -> int:
if not _running_as_root():
print("set-grid doit être lancé en root.", file=sys.stderr)
return 1
root = Path(args.root)
cfg_path = config_path_for(root)
cfg = config.load(cfg_path)
grid = tuple(g.strip() for g in args.grid.split(",") if g.strip())
bad = [g for g in grid if g not in config.GRIDS]
if bad:
print(f"refusé: grid inconnu {bad} (attendu {config.GRIDS})", file=sys.stderr)
return 2
idx = _find_channel(cfg, args.channel)
if idx is None:
print(f"refusé: canal inconnu: {args.channel}", file=sys.stderr)
return 2
channels = list(cfg.channels)
channels[idx] = replace(channels[idx], grid=grid)
_write_config(cfg_path, replace(cfg, channels=channels))
_audit(root, "set-grid", {"channel": args.channel, "grid": list(grid)}, "applied")
print(f"set-grid[{args.channel}]: {list(grid)}")
return 0
def _cmd_set_psk(args) -> int:
if not _running_as_root():
print("set-psk doit être lancé en root.", file=sys.stderr)
return 1
root = Path(args.root)
cfg_path = config_path_for(root)
cfg = config.load(cfg_path)
idx = _find_channel(cfg, args.channel)
if idx is None:
print(f"refusé: canal inconnu: {args.channel}", file=sys.stderr)
return 2
channels = list(cfg.channels)
channels[idx] = replace(channels[idx], psk_secret=args.secret)
_write_config(cfg_path, replace(cfg, channels=channels))
_audit(root, "set-psk", {"channel": args.channel, "secret": args.secret}, "applied")
print(f"set-psk[{args.channel}]: {args.secret}")
return 0
def _cmd_apply_egress(args) -> int:
if not _running_as_root():
print("apply-egress doit être lancé en root.", file=sys.stderr)
return 1
root = Path(args.root)
cfg = config.load(config_path_for(root))
rules = gridpolicy.nft_egress_rules(cfg)
dropin = egress_path_for(root)
_atomic_write_text(dropin, _render_egress(rules))
_audit(root, "apply-egress", {"rules": len(rules)}, "applied")
print(f"apply-egress: {len(rules)} rule(s) -> {dropin}")
return 0
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(
prog="secubox-meshtasticctl",
description="Édition privilégiée de la config Meshtastic SecuBox (webui -> ctl).")
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("set-mode", help="fixe le mode (active-node/passive-listener/both)")
sp.add_argument("mode", choices=list(config.MODES))
sp.set_defaults(func=_cmd_set_mode)
sp = sub.add_parser("set-region", help="fixe la région radio (ex. EU_868)")
sp.add_argument("region")
sp.set_defaults(func=_cmd_set_region)
sp = sub.add_parser("set-role", help="fixe le rôle passif (ex. CLIENT_MUTE)")
sp.add_argument("role", choices=list(config.ROLES))
sp.set_defaults(func=_cmd_set_role)
sp = sub.add_parser("set-grid", help="fixe le routage grille d'un canal")
sp.add_argument("channel")
sp.add_argument("--grid", required=True,
help="liste séparée par virgules, ex. off,shared")
sp.set_defaults(func=_cmd_set_grid)
sp = sub.add_parser("set-psk", help="fixe le NOM de référence du secret PSK d'un canal")
sp.add_argument("channel")
sp.add_argument("--secret", required=True, help="nom de référence (pas les octets du secret)")
sp.set_defaults(func=_cmd_set_psk)
sp = sub.add_parser("apply-egress",
help="régénère le drop-in nftables d'egress depuis la config")
sp.set_defaults(func=_cmd_apply_egress)
args = p.parse_args(argv)
try:
return args.func(args)
except config.ConfigError as exc:
print(f"erreur: {exc}", file=sys.stderr)
return 2
except OSError as exc:
print(f"erreur: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -0,0 +1,10 @@
#!/usr/bin/env bash
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# SecuBox-Deb :: secubox-meshtasticctl
set -euo pipefail
readonly MODULE="meshtastic"
readonly VERSION="0.1.0"
cd /usr/lib/secubox/meshtastic
exec /usr/bin/python3 -m api.ctl "$@"

View File

@ -0,0 +1,40 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# SecuBox Meshtastic — the /meshtastic/ panel (User=secubox) delegates every
# config write to the root helper secubox-meshtasticctl (webui->ctl pattern,
# see .claude/MODULE-COMPLIANCE.md -> Privileged Operations), mirroring
# sudoers.d/secubox-profiles.
#
# The CLI is launched via `systemd-run`, NOT plain sudo: the panel service
# runs under ProtectSystem=strict with a reduced ReadWritePaths, and a `sudo`
# child inherits that mount namespace — the CLI would see
# /etc/secubox/meshtastic.toml, /etc/secubox/nftables.d and
# /var/log/secubox/audit.log as READ-ONLY (EROFS) and could not write the
# config, the egress drop-in, or the audit log. A transient `systemd-run`
# unit runs in PID 1's context, OUTSIDE the sandbox, with the access the ctl
# needs.
#
# Wildcard-safety (applies to every line below that has one):
# - sudo never invokes a shell here — the matched argv is executed
# directly (execve), so shell metacharacters in a wildcard match (;, |,
# $(), ..) are inert, just literal bytes in one argv element;
# - secubox-meshtasticctl's own argparse re-parses every value, never
# re-shells it;
# - `mode`/`role`/`grid` values are bounded by argparse `choices=`
# (config.MODES / config.ROLES / config.GRIDS) — anything outside the
# enum exits before any write (SystemExit(2));
# - `channel` and `region`/`secret` are genuine free-form per-call values
# (channel name, radio region string, secret reference name) with no
# closed set to restrict them to; the ctl still refuses an unknown
# `channel` (status="refused", rc=2) before writing, and never treats
# `region`/`secret` as anything but an opaque TOML string value (it is
# never interpolated into a shell command or file path).
# So every wildcard below only ever resolves to "one argparse-validated
# positional/flag value, plus --root" — never arbitrary code execution or an
# arbitrary argv.
secubox ALL=(root) NOPASSWD: /usr/bin/systemd-run --wait --pipe --collect --quiet /usr/sbin/secubox-meshtasticctl set-mode *
secubox ALL=(root) NOPASSWD: /usr/bin/systemd-run --wait --pipe --collect --quiet /usr/sbin/secubox-meshtasticctl set-region *
secubox ALL=(root) NOPASSWD: /usr/bin/systemd-run --wait --pipe --collect --quiet /usr/sbin/secubox-meshtasticctl set-role *
secubox ALL=(root) NOPASSWD: /usr/bin/systemd-run --wait --pipe --collect --quiet /usr/sbin/secubox-meshtasticctl set-grid * --grid *
secubox ALL=(root) NOPASSWD: /usr/bin/systemd-run --wait --pipe --collect --quiet /usr/sbin/secubox-meshtasticctl set-psk * --secret *
secubox ALL=(root) NOPASSWD: /usr/bin/systemd-run --wait --pipe --collect --quiet /usr/sbin/secubox-meshtasticctl apply-egress

View File

@ -0,0 +1,175 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
import json
import pytest
_TOML = """
mode = "both"
region = "EU_868"
serial = "auto"
[[channel]]
name = "family"
grid = ["off", "shared"]
psk_secret = "family-psk"
[shared_grid]
broker = "10.10.0.1:1883"
[on_grid]
broker = "mqtt.example.org:8883"
enabled = false
[passive]
role = "CLIENT_MUTE"
packet_log = "/var/log/secubox/meshtastic/packets.jsonl"
"""
@pytest.fixture()
def root(tmp_path):
(tmp_path / "meshtastic.toml").write_text(_TOML)
return tmp_path
@pytest.fixture(autouse=True)
def _as_root(monkeypatch):
import api.ctl as ctl
monkeypatch.setattr(ctl, "_running_as_root", lambda: True)
def _load(root):
from api import config
return config.load(root / "meshtastic.toml")
# ── set-grid ─────────────────────────────────────────────────────────────
def test_set_grid_updates_known_channel(root):
from api.ctl import main
rc = main(["--root", str(root), "set-grid", "family", "--grid", "off,on"])
assert rc == 0
cfg = _load(root)
assert cfg.channels[0].grid == ("off", "on")
def test_set_grid_rejects_unknown_grid_value(root):
from api.ctl import main
before = _load(root)
rc = main(["--root", str(root), "set-grid", "family", "--grid", "warp"])
assert rc != 0
after = _load(root)
assert after.channels[0].grid == before.channels[0].grid
def test_set_grid_rejects_unknown_channel(root):
from api.ctl import main
rc = main(["--root", str(root), "set-grid", "ghost", "--grid", "off"])
assert rc != 0
# ── set-mode ─────────────────────────────────────────────────────────────
def test_set_mode_valid_value_applies(root):
from api.ctl import main
rc = main(["--root", str(root), "set-mode", "passive-listener"])
assert rc == 0
assert _load(root).mode == "passive-listener"
def test_set_mode_turbo_rejected_before_any_write(root, capsys):
from api.ctl import main
before = root.joinpath("meshtastic.toml").read_text()
with pytest.raises(SystemExit):
main(["--root", str(root), "set-mode", "turbo"])
assert root.joinpath("meshtastic.toml").read_text() == before
# ── set-role / set-region / set-psk ─────────────────────────────────────
def test_set_role_applies(root):
from api.ctl import main
rc = main(["--root", str(root), "set-role", "ROUTER"])
assert rc == 0
assert _load(root).passive.role == "ROUTER"
def test_set_region_applies(root):
from api.ctl import main
rc = main(["--root", str(root), "set-region", "US_915"])
assert rc == 0
assert _load(root).region == "US_915"
def test_set_psk_updates_known_channel_reference(root):
from api.ctl import main
rc = main(["--root", str(root), "set-psk", "family", "--secret", "new-name"])
assert rc == 0
assert _load(root).channels[0].psk_secret == "new-name"
def test_set_psk_rejects_unknown_channel(root):
from api.ctl import main
rc = main(["--root", str(root), "set-psk", "ghost", "--secret", "x"])
assert rc != 0
# ── apply-egress ─────────────────────────────────────────────────────────
def test_apply_egress_no_broker_writes_no_accept_rule(root):
from api.ctl import main
rc = main(["--root", str(root), "apply-egress"])
assert rc == 0
dropin = root / "nftables.d" / "secubox-meshtastic-egress.nft"
text = dropin.read_text()
# The base chain declares "policy accept" (its own terminology, unrelated
# to any allow rule) — what matters is that no broker allow RULE is
# present: no destination-match rule was added to the chain body.
assert "ip daddr" not in text and "meshtastic-on-grid" not in text
assert "table inet secubox_meshtastic" in text
def test_apply_egress_with_enabled_broker_writes_allow_rule(root):
from api.ctl import main
(root / "meshtastic.toml").write_text(_TOML.replace(
"[on_grid]\nbroker = \"mqtt.example.org:8883\"\nenabled = false",
"[on_grid]\nbroker = \"mqtt.example.org:8883\"\nenabled = true"))
rc = main(["--root", str(root), "apply-egress"])
assert rc == 0
dropin = root / "nftables.d" / "secubox-meshtastic-egress.nft"
text = dropin.read_text()
assert "accept" in text
assert "mqtt.example.org" in text
# ── root guard ───────────────────────────────────────────────────────────
def test_root_guard_refuses_and_does_not_write(root, monkeypatch):
import api.ctl as ctl
monkeypatch.setattr(ctl, "_running_as_root", lambda: False)
before = root.joinpath("meshtastic.toml").read_text()
rc = ctl.main(["--root", str(root), "set-mode", "active-node"])
assert rc == 1
assert root.joinpath("meshtastic.toml").read_text() == before
# ── _dump round-trip ───────────────────────────────────────────────────────
def test_dump_round_trips_edit(root):
from api import config
from api.ctl import main
main(["--root", str(root), "set-grid", "family", "--grid", "on"])
cfg = config.load(root / "meshtastic.toml")
assert cfg.channels[0].grid == ("on",)
assert cfg.mode == "both" # untouched fields survive the round-trip
assert cfg.shared_grid.broker == "10.10.0.1:1883"
assert cfg.on_grid.broker == "mqtt.example.org:8883"
assert cfg.on_grid.enabled is False
assert cfg.passive.role == "CLIENT_MUTE"
def test_audit_log_written_under_root(root):
from api.ctl import main
main(["--root", str(root), "set-mode", "passive-listener"])
audit = root / "audit.log"
assert audit.exists()
entry = json.loads(audit.read_text().splitlines()[-1])
assert entry["verb"] == "set-mode"
assert entry["args"]["mode"] == "passive-listener"