Security: Unauthenticated IDOR on team management endpoints via client-supplied user_id

src/magentic_ui/backend/web/routes/settingsroute.py

@@ -3,14 +3,21 @@
 import yaml
 from typing import Dict
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 
 from ...datamodel import Settings
 from ..deps import get_db
 
 router = APIRouter()
 
 
+def _get_authenticated_user_id(request: Request) -> str:
+    user_id = getattr(request.state, "user_id", None)
+    if not user_id:
+        raise HTTPException(status_code=401, detail="Unauthorized")
+    return user_id
+
+
 def _get_or_create_settings(user_id: str, db) -> Settings:
     """Get existing settings or create default ones for user_id"""
     response = db.get(Settings, filters={"user_id": user_id})
@@ -27,25 +34,32 @@ def _get_or_create_settings(user_id: str, db) -> Settings:
 
 
 @router.get("/")
-async def get_settings(user_id: str, db=Depends(get_db)) -> Dict:
+async def get_settings(request: Request, db=Depends(get_db)) -> Dict:
     try:
+        user_id = _get_authenticated_user_id(request)
         settings = _get_or_create_settings(user_id, db)
         return {"status": True, "data": settings}
+    except HTTPException:
+        raise
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
 
 
 @router.put("/")
-async def update_settings(settings: Settings, db=Depends(get_db)) -> Dict:
+async def update_settings(settings: Settings, request: Request, db=Depends(get_db)) -> Dict:
     try:
+        user_id = _get_authenticated_user_id(request)
+        settings.user_id = user_id
         # Get existing settings to preserve the id
-        existing = _get_or_create_settings(settings.user_id, db)
+        existing = _get_or_create_settings(user_id, db)
         settings.id = existing.id if hasattr(existing, "id") else existing["id"]
 
         response = db.upsert(settings)
         if not response.status:
             raise HTTPException(status_code=400, detail=response.message)
         return {"status": True, "data": response.data}
+    except HTTPException:
+        raise
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
 

src/magentic_ui/backend/web/routes/teams.py

@@ -1,41 +1,52 @@
 # api/routes/teams.py
 from typing import Dict
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 
 from ...datamodel import Team
 from ..deps import get_db
 
 router = APIRouter()
 
 
+def _get_authenticated_user_id(request: Request) -> str:
+    user_id = getattr(request.state, "user_id", None)
+    if not user_id:
+        raise HTTPException(status_code=401, detail="Unauthorized")
+    return user_id
+
+
 @router.get("/")
-async def list_teams(user_id: str, db=Depends(get_db)) -> Dict:
+async def list_teams(request: Request, db=Depends(get_db)) -> Dict:
     """List all teams for a user"""
+    user_id = _get_authenticated_user_id(request)
     response = db.get(Team, filters={"user_id": user_id})
     return {"status": True, "data": response.data}
 
 
 @router.get("/{team_id}")
-async def get_team(team_id: int, user_id: str, db=Depends(get_db)) -> Dict:
+async def get_team(team_id: int, request: Request, db=Depends(get_db)) -> Dict:
     """Get a specific team"""
+    user_id = _get_authenticated_user_id(request)
     response = db.get(Team, filters={"id": team_id, "user_id": user_id})
     if not response.status or not response.data:
         raise HTTPException(status_code=404, detail="Team not found")
     return {"status": True, "data": response.data[0]}
 
 
 @router.post("/")
-async def create_team(team: Team, db=Depends(get_db)) -> Dict:
+async def create_team(team: Team, request: Request, db=Depends(get_db)) -> Dict:
     """Create a new team"""
+    team.user_id = _get_authenticated_user_id(request)
     response = db.upsert(team)
     if not response.status:
         raise HTTPException(status_code=400, detail=response.message)
     return {"status": True, "data": response.data}
 
 
 @router.delete("/{team_id}")
-async def delete_team(team_id: int, user_id: str, db=Depends(get_db)) -> Dict:
+async def delete_team(team_id: int, request: Request, db=Depends(get_db)) -> Dict:
     """Delete a team"""
+    user_id = _get_authenticated_user_id(request)
     db.delete(filters={"id": team_id, "user_id": user_id}, model_class=Team)
     return {"status": True, "message": "Team deleted successfully"}