diff --git a/decnet/cli.py b/decnet/cli.py index 5f6845a..2cecf54 100644 --- a/decnet/cli.py +++ b/decnet/cli.py @@ -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: diff --git a/decnet/config.py b/decnet/config.py index dbeb4f6..32e6cd9 100644 --- a/decnet/config.py +++ b/decnet/config.py @@ -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 diff --git a/decnet/deployer.py b/decnet/deployer.py index 7520a0a..5c3bd8f 100644 --- a/decnet/deployer.py +++ b/decnet/deployer.py @@ -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,16 +91,27 @@ 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: - create_macvlan_network( - client, - interface=config.interface, - subnet=config.subnet, - gateway=config.gateway, - ip_range=decky_range, - ) - setup_host_macvlan(config.interface, host_ip, decky_range) + 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, + subnet=config.subnet, + gateway=config.gateway, + ip_range=decky_range, + ) + setup_host_macvlan(config.interface, host_ip, decky_range) # --- Compose generation --- compose_path = write_compose(config, COMPOSE_FILE) @@ -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) - teardown_host_macvlan(decky_range) + 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: diff --git a/decnet/network.py b/decnet/network.py index cbdaac2..aa829d0 100644 --- a/decnet/network.py +++ b/decnet/network.py @@ -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 # --------------------------------------------------------------------------- diff --git a/tests/test_network.py b/tests/test_network.py new file mode 100644 index 0000000..8a71b35 --- /dev/null +++ b/tests/test_network.py @@ -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 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)