Refactor Security Role Checks

- Replaced direct role checks with new role management functions across multiple API endpoints to enhance consistency and maintainability.
- Updated dependencies in billing.py, pricing_tables.py, private_ai_keys.py, products.py, regions.py, teams.py, users.py to utilize get_role_min_* functions.
- Streamlined role validation logic in rbac.py and roles.py for improved clarity and organization.
- Removed deprecated role assignment validation methods from tests, ensuring alignment with the new role management structure.

Author: PhiRho

app/api/billing.py

@@ -4,7 +4,7 @@
 import os
 from datetime import datetime, UTC
 from app.db.database import get_db
-from app.core.security import check_specific_team_admin, check_system_admin
+from app.core.security import get_role_min_specific_team_admin, get_role_min_system_admin
 from app.db.models import DBTeam, DBSystemSecret, DBProduct, DBTeamProduct
 from app.schemas.models import PricingTableSession, SubscriptionCreate, SubscriptionResponse
 from app.services.stripe import (
@@ -91,7 +91,7 @@ async def handle_events(
             detail="Error processing webhook"
         )
 
-@router.post("/teams/{team_id}/portal", dependencies=[Depends(check_specific_team_admin)])
+@router.post("/teams/{team_id}/portal", dependencies=[Depends(get_role_min_specific_team_admin)])
 async def get_portal(
     team_id: int,
     db: Session = Depends(get_db)
@@ -135,7 +135,7 @@ async def get_portal(
             detail="Error creating portal session"
         )
 
-@router.get("/teams/{team_id}/pricing-table-session", dependencies=[Depends(check_specific_team_admin)], response_model=PricingTableSession)
+@router.get("/teams/{team_id}/pricing-table-session", dependencies=[Depends(get_role_min_specific_team_admin)], response_model=PricingTableSession)
 async def get_pricing_table_session(
     team_id: int,
     db: Session = Depends(get_db)
@@ -178,7 +178,7 @@ async def get_pricing_table_session(
             detail="Error creating customer session"
         )
 
-@router.post("/teams/{team_id}/subscriptions", dependencies=[Depends(check_system_admin)], response_model=SubscriptionResponse, status_code=status.HTTP_201_CREATED)
+@router.post("/teams/{team_id}/subscriptions", dependencies=[Depends(get_role_min_system_admin)], response_model=SubscriptionResponse, status_code=status.HTTP_201_CREATED)
 async def create_team_subscription(
     team_id: int,
     subscription_data: SubscriptionCreate,

app/api/pricing_tables.py

@@ -5,7 +5,7 @@
 
 from app.db.database import get_db
 from app.db.models import DBTeam, DBPricingTable
-from app.core.security import check_system_admin, get_role_min_team_admin, get_current_user_from_auth
+from app.core.security import get_role_min_system_admin, get_role_min_team_admin, get_current_user_from_auth
 from app.schemas.models import PricingTableCreate, PricingTableResponse, PricingTablesResponse
 from app.core.config import settings
 
@@ -16,8 +16,8 @@
     tags=["pricing-tables"]
 )
 
-@router.post("", response_model=PricingTableResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(check_system_admin)])
-@router.post("/", response_model=PricingTableResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(check_system_admin)])
+@router.post("", response_model=PricingTableResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(get_role_min_system_admin)])
+@router.post("/", response_model=PricingTableResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(get_role_min_system_admin)])
 async def create_pricing_table(
     pricing_table: PricingTableCreate,
     db: Session = Depends(get_db)
@@ -113,8 +113,8 @@ async def get_pricing_table(
         updated_at=pricing_table.updated_at or pricing_table.created_at
     )
 
-@router.delete("", dependencies=[Depends(check_system_admin)])
-@router.delete("/", dependencies=[Depends(check_system_admin)])
+@router.delete("", dependencies=[Depends(get_role_min_system_admin)])
+@router.delete("/", dependencies=[Depends(get_role_min_system_admin)])
 async def delete_pricing_table(
     table_type: str,
     db: Session = Depends(get_db)
@@ -146,7 +146,7 @@ async def delete_pricing_table(
 
     return {"message": f"Pricing table of type '{table_type}' deleted successfully"}
 
-@router.get("/list", response_model=PricingTablesResponse, dependencies=[Depends(check_system_admin)])
+@router.get("/list", response_model=PricingTablesResponse, dependencies=[Depends(get_role_min_system_admin)])
 async def get_all_pricing_tables(
     db: Session = Depends(get_db)
 ):

app/api/private_ai_keys.py

@@ -17,7 +17,7 @@
     get_current_user_from_auth,
     get_role_min_team_admin,
     get_private_ai_access,
-    check_system_admin
+    get_role_min_system_admin
 )
 from app.core.roles import UserRole
 from app.core.config import settings
@@ -477,7 +477,7 @@ async def list_private_ai_keys(
     private_ai_keys = query.all()
     return [key.to_dict() for key in private_ai_keys]
 
-@router.get("/{key_id}", response_model=PrivateAIKeyDetail, dependencies=[Depends(check_system_admin)])
+@router.get("/{key_id}", response_model=PrivateAIKeyDetail, dependencies=[Depends(get_role_min_system_admin)])
 async def get_private_ai_key(
     key_id: int,
     current_user = Depends(get_current_user_from_auth),

app/api/products.py

@@ -5,15 +5,15 @@
 
 from app.db.database import get_db
 from app.db.models import DBProduct, DBTeamProduct, DBTeam
-from app.core.security import check_system_admin, get_current_user_from_auth, get_role_min_team_admin
+from app.core.security import get_role_min_system_admin, get_current_user_from_auth, get_role_min_team_admin
 from app.schemas.models import Product, ProductCreate, ProductUpdate
 
 router = APIRouter(
     tags=["products"]
 )
 
-@router.post("", response_model=Product, status_code=status.HTTP_201_CREATED, dependencies=[Depends(check_system_admin)])
-@router.post("/", response_model=Product, status_code=status.HTTP_201_CREATED, dependencies=[Depends(check_system_admin)])
+@router.post("", response_model=Product, status_code=status.HTTP_201_CREATED, dependencies=[Depends(get_role_min_system_admin)])
+@router.post("/", response_model=Product, status_code=status.HTTP_201_CREATED, dependencies=[Depends(get_role_min_system_admin)])
 async def create_product(
     product: ProductCreate,
     db: Session = Depends(get_db)
@@ -105,7 +105,7 @@ async def get_product(
         )
     return product
 
-@router.put("/{product_id}", response_model=Product, dependencies=[Depends(check_system_admin)])
+@router.put("/{product_id}", response_model=Product, dependencies=[Depends(get_role_min_system_admin)])
 async def update_product(
     product_id: str,
     product_update: ProductUpdate,
@@ -132,7 +132,7 @@ async def update_product(
 
     return product
 
-@router.delete("/{product_id}", dependencies=[Depends(check_system_admin)])
+@router.delete("/{product_id}", dependencies=[Depends(get_role_min_system_admin)])
 async def delete_product(
     product_id: str,
     db: Session = Depends(get_db)

app/api/regions.py

@@ -9,7 +9,7 @@
 from app.api.auth import get_current_user_from_auth
 from app.schemas.models import Region, RegionCreate, RegionResponse, User, RegionUpdate, TeamSummary
 from app.db.models import DBRegion, DBPrivateAIKey, DBTeamRegion, DBTeam
-from app.core.security import check_system_admin
+from app.core.security import get_role_min_system_admin
 
 logger = logging.getLogger(__name__)
 
@@ -89,8 +89,8 @@ async def validate_database_connection(host: str, port: int, user: str, password
             detail=f"Database connection validation failed: {str(e)}"
         )
 
-@router.post("", response_model=Region, dependencies=[Depends(check_system_admin)])
-@router.post("/", response_model=Region, dependencies=[Depends(check_system_admin)])
+@router.post("", response_model=Region, dependencies=[Depends(get_role_min_system_admin)])
+@router.post("/", response_model=Region, dependencies=[Depends(get_role_min_system_admin)])
 async def create_region(
     region: RegionCreate,
     db: Session = Depends(get_db)
@@ -158,13 +158,13 @@ async def list_regions(
 
     return non_dedicated_regions + team_dedicated_regions
 
-@router.get("/admin", response_model=List[Region], dependencies=[Depends(check_system_admin)])
+@router.get("/admin", response_model=List[Region], dependencies=[Depends(get_role_min_system_admin)])
 async def list_admin_regions(
     db: Session = Depends(get_db)
 ):
     return db.query(DBRegion).all()
 
-@router.get("/{region_id}", response_model=RegionResponse, dependencies=[Depends(check_system_admin)])
+@router.get("/{region_id}", response_model=RegionResponse, dependencies=[Depends(get_role_min_system_admin)])
 async def get_region(
     region_id: int,
     db: Session = Depends(get_db)
@@ -177,7 +177,7 @@ async def get_region(
         )
     return region
 
-@router.delete("/{region_id}", dependencies=[Depends(check_system_admin)])
+@router.delete("/{region_id}", dependencies=[Depends(get_role_min_system_admin)])
 async def delete_region(
     region_id: int,
     db: Session = Depends(get_db)
@@ -209,7 +209,7 @@ async def delete_region(
         )
     return {"message": "Region deleted successfully"}
 
-@router.put("/{region_id}", response_model=Region, dependencies=[Depends(check_system_admin)])
+@router.put("/{region_id}", response_model=Region, dependencies=[Depends(get_role_min_system_admin)])
 async def update_region(
     region_id: int,
     region: RegionUpdate,
@@ -251,7 +251,7 @@ async def update_region(
         )
     return db_region
 
-@router.post("/{region_id}/teams/{team_id}", dependencies=[Depends(check_system_admin)])
+@router.post("/{region_id}/teams/{team_id}", dependencies=[Depends(get_role_min_system_admin)])
 async def associate_team_with_region(
     region_id: int,
     team_id: int,
@@ -311,7 +311,7 @@ async def associate_team_with_region(
 
     return {"message": "Team associated with region successfully"}
 
-@router.delete("/{region_id}/teams/{team_id}", dependencies=[Depends(check_system_admin)])
+@router.delete("/{region_id}/teams/{team_id}", dependencies=[Depends(get_role_min_system_admin)])
 async def disassociate_team_from_region(
     region_id: int,
     team_id: int,
@@ -345,7 +345,7 @@ async def disassociate_team_from_region(
 
     return {"message": "Team disassociated from region successfully"}
 
-@router.get("/{region_id}/teams", response_model=List[TeamSummary], dependencies=[Depends(check_system_admin)])
+@router.get("/{region_id}/teams", response_model=List[TeamSummary], dependencies=[Depends(get_role_min_system_admin)])
 async def list_teams_for_region(
     region_id: int,
     db: Session = Depends(get_db)

app/api/teams.py

@@ -7,7 +7,7 @@
 
 from app.db.database import get_db
 from app.db.models import DBTeam, DBTeamProduct, DBUser, DBPrivateAIKey, DBRegion, DBTeamRegion, DBProduct
-from app.core.security import check_system_admin, check_specific_team_admin, get_current_user_from_auth, check_sales_or_higher
+from app.core.security import get_role_min_system_admin, get_role_min_specific_team_admin, get_current_user_from_auth, check_sales_or_higher
 from app.schemas.models import (
     Team, TeamCreate, TeamUpdate,
     TeamWithUsers, TeamMergeRequest, TeamMergeResponse
@@ -67,8 +67,8 @@ async def register_team(
 
     return db_team
 
-@router.get("", response_model=List[Team], dependencies=[Depends(check_system_admin)])
-@router.get("/", response_model=List[Team], dependencies=[Depends(check_system_admin)])
+@router.get("", response_model=List[Team], dependencies=[Depends(get_role_min_system_admin)])
+@router.get("/", response_model=List[Team], dependencies=[Depends(get_role_min_system_admin)])
 async def list_teams(
     db: Session = Depends(get_db)
 ):
@@ -77,7 +77,7 @@ async def list_teams(
     """
     return db.query(DBTeam).all()
 
-@router.get("/{team_id}", response_model=TeamWithUsers, dependencies=[Depends(check_specific_team_admin)])
+@router.get("/{team_id}", response_model=TeamWithUsers, dependencies=[Depends(get_role_min_specific_team_admin)])
 async def get_team(
     team_id: int,
     db: Session = Depends(get_db)
@@ -93,7 +93,7 @@ async def get_team(
     # Convert directly to TeamWithUsers model
     return TeamWithUsers.model_validate(db_team)
 
-@router.put("/{team_id}", response_model=Team, dependencies=[Depends(check_specific_team_admin)])
+@router.put("/{team_id}", response_model=Team, dependencies=[Depends(get_role_min_specific_team_admin)])
 async def update_team(
     team_id: int,
     team_update: TeamUpdate,
@@ -144,7 +144,7 @@ async def update_team(
 
     return db_team
 
-@router.delete("/{team_id}", dependencies=[Depends(check_system_admin)])
+@router.delete("/{team_id}", dependencies=[Depends(get_role_min_system_admin)])
 async def delete_team(
     team_id: int,
     db: Session = Depends(get_db)
@@ -167,7 +167,7 @@ async def delete_team(
 
     return {"message": "Team deleted successfully"}
 
-@router.post("/{team_id}/extend-trial", dependencies=[Depends(check_system_admin)])
+@router.post("/{team_id}/extend-trial", dependencies=[Depends(get_role_min_system_admin)])
 async def extend_team_trial(
     team_id: int,
     db: Session = Depends(get_db)
@@ -408,7 +408,7 @@ async def _resolve_key_conflicts(
     else:
         raise ValueError(f"Unknown conflict resolution strategy: {strategy}")
 
-@router.post("/{target_team_id}/merge", dependencies=[Depends(check_system_admin)])
+@router.post("/{target_team_id}/merge", dependencies=[Depends(get_role_min_system_admin)])
 async def merge_teams(
     target_team_id: int,
     merge_request: TeamMergeRequest,

app/api/users.py

@@ -6,15 +6,15 @@
 from app.db.database import get_db
 from app.schemas.models import User, UserUpdate, UserCreate, TeamOperation, UserRoleUpdate
 from app.db.models import DBUser, DBTeam
-from app.core.security import get_password_hash, check_system_admin, get_current_user_from_auth, get_role_min_team_admin
+from app.core.security import get_password_hash, get_role_min_system_admin, get_current_user_from_auth, get_role_min_team_admin
 from app.core.roles import UserRole
 from datetime import datetime, UTC
 
 router = APIRouter(
     tags=["users"]
 )
 
-@router.get("/search", response_model=List[User], dependencies=[Depends(check_system_admin)])
+@router.get("/search", response_model=List[User], dependencies=[Depends(get_role_min_system_admin)])
 async def search_users(
     email: str,
     db: Session = Depends(get_db)
@@ -199,7 +199,7 @@ async def add_user_to_team(
     db.refresh(db_user)
     return db_user
 
-@router.post("/{user_id}/remove-from-team", response_model=User, dependencies=[Depends(check_system_admin)])
+@router.post("/{user_id}/remove-from-team", response_model=User, dependencies=[Depends(get_role_min_system_admin)])
 async def remove_user_from_team(
     user_id: int,
     current_user: DBUser = Depends(get_current_user_from_auth),
@@ -225,7 +225,7 @@ async def remove_user_from_team(
     db.refresh(db_user)
     return db_user
 
-@router.delete("/{user_id}", dependencies=[Depends(check_system_admin)])
+@router.delete("/{user_id}", dependencies=[Depends(get_role_min_system_admin)])
 async def delete_user(
     user_id: int,
     current_user: DBUser = Depends(get_current_user_from_auth),

app/core/rbac.py

@@ -54,11 +54,11 @@ def _validate_user_type_constraints(self, user: DBUser) -> bool:
         effective_role = self._get_effective_role(user)
 
         # System users (team_id is None) cannot have team roles
-        if user.team_id is None and effective_role in ["admin", "key_creator", "read_only"]:
+        if user.team_id is None and effective_role in UserRole.get_team_roles():
             return True
 
         # Team users (team_id is not None) cannot have system roles
-        if user.team_id is not None and effective_role in ["system_admin", "user", "sales"]:
+        if user.team_id is not None and effective_role in UserRole.get_system_roles():
             return True
 
         return False
@@ -76,23 +76,23 @@ def require_system_admin():
 
 def require_team_admin():
     """Require team admin role or system admin"""
-    return RBACDependency([UserRole.TEAM_ADMIN, UserRole.SYSTEM_ADMIN], require_team_membership=True)
+    return RBACDependency(UserRole.ADMIN_ROLES, require_team_membership=True)
 
 def require_key_creator_or_higher():
     """Require key creator role or higher (team context)"""
-    return RBACDependency([UserRole.TEAM_ADMIN, UserRole.KEY_CREATOR, UserRole.SYSTEM_ADMIN], require_team_membership=True)
+    return RBACDependency(UserRole.KEY_MANAGEMENT_ROLES, require_team_membership=True)
 
 def require_private_ai_access():
     """Require access to private AI operations - allows system users or team key creators"""
-    return RBACDependency([UserRole.TEAM_ADMIN, UserRole.KEY_CREATOR, UserRole.SYSTEM_ADMIN, UserRole.USER], require_team_membership=False)
+    return RBACDependency(UserRole.KEY_MANAGEMENT_ROLES + [UserRole.USER], require_team_membership=False)
 
 def require_read_only_or_higher():
     """Require read only role or higher (team context)"""
-    return RBACDependency([UserRole.TEAM_ADMIN, UserRole.KEY_CREATOR, UserRole.READ_ONLY, UserRole.SYSTEM_ADMIN], require_team_membership=True)
+    return RBACDependency(UserRole.READ_ACCESS_ROLES, require_team_membership=True)
 
 def require_sales_or_higher():
     """Require sales role or higher (system context)"""
-    return RBACDependency([UserRole.SYSTEM_ADMIN, UserRole.SALES])
+    return RBACDependency(UserRole.SYSTEM_ACCESS_ROLES)
 
 def require_any_role():
     """Allow any authenticated user"""