Vulnerable code sample

import os
import importlib
from fastapi import FastAPI, Request, Depends, HTTPException, status
from pydantic import BaseModel, Field
from typing import Dict, Optional, Any

# --- Mocks and Simulated Environment for Demonstration ---

class User:
    """A mock user model."""
    def __init__(self, username: str, role: str):
        self.username = username
        self.role = role

# A hardcoded "database" of users and tokens to simulate an auth system.
FAKE_USERS_DB = {
    "user_token": User(username="normal_user", role="user"),
    "admin_token": User(username="admin_user", role="admin"),
}

def get_current_user(request: Request) -> User:
    """
    Simulates a dependency that gets the current authenticated user from a token.
    In the vulnerability, a check for an 'admin' role is missing in the endpoint
    that uses this dependency.
    """
    token = request.headers.get("Authorization")
    if token and token in FAKE_USERS_DB:
        return FAKE_USERS_DB[token]
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid or missing authentication token",
    )

# --- FastAPI Application ---

app = FastAPI()

# A global variable to simulate LiteLLM's in-memory configuration.
LITELLM_CONFIG: Dict[str, Any] = {
    "model_list": [
        {"model_name": "gpt-3.5-turbo", "litellm_params": {"model": "azure/gpt-3.5-turbo"}}
    ],
    "litellm_settings": {
        "set_verbose": False,
    },
}

# --- Pydantic Models for the API Endpoint ---

class ConfigUpdate(BaseModel):
    """
    Defines the payload for the /config/update endpoint, allowing modification
    of critical server settings.
    """
    environment_variables: Optional[Dict[str, str]] = None
    litellm_settings: Optional[Dict[str, Any]] = None
    model_list: Optional[list] = None
    custom_code_module: Optional[str] = Field(None, description="Module to import for custom logic (RCE vector)")

# --- Vulnerable Endpoint Definition (Pre-v1.83.0) ---

@app.post("/config/update")
async def update_config(
    config_update: ConfigUpdate,
    current_user: User = Depends(get_current_user)  # Step 1: User is authenticated.
):
    """
    This endpoint updates the server's configuration. It authenticates the user
    but fails to perform an authorization check to ensure the user is an admin.
    This allows any authenticated user to modify the server's behavior.
    """
    # Step 2: The authorization check is missing.
    # A correct implementation would have a check here like:
    # if current_user.role != "admin":
    #     raise HTTPException(status_code=403, detail="Not authorized")

    # Vector 1: Arbitrary Environment Variable Overwrite.
    # An attacker can set variables like 'UI_USERNAME', 'UI_PASSWORD' to take
    # over the admin account, or 'UI_LOGO_PATH' to read arbitrary files.
    if config_update.environment_variables:
        for key, value in config_update.environment_variables.items():
            os.environ[key] = str(value)

    # Vector 2: Proxy Configuration Modification.
    # An attacker can alter the model list or other settings.
    if config_update.model_list:
        LITELLM_CONFIG["model_list"] = config_update.model_list
    if config_update.litellm_settings:
        LITELLM_CONFIG["litellm_settings"].update(config_update.litellm_settings)

    # Vector 3: Remote Code Execution.
    # An attacker can specify a Python module path. If they can get a
    # malicious file onto the server, this will import and execute it.
    if config_update.custom_code_module:
        try:
            # Dynamically importing the module will execute its top-level code.
            importlib.import_module(config_update.custom_code_module)
        except ImportError as e:
            # The application might silently fail or log the error.
            print(f"Failed to import custom module: {e}")
            pass

    return {"status": "success", "message": "Configuration updated."}

@app.get("/config")
async def get_config(current_user: User = Depends(get_current_user)):
    """A helper endpoint to view the effects of the vulnerability."""
    return {
        "LITELLM_CONFIG": LITELLM_CONFIG,
        "UI_USERNAME": os.environ.get("UI_USERNAME"),
        "UI_PASSWORD": os.environ.get("UI_PASSWORD"),
        "UI_LOGO_PATH": os.environ.get("UI_LOGO_PATH"),
    }