gns3-server/tests/compute/qemu/test_qemu_kernel_datapath.py

#
# 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 <http://www.gnu.org/licenses/>.
#

"""
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")