Merge branch '3.1' into fix/docker-node-create-setattr-noise

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