api/kiwi_vpn_api/db/user.py

@@ -4,7 +4,7 @@ Python representation of `user` table.
 
 from __future__ import annotations
 
-from typing import Any, Sequence
+from typing import Any
 
 from pydantic import root_validator
 from sqlalchemy.exc import IntegrityError
@@ -13,7 +13,7 @@ from sqlmodel import Field, Relationship, SQLModel
 from ..config import Config
 from .connection import Connection
 from .device import Device
-from .user_capability import UserCapability, UserCapabilityType
+from .user_capability import UserCapabilityType, UserCapability
 
 
 class UserBase(SQLModel):
@@ -170,23 +170,14 @@ class User(UserBase, table=True):
             for capability in self.capabilities
         )
 
-    def can(
+    def can(self, capability: UserCapabilityType) -> bool:
-        self,
-        capability: UserCapabilityType,
-    ) -> bool:
         """
         Check if this user has a capability.
         """
 
-        return (
+        return capability in self.get_capabilities()
-            capability in self.get_capabilities()
-            or UserCapabilityType.admin in self.get_capabilities()
-        )
 
-    def set_capabilities(
+    def set_capabilities(self, capabilities: set[UserCapabilityType]) -> None:
-        self,
-        capabilities: Sequence[UserCapabilityType],
-    ) -> None:
         """
         Change the capabilities of this user.
         """

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 User, UserCapabilityType
+from ..db import UserCapabilityType, User
 
 oauth2_scheme = OAuth2PasswordBearer(
     tokenUrl=f"{Settings._.api_v1_prefix}/user/authenticate"

api/kiwi_vpn_api/routers/admin.py

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

api/kiwi_vpn_api/routers/user.py

@@ -7,7 +7,7 @@ from fastapi.security import OAuth2PasswordRequestForm
 from pydantic import BaseModel
 
 from ..config import Config
-from ..db import User, UserCapabilityType, UserCreate, UserRead
+from ..db import UserCapabilityType, User, UserCreate, UserRead
 from ._common import Responses, get_current_user, get_current_user_if_admin
 
 router = APIRouter(prefix="/user", tags=["user"])
@@ -47,10 +47,6 @@ async def login(
             headers={"WWW-Authenticate": "Bearer"},
         )
 
-    if not user.can(UserCapabilityType.login):
-        # user cannot login
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
-
     # authentication succeeded
     access_token = await current_config.jwt.create_token(user.name)
     return {"access_token": access_token, "token_type": "bearer"}
@@ -88,7 +84,6 @@ async def add_user(
 
     # actually create the new user
     new_user = User.create(**user.dict())
-    new_user.set_capabilities((UserCapabilityType.login))
 
     # fail if creation was unsuccessful
     if new_user is None: