feat(meshtastic): grid policy + nft egress allow-list

- Implement targets_for(channel, cfg) to return grid membership subset
- Implement nft_egress_rules(cfg) to generate allow-rules for enabled on-grid brokers
- TDD: 5 test cases all passing (offgrid, shared+on, on-disabled, empty rules, broker rules)
- Full test suite: 23 passed

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
CyberMind-FR 2026-07-22 07:37:56 +02:00
parent bd2f9ac48c
commit bda0e1280e
2 changed files with 69 additions and 0 deletions

View File

@ -0,0 +1,33 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""SecuBox-Deb :: meshtastic — per-channel grid routing + nft egress allow-list."""
from __future__ import annotations
from .config import Config
def targets_for(channel_name: str, cfg: Config) -> set[str]:
"""Return the set of grids ("shared", "on") that a channel bridges to.
A channel bridges to "shared" if its grid contains "shared" and shared_grid is configured.
A channel bridges to "on" if its grid contains "on" AND on_grid is configured AND on_grid.enabled.
"""
ch = next((c for c in cfg.channels if c.name == channel_name), None)
if ch is None:
return set()
out: set[str] = set()
if "shared" in ch.grid and cfg.shared_grid is not None:
out.add("shared")
if "on" in ch.grid and cfg.on_grid is not None and cfg.on_grid.enabled:
out.add("on")
return out
def nft_egress_rules(cfg: Config) -> list[str]:
"""Allow rules for ENABLED on-grid brokers only. Empty => DEFAULT DROP holds.
Rendered into the operator drop-in the ctl installs."""
if not (cfg.on_grid and cfg.on_grid.enabled):
return []
host, _, port = cfg.on_grid.broker.partition(":")
port = port or "8883"
return [f'# secubox-meshtastic on-grid egress (broker {cfg.on_grid.broker})',
f'ip daddr {host} tcp dport {port} accept comment "meshtastic-on-grid"']

View File

@ -0,0 +1,36 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""SecuBox-Deb :: meshtastic — grid policy tests."""
from api.config import Config, ChannelCfg, BrokerCfg
def _cfg(grid, on_enabled=False):
return Config(channels=[ChannelCfg("c", tuple(grid), "psk")],
shared_grid=BrokerCfg("10.10.0.1:1883"),
on_grid=BrokerCfg("mqtt.x.org:8883", on_enabled))
def test_offgrid_only_bridges_nowhere():
from api.gridpolicy import targets_for
assert targets_for("c", _cfg(["off"])) == set()
def test_shared_and_on_targets():
from api.gridpolicy import targets_for
assert targets_for("c", _cfg(["off", "shared", "on"], on_enabled=True)) == {"shared", "on"}
def test_on_target_dropped_when_broker_disabled():
from api.gridpolicy import targets_for
assert targets_for("c", _cfg(["off", "on"], on_enabled=False)) == set()
def test_nft_rules_empty_when_no_on_grid():
from api.gridpolicy import nft_egress_rules
assert nft_egress_rules(_cfg(["off"], on_enabled=False)) == []
def test_nft_rules_allow_only_enabled_broker():
from api.gridpolicy import nft_egress_rules
rules = nft_egress_rules(_cfg(["on"], on_enabled=True))
assert any("8883" in r and "mqtt.x.org" in r and "accept" in r for r in rules)