diff --git a/gns3server/compute/base_manager.py b/gns3server/compute/base_manager.py
index 5c1511e8e..4d29f2546 100644
--- a/gns3server/compute/base_manager.py
+++ b/gns3server/compute/base_manager.py
@@ -297,11 +297,12 @@ class BaseManager:
return node
@staticmethod
- def has_privileged_access(executable):
+ def has_privileged_access(executable, *, require_net_admin=False):
"""
Check if an executable have the right to attach to Ethernet and TAP adapters.
:param executable: executable path
+ :param require_net_admin: also require CAP_NET_ADMIN for network configuration
:returns: True or False
"""
@@ -324,8 +325,8 @@ class BaseManager:
try:
if sys.platform.startswith("linux") and "security.capability" in os.listxattr(executable):
caps = os.getxattr(executable, "security.capability")
- # test the 2nd byte and check if the 13th bit (CAP_NET_RAW) is set
- if struct.unpack(".
import logging
+import os
+import re
import subprocess
import sys
import gns3server.utils.asyncio
import gns3server.utils.interfaces
from gns3server.compute.ubridge.ubridge_error import UbridgeError
+from gns3server.schemas.compute.cloud_nodes import LINUX_TAP_NAME_PATTERN
from gns3server.utils.kernel_anchor import is_kernel_datapath_interface
from ...base_node import BaseNode
@@ -69,7 +72,7 @@ class Cloud(BaseNode):
for port in ports:
port["port_number"] = port_number
port_number += 1
- self._ports_mapping = ports
+ self._ports_mapping = self._normalize_ports_mapping(ports)
@property
def nios(self):
@@ -101,6 +104,9 @@ class Cloud(BaseNode):
}
)
+ if sys.platform.startswith("linux") and gns3server.utils.interfaces.is_interface_bridge(interface["name"]):
+ host_interfaces[-1]["is_bridge"] = True
+
return {
"name": self.name,
"usage": self.usage,
@@ -196,6 +202,11 @@ class Cloud(BaseNode):
self._remote_console_http_path = remote_console_http_path
+ @staticmethod
+ def _normalize_ports_mapping(ports):
+ # API responses include bridge=None; legacy projects omit it.
+ return [{key: value for key, value in port.items() if key != "bridge" or value is not None} for port in ports]
+
@property
def ports_mapping(self):
"""
@@ -214,6 +225,7 @@ class Cloud(BaseNode):
:param ports: ports info
"""
+ ports = self._normalize_ports_mapping(ports)
if ports != self._ports_mapping:
if len(self._nios) > 0:
raise NodeError("Cannot modify a cloud that is already connected.")
@@ -314,7 +326,9 @@ class Cloud(BaseNode):
await self._ubridge_apply_filters(bridge_name, nio.filters)
await self._ubridge_apply_markers(bridge_name, nio)
if port_info["type"] in ("ethernet", "tap"):
- if not self.manager.has_privileged_access(self.ubridge_path):
+ if not self.manager.has_privileged_access(
+ self.ubridge_path, require_net_admin=bool(port_info.get("bridge"))
+ ):
raise NodeError(
"uBridge requires root access or the capability to interact with Ethernet and TAP adapters"
)
@@ -334,9 +348,7 @@ class Cloud(BaseNode):
await self._add_windows_ethernet(port_info, bridge_name)
elif port_info["type"] == "tap":
- await self._ubridge_send(
- 'bridge add_nio_tap {name} "{interface}"'.format(name=bridge_name, interface=port_info["interface"])
- )
+ await self._add_tap_interface(port_info, bridge_name)
elif port_info["type"] == "udp":
await self._ubridge_send(
@@ -350,6 +362,32 @@ class Cloud(BaseNode):
await self._ubridge_send(f"bridge start {bridge_name}")
+ async def _add_tap_interface(self, port_info, bridge_name):
+ """Open a TAP and restore its optional host bridge assignment."""
+ target_bridge = port_info.get("bridge")
+ if target_bridge:
+ self._validate_tap_bridge(port_info["interface"], target_bridge)
+ await self._ubridge_send(
+ 'bridge add_nio_tap {name} "{interface}"'.format(name=bridge_name, interface=port_info["interface"])
+ )
+
+ if target_bridge:
+ await self._ubridge_send(f'brctl addif "{target_bridge}" "{port_info["interface"]}"')
+
+ def _validate_tap_bridge(self, tap, target_bridge):
+ """Validate host resources before creating or attaching a Cloud TAP."""
+ if not sys.platform.startswith("linux"):
+ raise NodeError("TAP bridge assignments are only supported on Linux")
+ if not all(re.fullmatch(LINUX_TAP_NAME_PATTERN, name) for name in (target_bridge, tap)):
+ raise NodeError("Invalid TAP or bridge interface name")
+ if target_bridge not in {
+ interface["name"] for interface in self._interfaces()
+ } or not gns3server.utils.interfaces.is_interface_bridge(target_bridge):
+ raise NodeError(f"Bridge '{target_bridge}' is missing or not allowed on this server")
+ master_path = os.path.join("/sys/class/net", tap, "master")
+ if os.path.exists(master_path) and os.path.basename(os.path.realpath(master_path)) != target_bridge:
+ raise NodeError(f"TAP '{tap}' already belongs to another bridge")
+
async def _add_linux_ethernet(self, port_info, bridge_name):
"""
Connects an Ethernet interface on Linux using raw sockets.
diff --git a/gns3server/schemas/compute/cloud_nodes.py b/gns3server/schemas/compute/cloud_nodes.py
index e2887bbec..274bf23ee 100644
--- a/gns3server/schemas/compute/cloud_nodes.py
+++ b/gns3server/schemas/compute/cloud_nodes.py
@@ -14,15 +14,18 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
+import re
from enum import Enum
from typing import List, Optional, Union
from uuid import UUID
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
from ..common import NodeStatus
from ..update import PartialUpdateModel
+LINUX_TAP_NAME_PATTERN = r"^[a-zA-Z0-9_.:-]{1,15}$"
+
class HostInterfaceType(str, Enum):
ethernet = "ethernet"
@@ -57,6 +60,7 @@ class HostInterface(BaseModel):
name: str = Field(..., description="Interface name")
type: HostInterfaceType = Field(..., description="Interface type")
special: bool = Field(..., description="Whether the interface is non standard")
+ is_bridge: bool = Field(False, description="Whether this is a Linux bridge")
ip_addresses: List[HostInterfaceIPAddress] = Field(
default_factory=list, description="All IPv4 and IPv6 addresses on this interface"
)
@@ -94,6 +98,19 @@ class TAPPort(BaseModel):
port_number: int
type: TAPType
interface: str
+ bridge: Optional[str] = Field(
+ None,
+ min_length=1,
+ max_length=15,
+ pattern=LINUX_TAP_NAME_PATTERN,
+ description="Existing Linux bridge to attach the TAP to",
+ )
+
+ @model_validator(mode="after")
+ def validate_bridged_interface(self):
+ if self.bridge and not re.fullmatch(LINUX_TAP_NAME_PATTERN, self.interface):
+ raise ValueError("A bridged TAP name must be 1-15 letters, digits, underscores, dots, colons or hyphens")
+ return self
class UDPType(str, Enum):
diff --git a/tests/api/routes/compute/test_cloud_nodes.py b/tests/api/routes/compute/test_cloud_nodes.py
index 0500356a8..fe1d58519 100644
--- a/tests/api/routes/compute/test_cloud_nodes.py
+++ b/tests/api/routes/compute/test_cloud_nodes.py
@@ -19,6 +19,7 @@ import pytest_asyncio
from fastapi import FastAPI, status
from httpx import AsyncClient
+from gns3server.compute.builtin import Builtin
from gns3server.compute.project import Project
from tests.utils import asyncio_patch
@@ -212,3 +213,36 @@ class TestCloudNodesRoutes:
# response = await compute_api.get("/projects/{project_id}/cloud/nodes/{node_id}/adapters/0/ports/0/pcap".format(project_id=compute_project.id, node_id=vm["node_id"]))
# assert response.status_code == 200
#
+
+ async def test_unbridged_connected_tap_update(self, app, compute_client, compute_project):
+ ports = [{"interface": "tap0", "name": "tap0", "port_number": 0, "type": "tap"}]
+ with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"):
+ response = await compute_client.post(
+ app.url_path_for("compute:create_cloud", project_id=compute_project.id),
+ json={"name": "Cloud TAP", "ports_mapping": ports},
+ )
+ assert response.status_code == 201
+ data = response.json()
+ node = Builtin.instance().get_node(data["node_id"], project_id=compute_project.id)
+ node.nios[0] = object()
+ response = await compute_client.put(
+ app.url_path_for("compute:update_cloud", project_id=compute_project.id, node_id=node.id),
+ json={"usage": "Updated usage", "ports_mapping": data["ports_mapping"]},
+ )
+ assert response.status_code == 200
+ assert response.json()["usage"] == "Updated usage"
+
+ @pytest.mark.parametrize("name", ["tap-interface-too-long", "tap bad", "/dev/tap0"])
+ async def test_invalid_bridged_tap_rejected_before_save(self, app, compute_client, compute_project, name):
+ with asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge") as start:
+ response = await compute_client.post(
+ app.url_path_for("compute:create_cloud", project_id=compute_project.id),
+ json={
+ "name": "Invalid TAP",
+ "ports_mapping": [
+ {"interface": name, "name": name, "port_number": 0, "type": "tap", "bridge": "virbr0"}
+ ],
+ },
+ )
+ assert response.status_code == 422
+ start.assert_not_called()
diff --git a/tests/compute/builtin/nodes/test_cloud.py b/tests/compute/builtin/nodes/test_cloud.py
index 180ab167d..89d1b7168 100644
--- a/tests/compute/builtin/nodes/test_cloud.py
+++ b/tests/compute/builtin/nodes/test_cloud.py
@@ -15,19 +15,34 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
+import struct
import uuid
from unittest.mock import MagicMock, call, patch
import pytest
import pytest_asyncio
+from fastapi.encoders import jsonable_encoder
+from pydantic import ValidationError
+from gns3server.compute.base_manager import BaseManager
from gns3server.compute.builtin.nodes.cloud import Cloud
+from gns3server.compute.error import NodeError
from gns3server.compute.nios.nio_udp import NIOUDP
+from gns3server.compute.ubridge.ubridge_error import UbridgeError
+from gns3server.schemas.compute.cloud_nodes import Cloud as CloudSchema
+from gns3server.schemas.compute.cloud_nodes import CloudUpdate, TAPPort
from tests.utils import asyncio_patch
pytestmark = pytest.mark.asyncio
+@pytest.fixture(autouse=True)
+def mock_bridge_detection():
+ # Interface inventories are mocked: never mix them with the host's sysfs.
+ with patch("gns3server.utils.interfaces.is_interface_bridge", return_value=False):
+ yield
+
+
@pytest.fixture
def nio():
@@ -256,3 +271,142 @@ async def test_linux_ethernet_raw_add_nio_bridge(linux_platform, compute_project
call(f"bridge start {cloud._id}-0"),
]
)
+
+
+@pytest.mark.parametrize("bridge", [None, "virbr0"])
+@pytest.mark.parametrize("existing_master", [False, True])
+async def test_tap_bridge_recreated(linux_platform, compute_project, manager, nio, bridge, existing_master):
+ ports = [{"interface": "tap0", "name": "tap0", "port_number": 0, "type": "tap"}]
+ if bridge:
+ ports[0]["bridge"] = bridge
+ cloud = Cloud("cloud1", str(uuid.uuid4()), compute_project, manager, ports=ports)
+ # Round-trip through the API schema, as happens when a project is restored.
+ restored_ports = CloudSchema(**cloud.asdict()).model_dump()["ports_mapping"]
+ restored = Cloud("cloud1", cloud.id, compute_project, manager, ports=restored_ports)
+ with (
+ asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._ubridge_send") as send,
+ asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._start_ubridge"),
+ patch.object(restored, "_interfaces", return_value=[{"name": "virbr0"}]),
+ patch("gns3server.utils.interfaces.is_interface_bridge", return_value=True),
+ patch("gns3server.compute.builtin.nodes.cloud.os.path.exists", return_value=existing_master),
+ patch("gns3server.compute.builtin.nodes.cloud.os.path.realpath", return_value="/sys/class/net/virbr0"),
+ ):
+ restored._nios[0] = nio
+ await restored.start()
+ tap_call = call(f'bridge add_nio_tap {cloud.id}-0 "tap0"')
+ assert tap_call in send.call_args_list
+ attach_call = call('brctl addif "virbr0" "tap0"')
+ if bridge:
+ assert send.call_args_list.index(tap_call) < send.call_args_list.index(attach_call)
+ else:
+ assert attach_call not in send.call_args_list
+
+
+@pytest.mark.parametrize("failure", ["missing", "not_bridge", "other_master", "invalid_name", "non_linux"])
+async def test_tap_bridge_validation(linux_platform, compute_project, manager, nio, failure):
+ ports = [{"interface": "tap0", "name": "tap0", "port_number": 0, "type": "tap", "bridge": "virbr0"}]
+ if failure == "invalid_name":
+ ports[0]["interface"] = 'tap0"\nbridge delete all'
+ cloud = Cloud("cloud1", str(uuid.uuid4()), compute_project, manager, ports=ports)
+ with (
+ asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._ubridge_send") as send,
+ patch.object(cloud, "_interfaces", return_value=[] if failure == "missing" else [{"name": "virbr0"}]),
+ patch("gns3server.utils.interfaces.is_interface_bridge", return_value=failure != "not_bridge"),
+ patch("gns3server.compute.builtin.nodes.cloud.os.path.exists", return_value=failure == "other_master"),
+ patch("gns3server.compute.builtin.nodes.cloud.os.path.realpath", return_value="/sys/class/net/another0"),
+ patch("gns3server.compute.builtin.nodes.cloud.sys.platform", "darwin" if failure == "non_linux" else "linux"),
+ ):
+ with pytest.raises(NodeError):
+ await cloud._add_ubridge_connection(nio, 0)
+ assert not any("add_nio_tap" in c.args[0] or "brctl addif" in c.args[0] for c in send.call_args_list)
+
+
+async def test_tap_bridge_attachment_failure(linux_platform, compute_project, manager, nio):
+ ports = [{"interface": "tap0", "name": "tap0", "port_number": 0, "type": "tap", "bridge": "virbr0"}]
+ cloud = Cloud("cloud1", str(uuid.uuid4()), compute_project, manager, ports=ports)
+ cloud.status = "started"
+
+ async def send(command):
+ if command.startswith("brctl addif"):
+ raise UbridgeError("Operation not permitted")
+
+ with (
+ asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._ubridge_send", side_effect=send),
+ asyncio_patch("gns3server.compute.builtin.nodes.cloud.Cloud._stop_ubridge") as stop,
+ patch.object(cloud, "_interfaces", return_value=[{"name": "virbr0"}]),
+ patch("gns3server.utils.interfaces.is_interface_bridge", return_value=True),
+ patch("gns3server.compute.builtin.nodes.cloud.os.path.exists", return_value=False),
+ ):
+ await cloud.add_nio(nio, 0)
+ stop.assert_called_once()
+ assert cloud.status == "stopped"
+
+
+async def test_cloud_lists_linux_bridges(linux_platform, compute_project, manager):
+ cloud = Cloud(
+ "cloud1",
+ str(uuid.uuid4()),
+ compute_project,
+ manager,
+ ports=[{"interface": "tap0", "name": "tap0", "port_number": 0, "type": "tap"}],
+ )
+ with (
+ patch(
+ "gns3server.utils.interfaces.interfaces",
+ return_value=[
+ {"name": "eth0", "type": "ethernet", "special": False},
+ {"name": "virbr0", "type": "ethernet", "special": True},
+ ],
+ ),
+ patch("gns3server.utils.interfaces.is_interface_bridge", side_effect=lambda name: name == "virbr0"),
+ ):
+ interfaces = CloudSchema(**cloud.asdict()).model_dump()["interfaces"]
+ assert [(interface["name"], interface["is_bridge"]) for interface in interfaces] == [
+ ("eth0", False),
+ ("virbr0", True),
+ ]
+
+
+@pytest.mark.parametrize("stored_bridge", [None, "omitted"])
+async def test_connected_tap_mapping_round_trip(compute_project, manager, stored_bridge):
+ port = {"interface": "tap0", "name": "tap0", "port_number": 0, "type": "tap"}
+ if stored_bridge is None:
+ port["bridge"] = None
+ cloud = Cloud("cloud1", str(uuid.uuid4()), compute_project, manager, ports=[port])
+ response = CloudSchema(**cloud.asdict()).model_dump(mode="json")
+ cloud._nios[0] = object()
+ ports = jsonable_encoder(CloudUpdate(ports_mapping=response["ports_mapping"]), exclude_unset=True)["ports_mapping"]
+ cloud.ports_mapping = ports
+ assert cloud.ports_mapping == [{"interface": "tap0", "name": "tap0", "port_number": 0, "type": "tap"}]
+ ports[0]["bridge"] = "virbr0"
+ with pytest.raises(NodeError, match="already connected"):
+ cloud.ports_mapping = ports
+
+
+@pytest.mark.parametrize("name", ["tap-interface-too-long", "tap bad", 'tap0"\nbridge delete all', "/dev/tap0"])
+async def test_bridged_tap_name_validation_preserves_legacy_names(name):
+ port = {"interface": name, "name": name, "port_number": 0, "type": "tap"}
+ assert TAPPort(**port).interface == name
+ with pytest.raises(ValidationError, match="bridged TAP name"):
+ TAPPort(**port, bridge="virbr0")
+
+
+@pytest.mark.parametrize(
+ "caps,legacy_allowed,bridge_allowed",
+ [
+ (0, False, False),
+ (1 << 12, False, False),
+ (1 << 13, True, False),
+ ((1 << 12) | (1 << 13), True, True),
+ ],
+)
+async def test_bridged_tap_requires_net_admin(linux_platform, tmp_path, caps, legacy_allowed, bridge_allowed):
+ executable = tmp_path / "ubridge"
+ executable.write_text("")
+ with (
+ patch("os.geteuid", return_value=1000),
+ patch("os.listxattr", return_value=["security.capability"]),
+ patch("os.getxattr", return_value=struct.pack("