Merge branch '3.1' into api-issue-06-8s9swu

This commit is contained in:
Jeremy Grossmann 2026-10-08 15:02:51 +02:00 committed by GitHub
commit 3187828325
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
46 changed files with 2104 additions and 220 deletions

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

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.",
}
}

View File

@ -80,7 +80,7 @@ def _iter_route_paths(routes: Sequence[BaseRoute], prefix: str = "", include_mou
yield from _iter_route_paths(mounted_routes, _join_paths(prefix, route.path), include_mounted_routes)
@router.get("/endpoints", status_code=status.HTTP_201_CREATED, dependencies=[Depends(has_privilege("ACE.Audit"))])
@router.get("/endpoints", dependencies=[Depends(has_privilege("ACE.Audit"))])
async def endpoints(
users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),

View File

@ -21,7 +21,7 @@ API routes for computes.
from typing import Any, List, Optional, Union
from uuid import UUID
from fastapi import APIRouter, Body, Depends, status
from fastapi import APIRouter, Body, Depends, Query, status
import gns3server.db.models as models
from gns3server import schemas
@ -106,14 +106,21 @@ async def get_compute(
)
async def get_computes(
computes_repo: ComputesRepository = Depends(get_repository(ComputesRepository)),
name: Optional[str] = Query(None, description="Return only computes whose name exactly matches (case-sensitive)"),
) -> List[models.Compute]:
"""
Return all computes known by the controller.
Required privilege: Compute.Audit
Query Parameters:
- name: Exact, case-sensitive match on the compute name.
"""
return await ComputesService(computes_repo).get_computes()
computes = await ComputesService(computes_repo).get_computes()
if name is not None:
computes = [compute for compute in computes if compute.name == name]
return computes
@router.put(

View File

@ -30,7 +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.openapi import ndjson_response
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
@ -147,7 +147,7 @@ def get_iou_license() -> schemas.IOULicense:
@router.put(
"/iou_license",
dependencies=[Depends(get_current_active_user)],
status_code=status.HTTP_201_CREATED,
status_code=status.HTTP_200_OK,
response_model=schemas.IOULicense,
)
async def update_iou_license(iou_license: schemas.IOULicense) -> schemas.IOULicense:
@ -257,13 +257,17 @@ async def statistics() -> dict:
@router.get(
"/notifications",
response_class=StreamingResponse,
responses={200: ndjson_response()},
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
@ -280,7 +284,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")
@ -288,7 +292,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:

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:

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)

View File

@ -196,14 +196,18 @@ async def get_images(
images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
image_type: Optional[schemas.ImageType] = None,
availability: Optional[Literal["unknown", "available", "missing", "unavailable", "invalid"]] = None,
name: Optional[str] = Query(None, description="Return only images whose filename exactly matches (case-sensitive)"),
) -> List[models.Image]:
"""
Return all images.
Required privilege: Image.Audit
Query Parameters:
- name: Exact, case-sensitive match on the image filename.
"""
return await images_repo.get_images(image_type, availability)
return await images_repo.get_images(image_type, availability, name)
@router.post(

View File

@ -20,12 +20,12 @@ API routes for links.
import logging
import os
from typing import Any, List, Union
from typing import Any, List, Optional, Union
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
@ -33,13 +33,21 @@ 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
@ -52,6 +60,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.
@ -111,21 +142,30 @@ 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()
@router.get("/{link_id}/available_filters", dependencies=[Depends(has_privilege("Link.Audit"))])
async def get_filters(link: Link = Depends(dep_link)) -> List[dict]:
@router.get(
"/{link_id}/available_filters",
response_model=List[schemas.LinkFilterDefinition],
response_model_exclude_unset=True,
dependencies=[Depends(has_privilege("Link.Audit"))],
)
async def get_filters(link: Link = Depends(dep_link)) -> List[schemas.LinkFilterDefinition]:
"""
Return all filters available for a given link.
@ -139,43 +179,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(

View File

@ -35,13 +35,25 @@ 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
@ -152,6 +164,7 @@ async def create_node(node_create: schemas.NodeCreate, project: Project = Depend
def get_nodes(
project: Project = Depends(dep_project),
tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)"),
name: Optional[str] = Query(None, description="Return only nodes whose name exactly matches (case-sensitive)"),
) -> List[schemas.Node]:
"""
Return all nodes belonging to a given project.
@ -161,6 +174,7 @@ def get_nodes(
Query Parameters:
- tags: Filter by tags. Multiple tags are ANDed together.
Example: ?tags=vendor:cisco&tags=model:7200
- name: Exact, case-sensitive match on the node name. Combined with other filters using AND.
"""
if project.status == "closed":
@ -169,6 +183,9 @@ def get_nodes(
else:
nodes = [v.asdict() for v in project.nodes.values()]
if name is not None:
nodes = [node for node in nodes if node.get("name") == name]
# Filter by tags if provided (all filter tags have to match the node tags)
if tags:
filtered_nodes = []
@ -252,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.
@ -262,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(
@ -327,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.
@ -376,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(

View File

@ -34,13 +34,25 @@ log = logging.getLogger()
from typing import Any, List, Optional
from uuid import UUID
from fastapi import APIRouter, Body, Depends, HTTPException, 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, ndjson_response
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
@ -56,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
@ -79,11 +99,15 @@ def dep_project(project_id: UUID) -> Project:
async def get_projects(
current_user: schemas.User = Depends(get_current_active_user),
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
name: Optional[str] = Query(None, description="Return only projects whose name exactly matches (case-sensitive)"),
) -> List[schemas.Project]:
"""
Return all projects.
Required privilege: Project.Audit
Query Parameters:
- name: Exact, case-sensitive match on the project name.
"""
controller = Controller.instance()
@ -92,7 +116,7 @@ async def get_projects(
if current_user.is_superadmin:
# super admin sees all projects
return [p.asdict() for p in controller.projects.values()]
return [p.asdict() for p in controller.projects.values() if name is None or p.name == name]
# Batch ACE + resource pool check (3 DB queries regardless of project count)
all_project_ids = list(controller.projects.keys())
@ -103,6 +127,8 @@ async def get_projects(
# Step 2: Filter direct ACE projects by created_by
# Direct project sharing is only available through resource pools
for p in controller.projects.values():
if name is not None and p.name != name:
continue
if p.id in direct_ace_ids and p.created_by == current_user.username:
if p.id not in seen_project_ids:
projects.append(p.asdict())
@ -110,6 +136,8 @@ async def get_projects(
# Step 3: Resource pool projects (no created_by filter)
for p in controller.projects.values():
if name is not None and p.name != name:
continue
if p.id in pool_accessible_ids:
if p.id not in seen_project_ids:
projects.append(p.asdict())
@ -143,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(
@ -488,7 +533,7 @@ async def close_project(project: Project = Depends(dep_project)) -> None:
@router.post(
"/{project_id}/open",
status_code=status.HTTP_201_CREATED,
status_code=status.HTTP_200_OK,
response_model=schemas.Project,
responses={**responses, 409: {"model": schemas.ErrorMessage, "description": "Could not open project"}},
dependencies=[Depends(has_privilege("Project.Allocate"))],
@ -526,13 +571,18 @@ async def load_project(path: str = Body(..., embed=True)) -> schemas.Project:
@router.get(
"/{project_id}/notifications",
response_class=StreamingResponse,
responses={200: ndjson_response()},
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
"""
@ -561,7 +611,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")
@ -571,7 +621,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
"""
@ -609,8 +660,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

View File

@ -106,7 +106,7 @@ async def delete_snapshot(
@router.post(
"/{snapshot_id}/restore",
status_code=status.HTTP_201_CREATED,
status_code=status.HTTP_200_OK,
response_model=schemas.Project,
dependencies=[Depends(has_privilege("Snapshot.Restore"))],
)

View File

@ -195,6 +195,7 @@ async def get_templates(
templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
current_user: schemas.User = Depends(get_current_active_user),
tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)"),
name: Optional[str] = Query(None, description="Return only templates whose name exactly matches (case-sensitive)"),
) -> List[dict]:
"""
Return all templates.
@ -204,10 +205,14 @@ async def get_templates(
Query Parameters:
- tags: Filter by tags. Multiple tags are ANDed together.
Example: ?tags=vendor:cisco&tags=model:7200
- name: Exact, case-sensitive match on the template name. Combined with other filters using AND.
"""
templates = await TemplatesService(templates_repo).get_templates()
if name is not None:
templates = [template for template in templates if template.get("name") == name]
# Filter by tags if provided (all filter tags have to match the node tags)
if tags:
filtered_templates = []

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)

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,13 +29,14 @@ 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.controller.controller_error import (
@ -71,12 +72,26 @@ 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",
description="This page describes the public controller API for GNS3.\n\n"
"## Notification streams\n\n"
"Notifications are available as HTTP streams of newline delimited JSON objects (`GET /v3/notifications` and "
"`GET /v3/projects/{project_id}/notifications`) and as WebSockets sending one JSON object per text frame "
"(`/v3/notifications/ws`, `/v3/projects/{project_id}/notifications/ws` and "
"`/v3/projects/{project_id}/notifications/markers/ws`). "
"OpenAPI cannot describe WebSocket routes, they carry the same `Notification` messages as the HTTP streams. "
"Marker events (`marker.match`) are only sent on the markers WebSocket.",
version="3.0.0",
docs_url=None,
redoc_url=None,
@ -151,82 +166,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})

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,

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

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

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

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],

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

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

View File

@ -83,7 +83,7 @@ class ImagesRepository(BaseRepository):
except (OSError, InvalidImageError):
return False
async def get_images(self, image_type=None, availability=None) -> List[models.Image]:
async def get_images(self, image_type=None, availability=None, name=None) -> List[models.Image]:
"""
Get all images.
"""
@ -94,6 +94,8 @@ class ImagesRepository(BaseRepository):
query = select(models.Image)
if availability:
query = query.where(models.Image.availability == availability)
if name is not None:
query = query.where(models.Image.filename == name)
result = await self._db_session.execute(query)
return list(result.scalars().all())

View File

@ -37,6 +37,10 @@ from .controller.links import (
Link,
LinkCapture,
LinkCreate,
LinkFilterDefinition,
LinkFilterParameter,
LinkFilters,
LinkFilterType,
LinkUpdate,
MarkerCreate,
MarkerDefinitionCreate,
@ -109,6 +113,7 @@ from .compute.vpcs_nodes import VPCS, VPCSCreate, VPCSUpdate
from .controller.capabilities import Capabilities
from .controller.iou_license import IOULicense
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, ACEUpdate, Privilege, Role, RoleCreate, RoleUpdate
from .controller.settings import SettingsResponse, SettingsUpdate, SettingsUpdateResponse
@ -244,6 +249,10 @@ __all__ = [
"Link",
"LinkCapture",
"LinkCreate",
"LinkFilterDefinition",
"LinkFilterParameter",
"LinkFilterType",
"LinkFilters",
"LinkUpdate",
"LoggedInUserUpdate",
"MarkerCreate",
@ -261,6 +270,8 @@ __all__ = [
"NodeDuplicate",
"NodeFile",
"NodeUpdate",
"Notification",
"NotificationAction",
"OpenAIMessage",
"OpenAIToolCall",
"Privilege",

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

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
from uuid import UUID, uuid4
from pydantic import BaseModel, Field, field_validator
from pydantic import BaseModel, ConfigDict, 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):

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

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

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

View File

@ -22,7 +22,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):

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

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:

View File

@ -161,3 +161,8 @@ class TestACLRoutes:
# the ACE should have been deleted after deleting the project
assert not await rbac_repo.get_ace_by_path(path)
async def test_get_endpoints(self, app: FastAPI, client: AsyncClient) -> None:
response = await client.get(app.url_path_for("endpoints"))
assert response.status_code == status.HTTP_200_OK

View File

@ -68,6 +68,31 @@ class TestComputeRoutes:
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) > 0
async def test_compute_list_filter_by_name(self, app: FastAPI, client: AsyncClient) -> None:
for name, port in (("compute-a", 85), ("compute-b", 86)):
params = {
"name": name,
"protocol": "http",
"host": "localhost",
"port": port,
"user": "julien",
"password": "secure",
}
response = await client.post(app.url_path_for("create_compute"), json=params)
assert response.status_code == status.HTTP_201_CREATED
response = await client.get(app.url_path_for("get_computes"), params={"name": "compute-a"})
assert response.status_code == status.HTTP_200_OK
assert [c["name"] for c in response.json()] == ["compute-a"]
response = await client.get(app.url_path_for("get_computes"), params={"name": "COMPUTE-A"})
assert response.status_code == status.HTTP_200_OK
assert response.json() == []
response = await client.get(app.url_path_for("get_computes"))
assert len(response.json()) >= 2
async def test_compute_get(self, app: FastAPI, client: AsyncClient, test_compute: Compute) -> None:
response = await client.get(app.url_path_for("get_compute", compute_id=test_compute.compute_id))

View File

@ -70,3 +70,10 @@ class TestControllerRoutes:
response = await client.get(app.url_path_for("statistics"))
assert response.status_code == status.HTTP_200_OK
async def test_update_iou_license(self, app: FastAPI, client: AsyncClient) -> None:
license_settings = {"iourc_content": "[license]\ngns3vm = 1234567890abcdef;", "license_check": False}
response = await client.put(app.url_path_for("update_iou_license"), json=license_settings)
assert response.status_code == status.HTTP_200_OK
assert response.json() == license_settings

View File

@ -168,6 +168,21 @@ class TestImageRoutes:
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 5 # 4 valid images uploaded before + 1 created
async def test_image_list_filter_by_name(self, app: FastAPI, client: AsyncClient, qcow2_image: str) -> None:
image_name = os.path.basename(qcow2_image)
response = await client.get(app.url_path_for("get_images"), params={"name": image_name})
assert response.status_code == status.HTTP_200_OK
assert [i["filename"] for i in response.json()] == [image_name]
response = await client.get(app.url_path_for("get_images"), params={"name": image_name.upper()})
assert response.status_code == status.HTTP_200_OK
assert response.json() == []
response = await client.get(app.url_path_for("get_images"), params={"name": image_name, "image_type": "iou"})
assert response.status_code == status.HTTP_200_OK
assert response.json() == []
async def test_image_get(self, app: FastAPI, client: AsyncClient, qcow2_image: str) -> None:
image_name = os.path.basename(qcow2_image)

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.

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
@ -156,6 +157,54 @@ class TestNodeRoutes:
else:
assert len(response.json()) == 0
@pytest.mark.parametrize(
"name, tags, expected_names",
(
(None, None, ["test", "test2"]),
("test", None, ["test"]),
("test2", None, ["test2"]),
("TEST", None, []),
("missing", None, []),
("test", ["tag1"], ["test"]),
("test", ["tag3"], []),
("test2", ["tag3", "tag4"], ["test2"]),
),
)
async def test_list_nodes_filter_by_name(
self,
app: FastAPI,
client: AsyncClient,
project: Project,
compute: Compute,
name: Optional[str],
tags: Optional[list],
expected_names: list,
) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
for node_name, node_tags in (("test", ["tag1", "tag2"]), ("test2", ["tag3", "tag4"])):
await client.post(
app.url_path_for("create_node", project_id=project.id),
json={
"name": node_name,
"node_type": "vpcs",
"compute_id": "example.com",
"tags": node_tags,
"properties": {"startup_script": "echo test"},
},
)
params = {}
if name is not None:
params["name"] = name
if tags is not None:
params["tags"] = tags
response = await client.get(app.url_path_for("get_nodes", project_id=project.id), params=params)
assert response.status_code == status.HTTP_200_OK
assert sorted(n["name"] for n in response.json()) == expected_names
async def test_get_node(self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
response = MagicMock()
@ -202,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()
@ -244,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
@ -252,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(
@ -260,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
@ -290,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

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"

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:
@ -164,6 +191,26 @@ class TestControllerProjectRoutes:
projects = response.json()
assert projects[0]["name"] == "test"
async def test_list_projects_filter_by_name(
self, app: FastAPI, client: AsyncClient, controller: Controller
) -> None:
for name in ("test", "other"):
params = {"name": name, "project_id": str(uuid.uuid4())}
response = await client.post(app.url_path_for("create_project"), json=params)
assert response.status_code == status.HTTP_201_CREATED
response = await client.get(app.url_path_for("get_projects"), params={"name": "test"})
assert response.status_code == status.HTTP_200_OK
assert [p["name"] for p in response.json()] == ["test"]
response = await client.get(app.url_path_for("get_projects"), params={"name": "TEST"})
assert response.status_code == status.HTTP_200_OK
assert response.json() == []
response = await client.get(app.url_path_for("get_projects"))
assert len(response.json()) == 2
async def test_get_project(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
response = await client.get(app.url_path_for("get_project", project_id=project.id))
@ -196,7 +243,7 @@ class TestControllerProjectRoutes:
with asyncio_patch("gns3server.controller.project.Project.open", return_value=True) as mock:
response = await client.post(app.url_path_for("open_project", project_id=project.id))
assert response.status_code == status.HTTP_201_CREATED
assert response.status_code == status.HTTP_200_OK
assert mock.called
async def test_load_project(self, app: FastAPI, client: AsyncClient, project: Project, config) -> None:

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
@ -72,7 +76,7 @@ class TestSnapshotRoutes:
response = await client.post(
app.url_path_for("restore_snapshot", project_id=project.id, snapshot_id=snapshot.id)
)
assert response.status_code == status.HTTP_201_CREATED
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == project.name
async def test_create_snapshot(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
@ -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"]

View File

@ -20,6 +20,7 @@ import os
import unittest.mock
import uuid
from pathlib import Path
from typing import Optional
import pytest
from fastapi import FastAPI, status
@ -89,6 +90,27 @@ class TestTemplateRoutes:
else:
assert len(response.json()) == 0
@pytest.mark.parametrize(
"name, tags, expected_names",
(
("VPCS_TEST", None, ["VPCS_TEST"]),
("vpcs_test", None, []),
("missing", None, []),
("VPCS_TEST", ["tag1"], ["VPCS_TEST"]),
("VPCS_TEST", ["tag42"], []),
),
)
async def test_template_list_filter_by_name(
self, app: FastAPI, client: AsyncClient, name: str, tags: Optional[list], expected_names: list
) -> None:
params = {"name": name}
if tags is not None:
params["tags"] = tags
response = await client.get(app.url_path_for("get_templates"), params=params)
assert response.status_code == status.HTTP_200_OK
assert [t["name"] for t in response.json()] == expected_names
async def test_template_get(self, app: FastAPI, client: AsyncClient) -> None:
template_id = str(uuid.uuid4())

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,

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:

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

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