Merge pull request #2964 from markparonyan/api-issue-17-asr039

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
 
@@ -143,43 +151,58 @@ async def get_filters(link: Link = Depends(dep_link)) -> List[schemas.LinkFilter
     "/{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
@@ -56,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
 
@@ -151,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(

tests/api/routes/controller/test_links.py

@@ -143,6 +143,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:

tests/api/routes/controller/test_nodes.py

@@ -250,6 +250,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()

tests/api/routes/controller/test_projects.py

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