Add validation to prevent privilege escalation via is_superuser field and enhance related tests
- Added explicit Pydantic validation to reject modifications to `is_superuser` in `UserUpdate` schema. - Updated backend logic in `users.py` to support defense-in-depth against privilege escalation. - Introduced comprehensive tests for `/users` and `/users/me` endpoints to ensure `is_superuser` validation works correctly. - Enhanced error handling and validation messages for better clarity and robustness.
This commit is contained in:
@@ -146,10 +146,10 @@ async def update_current_user(
|
||||
Users cannot elevate their own permissions (is_superuser).
|
||||
"""
|
||||
# Prevent users from making themselves superuser
|
||||
# NOTE: UserUpdate schema doesn't include is_superuser, so this is dead code
|
||||
if getattr(user_update, 'is_superuser', None) is not None: # pragma: no cover
|
||||
logger.warning(f"User {current_user.id} attempted to modify is_superuser field") # pragma: no cover
|
||||
raise AuthorizationError( # pragma: no cover
|
||||
# NOTE: Pydantic validator will reject is_superuser != None, but this provides defense in depth
|
||||
if getattr(user_update, 'is_superuser', None) is not None:
|
||||
logger.warning(f"User {current_user.id} attempted to modify is_superuser field")
|
||||
raise AuthorizationError(
|
||||
message="Cannot modify superuser status",
|
||||
error_code=ErrorCode.INSUFFICIENT_PERMISSIONS
|
||||
)
|
||||
@@ -266,10 +266,10 @@ async def update_user(
|
||||
)
|
||||
|
||||
# Prevent non-superusers from modifying superuser status
|
||||
# NOTE: UserUpdate schema doesn't include is_superuser, so this is dead code
|
||||
if getattr(user_update, 'is_superuser', None) is not None and not current_user.is_superuser: # pragma: no cover
|
||||
logger.warning(f"User {current_user.id} attempted to modify is_superuser field") # pragma: no cover
|
||||
raise AuthorizationError( # pragma: no cover
|
||||
# NOTE: Pydantic validator will reject is_superuser != None, but this provides defense in depth
|
||||
if getattr(user_update, 'is_superuser', None) is not None and not current_user.is_superuser:
|
||||
logger.warning(f"User {current_user.id} attempted to modify is_superuser field")
|
||||
raise AuthorizationError(
|
||||
message="Cannot modify superuser status",
|
||||
error_code=ErrorCode.INSUFFICIENT_PERMISSIONS
|
||||
)
|
||||
|
||||
@@ -38,6 +38,7 @@ class UserUpdate(BaseModel):
|
||||
password: Optional[str] = None
|
||||
preferences: Optional[Dict[str, Any]] = None
|
||||
is_active: Optional[bool] = None # Changed default from True to None to avoid unintended updates
|
||||
is_superuser: Optional[bool] = None # Explicitly reject privilege escalation attempts
|
||||
|
||||
@field_validator('phone_number')
|
||||
@classmethod
|
||||
@@ -52,6 +53,14 @@ class UserUpdate(BaseModel):
|
||||
return v
|
||||
return validate_password_strength(v)
|
||||
|
||||
@field_validator('is_superuser')
|
||||
@classmethod
|
||||
def prevent_superuser_modification(cls, v: Optional[bool]) -> Optional[bool]:
|
||||
"""Prevent users from modifying their superuser status via this schema."""
|
||||
if v is not None:
|
||||
raise ValueError("Cannot modify superuser status through user update")
|
||||
return v
|
||||
|
||||
|
||||
class UserInDB(UserBase):
|
||||
id: UUID
|
||||
|
||||
Reference in New Issue
Block a user