feat: add optional ETag/If-Match concurrency for projects, nodes and links

gns3server/api/routes/controller/dependencies/concurrency.py

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

gns3server/api/routes/controller/links.py

@@ -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
 
@@ -39,6 +39,14 @@ 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
 
@@ -138,43 +146,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(

gns3server/api/routes/controller/nodes.py

@@ -41,6 +41,14 @@ from gns3server.db.repositories.rbac import RbacRepository
 from gns3server.utils import force_unix_path
 from gns3server.utils.http_client import HTTPClient
 
+from .dependencies.concurrency import (
+    GET_RESPONSES,
+    PUT_RESPONSES,
+    check_if_match,
+    if_match_header,
+    serialize_updates,
+    set_etag,
+)
 from .dependencies.database import get_repository
 from .dependencies.rbac import has_privilege, has_privilege_on_websocket
 
@@ -256,8 +264,13 @@ async def reload_all_nodes(project: Project = Depends(dep_project)) -> None:
 _HOST_INTERFACE_NODE_TYPES = {"cloud", "nat"}
 
 
-@router.get("/{node_id}", response_model=schemas.Node, dependencies=[Depends(has_privilege("Node.Audit"))])
-async def get_node(node: Node = Depends(dep_node)) -> schemas.Node:
+@router.get(
+    "/{node_id}",
+    response_model=schemas.Node,
+    responses=GET_RESPONSES,
+    dependencies=[Depends(has_privilege("Node.Audit"))],
+)
+async def get_node(response: Response, node: Node = Depends(dep_node)) -> schemas.Node:
     """
     Return a node from a given project.
 
@@ -266,31 +279,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(

gns3server/api/routes/controller/projects.py

@@ -34,7 +34,18 @@ log = logging.getLogger()
 from typing import Any, List, Optional
 from uuid import UUID
 
-from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request, WebSocket, WebSocketDisconnect, status
+from fastapi import (
+    APIRouter,
+    Body,
+    Depends,
+    HTTPException,
+    Query,
+    Request,
+    Response,
+    WebSocket,
+    WebSocketDisconnect,
+    status,
+)
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import FileResponse, StreamingResponse
 from websockets.exceptions import ConnectionClosed, WebSocketException
@@ -55,6 +66,14 @@ from gns3server.utils.asyncio import aiozipstream
 from gns3server.utils.path import is_safe_path
 
 from .dependencies.authentication import get_current_active_user
+from .dependencies.concurrency import (
+    GET_RESPONSES,
+    PUT_RESPONSES,
+    check_if_match,
+    if_match_header,
+    serialize_updates,
+    set_etag,
+)
 from .dependencies.database import get_repository
 from .dependencies.rbac import has_privilege, has_privilege_on_websocket
 
@@ -150,34 +169,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(