check: user can login, "admin" can do everything

api/kiwi_vpn_api/db/user.py

@@ -4,7 +4,7 @@ Python representation of `user` table.
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Sequence
 
 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 UserCapabilityType, UserCapability
+from .user_capability import UserCapability, UserCapabilityType
 
 
 class UserBase(SQLModel):
@@ -170,14 +170,23 @@ class User(UserBase, table=True):
             for capability in self.capabilities
         )
 
-    def can(self, capability: UserCapabilityType) -> bool:
+    def can(
+        self,
+        capability: UserCapabilityType,
+    ) -> bool:
         """
         Check if this user has a capability.
         """
 
-        return capability in self.get_capabilities()
+        return (
+            capability in self.get_capabilities()
+            or UserCapabilityType.admin in self.get_capabilities()
+        )
 
-    def set_capabilities(self, capabilities: set[UserCapabilityType]) -> None:
+    def set_capabilities(
+        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 UserCapabilityType, User
+from ..db import User, UserCapabilityType
 
 oauth2_scheme = OAuth2PasswordBearer(
     tokenUrl=f"{Settings._.api_v1_prefix}/user/authenticate"

api/kiwi_vpn_api/routers/admin.py

@@ -63,10 +63,7 @@ async def create_initial_admin(
 
     # create an administrative user
     new_user = User.create(**admin_user.dict())
-    new_user.set_capabilities([
+    new_user.set_capabilities((UserCapabilityType.admin))
-        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 UserCapabilityType, User, UserCreate, UserRead
+from ..db import User, UserCapabilityType, UserCreate, UserRead
 from ._common import Responses, get_current_user, get_current_user_if_admin
 
 router = APIRouter(prefix="/user", tags=["user"])
@@ -47,6 +47,10 @@ 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"}
@@ -84,6 +88,7 @@ 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: