api/kiwi_vpn_api/db/__init__.py

@@ -3,7 +3,7 @@ Package `db`: ORM and schemas for database content.
 """
 
 from .connection import Connection
-from .device import Device, DeviceBase, DeviceCreate, DeviceRead
+from .device import Device, DeviceBase, DeviceCreate
 from .user import User, UserBase, UserCreate, UserRead
 from .user_capability import UserCapabilityType
 
@@ -12,7 +12,6 @@ __all__ = [
     "Device",
     "DeviceBase",
     "DeviceCreate",
-    "DeviceRead",
     "User",
     "UserBase",
     "UserCreate",

api/kiwi_vpn_api/db/device.py

@@ -31,6 +31,8 @@ class DeviceCreate(DeviceBase):
     Representation of a newly created device
     """
 
+    owner_name: str | None
+
 
 class DeviceRead(DeviceBase):
     """
@@ -60,47 +62,25 @@ class Device(DeviceBase, table=True):
     )
 
     @classmethod
-    def create(
+    def create(cls, **kwargs) -> Device | None:
-        cls,
-        *,
-        owner: User,
-        device: DeviceCreate,
-    ) -> Device | None:
         """
         Create a new device in the database.
         """
 
         try:
             with Connection.session as db:
-                new_device = cls.from_orm(device)
+                device = cls.from_orm(DeviceCreate(**kwargs))
-                new_device.owner = owner
 
-                db.add(new_device)
+                db.add(device)
                 db.commit()
-                db.refresh(new_device)
+                db.refresh(device)
 
-                return new_device
+                return device
 
         except IntegrityError:
             # device already existed
             return None
 
-    @classmethod
-    def create_kwargs(
-        cls,
-        *,
-        owner: User,
-        **kwargs
-    ) -> Device | None:
-        """
-        Create a new device in the database. Keywords version.
-        """
-
-        return cls.create(
-            owner=owner,
-            device=DeviceCreate(**kwargs),
-        )
-
     def update(self) -> None:
         """
         Update this device in the database.

api/kiwi_vpn_api/db/user.py

@@ -90,42 +90,25 @@ class User(UserBase, table=True):
     )
 
     @classmethod
-    def create(
+    def create(cls, **kwargs) -> User | None:
-        cls,
-        *,
-        user: UserCreate,
-    ) -> User | None:
         """
         Create a new user in the database.
         """
 
         try:
             with Connection.session as db:
-                new_user = cls.from_orm(user)
+                user = cls.from_orm(UserCreate(**kwargs))
 
-                db.add(new_user)
+                db.add(user)
                 db.commit()
-                db.refresh(new_user)
+                db.refresh(user)
 
-                return new_user
+                return user
 
         except IntegrityError:
             # user already existed
             return None
 
-    @classmethod
-    def create_kwargs(
-        cls,
-        **kwargs
-    ) -> User | None:
-        """
-        Create a new user in the database. Keywords version.
-        """
-
-        return cls.create(
-            user=UserCreate(**kwargs),
-        )
-
     @classmethod
     def get(cls, name: str) -> User | None:
         """

api/kiwi_vpn_api/routers/__init__.py

@@ -6,12 +6,11 @@ This file: Main API router definition.
 
 from fastapi import APIRouter
 
-from . import admin, device, user
+from . import admin, user
 
 main_router = APIRouter()
 
 main_router.include_router(admin.router)
-main_router.include_router(device.router)
 main_router.include_router(user.router)
 
 __all__ = [

api/kiwi_vpn_api/routers/_common.py

@@ -40,7 +40,7 @@ class Responses:
         "description": "Must be admin",
         "content": None,
     }
-    NEEDS_REQUESTED_USER = {
+    NEEDS_ADMIN_OR_SELF = {
         "description": "Must be the requested user",
         "content": None,
     }
@@ -99,29 +99,20 @@ async def get_current_user_if_admin(
     return current_user
 
 
-async def get_user_by_name(
+async def get_current_user_if_admin_or_self(
     user_name: str,
     current_user: User = Depends(get_current_user_if_exists),
 ) -> User:
     """
-    Get a user by name.
+    Get the currently logged-in user.
 
-    Works if a) the currently logged-in user is an admin,
+    Fails a) if the currently logged-in user is not the requested user,
-    or b) if it is the requested user.
+    and b) if it is not an admin.
     """
 
-    # check if current user is admin
+    # fail if not requested by an admin or self
-    if current_user.can(UserCapabilityType.admin):
+    if not (current_user.can(UserCapabilityType.admin)
-        # fail if requested user doesn't exist
+            or current_user.name == user_name):
-        if (user := User.get(user_name)) is None:
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
-
-    # check if current user is requested user
-    elif current_user.name == user_name:
-        pass
-
-    # current user is neither admin nor the requested user
-    else:
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
 
-    return user
+    return current_user

api/kiwi_vpn_api/routers/admin.py

@@ -63,7 +63,7 @@ async def create_initial_admin(
             raise HTTPException(status_code=status.HTTP_409_CONFLICT)
 
     # create an administrative user
-    new_user = User.create(admin_user)
+    new_user = User.create(**admin_user.dict())
     new_user.set_capabilities([UserCapabilityType.admin])
     new_user.update()
 

api/kiwi_vpn_api/routers/device.py

@@ -1,44 +0,0 @@
-"""
-/device endpoints.
-"""
-
-from fastapi import APIRouter, Depends, HTTPException, status
-
-from ..db import Device, DeviceCreate, DeviceRead, User
-from ._common import Responses, get_user_by_name
-
-router = APIRouter(prefix="/device", tags=["device"])
-
-
-@router.post(
-    "",
-    responses={
-        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_REQUESTED_USER,
-        status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
-        status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
-    },
-    response_model=DeviceRead,
-)
-async def add_device(
-    device: DeviceCreate,
-    user: User = Depends(get_user_by_name),
-) -> Device:
-    """
-    POST ./: Create a new device in the database.
-    """
-
-    # create the new device
-    new_device = Device.create(
-        owner=user,
-        device=device,
-    )
-
-    # fail if creation was unsuccessful
-    if new_device is None:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT)
-
-    # return the created device on success
-    return new_device

api/kiwi_vpn_api/routers/user.py

@@ -76,26 +76,24 @@ async def get_current_user(
         status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
         status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
     },
-    response_model=UserRead,
+    response_model=User,
 )
 async def add_user(
     user: UserCreate,
     _: User = Depends(get_current_user_if_admin),
-) -> User:
+):
     """
     POST ./: Create a new user in the database.
     """
 
     # actually create the new user
-    new_user = User.create(user=user)
+    new_user = User.create(**user.dict())
+    new_user.set_capabilities([UserCapabilityType.login])
 
     # fail if creation was unsuccessful
     if new_user is None:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT)
 
-    new_user.set_capabilities([UserCapabilityType.login])
-    new_user.update()
-
     # return the created user on success
     return new_user