arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2025-04-19 17:19:00 +00:00
Code Issues Releases Wiki Activity

Fix interface information API endpoint for Cloud/NAT devices

Browse Source
This commit is contained in:
grossmj 2025-04-16 15:26:56 +07:00
parent f525916195
commit 5019167098
No known key found for this signature in database
GPG Key ID: 1E7DD6DBB53FF3D7
4 changed files with 40 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
gns3server/api/routes/controller/links.py
View File

@ -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

2
gns3server/schemas/__init__.py
View File

@ -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

7
gns3server/schemas/controller/links.py
View File

@ -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

43
tests/api/routes/controller/test_links.py
View File

@ -16,6 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
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"
assert result["node_id"] == cloud_node.id
assert result["interface"] == "eth0"
assert result["type"] == "ethernet"