feat: add RBAC module (role-based access control)

Author: syed-amjad65

app/core/security.py

@@ -1,4 +1,4 @@
-from fastapi import Depends, HTTPException, status
+from fastapi import Depends, HTTPException
 from fastapi.security import OAuth2PasswordBearer
 from jose import jwt, JWTError
 
@@ -11,13 +11,28 @@ def get_current_user(token: str = Depends(oauth2_scheme)):
     try:
         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
         username: str = payload.get("sub")
-        if username is None:
+        role: str = payload.get("role")
+
+        if username is None or role is None:
             raise HTTPException(status_code=401, detail="Invalid token")
+
     except JWTError:
         raise HTTPException(status_code=401, detail="Invalid token")
 
     user = fake_users_db.get(username)
     if user is None:
         raise HTTPException(status_code=401, detail="User not found")
 
-    return user
+    return {"username": username, "role": role}
+
+
+def require_role(required_role: str):
+    def role_checker(current_user=Depends(get_current_user)):
+        if current_user["role"] != required_role:
+            raise HTTPException(
+                status_code=403,
+                detail=f"Access denied: requires '{required_role}' role"
+            )
+        return current_user
+
+    return role_checker

app/routers/asset_metadata.py

@@ -1,31 +1,30 @@
-from fastapi import APIRouter, HTTPException
-
+from fastapi import APIRouter, Depends, HTTPException
 from app.schemas.asset_metadata import AssetMetadata, CampaignMetadata
-from app.services.asset_metadata import (
-    get_all_assets,
-    get_asset_by_id,
-    get_all_campaigns,
-)
+from app.services.asset_metadata import get_all_assets, get_asset_by_id, get_all_campaigns
+from app.core.security import require_role
 
 router = APIRouter(
     prefix="/metadata",
     tags=["Asset Metadata"],
 )
 
 
+# ✅ Only ADMIN can view all assets
 @router.get("/assets", response_model=list[AssetMetadata])
-def list_assets():
+def list_assets(current_user=Depends(require_role("admin"))):
     return get_all_assets()
 
 
+# ✅ Only ADMIN can view a single asset
 @router.get("/assets/{asset_id}", response_model=AssetMetadata)
-def get_asset(asset_id: str):
+def get_asset(asset_id: str, current_user=Depends(require_role("admin"))):
     asset = get_asset_by_id(asset_id)
     if asset is None:
         raise HTTPException(status_code=404, detail="Asset not found")
     return asset
 
 
+# ✅ Only ANALYST can view campaign metadata
 @router.get("/campaigns", response_model=list[CampaignMetadata])
-def list_campaigns():
+def list_campaigns(current_user=Depends(require_role("analyst"))):
     return get_all_campaigns()

app/routers/auth.py

@@ -1,7 +1,7 @@
-from fastapi import APIRouter, HTTPException, Depends
+from fastapi import APIRouter, Depends, HTTPException
 from fastapi.security import OAuth2PasswordRequestForm
 
-from app.schemas.auth import UserLoginRequest, TokenResponse
+from app.schemas.auth import TokenResponse
 from app.services.auth import authenticate_user, create_access_token
 from app.core.security import get_current_user
 
@@ -17,13 +17,15 @@ def login(form_data: OAuth2PasswordRequestForm = Depends()):
     if not user:
         raise HTTPException(status_code=401, detail="Invalid username or password")
 
-    token = create_access_token({"sub": user["username"]})
+    # ✅ Token now includes role
+    token = create_access_token({
+        "sub": user["username"],
+        "role": user["role"]
+    })
+
     return TokenResponse(access_token=token)
 
 
 @router.get("/me")
-def get_me(current_user: dict = Depends(get_current_user)):
-    return {
-        "username": current_user["username"],
-        "role": current_user["role"],
-    }
+def get_me(current_user=Depends(get_current_user)):
+    return current_user

app/routers/cx_analytics.py

@@ -1,38 +1,15 @@
-from fastapi import APIRouter
-
-from app.schemas.cx_analytics import (
-    AnomalyRequest,
-    AnomalyResponse,
-    EventValidationRequest,
-    EventValidationResponse,
-)
-from app.services.cx_analytics import detect_anomalies, validate_event_sequence
+from fastapi import APIRouter, Depends
+from app.schemas.cx_analytics import AnomalyRequest, AnomalyResponse
+from app.services.cx_analytics import detect_anomaly
+from app.core.security import require_role
 
 router = APIRouter(
     prefix="/cx",
     tags=["CX Analytics"],
 )
 
 
+# ✅ Only ANALYST can run anomaly detection
 @router.post("/anomaly", response_model=AnomalyResponse)
-def anomaly_detection(request: AnomalyRequest):
-    result = detect_anomalies(request.events, request.expected_events)
-
-    return AnomalyResponse(
-        campaign=request.campaign,
-        missing_events=result["missing_events"],
-        unexpected_events=result["unexpected_events"],
-        anomaly_score=result["anomaly_score"],
-        status=result["status"],
-    )
-
-
-@router.post("/validate-events", response_model=EventValidationResponse)
-def validate_events(request: EventValidationRequest):
-    issues = validate_event_sequence(request.events)
-
-    return EventValidationResponse(
-        session_id=request.session_id,
-        valid=len(issues) == 0,
-        issues=issues,
-    )
+def anomaly_detection(request: AnomalyRequest, current_user=Depends(require_role("analyst"))):
+    return detect_anomaly(request)