NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-10-12 13:35:31 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch '3.1' into api-issue-10-0eztih

Browse Source
This commit is contained in:
Jeremy Grossmann 2026-10-08 19:21:07 +02:00 committed by GitHub
commit aeac649c7f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
58 changed files with 2455 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/testing.yml vendored
View File

@ -17,7 +17,7 @@ jobs:
strategy:
matrix:
python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
python-version: ["3.11", "3.12", "3.13", "3.14"]
steps:
- uses: actions/checkout@v7

7
gns3server/agent/web_wireshark/docker_client.py
View File

@ -23,16 +23,11 @@ This module handles Docker API operations via aiohttp Unix socket.
import asyncio
import logging
import sys
from asyncio import timeout as async_timeout
from typing import Any
import aiohttp
if sys.version_info >= (3, 11):
from asyncio import timeout as async_timeout
else:
from async_timeout import timeout as async_timeout
from gns3server.utils import parse_version
logger = logging.getLogger(__name__)

79
gns3server/api/errors.py Normal file
View File

@ -0,0 +1,79 @@
#!/usr/bin/env python
#
# 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/>.
"""
Error response helpers shared by the exception handlers and the OpenAPI schema.
"""
from collections.abc import Mapping
from http import HTTPStatus
from typing import Any
from fastapi.responses import JSONResponse
from gns3server.schemas.common import ErrorMessage
AUTHENTICATED_ERROR_RESPONSES = {
401: "Authentication is required or the credentials are invalid",
403: "Not enough privileges to perform this operation",
422: "Request validation error",
}
def code_for_status(status_code: int) -> str:
try:
phrase = HTTPStatus(status_code).phrase
except ValueError:
return "error"
return phrase.lower().replace(" ", "_").replace("-", "_")
def error_response(
status_code: int,
message: Any,
code: str | None = None,
details: dict[str, Any] | None = None,
headers: Mapping[str, str] | None = None,
) -> JSONResponse:
content: dict[str, Any] = {"message": message, "code": code or code_for_status(status_code)}
if details:
content["details"] = details
return JSONResponse(status_code=status_code, content=content, headers=headers)
def customize_openapi_errors(schema: dict[str, Any]) -> dict[str, Any]:
components = schema.setdefault("components", {}).setdefault("schemas", {})
components.pop("HTTPValidationError", None)
components.pop("ValidationError", None)
if "ErrorMessage" not in components:
error_schema = ErrorMessage.model_json_schema(ref_template="#/components/schemas/{model}")
components["ErrorMessage"] = {k: v for k, v in error_schema.items() if k != "$defs"}
for path_item in schema.get("paths", {}).values():
for operation in path_item.values():
if not isinstance(operation, dict) or "responses" not in operation:
continue
responses = operation["responses"]
wanted = ["422"] if "422" in responses else []
if operation.get("security"):
wanted = ["401", "403", "422"]
for status_code in wanted:
responses[status_code] = {
"description": AUTHENTICATED_ERROR_RESPONSES[int(status_code)],
"content": {"application/json": {"schema": {"$ref": "#/components/schemas/ErrorMessage"}}},
}
return schema

47
gns3server/api/openapi.py Normal file
View File

@ -0,0 +1,47 @@
#
# 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/>.
from typing import Any, Dict
BINARY_SCHEMA: Dict[str, Any] = {"type": "string", "format": "binary"}
NOTIFICATION_LINE_SCHEMA: Dict[str, Any] = {
"type": "object",
"description": "One notification event per line",
"properties": {
"action": {"type": "string"},
"event": {"description": "Event payload, its shape depends on the action"},
},
"required": ["action", "event"],
"additionalProperties": True,
}
PCAP_MEDIA_TYPE = "application/vnd.tcpdump.pcap"
NDJSON_MEDIA_TYPE = "application/x-ndjson"
OCTET_STREAM_MEDIA_TYPE = "application/octet-stream"
ZIP_MEDIA_TYPE = "application/zip"
def binary_response(media_type: str = OCTET_STREAM_MEDIA_TYPE, description: str = "Binary content") -> Dict[str, Any]:
return {"description": description, "content": {media_type: {"schema": BINARY_SCHEMA}}}
def ndjson_response(description: str = "Stream of newline-delimited JSON notifications") -> Dict[str, Any]:
return {"description": description, "content": {NDJSON_MEDIA_TYPE: {"schema": NOTIFICATION_LINE_SCHEMA}}}
def binary_request_body(media_type: str = OCTET_STREAM_MEDIA_TYPE) -> Dict[str, Any]:
return {"requestBody": {"required": True, "content": {media_type: {"schema": BINARY_SCHEMA}}}}

41
gns3server/api/responses.py Normal file
View File

@ -0,0 +1,41 @@
#
# 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/>.
from typing import Any
from fastapi.responses import JSONResponse
from gns3server import schemas
NDJSON_MEDIA_TYPE = "application/x-ndjson"
class NDJSONResponse(JSONResponse):
"""
Only used as the response_class of streaming routes to publish the application/x-ndjson media type
and the schema of the streamed objects in the OpenAPI document.
"""
media_type = NDJSON_MEDIA_TYPE
NOTIFICATION_STREAM_RESPONSES: dict[int | str, dict[str, Any]] = {
200: {
"model": schemas.Notification,
"description": "Stream of notifications, one JSON object per line. "
"Each line is a Notification, the first one is a ping and then a ping is sent at least every 5 seconds.",
}
}

4
gns3server/api/routes/compute/__init__.py
View File

@ -31,6 +31,7 @@ from gns3server.compute.compute_error import (
from gns3server.compute.error import ImageMissingError, NodeError
from gns3server.compute.ubridge.ubridge_error import UbridgeError
from gns3server.controller.gns3vm.gns3_vm_error import GNS3VMError
from gns3server.version import __api_version__
from . import (
atm_switch_nodes,
@ -60,7 +61,8 @@ log = logging.getLogger(__name__)
compute_api = FastAPI(
title="GNS3 compute API",
description="This page describes the private compute API for GNS3. PLEASE DO NOT USE DIRECTLY!",
version="3.0.0",
version=__api_version__,
servers=[{"url": "/v3/compute"}],
)
compute_api.state.controller_host = None

24
gns3server/api/routes/controller/__init__.py
View File

@ -18,6 +18,7 @@ from fastapi import APIRouter, Depends, HTTPException, status
# Check AI Copilot availability
from gns3server.agent import AI_COPILOT_AVAILABLE
from gns3server.config import Config
# Conditionally import AI-dependent routes
if AI_COPILOT_AVAILABLE:
@ -32,9 +33,11 @@ else:
_copilot_router = APIRouter()
_llm_router = APIRouter()
@_chat_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"])
@_copilot_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"])
@_llm_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"])
@_chat_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"], include_in_schema=False)
@_copilot_router.api_route(
"/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"], include_in_schema=False
)
@_llm_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"], include_in_schema=False)
async def ai_not_available(path: str = ""):
raise HTTPException(
status_code=status.HTTP_501_NOT_IMPLEMENTED,
@ -67,6 +70,8 @@ from . import (
)
from .dependencies.authentication import get_current_active_user
_include_ai_in_schema = Config.instance().settings.Server.openapi_include_ai
router = APIRouter()
router.include_router(controller.router, tags=["Controller"])
@ -116,11 +121,19 @@ router.include_router(
)
router.include_router(
_llm_router, prefix="/access", dependencies=[Depends(get_current_active_user)], tags=["LLM Model Configurations"]
_llm_router,
prefix="/access",
dependencies=[Depends(get_current_active_user)],
tags=["LLM Model Configurations"],
include_in_schema=_include_ai_in_schema,
)
router.include_router(
_copilot_router, prefix="/copilot", dependencies=[Depends(get_current_active_user)], tags=["GNS3 Copilot"]
_copilot_router,
prefix="/copilot",
dependencies=[Depends(get_current_active_user)],
tags=["GNS3 Copilot"],
include_in_schema=_include_ai_in_schema,
)
router.include_router(
@ -128,6 +141,7 @@ router.include_router(
prefix="/copilot/projects/{project_id}/chat",
dependencies=[Depends(get_current_active_user)],
tags=["GNS3 Copilot"],
include_in_schema=_include_ai_in_schema,
)
router.include_router(api_keys.router, dependencies=[Depends(get_current_active_user)], tags=["API Keys"])

8
gns3server/api/routes/controller/appliances.py
View File

@ -19,7 +19,7 @@ API routes for appliances.
"""
import logging
from typing import List, Optional, Union
from typing import List, Optional
from uuid import UUID
from fastapi import APIRouter, Depends, status
@ -87,9 +87,7 @@ def get_appliance(appliance_id: UUID) -> schemas.Appliance:
status_code=status.HTTP_201_CREATED,
dependencies=[Depends(has_privilege("Appliance.Allocate"))],
)
def add_appliance_version(
appliance_id: UUID, appliance_version: Union[schemas.ApplianceVersion, schemas.ApplianceVersionV8]
) -> dict:
def add_appliance_version(appliance_id: UUID, appliance_version: schemas.ApplianceVersionCreate) -> dict:
"""
Add a version to an appliance.
@ -111,7 +109,7 @@ def add_appliance_version(
if version.get("name") == appliance_version.name:
raise ControllerError(message=f"Appliance '{appliance_id}' already has version '{appliance_version.name}'")
appliance.versions.append(appliance_version.model_dump(exclude_unset=True))
appliance.versions.append(appliance_version.model_dump(exclude_unset=True, exclude={"registry_version"}))
return appliance.asdict()

19
gns3server/api/routes/controller/controller.py
View File

@ -30,6 +30,7 @@ from websockets.exceptions import ConnectionClosed, WebSocketException
from gns3server import schemas
from gns3server.agent.web_wireshark.stats import collect_webwireshark_stats
from gns3server.api.responses import NDJSON_MEDIA_TYPE, NOTIFICATION_STREAM_RESPONSES, NDJSONResponse
from gns3server.config import Config
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError, ControllerForbiddenError
@ -259,10 +260,19 @@ async def statistics() -> dict:
}
@router.get("/notifications", dependencies=[Depends(get_current_active_user)])
@router.get(
"/notifications",
dependencies=[Depends(get_current_active_user)],
response_class=NDJSONResponse,
responses=NOTIFICATION_STREAM_RESPONSES,
)
async def controller_http_notifications(request: Request) -> StreamingResponse:
"""
Receive controller notifications about the controller from HTTP stream.
Receive controller notifications from an HTTP stream of newline delimited JSON objects (see the Notification schema).
The same messages are available as WebSocket text frames on `/v3/notifications/ws`.
Controller actions: `compute.*`, `template.*`, `project.created`, `project.updated`, `project.opened`,
`project.closed`, `project.deleted`, `settings.updated`, `log.*` and `ping`.
"""
from gns3server.api.server import app
@ -279,7 +289,7 @@ async def controller_http_notifications(request: Request) -> StreamingResponse:
finally:
log.info(f"Client {client} has disconnected from controller HTTP notification stream")
return StreamingResponse(event_stream(), media_type="application/json")
return StreamingResponse(event_stream(), media_type=NDJSON_MEDIA_TYPE)
@router.websocket("/notifications/ws")
@ -287,7 +297,8 @@ async def controller_ws_notifications(
websocket: WebSocket, current_user: schemas.User = Depends(get_current_active_user_from_websocket)
) -> None:
"""
Receive project notifications about the controller from WebSocket.
Receive controller notifications from a WebSocket, one Notification JSON object per text frame.
Same messages as the HTTP stream `/v3/notifications`.
"""
if current_user is None:

7
gns3server/api/routes/controller/dependencies/authentication.py
View File

@ -36,7 +36,12 @@ from gns3server.services.access_tickets import TICKET_PREFIX
from .database import get_repository
log = logging.getLogger(__name__)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/v3/access/users/login", auto_error=False)
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="/v3/access/users/login",
auto_error=False,
description="Send the access token as a bearer token. "
"An API key (gns3_<api_key_id>_<secret>) is also accepted as the bearer token.",
)
def _reject_refresh_token(token_data) -> None:

115
gns3server/api/routes/controller/dependencies/concurrency.py Normal file
View File

@ -0,0 +1,115 @@
#
# 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/>.
"""
Optional optimistic concurrency (ETag / If-Match) for controller resources.
"""
import asyncio
import contextlib
import hashlib
import json
from typing import Any, AsyncIterator, Optional
from fastapi import Header, HTTPException, Response, status
from fastapi.encoders import jsonable_encoder
from gns3server import schemas
ETAG_RESPONSE_HEADERS: dict[str, Any] = {
"ETag": {
"description": "Entity tag of the current representation of the resource",
"schema": {"type": "string"},
}
}
PRECONDITION_FAILED_RESPONSE: dict[int | str, dict[str, Any]] = {
status.HTTP_412_PRECONDITION_FAILED: {
"model": schemas.ErrorMessage,
"description": "The If-Match header does not match the current ETag of the resource",
}
}
GET_RESPONSES: dict[int | str, dict[str, Any]] = {status.HTTP_200_OK: {"headers": ETAG_RESPONSE_HEADERS}}
PUT_RESPONSES: dict[int | str, dict[str, Any]] = {**GET_RESPONSES, **PRECONDITION_FAILED_RESPONSE}
_locks: dict[str, list[Any]] = {}
def compute_etag(resource: Any) -> str:
"""
Compute a strong entity tag from the serialized resource.
"""
payload = json.dumps(jsonable_encoder(resource), sort_keys=True, separators=(",", ":"), default=str)
return f'"{hashlib.sha256(payload.encode()).hexdigest()[:32]}"'
def set_etag(response: Response, resource: Any) -> None:
"""
Add the ETag header of the resource to the response.
"""
response.headers["ETag"] = compute_etag(resource)
def if_match_header(
if_match: Optional[str] = Header(
None,
alias="If-Match",
description="Only apply the update if this entity tag matches the current resource",
),
) -> Optional[str]:
"""
Dependency returning the raw If-Match request header.
"""
return if_match
def check_if_match(if_match: Optional[str], resource: Any) -> None:
"""
Raise a 412 error if If-Match is present and does not match the resource.
"""
if if_match is None:
return
candidates = [tag.strip() for tag in if_match.split(",")]
if "*" in candidates:
return
if compute_etag(resource) not in candidates:
raise HTTPException(
status_code=status.HTTP_412_PRECONDITION_FAILED,
detail="The resource was modified since it was last retrieved (If-Match does not match the current ETag)",
)
@contextlib.asynccontextmanager
async def serialize_updates(key: str) -> AsyncIterator[None]:
"""
Serialize updates of the same resource so that the If-Match check and the update are atomic.
"""
entry = _locks.setdefault(key, [asyncio.Lock(), 0])
entry[1] += 1
try:
async with entry[0]:
yield
finally:
entry[1] -= 1
if entry[1] <= 0:
_locks.pop(key, None)

2
gns3server/api/routes/controller/images.py
View File

@ -32,6 +32,7 @@ from starlette.requests import ClientDisconnect
import gns3server.db.models as models
from gns3server import schemas
from gns3server.api.openapi import binary_request_body
from gns3server.compute.qemu import Qemu
from gns3server.config import Config
from gns3server.controller import Controller
@ -214,6 +215,7 @@ async def get_images(
response_model=schemas.ImageUpload,
response_model_exclude_unset=True,
status_code=status.HTTP_201_CREATED,
openapi_extra=binary_request_body(),
dependencies=[Depends(has_privilege("Image.Allocate"))],
)
async def upload_image(

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

@ -21,24 +21,34 @@ API routes for links.
import logging
import os
from typing import Any, Dict, List, Union
from typing import Any, List, Optional
from uuid import UUID, uuid4
import aiohttp
import multidict
from fastapi import APIRouter, Depends, Request, WebSocket, status
from fastapi import APIRouter, Depends, Request, Response, WebSocket, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import FileResponse, StreamingResponse
from gns3server import schemas
from gns3server.agent.web_wireshark.manager import WebWiresharkManager
from gns3server.api.openapi import PCAP_MEDIA_TYPE, binary_response
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError
from gns3server.controller.controller_error import ControllerError, ControllerNotFoundError
from gns3server.controller.link import _UNSET, Link
from gns3server.db.repositories.rbac import RbacRepository
from gns3server.utils.http_client import HTTPClient
from gns3server.utils.port_allocator import link_id_to_port
from gns3server.utils.websocket_to_websocket import websocket_proxy
from .dependencies.concurrency import (
GET_RESPONSES,
PUT_RESPONSES,
check_if_match,
if_match_header,
serialize_updates,
set_etag,
)
from .dependencies.database import get_repository
from .dependencies.rbac import has_privilege, has_privilege_on_websocket
@ -51,6 +61,29 @@ responses: dict[int | str, dict[str, Any]] = {
router = APIRouter(responses=responses)
def _select_port(node, endpoint, other_port):
if endpoint.get("port_name") is not None:
port = node.get_port_by_name(endpoint["port_name"])
if port is None:
raise ControllerNotFoundError(f"Port named {endpoint['port_name']} for {node.name} not found")
return port
if endpoint.get("port") == "auto":
port = node.get_free_port(other_port.link_type if other_port else None)
if port is None:
raise ControllerError(f"No free port available on {node.name}", code="no_free_port")
return port
return None
def _peer_port(project, endpoint):
if endpoint.get("port") == "auto":
return None
node = project.get_node(endpoint["node_id"])
if endpoint.get("port_name") is not None:
return node.get_port_by_name(endpoint["port_name"])
return node.get_port(endpoint["adapter_number"], endpoint["port_number"])
async def dep_link(project_id: UUID, link_id: UUID) -> Link:
"""
Dependency to retrieve a link.
@ -110,14 +143,18 @@ async def create_link(project_id: UUID, link_create: schemas.LinkCreate) -> sche
if "show_filters_icon" in link_data:
await link.update_show_filters_icon(link_data["show_filters_icon"])
try:
for node in link_data["nodes"]:
await link.add_node(
project.get_node(node["node_id"]),
node.get("adapter_number", 0),
node.get("port_number", 0),
label=node.get("label"),
)
endpoints = link_data["nodes"]
attached_port = None
for index, endpoint in enumerate(endpoints):
node = project.get_node(endpoint["node_id"])
peer_port = attached_port or _peer_port(project, endpoints[1 - index])
port = _select_port(node, endpoint, peer_port)
adapter_number = port.adapter_number if port else endpoint["adapter_number"]
port_number = port.port_number if port else endpoint["port_number"]
await link.add_node(node, adapter_number, port_number, label=endpoint.get("label"))
attached_port = node.get_port(adapter_number, port_number)
except ControllerError as e:
link.release_ports()
await project.delete_link(link.id)
raise e
return link.asdict()
@ -143,43 +180,58 @@ async def get_filters(link: Link = Depends(dep_link)) -> List[dict]:
"/{link_id}",
response_model=schemas.Link,
response_model_exclude_unset=True,
responses=GET_RESPONSES,
dependencies=[Depends(has_privilege("Link.Audit"))],
)
async def get_link(link: Link = Depends(dep_link)) -> schemas.Link:
async def get_link(response: Response, link: Link = Depends(dep_link)) -> schemas.Link:
"""
Return a link.
Required privilege: Link.Audit
"""
return link.asdict()
link_dict = link.asdict()
set_etag(response, link_dict)
return link_dict
@router.put(
"/{link_id}",
response_model=schemas.Link,
response_model_exclude_unset=True,
responses=PUT_RESPONSES,
dependencies=[Depends(has_privilege("Link.Modify"))],
)
async def update_link(link_update: schemas.LinkUpdate, link: Link = Depends(dep_link)) -> schemas.Link:
async def update_link(
link_update: schemas.LinkUpdate,
response: Response,
link: Link = Depends(dep_link),
if_match: Optional[str] = Depends(if_match_header),
) -> schemas.Link:
"""
Update a link.
If the If-Match header is present, the update is only applied when it matches the current ETag.
Required privilege: Link.Modify
"""
link_data = jsonable_encoder(link_update, exclude_unset=True)
if "filters" in link_data:
await link.update_filters(link_data["filters"])
if "link_style" in link_data:
await link.update_link_style(link_data["link_style"])
if "suspend" in link_data:
await link.update_suspend(link_data["suspend"])
if "show_filters_icon" in link_data:
await link.update_show_filters_icon(link_data["show_filters_icon"])
if "nodes" in link_data:
await link.update_nodes(link_data["nodes"])
return link.asdict()
async with serialize_updates(f"link:{link.id}"):
check_if_match(if_match, link.asdict())
if "filters" in link_data:
await link.update_filters(link_data["filters"])
if "link_style" in link_data:
await link.update_link_style(link_data["link_style"])
if "suspend" in link_data:
await link.update_suspend(link_data["suspend"])
if "show_filters_icon" in link_data:
await link.update_show_filters_icon(link_data["show_filters_icon"])
if "nodes" in link_data:
await link.update_nodes(link_data["nodes"])
link_dict = link.asdict()
set_etag(response, link_dict)
return link_dict
@router.delete(
@ -281,7 +333,12 @@ async def restart_wireshark(http_request: Request, link: Link = Depends(dep_link
return {"status": "restarted"}
@router.get("/{link_id}/capture/stream", dependencies=[Depends(has_privilege("Link.Capture"))])
@router.get(
"/{link_id}/capture/stream",
response_class=StreamingResponse,
responses={200: binary_response(PCAP_MEDIA_TYPE, "Packet capture stream")},
dependencies=[Depends(has_privilege("Link.Capture"))],
)
async def stream_pcap(request: Request, link: Link = Depends(dep_link)) -> StreamingResponse:
"""
Stream the PCAP capture file from compute.
@ -504,7 +561,7 @@ async def update_marker(marker_name: str, marker_data: schemas.MarkerUpdate, lin
@router.get(
"/{link_id}/iface",
response_model=Union[schemas.UDPPortInfo, schemas.EthernetPortInfo],
response_model=schemas.LinkIfaceInfo,
dependencies=[Depends(has_privilege("Link.Audit"))],
)
async def get_iface(link: Link = Depends(dep_link)) -> dict:
@ -535,6 +592,7 @@ async def get_iface(link: Link = Depends(dep_link)) -> dict:
port_type = port.get("type", "")
if "udp" in port_type.lower():
ifaces_info = {
"kind": "udp",
"node_id": node.id,
"type": f"{port_type}",
"lport": port["lport"],
@ -543,6 +601,7 @@ async def get_iface(link: Link = Depends(dep_link)) -> dict:
}
else:
ifaces_info = {
"kind": "ethernet",
"node_id": node.id,
"type": f"{port_type}",
"interface": port["interface"],

162
gns3server/api/routes/controller/nodes.py
View File

@ -32,15 +32,28 @@ from fastapi.responses import StreamingResponse
from fastapi.routing import APIRoute
from gns3server import schemas
from gns3server.api.openapi import binary_request_body, binary_response
from gns3server.config import Config
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerBadRequestError, ControllerForbiddenError
from gns3server.controller.controller_error import (
ControllerBadRequestError,
ControllerForbiddenError,
ControllerTimeoutError,
)
from gns3server.controller.node import Node
from gns3server.controller.project import Project
from gns3server.db.repositories.rbac import RbacRepository
from gns3server.utils import force_unix_path
from gns3server.utils.http_client import HTTPClient
from .dependencies.concurrency import (
GET_RESPONSES,
PUT_RESPONSES,
check_if_match,
if_match_header,
serialize_updates,
set_etag,
)
from .dependencies.database import get_repository
from .dependencies.rbac import has_privilege, has_privilege_on_websocket
@ -256,8 +269,13 @@ async def reload_all_nodes(project: Project = Depends(dep_project)) -> None:
_HOST_INTERFACE_NODE_TYPES = {"cloud", "nat"}
@router.get("/{node_id}", response_model=schemas.Node, dependencies=[Depends(has_privilege("Node.Audit"))])
async def get_node(node: Node = Depends(dep_node)) -> schemas.Node:
@router.get(
"/{node_id}",
response_model=schemas.Node,
responses=GET_RESPONSES,
dependencies=[Depends(has_privilege("Node.Audit"))],
)
async def get_node(response: Response, node: Node = Depends(dep_node)) -> schemas.Node:
"""
Return a node from a given project.
@ -266,31 +284,50 @@ async def get_node(node: Node = Depends(dep_node)) -> schemas.Node:
if node.node_type in _HOST_INTERFACE_NODE_TYPES:
try:
response = await node.get()
await node.parse_node_response(response.json)
node_response = await node.get()
await node.parse_node_response(node_response.json)
except Exception:
# If compute is unreachable, still return cached data
log.warning(f"Could not refresh node {node.id} from compute, returning cached data")
return node.asdict()
node_dict = node.asdict()
set_etag(response, node_dict)
return node_dict
@router.put(
"/{node_id}",
response_model=schemas.Node,
response_model_exclude_unset=True,
responses=PUT_RESPONSES,
dependencies=[Depends(has_privilege("Node.Modify"))],
)
async def update_node(node_update: schemas.NodeUpdate, node: Node = Depends(dep_node)) -> schemas.Node:
async def update_node(
node_update: schemas.NodeUpdate,
response: Response,
node: Node = Depends(dep_node),
if_match: Optional[str] = Depends(if_match_header),
) -> schemas.Node:
"""
Update a node.
If the If-Match header is present, the update is only applied when it matches the current ETag.
Required privilege: Node.Modify
"""
node_data = jsonable_encoder(node_update, exclude_unset=True)
await node.update(**node_data)
return node.asdict()
# Ignore these because we only use them when creating a node
node_data.pop("node_id", None)
node_data.pop("node_type", None)
node_data.pop("compute_id", None)
async with serialize_updates(f"node:{node.id}"):
check_if_match(if_match, node.asdict())
await node.update(**node_data)
node_dict = node.asdict()
set_etag(response, node_dict)
return node_dict
@router.delete(
@ -331,46 +368,88 @@ async def duplicate_node(duplicate_data: schemas.NodeDuplicate, node: Node = Dep
return new_node.asdict()
LIFECYCLE_RESPONSES: dict[int | str, dict[str, Any]] = {
status.HTTP_408_REQUEST_TIMEOUT: {"model": schemas.ErrorMessage, "description": "Timeout waiting for the node"},
}
async def _run_lifecycle(
node: Node, action: Callable, target: str, wait: bool, timeout: int, tolerate_unsupported: bool = True
) -> dict:
async def _run():
try:
await action()
except HTTPException as e:
if not tolerate_unsupported or not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
raise
return
if wait and not node.is_always_running():
await node.wait_for_status(target)
if not wait:
await _run()
return node.asdict()
try:
await asyncio.wait_for(_run(), timeout)
except asyncio.TimeoutError:
raise ControllerTimeoutError(f"Timeout when waiting for {node.name} to be {target}")
return node.asdict()
@router.post(
"/{node_id}/start", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(has_privilege("Node.PowerMgmt"))]
"/{node_id}/start",
response_model=schemas.Node,
responses=LIFECYCLE_RESPONSES,
dependencies=[Depends(has_privilege("Node.PowerMgmt"))],
)
async def start_node(start_data: Optional[dict] = None, node: Node = Depends(dep_node)) -> None:
async def start_node(
start_data: Optional[dict] = None,
wait: bool = Query(False, description="Return only when the node status is 'started'"),
timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
node: Node = Depends(dep_node),
) -> dict:
"""
Start a node.
Required privilege: Node.PowerMgmt
"""
try:
await node.start(data=start_data)
except HTTPException as e:
if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
raise
return await _run_lifecycle(node, lambda: node.start(data=start_data), "started", wait, timeout)
@router.post(
"/{node_id}/stop", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(has_privilege("Node.PowerMgmt"))]
"/{node_id}/stop",
response_model=schemas.Node,
responses=LIFECYCLE_RESPONSES,
dependencies=[Depends(has_privilege("Node.PowerMgmt"))],
)
async def stop_node(node: Node = Depends(dep_node)) -> None:
async def stop_node(
wait: bool = Query(False, description="Return only when the node status is 'stopped'"),
timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
node: Node = Depends(dep_node),
) -> dict:
"""
Stop a node.
Errors reported while stopping are returned to the caller.
Required privilege: Node.PowerMgmt
"""
try:
await node.stop()
except HTTPException as e:
if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
raise
return await _run_lifecycle(node, lambda: node.stop(strict=True), "stopped", wait, timeout)
@router.post(
"/{node_id}/suspend",
status_code=status.HTTP_204_NO_CONTENT,
response_model=schemas.Node,
responses=LIFECYCLE_RESPONSES,
dependencies=[Depends(has_privilege("Node.PowerMgmt"))],
)
async def suspend_node(node: Node = Depends(dep_node)) -> None:
async def suspend_node(
wait: bool = Query(False, description="Return only when the node status is 'suspended'"),
timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
node: Node = Depends(dep_node),
) -> dict:
"""
Suspend a node.
@ -380,24 +459,27 @@ async def suspend_node(node: Node = Depends(dep_node)) -> None:
Required privilege: Node.PowerMgmt
"""
await node.suspend()
return await _run_lifecycle(node, node.suspend, "suspended", wait, timeout, tolerate_unsupported=False)
@router.post(
"/{node_id}/reload", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(has_privilege("Node.PowerMgmt"))]
"/{node_id}/reload",
response_model=schemas.Node,
responses=LIFECYCLE_RESPONSES,
dependencies=[Depends(has_privilege("Node.PowerMgmt"))],
)
async def reload_node(node: Node = Depends(dep_node)) -> None:
async def reload_node(
wait: bool = Query(False, description="Return only when the node status is 'started'"),
timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
node: Node = Depends(dep_node),
) -> dict:
"""
Reload a node.
Required privilege: Node.PowerMgmt
"""
try:
await node.reload()
except HTTPException as e:
if not e.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
raise
return await _run_lifecycle(node, node.reload, "started", wait, timeout)
@router.post(
@ -555,7 +637,12 @@ async def list_node_files(
return res.json
@router.get("/{node_id}/files/{file_path:path}", dependencies=[Depends(has_privilege("Node.Audit"))])
@router.get(
"/{node_id}/files/{file_path:path}",
response_class=StreamingResponse,
responses={200: binary_response(description="File content")},
dependencies=[Depends(has_privilege("Node.Audit"))],
)
async def get_file(file_path: str, node: Node = Depends(dep_node)) -> Response:
"""
Return a file from the node directory.
@ -596,6 +683,13 @@ async def get_file(file_path: str, node: Node = Depends(dep_node)) -> Response:
@router.post(
"/{node_id}/files/{file_path:path}",
status_code=status.HTTP_201_CREATED,
responses={
201: {
"description": "File written",
"content": {"application/json": {"schema": {"type": "null"}}},
}
},
openapi_extra=binary_request_body(),
dependencies=[Depends(has_privilege("Node.Modify"))],
)
async def post_file(file_path: str, request: Request, node: Node = Depends(dep_node)):

103
gns3server/api/routes/controller/projects.py
View File

@ -34,12 +34,25 @@ log = logging.getLogger()
from typing import Any, Dict, List, Optional
from uuid import UUID
from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request, WebSocket, WebSocketDisconnect, status
from fastapi import (
APIRouter,
Body,
Depends,
HTTPException,
Query,
Request,
Response,
WebSocket,
WebSocketDisconnect,
status,
)
from fastapi.encoders import jsonable_encoder
from fastapi.responses import FileResponse, StreamingResponse
from websockets.exceptions import ConnectionClosed, WebSocketException
from gns3server import schemas
from gns3server.api.openapi import ZIP_MEDIA_TYPE, binary_request_body, binary_response
from gns3server.api.responses import NDJSON_MEDIA_TYPE, NOTIFICATION_STREAM_RESPONSES, NDJSONResponse
from gns3server.controller import Controller, marker_replay
from gns3server.controller.controller_error import ControllerBadRequestError, ControllerError
from gns3server.controller.export_project import export_project as export_controller_project
@ -55,6 +68,14 @@ from gns3server.utils.asyncio import aiozipstream
from gns3server.utils.path import is_safe_path
from .dependencies.authentication import get_current_active_user
from .dependencies.concurrency import (
GET_RESPONSES,
PUT_RESPONSES,
check_if_match,
if_match_header,
serialize_updates,
set_etag,
)
from .dependencies.database import get_repository
from .dependencies.rbac import has_privilege, has_privilege_on_websocket
@ -150,34 +171,51 @@ async def create_project(
return project.asdict()
@router.get("/{project_id}", response_model=schemas.Project, dependencies=[Depends(has_privilege("Project.Audit"))])
def get_project(project: Project = Depends(dep_project)) -> schemas.Project:
@router.get(
"/{project_id}",
response_model=schemas.Project,
responses=GET_RESPONSES,
dependencies=[Depends(has_privilege("Project.Audit"))],
)
def get_project(response: Response, project: Project = Depends(dep_project)) -> schemas.Project:
"""
Return a project.
Required privilege: Project.Audit
"""
return project.asdict()
project_dict = project.asdict()
set_etag(response, project_dict)
return project_dict
@router.put(
"/{project_id}",
response_model=schemas.Project,
response_model_exclude_unset=True,
responses=PUT_RESPONSES,
dependencies=[Depends(has_privilege("Project.Modify"))],
)
async def update_project(
project_data: schemas.ProjectUpdate, project: Project = Depends(dep_project)
project_data: schemas.ProjectUpdate,
response: Response,
project: Project = Depends(dep_project),
if_match: Optional[str] = Depends(if_match_header),
) -> schemas.Project:
"""
Update a project.
If the If-Match header is present, the update is only applied when it matches the current ETag.
Required privilege: Project.Modify
"""
await project.update(**jsonable_encoder(project_data, exclude_unset=True))
return project.asdict()
async with serialize_updates(f"project:{project.id}"):
check_if_match(if_match, project.asdict())
await project.update(**jsonable_encoder(project_data, exclude_unset=True))
project_dict = project.asdict()
set_etag(response, project_dict)
return project_dict
@router.delete(
@ -554,10 +592,20 @@ async def load_project(path: str = Body(..., embed=True)) -> schemas.Project:
return project.asdict()
@router.get("/{project_id}/notifications", dependencies=[Depends(has_privilege("Project.Audit"))])
@router.get(
"/{project_id}/notifications",
dependencies=[Depends(has_privilege("Project.Audit"))],
response_class=NDJSONResponse,
responses=NOTIFICATION_STREAM_RESPONSES,
)
async def project_http_notifications(project_id: UUID) -> StreamingResponse:
"""
Receive project notifications about the controller from HTTP stream.
Receive project notifications from an HTTP stream of newline delimited JSON objects (see the Notification schema).
The same messages are available as WebSocket text frames on `/v3/projects/{project_id}/notifications/ws`.
Marker events (`marker.match`) are not sent here, they are only available on
`/v3/projects/{project_id}/notifications/markers/ws`.
Project actions: `node.*`, `link.*`, `drawing.*`, `project.updated`, `snapshot.restored`, `log.*` and `ping`.
Required privilege: Project.Audit
"""
@ -586,7 +634,7 @@ async def project_http_notifications(project_id: UUID) -> StreamingResponse:
log.info(f"Project '{project.id}' is automatically closing due to no client listening")
await project.close()
return StreamingResponse(event_stream(), media_type="application/json")
return StreamingResponse(event_stream(), media_type=NDJSON_MEDIA_TYPE)
@router.websocket("/{project_id}/notifications/ws")
@ -596,7 +644,8 @@ async def project_ws_notifications(
current_user: schemas.User = Depends(has_privilege_on_websocket("Project.Audit")),
) -> None:
"""
Receive project notifications about the controller from WebSocket.
Receive project notifications from a WebSocket, one Notification JSON object per text frame.
Same messages as the HTTP stream `/v3/projects/{project_id}/notifications`.
Required privilege: Project.Audit
"""
@ -634,8 +683,8 @@ async def project_marker_ws_notifications(
current_user: schemas.User = Depends(has_privilege_on_websocket("Project.Audit")),
) -> None:
"""
Receive marker notifications (e.g. marker.match) for a project on a
dedicated WebSocket, separate from the main project stream so high-frequency
Receive marker notifications (`marker.match`, see the Notification schema) for a project on a
dedicated WebSocket, one JSON object per text frame, separate from the main project stream so high-frequency
marker.matches do not block topology events (node.*/link.*).
Required privilege: Project.Audit
@ -663,7 +712,12 @@ async def project_marker_ws_notifications(
log.warning(f"Error while sending marker event to WebSocket client: {e}")
@router.get("/{project_id}/export", dependencies=[Depends(has_privilege("Project.Audit"))])
@router.get(
"/{project_id}/export",
response_class=StreamingResponse,
responses={200: binary_response(ZIP_MEDIA_TYPE, "Portable project archive")},
dependencies=[Depends(has_privilege("Project.Audit"))],
)
async def export_project(
project: Project = Depends(dep_project),
include_snapshots: bool = False,
@ -749,6 +803,7 @@ async def export_project(
"/{project_id}/import",
status_code=status.HTTP_201_CREATED,
response_model=schemas.Project,
openapi_extra=binary_request_body(ZIP_MEDIA_TYPE),
dependencies=[Depends(has_privilege("Project.Allocate"))],
)
async def import_project(project_id: UUID, request: Request, name: Optional[str] = None) -> schemas.Project:
@ -873,7 +928,12 @@ async def unlock_project(project: Project = Depends(dep_project)) -> None:
project.unlock()
@router.get("/{project_id}/files/{file_path:path}", dependencies=[Depends(has_privilege("Project.Audit"))])
@router.get(
"/{project_id}/files/{file_path:path}",
response_class=FileResponse,
responses={200: binary_response(description="File content")},
dependencies=[Depends(has_privilege("Project.Audit"))],
)
async def get_file(file_path: str, project: Project = Depends(dep_project)) -> FileResponse:
"""
Return a file from a project.
@ -895,7 +955,17 @@ async def get_file(file_path: str, project: Project = Depends(dep_project)) -> F
return FileResponse(path, media_type="application/octet-stream")
@router.get("/{project_id}/gns3file", dependencies=[Depends(has_privilege("Project.Audit"))])
@router.get(
"/{project_id}/gns3file",
response_class=FileResponse,
responses={
200: {
"description": "Project topology file",
"content": {"application/json": {"schema": {"type": "object"}}},
}
},
dependencies=[Depends(has_privilege("Project.Audit"))],
)
async def get_project_gns3_file(project: Project = Depends(dep_project)) -> FileResponse:
"""
Return the .gns3 topology file of a project.
@ -913,6 +983,7 @@ async def get_project_gns3_file(project: Project = Depends(dep_project)) -> File
@router.post(
"/{project_id}/files/{file_path:path}",
status_code=status.HTTP_204_NO_CONTENT,
openapi_extra=binary_request_body(),
dependencies=[Depends(has_privilege("Project.Modify"))],
)
async def write_file(file_path: str, request: Request, project: Project = Depends(dep_project)) -> None:

8
gns3server/api/routes/controller/symbols.py
View File

@ -27,6 +27,7 @@ from fastapi import APIRouter, Depends, Request, Response, status
from fastapi.responses import FileResponse
from gns3server import schemas
from gns3server.api.openapi import binary_request_body, binary_response
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError, ControllerForbiddenError, ControllerNotFoundError
from gns3server.utils.get_resource import get_resource
@ -53,7 +54,11 @@ def get_symbols() -> List[dict]:
@router.get(
"/{symbol_id:path}/raw",
responses={404: {"model": schemas.ErrorMessage, "description": "Could not find symbol"}},
response_class=FileResponse,
responses={
200: binary_response(description="Symbol file"),
404: {"model": schemas.ErrorMessage, "description": "Could not find symbol"},
},
dependencies=[Depends(has_privilege("Symbol.Audit"))],
)
async def get_symbol(symbol_id: str, request: Request) -> Response:
@ -117,6 +122,7 @@ def get_default_symbols() -> dict:
@router.post(
"/{symbol_id:path}/raw",
status_code=status.HTTP_204_NO_CONTENT,
openapi_extra=binary_request_body(),
dependencies=[Depends(has_privilege("Symbol.Allocate"))],
)
async def upload_symbol(symbol_id: str, request: Request) -> None:

20
gns3server/api/routes/controller/users.py
View File

@ -65,12 +65,7 @@ async def login(
headers={"WWW-Authenticate": "Bearer"},
)
token = schemas.Token(
access_token=auth_service.create_access_token(user.username, token_version=user.token_version),
token_type="bearer",
refresh_token=auth_service.create_refresh_token(user.username, token_version=user.token_version),
)
return token
return auth_service.create_token_response(user)
@router.post("/authenticate", response_model=schemas.Token)
@ -91,12 +86,7 @@ async def authenticate(
headers={"WWW-Authenticate": "Bearer"},
)
token = schemas.Token(
access_token=auth_service.create_access_token(user.username, token_version=user.token_version),
token_type="bearer",
refresh_token=auth_service.create_refresh_token(user.username, token_version=user.token_version),
)
return token
return auth_service.create_token_response(user)
@router.post("/refresh", response_model=schemas.Token)
@ -135,11 +125,7 @@ async def refresh_access_token(
headers={"WWW-Authenticate": "Bearer"},
)
return schemas.Token(
access_token=auth_service.create_access_token(user.username, token_version=user.token_version),
token_type="bearer",
refresh_token=auth_service.create_refresh_token(user.username, token_version=user.token_version),
)
return auth_service.create_token_response(user)
@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)

83
gns3server/api/server.py
View File

@ -19,7 +19,7 @@
FastAPI app
"""
from typing import cast
from typing import Any, cast
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.exceptions import RequestValidationError
@ -29,15 +29,17 @@ from fastapi.openapi.docs import (
get_swagger_ui_html,
get_swagger_ui_oauth2_redirect_html,
)
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from sqlalchemy.exc import SQLAlchemyError
from starlette.exceptions import HTTPException as StarletteHTTPException
from uvicorn.main import Server as UvicornServer
# MCP is an optional feature — import only if dependencies are installed
from gns3server.agent import MCP_AVAILABLE
from gns3server.api.errors import customize_openapi_errors, error_response
from gns3server.api.routes import controller, index
from gns3server.api.routes.compute import compute_api
from gns3server.config import Config
from gns3server.controller.controller_error import (
ComputeConflictError,
ControllerBadRequestError,
@ -48,6 +50,7 @@ from gns3server.controller.controller_error import (
ControllerUnauthorizedError,
)
from gns3server.core import tasks
from gns3server.version import __api_version__
if MCP_AVAILABLE:
from gns3server.agent import mcp
@ -58,7 +61,7 @@ else:
_mcp_router = APIRouter(prefix="/mcp", tags=["MCP"])
@_mcp_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"])
@_mcp_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"], include_in_schema=False)
async def mcp_not_available(path: str = ""):
raise HTTPException(
status_code=status.HTTP_501_NOT_IMPLEMENTED,
@ -71,13 +74,21 @@ import logging
log = logging.getLogger(__name__)
class ControllerAPI(FastAPI):
def openapi(self) -> dict[str, Any]:
if self.openapi_schema is None:
customize_openapi_errors(super().openapi())
return cast(dict[str, Any], self.openapi_schema)
def get_application() -> FastAPI:
application = FastAPI(
application = ControllerAPI(
lifespan=tasks.lifespan,
title="GNS3 controller API",
description="This page describes the public controller API for GNS3",
version="3.0.0",
version=__api_version__,
servers=[{"url": "/"}],
docs_url=None,
redoc_url=None,
)
@ -96,7 +107,12 @@ def get_application() -> FastAPI:
application.mount("/v3/compute", compute_api, name="compute")
# Register MCP routes (stub returns 501 if MCP dependencies are not installed)
application.include_router(_mcp_router, prefix="/v3", tags=["MCP"])
application.include_router(
_mcp_router,
prefix="/v3",
tags=["MCP"],
include_in_schema=Config.instance().settings.Server.openapi_include_ai,
)
return application
@ -151,82 +167,65 @@ async def redoc_html():
@app.exception_handler(ControllerError)
async def controller_error_handler(request: Request, exc: ControllerError):
log.error(f"Controller error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content={"message": str(exc)},
)
return error_response(status.HTTP_409_CONFLICT, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerTimeoutError)
async def controller_timeout_error_handler(request: Request, exc: ControllerTimeoutError):
log.error(f"Controller timeout error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_408_REQUEST_TIMEOUT,
content={"message": str(exc)},
)
return error_response(status.HTTP_408_REQUEST_TIMEOUT, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerUnauthorizedError)
async def controller_unauthorized_error_handler(request: Request, exc: ControllerUnauthorizedError):
log.error(f"Controller unauthorized error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"message": str(exc)},
)
return error_response(status.HTTP_401_UNAUTHORIZED, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerForbiddenError)
async def controller_forbidden_error_handler(request: Request, exc: ControllerForbiddenError):
log.error(f"Controller forbidden error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_403_FORBIDDEN,
content={"message": str(exc)},
)
return error_response(status.HTTP_403_FORBIDDEN, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerNotFoundError)
async def controller_not_found_error_handler(request: Request, exc: ControllerNotFoundError):
log.error(f"Controller not found error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={"message": str(exc)},
)
return error_response(status.HTTP_404_NOT_FOUND, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerBadRequestError)
async def controller_bad_request_error_handler(request: Request, exc: ControllerBadRequestError):
log.error(f"Controller bad request error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={"message": str(exc)},
)
return error_response(status.HTTP_400_BAD_REQUEST, str(exc), exc.code, exc.details)
@app.exception_handler(ComputeConflictError)
async def compute_conflict_error_handler(request: Request, exc: ComputeConflictError):
log.error(f"Controller received error from compute for request '{exc.url()}': {exc}")
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content={"message": str(exc)},
)
return error_response(status.HTTP_409_CONFLICT, str(exc), exc.code, exc.details)
# make sure the content key is "message", not "detail" per default
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(status_code=exc.status_code, content={"message": exc.detail}, headers=exc.headers)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return error_response(exc.status_code, exc.detail, headers=exc.headers)
@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_error_handler(request: Request, exc: SQLAlchemyError):
log.error(f"Controller database error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"message": "Database error detected, please check logs to find details"},
return error_response(
status.HTTP_500_INTERNAL_SERVER_ERROR,
"Database error detected, please check logs to find details",
"database_error",
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
log.error(f"Request validation error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, content={"message": str(exc)})
errors = [
{"loc": list(error.get("loc", ())), "msg": error.get("msg"), "type": error.get("type")}
for error in exc.errors()
]
return error_response(status.HTTP_422_UNPROCESSABLE_CONTENT, str(exc), "validation_error", {"errors": errors})

12
gns3server/compute/base_node.py
View File

@ -671,11 +671,7 @@ class BaseNode:
)
# keep forwarding websocket data in both direction
if sys.version_info >= (3, 11, 0):
# Starting with Python 3.11, passing coroutine objects to wait() directly is forbidden.
aws = [asyncio.create_task(ws_forward(telnet_writer)), asyncio.create_task(telnet_forward(telnet_reader))]
else:
aws = [ws_forward(telnet_writer), telnet_forward(telnet_reader)]
aws = [asyncio.create_task(ws_forward(telnet_writer)), asyncio.create_task(telnet_forward(telnet_reader))]
done, pending = await asyncio.wait(aws, return_when=asyncio.FIRST_COMPLETED)
for task in done:
@ -762,11 +758,7 @@ class BaseNode:
log.warning(f"Exception while forwarding VNC data to WebSocket: {e!r}")
# Keep forwarding WebSocket data in both directions
if sys.version_info >= (3, 11, 0):
# Starting with Python 3.11, passing coroutine objects to wait() directly is forbidden.
aws = [asyncio.create_task(ws_forward(vnc_writer)), asyncio.create_task(vnc_forward(vnc_reader))]
else:
aws = [ws_forward(vnc_writer), vnc_forward(vnc_reader)]
aws = [asyncio.create_task(ws_forward(vnc_writer)), asyncio.create_task(vnc_forward(vnc_reader))]
done, pending = await asyncio.wait(aws, return_when=asyncio.FIRST_COMPLETED)
for task in done:

10
gns3server/compute/project.py
View File

@ -480,11 +480,13 @@ class Project:
stat_info = await wait_run_in_executor(lambda e=entry: e.stat())
is_dir = await wait_run_in_executor(lambda e=entry: e.is_dir())
if is_dir:
created_at: str | None = None
modified_at: str | None = None
try:
created_at = datetime.datetime.fromtimestamp(stat_info.st_ctime).isoformat()
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError):
created_at = modified_at = ""
created_at = modified_at = None
files.append(
{
"path": rel_path,
@ -506,7 +508,7 @@ class Project:
created_at = datetime.datetime.fromtimestamp(stat_info.st_ctime).isoformat()
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError):
created_at = modified_at = ""
created_at = modified_at = None
files.append(
{
"path": rel_path,
@ -537,7 +539,7 @@ class Project:
created_at = datetime.datetime.fromtimestamp(stat_info.st_ctime).isoformat()
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError):
created_at = modified_at = ""
created_at = modified_at = None
files.append(
{
"path": os.path.relpath(dir_full_path, base_path),
@ -567,7 +569,7 @@ class Project:
modified_at = datetime.datetime.fromtimestamp(stat_info.st_mtime).isoformat()
except (OSError, OverflowError, ValueError) as e:
log.warning(f"Invalid timestamp for '{rel_path}': {e}")
created_at = modified_at = ""
created_at = modified_at = None
files.append(
{
"path": rel_path,

8
gns3server/controller/__init__.py
View File

@ -710,7 +710,9 @@ class Controller:
except KeyError:
if compute_id == "vm":
raise ControllerNotFoundError("Cannot use a node on the GNS3 VM server with the GNS3 VM not configured")
raise ControllerNotFoundError(f"Compute ID {compute_id} doesn't exist")
raise ControllerNotFoundError(
f"Compute ID {compute_id} doesn't exist", code="compute_not_found", details={"compute_id": compute_id}
)
def has_compute(self, compute_id):
"""
@ -748,7 +750,9 @@ class Controller:
try:
return self._projects[project_id]
except KeyError:
raise ControllerNotFoundError(f"Project ID {project_id} doesn't exist")
raise ControllerNotFoundError(
f"Project ID {project_id} doesn't exist", code="project_not_found", details={"project_id": project_id}
)
async def get_loaded_project(self, project_id):
"""

9
gns3server/controller/compute.py
View File

@ -19,19 +19,14 @@ import asyncio
import io
import ipaddress
import json
import logging
import socket
import sys
from asyncio import timeout as asynctimeout
import aiohttp
from fastapi import HTTPException
if sys.version_info >= (3, 11):
from asyncio import timeout as asynctimeout
else:
from async_timeout import timeout as asynctimeout
import logging
from ..controller.controller_error import (
ComputeConflictError,
ComputeError,

35
gns3server/controller/controller_error.py
View File

@ -15,11 +15,17 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from typing import Any
class ControllerError(Exception):
def __init__(self, message: str):
default_code = "conflict"
def __init__(self, message: str, code: str | None = None, details: dict[str, Any] | None = None):
super().__init__()
self._message = message
self._code = code or self.default_code
self._details = details
def __repr__(self):
return self._message
@ -27,30 +33,33 @@ class ControllerError(Exception):
def __str__(self):
return self._message
@property
def code(self) -> str:
return self._code
@property
def details(self) -> dict[str, Any] | None:
return self._details
class ControllerNotFoundError(ControllerError):
def __init__(self, message: str):
super().__init__(message)
default_code = "not_found"
class ControllerBadRequestError(ControllerError):
def __init__(self, message: str):
super().__init__(message)
default_code = "bad_request"
class ControllerUnauthorizedError(ControllerError):
def __init__(self, message: str):
super().__init__(message)
default_code = "unauthorized"
class ControllerForbiddenError(ControllerError):
def __init__(self, message: str):
super().__init__(message)
default_code = "forbidden"
class ControllerTimeoutError(ControllerError):
def __init__(self, message: str):
super().__init__(message)
default_code = "timeout"
class ComputeError(ControllerError):
@ -65,8 +74,10 @@ class ComputeConflictError(ComputeError):
:param response: compute JSON response
"""
default_code = "compute_conflict"
def __init__(self, url, response):
super().__init__(response["message"])
super().__init__(response["message"], code=response.get("code"), details=response.get("details"))
self._url = url
self._response = response

39
gns3server/controller/link.py
View File

@ -274,9 +274,22 @@ class Link:
port = node.get_port(adapter_number, port_number)
if port is None:
raise ControllerNotFoundError(f"Port {adapter_number}/{port_number} for {node.name} not found")
raise ControllerNotFoundError(
f"Port {adapter_number}/{port_number} for {node.name} not found",
code="port_not_found",
details={"node_id": node.id, "adapter_number": adapter_number, "port_number": port_number},
)
if port.link is not None:
raise ControllerError("Port is already used")
raise ControllerError(
"Port is already used",
code="port_in_use",
details={
"node_id": node.id,
"adapter_number": adapter_number,
"port_number": port_number,
"link_id": port.link.id,
},
)
self._link_type = port.link_type
@ -313,17 +326,29 @@ class Link:
{"node": node, "adapter_number": adapter_number, "port_number": port_number, "port": port, "label": label}
)
if not batch:
port.link = self
if len(self._nodes) == 2 and not batch:
await self.create()
try:
await self.create()
except BaseException:
self.release_ports()
raise
for n in self._nodes:
n["node"].add_link(self)
n["port"].link = self
self._created = True
self._project.emit_notification("link.created", self.asdict())
if dump:
self._project.dump()
def release_ports(self):
for n in self._nodes:
if n["port"].link == self:
n["port"].link = None
n["node"].links.discard(self)
async def update_nodes(self, nodes):
for node_data in nodes:
node = self._project.get_node(node_data["node_id"])
@ -352,11 +377,7 @@ class Link:
"""
Delete the link
"""
for n in self._nodes:
# It could be different of self if we rollback an already existing link
if n["port"].link == self:
n["port"].link = None
n["node"].remove_link(self)
self.release_ports()
async def reset(self):
"""

120
gns3server/controller/node.py
View File

@ -16,6 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio
import contextlib
import copy
import html
import logging
@ -134,6 +135,8 @@ class Node:
self._command_line = None
self._node_directory = None
self._status = "stopped"
self._status_changed = asyncio.Event()
self._transition = None
self._template_id = template_id
self._x = 0
self._y = 0
@ -785,7 +788,7 @@ class Node:
elif key == "command_line":
self._command_line = value
elif key == "status":
self._status = value
self._set_status(value)
elif key == "console_type":
self._console_type = value
elif key == "aux_type":
@ -901,6 +904,30 @@ class Node:
async def destroy(self):
await self.delete()
@contextlib.asynccontextmanager
async def _transitioning(self, transition):
"""
Report a transitional status while a lifecycle call is in flight
"""
self._transition = transition
self.project.emit_notification("node.updated", self.asdict())
try:
yield
finally:
self._transition = None
async def wait_for_status(self, status):
"""
Wait until the controller side status of the node matches
:param status: expected status
"""
while self._status != status:
self._status_changed.clear()
await self._status_changed.wait()
async def start(self, data=None):
"""
Start a node
@ -924,32 +951,56 @@ class Node:
f"Cannot start node '{self._name}': {len(failed_links)} deferred link(s) "
"could not be restored. Please try again."
)
try:
# For IOU: we need to send the licence everytime we start a node
if self.node_type == "iou":
license_check = self._project.controller.iou_license.get("license_check", True)
iourc_content = self._project.controller.iou_license.get("iourc_content", None)
await self.post(
"/start", timeout=240, data={"license_check": license_check, "iourc_content": iourc_content}
)
else:
await self.post("/start", data=data, timeout=240)
except asyncio.TimeoutError:
raise ControllerTimeoutError(f"Timeout when starting {self._name}")
async with self._transitioning("starting"):
try:
# For IOU: we need to send the licence everytime we start a node
if self.node_type == "iou":
license_check = self._project.controller.iou_license.get("license_check", True)
iourc_content = self._project.controller.iou_license.get("iourc_content", None)
await self.post(
"/start", timeout=240, data={"license_check": license_check, "iourc_content": iourc_content}
)
else:
await self.post("/start", data=data, timeout=240)
except asyncio.TimeoutError:
raise ControllerTimeoutError(f"Timeout when starting {self._name}")
self._confirm_status("started")
async def stop(self):
async def stop(self, strict=False):
"""
Stop a node
:param strict: Raise the errors reported by the compute instead of ignoring them
"""
if self.missing_image:
return
try:
await self.post("/stop", timeout=240, dont_connect=True)
# We don't care if a node is down at this step
except (ComputeError, ControllerError):
pass
except asyncio.TimeoutError:
raise ControllerTimeoutError(f"Timeout when stopping {self._name}")
stopped = False
async with self._transitioning("stopping"):
try:
await self.post("/stop", timeout=240, dont_connect=True)
stopped = True
except (ComputeError, ControllerError):
# We don't care if a node is down at this step
if strict:
raise
except asyncio.TimeoutError:
raise ControllerTimeoutError(f"Timeout when stopping {self._name}")
if stopped:
self._confirm_status("stopped")
def _set_status(self, status):
self._status = status
self._status_changed.set()
def _confirm_status(self, status):
"""
Record the state reached once the compute accepted a lifecycle call,
without waiting for the asynchronous compute notification
"""
if not self.is_always_running():
self._set_status(status)
self.project.emit_notification("node.updated", self.asdict())
async def suspend(self):
"""
@ -958,7 +1009,8 @@ class Node:
try:
await self.post("/suspend", timeout=240)
except asyncio.TimeoutError:
raise ControllerTimeoutError(f"Timeout when reloading {self._name}")
raise ControllerTimeoutError(f"Timeout when suspending {self._name}")
self._confirm_status("suspended")
async def reload(self):
"""
@ -968,6 +1020,7 @@ class Node:
await self.post("/reload", timeout=240)
except asyncio.TimeoutError:
raise ControllerTimeoutError(f"Timeout when reloading {self._name}")
self._confirm_status("started")
async def reset_console(self):
"""
@ -1142,6 +1195,27 @@ class Node:
return port
return None
def get_port_by_name(self, name):
"""
Return the port with this name or None if the port is not found
"""
for port in self.ports:
if port.name == name:
return port
return None
def get_free_port(self, link_type=None):
"""
Return the lowest free port, optionally restricted to a link type,
or None if there is no such port
"""
free_ports = [
port for port in self.ports if port.link is None and (link_type is None or port.link_type == link_type)
]
if not free_ports:
return None
return min(free_ports, key=lambda port: (port.adapter_number, port.port_number))
def _list_ports(self):
"""
Generate the list of port display in the client
@ -1280,7 +1354,7 @@ class Node:
additional_data = {
"project_id": self._project.id,
"command_line": self._command_line,
"status": self._status,
"status": self._transition or self._status,
"console_host": str(self._compute.console_host),
"node_directory": self._node_directory,
"ports": [port.asdict() for port in self.ports],

4
gns3server/controller/ports/port.py
View File

@ -41,6 +41,10 @@ class Port:
def link(self, val):
self._link = val
@property
def name(self):
return self._name
@property
def adapter_number(self):
return self._adapter_number

18
gns3server/controller/project.py
View File

@ -748,7 +748,9 @@ class Project:
try:
return self._nodes[node_id]
except KeyError:
raise ControllerNotFoundError(f"Node ID {node_id} doesn't exist")
raise ControllerNotFoundError(
f"Node ID {node_id} doesn't exist", code="node_not_found", details={"node_id": node_id}
)
def _get_closed_data(self, section, id_key):
"""
@ -817,7 +819,9 @@ class Project:
try:
return self._drawings[drawing_id]
except KeyError:
raise ControllerNotFoundError(f"Drawing ID {drawing_id} doesn't exist")
raise ControllerNotFoundError(
f"Drawing ID {drawing_id} doesn't exist", code="drawing_not_found", details={"drawing_id": drawing_id}
)
@open_required
async def delete_drawing(self, drawing_id):
@ -1101,7 +1105,9 @@ class Project:
try:
return self._links[link_id]
except KeyError:
raise ControllerNotFoundError(f"Link ID {link_id} doesn't exist")
raise ControllerNotFoundError(
f"Link ID {link_id} doesn't exist", code="link_not_found", details={"link_id": link_id}
)
@property
def links(self):
@ -1462,7 +1468,11 @@ class Project:
try:
return self._snapshots[snapshot_id]
except KeyError:
raise ControllerNotFoundError(f"Snapshot ID {snapshot_id} doesn't exist")
raise ControllerNotFoundError(
f"Snapshot ID {snapshot_id} doesn't exist",
code="snapshot_not_found",
details={"snapshot_id": snapshot_id},
)
def _load_snapshot_config(self):

8
gns3server/db/repositories/rbac.py
View File

@ -16,11 +16,11 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import logging
from typing import List, Optional, Union, cast
from typing import Any, List, Optional, Union, cast
from urllib.parse import urlparse
from uuid import UUID
from sqlalchemy import delete, select, update
from sqlalchemy import Select, delete, select, update
from sqlalchemy.engine import CursorResult
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@ -331,7 +331,7 @@ class RbacRepository(BaseRepository):
Retrieve all user ACEs matching the user_id and privilege name.
"""
query = (
query: Select[Any, Any, Any, Any] = (
select(models.ACE.path, models.ACE.propagate, models.ACE.allowed, models.Privilege.name)
.join(models.Privilege.roles)
.join(models.Role.acl_entries)
@ -349,7 +349,7 @@ class RbacRepository(BaseRepository):
Retrieve all group ACEs matching the user_id and privilege name.
"""
query = (
query: Select[Any, Any, Any, Any] = (
select(models.ACE.path, models.ACE.propagate, models.ACE.allowed, models.Privilege.name)
.join(models.Privilege.roles)
.join(models.Role.acl_entries)

14
gns3server/schemas/__init__.py
View File

@ -17,7 +17,7 @@
# General schemas
from .common import ErrorMessage
from .config import ServerConfig
from .controller.appliances import Appliance, ApplianceVersion, ApplianceVersionV8
from .controller.appliances import Appliance, ApplianceVersion, ApplianceVersionCreate, ApplianceVersionV8
from .controller.computes import (
AutoIdlePC,
Compute,
@ -49,6 +49,10 @@ from .controller.links import (
LinkCreate,
LinkFilterInfo,
LinkFilterParameter,
LinkFilterParameter,
LinkFilters,
LinkFilterType,
LinkIfaceInfo,
LinkUpdate,
LinkWiresharkRestart,
Marker,
@ -136,6 +140,7 @@ from .controller.marker_replay import (
ReplayTreeNode,
)
from .controller.netmiko import NetmikoDeviceType, NetmikoDeviceTypeList
from .controller.notifications import Notification, NotificationAction
from .controller.pools import Resource, ResourceCreate, ResourcePool, ResourcePoolCreate, ResourcePoolUpdate
from .controller.rbac import ACE, ACECreate, ACEEndpoint, ACEUpdate, Privilege, Role, RoleCreate, RoleUpdate
from .controller.settings import SettingsResponse, SettingsUpdate, SettingsUpdateResponse
@ -204,6 +209,7 @@ __all__ = [
"ApiKeyMessage",
"Appliance",
"ApplianceVersion",
"ApplianceVersionCreate",
"ApplianceVersionV8",
"AutoIdlePC",
"BatchNIOCreate",
@ -297,6 +303,10 @@ __all__ = [
"LinkFilterInfo",
"LinkFilterParameter",
"LinkStatistics",
"LinkFilterParameter",
"LinkFilterType",
"LinkFilters",
"LinkIfaceInfo",
"LinkUpdate",
"LinkWiresharkRestart",
"LoggedInUserUpdate",
@ -319,6 +329,8 @@ __all__ = [
"NodeFile",
"NodeStatistics",
"NodeUpdate",
"Notification",
"NotificationAction",
"OpenAIMessage",
"OpenAIToolCall",
"Privilege",

9
gns3server/schemas/common.py
View File

@ -15,7 +15,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from enum import Enum
from typing import Optional
from typing import Any, Optional
from pydantic import BaseModel, Field, field_validator
@ -26,6 +26,11 @@ class ErrorMessage(BaseModel):
"""
message: str
code: Optional[str] = Field(
None,
description="Stable machine-readable error code, for example port_in_use or node_not_found",
)
details: Optional[dict[str, Any]] = Field(None, description="Optional structured data about the error")
class NodeStatus(str, Enum):
@ -36,6 +41,8 @@ class NodeStatus(str, Enum):
stopped = "stopped"
started = "started"
suspended = "suspended"
starting = "starting"
stopping = "stopping"
class CustomAdapter(BaseModel):

3
gns3server/schemas/config.py
View File

@ -155,6 +155,9 @@ class ServerSettings(BaseModel):
False, description="Local server mode, set by the --local command line argument (not meant to be set by hand)"
)
enable_http_auth: bool = Field(True, description="Enable compute HTTP authentication")
openapi_include_ai: bool = Field(
True, description="Include the AI Copilot, LLM and MCP routes in the generated OpenAPI specification"
)
name: str = Field(
f"{socket.gethostname()} (controller)",
description="Server name, default is what is returned by socket.gethostname()",

35
gns3server/schemas/controller/appliances.py
View File

@ -20,7 +20,7 @@ from enum import Enum
from typing import Annotated, List, Literal, Optional, Union
from uuid import UUID
from pydantic import AnyUrl, BaseModel, Discriminator, EmailStr, Field, Tag
from pydantic import AnyUrl, BaseModel, BeforeValidator, Discriminator, EmailStr, Field, Tag
from ..common import ExtraConfig
@ -648,6 +648,39 @@ class ApplianceVersionV8(BaseModel):
images: Optional[ApplianceVersionImages] = Field(None, title="Images used for this version")
_V8_VERSION_ONLY_FIELDS = frozenset(set(ApplianceVersionV8.model_fields) - set(ApplianceVersion.model_fields))
class ApplianceVersionCreateV1_6(ApplianceVersion):
"""Request body to add a version to an appliance with registry version 1-6"""
registry_version: Literal[1, 2, 3, 4, 5, 6] = Field(
1, title="Version of the registry compatible with the appliance receiving the version"
)
class ApplianceVersionCreateV8(ApplianceVersionV8):
"""Request body to add a version to an appliance with registry version 8"""
registry_version: Literal[8] = Field(
8, title="Version of the registry compatible with the appliance receiving the version"
)
def _infer_version_registry_version(data):
if isinstance(data, dict) and "registry_version" not in data:
data = dict(data)
data["registry_version"] = 8 if _V8_VERSION_ONLY_FIELDS.intersection(data) else 1
return data
ApplianceVersionCreate = Annotated[
Union[ApplianceVersionCreateV1_6, ApplianceVersionCreateV8],
BeforeValidator(_infer_version_registry_version),
Discriminator("registry_version"),
]
# ============================================================================
# Child Models with Discriminated Union
# ============================================================================

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

@ -15,10 +15,10 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from enum import Enum
from typing import List, Optional, Tuple
from typing import Annotated, List, Literal, Optional, Tuple, Union
from uuid import UUID, uuid4
from pydantic import BaseModel, Field, field_validator
from pydantic import BaseModel, ConfigDict, Discriminator, Field, field_validator, model_validator
from ..update import PartialUpdateModel
from .labels import Label
@ -35,6 +35,32 @@ class LinkNode(BaseModel):
label: Optional[Label] = None
class LinkNodeCreate(BaseModel):
"""
Link node data for link creation.
The port is given either by adapter_number and port_number, by port_name
or by port set to "auto" (first free compatible port).
"""
node_id: UUID
adapter_number: Optional[int] = None
port_number: Optional[int] = None
port_name: Optional[str] = Field(None, description="Name of the port on the node")
port: Optional[Literal["auto"]] = Field(None, description='Use "auto" to select the first free compatible port')
label: Optional[Label] = None
@model_validator(mode="after")
def check_port_selector(self):
numbered = self.adapter_number is not None or self.port_number is not None
if numbered and (self.adapter_number is None or self.port_number is None):
raise ValueError("adapter_number and port_number must be given together")
forms = sum([numbered, self.port_name is not None, self.port is not None])
if forms != 1:
raise ValueError("exactly one of adapter_number and port_number, port_name or port must be given")
return self
class LinkType(str, Enum):
"""
Link type.
@ -54,6 +80,77 @@ class LinkStyle(BaseModel):
control_offset: Optional[Tuple[float, float]] = None
class LinkFilterType(str, Enum):
"""
Packet filter type.
"""
frequency_drop = "frequency_drop"
packet_loss = "packet_loss"
delay = "delay"
corrupt = "corrupt"
bpf = "bpf"
class LinkFilters(BaseModel):
"""
Packet filters applied on a link. Each filter is an array of positional values.
A filter set to an empty array or to its disabled value (0, 0 for delay, empty text for bpf) is not applied.
"""
model_config = ConfigDict(extra="forbid")
frequency_drop: Optional[List[Annotated[int, Field(ge=-1, le=32767)]]] = Field(
None,
max_length=1,
description="[frequency]: -1 drops every packet, N > 0 drops every Nth packet, 0 disables the filter",
)
packet_loss: Optional[List[Annotated[int, Field(ge=0, le=100)]]] = Field(
None,
max_length=1,
description="[chance]: percentage chance for a packet to be lost, 0 disables the filter",
)
delay: Optional[List[Annotated[int, Field(ge=0, le=32767)]]] = Field(
None,
max_length=2,
description="[latency, jitter] in milliseconds: latency must be 1 to 32767 unless both values are 0, "
"which disables the filter; jitter is 0 to 32767",
)
corrupt: Optional[List[Annotated[int, Field(ge=0, le=100)]]] = Field(
None,
max_length=1,
description="[chance]: percentage chance for a packet to be corrupted, 0 disables the filter",
)
bpf: Optional[List[str]] = Field(
None,
max_length=1,
description="[expressions]: BPF expressions, one per line, matching packets are dropped",
)
class LinkFilterParameter(BaseModel):
"""
Parameter of a packet filter.
"""
name: str
type: Literal["int", "text"]
minimum: Optional[int] = None
maximum: Optional[int] = None
unit: Optional[str] = None
class LinkFilterDefinition(BaseModel):
"""
Packet filter available on a link.
"""
type: LinkFilterType
name: str
description: str
parameters: List[LinkFilterParameter]
class LinkBase(BaseModel):
"""
Link data.
@ -62,7 +159,7 @@ class LinkBase(BaseModel):
nodes: Optional[List[LinkNode]] = Field(None, min_length=0, max_length=2)
suspend: Optional[bool] = None
link_style: Optional[LinkStyle] = None
filters: Optional[dict] = None
filters: Optional[LinkFilters] = None
markers: Optional[dict] = Field(
None, description="Traffic-insight markers on this link: name → {bpf, tag, enabled}"
)
@ -71,7 +168,8 @@ class LinkBase(BaseModel):
class LinkCreate(LinkBase):
link_id: UUID = Field(default_factory=uuid4)
nodes: List[LinkNode] = Field(..., min_length=2, max_length=2)
# LinkNodeCreate narrows the port selector, so it cannot be a LinkNode subclass
nodes: List[LinkNodeCreate] = Field(..., min_length=2, max_length=2) # type: ignore[assignment]
class LinkUpdate(PartialUpdateModel, LinkBase):
@ -107,6 +205,7 @@ class UDPPortInfo(BaseModel):
rhost: str
rport: int
type: str
kind: Literal["udp"]
class EthernetPortInfo(BaseModel):
@ -117,6 +216,10 @@ class EthernetPortInfo(BaseModel):
node_id: UUID
interface: str
type: str
kind: Literal["ethernet"]
LinkIfaceInfo = Annotated[Union[UDPPortInfo, EthernetPortInfo], Discriminator("kind")]
class LinkCapture(BaseModel):

154
gns3server/schemas/controller/notifications.py Normal file
View File

@ -0,0 +1,154 @@
#
# 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/>.
from enum import Enum
from typing import Any, Dict, List, Optional, Union
from uuid import UUID
from pydantic import BaseModel, Field
from .computes import Compute
from .drawings import Drawing
from .links import Link
from .nodes import Node
from .projects import Project
from .snapshots import Snapshot
class NotificationAction(str, Enum):
"""
Values of the "action" field of a notification.
"""
ping = "ping"
log_info = "log.info"
log_warning = "log.warning"
log_error = "log.error"
node_created = "node.created"
node_updated = "node.updated"
node_deleted = "node.deleted"
node_interface_status = "node.interface_status"
link_created = "link.created"
link_updated = "link.updated"
link_deleted = "link.deleted"
link_web_wireshark_started = "link.web_wireshark_started"
drawing_created = "drawing.created"
drawing_updated = "drawing.updated"
drawing_deleted = "drawing.deleted"
marker_match = "marker.match"
snapshot_restored = "snapshot.restored"
project_created = "project.created"
project_updated = "project.updated"
project_opened = "project.opened"
project_closed = "project.closed"
project_deleted = "project.deleted"
compute_created = "compute.created"
compute_updated = "compute.updated"
compute_deleted = "compute.deleted"
template_created = "template.created"
template_updated = "template.updated"
template_deleted = "template.deleted"
settings_updated = "settings.updated"
class PingEvent(BaseModel):
"""
Event of a "ping" notification, sent when the stream is opened and then at least every 5 seconds.
"""
cpu_usage_percent: float = Field(..., description="CPU usage of the server in percent")
memory_usage_percent: float = Field(..., description="Memory usage of the server in percent")
disk_usage_percent: float = Field(..., description="Disk usage of the project directory in percent")
compute_id: Optional[str] = Field(None, description="Compute that sent the ping (project streams only)")
class LogEvent(BaseModel):
"""
Event of a "log.info", "log.warning" or "log.error" notification.
"""
message: str = Field(..., description="Log message")
class NodeInterfaceStatusEvent(BaseModel):
"""
Event of a "node.interface_status" notification.
"""
project_id: UUID
node_id: UUID
adapter_number: int
port_number: int
status: str = Field(..., description="Status of the interface")
class MarkerMatchEvent(BaseModel):
"""
Event of a "marker.match" notification, only sent on the markers WebSocket.
"""
project_id: UUID
node_id: UUID
link_id: Optional[UUID] = None
filter: str = Field(..., description="Name of the packet filter that matched")
tag: Optional[int] = Field(None, description="Tag of the marker")
ts: float = Field(..., description="Timestamp reported by uBridge (seconds since the epoch)")
len: int = Field(0, description="Length of the matching packet")
dir: Optional[str] = Field(None, description="Direction of the packet relative to the capture node")
class LinkWebWiresharkStartedEvent(BaseModel):
"""
Event of a "link.web_wireshark_started" notification.
"""
link_id: UUID
ws_url: Optional[str] = Field(None, description="URL of the Web Wireshark session")
class SettingsUpdatedEvent(BaseModel):
"""
Event of a "settings.updated" notification.
"""
changed: List[str] = Field(..., description="Names of the settings that changed (values are never sent)")
restart_required: List[str] = Field(..., description="Names of the changed settings that need a server restart")
class Notification(BaseModel):
"""
A message of the notification streams. HTTP streams send one JSON object per line (application/x-ndjson),
WebSocket streams send one JSON object per text frame.
"""
action: NotificationAction = Field(..., description="Type of the notification")
event: Union[
PingEvent,
LogEvent,
NodeInterfaceStatusEvent,
MarkerMatchEvent,
LinkWebWiresharkStartedEvent,
SettingsUpdatedEvent,
Node,
Link,
Drawing,
Project,
Compute,
Snapshot,
Dict[str, Any],
] = Field(..., description="Payload of the notification, its content depends on the action")
project_id: Optional[UUID] = Field(None, description="Project the notification belongs to, when set by the sender")
compute_id: Optional[str] = Field(None, description="Compute that emitted the notification, when set by the sender")

5
gns3server/schemas/controller/projects.py
View File

@ -15,6 +15,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from datetime import datetime
from enum import Enum
from typing import List, Optional
from uuid import UUID
@ -111,8 +112,8 @@ class NodeFile(BaseModel):
path: str = Field(..., description="File name")
size: int = Field(..., description="File size in bytes")
created_at: str = Field(..., description="File creation time (ISO 8601)")
modified_at: str = Field(..., description="File modification time (ISO 8601)")
created_at: Optional[datetime] = Field(None, description="File creation time (ISO 8601)")
modified_at: Optional[datetime] = Field(None, description="File modification time (ISO 8601)")
file_type: str = Field(..., description="File type determined by the file command")

1
gns3server/schemas/controller/settings.py
View File

@ -103,6 +103,7 @@ class ServerSettingsUpdate(BaseModel):
local: Optional[bool] = None
enable_http_auth: Optional[bool] = None
openapi_include_ai: Optional[bool] = None
name: Optional[str] = None
protocol: Optional[ServerProtocol] = None
host: Optional[str] = None

15
gns3server/schemas/controller/snapshots.py
View File

@ -15,10 +15,11 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from datetime import datetime, timezone
from typing import Optional
from uuid import UUID
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, model_validator
class SnapshotBase(BaseModel):
@ -44,4 +45,14 @@ class Snapshot(SnapshotBase):
name: str = Field(..., description="Name of the snapshot")
filename: str = Field(..., description="Filename of the snapshot")
description: str = Field(..., description="Description of the snapshot")
created_at: int = Field(..., description="Date of the snapshot (UTC timestamp)")
created_at: int = Field(
..., description="Date of the snapshot (UTC timestamp). Deprecated: use created", deprecated=True
)
created: datetime = Field(..., description="Date of the snapshot (ISO 8601, UTC)")
@model_validator(mode="before")
@classmethod
def _derive_created(cls, data):
if isinstance(data, dict) and "created" not in data and data.get("created_at") is not None:
data = {**data, "created": datetime.fromtimestamp(data["created_at"], tz=timezone.utc)}
return data

2
gns3server/schemas/controller/tokens.py
View File

@ -23,7 +23,9 @@ from pydantic import BaseModel
class Token(BaseModel):
access_token: str
token_type: str
expires_in: int
refresh_token: Optional[str] = None
refresh_expires_in: Optional[int] = None
class TokenData(BaseModel):

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

@ -32,6 +32,11 @@ from .nodes import Node
from .projects import Supplier, Variable
class TopologyLink(Link):
# Legacy topology files store filters as untyped dicts
filters: Optional[dict] = None # type: ignore[assignment]
class TopologyType(str, Enum):
topology = "topology"
@ -39,7 +44,7 @@ class TopologyType(str, Enum):
class TopologyContent(BaseModel):
computes: List[Compute] = Field(..., description="List of computes")
drawings: List[Drawing] = Field(..., description="List of drawings")
links: List[Link] = Field(..., description="List of links")
links: List[TopologyLink] = Field(..., description="List of links")
nodes: List[Node] = Field(..., description="List of nodes")

6
gns3server/server.py
View File

@ -255,9 +255,9 @@ class Server:
else:
log.info(f"Compute authentication is enabled with username '{config.Server.compute_username}'")
# we only support Python 3 version >= 3.9
if sys.version_info < (3, 9, 0):
raise SystemExit("Python 3.9 or higher is required")
# we only support Python 3 version >= 3.11
if sys.version_info < (3, 11, 0):
raise SystemExit("Python 3.11 or higher is required")
log.info(
f"Running with Python {sys.version_info[0]}.{sys.version_info[1]}.{sys.version_info[2]} and has PID {os.getpid()}"

21
gns3server/services/authentication.py
View File

@ -29,7 +29,7 @@ from joserfc.jwk import OctKey
from pydantic import ValidationError
from gns3server.config import Config
from gns3server.schemas.controller.tokens import TokenData
from gns3server.schemas.controller.tokens import Token, TokenData
log = logging.getLogger(__name__)
@ -47,6 +47,9 @@ def _extract_alg(token: str) -> str:
return "<undecodable>"
BEARER_TOKEN_TYPE = "bearer"
class AuthService:
def hash_password(self, password: str) -> str:
@ -61,8 +64,9 @@ class AuthService:
def _create_token(self, username, token_version, token_type, expires_in, secret_key=None) -> str:
"""Shared helper to create any kind of signed JWT token."""
expire = datetime.now(timezone.utc) + timedelta(minutes=expires_in)
to_encode = {"sub": username, "exp": expire, "ver": token_version, "type": token_type}
issued_at = datetime.now(timezone.utc)
expire = issued_at + timedelta(minutes=expires_in)
to_encode = {"sub": username, "iat": issued_at, "exp": expire, "ver": token_version, "type": token_type}
if secret_key is None:
secret_key = Config.instance().settings.Controller.jwt_secret_key
if secret_key is None:
@ -89,6 +93,17 @@ class AuthService:
expires_in = Config.instance().settings.Controller.jwt_refresh_token_expire_minutes
return self._create_token(username, token_version, "refresh", expires_in, secret_key)
def create_token_response(self, user) -> Token:
settings = Config.instance().settings.Controller
return Token(
access_token=self.create_access_token(user.username, token_version=user.token_version),
token_type=BEARER_TOKEN_TYPE,
expires_in=settings.jwt_access_token_expire_minutes * 60,
refresh_token=self.create_refresh_token(user.username, token_version=user.token_version),
refresh_expires_in=settings.jwt_refresh_token_expire_minutes * 60,
)
def get_token_data(self, token: str, secret_key: Optional[str] = None) -> TokenData:
def auth_error(detail: str) -> HTTPException:

2
gns3server/version.py
View File

@ -35,3 +35,5 @@ if "dev" in __version__:
__version__ += "+" + r
except Exception as e:
print(e)
__api_version__ = __version__.split("+", 1)[0]

7
pyproject.toml
View File

@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
[project]
name = "gns3-server"
description = "GNS3 graphical interface for the GNS3 server."
description = "GNS3 server (controller and compute) for GNS3 network simulator"
license-files = ["LICENSE"]
authors = [
{ name = "Jeremy Grossmann", email = "developers@gns3.com" }
]
readme = "README.md"
requires-python = ">=3.10"
requires-python = ">=3.11"
classifiers = [
"Development Status :: 5 - Production/Stable",
"Environment :: Console",
@ -21,10 +21,10 @@ classifiers = [
"Natural Language :: English",
"Operating System :: POSIX :: Linux",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Programming Language :: Python :: Implementation :: CPython"
]
@ -219,6 +219,7 @@ extend-immutable-calls = [
"scripts/welcome.py" = ["C901", "E722", "PLW1510", "S108", "S112", "S603", "S605", "S606", "S607"]
"tests/agent/mcp/test_handlers.py" = ["B905"]
"tests/agent/mcp/test_tool_params.py" = ["C901"]
"tests/api/test_openapi_spec.py" = ["S603"]
"tests/api/routes/controller/test_images.py" = ["S324", "W292"]
"tests/api/routes/controller/test_links.py" = ["RUF015", "W292"]
"tests/api/routes/controller/test_users.py" = ["F811"]

12
requirements.txt
View File

@ -1,19 +1,17 @@
# GNS3 Server Core Dependencies
uvicorn==0.54.0
pydantic==2.13.5
fastapi==0.141.1
fastapi==0.142.4
python-multipart==0.0.32
websockets==16.1.1 # version 16.1.1 is the last version that supports Python 3.10
aiohttp>=3.14.3,<3.15
websockets==16.1.1 # google-genai 2.10.0 has requirement websockets<17.0
aiohttp>=3.14.4,<3.15
aiofiles>=25.1.0,<26.0
Jinja2>=3.1.6,<3.2
sentry-sdk>=2.70.0,<3 # optional dependency
psutil>=7.2.2
async-timeout>=5.0.1,<5.1; python_version < '3.11' # this library has effectively been upstreamed into Python 3.11+
distro>=1.9.0
py-cpuinfo>=9.0.0,<10.0
greenlet==3.5.3; python_version >= '3.13' # necessary to run sqlalchemy on Python >= 3.13
sqlalchemy==2.0.54 # version 2.0.54 is the last version that supports Python 3.10
sqlalchemy[asyncio]==2.1.4
aiosqlite==0.22.1
alembic==1.20
bcrypt==5.0.0
@ -22,7 +20,7 @@ email-validator==2.3.0
watchdog==6.0.0
zstandard==0.25.0
platformdirs>=2.4.0 # fastmcp-slim >=3.4 requires >=4.0.0; upper bound removed for compatibility
truststore>=0.10.4; python_version >= '3.10'
truststore>=0.10.4
# Shared dependencies (also used by AI Copilot)
telnetlib3==5.0.1,<6

30
tests/api/routes/controller/test_appliances.py
View File

@ -101,3 +101,33 @@ class TestApplianceRoutes:
app.url_path_for("add_appliance_version", appliance_id=appliance_id), json=new_version
)
assert response.status_code == status.HTTP_409_CONFLICT
async def test_add_v8_version_appliance(self, app: FastAPI, client: AsyncClient) -> None:
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
new_version = {"name": "v8-99G", "category": "guest", "images": {"hda_disk_image": "empty99G.qcow2"}}
response = await client.post(
app.url_path_for("add_appliance_version", appliance_id=appliance_id), json=new_version
)
assert response.status_code == status.HTTP_201_CREATED
assert new_version in response.json()["versions"]
async def test_add_version_appliance_with_registry_version(self, app: FastAPI, client: AsyncClient) -> None:
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
stored_version = {"name": "explicit-99G", "images": {"hda_disk_image": "empty99G.qcow2"}}
response = await client.post(
app.url_path_for("add_appliance_version", appliance_id=appliance_id),
json=dict(stored_version, registry_version=3),
)
assert response.status_code == status.HTTP_201_CREATED
assert stored_version in response.json()["versions"]
async def test_add_version_appliance_unknown_registry_version(self, app: FastAPI, client: AsyncClient) -> None:
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
response = await client.post(
app.url_path_for("add_appliance_version", appliance_id=appliance_id),
json={"name": "bad", "registry_version": 7, "images": {"hda_disk_image": "empty99G.qcow2"}},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY

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

@ -115,6 +115,110 @@ class TestLinkRoutes:
assert response.status_code == status.HTTP_409_CONFLICT
assert len(project.links) == 0
async def test_create_link_by_port_name(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create"):
response = await client.post(
app.url_path_for("create_link", project_id=project.id),
json={
"nodes": [
{"node_id": node1.id, "port_name": "E0"},
{"node_id": node2.id, "adapter_number": 2, "port_number": 4},
]
},
)
assert response.status_code == status.HTTP_201_CREATED
first, second = response.json()["nodes"]
assert (first["adapter_number"], first["port_number"]) == (0, 3)
assert (second["adapter_number"], second["port_number"]) == (2, 4)
async def test_create_link_unknown_port_name(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
response = await client.post(
app.url_path_for("create_link", project_id=project.id),
json={
"nodes": [
{"node_id": node1.id, "port_name": "nope"},
{"node_id": node2.id, "adapter_number": 2, "port_number": 4},
]
},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert len(project.links) == 0
async def test_create_link_auto_port(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
node1._ports = [EthernetPort("E1", 0, 0, 5), EthernetPort("E0", 0, 0, 3)]
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create"):
response = await client.post(
app.url_path_for("create_link", project_id=project.id),
json={"nodes": [{"node_id": node1.id, "port": "auto"}, {"node_id": node2.id, "port": "auto"}]},
)
assert response.status_code == status.HTTP_201_CREATED
assert response.json()["nodes"][0]["port_number"] == 3
response = await client.post(
app.url_path_for("create_link", project_id=project.id),
json={"nodes": [{"node_id": node1.id, "port": "auto"}, {"node_id": node2.id, "port": "auto"}]},
)
assert response.status_code == status.HTTP_409_CONFLICT
assert response.json()["code"] == "no_free_port"
assert len(project.links) == 1
async def test_create_link_port_in_use_code(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
body = {
"nodes": [
{"node_id": node1.id, "adapter_number": 0, "port_number": 3},
{"node_id": node2.id, "adapter_number": 2, "port_number": 4},
]
}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create"):
first = await client.post(app.url_path_for("create_link", project_id=project.id), json=body)
second = await client.post(app.url_path_for("create_link", project_id=project.id), json=body)
assert first.status_code == status.HTTP_201_CREATED
assert second.status_code == status.HTTP_409_CONFLICT
assert second.json()["code"] == "port_in_use"
assert len(project.links) == 1
async def test_create_link_invalid_port_selector(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
valid = {"node_id": node2.id, "adapter_number": 2, "port_number": 4}
for bad in (
{"node_id": node1.id},
{"node_id": node1.id, "adapter_number": 0},
{"node_id": node1.id, "port_name": "E0", "port": "auto"},
{"node_id": node1.id, "adapter_number": 0, "port_number": 3, "port_name": "E0"},
):
response = await client.post(
app.url_path_for("create_link", project_id=project.id), json={"nodes": [bad, valid]}
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
async def test_get_link(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
@ -143,6 +247,43 @@ class TestLinkRoutes:
assert response.status_code == status.HTTP_200_OK
assert response.json()["nodes"][0]["label"]["x"] == 42
async def test_link_etag_and_if_match(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create"):
response = await client.post(
app.url_path_for("create_link", project_id=project.id),
json={
"nodes": [
{"node_id": node1.id, "adapter_number": 0, "port_number": 3},
{"node_id": node2.id, "adapter_number": 2, "port_number": 4},
]
},
)
link_id = response.json()["link_id"]
url = app.url_path_for("get_link", project_id=project.id, link_id=link_id)
response = await client.get(url)
assert response.status_code == status.HTTP_200_OK
etag = response.headers["ETag"]
assert (await client.get(url)).headers["ETag"] == etag
response = await client.put(url, json={"show_filters_icon": False}, headers={"If-Match": etag})
assert response.status_code == status.HTTP_200_OK
assert response.headers["ETag"] != etag
assert (await client.get(url)).headers["ETag"] == response.headers["ETag"]
response = await client.put(url, json={"show_filters_icon": True}, headers={"If-Match": etag})
assert response.status_code == status.HTTP_412_PRECONDITION_FAILED
assert "message" in response.json()
assert (await client.get(url)).json()["show_filters_icon"] is False
response = await client.put(url, json={"show_filters_icon": True})
assert response.status_code == status.HTTP_200_OK
assert response.json()["show_filters_icon"] is True
async def test_update_link_suspend(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
@ -335,6 +476,97 @@ class TestLinkRoutes:
assert response.status_code == status.HTTP_200_OK
assert response.json() == FILTERS
async def test_update_link_filters(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
filters = {
"frequency_drop": [50],
"packet_loss": [10],
"delay": [10, 5],
"corrupt": [3],
"bpf": ["icmp"],
}
response = await client.put(
app.url_path_for("update_link", project_id=project.id, link_id=link.id), json={"filters": filters}
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["filters"] == filters
assert link.filters == filters
async def test_update_link_disabled_filters(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
response = await client.put(
app.url_path_for("update_link", project_id=project.id, link_id=link.id),
json={"filters": {"delay": [0, 0], "packet_loss": [0], "corrupt": [], "bpf": [""]}},
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["filters"] == {}
@pytest.mark.parametrize(
"filters",
[
{"packet_loss": [101]},
{"packet_loss": [-1]},
{"corrupt": [101]},
{"frequency_drop": [-2]},
{"frequency_drop": [32768]},
{"delay": [32768, 0]},
{"delay": [10, -1]},
{"packet_loss": [1, 2]},
{"bpf": [1]},
{"packet_loss": ["abc"]},
],
)
async def test_update_link_invalid_filter_values(
self, app: FastAPI, client: AsyncClient, project: Project, filters: dict
) -> None:
link = Link(project)
project._links = {link.id: link}
response = await client.put(
app.url_path_for("update_link", project_id=project.id, link_id=link.id), json={"filters": filters}
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
assert "message" in response.json()
assert link.filters == {}
async def test_update_link_unknown_filter_type(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
response = await client.put(
app.url_path_for("update_link", project_id=project.id, link_id=link.id),
json={"filters": {"unknown": [1]}},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
assert "message" in response.json()
assert link.filters == {}
async def test_update_link_invalid_delay_latency(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
link = Link(project)
project._links = {link.id: link}
response = await client.put(
app.url_path_for("update_link", project_id=project.id, link_id=link.id), json={"filters": {"delay": [0, 5]}}
)
assert response.status_code == status.HTTP_409_CONFLICT
async def test_openapi_link_filters(self, app: FastAPI) -> None:
schemas = app.openapi()["components"]["schemas"]
link_filters = schemas["LinkFilters"]
assert link_filters["additionalProperties"] is False
assert set(link_filters["properties"]) == {"frequency_drop", "packet_loss", "delay", "corrupt", "bpf"}
assert schemas["LinkFilterDefinition"]["properties"]["parameters"]["items"] == {
"$ref": "#/components/schemas/LinkFilterParameter"
}
path = "/v3/projects/{project_id}/links/{link_id}/available_filters"
responses = app.openapi()["paths"][path]["get"]["responses"]["200"]["content"]["application/json"]["schema"]
assert responses["items"] == {"$ref": "#/components/schemas/LinkFilterDefinition"}
async def test_get_udp_interface(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
"""
Test getting UDP tunnel interface information from a link.
@ -376,6 +608,7 @@ class TestLinkRoutes:
assert result["rhost"] == "127.0.0.1"
assert result["rport"] == 30000
assert result["type"] == "udp"
assert result["kind"] == "udp"
async def test_get_ethernet_interface(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
"""
@ -406,3 +639,4 @@ class TestLinkRoutes:
assert result["node_id"] == cloud_node.id
assert result["interface"] == "eth0"
assert result["type"] == "ethernet"
assert result["kind"] == "ethernet"

159
tests/api/routes/controller/test_nodes.py
View File

@ -34,6 +34,7 @@ from pydantic import SecretStr
from gns3server.api.routes.controller.nodes import vnc_console, ws_console
from gns3server.config import Config
from gns3server.controller.compute import Compute
from gns3server.controller.controller_error import ComputeError
from gns3server.controller.node import Node
from gns3server.controller.project import Project
from gns3server.services import auth_service
@ -250,6 +251,77 @@ class TestNodeRoutes:
assert "name" not in response.json()["properties"]
assert response.json()["tags"] == ["tag1", "tag2"]
async def test_node_etag_on_get_and_put(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
compute.put = AsyncioMagicMock(return_value=MagicMock(json={"console": 2048}))
url = app.url_path_for("update_node", project_id=project.id, node_id=node.id)
response = await client.get(url)
assert response.status_code == status.HTTP_200_OK
etag = response.headers["ETag"]
assert etag.startswith('"') and etag.endswith('"')
assert (await client.get(url)).headers["ETag"] == etag
response = await client.put(url, json={"name": "renamed"})
assert response.status_code == status.HTTP_200_OK
assert response.headers["ETag"] != etag
assert (await client.get(url)).headers["ETag"] == response.headers["ETag"]
async def test_update_node_with_matching_if_match(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
compute.put = AsyncioMagicMock(return_value=MagicMock(json={"console": 2048}))
url = app.url_path_for("update_node", project_id=project.id, node_id=node.id)
etag = (await client.get(url)).headers["ETag"]
response = await client.put(url, json={"name": "renamed"}, headers={"If-Match": etag})
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "renamed"
assert response.headers["ETag"] != etag
async def test_update_node_with_stale_if_match(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
compute.put = AsyncioMagicMock(return_value=MagicMock(json={"console": 2048}))
url = app.url_path_for("update_node", project_id=project.id, node_id=node.id)
stale_etag = (await client.get(url)).headers["ETag"]
assert (await client.put(url, json={"name": "first"})).status_code == status.HTTP_200_OK
response = await client.put(url, json={"name": "second"}, headers={"If-Match": stale_etag})
assert response.status_code == status.HTTP_412_PRECONDITION_FAILED
assert "message" in response.json()
assert node.name == "first"
async def test_update_node_without_if_match(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
compute.put = AsyncioMagicMock(return_value=MagicMock(json={"console": 2048}))
url = app.url_path_for("update_node", project_id=project.id, node_id=node.id)
assert (await client.put(url, json={"name": "first"})).status_code == status.HTTP_200_OK
assert (await client.put(url, json={"name": "second"})).status_code == status.HTTP_200_OK
assert node.name == "second"
async def test_update_node_with_wildcard_and_list_if_match(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
compute.put = AsyncioMagicMock(return_value=MagicMock(json={"console": 2048}))
url = app.url_path_for("update_node", project_id=project.id, node_id=node.id)
etag = (await client.get(url)).headers["ETag"]
response = await client.put(url, json={"name": "a"}, headers={"If-Match": "*"})
assert response.status_code == status.HTTP_200_OK
etag = (await client.get(url)).headers["ETag"]
response = await client.put(url, json={"name": "b"}, headers={"If-Match": f'"other", {etag}'})
assert response.status_code == status.HTTP_200_OK
async def test_start_all_nodes(self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
compute.post = AsyncioMagicMock()
@ -292,7 +364,8 @@ class TestNodeRoutes:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("start_node", project_id=project.id, node_id=node.id), json={})
assert response.status_code == status.HTTP_204_NO_CONTENT
assert response.status_code == status.HTTP_200_OK
assert response.json()["status"] == "started"
async def test_stop_node(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
@ -300,6 +373,84 @@ class TestNodeRoutes:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_200_OK
assert response.json()["status"] == "stopped"
async def test_start_node_wait(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
compute.post = AsyncioMagicMock()
response = await client.post(
app.url_path_for("start_node", project_id=project.id, node_id=node.id) + "?wait=true&timeout=5", json={}
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["status"] == "started"
response = await client.get(app.url_path_for("get_node", project_id=project.id, node_id=node.id))
assert response.json()["status"] == "started"
async def test_start_node_reports_starting_while_in_flight(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
seen = []
async def post(*args, **kwargs):
seen.append(node.asdict()["status"])
compute.post = post
response = await client.post(app.url_path_for("start_node", project_id=project.id, node_id=node.id), json={})
assert response.status_code == status.HTTP_200_OK
assert seen == ["starting"]
assert response.json()["status"] == "started"
async def test_stop_node_reports_stopping_while_in_flight(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
seen = []
async def post(*args, **kwargs):
seen.append(node.asdict()["status"])
compute.post = post
node._status = "started"
response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_200_OK
assert seen == ["stopping"]
assert response.json()["status"] == "stopped"
async def test_start_node_wait_timeout(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
async def post(*args, **kwargs):
await asyncio.sleep(10)
compute.post = post
response = await client.post(
app.url_path_for("start_node", project_id=project.id, node_id=node.id) + "?wait=true&timeout=1", json={}
)
assert response.status_code == status.HTTP_408_REQUEST_TIMEOUT
assert "message" in response.json()
assert node.asdict()["status"] == "stopped"
async def test_stop_node_reports_compute_error(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
compute.post = AsyncioMagicMock(side_effect=ComputeError("compute unreachable"))
node._status = "started"
response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_409_CONFLICT
assert node.asdict()["status"] == "started"
async def test_stop_all_nodes_tolerates_compute_error(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
) -> None:
compute.post = AsyncioMagicMock(side_effect=ComputeError("compute unreachable"))
response = await client.post(app.url_path_for("stop_all_nodes", project_id=project.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_suspend_node(
@ -308,7 +459,8 @@ class TestNodeRoutes:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("suspend_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert response.status_code == status.HTTP_200_OK
assert response.json()["status"] == "suspended"
async def test_suspend_node_unsupported(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node
@ -338,7 +490,8 @@ class TestNodeRoutes:
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("reload_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
assert response.status_code == status.HTTP_200_OK
assert response.json()["status"] == "started"
async def test_isolate_node(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node

88
tests/api/routes/controller/test_notifications.py Normal file
View File

@ -0,0 +1,88 @@
#
# 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/>.
import json
import uuid
import pytest
from fastapi import FastAPI
from pydantic import ValidationError
from gns3server import schemas
from gns3server.utils.notification_queue import NotificationQueue
STREAM_PATHS = ["/v3/notifications", "/v3/projects/{project_id}/notifications"]
class TestNotificationOpenAPI:
def test_notification_schema_in_spec(self, app: FastAPI) -> None:
spec = app.openapi()
schema = spec["components"]["schemas"]["Notification"]
assert set(schema["required"]) == {"action", "event"}
assert {"action", "event", "project_id", "compute_id"} <= set(schema["properties"])
action_schema = spec["components"]["schemas"]["NotificationAction"]
assert {"ping", "node.created", "node.updated", "link.created", "marker.match"} <= set(action_schema["enum"])
@pytest.mark.parametrize("path", STREAM_PATHS)
def test_stream_responses_reference_schema(self, app: FastAPI, path: str) -> None:
content = app.openapi()["paths"][path]["get"]["responses"]["200"]["content"]
assert list(content) == ["application/x-ndjson"]
assert content["application/x-ndjson"]["schema"] == {"$ref": "#/components/schemas/Notification"}
class TestNotificationSchema:
def test_ping(self) -> None:
message = {
"action": "ping",
"event": {"cpu_usage_percent": 1.5, "memory_usage_percent": 40.0, "disk_usage_percent": 10.0},
}
notification = schemas.Notification.model_validate(message)
assert notification.action == schemas.NotificationAction.ping
def test_unknown_action(self) -> None:
with pytest.raises(ValidationError):
schemas.Notification.model_validate({"action": "unknown.action", "event": {}})
def test_optional_ids(self) -> None:
project_id = str(uuid.uuid4())
notification = schemas.Notification.model_validate(
{"action": "log.info", "event": {"message": "hello"}, "project_id": project_id, "compute_id": "local"}
)
assert str(notification.project_id) == project_id
assert notification.compute_id == "local"
def test_generic_event(self) -> None:
notification = schemas.Notification.model_validate(
{"action": "template.deleted", "event": {"template_id": str(uuid.uuid4())}}
)
assert notification.action == schemas.NotificationAction.template_deleted
@pytest.mark.asyncio
async def test_queue_message_matches_schema(self) -> None:
queue = NotificationQueue()
queue.put_nowait(("log.warning", {"message": "disk space is low"}, {"compute_id": "local"}))
await queue.get_json(5)
message = json.loads(await queue.get_json(5))
notification = schemas.Notification.model_validate(message)
assert notification.action == schemas.NotificationAction.log_warning
assert notification.compute_id == "local"

27
tests/api/routes/controller/test_projects.py
View File

@ -139,6 +139,33 @@ class TestControllerProjectRoutes:
assert response.json()["path"] == path
assert response.json()["project_id"] == "10010203-0405-0607-0809-0a0b0c0d0e0f"
async def test_project_etag_and_if_match(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None:
params = {"name": "test", "project_id": "10010203-0405-0607-0809-0a0b0c0d0e0f"}
assert (await client.post(app.url_path_for("create_project"), json=params)).status_code == 201
get_url = app.url_path_for("get_project", project_id="10010203-0405-0607-0809-0a0b0c0d0e0f")
put_url = app.url_path_for("update_project", project_id="10010203-0405-0607-0809-0a0b0c0d0e0f")
response = await client.get(get_url)
assert response.status_code == status.HTTP_200_OK
etag = response.headers["ETag"]
assert (await client.get(get_url)).headers["ETag"] == etag
response = await client.put(put_url, json={"name": "test2"}, headers={"If-Match": etag})
assert response.status_code == status.HTTP_200_OK
new_etag = response.headers["ETag"]
assert new_etag != etag
assert (await client.get(get_url)).headers["ETag"] == new_etag
response = await client.put(put_url, json={"name": "test3"}, headers={"If-Match": etag})
assert response.status_code == status.HTTP_412_PRECONDITION_FAILED
assert "message" in response.json()
assert (await client.get(get_url)).json()["name"] == "test2"
response = await client.put(put_url, json={"name": "test4"})
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test4"
async def test_update_project_with_variables(
self, app: FastAPI, client: AsyncClient, controller: Controller
) -> None:

16
tests/api/routes/controller/test_snapshots.py
View File

@ -17,6 +17,7 @@
import os
import uuid
from datetime import datetime, timezone
import pytest
import pytest_asyncio
@ -54,6 +55,9 @@ class TestSnapshotRoutes:
response = await client.get(app.url_path_for("get_snapshots", project_id=project.id))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 1
data = response.json()[0]
assert isinstance(data["created_at"], int)
assert datetime.fromisoformat(data["created"]) == datetime.fromtimestamp(data["created_at"], tz=timezone.utc)
async def test_delete_snapshot(
self, app: FastAPI, client: AsyncClient, project: Project, snapshot: Snapshot
@ -80,3 +84,15 @@ class TestSnapshotRoutes:
response = await client.post(app.url_path_for("create_snapshot", project_id=project.id), json={"name": "snap1"})
assert response.status_code == status.HTTP_201_CREATED
assert len(os.listdir(os.path.join(project.path, "snapshots"))) == 2
data = response.json()
assert isinstance(data["created_at"], int)
assert datetime.fromisoformat(data["created"]) == datetime.fromtimestamp(data["created_at"], tz=timezone.utc)
async def test_snapshot_schema_timestamps(self, app: FastAPI) -> None:
schemas = app.openapi()["components"]["schemas"]
assert schemas["Snapshot"]["properties"]["created"]["format"] == "date-time"
assert schemas["Snapshot"]["properties"]["created_at"]["type"] == "integer"
assert schemas["Snapshot"]["properties"]["created_at"]["deprecated"] is True
for field in ("created_at", "modified_at"):
assert {"type": "string", "format": "date-time"} in schemas["NodeFile"]["properties"][field]["anyOf"]

68
tests/api/routes/controller/test_users.py
View File

@ -446,6 +446,74 @@ class TestLogout:
assert response.status_code == status.HTTP_401_UNAUTHORIZED
class TestTokenLifetime:
async def _issue(self, app, client, user, endpoint):
credentials = {"username": user.username, "password": "user1_password"}
if endpoint == "login":
return await client.post(
app.url_path_for("login"),
data=credentials,
headers={"content-type": "application/x-www-form-urlencoded"},
)
auth = await client.post(app.url_path_for("authenticate"), json=credentials)
if endpoint == "authenticate":
return auth
return await client.post(
app.url_path_for("refresh_access_token"), json={"refresh_token": auth.json()["refresh_token"]}
)
@pytest.mark.parametrize("endpoint", ("login", "authenticate", "refresh_access_token"))
async def test_token_response_includes_lifetimes(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
config: Config,
endpoint: str,
) -> None:
response = await self._issue(app, unauthorized_client, test_user, endpoint)
assert response.status_code == status.HTTP_200_OK
body = response.json()
controller = config.settings.Controller
assert body["expires_in"] == controller.jwt_access_token_expire_minutes * 60
assert body["refresh_expires_in"] == controller.jwt_refresh_token_expire_minutes * 60
assert body["access_token"]
assert body["token_type"] == "bearer"
assert body["refresh_token"]
@pytest.mark.parametrize("endpoint", ("login", "authenticate", "refresh_access_token"))
async def test_expires_in_matches_jwt_claims(
self,
app: FastAPI,
unauthorized_client: AsyncClient,
test_user: User,
config: Config,
endpoint: str,
) -> None:
response = await self._issue(app, unauthorized_client, test_user, endpoint)
body = response.json()
key = OctKey.import_key(config.settings.Controller.jwt_secret_key)
access = jwt.decode(body["access_token"], key, algorithms=["HS256"]).claims
refresh = jwt.decode(body["refresh_token"], key, algorithms=["HS256"]).claims
assert access["exp"] - access["iat"] == body["expires_in"]
assert refresh["exp"] - refresh["iat"] == body["refresh_expires_in"]
async def test_openapi_declares_token_lifetimes_as_integers(self, app: FastAPI) -> None:
schema = app.openapi()["components"]["schemas"]["Token"]
assert schema["properties"]["expires_in"]["type"] == "integer"
refresh_expires_in = schema["properties"]["refresh_expires_in"]
assert "integer" in str(refresh_expires_in)
assert "expires_in" in schema["required"]
async def test_openapi_security_scheme_mentions_api_keys(self, app: FastAPI) -> None:
schemes = app.openapi()["components"]["securitySchemes"]
assert any("API key" in scheme.get("description", "") for scheme in schemes.values())
class TestRefreshToken:
async def test_login_returns_refresh_token(
self,

5
tests/api/routes/controller/test_version.py
View File

@ -42,7 +42,10 @@ class TestVersionRoutes:
params = {"version": "0.4.2"}
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_409_CONFLICT
assert response.json() == {"message": f"Client version 0.4.2 is not the same as server version {__version__}"}
assert response.json() == {
"message": f"Client version 0.4.2 is not the same as server version {__version__}",
"code": "conflict",
}
async def test_version_invalid_input_schema(self, app: FastAPI, client: AsyncClient) -> None:

221
tests/api/routes/test_error_responses.py Normal file
View File

@ -0,0 +1,221 @@
#
# 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/>.
import uuid
from typing import Tuple
from unittest.mock import MagicMock
import pytest
import pytest_asyncio
from fastapi import FastAPI, status
from httpx import AsyncClient
from gns3server.api.errors import code_for_status
from gns3server.controller.compute import Compute
from gns3server.controller.controller_error import ComputeConflictError, ControllerError
from gns3server.controller.node import Node
from gns3server.controller.ports.ethernet_port import EthernetPort
from gns3server.controller.project import Project
from gns3server.db.models import User
from gns3server.schemas import ErrorMessage
from gns3server.services import auth_service
from gns3server.services.authentication import DEFAULT_JWT_SECRET_KEY
from tests.utils import AsyncioMagicMock, asyncio_patch
pytestmark = pytest.mark.asyncio
class TestErrorResponses:
@pytest_asyncio.fixture
async def nodes(self, compute: Compute, project: Project) -> Tuple[Node, Node]:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
node1 = await project.add_node(compute, "node1", None, node_type="qemu")
node1._ports = [EthernetPort("E0", 0, 0, 3)]
node2 = await project.add_node(compute, "node2", None, node_type="qemu")
node2._ports = [EthernetPort("E0", 0, 2, 4)]
return node1, node2
async def test_unauthorized(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
response = await client.get(
app.url_path_for("get_project", project_id=project.id),
headers={"Authorization": "Bearer invalid_token"},
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
body = response.json()
assert isinstance(body["message"], str)
assert body["code"] == "unauthorized"
assert "detail" not in body
async def test_forbidden(self, app: FastAPI, client: AsyncClient, test_user: User, project: Project) -> None:
token = auth_service.create_access_token(test_user.username, secret_key=DEFAULT_JWT_SECRET_KEY)
response = await client.delete(
app.url_path_for("delete_project", project_id=project.id),
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == status.HTTP_403_FORBIDDEN
body = response.json()
assert isinstance(body["message"], str)
assert body["code"] == "forbidden"
async def test_unknown_node(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
node_id = str(uuid.uuid4())
response = await client.get(app.url_path_for("get_node", project_id=project.id, node_id=node_id))
assert response.status_code == status.HTTP_404_NOT_FOUND
body = response.json()
assert body["message"] == f"Node ID {node_id} doesn't exist"
assert body["code"] == "node_not_found"
assert body["details"] == {"node_id": node_id}
async def test_unknown_project(self, app: FastAPI, client: AsyncClient) -> None:
project_id = str(uuid.uuid4())
response = await client.get(app.url_path_for("get_project", project_id=project_id))
assert response.status_code == status.HTTP_404_NOT_FOUND
body = response.json()
assert body["code"] == "project_not_found"
assert body["details"] == {"project_id": project_id}
async def test_unknown_route(self, client: AsyncClient) -> None:
response = await client.get("/v3/no/such/route")
assert response.status_code == status.HTTP_404_NOT_FOUND
body = response.json()
assert body["message"] == "Not Found"
assert body["code"] == "not_found"
assert "detail" not in body
async def test_port_in_use(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
payload = {
"nodes": [
{"node_id": node1.id, "adapter_number": 0, "port_number": 3},
{"node_id": node2.id, "adapter_number": 2, "port_number": 4},
]
}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create"):
first = await client.post(app.url_path_for("create_link", project_id=project.id), json=payload)
assert first.status_code == status.HTTP_201_CREATED
second = await client.post(app.url_path_for("create_link", project_id=project.id), json=payload)
assert second.status_code == status.HTTP_409_CONFLICT
body = second.json()
assert body["message"] == "Port is already used"
assert body["code"] == "port_in_use"
assert body["details"]["node_id"] == node1.id
assert body["details"]["adapter_number"] == 0
assert body["details"]["port_number"] == 3
assert body["details"]["link_id"] == first.json()["link_id"]
async def test_generic_conflict(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, _ = nodes
node1._ports = [EthernetPort("E0", 0, 0, 3), EthernetPort("E1", 0, 0, 4)]
response = await client.post(
app.url_path_for("create_link", project_id=project.id),
json={
"nodes": [
{"node_id": node1.id, "adapter_number": 0, "port_number": 3},
{"node_id": node1.id, "adapter_number": 0, "port_number": 4},
]
},
)
assert response.status_code == status.HTTP_409_CONFLICT
body = response.json()
assert body["message"] == "Cannot connect to itself"
assert body["code"] == "conflict"
assert "details" not in body
async def test_validation_error(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
response = await client.post(app.url_path_for("create_link", project_id=project.id), json={"nodes": "bad"})
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
body = response.json()
assert isinstance(body["message"], str)
assert body["code"] == "validation_error"
assert body["details"]["errors"]
assert {"loc", "msg", "type"} <= set(body["details"]["errors"][0])
assert "detail" not in body
async def test_message_only_clients_keep_working(self, app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("get_project", project_id=str(uuid.uuid4())))
assert ErrorMessage.model_validate({"message": response.json()["message"]}).code is None
assert isinstance(response.json()["message"], str)
class TestErrorCodes:
async def test_default_codes(self) -> None:
assert ControllerError("x").code == "conflict"
assert ControllerError("x", code="custom", details={"a": 1}).details == {"a": 1}
async def test_compute_conflict_code(self) -> None:
error = ComputeConflictError("http://c/x", {"message": "boom", "code": "port_in_use"})
assert error.code == "port_in_use"
assert ComputeConflictError("http://c/x", {"message": "boom"}).code == "compute_conflict"
@pytest.mark.parametrize(
"status_code,code",
[(401, "unauthorized"), (403, "forbidden"), (404, "not_found"), (409, "conflict"), (501, "not_implemented")],
)
async def test_code_for_status(self, status_code: int, code: str) -> None:
assert code_for_status(status_code) == code
class TestOpenAPIErrorSchemas:
async def test_error_schemas(self, client: AsyncClient) -> None:
response = await client.get("/openapi.json")
assert response.status_code == status.HTTP_200_OK
spec = response.json()
schemas = spec["components"]["schemas"]
assert "HTTPValidationError" not in schemas
assert "ValidationError" not in schemas
assert set(schemas["ErrorMessage"]["properties"]) == {"message", "code", "details"}
assert schemas["ErrorMessage"]["required"] == ["message"]
error_ref = {"$ref": "#/components/schemas/ErrorMessage"}
authenticated = 0
for path, path_item in spec["paths"].items():
for method, operation in path_item.items():
responses = operation["responses"]
if "422" in responses:
assert responses["422"]["content"]["application/json"]["schema"] == error_ref, (method, path)
if operation.get("security"):
authenticated += 1
for code in ("401", "403", "422"):
assert responses[code]["content"]["application/json"]["schema"] == error_ref, (
method,
path,
code,
)
assert authenticated > 0
assert "HTTPValidationError" not in response.text

68
tests/api/routes/test_openapi.py Normal file
View File

@ -0,0 +1,68 @@
#
# 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/>.
import pytest
from fastapi import FastAPI
pytestmark = pytest.mark.asyncio
STREAM_AND_BINARY_RESPONSES = [
("/v3/notifications", "get", "200", "application/x-ndjson"),
("/v3/projects/{project_id}/notifications", "get", "200", "application/x-ndjson"),
("/v3/projects/{project_id}/export", "get", "200", "application/zip"),
("/v3/projects/{project_id}/gns3file", "get", "200", "application/json"),
("/v3/projects/{project_id}/files/{file_path}", "get", "200", "application/octet-stream"),
("/v3/projects/{project_id}/nodes/{node_id}/files/{file_path}", "get", "200", "application/octet-stream"),
("/v3/projects/{project_id}/nodes/{node_id}/files/{file_path}", "post", "201", "application/json"),
("/v3/projects/{project_id}/links/{link_id}/capture/stream", "get", "200", "application/vnd.tcpdump.pcap"),
("/v3/symbols/{symbol_id}/raw", "get", "200", "application/octet-stream"),
]
RAW_BODY_REQUESTS = [
("/v3/images/upload/{image_path}", "application/octet-stream"),
("/v3/symbols/{symbol_id}/raw", "application/octet-stream"),
("/v3/projects/{project_id}/files/{file_path}", "application/octet-stream"),
("/v3/projects/{project_id}/nodes/{node_id}/files/{file_path}", "application/octet-stream"),
("/v3/projects/{project_id}/import", "application/zip"),
]
class TestOpenAPIBinaryRoutes:
@pytest.mark.parametrize("path, method, status_code, media_type", STREAM_AND_BINARY_RESPONSES)
async def test_response_content(
self, app: FastAPI, path: str, method: str, status_code: str, media_type: str
) -> None:
content = app.openapi()["paths"][path][method]["responses"][status_code]["content"]
assert media_type in content
assert content[media_type]["schema"] != {}
if media_type != "application/json":
assert "application/json" not in content
@pytest.mark.parametrize("path, media_type", RAW_BODY_REQUESTS)
async def test_request_body(self, app: FastAPI, path: str, media_type: str) -> None:
request_body = app.openapi()["paths"][path]["post"]["requestBody"]
assert request_body["required"] is True
assert request_body["content"][media_type]["schema"] == {"type": "string", "format": "binary"}
async def test_binary_responses_use_binary_schema(self, app: FastAPI) -> None:
paths = app.openapi()["paths"]
for path, method, status_code, media_type in STREAM_AND_BINARY_RESPONSES:
if media_type in ("application/octet-stream", "application/zip", "application/vnd.tcpdump.pcap"):
schema = paths[path][method]["responses"][status_code]["content"][media_type]["schema"]
assert schema == {"type": "string", "format": "binary"}

65
tests/api/test_openapi_discriminators.py Normal file
View File

@ -0,0 +1,65 @@
#!/usr/bin/env python
#
# 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/>.
import pytest
from fastapi import FastAPI
pytestmark = pytest.mark.asyncio
@pytest.fixture
def openapi(app: FastAPI) -> dict:
return app.openapi()
def _ref(name: str) -> str:
return f"#/components/schemas/{name}"
def _assert_discriminated(schema: dict, property_name: str, mapping: dict) -> None:
assert [item["$ref"] for item in schema["oneOf"]] == [_ref(name) for name in dict.fromkeys(mapping.values())]
assert schema["discriminator"]["propertyName"] == property_name
assert schema["discriminator"]["mapping"] == {key: _ref(name) for key, name in mapping.items()}
async def test_link_iface_response_is_discriminated(openapi: dict) -> None:
path = "/v3/projects/{project_id}/links/{link_id}/iface"
schema = openapi["paths"][path]["get"]["responses"]["200"]["content"]["application/json"]["schema"]
_assert_discriminated(schema, "kind", {"udp": "UDPPortInfo", "ethernet": "EthernetPortInfo"})
async def test_appliance_version_request_is_discriminated(openapi: dict) -> None:
schema = openapi["paths"]["/v3/appliances/{appliance_id}/version"]["post"]["requestBody"]["content"][
"application/json"
]["schema"]
mapping = {str(version): "ApplianceVersionCreateV1_6" for version in range(1, 7)}
mapping["8"] = "ApplianceVersionCreateV8"
_assert_discriminated(schema, "registry_version", mapping)
async def test_appliance_settings_are_discriminated(openapi: dict) -> None:
schema = openapi["components"]["schemas"]["ApplianceV8"]["properties"]["settings"]["items"]
_assert_discriminated(
schema,
"template_type",
{
"qemu": "QemuTemplateSetting",
"dynamips": "DynamipsTemplateSetting",
"iou": "IouTemplateSetting",
"docker": "DockerTemplateSetting",
},
)

114
tests/api/test_openapi_spec.py Normal file
View File

@ -0,0 +1,114 @@
#
# 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/>.
import json
import subprocess
import sys
import pytest
from fastapi import FastAPI
from gns3server.agent import AI_COPILOT_AVAILABLE, MCP_AVAILABLE
from gns3server.version import __api_version__, __version__
pytestmark = pytest.mark.asyncio
AI_PATH_PREFIXES = (
"/v3/copilot",
"/v3/mcp",
"/v3/access/llm",
"/v3/access/users/{user_id}/llm-model-configs",
"/v3/access/groups/{group_id}/llm-model-configs",
)
SPEC_SCRIPT = """
import json
import sys
from gns3server.config import Config
Config.instance(files=[sys.argv[1]])
from gns3server.api.server import app
print(json.dumps(app.openapi()))
"""
def _operation_ids(spec: dict) -> list:
return [
operation["operationId"]
for path_item in spec["paths"].values()
for operation in path_item.values()
if isinstance(operation, dict) and "operationId" in operation
]
def _build_spec(tmp_path, include_ai: bool) -> dict:
config_file = tmp_path / "gns3_server.conf"
config_file.write_text(f"[Server]\nopenapi_include_ai = {str(include_ai).lower()}\n")
result = subprocess.run(
[sys.executable, "-c", SPEC_SCRIPT, str(config_file)],
capture_output=True,
text=True,
check=True,
timeout=120,
)
return json.loads(result.stdout.splitlines()[-1])
class TestOpenAPISpec:
async def test_controller_version_matches_package(self, app: FastAPI) -> None:
spec = app.openapi()
assert spec["info"]["version"] == __api_version__
assert spec["info"]["version"] == __version__.split("+")[0]
assert "+" not in spec["info"]["version"]
async def test_compute_version_matches_package(self, app: FastAPI) -> None:
compute_app = next(route.app for route in app.routes if getattr(route, "path", None) == "/v3/compute")
assert compute_app.openapi()["info"]["version"] == __api_version__
async def test_servers_entry(self, app: FastAPI) -> None:
assert app.openapi()["servers"] == [{"url": "/"}]
async def test_no_catch_all_operations(self, app: FastAPI) -> None:
paths = app.openapi()["paths"]
assert [path for path in paths if "{path}" in path] == []
async def test_operation_ids_are_unique(self, app: FastAPI) -> None:
operation_ids = _operation_ids(app.openapi())
assert len(operation_ids) == len(set(operation_ids))
@pytest.mark.skipif(AI_COPILOT_AVAILABLE or MCP_AVAILABLE, reason="AI extras are installed")
async def test_ai_routes_absent_without_extras(self, app: FastAPI) -> None:
paths = app.openapi()["paths"]
assert [path for path in paths if path.startswith(AI_PATH_PREFIXES)] == []
async def test_setting_hides_ai_routes_and_keeps_the_rest(self, tmp_path) -> None:
with_ai = _build_spec(tmp_path, include_ai=True)
without_ai = _build_spec(tmp_path, include_ai=False)
assert [path for path in without_ai["paths"] if path.startswith(AI_PATH_PREFIXES)] == []
non_ai_paths = {path: item for path, item in with_ai["paths"].items() if not path.startswith(AI_PATH_PREFIXES)}
assert without_ai["paths"] == non_ai_paths
assert len(_operation_ids(without_ai)) == len(set(_operation_ids(without_ai)))

54
tests/controller/test_appliance.py
View File

@ -17,10 +17,21 @@
import pydantic
import pytest
from pydantic import TypeAdapter
from gns3server.controller.appliance import Appliance
from gns3server.controller.appliance_to_template import ApplianceToTemplate
from gns3server.schemas.controller.appliances import ApplianceModel
from gns3server.schemas.controller.appliances import (
ApplianceModel,
ApplianceVersionCreate,
ApplianceVersionCreateV1_6,
ApplianceVersionCreateV8,
DockerTemplateSetting,
DynamipsTemplateSetting,
IouTemplateSetting,
QemuTemplateSetting,
TemplateSetting,
)
# v8 mirror of the XRd Control Plane appliance shape (docker, custom_adapters, no versions)
XRD_V8 = {
@ -242,3 +253,44 @@ def test_v8_netmiko_device_type_empty_string_clears():
appliance = dict(XRD_V8, netmiko_device_type="")
model = ApplianceModel.model_validate(appliance)
assert model.netmiko_device_type == ""
@pytest.mark.parametrize(
"template_type, properties, expected",
[
("qemu", {"ram": 512}, QemuTemplateSetting),
("dynamips", {"platform": "c7200", "ram": 256}, DynamipsTemplateSetting),
("iou", {"ram": 256}, IouTemplateSetting),
("docker", {"image": "alpine:latest"}, DockerTemplateSetting),
],
)
def test_template_setting_decoded_through_discriminator(template_type, properties, expected):
setting = TypeAdapter(TemplateSetting).validate_python(
{"name": "a", "template_type": template_type, "template_properties": properties}
)
assert type(setting) is expected
assert setting.template_type == template_type
def test_template_setting_unknown_template_type_rejected():
with pytest.raises(pydantic.ValidationError):
TypeAdapter(TemplateSetting).validate_python({"template_type": "vpcs", "template_properties": {}})
@pytest.mark.parametrize(
"data, expected",
[
({"name": "1", "registry_version": 1}, ApplianceVersionCreateV1_6),
({"name": "6", "registry_version": 6}, ApplianceVersionCreateV1_6),
({"name": "8", "registry_version": 8}, ApplianceVersionCreateV8),
({"name": "plain", "images": {"hda_disk_image": "a.qcow2"}}, ApplianceVersionCreateV1_6),
({"name": "v8", "category": "guest"}, ApplianceVersionCreateV8),
],
)
def test_appliance_version_decoded_through_discriminator(data, expected):
assert type(TypeAdapter(ApplianceVersionCreate).validate_python(data)) is expected
def test_appliance_version_unknown_registry_version_rejected():
with pytest.raises(pydantic.ValidationError):
TypeAdapter(ApplianceVersionCreate).validate_python({"name": "x", "registry_version": 7})

65
tests/controller/test_link.py
View File

@ -15,6 +15,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio
from unittest.mock import MagicMock
import pytest
@ -417,3 +418,67 @@ async def test_update_link_style(project, compute):
}
link._project.emit_notification.assert_called_with("link.updated", link.asdict())
assert project.dump.called
@pytest.mark.asyncio
async def test_add_node_concurrent_same_port(project, compute):
project.dump = MagicMock()
node1 = Node(project, compute, "node1", node_type="qemu")
node1._ports = [EthernetPort("E0", 0, 0, 4)]
node2 = Node(project, compute, "node2", node_type="qemu")
node2._ports = [EthernetPort("E0", 0, 0, 4), EthernetPort("E1", 0, 0, 5)]
node3 = Node(project, compute, "node3", node_type="qemu")
node3._ports = [EthernetPort("E0", 0, 0, 4)]
async def slow_create():
await asyncio.sleep(0)
async def connect(other, other_port):
link = Link(project)
link.create = slow_create
try:
await link.add_node(other, 0, other_port)
await link.add_node(node1, 0, 4)
except ControllerError as e:
return e
return link
results = await asyncio.gather(connect(node2, 4), connect(node3, 4))
errors = [r for r in results if isinstance(r, ControllerError)]
assert len(errors) == 1
assert errors[0].code == "port_in_use"
assert node1._ports[0].link in results
@pytest.mark.asyncio
async def test_add_node_create_failure_releases_ports(project, compute):
project.dump = MagicMock()
node1 = Node(project, compute, "node1", node_type="qemu")
node1._ports = [EthernetPort("E0", 0, 0, 4)]
node2 = Node(project, compute, "node2", node_type="qemu")
node2._ports = [EthernetPort("E0", 0, 0, 4)]
link = Link(project)
link.create = AsyncioMagicMock(side_effect=ControllerError("boom"))
await link.add_node(node1, 0, 4)
with pytest.raises(ControllerError):
await link.add_node(node2, 0, 4)
assert node1._ports[0].link is None
assert node2._ports[0].link is None
assert not node1.links
assert not node2.links
def test_node_get_port_by_name_and_free_port(project, compute):
node = Node(project, compute, "node1", node_type="qemu")
node._ports = [EthernetPort("E1", 0, 0, 1), EthernetPort("E0", 0, 0, 0), SerialPort("S0", 0, 1, 0)]
assert node.get_port_by_name("E0") is node._ports[1]
assert node.get_port_by_name("nope") is None
assert node.get_free_port() is node._ports[1]
assert node.get_free_port("serial") is node._ports[2]
node._ports[1].link = object()
assert node.get_free_port("ethernet") is node._ports[0]
node._ports[0].link = object()
assert node.get_free_port("ethernet") is None