fix: breaking change with dependency FastAPI v0.137.0

https://fastapi.tiangolo.com/release-notes/#specific-breaking-changes

gns3server/api/routes/controller/acl.py

@@ -22,9 +22,10 @@ API routes for ACL.
 import re
 
 from fastapi import APIRouter, Depends, Request, status
-from fastapi.routing import APIRoute
+from fastapi.routing import APIRoute, _IncludedRouter
+from starlette.routing import BaseRoute, Mount
 from uuid import UUID
-from typing import List
+from typing import Iterator, List, Sequence
 
 
 from gns3server import schemas
@@ -48,6 +49,39 @@ log = logging.getLogger(__name__)
 router = APIRouter()
 
 
+def _join_paths(prefix: str, path: str) -> str:
+
+    if not prefix:
+        return path
+
+    normalized_prefix = prefix
+    while normalized_prefix.endswith("/"):
+        normalized_prefix = normalized_prefix[:-1]
+
+    normalized_path = path
+    while normalized_path.startswith("/"):
+        normalized_path = normalized_path[1:]
+
+    return f"{normalized_prefix}/{normalized_path}".rstrip("/")
+
+
+def _iter_route_paths(routes: Sequence[BaseRoute], prefix: str = "", include_mounted_routes=False) -> Iterator[str]:
+    for route in routes:
+        if isinstance(route, _IncludedRouter):
+            include_prefix = route.include_context.prefix or ""
+            yield from _iter_route_paths(route.original_router.routes, _join_paths(prefix, include_prefix), include_mounted_routes)
+            continue
+
+        if isinstance(route, APIRoute):
+            yield _join_paths(prefix, route.path)
+            continue
+
+        if isinstance(route, Mount) and include_mounted_routes:
+            mounted_routes = getattr(route, "routes", None)
+            if isinstance(mounted_routes, Sequence):
+                yield from _iter_route_paths(mounted_routes, _join_paths(prefix, route.path), include_mounted_routes)
+
+
 @router.get(
     "/endpoints",
     status_code=status.HTTP_201_CREATED,
@@ -179,19 +213,21 @@ async def create_ace(
     Required privilege: ACE.Allocate
     """
 
-    for route in request.app.routes:
-        if isinstance(route, APIRoute):
+    for route_path in _iter_route_paths(request.app.routes, include_mounted_routes=True):
+        print(route_path)
 
-            # remove the prefix (e.g. "/v3") from the route path
-            route_path = re.sub(r"^/v[0-9]", "", route.path)
-            # replace route path ID parameters by a UUID regex
-            route_path = re.sub(r"{\w+_id}", "[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}", route_path)
-            # replace remaining route path parameters by a word matching regex
-            route_path = re.sub(r"/{[\w:]+}", r"/\\w+", route_path)
+    for route_path in _iter_route_paths(request.app.routes, include_mounted_routes=True):
 
-            if re.fullmatch(route_path, ace_create.path):
-                log.info(f"Creating ACE for route path {route_path}")
-                return await rbac_repo.create_ace(ace_create)
+        # remove the prefix (e.g. "/v3") from the route path
+        normalized_path = re.sub(r"^/v[0-9]", "", route_path)
+        # replace route path ID parameters by a UUID regex
+        normalized_path = re.sub(r"{\w+_id}", "[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}", normalized_path)
+        # replace remaining route path parameters by a word matching regex
+        normalized_path = re.sub(r"/{[\w:]+}", r"/\\w+", normalized_path)
+
+        if re.fullmatch(normalized_path, ace_create.path):
+            log.info(f"Creating ACE for route path {route_path}")
+            return await rbac_repo.create_ace(ace_create)
 
     raise ControllerBadRequestError(f"Path '{ace_create.path}' doesn't match any existing endpoint")
 

tests/api/routes/controller/test_acl.py

@@ -87,6 +87,7 @@ class TestACLRoutes:
         response = await authorized_client.post(app.url_path_for("create_project"), json={"name": "test"})
         assert response.status_code == status.HTTP_201_CREATED
 
+
     async def test_create_ace_not_existing_endpoint(
             self,
             app: FastAPI,

tests/api/routes/test_routes.py

@@ -17,11 +17,12 @@
 
 import pytest
 from fastapi import FastAPI, status
-from fastapi.routing import APIRoute, APIWebSocketRoute
-from starlette.routing import Mount
+from fastapi.routing import APIRoute, APIWebSocketRoute, _IncludedRouter
+from starlette.routing import BaseRoute, Mount
 from httpx import AsyncClient
 from httpx_ws import aconnect_ws
 from httpx_ws.transport import ASGIWebSocketTransport
+from typing import Iterator, Sequence, Tuple
 
 
 
@@ -46,6 +47,43 @@ ALLOWED_CONTROLLER_ENDPOINTS = [
     ("/v3/mcp/", "GET"),
 ]
 
+
+def _join_paths(prefix: str, path: str) -> str:
+    if not prefix:
+        return path
+
+    normalized_prefix = prefix
+    while normalized_prefix.endswith("/"):
+        normalized_prefix = normalized_prefix[:-1]
+
+    normalized_path = path
+    while normalized_path.startswith("/"):
+        normalized_path = normalized_path[1:]
+
+    return f"{normalized_prefix}/{normalized_path}"
+
+
+def _iter_routes(
+        routes: Sequence[BaseRoute],
+        prefix: str = "",
+        include_mounted_routes: bool = False
+) -> Iterator[Tuple[str, BaseRoute]]:
+    for route in routes:
+        if isinstance(route, _IncludedRouter):
+            include_prefix = route.include_context.prefix or ""
+            yield from _iter_routes(route.original_router.routes, _join_paths(prefix, include_prefix), include_mounted_routes)
+            continue
+
+        if isinstance(route, (APIRoute, APIWebSocketRoute)):
+            yield _join_paths(prefix, route.path), route
+            continue
+
+        if isinstance(route, Mount) and include_mounted_routes:
+            mounted_routes = getattr(route, "routes", None)
+            if isinstance(mounted_routes, Sequence):
+                yield from _iter_routes(mounted_routes, _join_paths(prefix, route.path), include_mounted_routes)
+
+
 class TestRoutes:
 
     # Controller endpoints have a OAuth2 bearer token authentication
@@ -55,16 +93,17 @@ class TestRoutes:
             unauthorized_client: AsyncClient
     ) -> None:
 
-        for route in app.routes:
+        for path, route in _iter_routes(app.routes):
             if isinstance(route, APIRoute):
                 for method in list(route.methods):
-                    if (route.path, method) not in ALLOWED_CONTROLLER_ENDPOINTS:
-                        response = await getattr(unauthorized_client, method.lower())(route.path)
-                        assert response.status_code == status.HTTP_401_UNAUTHORIZED
-            elif isinstance(route, APIWebSocketRoute):
+                    if (path, method) not in ALLOWED_CONTROLLER_ENDPOINTS:
+                        request_path = path.rstrip("/")
+                        response = await getattr(unauthorized_client, method.lower())(request_path)
+                        assert response.status_code == status.HTTP_401_UNAUTHORIZED, f"{method} {request_path} -> {response.status_code}"
+            elif isinstance(route, APIWebSocketRoute) and not path.startswith("/v3/compute"):
                 params = {"token": "wrong_token"}
                 async with AsyncClient(base_url="http://test-api", transport=ASGIWebSocketTransport(app=app)) as client:
-                    async with aconnect_ws(route.path, client, params=params) as ws:
+                    async with aconnect_ws(path, client, params=params) as ws:
                         json_notification = await ws.receive_json()
                         assert json_notification['event'] == {
                             'message': 'Could not authenticate while connecting to controller WebSocket: Could not validate credentials'
@@ -78,17 +117,21 @@ class TestRoutes:
             unauthorized_client: AsyncClient
     ) -> None:
 
-        for route in app.routes:
-            if isinstance(route, Mount):
-                for compute_route in route.routes:
-                    if isinstance(compute_route, APIRoute):
-                        for method in list(compute_route.methods):
-                            response = await getattr(unauthorized_client, method.lower())(route.path + compute_route.path)
-                            assert response.status_code == status.HTTP_401_UNAUTHORIZED
-                    elif isinstance(compute_route, APIWebSocketRoute):
-                        async with AsyncClient(base_url="http://test-api", transport=ASGIWebSocketTransport(app=app)) as client:
-                            async with aconnect_ws(route.path + compute_route.path, client, auth=("wrong_user", "password123")) as ws:
-                                json_notification = await ws.receive_json()
-                                assert json_notification['event'] == {
-                                    'message': 'Could not authenticate while connecting to compute WebSocket: Could not validate credentials'
-                                }
+        for path, route in _iter_routes(app.routes, include_mounted_routes=True):
+            if not path.startswith("/v3/compute"):
+                continue
+
+            if isinstance(route, APIRoute):
+                for method in list(route.methods):
+                    request_path = path.rstrip("/")
+                    response = await getattr(unauthorized_client, method.lower())(request_path)
+                    #if response.status_code == status.HTTP_307_TEMPORARY_REDIRECT:
+                    #    response = await getattr(unauthorized_client, method.lower())(response.headers["location"])
+                    assert response.status_code == status.HTTP_401_UNAUTHORIZED, f"{method} {request_path} -> {response.status_code}"
+            elif isinstance(route, APIWebSocketRoute):
+                async with AsyncClient(base_url="http://test-api", transport=ASGIWebSocketTransport(app=app)) as client:
+                    async with aconnect_ws(path, client, auth=("wrong_user", "password123")) as ws:
+                        json_notification = await ws.receive_json()
+                        assert json_notification['event'] == {
+                            'message': 'Could not authenticate while connecting to compute WebSocket: Could not validate credentials'
+                        }