From 0166992f6b15c541f53dfb8d3a7a61d3c7463791 Mon Sep 17 00:00:00 2001 From: user <70670632+stuxf@users.noreply.github.com> Date: Wed, 29 Apr 2026 21:09:37 +0000 Subject: [PATCH] fix(proxy): contain UI_LOGO_PATH and LITELLM_FAVICON_URL to allowed asset roots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unauthenticated ``/get_image`` and ``/get_favicon`` endpoints accept the admin-set env vars ``UI_LOGO_PATH`` and ``LITELLM_FAVICON_URL`` and return whatever bytes they resolve to, with a hard-coded ``image/jpeg`` or ``image/x-icon`` content-type. Two attack shapes: * ``UI_LOGO_PATH=/etc/passwd`` (or any other readable file path) — any unauthenticated caller exfiltrates the file via ``GET /get_image``. The previous gate was ``os.path.exists(logo_path)`` which fires on every readable file. Same shape for the favicon endpoint. * ``UI_LOGO_PATH=http://169.254.169.254/iam`` (or any internal HTTP service the admin pointed at) — the proxy fetches it server-side and streams the response body to the unauthenticated caller. No URL validation, no Content-Type validation; ``application/json`` AWS metadata gets tunneled out under the ``image/jpeg`` wrapper. New helper module ``litellm/proxy/common_utils/static_asset_utils.py``: * ``resolve_local_asset_path(candidate, allowed_roots)`` — returns the resolved absolute path only if it lives within one of the allowed asset roots. Uses ``realpath`` so symlinks pointing outside the roots are caught. * ``fetch_validated_image_bytes(url)`` — runs the URL through ``validate_url`` (rejecting private / cloud-metadata / loopback targets) and only returns the response body if the upstream Content-Type is in a small allowlist of image MIME types. Both ``/get_image`` and ``/get_favicon`` are wired through the helpers. The SSRF gate is enforced unconditionally — these endpoints are unauthenticated, so the admin-facing ``litellm.user_url_validation`` toggle does not apply (an admin who opted out of URL validation for LLM provider paths shouldn't also expose ``/get_image`` to SSRF). Tests: - ``TestResolveLocalAssetPath``: 10 cases covering legitimate paths, ``/etc/passwd``, ``/proc/self/environ``, symlink-out, ``..`` traversal, directories, missing files, and root list edge cases. - ``TestFetchValidatedImageBytes``: 7 cases covering SSRF block, non- image content-type rejection, valid image passthrough, non-200 response, fetch exception, empty URL, and parametrized coverage of every allowed image MIME type. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../proxy/common_utils/static_asset_utils.py | 132 +++++++++ litellm/proxy/proxy_server.py | 121 ++++----- .../common_utils/test_static_asset_utils.py | 252 ++++++++++++++++++ 3 files changed, 442 insertions(+), 63 deletions(-) create mode 100644 litellm/proxy/common_utils/static_asset_utils.py create mode 100644 tests/test_litellm/proxy/common_utils/test_static_asset_utils.py diff --git a/litellm/proxy/common_utils/static_asset_utils.py b/litellm/proxy/common_utils/static_asset_utils.py new file mode 100644 index 00000000000..0643572118b --- /dev/null +++ b/litellm/proxy/common_utils/static_asset_utils.py @@ -0,0 +1,132 @@ +""" +Helpers for the unauthenticated logo / favicon endpoints (``/get_image`` and +``/get_favicon``). Both read an admin-set environment variable that may be a +local filesystem path or an HTTP URL, fetch the resource, and return the +bytes verbatim to any unauthenticated caller. + +Without these helpers: + +* a misconfigured / hostile env var like ``UI_LOGO_PATH=/etc/passwd`` lets + any unauthenticated caller exfiltrate the file (LFI — GHSA-3pcp-536p-ghjc). +* a legitimate-looking ``UI_LOGO_PATH=http://internal-service/branding.png`` + pointing at a private host lets any unauthenticated caller exfiltrate + whatever that host returns (SSRF — GHSA-pjc9-2hw6-78rr), regardless of + whether the body is actually an image. +""" + +import os +from typing import List, Optional + +from litellm._logging import verbose_proxy_logger +from litellm.litellm_core_utils.url_utils import SSRFError, validate_url +from litellm.llms.custom_httpx.http_handler import get_async_httpx_client +from litellm.types.llms.custom_http import httpxSpecialProvider + +# Conservative allowlist of image MIME types. Anything else is refused — +# without this, an admin-configured URL whose upstream returns +# ``application/json`` (e.g. cloud metadata, internal API) would still be +# served back to the caller verbatim. +ALLOWED_IMAGE_CONTENT_TYPES = frozenset( + { + "image/jpeg", + "image/jpg", + "image/png", + "image/gif", + "image/svg+xml", + "image/webp", + "image/x-icon", + "image/vnd.microsoft.icon", + } +) + + +def resolve_local_asset_path(candidate: str, allowed_roots: List[str]) -> Optional[str]: + """ + Resolve ``candidate`` and return its absolute path only if it lives + within one of ``allowed_roots``. Returns None on any miss (caller + falls back to the bundled default asset). + + Resolution uses ``realpath`` to follow symlinks, so a symlink inside + ``allowed_roots`` pointing at ``/etc/passwd`` is rejected the same as + a direct ``/etc/passwd`` config. + """ + if not candidate: + return None + try: + resolved = os.path.realpath(os.path.expanduser(candidate)) + except (OSError, ValueError): + return None + if not os.path.isfile(resolved): + return None + for root in allowed_roots: + if not root: + continue + try: + root_resolved = os.path.realpath(root) + except (OSError, ValueError): + continue + if resolved == root_resolved: + return resolved + if resolved.startswith(root_resolved + os.sep): + return resolved + return None + + +async def fetch_validated_image_bytes( + url: str, *, timeout_s: float = 5.0 +) -> Optional[bytes]: + """ + Fetch ``url`` with SSRF protection (always-on) and Content-Type + validation. Returns the raw bytes on success, ``None`` on any + failure (blocked target, non-200, or non-image response). + + The SSRF guard is enforced unconditionally — these endpoints are + unauthenticated, so the admin-facing ``litellm.user_url_validation`` + toggle does not apply. An admin who opted out of URL validation for + LLM provider paths should not also expose ``/get_image`` to SSRF. + """ + if not url: + return None + try: + rewritten_url, host_header = validate_url(url) + except SSRFError as exc: + verbose_proxy_logger.warning( + "Blocked unauthenticated asset fetch — SSRF guard rejected %r: %s", + url, + exc, + ) + return None + + # ``validate_url`` rewrites HTTP URLs to point at a validated IP and + # returns the original hostname for the Host header. For HTTPS with + # ssl_verify enabled, it returns the URL unchanged (TLS hostname + # validation handles DNS rebinding). + request_kwargs = {} + if rewritten_url != url: + request_kwargs["headers"] = {"host": host_header} + + async_client = get_async_httpx_client( + llm_provider=httpxSpecialProvider.UI, + params={"timeout": timeout_s}, + ) + try: + response = await async_client.get(rewritten_url, **request_kwargs) + except Exception as exc: + verbose_proxy_logger.debug("Asset fetch failed for %r: %s", url, exc) + return None + + if response.status_code != 200: + return None + + content_type = ( + (response.headers.get("content-type") or "").split(";")[0].strip().lower() + ) + if content_type not in ALLOWED_IMAGE_CONTENT_TYPES: + verbose_proxy_logger.warning( + "Asset fetch from %r returned non-image content-type %r — refusing to serve.", + url, + content_type, + ) + return None + + return response.content diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 870ea78f17a..bbd528072fd 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -12270,13 +12270,25 @@ async def get_image(): logo_path = os.getenv("UI_LOGO_PATH", default_logo) verbose_proxy_logger.debug("Reading logo from path: %s", logo_path) - # If UI_LOGO_PATH points to a local file, serve it directly (skip cache) + # ``/get_image`` is unauthenticated. Validate any admin-configured local + # path against an allowlist of asset roots — without this guard, an + # env var like ``UI_LOGO_PATH=/etc/passwd`` lets any caller exfiltrate + # the file via this endpoint. + from litellm.proxy.common_utils.static_asset_utils import ( + fetch_validated_image_bytes, + resolve_local_asset_path, + ) + + allowed_local_roots = [assets_dir, current_dir] + if logo_path != default_logo and not logo_path.startswith(("http://", "https://")): - if os.path.exists(logo_path): - return FileResponse(logo_path, media_type="image/jpeg") - # Custom path doesn't exist — fall back to default + safe_logo = resolve_local_asset_path(logo_path, allowed_local_roots) + if safe_logo is not None: + return FileResponse(safe_logo, media_type="image/jpeg") verbose_proxy_logger.warning( - f"UI_LOGO_PATH '{logo_path}' does not exist, falling back to default logo" + "UI_LOGO_PATH %r is outside the allowed asset roots or does not " + "exist, falling back to default logo", + logo_path, ) logo_path = default_logo @@ -12286,32 +12298,21 @@ async def get_image(): # Check if the logo path is an HTTP/HTTPS URL if logo_path.startswith(("http://", "https://")): - try: - # Download the image and cache it - from litellm.llms.custom_httpx.http_handler import get_async_httpx_client - from litellm.types.llms.custom_http import httpxSpecialProvider - - async_client = get_async_httpx_client( - llm_provider=httpxSpecialProvider.UI, - params={"timeout": 5.0}, - ) - response = await async_client.get(logo_path) - if response.status_code == 200: - # Save the image to a local file + # SSRF + content-type validation — the helper rejects + # private/internal/cloud-metadata targets and non-image responses. + image_bytes = await fetch_validated_image_bytes(logo_path) + if image_bytes is not None: + try: with open(cache_path, "wb") as f: - f.write(response.content) - - # Return the cached image as a FileResponse + f.write(image_bytes) return FileResponse(cache_path, media_type="image/jpeg") - else: - # Handle the case when the image cannot be downloaded - return FileResponse(default_logo, media_type="image/jpeg") - except Exception as e: - # Handle any exceptions during the download (e.g., timeout, connection error) - verbose_proxy_logger.debug(f"Error downloading logo from {logo_path}: {e}") - return FileResponse(default_logo, media_type="image/jpeg") + except OSError as e: + verbose_proxy_logger.debug( + "Could not write logo cache to %s: %s", cache_path, e + ) + return FileResponse(default_logo, media_type="image/jpeg") else: - # Return the local image file if the logo path is not an HTTP/HTTPS URL + # Default logo (resolved from the bundled asset, not user-controlled). return FileResponse(logo_path, media_type="image/jpeg") @@ -12320,8 +12321,14 @@ async def get_favicon(): """Get custom favicon for the admin UI.""" from fastapi.responses import Response + from litellm.proxy.common_utils.static_asset_utils import ( + fetch_validated_image_bytes, + resolve_local_asset_path, + ) + current_dir = os.path.dirname(os.path.abspath(__file__)) default_favicon = os.path.join(current_dir, "_experimental", "out", "favicon.ico") + favicon_assets_dir = os.path.dirname(default_favicon) favicon_url = os.getenv("LITELLM_FAVICON_URL", "") @@ -12331,42 +12338,30 @@ async def get_favicon(): raise HTTPException(status_code=404, detail="Default favicon not found") if favicon_url.startswith(("http://", "https://")): - try: - from litellm.llms.custom_httpx.http_handler import get_async_httpx_client - from litellm.types.llms.custom_http import httpxSpecialProvider - - async_client = get_async_httpx_client( - llm_provider=httpxSpecialProvider.UI, - params={"timeout": 5.0}, - ) - response = await async_client.get(favicon_url) - if response.status_code == 200: - content_type = response.headers.get("content-type", "image/x-icon") - return Response( - content=response.content, - media_type=content_type, - ) - else: - verbose_proxy_logger.warning( - "Failed to fetch favicon from %s: status %s", - favicon_url, - response.status_code, - ) - if os.path.exists(default_favicon): - return FileResponse(default_favicon, media_type="image/x-icon") - raise HTTPException(status_code=404, detail="Favicon not found") - except HTTPException: - raise - except Exception as e: - verbose_proxy_logger.debug( - "Error downloading favicon from %s: %s", favicon_url, e - ) - if os.path.exists(default_favicon): - return FileResponse(default_favicon, media_type="image/x-icon") - raise HTTPException(status_code=404, detail="Favicon not found") + # SSRF + content-type validation — the helper rejects + # private/internal/cloud-metadata targets and non-image responses. + image_bytes = await fetch_validated_image_bytes(favicon_url) + if image_bytes is not None: + return Response(content=image_bytes, media_type="image/x-icon") + verbose_proxy_logger.warning( + "Failed to fetch favicon from %s — falling back to default", favicon_url + ) + if os.path.exists(default_favicon): + return FileResponse(default_favicon, media_type="image/x-icon") + raise HTTPException(status_code=404, detail="Favicon not found") else: - if os.path.exists(favicon_url): - return FileResponse(favicon_url, media_type="image/x-icon") + # ``/get_favicon`` is unauthenticated. Validate any admin-configured + # local path against an allowlist of asset roots — see ``/get_image`` + # for the LFI threat-model rationale. + allowed_local_roots = [favicon_assets_dir, current_dir] + safe_favicon = resolve_local_asset_path(favicon_url, allowed_local_roots) + if safe_favicon is not None: + return FileResponse(safe_favicon, media_type="image/x-icon") + verbose_proxy_logger.warning( + "LITELLM_FAVICON_URL %r is outside the allowed asset roots or " + "does not exist, falling back to default favicon", + favicon_url, + ) if os.path.exists(default_favicon): return FileResponse(default_favicon, media_type="image/x-icon") raise HTTPException(status_code=404, detail="Favicon not found") diff --git a/tests/test_litellm/proxy/common_utils/test_static_asset_utils.py b/tests/test_litellm/proxy/common_utils/test_static_asset_utils.py new file mode 100644 index 00000000000..6fe3b04bf02 --- /dev/null +++ b/tests/test_litellm/proxy/common_utils/test_static_asset_utils.py @@ -0,0 +1,252 @@ +""" +Unit tests for the unauthenticated logo / favicon endpoint helpers. + +Closes the LFI half of GHSA-3pcp-536p-ghjc and the SSRF half of +GHSA-pjc9-2hw6-78rr — both endpoints accept an admin-set env var and +return its contents unauthenticated, so the helpers must reject: + +* local paths outside the allowed asset roots (LFI) +* HTTP URLs resolving to private / cloud-metadata addresses (SSRF) +* non-image responses (smuggling JSON / credentials through the + ``image/jpeg`` response wrapper) +""" + +import os +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +sys.path.insert(0, os.path.abspath("../../../..")) + +from litellm.litellm_core_utils.url_utils import SSRFError +from litellm.proxy.common_utils.static_asset_utils import ( + ALLOWED_IMAGE_CONTENT_TYPES, + fetch_validated_image_bytes, + resolve_local_asset_path, +) + + +class TestResolveLocalAssetPath: + @pytest.fixture + def assets_dir(self, tmp_path): + d = tmp_path / "assets" + d.mkdir() + return d + + def test_returns_resolved_path_for_file_inside_allowed_root(self, assets_dir): + logo = assets_dir / "logo.jpg" + logo.write_bytes(b"\xff\xd8\xff") # JPEG header + + result = resolve_local_asset_path(str(logo), [str(assets_dir)]) + assert result == str(logo.resolve()) + + def test_rejects_path_outside_allowed_roots(self, tmp_path, assets_dir): + outside = tmp_path / "secret.txt" + outside.write_text("password=hunter2") + + result = resolve_local_asset_path(str(outside), [str(assets_dir)]) + assert result is None + + def test_rejects_etc_passwd(self, assets_dir): + # The canonical LFI shape from GHSA-3pcp-536p-ghjc. + result = resolve_local_asset_path("/etc/passwd", [str(assets_dir)]) + assert result is None + + def test_rejects_proc_self_environ(self, assets_dir): + # Process environment exfil — same shape as /etc/passwd attack. + result = resolve_local_asset_path("/proc/self/environ", [str(assets_dir)]) + assert result is None + + def test_rejects_symlink_pointing_outside_allowed_roots(self, tmp_path, assets_dir): + secret = tmp_path / "secret.txt" + secret.write_text("password=hunter2") + sneaky = assets_dir / "logo.jpg" + os.symlink(str(secret), str(sneaky)) + + result = resolve_local_asset_path(str(sneaky), [str(assets_dir)]) + assert result is None + + def test_rejects_path_traversal_with_dotdot(self, tmp_path, assets_dir): + outside = tmp_path / "secret.txt" + outside.write_text("nope") + traversal = str(assets_dir / ".." / "secret.txt") + + result = resolve_local_asset_path(traversal, [str(assets_dir)]) + assert result is None + + def test_rejects_directory(self, assets_dir): + # Path containment requires the resolved entry to be a regular file. + result = resolve_local_asset_path(str(assets_dir), [str(assets_dir)]) + assert result is None + + def test_rejects_nonexistent_file_inside_allowed_root(self, assets_dir): + # Even a path that *would* be inside the allowed root must point at + # an existing file — otherwise we shouldn't pretend it resolves. + result = resolve_local_asset_path( + str(assets_dir / "missing.jpg"), [str(assets_dir)] + ) + assert result is None + + def test_rejects_empty_or_none(self, assets_dir): + assert resolve_local_asset_path("", [str(assets_dir)]) is None + + def test_skips_empty_or_invalid_roots(self, assets_dir): + logo = assets_dir / "logo.jpg" + logo.write_bytes(b"\xff\xd8\xff") + result = resolve_local_asset_path( + str(logo), ["", str(assets_dir), "/nonexistent/root"] + ) + assert result == str(logo.resolve()) + + +class TestFetchValidatedImageBytes: + @pytest.fixture + def mock_async_client(self): + client = MagicMock() + client.get = AsyncMock() + return client + + @pytest.mark.asyncio + async def test_blocks_private_ip_via_validate_url(self, mock_async_client): + # The SSRF half of GHSA-pjc9-2hw6-78rr — admin sets logo URL to + # http://169.254.169.254/iam, attacker hits /get_image, exfils creds. + with ( + patch( + "litellm.proxy.common_utils.static_asset_utils.validate_url", + side_effect=SSRFError("blocked: 169.254.169.254"), + ), + patch( + "litellm.proxy.common_utils.static_asset_utils.get_async_httpx_client", + return_value=mock_async_client, + ), + ): + result = await fetch_validated_image_bytes("http://169.254.169.254/iam") + + assert result is None + # The fetch must not be attempted when the URL is rejected. + mock_async_client.get.assert_not_called() + + @pytest.mark.asyncio + async def test_rejects_non_image_content_type(self, mock_async_client): + # Even when the URL passes SSRF, the upstream response must be an + # image. Otherwise an attacker could redirect to an upstream that + # returns ``application/json`` AWS creds and have them tunneled + # through the ``image/jpeg`` response wrapper. + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"content-type": "application/json"} + mock_response.content = b'{"AccessKeyId": "..."}' + mock_async_client.get.return_value = mock_response + + with ( + patch( + "litellm.proxy.common_utils.static_asset_utils.validate_url", + return_value=("http://cdn.example/logo", "cdn.example"), + ), + patch( + "litellm.proxy.common_utils.static_asset_utils.get_async_httpx_client", + return_value=mock_async_client, + ), + ): + result = await fetch_validated_image_bytes("http://cdn.example/logo") + + assert result is None + + @pytest.mark.asyncio + async def test_returns_bytes_for_valid_image_response(self, mock_async_client): + png_bytes = b"\x89PNG\r\n\x1a\nfake png body" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"content-type": "image/png; charset=binary"} + mock_response.content = png_bytes + mock_async_client.get.return_value = mock_response + + with ( + patch( + "litellm.proxy.common_utils.static_asset_utils.validate_url", + return_value=( + "https://cdn.example/logo.png", + "cdn.example", + ), + ), + patch( + "litellm.proxy.common_utils.static_asset_utils.get_async_httpx_client", + return_value=mock_async_client, + ), + ): + result = await fetch_validated_image_bytes("https://cdn.example/logo.png") + + assert result == png_bytes + + @pytest.mark.asyncio + async def test_returns_none_on_non_200_response(self, mock_async_client): + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.headers = {"content-type": "image/png"} + mock_async_client.get.return_value = mock_response + + with ( + patch( + "litellm.proxy.common_utils.static_asset_utils.validate_url", + return_value=("https://cdn.example/logo", "cdn.example"), + ), + patch( + "litellm.proxy.common_utils.static_asset_utils.get_async_httpx_client", + return_value=mock_async_client, + ), + ): + result = await fetch_validated_image_bytes("https://cdn.example/logo") + + assert result is None + + @pytest.mark.asyncio + async def test_returns_none_on_fetch_exception(self, mock_async_client): + mock_async_client.get.side_effect = Exception("connection reset") + + with ( + patch( + "litellm.proxy.common_utils.static_asset_utils.validate_url", + return_value=("https://cdn.example/logo", "cdn.example"), + ), + patch( + "litellm.proxy.common_utils.static_asset_utils.get_async_httpx_client", + return_value=mock_async_client, + ), + ): + result = await fetch_validated_image_bytes("https://cdn.example/logo") + + assert result is None + + @pytest.mark.asyncio + async def test_returns_none_for_empty_url(self): + result = await fetch_validated_image_bytes("") + assert result is None + + @pytest.mark.parametrize( + "content_type", + sorted(ALLOWED_IMAGE_CONTENT_TYPES), + ) + @pytest.mark.asyncio + async def test_accepts_each_allowed_image_content_type( + self, mock_async_client, content_type + ): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"content-type": content_type} + mock_response.content = b"image-bytes" + mock_async_client.get.return_value = mock_response + + with ( + patch( + "litellm.proxy.common_utils.static_asset_utils.validate_url", + return_value=("https://cdn.example/logo", "cdn.example"), + ), + patch( + "litellm.proxy.common_utils.static_asset_utils.get_async_httpx_client", + return_value=mock_async_client, + ), + ): + result = await fetch_validated_image_bytes("https://cdn.example/logo") + + assert result == b"image-bytes"