Preventing Privilege Escalation

Introduction

Welcome to the fourth lesson of the " Broken Access Control " course! In this lesson, we will explore the concept of privilege escalation, a critical aspect of broken access control vulnerabilities. By understanding how attackers can exploit these vulnerabilities to gain unauthorized access to higher-level privileges, you'll be better equipped to secure your applications. Let's dive in and learn how to prevent privilege escalation! 🚀

Understanding Privilege Escalation

Privilege escalation occurs when an attacker gains elevated access to resources that are normally protected from an application or user. This is a common and severe vulnerability that can give an attacker full control over an application or system. There are two main types of privilege escalation: Vertical Privilege Escalation : This is when an attacker gains higher-level privileges. It's like a regular hotel guest getting a master keycard that opens every room. Example : A regular user manipulates a request to change their role to admin . Example : An attacker accesses an admin-only dashboard (e.g., /admin/dashboard) that lacks proper access control checks. Horizontal Privilege Escalation : This is when an attacker gains access to the resources of another user with the same level of privileges. It's like a hotel guest using their keycard to open the room next door. Example : A user changes the id in a URL like /orders/123 to /orders/124 and successfully views another user's order. Example : An attacker accesses another user's private messages or profile information. Understanding these concepts is vital for securing web applications against unauthorized access. Let's look at a vulnerable code example to see how these vulnerabilities can manifest in real applications.

Vulnerable Code Example

Let's examine a code snippet that demonstrates a vulnerability allowing vertical privilege escalation. This example shows how a lack of proper validation can lead to unauthorized role changes. Python# VULNERABLE: no auth, no field whitelist; allows role escalation @router.put("/users/{id}") async def update_user(id: int, request: Request, db: AsyncSession = Depends(get_db)): data = await request.json() result = await db.execute(select(User).where(User.id == id)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="User not found") # Blindly update all fields from the request payload for key, value in data.items(): setattr(user, key, value) await db.commit() await db.refresh(user) return {"id": user.id, "username": user.username, "role": user.role.value}# VULNERABLE: no auth, no field whitelist; allows role escalation @router.put("/users/{id}") async def update_user(id: int, request: Request, db: AsyncSession = Depends(get_db)): data = await request.json() result = await db.execute(select(User).where(User.id == id)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="User not found") # Blindly update all fields from the request payload for key, value in data.items(): setattr(user, key, value) await db.commit() await db.refresh(user) return {"id": user.id, "username": user.username, "role": user.role.value} The vulnerability lies in the for loop. It iterates through all key-value pairs in the JSON payload sent by the client and uses setattr to update the user object's attributes directly. This is known as mass assignment. Because there is no filter, an attacker can include sensitive fields like role in their request, and the server will blindly update them.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Securing Endpoints with JWT Authentication

The first step to fixing this is to ensure users can only modify their own data. As we saw in the previous lesson, we can use JWT authentication to securely identify the current user.

Instead of accepting a user ID from the URL path (e.g., /users/{id}), which can be easily manipulated, we should create a dedicated endpoint like /users/profile that always acts on the authenticated user. We can extract the user's ID directly from their JWT token using a helper function:

Python

async def get_current_user_id(authorization: str | None, db: AsyncSession) -> int:
    """Extract and validate user ID from JWT token"""
    if not authorization:
        raise HTTPException(status_code=401, detail="Authentication required")
    token = authorization.split(" ")[1] if authorization.startswith("Bearer ") else authorization
    try:
        decoded = jwt.decode(token, JWT_SECRET_KEY, algorithms=["HS256"])
        return int(decoded.get("userId"))
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@router.put("/users/profile")
async def update_own_profile(request: Request, db: AsyncSession = Depends(get_db)):
    # Get the authorization header from the request
    authz = request.headers.get("authorization")
    # Extract the user ID from the JWT token
    user_id = await get_current_user_id(authz, db)

    # Fetch the user from the database using the ID from the token
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Continue with update logic...

async def get_current_user_id(authorization: str | None, db: AsyncSession) -> int:
    """Extract and validate user ID from JWT token"""
    if not authorization:
        raise HTTPException(status_code=401, detail="Authentication required")
    token = authorization.split(" ")[1] if authorization.startswith("Bearer ") else authorization
    try:
        decoded = jwt.decode(token, JWT_SECRET_KEY, algorithms=["HS256"])
        return int(decoded.get("userId"))
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@router.put("/users/profile")
async def update_own_profile(request: Request, db: AsyncSession = Depends(get_db)):
    # Get the authorization header from the request
    authz = request.headers.get("authorization")
    # Extract the user ID from the JWT token
    user_id = await get_current_user_id(authz, db)

    # Fetch the user from the database using the ID from the token
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Continue with update logic...

This modification ensures that users can only update their own profiles, as the ID comes from their authenticated token, not the URL. This effectively prevents horizontal privilege escalation. However, we still need to restrict which fields they can update to prevent vertical privilege escalation.

Preventing Mass Assignment with Whitelisting

To prevent a user from changing their own role, we must stop blindly accepting all fields from the request payload. The best practice is to use a whitelist (or "allow list") of fields that a non-admin user is permitted to change.

We can check the user's role. If they are not an admin, we only process the fields present in our allowed_fields set.

Python

    payload = await request.json()
    
    # Whitelist: only 'username' can be updated by non-admins
    allowed_fields = {"username"}
    safe_update_data = {k: v for k, v in payload.items() if k in allowed_fields}
    
    # If the current user is an admin, they can update any field
    if user.role == Role.ADMIN:
        for key, value in payload.items():
            setattr(user, key, value)
    # Otherwise, only update the fields that are in our whitelist
    else:
        for key, value in safe_update_data.items():
            setattr(user, key, value)

    await db.commit()
    await db.refresh(user)
    return {"id": user.id, "username": user.username, "role": user.role.value}

    payload = await request.json()
    
    # Whitelist: only 'username' can be updated by non-admins
    allowed_fields = {"username"}
    safe_update_data = {k: v for k, v in payload.items() if k in allowed_fields}
    
    # If the current user is an admin, they can update any field
    if user.role == Role.ADMIN:
        for key, value in payload.items():
            setattr(user, key, value)
    # Otherwise, only update the fields that are in our whitelist
    else:
        for key, value in safe_update_data.items():
            setattr(user, key, value)

    await db.commit()
    await db.refresh(user)
    return {"id": user.id, "username": user.username, "role": user.role.value}

You'll notice that we've excluded the password field from our allowed_fields set. This is intentional. Password updates require special handling because they must be hashed before storage, as we learned in the previous lesson. In production applications, password changes should always be handled through a dedicated endpoint that validates the current password and properly hashes the new one. Mixing password updates with general profile updates is a security anti-pattern that can lead to plaintext password storage if the hashing step is forgotten.

In this secure version:

We define an allowed_fields set containing fields that are safe for a user to modify.
We create a safe_update_data dictionary using a dictionary comprehension that contains only the keys from the payload that are also in our allowed_fields set. This prevents them from submitting and changing sensitive fields like role.
We check if the user has the ADMIN role.
If they are an admin, we allow them to update any field, providing administrative flexibility while keeping the default behavior secure.
If they are not an admin, we only apply the filtered safe_update_data to prevent privilege escalation.

This combination of role-checking and field whitelisting provides a robust defense against vertical privilege escalation. By explicitly defining what users can and cannot change, we close the loophole that allowed the mass assignment vulnerability.