arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-25 09:48:09 +00:00
Code Issues Releases Wiki Activity

Merge remote-tracking branch 'origin/3.0' into gh-pages

Browse Source
This commit is contained in:
github-actions 2021-08-17 12:27:31 +00:00
commit 4bccfd4e06
5 changed files with 102 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
gns3server/api/routes/controller/dependencies/authentication.py
View File

@ -61,11 +61,7 @@ async def get_current_active_user(
)
# remove the prefix (e.g. "/v3") from URL path
match = re.search(r"^(/v[0-9]+).*", request.url.path)
if match:
path = request.url.path[len(match.group(1)):]
else:
path = request.url.path
path = re.sub(r"^/v[0-9]", "", request.url.path)
# special case: always authorize access to the "/users/me" endpoint
if path == "/users/me":

48
gns3server/api/routes/controller/permissions.py
View File

@ -19,10 +19,14 @@
API routes for permissions.
"""
from fastapi import APIRouter, Depends, Response, status
import re
from fastapi import APIRouter, Depends, Response, 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,
@ -53,6 +57,7 @@ async def get_permissions(
@router.post("", response_model=schemas.Permission, status_code=status.HTTP_201_CREATED)
async def create_permission(
request: Request,
permission_create: schemas.PermissionCreate,
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
) -> schemas.Permission:
@ -60,11 +65,32 @@ async def create_permission(
Create a new permission.
"""
if await rbac_repo.check_permission_exists(permission_create):
raise ControllerBadRequestError(f"Permission '{permission_create.methods} {permission_create.path} "
f"{permission_create.action}' already exists")
# TODO: should we prevent having multiple permissions with same methods/path?
#if await rbac_repo.check_permission_exists(permission_create):
# raise ControllerBadRequestError(f"Permission '{permission_create.methods} {permission_create.path} "
# f"{permission_create.action}' already exists")
return await rbac_repo.create_permission(permission_create)
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 an 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 an word matching regex
route_path = re.sub(r"/{[\w:]+}", r"/\\w+", route_path)
# the permission can match multiple routes
if permission_create.path.endswith("/*"):
route_path += r"/\*"
if re.fullmatch(route_path, permission_create.path):
for method in permission_create.methods:
if method in list(route.methods):
return await rbac_repo.create_permission(permission_create)
raise ControllerBadRequestError(f"Permission '{permission_create.methods} {permission_create.path}' "
f"doesn't match any existing endpoint")
@router.get("/{permission_id}", response_model=schemas.Permission)
@ -117,3 +143,15 @@ async def delete_permission(
raise ControllerNotFoundError(f"Permission '{permission_id}' could not be deleted")
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.post("/prune", status_code=status.HTTP_204_NO_CONTENT)
async def prune_permissions(
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
) -> Response:
"""
Prune orphaned permissions.
"""
await rbac_repo.prune_permissions()
return Response(status_code=status.HTTP_204_NO_CONTENT)

33
gns3server/db/repositories/rbac.py
View File

@ -17,7 +17,7 @@
from uuid import UUID
from typing import Optional, List, Union
from sqlalchemy import select, update, delete
from sqlalchemy import select, update, delete, null
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@ -194,7 +194,8 @@ class RbacRepository(BaseRepository):
Get all permissions.
"""
query = select(models.Permission)
query = select(models.Permission).\
order_by(models.Permission.path.desc())
result = await self._db_session.execute(query)
return result.scalars().all()
@ -257,6 +258,22 @@ class RbacRepository(BaseRepository):
await self._db_session.commit()
return result.rowcount > 0
async def prune_permissions(self) -> int:
"""
Prune orphaned permissions.
"""
query = select(models.Permission).\
filter((~models.Permission.roles.any()) & (models.Permission.user_id == null()))
result = await self._db_session.execute(query)
permissions = result.scalars().all()
permissions_deleted = 0
for permission in permissions:
if await self.delete_permission(permission.permission_id):
permissions_deleted += 1
log.info(f"{permissions_deleted} orphaned permissions have been deleted")
return permissions_deleted
def _match_permission(
self,
permissions: List[models.Permission],
@ -282,9 +299,9 @@ class RbacRepository(BaseRepository):
"""
query = select(models.Permission).\
join(models.User.permissions). \
join(models.User.permissions).\
filter(models.User.user_id == user_id).\
order_by(models.Permission.path)
order_by(models.Permission.path.desc())
result = await self._db_session.execute(query)
return result.scalars().all()
@ -379,11 +396,11 @@ class RbacRepository(BaseRepository):
"""
query = select(models.Permission).\
join(models.Permission.roles). \
join(models.Role.groups). \
join(models.UserGroup.users). \
join(models.Permission.roles).\
join(models.Role.groups).\
join(models.UserGroup.users).\
filter(models.User.user_id == user_id).\
order_by(models.Permission.path)
order_by(models.Permission.path.desc())
result = await self._db_session.execute(query)
permissions = result.scalars().all()

2
requirements.txt
View File

@ -1,4 +1,4 @@
uvicorn==0.14.0
uvicorn==0.15.0
fastapi==0.68.0
websockets==9.1
python-multipart==0.0.5

35
tests/api/routes/controller/test_permissions.py
View File

@ -32,12 +32,32 @@ class TestPermissionRoutes:
new_permission = {
"methods": ["GET"],
"path": "/templates",
"path": "/templates/f6113095-a703-4967-b039-ab95ac3eb4f5",
"action": "ALLOW"
}
response = await client.post(app.url_path_for("create_permission"), json=new_permission)
assert response.status_code == status.HTTP_201_CREATED
async def test_create_wildcard_permission(self, app: FastAPI, client: AsyncClient) -> None:
new_permission = {
"methods": ["GET"],
"path": "/templates/*",
"action": "ALLOW"
}
response = await client.post(app.url_path_for("create_permission"), json=new_permission)
assert response.status_code == status.HTTP_201_CREATED
async def test_create_invalid_permission(self, app: FastAPI, client: AsyncClient) -> None:
new_permission = {
"methods": ["GET"],
"path": "/templates/invalid",
"action": "ALLOW"
}
response = await client.post(app.url_path_for("create_permission"), json=new_permission)
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_get_permission(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
rbac_repo = RbacRepository(db_session)
@ -50,12 +70,12 @@ class TestPermissionRoutes:
response = await client.get(app.url_path_for("get_permissions"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 6 # 5 default permissions + 1 custom permission
assert len(response.json()) == 7 # 5 default permissions + 2 custom permissions
async def test_update_permission(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
rbac_repo = RbacRepository(db_session)
permission_in_db = await rbac_repo.get_permission_by_path("/templates")
permission_in_db = await rbac_repo.get_permission_by_path("/templates/*")
update_permission = {
"methods": ["GET"],
@ -81,3 +101,12 @@ class TestPermissionRoutes:
permission_in_db = await rbac_repo.get_permission_by_path("/appliances")
response = await client.delete(app.url_path_for("delete_permission", permission_id=permission_in_db.permission_id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_prune_permissions(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
response = await client.post(app.url_path_for("prune_permissions"))
assert response.status_code == status.HTTP_204_NO_CONTENT
rbac_repo = RbacRepository(db_session)
permissions_in_db = await rbac_repo.get_permissions()
assert len(permissions_in_db) == 5 # 5 default permissions