dirty commit

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)
         )

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"))
 
 

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

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.

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)