Merge branch '3.1' into api-issue-15-r12z73

This commit is contained in:
Jeremy Grossmann authored and GitHub committed 2026-10-08 16:42:57 +02:00
commit 029983260c
46 files changed
+2103 -215

No files matched your search

+79
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
+41
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.",
}
}
+1 -1
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)),
+9 -2
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(
+16 -5
View File
@@ -30,6 +30,7 @@ from websockets.exceptions import ConnectionClosed, WebSocketException
from gns3server import schemas
from gns3server.agent.web_wireshark.stats import collect_webwireshark_stats
from gns3server.api.responses import NDJSON_MEDIA_TYPE, NOTIFICATION_STREAM_RESPONSES, NDJSONResponse
from gns3server.config import Config
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError, ControllerForbiddenError
@@ -146,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:
@@ -254,10 +255,19 @@ async def statistics() -> dict:
}
@router.get("/notifications", dependencies=[Depends(get_current_active_user)])
@router.get(
"/notifications",
dependencies=[Depends(get_current_active_user)],
response_class=NDJSONResponse,
responses=NOTIFICATION_STREAM_RESPONSES,
)
async def controller_http_notifications(request: Request) -> StreamingResponse:
"""
Receive controller notifications about the controller from HTTP stream.
Receive controller notifications from an HTTP stream of newline delimited JSON objects (see the Notification schema).
The same messages are available as WebSocket text frames on `/v3/notifications/ws`.
Controller actions: `compute.*`, `template.*`, `project.created`, `project.updated`, `project.opened`,
`project.closed`, `project.deleted`, `settings.updated`, `log.*` and `ping`.
"""
from gns3server.api.server import app
@@ -274,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")
@@ -282,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:
@@ -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:
@@ -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)
+5 -1
View File
@@ -195,14 +195,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(
+81 -26
View File
@@ -20,25 +20,33 @@ 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
from gns3server import schemas
from gns3server.agent.web_wireshark.manager import WebWiresharkManager
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError
from gns3server.controller.controller_error import ControllerError, ControllerNotFoundError
from gns3server.controller.link import _UNSET, Link
from gns3server.db.repositories.rbac import RbacRepository
from gns3server.utils.http_client import HTTPClient
from gns3server.utils.port_allocator import link_id_to_port
from gns3server.utils.websocket_to_websocket import websocket_proxy
from .dependencies.concurrency import (
GET_RESPONSES,
PUT_RESPONSES,
check_if_match,
if_match_header,
serialize_updates,
set_etag,
)
from .dependencies.database import get_repository
from .dependencies.rbac import has_privilege, has_privilege_on_websocket
@@ -51,6 +59,29 @@ responses: dict[int | str, dict[str, Any]] = {
router = APIRouter(responses=responses)
def _select_port(node, endpoint, other_port):
if endpoint.get("port_name") is not None:
port = node.get_port_by_name(endpoint["port_name"])
if port is None:
raise ControllerNotFoundError(f"Port named {endpoint['port_name']} for {node.name} not found")
return port
if endpoint.get("port") == "auto":
port = node.get_free_port(other_port.link_type if other_port else None)
if port is None:
raise ControllerError(f"No free port available on {node.name}", code="no_free_port")
return port
return None
def _peer_port(project, endpoint):
if endpoint.get("port") == "auto":
return None
node = project.get_node(endpoint["node_id"])
if endpoint.get("port_name") is not None:
return node.get_port_by_name(endpoint["port_name"])
return node.get_port(endpoint["adapter_number"], endpoint["port_number"])
async def dep_link(project_id: UUID, link_id: UUID) -> Link:
"""
Dependency to retrieve a link.
@@ -110,21 +141,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.
@@ -138,43 +178,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(
+119 -33
View File
@@ -34,13 +34,25 @@ from fastapi.routing import APIRoute
from gns3server import schemas
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
@@ -151,6 +163,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.
@@ -160,6 +173,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":
@@ -168,6 +182,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 = []
@@ -251,8 +268,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.
@@ -261,31 +283,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(
@@ -326,46 +367,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.
@@ -375,24 +458,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(
+71 -15
View File
@@ -34,12 +34,24 @@ 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.responses import NDJSON_MEDIA_TYPE, NOTIFICATION_STREAM_RESPONSES, NDJSONResponse
from gns3server.controller import Controller, marker_replay
from gns3server.controller.controller_error import ControllerBadRequestError, ControllerError
from gns3server.controller.export_project import export_project as export_controller_project
@@ -55,6 +67,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
@@ -78,11 +98,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()
@@ -91,7 +115,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())
@@ -102,6 +126,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())
@@ -109,6 +135,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())
@@ -142,34 +170,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(
@@ -487,7 +532,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"))],
@@ -523,10 +568,20 @@ async def load_project(path: str = Body(..., embed=True)) -> schemas.Project:
return project.asdict()
@router.get("/{project_id}/notifications", dependencies=[Depends(has_privilege("Project.Audit"))])
@router.get(
"/{project_id}/notifications",
dependencies=[Depends(has_privilege("Project.Audit"))],
response_class=NDJSONResponse,
responses=NOTIFICATION_STREAM_RESPONSES,
)
async def project_http_notifications(project_id: UUID) -> StreamingResponse:
"""
Receive project notifications about the controller from HTTP stream.
Receive project notifications from an HTTP stream of newline delimited JSON objects (see the Notification schema).
The same messages are available as WebSocket text frames on `/v3/projects/{project_id}/notifications/ws`.
Marker events (`marker.match`) are not sent here, they are only available on
`/v3/projects/{project_id}/notifications/markers/ws`.
Project actions: `node.*`, `link.*`, `drawing.*`, `project.updated`, `snapshot.restored`, `log.*` and `ping`.
Required privilege: Project.Audit
"""
@@ -555,7 +610,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")
@@ -565,7 +620,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
"""
@@ -603,8 +659,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
@@ -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"))],
)
@@ -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 = []
+3 -17
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)
+30 -39
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.config import Config
@@ -73,9 +74,16 @@ 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",
@@ -159,82 +167,65 @@ async def redoc_html():
@app.exception_handler(ControllerError)
async def controller_error_handler(request: Request, exc: ControllerError):
log.error(f"Controller error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content={"message": str(exc)},
)
return error_response(status.HTTP_409_CONFLICT, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerTimeoutError)
async def controller_timeout_error_handler(request: Request, exc: ControllerTimeoutError):
log.error(f"Controller timeout error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_408_REQUEST_TIMEOUT,
content={"message": str(exc)},
)
return error_response(status.HTTP_408_REQUEST_TIMEOUT, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerUnauthorizedError)
async def controller_unauthorized_error_handler(request: Request, exc: ControllerUnauthorizedError):
log.error(f"Controller unauthorized error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"message": str(exc)},
)
return error_response(status.HTTP_401_UNAUTHORIZED, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerForbiddenError)
async def controller_forbidden_error_handler(request: Request, exc: ControllerForbiddenError):
log.error(f"Controller forbidden error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_403_FORBIDDEN,
content={"message": str(exc)},
)
return error_response(status.HTTP_403_FORBIDDEN, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerNotFoundError)
async def controller_not_found_error_handler(request: Request, exc: ControllerNotFoundError):
log.error(f"Controller not found error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={"message": str(exc)},
)
return error_response(status.HTTP_404_NOT_FOUND, str(exc), exc.code, exc.details)
@app.exception_handler(ControllerBadRequestError)
async def controller_bad_request_error_handler(request: Request, exc: ControllerBadRequestError):
log.error(f"Controller bad request error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={"message": str(exc)},
)
return error_response(status.HTTP_400_BAD_REQUEST, str(exc), exc.code, exc.details)
@app.exception_handler(ComputeConflictError)
async def compute_conflict_error_handler(request: Request, exc: ComputeConflictError):
log.error(f"Controller received error from compute for request '{exc.url()}': {exc}")
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content={"message": str(exc)},
)
return error_response(status.HTTP_409_CONFLICT, str(exc), exc.code, exc.details)
# make sure the content key is "message", not "detail" per default
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(status_code=exc.status_code, content={"message": exc.detail}, headers=exc.headers)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return error_response(exc.status_code, exc.detail, headers=exc.headers)
@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_error_handler(request: Request, exc: SQLAlchemyError):
log.error(f"Controller database error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"message": "Database error detected, please check logs to find details"},
return error_response(
status.HTTP_500_INTERNAL_SERVER_ERROR,
"Database error detected, please check logs to find details",
"database_error",
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
log.error(f"Request validation error in {request.url.path} ({request.method}): {exc}")
return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, content={"message": str(exc)})
errors = [
{"loc": list(error.get("loc", ())), "msg": error.get("msg"), "type": error.get("type")}
for error in exc.errors()
]
return error_response(status.HTTP_422_UNPROCESSABLE_CONTENT, str(exc), "validation_error", {"errors": errors})