feat: RBAC Phase 1 — consolidate user roles into 4-value enum
Some checks failed
Some checks failed
Consolidate User.role (2-value: admin/store) + User.is_super_admin (boolean) into a single 4-value UserRole enum: super_admin, platform_admin, merchant_owner, store_member. Drop stale StoreUser.user_type column. Fix role="user" bug in merchant creation. Key changes: - Expand UserRole enum from 2 to 4 values with computed properties (is_admin, is_super_admin, is_platform_admin, is_merchant_owner, is_store_user) - Add Alembic migration (tenancy_003) for data migration + column drops - Remove is_super_admin from JWT token payload - Update all auth dependencies, services, routes, templates, JS, and tests - Update all RBAC documentation 66 files changed, 1219 unit tests passing. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -344,7 +344,7 @@ class AdminPlatformService:
|
||||
reason="User must be an admin to be promoted to super admin",
|
||||
)
|
||||
|
||||
user.is_super_admin = is_super_admin
|
||||
user.role = "super_admin" if is_super_admin else "platform_admin"
|
||||
user.updated_at = datetime.now(UTC)
|
||||
db.flush()
|
||||
db.refresh(user)
|
||||
@@ -402,9 +402,8 @@ class AdminPlatformService:
|
||||
hashed_password=auth_manager.hash_password(password),
|
||||
first_name=first_name,
|
||||
last_name=last_name,
|
||||
role="admin",
|
||||
role="platform_admin",
|
||||
is_active=True,
|
||||
is_super_admin=False, # Platform admin, not super admin
|
||||
)
|
||||
db.add(user)
|
||||
db.flush()
|
||||
@@ -459,10 +458,12 @@ class AdminPlatformService:
|
||||
Returns:
|
||||
Tuple of (list of User objects, total count)
|
||||
"""
|
||||
query = db.query(User).filter(User.role == "admin")
|
||||
query = db.query(User).filter(
|
||||
User.role.in_(["super_admin", "platform_admin"])
|
||||
)
|
||||
|
||||
if not include_super_admins:
|
||||
query = query.filter(User.is_super_admin == False)
|
||||
query = query.filter(User.role == "platform_admin")
|
||||
|
||||
if is_active is not None:
|
||||
query = query.filter(User.is_active == is_active)
|
||||
@@ -508,7 +509,10 @@ class AdminPlatformService:
|
||||
admin = (
|
||||
db.query(User)
|
||||
.options(joinedload(User.admin_platforms))
|
||||
.filter(User.id == user_id, User.role == "admin")
|
||||
.filter(
|
||||
User.id == user_id,
|
||||
User.role.in_(["super_admin", "platform_admin"]),
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
@@ -560,8 +564,7 @@ class AdminPlatformService:
|
||||
hashed_password=get_password_hash(password),
|
||||
first_name=first_name,
|
||||
last_name=last_name,
|
||||
role="admin",
|
||||
is_super_admin=True,
|
||||
role="super_admin",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(user)
|
||||
@@ -601,7 +604,10 @@ class AdminPlatformService:
|
||||
operation="deactivate own account",
|
||||
)
|
||||
|
||||
admin = db.query(User).filter(User.id == user_id, User.role == "admin").first()
|
||||
admin = db.query(User).filter(
|
||||
User.id == user_id,
|
||||
User.role.in_(["super_admin", "platform_admin"]),
|
||||
).first()
|
||||
|
||||
if not admin:
|
||||
raise UserNotFoundException(str(user_id))
|
||||
@@ -642,7 +648,10 @@ class AdminPlatformService:
|
||||
operation="delete own account",
|
||||
)
|
||||
|
||||
admin = db.query(User).filter(User.id == user_id, User.role == "admin").first()
|
||||
admin = db.query(User).filter(
|
||||
User.id == user_id,
|
||||
User.role.in_(["super_admin", "platform_admin"]),
|
||||
).first()
|
||||
|
||||
if not admin:
|
||||
raise UserNotFoundException(str(user_id))
|
||||
|
||||
@@ -69,7 +69,7 @@ class AdminService:
|
||||
raise CannotModifySelfException(user_id, "deactivate account")
|
||||
|
||||
# Check if user is another admin
|
||||
if user.role == "admin" and user.id != current_admin_id:
|
||||
if user.is_admin and user.id != current_admin_id:
|
||||
raise UserStatusChangeException(
|
||||
user_id=user_id,
|
||||
current_status="admin",
|
||||
@@ -248,7 +248,7 @@ class AdminService:
|
||||
user = self._get_user_by_id_or_raise(db, user_id)
|
||||
|
||||
# Prevent changing own admin status
|
||||
if user.id == current_admin_id and role and role != "admin":
|
||||
if user.id == current_admin_id and role and role not in ("super_admin", "platform_admin"):
|
||||
raise UserRoleChangeException(
|
||||
user_id=user_id,
|
||||
current_role=user.role,
|
||||
|
||||
@@ -69,7 +69,7 @@ class MerchantService:
|
||||
username=merchant_data.owner_email.split("@")[0],
|
||||
email=merchant_data.owner_email,
|
||||
hashed_password=auth_manager.hash_password(temp_password),
|
||||
role="user",
|
||||
role="merchant_owner",
|
||||
is_active=True,
|
||||
is_email_verified=False,
|
||||
)
|
||||
|
||||
@@ -80,7 +80,7 @@ class StoreService:
|
||||
|
||||
# Check if user is merchant owner or admin
|
||||
if (
|
||||
current_user.role != "admin"
|
||||
not current_user.is_admin
|
||||
and merchant.owner_user_id != current_user.id
|
||||
):
|
||||
raise UnauthorizedStoreAccessException(
|
||||
@@ -105,7 +105,7 @@ class StoreService:
|
||||
letzshop_csv_url_en=store_data.letzshop_csv_url_en,
|
||||
letzshop_csv_url_de=store_data.letzshop_csv_url_de,
|
||||
is_active=True,
|
||||
is_verified=(current_user.role == "admin"),
|
||||
is_verified=current_user.is_admin,
|
||||
)
|
||||
|
||||
db.add(new_store)
|
||||
@@ -153,7 +153,7 @@ class StoreService:
|
||||
query = db.query(Store)
|
||||
|
||||
# Non-admin users can only see active and verified stores, plus their own
|
||||
if current_user.role != "admin":
|
||||
if not current_user.is_admin:
|
||||
# Get store IDs the user owns through merchants
|
||||
from app.modules.tenancy.models import Merchant
|
||||
|
||||
@@ -457,7 +457,7 @@ class StoreService:
|
||||
def _can_access_store(self, store: Store, user: User) -> bool:
|
||||
"""Check if user can access store."""
|
||||
# Admins can always access
|
||||
if user.role == "admin":
|
||||
if user.is_admin:
|
||||
return True
|
||||
|
||||
# Merchant owners can access their stores
|
||||
@@ -481,7 +481,7 @@ class StoreService:
|
||||
- Team members with appropriate role (owner role in StoreUser)
|
||||
"""
|
||||
# Admins can always update
|
||||
if user.role == "admin":
|
||||
if user.is_admin:
|
||||
return True
|
||||
|
||||
# Check if user is store owner via merchant
|
||||
|
||||
@@ -33,7 +33,7 @@ from app.modules.tenancy.exceptions import (
|
||||
TeamMemberAlreadyExistsException,
|
||||
UserNotFoundException,
|
||||
)
|
||||
from app.modules.tenancy.models import Role, Store, StoreUser, StoreUserType, User
|
||||
from app.modules.tenancy.models import Role, Store, StoreUser, User
|
||||
from middleware.auth import AuthManager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -134,7 +134,7 @@ class StoreTeamService:
|
||||
email=email,
|
||||
username=username,
|
||||
hashed_password=self.auth_manager.hash_password(temp_password),
|
||||
role="store", # Platform role
|
||||
role="store_member",
|
||||
is_active=False, # Will be activated when invitation accepted
|
||||
is_email_verified=False,
|
||||
)
|
||||
@@ -157,7 +157,6 @@ class StoreTeamService:
|
||||
store_user = StoreUser(
|
||||
store_id=store.id,
|
||||
user_id=user.id,
|
||||
user_type=StoreUserType.TEAM_MEMBER.value,
|
||||
role_id=role.id,
|
||||
invited_by=inviter.id,
|
||||
invitation_token=invitation_token,
|
||||
@@ -416,7 +415,6 @@ class StoreTeamService:
|
||||
"first_name": vu.user.first_name,
|
||||
"last_name": vu.user.last_name,
|
||||
"full_name": vu.user.full_name,
|
||||
"user_type": vu.user_type,
|
||||
"role_name": vu.role.name if vu.role else "owner",
|
||||
"role_id": vu.role.id if vu.role else None,
|
||||
"permissions": vu.get_all_permissions(),
|
||||
|
||||
@@ -216,7 +216,7 @@ class TenancyMetricsProvider:
|
||||
db.query(User)
|
||||
.filter(
|
||||
User.id.in_(platform_user_ids),
|
||||
User.role == "admin",
|
||||
User.role.in_(["super_admin", "platform_admin"]),
|
||||
)
|
||||
.count()
|
||||
)
|
||||
@@ -230,11 +230,9 @@ class TenancyMetricsProvider:
|
||||
)
|
||||
|
||||
# Team members: distinct StoreUser users who are NOT merchant owners
|
||||
# Uses user_type="member" AND excludes owner user IDs to avoid overlap
|
||||
team_members = (
|
||||
db.query(func.count(func.distinct(StoreUser.user_id)))
|
||||
.filter(
|
||||
StoreUser.user_type == "member",
|
||||
~StoreUser.user_id.in_(db.query(Merchant.owner_user_id)),
|
||||
)
|
||||
.scalar() or 0
|
||||
|
||||
Reference in New Issue
Block a user