Working on RBAC.

backend/apps/users/services.py

@@ -16,19 +16,7 @@ from middleware.db_connection import get_db
 from settings import config
 
 ALGORITHM = "HS256"
-
-fake_users_db = {
-    "johndoe": {
-        "username": "johndoe",
-        "full_name": "John Doe",
-        "email": "johndoe@example.com",
-        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
-        "disabled": False,
-    }
-}
-
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
@@ -89,7 +77,6 @@ async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: Se
         raise credentials_exception
     return user
 
-
 async def get_current_active_user(
         current_user: Annotated[User, Depends(get_current_user)]
 ):
@@ -98,11 +85,10 @@ async def get_current_active_user(
     return current_user
 
 
-def permission_required(permission: Permission):
+def permission_required(*permissions: Permission):
     def decorator(func):
         @wraps(func)
         async def decorated_function(*args, **kwargs):
-            print("test", kwargs)
             request: Request = kwargs.get("request")
             db: Session = kwargs.get("db")
             if not request:
@@ -110,30 +96,20 @@ def permission_required(permission: Permission):
                     status_code=status.HTTP_400_BAD_REQUEST,
                     detail="Missing request object",
                 )
-
-            # Get the authorization header
             authorization_header = request.headers.get("Authorization")
-
-            # Extract the token from the authorization header
             if not authorization_header:
                 raise HTTPException(
                     status_code=status.HTTP_401_UNAUTHORIZED,
                     detail="Missing authorization header",
                 )
-
-            # Assuming the token is in the format "Bearer <token>"
             token = authorization_header.split(" ")[1]
-
             current_user = await get_current_user(token=token, db=db)
-            # Check if the user has the required permission
+            for permission in permissions:
                 if permission not in role_permissions[current_user.role]:
                     raise HTTPException(
                         status_code=status.HTTP_403_FORBIDDEN,
-                    detail="You do not have the required permission",
+                        detail=f"You do not have the required permission: {permission.value}",
                     )
-            # Call the decorated function with the original arguments
             return await func(*args, **kwargs)
-
         return decorated_function
-
     return decorator

backend/routes/http/agent.py

@@ -9,14 +9,11 @@ from middleware.db_connection import get_db
 
 router = APIRouter(prefix="/agents", tags=["agents"])
 
-
 class Agent(BaseModel):
     name: str
 
-
-# Example usage:
 @router.post("/", dependencies=[Depends(get_current_active_user)])
-@permission_required(Permission.CREATE)
+@permission_required(Permission.CREATE, Permission.READ)
 async def create_agent(agent: Agent, request: Request, db: Session = Depends(get_db)):
     print(agent)
     return True

backend/routes/http/user.py

@@ -5,7 +5,7 @@ from sqlalchemy.orm import Session
 from typing import Annotated
 
 from apps.users.roles import Permission
-from apps.users.services import authenticate_user, fake_users_db, create_access_token, get_current_active_user
+from apps.users.services import authenticate_user, create_access_token, get_current_active_user
 from fastapi.security import OAuth2PasswordRequestForm
 
 from apps.users import crud, models, schemas