feat(api): add machine-readable error codes and align OpenAPI error schemas

gns3server/api/errors.py

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

gns3server/api/server.py

@@ -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,9 +72,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",
@@ -151,82 +159,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})