Improve permission handling

README.md

@@ -117,11 +117,10 @@ The structure of gooseBit is as follows:
 -   `updater`: DDI API handler and device update manager.
 -   `updates`: SWUpdate file parsing.
 -   `realtime`: Realtime API functionality with websockets.
--   `auth`: Authentication functions.
+-   `auth`: Authentication functions and permission handling.
 -   `models`: Database models.
 -   `db`: Database config and initialization.
 -   `schema`: Pydantic models used for API type hinting.
--   `permissions`: Permission handling and permission enums.
 -   `settings`: Settings loader and handler.
 -   `telemetry`: Telemetry data handlers.
 -   `routes`: Routes for a giving endpoint, including the router.

docs/development.md

@@ -72,11 +72,10 @@ The structure of gooseBit is as follows:
 -   `updater`: DDI API handler and device update manager.
 -   `updates`: SWUpdate file parsing.
 -   `realtime`: Realtime API functionality with websockets.
--   `auth`: Authentication functions.
+-   `auth`: Authentication functions and permission handling.
 -   `models`: Database models.
 -   `db`: Database config and initialization.
 -   `schema`: Pydantic models used for API type hinting.
--   `permissions`: Permission handling and permission enums.
 -   `settings`: Settings loader and handler.
 -   `telemetry`: Telemetry data handlers.
 -   `routes`: Routes for a giving endpoint, including the router.

goosebit/api/v1/devices/device/routes.py

@@ -5,23 +5,22 @@
 
 from goosebit.api.v1.devices.device.responses import DeviceLogResponse, DeviceResponse
 from goosebit.auth import validate_user_permissions
-from goosebit.permissions import Permissions
 from goosebit.updater.manager import UpdateManager, get_update_manager
 
 router = APIRouter(prefix="/{dev_id}")
 
 
 @router.get(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.HOME.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["home.read"])],
 )
 async def device_get(_: Request, updater: UpdateManager = Depends(get_update_manager)) -> DeviceResponse:
     return await DeviceResponse.convert(await updater.get_device())
 
 
 @router.get(
     "/log",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.HOME.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["home.read"])],
 )
 async def device_logs(_: Request, updater: UpdateManager = Depends(get_update_manager)) -> DeviceLogResponse:
     device = await updater.get_device()

goosebit/api/v1/devices/routes.py

@@ -6,7 +6,6 @@
 from goosebit.api.responses import StatusResponse
 from goosebit.auth import validate_user_permissions
 from goosebit.models import Device, Software, UpdateModeEnum
-from goosebit.permissions import Permissions
 from goosebit.updater.manager import delete_devices, get_update_manager
 
 from . import device
@@ -18,15 +17,15 @@
 
 @router.get(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.HOME.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["home.read"])],
 )
 async def devices_get(_: Request) -> DevicesResponse:
     return await DevicesResponse.convert(await Device.all().prefetch_related("assigned_software", "hardware"))
 
 
 @router.patch(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.DEVICE.WRITE])],
+    dependencies=[Security(validate_user_permissions, scopes=["device.write"])],
 )
 async def devices_patch(_: Request, config: DevicesPatchRequest) -> StatusResponse:
     for uuid in config.devices:
@@ -52,7 +51,7 @@ async def devices_patch(_: Request, config: DevicesPatchRequest) -> StatusRespon
 
 @router.delete(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.DEVICE.DELETE])],
+    dependencies=[Security(validate_user_permissions, scopes=["device.delete"])],
 )
 async def devices_delete(_: Request, config: DevicesDeleteRequest) -> StatusResponse:
     await delete_devices(config.devices)

goosebit/api/v1/rollouts/routes.py

@@ -4,7 +4,6 @@
 from goosebit.api.responses import StatusResponse
 from goosebit.auth import validate_user_permissions
 from goosebit.models import Rollout
-from goosebit.permissions import Permissions
 
 from .requests import RolloutsDeleteRequest, RolloutsPatchRequest, RolloutsPutRequest
 from .responses import RolloutsPutResponse, RolloutsResponse
@@ -14,15 +13,15 @@
 
 @router.get(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.ROLLOUT.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["rollout.read"])],
 )
 async def rollouts_get(_: Request) -> RolloutsResponse:
     return await RolloutsResponse.convert(await Rollout.all().prefetch_related("software"))
 
 
 @router.post(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.ROLLOUT.WRITE])],
+    dependencies=[Security(validate_user_permissions, scopes=["rollout.write"])],
 )
 async def rollouts_put(_: Request, rollout: RolloutsPutRequest) -> RolloutsPutResponse:
     rollout = await Rollout.create(
@@ -35,7 +34,7 @@ async def rollouts_put(_: Request, rollout: RolloutsPutRequest) -> RolloutsPutRe
 
 @router.patch(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.ROLLOUT.WRITE])],
+    dependencies=[Security(validate_user_permissions, scopes=["rollout.write"])],
 )
 async def rollouts_patch(_: Request, rollouts: RolloutsPatchRequest) -> StatusResponse:
     await Rollout.filter(id__in=rollouts.ids).update(paused=rollouts.paused)
@@ -44,7 +43,7 @@ async def rollouts_patch(_: Request, rollouts: RolloutsPatchRequest) -> StatusRe
 
 @router.delete(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.ROLLOUT.DELETE])],
+    dependencies=[Security(validate_user_permissions, scopes=["rollout.delete"])],
 )
 async def rollouts_delete(_: Request, rollouts: RolloutsDeleteRequest) -> StatusResponse:
     await Rollout.filter(id__in=rollouts.ids).delete()

goosebit/api/v1/software/routes.py

@@ -4,7 +4,6 @@
 from goosebit.api.responses import StatusResponse
 from goosebit.auth import validate_user_permissions
 from goosebit.models import Rollout, Software
-from goosebit.permissions import Permissions
 
 from .requests import SoftwareDeleteRequest
 from .responses import SoftwareResponse
@@ -14,15 +13,15 @@
 
 @router.get(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.SOFTWARE.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["software.read"])],
 )
 async def software_get(_: Request) -> SoftwareResponse:
     return await SoftwareResponse.convert(await Software.all().prefetch_related("compatibility"))
 
 
 @router.delete(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.SOFTWARE.DELETE])],
+    dependencies=[Security(validate_user_permissions, scopes=["software.delete"])],
 )
 async def software_delete(_: Request, config: SoftwareDeleteRequest) -> StatusResponse:
     success = False

goosebit/auth/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Annotated
+from typing import Annotated, Iterable
 
 from argon2.exceptions import VerifyMismatchError
 from fastapi import Depends, HTTPException
@@ -106,13 +106,34 @@ def validate_user_permissions(
     security: SecurityScopes,
     user: User = Depends(get_current_user),
 ) -> HTTPConnection:
-    if security.scopes is None:
-        return connection
-    for scope in security.scopes:
-        if scope not in user.permissions:
-            logger.warning(f"User {user.username} does not have permission {scope}")
-            raise HTTPException(
-                status_code=403,
-                detail="Not enough permissions",
-                headers={"WWW-Authenticate": "Bearer"},
-            )
+    if not check_permissions(security.scopes, user.permissions):
+        logger.warning(f"{user.username} does not have sufficient permissions")
+        raise HTTPException(
+            status_code=403,
+            detail="Not enough permissions",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    return connection
+
+
+def check_permissions(scopes: Iterable[str] | None, permissions: Iterable[str]) -> bool:
+    deny_permissions = [p.lstrip("!") for p in permissions if p.startswith("!")]
+    allow_permissions = [p for p in permissions if not p.startswith("!")]
+    if scopes is None:
+        return True
+    for scope in scopes:
+        if any([_check_permission(scope, permission) for permission in deny_permissions]):
+            return False
+        if not any([_check_permission(scope, permission) for permission in allow_permissions]):
+            return False
+    return True
+
+
+def _check_permission(scope: str, permission: str) -> bool:
+    split_scope = scope.split(".")
+    for idx, permission in enumerate(permission.split(".")):
+        if permission == "*":
+            continue
+        if not split_scope[idx] == permission:
+            return False
+    return True

goosebit/realtime/logs.py

@@ -6,7 +6,6 @@
 from websockets.exceptions import ConnectionClosed
 
 from goosebit.auth import validate_user_permissions
-from goosebit.permissions import Permissions
 from goosebit.updater.manager import get_update_manager
 
 router = APIRouter(prefix="/logs")
@@ -20,7 +19,7 @@ class RealtimeLogModel(BaseModel):
 
 @router.websocket(
     "/{dev_id}",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.HOME.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["home.read"])],
 )
 async def device_logs(websocket: WebSocket, dev_id: str):
     await websocket.accept()

goosebit/settings/schema.py

@@ -1,6 +1,6 @@
 import secrets
 from pathlib import Path
-from typing import Annotated, Iterable
+from typing import Annotated
 
 from joserfc.rfc7518.oct_key import OctKey
 from pydantic import BaseModel, BeforeValidator, Field
@@ -11,22 +11,13 @@
     YamlConfigSettingsSource,
 )
 
-from goosebit.permissions import Permissions, PermissionsBase
-
 from .const import BASE_DIR, LOGGING_DEFAULT, PWD_CXT
 
 
-def parse_permissions(items: Iterable[str]):
-    permissions = set()
-    for p in items:
-        permissions.update(Permissions.from_str(p))
-    return permissions
-
-
 class User(BaseModel):
     username: str
     hashed_pwd: Annotated[str, BeforeValidator(PWD_CXT.hash)] = Field(validation_alias="password")
-    permissions: Annotated[set[PermissionsBase], BeforeValidator(parse_permissions)]
+    permissions: set[str]
 
     def get_json_permissions(self):
         return [str(p) for p in self.permissions]

goosebit/ui/bff/devices/routes.py

@@ -6,7 +6,6 @@
 
 from goosebit.auth import validate_user_permissions
 from goosebit.models import Device, UpdateModeEnum, UpdateStateEnum
-from goosebit.permissions import Permissions
 
 from .responses import BFFDeviceResponse
 
@@ -15,7 +14,7 @@
 
 @router.get(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.HOME.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["home.read"])],
 )
 async def devices_get(request: Request) -> BFFDeviceResponse:
     def search_filter(search_value):

goosebit/ui/bff/rollouts/routes.py

@@ -4,7 +4,6 @@
 
 from goosebit.auth import validate_user_permissions
 from goosebit.models import Rollout
-from goosebit.permissions import Permissions
 
 from .responses import BFFRolloutsResponse
 
@@ -13,7 +12,7 @@
 
 @router.get(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.ROLLOUT.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["rollout.read"])],
 )
 async def rollouts_get(request: Request) -> BFFRolloutsResponse:
     def search_filter(search_value):

goosebit/ui/bff/software/routes.py

@@ -6,15 +6,14 @@
 
 from goosebit.auth import validate_user_permissions
 from goosebit.models import Software
-from goosebit.permissions import Permissions
 from goosebit.ui.bff.software.responses import BFFSoftwareResponse
 
 router = APIRouter(prefix="/software")
 
 
 @router.get(
     "",
-    dependencies=[Security(validate_user_permissions, scopes=[Permissions.SOFTWARE.READ])],
+    dependencies=[Security(validate_user_permissions, scopes=["software.read"])],
 )
 async def software_get(request: Request) -> BFFSoftwareResponse:
     def search_filter(search_value):