Add --ipvlan flag for WiFi-compatible deployments
MACVLAN assigns unique MACs per container; WiFi APs typically reject frames from unregistered MACs, making deckies unreachable from other LAN devices. IPvlan L2 shares the host's MAC, so all traffic passes through the AP normally. - network.py: add create_ipvlan_network, setup_host_ipvlan, teardown_host_ipvlan, HOST_IPVLAN_IFACE - config.py: add ipvlan: bool = False to DecnetConfig (persisted to state so teardown uses the right driver) - deployer.py: branch on config.ipvlan for create/setup/teardown - cli.py: add --ipvlan flag, wire into DecnetConfig - tests/test_network.py: new test module covering ips_to_range, create_macvlan/ipvlan, setup/teardown for both drivers Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -188,6 +188,7 @@ def deploy(
|
||||
log_file: Optional[str] = typer.Option(None, "--log-file", help="Write RFC 5424 syslog to this path inside containers (e.g. /var/log/decnet/decnet.log)"),
|
||||
dry_run: bool = typer.Option(False, "--dry-run", help="Generate compose file without starting containers"),
|
||||
no_cache: bool = typer.Option(False, "--no-cache", help="Force rebuild all images, ignoring Docker layer cache"),
|
||||
ipvlan: bool = typer.Option(False, "--ipvlan", help="Use IPvlan L2 instead of MACVLAN (required on WiFi interfaces)"),
|
||||
config_file: Optional[str] = typer.Option(None, "--config", "-c", help="Path to INI config file"),
|
||||
) -> None:
|
||||
"""Deploy deckies to the LAN."""
|
||||
@@ -294,6 +295,7 @@ def deploy(
|
||||
deckies=decky_configs,
|
||||
log_target=effective_log_target,
|
||||
log_file=effective_log_file,
|
||||
ipvlan=ipvlan,
|
||||
)
|
||||
|
||||
if effective_log_target and not dry_run:
|
||||
|
||||
@@ -44,6 +44,7 @@ class DecnetConfig(BaseModel):
|
||||
deckies: list[DeckyConfig]
|
||||
log_target: str | None = None # "ip:port" or None
|
||||
log_file: str | None = None # path for RFC 5424 syslog file output
|
||||
ipvlan: bool = False # use IPvlan L2 instead of MACVLAN (WiFi-friendly)
|
||||
|
||||
@field_validator("log_target")
|
||||
@classmethod
|
||||
|
||||
@@ -15,13 +15,16 @@ from decnet.composer import write_compose
|
||||
from decnet.network import (
|
||||
MACVLAN_NETWORK_NAME,
|
||||
allocate_ips,
|
||||
create_ipvlan_network,
|
||||
create_macvlan_network,
|
||||
detect_interface,
|
||||
detect_subnet,
|
||||
get_host_ip,
|
||||
ips_to_range,
|
||||
remove_macvlan_network,
|
||||
setup_host_ipvlan,
|
||||
setup_host_macvlan,
|
||||
teardown_host_ipvlan,
|
||||
teardown_host_macvlan,
|
||||
)
|
||||
|
||||
@@ -88,8 +91,19 @@ def deploy(config: DecnetConfig, dry_run: bool = False, no_cache: bool = False)
|
||||
decky_range = ips_to_range(ip_list)
|
||||
host_ip = get_host_ip(config.interface)
|
||||
|
||||
console.print(f"[bold cyan]Creating MACVLAN network[/] ({MACVLAN_NETWORK_NAME}) on {config.interface}")
|
||||
net_driver = "IPvlan L2" if config.ipvlan else "MACVLAN"
|
||||
console.print(f"[bold cyan]Creating {net_driver} network[/] ({MACVLAN_NETWORK_NAME}) on {config.interface}")
|
||||
if not dry_run:
|
||||
if config.ipvlan:
|
||||
create_ipvlan_network(
|
||||
client,
|
||||
interface=config.interface,
|
||||
subnet=config.subnet,
|
||||
gateway=config.gateway,
|
||||
ip_range=decky_range,
|
||||
)
|
||||
setup_host_ipvlan(config.interface, host_ip, decky_range)
|
||||
else:
|
||||
create_macvlan_network(
|
||||
client,
|
||||
interface=config.interface,
|
||||
@@ -142,10 +156,14 @@ def teardown(decky_id: str | None = None) -> None:
|
||||
|
||||
ip_list = [d.ip for d in config.deckies]
|
||||
decky_range = ips_to_range(ip_list)
|
||||
if config.ipvlan:
|
||||
teardown_host_ipvlan(decky_range)
|
||||
else:
|
||||
teardown_host_macvlan(decky_range)
|
||||
remove_macvlan_network(client)
|
||||
clear_state()
|
||||
console.print("[green]All deckies torn down. MACVLAN network removed.[/]")
|
||||
net_driver = "IPvlan" if config.ipvlan else "MACVLAN"
|
||||
console.print(f"[green]All deckies torn down. {net_driver} network removed.[/]")
|
||||
|
||||
|
||||
def status() -> None:
|
||||
|
||||
@@ -19,6 +19,7 @@ import docker
|
||||
|
||||
MACVLAN_NETWORK_NAME = "decnet_lan"
|
||||
HOST_MACVLAN_IFACE = "decnet_macvlan0"
|
||||
HOST_IPVLAN_IFACE = "decnet_ipvlan0"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -157,6 +158,35 @@ def create_macvlan_network(
|
||||
)
|
||||
|
||||
|
||||
def create_ipvlan_network(
|
||||
client: docker.DockerClient,
|
||||
interface: str,
|
||||
subnet: str,
|
||||
gateway: str,
|
||||
ip_range: str,
|
||||
) -> None:
|
||||
"""Create an IPvlan L2 Docker network. No-op if it already exists."""
|
||||
existing = [n.name for n in client.networks.list()]
|
||||
if MACVLAN_NETWORK_NAME in existing:
|
||||
return
|
||||
|
||||
client.networks.create(
|
||||
name=MACVLAN_NETWORK_NAME,
|
||||
driver="ipvlan",
|
||||
options={"parent": interface, "ipvlan_mode": "l2"},
|
||||
ipam=docker.types.IPAMConfig(
|
||||
driver="default",
|
||||
pool_configs=[
|
||||
docker.types.IPAMPool(
|
||||
subnet=subnet,
|
||||
gateway=gateway,
|
||||
iprange=ip_range,
|
||||
)
|
||||
],
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def remove_macvlan_network(client: docker.DockerClient) -> None:
|
||||
nets = [n for n in client.networks.list() if n.name == MACVLAN_NETWORK_NAME]
|
||||
for n in nets:
|
||||
@@ -197,6 +227,28 @@ def teardown_host_macvlan(decky_ip_range: str) -> None:
|
||||
_run(["ip", "link", "del", HOST_MACVLAN_IFACE], check=False)
|
||||
|
||||
|
||||
def setup_host_ipvlan(interface: str, host_ipvlan_ip: str, decky_ip_range: str) -> None:
|
||||
"""
|
||||
Create an IPvlan interface on the host so the deployer can reach deckies.
|
||||
Idempotent — skips steps that are already done.
|
||||
"""
|
||||
_require_root()
|
||||
|
||||
result = _run(["ip", "link", "show", HOST_IPVLAN_IFACE], check=False)
|
||||
if result.returncode != 0:
|
||||
_run(["ip", "link", "add", HOST_IPVLAN_IFACE, "link", interface, "type", "ipvlan", "mode", "l2"])
|
||||
|
||||
_run(["ip", "addr", "add", f"{host_ipvlan_ip}/32", "dev", HOST_IPVLAN_IFACE], check=False)
|
||||
_run(["ip", "link", "set", HOST_IPVLAN_IFACE, "up"])
|
||||
_run(["ip", "route", "add", decky_ip_range, "dev", HOST_IPVLAN_IFACE], check=False)
|
||||
|
||||
|
||||
def teardown_host_ipvlan(decky_ip_range: str) -> None:
|
||||
_require_root()
|
||||
_run(["ip", "route", "del", decky_ip_range, "dev", HOST_IPVLAN_IFACE], check=False)
|
||||
_run(["ip", "link", "del", HOST_IPVLAN_IFACE], check=False)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Compute an ip_range CIDR that covers a list of IPs
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
196
tests/test_network.py
Normal file
196
tests/test_network.py
Normal file
@@ -0,0 +1,196 @@
|
||||
"""
|
||||
Tests for decnet.network utility functions.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, call, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from decnet.network import (
|
||||
HOST_IPVLAN_IFACE,
|
||||
HOST_MACVLAN_IFACE,
|
||||
MACVLAN_NETWORK_NAME,
|
||||
create_ipvlan_network,
|
||||
create_macvlan_network,
|
||||
ips_to_range,
|
||||
setup_host_ipvlan,
|
||||
setup_host_macvlan,
|
||||
teardown_host_ipvlan,
|
||||
teardown_host_macvlan,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ips_to_range
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestIpsToRange:
|
||||
def test_single_ip(self):
|
||||
assert ips_to_range(["192.168.1.100"]) == "192.168.1.100/32"
|
||||
|
||||
def test_consecutive_small_range(self):
|
||||
# .97–.101: max^min = 4, bit_length=3, prefix=29 → .96/29
|
||||
result = ips_to_range([f"192.168.1.{i}" for i in range(97, 102)])
|
||||
from ipaddress import IPv4Network, IPv4Address
|
||||
net = IPv4Network(result)
|
||||
for i in range(97, 102):
|
||||
assert IPv4Address(f"192.168.1.{i}") in net
|
||||
|
||||
def test_range_crossing_cidr_boundary(self):
|
||||
# .110–.119 crosses the /28 boundary (.96–.111 vs .112–.127)
|
||||
# Subtraction gives /28 (wrong), XOR gives /27 (correct)
|
||||
ips = [f"192.168.1.{i}" for i in range(110, 120)]
|
||||
result = ips_to_range(ips)
|
||||
from ipaddress import IPv4Network, IPv4Address
|
||||
net = IPv4Network(result)
|
||||
for i in range(110, 120):
|
||||
assert IPv4Address(f"192.168.1.{i}") in net, (
|
||||
f"192.168.1.{i} not in computed range {result}"
|
||||
)
|
||||
|
||||
def test_all_ips_covered(self):
|
||||
# Larger spread: .10–.200
|
||||
ips = [f"10.0.0.{i}" for i in range(10, 201)]
|
||||
result = ips_to_range(ips)
|
||||
from ipaddress import IPv4Network, IPv4Address
|
||||
net = IPv4Network(result)
|
||||
for i in range(10, 201):
|
||||
assert IPv4Address(f"10.0.0.{i}") in net
|
||||
|
||||
def test_two_ips_same_cidr(self):
|
||||
# .100 and .101 share /31
|
||||
result = ips_to_range(["192.168.1.100", "192.168.1.101"])
|
||||
from ipaddress import IPv4Network, IPv4Address
|
||||
net = IPv4Network(result)
|
||||
assert IPv4Address("192.168.1.100") in net
|
||||
assert IPv4Address("192.168.1.101") in net
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# create_macvlan_network
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCreateMacvlanNetwork:
|
||||
def _make_client(self, existing=None):
|
||||
client = MagicMock()
|
||||
nets = [MagicMock(name=n) for n in (existing or [])]
|
||||
for net, n in zip(nets, (existing or [])):
|
||||
net.name = n
|
||||
client.networks.list.return_value = nets
|
||||
return client
|
||||
|
||||
def test_creates_network_when_absent(self):
|
||||
client = self._make_client([])
|
||||
create_macvlan_network(client, "eth0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||||
client.networks.create.assert_called_once()
|
||||
kwargs = client.networks.create.call_args
|
||||
assert kwargs[1]["driver"] == "macvlan"
|
||||
assert kwargs[1]["name"] == MACVLAN_NETWORK_NAME
|
||||
assert kwargs[1]["options"]["parent"] == "eth0"
|
||||
|
||||
def test_noop_when_network_exists(self):
|
||||
client = self._make_client([MACVLAN_NETWORK_NAME])
|
||||
create_macvlan_network(client, "eth0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||||
client.networks.create.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# create_ipvlan_network
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCreateIpvlanNetwork:
|
||||
def _make_client(self, existing=None):
|
||||
client = MagicMock()
|
||||
nets = [MagicMock(name=n) for n in (existing or [])]
|
||||
for net, n in zip(nets, (existing or [])):
|
||||
net.name = n
|
||||
client.networks.list.return_value = nets
|
||||
return client
|
||||
|
||||
def test_creates_ipvlan_network(self):
|
||||
client = self._make_client([])
|
||||
create_ipvlan_network(client, "wlan0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||||
client.networks.create.assert_called_once()
|
||||
kwargs = client.networks.create.call_args
|
||||
assert kwargs[1]["driver"] == "ipvlan"
|
||||
assert kwargs[1]["options"]["parent"] == "wlan0"
|
||||
assert kwargs[1]["options"]["ipvlan_mode"] == "l2"
|
||||
|
||||
def test_noop_when_network_exists(self):
|
||||
client = self._make_client([MACVLAN_NETWORK_NAME])
|
||||
create_ipvlan_network(client, "wlan0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||||
client.networks.create.assert_not_called()
|
||||
|
||||
def test_uses_same_network_name_as_macvlan(self):
|
||||
"""Both drivers share the same logical network name so compose files are identical."""
|
||||
client = self._make_client([])
|
||||
create_ipvlan_network(client, "wlan0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||||
assert client.networks.create.call_args[1]["name"] == MACVLAN_NETWORK_NAME
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# setup_host_macvlan / teardown_host_macvlan
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSetupHostMacvlan:
|
||||
@patch("decnet.network.os.geteuid", return_value=0)
|
||||
@patch("decnet.network._run")
|
||||
def test_creates_interface_when_absent(self, mock_run, _):
|
||||
# Simulate interface not existing (returncode != 0)
|
||||
mock_run.side_effect = lambda cmd, **kw: MagicMock(returncode=1) if "show" in cmd else MagicMock(returncode=0)
|
||||
setup_host_macvlan("eth0", "192.168.1.5", "192.168.1.96/27")
|
||||
calls = [str(c) for c in mock_run.call_args_list]
|
||||
assert any("macvlan" in c for c in calls)
|
||||
assert any("mode" in c and "bridge" in c for c in calls)
|
||||
|
||||
@patch("decnet.network.os.geteuid", return_value=0)
|
||||
@patch("decnet.network._run")
|
||||
def test_skips_create_when_interface_exists(self, mock_run, _):
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
setup_host_macvlan("eth0", "192.168.1.5", "192.168.1.96/27")
|
||||
calls = [c[0][0] for c in mock_run.call_args_list]
|
||||
# "ip link add <iface> link ..." should not be called when iface exists
|
||||
assert not any("link" in cmd and "add" in cmd and HOST_MACVLAN_IFACE in cmd for cmd in calls)
|
||||
|
||||
@patch("decnet.network.os.geteuid", return_value=1)
|
||||
def test_requires_root(self, _):
|
||||
with pytest.raises(PermissionError):
|
||||
setup_host_macvlan("eth0", "192.168.1.5", "192.168.1.96/27")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# setup_host_ipvlan / teardown_host_ipvlan
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSetupHostIpvlan:
|
||||
@patch("decnet.network.os.geteuid", return_value=0)
|
||||
@patch("decnet.network._run")
|
||||
def test_creates_ipvlan_interface(self, mock_run, _):
|
||||
mock_run.side_effect = lambda cmd, **kw: MagicMock(returncode=1) if "show" in cmd else MagicMock(returncode=0)
|
||||
setup_host_ipvlan("wlan0", "192.168.1.5", "192.168.1.96/27")
|
||||
calls = [str(c) for c in mock_run.call_args_list]
|
||||
assert any("ipvlan" in c for c in calls)
|
||||
assert any("mode" in c and "l2" in c for c in calls)
|
||||
|
||||
@patch("decnet.network.os.geteuid", return_value=0)
|
||||
@patch("decnet.network._run")
|
||||
def test_uses_ipvlan_iface_name(self, mock_run, _):
|
||||
mock_run.side_effect = lambda cmd, **kw: MagicMock(returncode=1) if "show" in cmd else MagicMock(returncode=0)
|
||||
setup_host_ipvlan("wlan0", "192.168.1.5", "192.168.1.96/27")
|
||||
calls = [str(c) for c in mock_run.call_args_list]
|
||||
assert any(HOST_IPVLAN_IFACE in c for c in calls)
|
||||
assert not any(HOST_MACVLAN_IFACE in c for c in calls)
|
||||
|
||||
@patch("decnet.network.os.geteuid", return_value=1)
|
||||
def test_requires_root(self, _):
|
||||
with pytest.raises(PermissionError):
|
||||
setup_host_ipvlan("wlan0", "192.168.1.5", "192.168.1.96/27")
|
||||
|
||||
@patch("decnet.network.os.geteuid", return_value=0)
|
||||
@patch("decnet.network._run")
|
||||
def test_teardown_uses_ipvlan_iface(self, mock_run, _):
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
teardown_host_ipvlan("192.168.1.96/27")
|
||||
calls = [str(c) for c in mock_run.call_args_list]
|
||||
assert any(HOST_IPVLAN_IFACE in c for c in calls)
|
||||
assert not any(HOST_MACVLAN_IFACE in c for c in calls)
|
||||
Reference in New Issue
Block a user