diff --git a/routes/gallery_routes.py b/routes/gallery_routes.py index 8ec2176..dd62ca6 100644 --- a/routes/gallery_routes.py +++ b/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 diff --git a/tests/test_gallery_endpoint_ssrf.py b/tests/test_gallery_endpoint_ssrf.py new file mode 100644 index 0000000..b167919 --- /dev/null +++ b/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