Patched code sample

import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Path
from pydantic import BaseModel
from typing import Dict, Literal, Optional

# --- Mock Data Models ---
# Role hierarchy: Owner > Admin > Member
ROLE_HIERARCHY = {"owner": 3, "admin": 2, "member": 1}
UserRole = Literal["owner", "admin", "member"]

class User(BaseModel):
    id: int
    username: str

class WorkspaceMember(BaseModel):
    user_id: int
    workspace_id: str
    role: UserRole

class Agent(BaseModel):
    id: str
    name: str
    workspace_id: str  # The crucial field for the fix

# --- Mock Database ---
# A simple in-memory "database" to simulate data storage
DB_USERS = {1: User(id=1, username="user_one")}
DB_AGENTS = {
    "agent_abc": Agent(id="agent_abc", name="Agent from Workspace A", workspace_id="workspace_A"),
    "agent_xyz": Agent(id="agent_xyz", name="Agent from Workspace B", workspace_id="workspace_B"),
}
DB_MEMBERSHIPS = {
    # User 1 is a 'member' in Workspace A and an 'owner' in Workspace C
    (1, "workspace_A"): WorkspaceMember(user_id=1, workspace_id="workspace_A", role="member"),
    (1, "workspace_C"): WorkspaceMember(user_id=1, workspace_id="workspace_C", role="owner"),
}

# --- Mock Authentication and DB Access ---
def get_current_user() -> User:
    """Simulates getting the currently authenticated user."""
    return DB_USERS[1]

def get_agent_from_db(agent_id: str) -> Optional[Agent]:
    """Simulates fetching an agent by its primary key."""
    return DB_AGENTS.get(agent_id)

# --- Patched Authorization Dependency ---
def require_workspace_member(min_role: UserRole = "member"):
    """
    This is the patched dependency factory.
    FIX 1: It now accepts a `min_role` to enforce role-based access control.
    """
    def dependency(
        workspace_id: str = Path(...),
        current_user: User = Depends(get_current_user)
    ) -> WorkspaceMember:
        membership = DB_MEMBERSHIPS.get((current_user.id, workspace_id))

        if not membership:
            raise HTTPException(status_code=403, detail="Not a member of this workspace")

        # This check enforces the role hierarchy, fixing the privilege escalation flaw.
        # A basic 'member' can no longer call routes requiring 'admin' or 'owner'.
        if ROLE_HIERARCHY[membership.role] < ROLE_HIERARCHY[min_role]:
            raise HTTPException(status_code=403, detail=f"Insufficient permissions. Requires role: {min_role}")

        return membership
    return dependency

# --- FastAPI Application with Patched Routes ---
app = FastAPI()

@app.get("/api/v1/workspaces/{workspace_id}/agents/{agent_id}", response_model=Agent)
def get_agent(
    agent_id: str = Path(...),
    # The dependency still checks for basic membership for read operations.
    membership: WorkspaceMember = Depends(require_workspace_member(min_role="member")),
):
    """
    This route demonstrates the fix for the Insecure Direct Object Reference (IDOR).
    A user in 'workspace_A' cannot access an agent from 'workspace_B'.
    """
    agent = get_agent_from_db(agent_id)

    # FIX 2: This check ensures the retrieved resource belongs to the workspace specified
    # in the URL, preventing cross-tenant data leakage.
    if not agent or agent.workspace_id != membership.workspace_id:
        # We return 404 instead of 403 to avoid leaking information about
        # whether the agent_id exists in another tenant's workspace.
        raise HTTPException(status_code=404, detail="Agent not found")

    return agent

@app.delete("/api/v1/workspaces/{workspace_id}")
def delete_workspace(
    # This route now correctly requires the 'owner' role by passing it to the
    # patched dependency, fixing the privilege escalation vulnerability.
    membership: WorkspaceMember = Depends(require_workspace_member(min_role="owner")),
):
    """
    This route demonstrates the fix for the privilege escalation vulnerability.
    Only an 'owner' can delete the workspace.
    """
    # In a real app, logic to delete the workspace would go here.
    # e.g., del DB_WORKSPACES[membership.workspace_id]
    return {"message": f"Workspace '{membership.workspace_id}' deleted successfully."}

# To run this example:
# 1. Install dependencies: pip install fastapi "uvicorn[standard]"
# 2. Run the server: uvicorn name_of_this_file:app --reload
#
# Example failing curl (IDOR):
# User from 'workspace_A' tries to access agent from 'workspace_B'. This will now fail with 404.
# curl -X GET http://127.0.0.1:8000/api/v1/workspaces/workspace_A/agents/agent_xyz
#
# Example failing curl (Privilege Escalation):
# 'member' of 'workspace_A' tries to delete it. This will now fail with 403.
# curl -X DELETE http://127.0.0.1:8000/api/v1/workspaces/workspace_A
#
# Example successful curl:
# 'owner' of 'workspace_C' deletes it. This will succeed.
# curl -X DELETE http://127.0.0.1:8000/api/v1/workspaces/workspace_C