fix: validate client-supplied image _endpoint to prevent SSRF (gallery proxies) (#1718)

POST /api/image/harmonize and POST /api/image/inpaint read an `_endpoint` from
the request body and issue server-side httpx POSTs to it with no validation. A
caller can set `_endpoint` to http://169.254.169.254/ (cloud instance metadata)
or any internal/loopback address the server can reach, turning these routes into
an SSRF primitive.

routes/embedding_routes.py already runs its user-supplied endpoint through
src.url_safety.check_outbound_url; these two routes were missing the same guard.
Validate `_endpoint` the same way before any outbound request: non-HTTP(S)
schemes and the link-local metadata range are always rejected, and
IMAGE_BLOCK_PRIVATE_IPS=true blocks private/loopback for full lockdown (the
local-first default still allows LAN diffusion servers).

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

routes/gallery_routes.py

@@ -923,6 +923,16 @@ def setup_gallery_routes() -> APIRouter:
         body = await request.json()
         # Use endpoint from request body (editor dropdown) or fall back to DB lookup
         base = (body.pop("_endpoint", "") or "").rstrip("/")
+        # SSRF hardening: validate a client-supplied endpoint before any
+        # outbound request (mirrors routes/embedding_routes.py).
+        if base:
+            from src.url_safety import check_outbound_url
+            ok, reason = check_outbound_url(
+                base,
+                block_private=os.getenv("IMAGE_BLOCK_PRIVATE_IPS", "false").lower() == "true",
+            )
+            if not ok:
+                raise HTTPException(400, f"Rejected endpoint URL: {reason}")
         chosen_model = (body.pop("_model", "") or "").strip()
         api_key = None
         if not base:
@@ -1115,6 +1125,18 @@ def setup_gallery_routes() -> APIRouter:
             raise HTTPException(400, "No image provided")
 
         endpoint = (body.get("_endpoint") or "").rstrip("/")
+        # SSRF hardening: a client-supplied endpoint is fetched server-side
+        # below, so validate it first (mirrors routes/embedding_routes.py).
+        # Local-first means loopback/LAN is allowed by default; the cloud
+        # metadata range and non-HTTP(S) schemes are always rejected.
+        if endpoint:
+            from src.url_safety import check_outbound_url
+            ok, reason = check_outbound_url(
+                endpoint,
+                block_private=os.getenv("IMAGE_BLOCK_PRIVATE_IPS", "false").lower() == "true",
+            )
+            if not ok:
+                raise HTTPException(400, f"Rejected endpoint URL: {reason}")
         model = (body.get("_model") or "").strip()
 
         base = endpoint

tests/test_gallery_endpoint_ssrf.py

@@ -0,0 +1,44 @@
+"""Regression: the gallery image-edit proxies must validate a client-supplied
+``_endpoint`` through ``check_outbound_url`` before fetching it server-side.
+
+``POST /api/image/harmonize`` and ``POST /api/image/inpaint`` accept an
+``_endpoint`` field in the request body and then issue outbound httpx POSTs to
+it. With no validation this is a server-side request forgery primitive: a caller
+can point ``_endpoint`` at ``http://169.254.169.254/`` (cloud instance metadata)
+or at internal/loopback services the server can reach but the caller cannot.
+
+The analogous user-supplied endpoint in ``routes/embedding_routes.py`` already
+goes through ``check_outbound_url``; these two routes were missing the same
+guard. This test pins the guard in place and confirms the validator rejects the
+metadata range.
+"""
+import ast
+from pathlib import Path
+
+SRC = Path(__file__).resolve().parent.parent / "routes" / "gallery_routes.py"
+
+
+def _function_source(src_text: str, func_name: str) -> str:
+    tree = ast.parse(src_text)
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == func_name:
+            return ast.get_source_segment(src_text, node)
+    raise AssertionError(f"{func_name} not found in {SRC}")
+
+
+def test_endpoint_validated_before_fetch():
+    src = SRC.read_text()
+    for func in ("harmonize_image", "inpaint_proxy"):
+        body = _function_source(src, func)
+        assert "check_outbound_url" in body, (
+            f"{func} must validate the client-supplied _endpoint via "
+            "check_outbound_url before issuing an outbound request"
+        )
+
+
+def test_url_safety_blocks_metadata_endpoint():
+    # The guard is only as strong as the checker: confirm the link-local cloud
+    # metadata address is rejected even with private IPs otherwise allowed.
+    from src.url_safety import check_outbound_url
+    ok, _ = check_outbound_url("http://169.254.169.254/latest/meta-data")
+    assert ok is False