diff --git a/api/kiwi_vpn_api/db/user.py b/api/kiwi_vpn_api/db/user.py index 8e6e5f7..eac3c03 100644 --- a/api/kiwi_vpn_api/db/user.py +++ b/api/kiwi_vpn_api/db/user.py @@ -174,20 +174,6 @@ class User(UserBase, table=True): for capability in self.capabilities ) - def can( - self, - capability: UserCapabilityType, - ) -> bool: - """ - Check if this user has a capability. - """ - - return ( - capability in self.get_capabilities() - # admin can do everything - or UserCapabilityType.admin in self.get_capabilities() - ) - def set_capabilities( self, capabilities: Sequence[UserCapabilityType], @@ -203,6 +189,20 @@ class User(UserBase, table=True): ) for capability in capabilities ] + def _can( + self, + capability: UserCapabilityType, + ) -> bool: + """ + Check if this user has a capability. + """ + + return ( + capability in self.get_capabilities() + # admin can do everything + or UserCapabilityType.admin in self.get_capabilities() + ) + def can_edit( self, user: User, @@ -214,7 +214,61 @@ class User(UserBase, table=True): return ( user.name == self.name # admin can edit everything - or self.can(UserCapabilityType.admin) + or self._can(UserCapabilityType.admin) + ) + + def is_admin( + self, + ) -> bool: + """ + Check if this user is an admin. + """ + + # is admin with "admin" capability + return self._can(UserCapabilityType.admin) + + def can_login( + self, + ) -> bool: + """ + Check if this user can log in. + """ + + return ( + # can login with "login" capability + self._can(UserCapabilityType.login) + # admins can always login + or self.is_admin() + ) + + def can_be_edited_by( + self, + user: User, + ) -> bool: + """ + Check if this user can be edited by another user. + """ + + return ( + # user can edit itself + self.name == user.name + # admin can edit every user + or user._can(UserCapabilityType.admin) + ) + + def can_be_deleted_by( + self, + user: User, + ) -> bool: + """ + Check if this user can be deleted by another user. + """ + + return ( + # only admin can delete users + user._can(UserCapabilityType.admin) + # even admin cannot delete itself + and self.name != user.name ) def owns( @@ -228,5 +282,5 @@ class User(UserBase, table=True): return ( device.owner_name == self.name # admin owns everything - or self.can(UserCapabilityType.admin) + or self._can(UserCapabilityType.admin) ) diff --git a/api/kiwi_vpn_api/main.py b/api/kiwi_vpn_api/main.py index 86ae4f2..eb69de2 100755 --- a/api/kiwi_vpn_api/main.py +++ b/api/kiwi_vpn_api/main.py @@ -13,7 +13,7 @@ import uvicorn from fastapi import FastAPI from .config import Config, Settings -from .db import Connection, User +from .db import Connection, User, UserRead from .routers import main_router app = FastAPI( @@ -43,7 +43,7 @@ async def on_startup() -> None: Connection.connect(current_config.db.uri) # some testing - print(User.get("admin")) + print(UserRead.from_orm(User.get("admin"))) print(User.get("nonexistent")) diff --git a/api/kiwi_vpn_api/routers/_common.py b/api/kiwi_vpn_api/routers/_common.py index 092bb7e..b017952 100644 --- a/api/kiwi_vpn_api/routers/_common.py +++ b/api/kiwi_vpn_api/routers/_common.py @@ -7,7 +7,7 @@ from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from ..config import Config, Settings -from ..db import Device, User, UserCapabilityType +from ..db import Device, User oauth2_scheme = OAuth2PasswordBearer( tokenUrl=f"{Settings._.api_v1_prefix}/user/authenticate" @@ -97,7 +97,7 @@ async def get_current_user_if_admin( Fail if the currently logged-in user is not an admin. """ - if not current_user.can(UserCapabilityType.admin): + if not current_user.is_admin(): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) return current_user diff --git a/api/kiwi_vpn_api/routers/device.py b/api/kiwi_vpn_api/routers/device.py index 7b1e58c..21f774f 100644 --- a/api/kiwi_vpn_api/routers/device.py +++ b/api/kiwi_vpn_api/routers/device.py @@ -5,7 +5,8 @@ from fastapi import APIRouter, Depends, HTTPException, status from ..db import Device, DeviceCreate, DeviceRead, User -from ._common import Responses, get_device_by_id_if_editable, get_user_by_name +from ._common import (Responses, get_device_by_id_if_editable, + get_user_by_name_if_editable) router = APIRouter(prefix="/device", tags=["device"]) @@ -24,7 +25,7 @@ router = APIRouter(prefix="/device", tags=["device"]) ) async def add_device( device: DeviceCreate, - user: User = Depends(get_user_by_name), + user: User = Depends(get_user_by_name_if_editable), ) -> Device: """ POST ./: Create a new device in the database. diff --git a/api/kiwi_vpn_api/routers/user.py b/api/kiwi_vpn_api/routers/user.py index 8bd9b74..5772d5b 100644 --- a/api/kiwi_vpn_api/routers/user.py +++ b/api/kiwi_vpn_api/routers/user.py @@ -48,7 +48,7 @@ async def login( headers={"WWW-Authenticate": "Bearer"}, ) - if not user.can(UserCapabilityType.login): + if not user.can_login(): # user cannot login raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)