refac

backend/open_webui/routers/terminals.py

@@ -6,6 +6,8 @@ Routes:
 """
 
 import logging
+import posixpath
+from urllib.parse import unquote
 
 import aiohttp
 from fastapi import APIRouter, Depends, Request, Response, WebSocket
@@ -27,6 +29,21 @@ STRIPPED_RESPONSE_HEADERS = frozenset(
 )
 
 
+def _sanitize_proxy_path(path: str) -> str | None:
+    """Sanitize a proxy path to prevent directory traversal / SSRF.
+
+    Returns the cleaned path, or None if the path is invalid.
+    """
+    decoded = unquote(path)
+    normalized = posixpath.normpath(decoded)
+    # Remove any leading slashes that would reset the base
+    cleaned = normalized.lstrip("/")
+    # Reject if normpath resolved to parent traversal or current-dir only
+    if cleaned.startswith("..") or cleaned == ".":
+        return None
+    return cleaned
+
+
 @router.get("/")
 async def list_terminal_servers(request: Request, user=Depends(get_verified_user)):
     """Return terminal servers the authenticated user has access to."""
@@ -74,12 +91,16 @@ async def proxy_terminal(
             {"error": "Terminal server URL not configured"}, status_code=503
         )
 
-    target_url = f"{base_url}/{path}"
+    safe_path = _sanitize_proxy_path(path)
+    if safe_path is None:
+        return JSONResponse({"error": "Invalid path"}, status_code=400)
+
+    target_url = f"{base_url}/{safe_path}"
 
     # Route through orchestrator policy endpoint if policy_id is set
     policy_id = connection.get("policy_id")
     if policy_id:
-        target_url = f"{base_url}/p/{policy_id}/{path}"
+        target_url = f"{base_url}/p/{policy_id}/{safe_path}"
 
     if request.query_params:
         target_url += f"?{request.query_params}"