vulnerability-fixes.md

January 14, 2025 · View on GitHub

Se puede encontrar docs de la API en http://localhost:8091/docs

Level 0 - Technology Details Exposed Via Http Header

El endpoint /healthcheck esta exponiendo versiones de tecnologias utilizadas. Debemos removerlo.

Vuln code:

@router.get("/healthcheck")
def healthcheck(response: Response):
    response.headers["X-Powered-By"] = "Python 3.8, FastAPI ^0.103.0"
    return {"ok": True}

Fixed code:

@router.get("/healthcheck")
def healthcheck():
    return {"ok": True}

Cualquiera puede usar el endpoint /menu/{id} con el metodo DELETE para borrar items del menu, debemos ponerle un control de accesos (basado en roles).

Vuln code:

@router.delete("/menu/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_menu_item(
    item_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Session = Depends(get_db),
    # auth=Depends(RolesBasedAuthChecker([UserRole.EMPLOYEE, UserRole.CHEF])),
):
    utils.delete_menu_item(db, item_id)

Fixed code:

@router.delete("/menu/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_menu_item(
    item_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Session = Depends(get_db),
    auth=Depends(RolesBasedAuthChecker([UserRole.EMPLOYEE, UserRole.CHEF])),
):
    utils.delete_menu_item(db, item_id)

Level 2 - Unrestricted Profile Update Idor

Se pueden modificar los detalles de cualquier usuario al enviar un username en la requests al endpoint /profile con el metodo PUT.

Vuln code:

class UserUpdate(BaseModel):
    username: str
    first_name: Union[str, None] = None
    last_name: Union[str, None] = None
    phone_number: Union[str, None] = None

@router.put("/profile", response_model=UserUpdate, status_code=status.HTTP_200_OK)
def update_profile(
    user: UserUpdate,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Session = Depends(get_db),
    
):
    
    db_user = get_user_by_username(db, user.username)

    for var, value in user.dict().items():
        if value:
            setattr(db_user, var, value)

    db.add(db_user)
    db.commit()
    db.refresh(db_user)

    return db_user

Fixed code:

class UserUpdate(BaseModel):
    first_name: Union[str, None] = None
    last_name: Union[str, None] = None
    phone_number: Union[str, None] = None

@router.put("/profile", response_model=UserUpdate, status_code=status.HTTP_200_OK)
def update_profile(
    user: UserUpdate,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Session = Depends(get_db),
):
    db_user = get_user_by_username(db, current_user.username)

    for var, value in user.dict().items():
        if value:
            setattr(db_user, var, value)

    db.add(db_user)
    db.commit()
    db.refresh(db_user)

    return db_user

Si bien esa solucion funciono, se puede hacer mas facil tomando directamente el username desde la sesion:

### SEGURO
db_user = get_user_by_username(db, current_user.username)

### INSEGURO
db_user = get_user_by_username(db, user.username)

No me convencian las soluciones habia hecho, asi que segui probando con otras cosas y buscando un poco mas y esta es la mejor version funcional que saque (importe HTTPException de FastAPI):

@router.put("/profile", response_model=UserUpdate, status_code=status.HTTP_200_OK)
def update_profile(
    user: UserUpdate,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Session = Depends(get_db),
    
):
    
    if user.username != current_user.username:
        raise HTTPException(status_code=403, detail={'error':'Malicious activity detected'})
    else:
        
        db_user = get_user_by_username(db, current_user.username)

        for var, value in user.dict().items():
            if value:
                setattr(db_user, var, value)

        db.add(db_user)
        db.commit()
        db.refresh(db_user)

        return db_user

Level 3 - Privilege Escalation

Es posible escalar privilegios de customer a employee. Esto se puede lograr usando el endpoint /users/update_role y cambiar el rol.

Vuln code:

@router.put("/users/update_role", response_model=UserRoleUpdate)
async def update_user_role(
    user: UserRoleUpdate,
    current_user: Annotated[models.User, Depends(get_current_user)],
    db: Session = Depends(get_db),
):
    # this method allows staff to give Employee role to other users
    # Chef role is restricted
    if user.role == models.UserRole.CHEF.value:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Only Chef is authorized to add Chef role!",
        )

    db_user = update_user(db, user.username, user)
    return current_user

Fixed code:

@router.put("/users/update_role", response_model=UserRoleUpdate)
async def update_user_role(
    user: UserRoleUpdate,
    current_user: Annotated[models.User, Depends(get_current_user)],
    db: Session = Depends(get_db),
    auth = Depends(RolesBasedAuthChecker([models.UserRole.EMPLOYEE, models.UserRole.CHEF])),
):
    # this method allows staff to give Employee role to other users
    # Chef role is restricted
    if user.role == models.UserRole.CHEF.value:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Only Chef is authorized to add Chef role!",
        )

    db_user = update_user(db, user.username, user)
    return current_user

Level 4 - Server Side Request Forgery

Aparentemente, usando el rol de Employee, se puede acceder a mas endpoints y explotar mas vulns en esos endpoints. Hay un PUT en el endpoint /menu que permite crear items en el menu y setear su imagen via URL… Tambien hay un endpoint escondido que solo puede ser accedido desde localhost, por ende, se puede explotar con SSRF, que es /admin/reset-chef-password…

Esto se podria fixear solo permitiendo dominios especificos para hostear imagenes, y tambien se puede restringir para que solo acepte archivos. Podemos hacerlo en: app/apis/menu/utils.py en la funcion _image_url_to_base64

o en app/apis/menu/servics/update_menu_item_service.py en la funcion update_menu_item

Vuln code /menu/utils.py:

def _image_url_to_base64(image_url: str):
    response = requests.get(image_url)
    return base64.b64encode(response.content).decode()

Vuln code /menu/services/update_menu_item_service.py:

@router.put("/menu/{item_id}", response_model=schemas.MenuItem)
def update_menu_item(
    item_id: int,
    menu_item: schemas.MenuItemCreate,
    current_user: Annotated[User, Depends(get_current_user)],
    db: Session = Depends(get_db),
    auth=Depends(RolesBasedAuthChecker([UserRole.EMPLOYEE, UserRole.CHEF])),
):
    db_item = utils.update_menu_item(db, item_id, menu_item)
    return db_item

Para fixear esta vulnerabilidad decidi hacer todo en utils.py, usando urllib.parse para extraer el host y puerto, y validar con un bucket que seria propio y tambien validando que el content type sea png, y que la extension sea png tambien. Fixed code:

def _image_url_to_base64(image_url: str):
    parsed_url = urllib.parse.urlparse(image_url)
    if parsed_url.netloc is not 'bucket.s3.amazonaws.com:443':
        raise HTTPException(status_code=403, detail="{'error':'Malicious activity detected'}")
    response = requests.get(image_url)
    if response.headers['Content-Type'] == 'image/png':
        if not response.url.endswith('.png'):
            raise HTTPException(status_code=400, detail="{'error':'File type must be png'}")
    else:
        raise HTTPException(status_code=400, detail="{'error':'File type must be png'}")
            
    return base64.b64encode(response.content).decode(

Level 5 - Remote Code Execution

Abusando de la vulnerabilidad anterior, podemos resetar y obtener la passwd del chef:

Si desencodeamos esto, nos devuelve lo siguiente: {"password":"bR(dPWK77NoPm4JXuJreq:sHj9)DV$KC"}

Ahora, logueados como chef, podemos usar el endpoint /admin/stats/disk para checkear el uso del disco. Este endpoint usa la query “parameters”” para pasarle mas argumentos al comando “df” que se ejecuta en el server. Manipulando esto, se pueden inyectar comandos en el servidor.

Vuln code:

def get_disk_usage(parameters: str):
    command = "df -h " + parameters

    try:
        result = subprocess.run(
            command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
        )
        usage = result.stdout.strip().decode()
    except:
        raise Exception("An unexpected error was observed")

    return usage

Para fixearlo, pase los parametros como lista (de esta forma evitamos que sean tomados como string). Aparte de esto:

Hice un split del input del usuario para obtenerlo y compararlo
valide contra una lista de parametros permitidos
Elimine la flag Shell=True de la funcion Gracias por el feedback Krzysztof :) Fixed code:

import subprocess


def get_disk_usage(parameters: str):
    allowed_parameters = [
        "-a", "--all", "-B", "--block-size", "-h", "--human-readable", 
        "-H", "--si", "-i", "--inodes", "-k", "-l", "--local", 
        "--no-sync", "--output", "-P", "--portability", "--sync", 
        "--total", "-t", "--type", "-T", "-x", "--exclude-type"
    ]
    
    input_params = parameters.split()
    for param in input_params:
        if param not in allowed_parameters:
            raise ValueError(f"Invalid parameter: {parameters}")
    
    command = ["df", "-h"]
    command.extend(input_params)

    try:
        result = subprocess.run(
            command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )
        usage = result.stdout.strip().decode()
    except:
        raise Exception("An unexpected error was observed")

    return usage

Level 0 - Technology Details Exposed Via Http Header

Level 1 - Unrestricted Menu Item Deletion

Level 2 - Unrestricted Profile Update Idor

Level 3 - Privilege Escalation

Level 4 - Server Side Request Forgery

Level 5 - Remote Code Execution