#!/usr/bin/env python # # Copyright (C) 2023 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 . """ API routes for ACL. """ import re from fastapi import APIRouter, Depends, Request, status from fastapi.routing import APIRoute, _IncludedRouter from starlette.routing import BaseRoute, Mount from uuid import UUID from typing import Iterator, List, Optional, Sequence from gns3server import schemas from gns3server.controller.controller_error import ControllerBadRequestError, ControllerNotFoundError from gns3server.controller import Controller import gns3server.db.models as models from gns3server.db.repositories.users import UsersRepository from gns3server.db.repositories.rbac import RbacRepository from gns3server.db.repositories.images import ImagesRepository from gns3server.db.repositories.templates import TemplatesRepository from .dependencies.database import get_repository from .dependencies.rbac import has_privilege import logging 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, dependencies=[Depends(has_privilege("ACE.Audit"))]) async def endpoints( users_repo: UsersRepository = Depends(get_repository(UsersRepository)), rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)), images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)), templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)), ) -> List[dict]: """ List all endpoints to be used in ACL entries. """ controller = Controller.instance() endpoints = [{"endpoint": "/", "name": "All endpoints", "endpoint_type": "root"}] def add_to_endpoints(endpoint: str, name: str, endpoint_type: str) -> None: if not any(e["endpoint"] == endpoint for e in endpoints): endpoints.append({"endpoint": endpoint, "name": name, "endpoint_type": endpoint_type}) # projects add_to_endpoints("/projects", "All projects", "project") projects = [p for p in controller.projects.values()] for project in projects: add_to_endpoints(f"/projects/{project.id}", f'Project "{project.name}"', "project") if project.status == "closed": nodes = project.nodes.values() links = project.links.values() else: nodes = [v.asdict() for v in project.nodes.values()] links = [v.asdict() for v in project.links.values()] # nodes add_to_endpoints(f"/projects/{project.id}/nodes", f'All nodes in project "{project.name}"', "node") for node in nodes: add_to_endpoints( f"/projects/{project.id}/nodes/{node['node_id']}", f'Node "{node["name"]}" in project "{project.name}"', endpoint_type="node", ) # links add_to_endpoints(f"/projects/{project.id}/links", f'All links in project "{project.name}"', "link") for link in links: node_id_1 = link["nodes"][0]["node_id"] node_id_2 = link["nodes"][1]["node_id"] node_name_1 = node_name_2 = "N/A" for node in nodes: if node["node_id"] == node_id_1: node_name_1 = node["name"] if node["node_id"] == node_id_2: node_name_2 = node["name"] add_to_endpoints( f"/projects/{project.id}/links/{link['link_id']}", f'Link from "{node_name_1}" to "{node_name_2}" in project "{project.name}"', endpoint_type="link", ) # users add_to_endpoints("/access/users", "All users", "user") users = await users_repo.get_users() for user in users: add_to_endpoints(f"/access/users/{user.user_id}", f'User "{user.username}"', "user") # groups add_to_endpoints("/access/groups", "All groups", "group") groups = await users_repo.get_user_groups() for group in groups: add_to_endpoints(f"/access/groups/{group.user_group_id}", f'Group "{group.name}"', "group") # roles add_to_endpoints("/access/roles", "All roles", "role") roles = await rbac_repo.get_roles() for role in roles: add_to_endpoints(f"/access/roles/{role.role_id}", f'Role "{role.name}"', "role") # images add_to_endpoints("/images", "All images", "image") images = await images_repo.get_images() for image in images: add_to_endpoints(f"/images/{image.filename}", f'Image "{image.filename}"', "image") # templates add_to_endpoints("/templates", "All templates", "template") templates = await templates_repo.get_templates() for template in templates: add_to_endpoints(f"/templates/{template.template_id}", f'Template "{template.name}"', "template") # Resource pools are not included in "all endpoints" to prevent accidental access # They must be explicitly configured for team sharing return endpoints @router.get("", response_model=List[schemas.ACE], dependencies=[Depends(has_privilege("ACE.Audit"))]) async def get_aces(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[models.ACE]: """ Get all ACL entries. Required privilege: ACE.Audit """ return await rbac_repo.get_aces() @router.post( "", response_model=schemas.ACE, status_code=status.HTTP_201_CREATED, dependencies=[Depends(has_privilege("ACE.Allocate"))], ) async def create_ace( request: Request, ace_create: schemas.ACECreate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)) ) -> models.ACE: """ Create a new ACL entry. Required privilege: ACE.Allocate """ for route_path in _iter_route_paths(request.app.routes, include_mounted_routes=True): print(route_path) for route_path in _iter_route_paths(request.app.routes, include_mounted_routes=True): # 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") @router.get("/{ace_id}", response_model=schemas.ACE, dependencies=[Depends(has_privilege("ACE.Audit"))]) async def get_ace( ace_id: UUID, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)), ) -> models.ACE: """ Get an ACL entry. Required privilege: ACE.Audit """ ace = await rbac_repo.get_ace(ace_id) if not ace: raise ControllerNotFoundError(f"ACL entry '{ace_id}' not found") return ace @router.put("/{ace_id}", response_model=schemas.ACE, dependencies=[Depends(has_privilege("ACE.Modify"))]) async def update_ace( ace_id: UUID, ace_update: schemas.ACEUpdate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)) ) -> Optional[models.ACE]: """ Update an ACL entry. Required privilege: ACE.Modify """ ace = await rbac_repo.get_ace(ace_id) if not ace: raise ControllerNotFoundError(f"ACL entry '{ace_id}' not found") return await rbac_repo.update_ace(ace_id, ace_update) @router.delete( "/{ace_id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(has_privilege("ACE.Allocate"))] ) async def delete_ace( ace_id: UUID, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)), ) -> None: """ Delete an ACL entry. Required privilege: ACE.Allocate """ ace = await rbac_repo.get_ace(ace_id) if not ace: raise ControllerNotFoundError(f"ACL entry '{ace_id}' not found") success = await rbac_repo.delete_ace(ace_id) if not success: raise ControllerNotFoundError(f"ACL entry '{ace_id}' could not be deleted")