Security: Settings update allows cross-user modification through untrusted user_id in body

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

@@ -3,14 +3,31 @@
 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 user_id:
+        return user_id
+
+    user = getattr(request.state, "user", None)
+    if isinstance(user, dict):
+        user_id = user.get("user_id") or user.get("id")
+    else:
+        user_id = getattr(user, "user_id", None) or getattr(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})
@@ -36,16 +53,21 @@ async def get_settings(user_id: str, db=Depends(get_db)) -> Dict:
 
 
 @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