# # Copyright (C) 2026 GNS3 Technologies Inc. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # """ Kernel-datapath wiring for QEMU: the persistent TAP every adapter owns, its netdev, kernel-link enslavement, relay attach on the anchor, capture, and the legacy socket-netdev fallback for a uBridge without the tap module. """ import os import stat import sys import pytest import pytest_asyncio from unittest.mock import MagicMock, patch from tests.utils import asyncio_patch, AsyncioMagicMock from gns3server.compute.qemu import Qemu from gns3server.compute.qemu.qemu_vm import QemuVM from gns3server.compute.qemu.qemu_error import QemuError from gns3server.compute.nios.nio_bridge import NIOBridge from gns3server.compute.ubridge.ubridge_error import UbridgeError BRIDGE = "gns3a1b2c3d4e5f" TAP0 = "gq00010203e0p0" # the vm fixture's node id, adapter 0 @pytest_asyncio.fixture async def manager(port_manager): m = Qemu.instance() m.port_manager = port_manager return m @pytest.fixture def fake_qemu_binary(monkeypatch, tmpdir): monkeypatch.setenv("PATH", str(tmpdir)) bin_path = os.path.join(os.environ["PATH"], "qemu-system-x86_64") with open(bin_path, "w+") as f: f.write("1") os.chmod(bin_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR) return bin_path @pytest_asyncio.fixture(scope="function") async def vm(compute_project, manager, fake_qemu_binary): vm = QemuVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager, qemu_path=fake_qemu_binary) vm._process_priority = "normal" vm._ubridge_hypervisor = MagicMock() vm._ubridge_hypervisor.is_running.return_value = True vm._ubridge_send = AsyncioMagicMock() return vm def _running(vm): """Make is_running() true without a real process.""" vm._process = MagicMock(returncode=None) # --------------------------------------------------------------------------- # TAP lifecycle # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_prepare_tap_datapath_creates_adapter_taps(vm): """ The tap module is probed once per start; on success every adapter gets its persistent TAP, owned by this user (so unprivileged QEMU can open it) and born down (carrier off until a link attaches). A leftover from a previous run is swept first. """ await vm._prepare_tap_datapath() assert vm._tap_datapath is True assert vm._kernel_taps == {(0, 0): TAP0} vm._ubridge_send.assert_any_call('tap create "gq00010203prob"') vm._ubridge_send.assert_any_call('tap delete "gq00010203prob"') vm._ubridge_send.assert_any_call(f'tap delete "{TAP0}"') vm._ubridge_send.assert_any_call(f'tap create "{TAP0}"') vm._ubridge_send.assert_any_call(f"tap set_owner {TAP0} {os.getuid()}") vm._ubridge_send.assert_any_call(f'link set "{TAP0}" down') @pytest.mark.asyncio async def test_prepare_tap_datapath_falls_back_without_the_tap_module(vm): """ A uBridge build without the tap module keeps the legacy socket-netdev datapath — the node still starts, it just cannot carry kernel links. """ async def send(command): if command.startswith("tap create"): raise UbridgeError("202-Unknown command 'create'") vm._ubridge_send = AsyncioMagicMock(side_effect=send) with patch.object(vm.project, "emit") as emit: await vm._prepare_tap_datapath() assert vm._tap_datapath is False assert vm._kernel_taps == {} assert emit.called @pytest.mark.asyncio async def test_tap_name_fits_ifnamsiz(vm): """ The anchor name has to fit IFNAMSIZ (15) even for a late adapter. """ vm.adapters = 32 assert len(vm._tap_name(31)) <= 15 @pytest.mark.asyncio async def test_network_options_use_the_tap_anchor(vm): """ On the TAP datapath the netdev is the adapter's anchor, not a UDP socket. """ vm._tap_datapath = True vm._kernel_taps[(0, 0)] = TAP0 options = await vm._network_options() assert f"tap,id=gns3-0,ifname={TAP0},script=no,downscript=no" in options assert not any(option.startswith("socket,") for option in options) assert not vm._local_udp_tunnels @pytest.mark.asyncio async def test_network_options_keep_the_legacy_socket_netdev(vm): """ Without the tap module nothing changes: the local UDP tunnel into uBridge is still the adapter's netdev. """ vm._tap_datapath = False options = await vm._network_options() assert any(option.startswith("socket,id=gns3-0,udp=") for option in options) assert vm._local_udp_tunnels # --------------------------------------------------------------------------- # Link attach / detach # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_connect_nio_kernel_link_enslaves_the_tap(vm): """ A kernel link enslaves the adapter's TAP into the per-link kernel bridge, with the bridge brought up first (a bridge that is down keeps its ports DISABLED and forwards nothing, silently). """ vm._tap_datapath = True vm._kernel_taps[(0, 0)] = TAP0 nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}) await vm._connect_nio(0, nio) vm._ubridge_send.assert_any_call(f'brctl create "{BRIDGE}"') vm._ubridge_send.assert_any_call(f'link set "{BRIDGE}" up') vm._ubridge_send.assert_any_call(f'brctl addif "{BRIDGE}" "{TAP0}"') @pytest.mark.asyncio async def test_connect_nio_relay_uses_the_tap_as_af_packet_endpoint(vm): """ A relay link attaches the TAP to the uBridge relay bridge via AF_PACKET (not the TAP NIO: that one takes the fd QEMU holds). """ vm._tap_datapath = True vm._kernel_taps[(0, 0)] = TAP0 nio = vm.manager.create_nio({"type": "nio_udp", "lport": 20000, "rhost": "127.0.0.1", "rport": 20001}) await vm._connect_nio(0, nio) bridge = f"QEMU-{vm.id}-0" vm._ubridge_send.assert_any_call(f"bridge create {bridge}") vm._ubridge_send.assert_any_call(f'link set "{TAP0}" up') vm._ubridge_send.assert_any_call(f'bridge add_nio_ethernet {bridge} "{TAP0}"') vm._ubridge_send.assert_any_call(f"bridge add_nio_udp {bridge} 20000 127.0.0.1 20001") vm._ubridge_send.assert_any_call(f"bridge start {bridge}") @pytest.mark.asyncio async def test_connect_nio_without_anchor_raises(vm): """ A kernel link on an adapter whose TAP does not exist (a VM started before the TAP datapath) fails with an actionable message. """ vm._tap_datapath = True nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}) with pytest.raises(QemuError, match="no TAP interface"): await vm._connect_nio(0, nio) @pytest.mark.asyncio async def test_legacy_datapath_rejects_kernel_link(vm): """ The legacy datapath has no anchor to enslave: a kernel link there is an error, not a silent fallback to the relay. """ vm._tap_datapath = False nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}) with pytest.raises(QemuError, match="legacy relay datapath"): await vm._connect_nio(0, nio) @pytest.mark.asyncio async def test_add_nio_binding_on_a_running_vm_attaches_and_brings_the_carrier_up(vm): _running(vm) vm._tap_datapath = True vm._kernel_taps[(0, 0)] = TAP0 nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}) await vm.adapter_add_nio_binding(0, nio) vm._ubridge_send.assert_any_call(f'brctl addif "{BRIDGE}" "{TAP0}"') vm._ubridge_send.assert_any_call(f'link set "{TAP0}" up') @pytest.mark.asyncio async def test_update_nio_binding_reapplies_filters_on_the_anchor(vm): """ A filter update on an attached kernel link reconciles on the anchor without re-enslaving it (no second addif). """ _running(vm) vm._tap_datapath = True vm._kernel_taps[(0, 0)] = TAP0 nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}) vm._ethernet_adapters[0].add_nio(0, nio) nio.filters = {"delay": [10]} await vm.adapter_update_nio_binding(0, nio) vm._ubridge_send.assert_any_call(f'tc reset "{TAP0}"') vm._ubridge_send.assert_any_call(f'tc netem set "{TAP0}" delay 10') assert not any(f"brctl addif" in str(c) for c in vm._ubridge_send.call_args_list) @pytest.mark.asyncio async def test_remove_nio_binding_detaches_the_anchor(vm): """ Removing a kernel link resets the anchor (no impairment survives on the adapter) and detaches it from the bridge. """ _running(vm) vm._tap_datapath = True vm._kernel_taps[(0, 0)] = TAP0 nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}) vm._ethernet_adapters[0].add_nio(0, nio) with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.stop_capture"): await vm.adapter_remove_nio_binding(0) vm._ubridge_send.assert_any_call(f'tc reset "{TAP0}"') vm._ubridge_send.assert_any_call(f'brctl delif "{BRIDGE}" "{TAP0}"') vm._ubridge_send.assert_any_call(f'brctl delete "{BRIDGE}"') @pytest.mark.asyncio async def test_stop_ubridge_removes_taps_and_orphan_bridges(vm): """ A stopped VM has no links attached, so the TAPs and any per-link bridge it still holds are removed while the uBridge control channel is up. """ vm._kernel_taps[(0, 0)] = TAP0 nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}) vm._ethernet_adapters[0].add_nio(0, nio) with asyncio_patch("gns3server.compute.base_node.BaseNode._stop_ubridge") as stop: await vm._stop_ubridge() vm._ubridge_send.assert_any_call(f'tap delete "{TAP0}"') vm._ubridge_send.assert_any_call(f'brctl delete "{BRIDGE}"') assert stop.called assert vm._kernel_taps == {} assert vm._ubridge_tc_caps is None # --------------------------------------------------------------------------- # Capture # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_capture_on_a_kernel_link_uses_the_anchor(vm): vm._tap_datapath = True vm._kernel_taps[(0, 0)] = TAP0 nio = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE}) vm._ethernet_adapters[0].add_nio(0, nio) await vm.start_capture(0, "/tmp/test.pcap") vm._ubridge_send.assert_any_call(f'capture start_kernel {TAP0} "/tmp/test.pcap"') await vm.stop_capture(0) vm._ubridge_send.assert_any_call("capture stop_kernel") @pytest.mark.asyncio async def test_capture_on_a_relay_link_uses_the_relay_bridge(vm): vm._tap_datapath = True vm._kernel_taps[(0, 0)] = TAP0 nio = vm.manager.create_nio({"type": "nio_udp", "lport": 20000, "rhost": "127.0.0.1", "rport": 20001}) vm._ethernet_adapters[0].add_nio(0, nio) await vm.start_capture(0, "/tmp/test.pcap") vm._ubridge_send.assert_any_call(f'bridge start_capture QEMU-{vm.id}-0 "/tmp/test.pcap"') await vm.stop_capture(0) vm._ubridge_send.assert_any_call(f"bridge stop_capture QEMU-{vm.id}-0") @pytest.mark.asyncio async def test_stop_stops_the_process_before_deleting_the_taps(vm): """ The QEMU process is the guest side of every adapter TAP, and uBridge's tap delete refuses a held device ("Device or resource busy") — that best-effort delete is suppressed, so removing the taps while QEMU still ran leaked the persistent TAPs. The process must be gone first (the same lesson as IOU's reverse stop order: process side, then the devices). """ _running(vm) order = [] async def fake_termination(process, timeout=None): order.append("process") async def fake_stop_ubridge(): order.append("taps") with ( patch("gns3server.utils.asyncio.wait_for_process_termination", new=fake_termination), patch.object(vm, "_stop_ubridge", new=fake_stop_ubridge), asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM._export_config"), asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM._clear_save_vm_stated"), asyncio_patch("gns3server.compute.base_node.BaseNode.stop"), ): await vm.stop() assert order == ["process", "taps"]