scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 85

fix(proxy): contain UI_LOGO_PATH and LITELLM_FAVICON_URL to allowed asset roots

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
user 2026-04-29 21:09:37 +00:00
parent 9b3cd5ca25
commit 0166992f6b
No known key found for this signature in database
3 changed files with 442 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

132
litellm/proxy/common_utils/static_asset_utils.py Normal file
View file

@ -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

121
litellm/proxy/proxy_server.py
View file

@ -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")

252
tests/test_litellm/proxy/common_utils/test_static_asset_utils.py Normal file
View file

@ -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"