diff --git a/gns3server/api/responses.py b/gns3server/api/responses.py new file mode 100644 index 000000000..55330c038 --- /dev/null +++ b/gns3server/api/responses.py @@ -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 . + +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.", + } +} diff --git a/gns3server/api/routes/controller/controller.py b/gns3server/api/routes/controller/controller.py index e1f03b7e3..cc32fb90f 100644 --- a/gns3server/api/routes/controller/controller.py +++ b/gns3server/api/routes/controller/controller.py @@ -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 @@ -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: diff --git a/gns3server/api/routes/controller/projects.py b/gns3server/api/routes/controller/projects.py index fab2691d2..6972dee3f 100644 --- a/gns3server/api/routes/controller/projects.py +++ b/gns3server/api/routes/controller/projects.py @@ -40,6 +40,7 @@ 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 @@ -523,10 +524,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 +566,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 +576,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 +615,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 diff --git a/gns3server/api/server.py b/gns3server/api/server.py index 4ea037e83..d1d59354d 100644 --- a/gns3server/api/server.py +++ b/gns3server/api/server.py @@ -76,7 +76,14 @@ def get_application() -> FastAPI: application = FastAPI( 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, diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py index a5eb8406c..b01b737ed 100644 --- a/gns3server/schemas/__init__.py +++ b/gns3server/schemas/__init__.py @@ -109,6 +109,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 @@ -261,6 +262,8 @@ __all__ = [ "NodeDuplicate", "NodeFile", "NodeUpdate", + "Notification", + "NotificationAction", "OpenAIMessage", "OpenAIToolCall", "Privilege", diff --git a/gns3server/schemas/controller/notifications.py b/gns3server/schemas/controller/notifications.py new file mode 100644 index 000000000..11ed85e4c --- /dev/null +++ b/gns3server/schemas/controller/notifications.py @@ -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 . + +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") diff --git a/tests/api/routes/controller/test_notifications.py b/tests/api/routes/controller/test_notifications.py new file mode 100644 index 000000000..ff6209e19 --- /dev/null +++ b/tests/api/routes/controller/test_notifications.py @@ -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 . + +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"