#!/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 from uuid import UUID from typing import List from gns3server import schemas from gns3server.controller.controller_error import ( ControllerBadRequestError, ControllerNotFoundError ) from gns3server.controller import Controller 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() @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 endpoint not 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") # nodes add_to_endpoints(f"/projects/{project.id}/nodes", f'All nodes in project "{project.name}"', "node") for node in project.nodes.values(): 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 project.links.values(): node_id_1 = link["nodes"][0]["node_id"] node_id_2 = link["nodes"][1]["node_id"] node_name_1 = project.nodes[node_id_1]["name"] node_name_2 = project.nodes[node_id_2]["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"/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"/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"/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") 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[schemas.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)) ) -> schemas.ACE: """ Create a new ACL entry. Required privilege: ACE.Allocate """ for route in request.app.routes: if isinstance(route, APIRoute): # 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) if re.fullmatch(route_path, ace_create.path): log.info("Creating ACE for route path", ace_create.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)), ) -> schemas.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)) ) -> schemas.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")