api/kiwi_vpn_api/db/user.py

@@ -143,8 +143,7 @@ class User(UserBase, table=True):
             # password hash mismatch
             return None
 
-        if not (user.has_tag(TagValue.login)
+        if user.has_tag(TagValue.login):
-                or user.has_tag(TagValue.admin)):
             # no login permission
             return None
 
@@ -184,7 +183,7 @@ class User(UserBase, table=True):
         Check if this user has a tag.
         """
 
-        return tag in self._get_tags()
+        return tag in self._get_tags
 
     def add_tags(
         self,

api/kiwi_vpn_api/routers/_common.py

@@ -5,6 +5,7 @@ Common dependencies for routers.
 
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
+from kiwi_vpn_api.db.tag import TagValue
 
 from ..config import Config, Settings
 from ..db import Device, User
@@ -36,8 +37,12 @@ class Responses:
         "description": "Must be logged in",
         "content": None,
     }
-    NEEDS_PERMISSION = {
+    NEEDS_ADMIN = {
-        "description": "You're not allowed that operation",
+        "description": "Must be admin",
+        "content": None,
+    }
+    PERMISSION_ERROR = {
+        "description": "You're not allowed that action",
         "content": None,
     }
     ENTRY_EXISTS = {
@@ -48,6 +53,10 @@ class Responses:
         "description": "Entry does not exist in database",
         "content": None,
     }
+    CANT_TARGET_SELF = {
+        "description": "Operation can't target yourself",
+        "content": None,
+    }
 
 
 async def get_current_user(
@@ -72,6 +81,19 @@ async def get_current_user(
     return user
 
 
+async def get_current_user_if_admin(
+    current_user: User = Depends(get_current_user),
+) -> User:
+    """
+    Fail if the currently logged-in user is not an admin.
+    """
+
+    if current_user.has_tag(TagValue.admin):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+
+    return current_user
+
+
 async def get_user_by_name(
     user_name: str,
     current_config: Config | None = Depends(Config.load),

api/kiwi_vpn_api/routers/admin.py

@@ -8,7 +8,7 @@ from sqlmodel import select
 
 from ..config import Config
 from ..db import Connection, TagValue, User, UserCreate
-from ._common import Responses, get_current_user
+from ._common import Responses, get_current_user_if_admin
 
 router = APIRouter(prefix="/admin", tags=["admin"])
 
@@ -74,21 +74,17 @@ async def create_initial_admin(
         status.HTTP_200_OK: Responses.OK,
         status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
         status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
-        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
     },
 )
 async def set_config(
     config: Config,
-    current_user: User = Depends(get_current_user),
+    _: User = Depends(get_current_user_if_admin),
 ):
     """
     PUT ./config: Edit `kiwi-vpn` main config.
     """
 
-    # check permissions
-    if not current_user.has_tag(TagValue.admin):
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
-
     # update config file, reconnect to database
     config.save()
     Connection.connect(config.db.uri)

api/kiwi_vpn_api/routers/device.py

@@ -17,7 +17,7 @@ router = APIRouter(prefix="/device", tags=["device"])
         status.HTTP_200_OK: Responses.OK,
         status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
         status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
-        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
+        status.HTTP_403_FORBIDDEN: Responses.PERMISSION_ERROR,
         status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
         status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
     },
@@ -56,7 +56,7 @@ async def add_device(
         status.HTTP_200_OK: Responses.OK,
         status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
         status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
-        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
         status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
     },
     response_model=User,

api/kiwi_vpn_api/routers/user.py

@@ -8,7 +8,8 @@ from pydantic import BaseModel
 
 from ..config import Config
 from ..db import TagValue, User, UserCreate, UserRead
-from ._common import Responses, get_current_user, get_user_by_name
+from ._common import (Responses, get_current_user, get_current_user_if_admin,
+                      get_user_by_name)
 
 router = APIRouter(prefix="/user", tags=["user"])
 
@@ -53,7 +54,7 @@ async def login(
 
 
 @router.get("/current", response_model=UserRead)
-async def get_current_user_route(
+async def get_current_user(
     current_user: User = Depends(get_current_user),
 ):
     """
@@ -69,24 +70,20 @@ async def get_current_user_route(
         status.HTTP_200_OK: Responses.OK,
         status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
         status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
-        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
         status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
     },
     response_model=UserRead,
 )
 async def add_user(
     user: UserCreate,
-    current_user: User = Depends(get_current_user),
+    _: User = Depends(get_current_user_if_admin),
 ) -> User:
     """
     POST ./: Create a new user in the database.
     """
 
-    # check permissions
+    # actually create the new user
-    if not current_user.can_create(User):
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
-
-    # create the new user
     new_user = User.create(user=user)
 
     # fail if creation was unsuccessful
@@ -106,22 +103,23 @@ async def add_user(
         status.HTTP_200_OK: Responses.OK,
         status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
         status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
-        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
         status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
+        status.HTTP_406_NOT_ACCEPTABLE: Responses.CANT_TARGET_SELF,
     },
     response_model=User,
 )
 async def remove_user(
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(get_current_user_if_admin),
     user: User = Depends(get_user_by_name),
 ):
     """
     DELETE ./{user_name}: Remove a user from the database.
     """
 
-    # check permissions
+    # stop inting
-    if not current_user.can_admin(user):
+    if current_user.name == user.name:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+        raise HTTPException(status_code=status.HTTP_406_NOT_ACCEPTABLE)
 
     # delete user
     user.delete()
@@ -133,24 +131,20 @@ async def remove_user(
         status.HTTP_200_OK: Responses.OK,
         status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
         status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
-        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
     },
 )
 async def extend_tags(
     tags: list[TagValue],
-    current_user: User = Depends(get_current_user),
+    _: User = Depends(get_current_user_if_admin),
     user: User = Depends(get_user_by_name),
 ):
     """
     POST ./{user_name}/tags: Add tags to a user.
     """
 
-    # check permissions
-    if not current_user.can_admin(user):
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
-
-    # change user
     user.add_tags(tags)
+
     user.update()
 
 
@@ -160,22 +154,18 @@ async def extend_tags(
         status.HTTP_200_OK: Responses.OK,
         status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
         status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
-        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
     },
 )
 async def remove_tags(
     tags: list[TagValue],
-    current_user: User = Depends(get_current_user),
+    _: User = Depends(get_current_user_if_admin),
     user: User = Depends(get_user_by_name),
 ):
     """
     DELETE ./{user_name}/tags: Remove tags from a user.
     """
 
-    # check permissions
-    if not current_user.can_admin(user):
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
-
-    # change user
     user.remove_tags(tags)
+
     user.update()