From 50191670981c92441a37806154101a0faf870ded Mon Sep 17 00:00:00 2001 From: grossmj Date: Wed, 16 Apr 2025 15:26:56 +0700 Subject: [PATCH] Fix interface information API endpoint for Cloud/NAT devices --- gns3server/api/routes/controller/links.py | 32 ++++++++--------- gns3server/schemas/__init__.py | 2 +- gns3server/schemas/controller/links.py | 7 ++-- tests/api/routes/controller/test_links.py | 43 +++++++++++------------ 4 files changed, 40 insertions(+), 44 deletions(-) diff --git a/gns3server/api/routes/controller/links.py b/gns3server/api/routes/controller/links.py index 8c438de2..67041eee 100644 --- a/gns3server/api/routes/controller/links.py +++ b/gns3server/api/routes/controller/links.py @@ -24,7 +24,7 @@ import aiohttp from fastapi import APIRouter, Depends, Request, status from fastapi.responses import StreamingResponse from fastapi.encoders import jsonable_encoder -from typing import List +from typing import List, Union from uuid import UUID from gns3server.controller import Controller @@ -289,15 +289,16 @@ async def stream_pcap(request: Request, link: Link = Depends(dep_link)) -> Strea @router.get( "/{link_id}/iface", - response_model=dict[str, schemas.UdpPort | schemas.EthernetPort], + response_model=Union[schemas.UDPPortInfo, schemas.EthernetPortInfo], dependencies=[Depends(has_privilege("Link.Audit"))] ) -async def get_iface(link: Link = Depends(dep_link)) -> dict[str, schemas.UdpPort | schemas.EthernetPort]: +async def get_iface(link: Link = Depends(dep_link)) -> Union[schemas.UDPPortInfo, schemas.EthernetPortInfo]: """ Return iface info for links to Cloud or NAT devices. Required privilege: Link.Audit """ + ifaces_info = {} for node_data in link._nodes: node = node_data["node"] @@ -305,17 +306,12 @@ async def get_iface(link: Link = Depends(dep_link)) -> dict[str, schemas.UdpPort continue port_number = node_data["port_number"] - compute = node.compute project_id = link.project.id - response = await compute.get( - f"/projects/{project_id}/{node.node_type}/nodes/{node.id}" - ) - node_info = response.json - - if "ports_mapping" not in node_info: + response = await compute.get(f"/projects/{project_id}/{node.node_type}/nodes/{node.id}") + if "ports_mapping" not in response.json: continue - ports_mapping = node_info["ports_mapping"] + ports_mapping = response.json["ports_mapping"] for port in ports_mapping: port_num = port.get("port_number") @@ -323,20 +319,20 @@ async def get_iface(link: Link = Depends(dep_link)) -> dict[str, schemas.UdpPort if port_num and int(port_num) == int(port_number): port_type = port.get("type", "") if "udp" in port_type.lower(): - ifaces_info[node.id] = { + ifaces_info = { + "node_id": node.id, "type": f"{port_type}", - "rhost": port["rhost"], "lport": port["lport"], - "rport": port["rport"], + "rhost": port["rhost"], + "rport": port["rport"] } else: - ifaces_info[node.id] = { + ifaces_info = { + "node_id": node.id, "type": f"{port_type}", "interface": port["interface"], } if not ifaces_info: - raise ControllerError( - "Link not connected to Cloud/NAT" - ) + raise ControllerError("Link not connected to Cloud/NAT") return ifaces_info diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py index a4fb297d..a23d61d9 100644 --- a/gns3server/schemas/__init__.py +++ b/gns3server/schemas/__init__.py @@ -20,7 +20,7 @@ from .common import ErrorMessage from .version import Version # Controller schemas -from .controller.links import LinkCreate, LinkUpdate, Link, UdpPort, EthernetPort +from .controller.links import LinkCreate, LinkUpdate, Link, UDPPortInfo, EthernetPortInfo from .controller.computes import ComputeCreate, ComputeUpdate, ComputeVirtualBoxVM, ComputeVMwareVM, ComputeDockerImage, AutoIdlePC, Compute from .controller.templates import TemplateCreate, TemplateUpdate, TemplateUsage, Template from .controller.images import Image, ImageType diff --git a/gns3server/schemas/controller/links.py b/gns3server/schemas/controller/links.py index ac82c0d7..cdcf6047 100644 --- a/gns3server/schemas/controller/links.py +++ b/gns3server/schemas/controller/links.py @@ -94,21 +94,22 @@ class Link(LinkBase): ) -class UdpPort(BaseModel): +class UDPPortInfo(BaseModel): """ UDP port information. """ + node_id: UUID lport: int rhost: str rport: int type: str - -class EthernetPort(BaseModel): +class EthernetPortInfo(BaseModel): """ Ethernet port information. """ + node_id: UUID interface: str type: str diff --git a/tests/api/routes/controller/test_links.py b/tests/api/routes/controller/test_links.py index a5aea4d5..5068fc8e 100644 --- a/tests/api/routes/controller/test_links.py +++ b/tests/api/routes/controller/test_links.py @@ -16,6 +16,7 @@ # along with this program. If not, see . import pytest +import uuid import pytest_asyncio from typing import Tuple @@ -424,16 +425,18 @@ class TestLinkRoutes: assert response.status_code == status.HTTP_200_OK assert response.json() == FILTERS + async def test_get_udp_interface(self, app: FastAPI, client: AsyncClient, project: Project) -> None: """ Test getting UDP tunnel interface information from a link. """ + link = Link(project) project._links = {link.id: link} cloud_node = MagicMock() cloud_node.node_type = "cloud" - cloud_node.id = "cloud-node-id" + cloud_node.id = str(uuid.uuid4()) cloud_node.name = "Cloud1" compute = MagicMock() @@ -456,29 +459,28 @@ class TestLinkRoutes: link._nodes = [{"node": cloud_node, "port_number": 1}] response = await client.get(app.url_path_for("get_iface", project_id=project.id, link_id=link.id)) - + assert response.status_code == status.HTTP_200_OK result = response.json() - - assert "cloud-node-id" in result - - udp_info = result["cloud-node-id"] - assert udp_info["lport"] == 20000 - assert udp_info["rhost"] == "127.0.0.1" - assert udp_info["rport"] == 30000 - assert udp_info["type"] == "udp" + assert result["node_id"] == cloud_node.id + assert result["lport"] == 20000 + assert result["rhost"] == "127.0.0.1" + assert result["rport"] == 30000 + assert result["type"] == "udp" + + async def test_get_ethernet_interface(self, app: FastAPI, client: AsyncClient, project: Project) -> None: """ Test getting ethernet interface information from a link. """ link = Link(project) project._links = {link.id: link} - + cloud_node = MagicMock() cloud_node.node_type = "cloud" - cloud_node.id = "cloud-node-id" + cloud_node.id = str(uuid.uuid4()) cloud_node.name = "Cloud1" - + compute = MagicMock() response = MagicMock() response.json = { @@ -493,16 +495,13 @@ class TestLinkRoutes: } compute.get = AsyncioMagicMock(return_value=response) cloud_node.compute = compute - + link._nodes = [{"node": cloud_node, "port_number": 1}] - + response = await client.get(app.url_path_for("get_iface", project_id=project.id, link_id=link.id)) - + assert response.status_code == status.HTTP_200_OK result = response.json() - - assert "cloud-node-id" in result - - interface_info = result["cloud-node-id"] - assert interface_info["interface"] == "eth0" - assert interface_info["type"] == "ethernet" \ No newline at end of file + assert result["node_id"] == cloud_node.id + assert result["interface"] == "eth0" + assert result["type"] == "ethernet" \ No newline at end of file