mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-08 03:08:45 +00:00
fix(security): patch 3 critical vulnerabilities reported by Antiproof/UC Berkeley
1. Custom Code Guardrail RCE: Add PROXY_ADMIN role check to /apply_guardrail endpoint and block frame traversal patterns (cr_frame, gi_frame, f_back, f_globals, etc.) in the code validator to prevent sandbox escape. 2. Skills Sandbox Path Traversal: Sanitize ZIP entry names in extract_all_files() to reject .. and absolute paths, and add realpath containment check in sandbox_executor before writing files. 3. OIDC Arbitrary File Read: Add path validation blocking sensitive directories (/etc, /root, /proc, /sys, /dev) in the OIDC file provider, and restrict /health/test_connection endpoint to admin users only. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
789cf45781
commit
8cdc9ea02b
9 changed files with 264 additions and 5 deletions
|
|
@ -5,6 +5,7 @@ Handles extraction of skill content (SKILL.md) from stored ZIP files
|
|||
and injection into the system prompt for non-Anthropic models.
|
||||
"""
|
||||
|
||||
import os
|
||||
import zipfile
|
||||
from io import BytesIO
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
|
@ -103,8 +104,25 @@ class SkillPromptInjectionHandler:
|
|||
else:
|
||||
clean_path = name
|
||||
|
||||
if clean_path:
|
||||
files[clean_path] = zf.read(name)
|
||||
if not clean_path:
|
||||
continue
|
||||
|
||||
# Reject path traversal attempts
|
||||
if ".." in clean_path.split("/") or clean_path.startswith("/"):
|
||||
verbose_logger.warning(
|
||||
f"SkillPromptInjectionHandler: Skipping malicious ZIP entry: {name}"
|
||||
)
|
||||
continue
|
||||
|
||||
# Normalize and verify no traversal
|
||||
normalized = os.path.normpath(clean_path)
|
||||
if normalized.startswith("..") or os.path.isabs(normalized):
|
||||
verbose_logger.warning(
|
||||
f"SkillPromptInjectionHandler: Skipping path traversal in ZIP entry: {name}"
|
||||
)
|
||||
continue
|
||||
|
||||
files[clean_path] = zf.read(name)
|
||||
except Exception as e:
|
||||
verbose_logger.warning(
|
||||
f"SkillPromptInjectionHandler: Error extracting files from skill {skill.skill_id}: {e}"
|
||||
|
|
|
|||
|
|
@ -97,6 +97,12 @@ class SkillsSandboxExecutor:
|
|||
for path, content in skill_files.items():
|
||||
# Create the file in temp directory
|
||||
local_path = os.path.join(tmpdir, path)
|
||||
# Ensure resolved path stays within tmpdir (prevent path traversal)
|
||||
real_path = os.path.realpath(local_path)
|
||||
if not real_path.startswith(os.path.realpath(tmpdir)):
|
||||
raise ValueError(
|
||||
f"Path traversal detected in skill file: {path}"
|
||||
)
|
||||
os.makedirs(os.path.dirname(local_path), exist_ok=True)
|
||||
with open(local_path, "wb") as f:
|
||||
f.write(content)
|
||||
|
|
|
|||
|
|
@ -2080,6 +2080,12 @@ async def apply_guardrail(
|
|||
"""
|
||||
from litellm.proxy.utils import handle_exception_on_proxy
|
||||
|
||||
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="Admin access required to apply guardrails",
|
||||
)
|
||||
|
||||
try:
|
||||
active_guardrail: Optional[
|
||||
CustomGuardrail
|
||||
|
|
|
|||
|
|
@ -33,6 +33,19 @@ FORBIDDEN_PATTERNS: List[Tuple[str, str]] = [
|
|||
(r"__getattribute__", "__getattribute__ access is not allowed"),
|
||||
(r"__reduce__", "__reduce__ access is not allowed"),
|
||||
(r"__reduce_ex__", "__reduce_ex__ access is not allowed"),
|
||||
# Frame traversal - sandbox escape vectors
|
||||
(r"\.cr_frame", "coroutine frame access is not allowed"),
|
||||
(r"\.gi_frame", "generator frame access is not allowed"),
|
||||
(r"\.ag_frame", "async generator frame access is not allowed"),
|
||||
(r"\.f_back", "frame traversal via f_back is not allowed"),
|
||||
(r"\.f_globals", "frame globals access is not allowed"),
|
||||
(r"\.f_locals", "frame locals access is not allowed"),
|
||||
(r"\.f_code", "frame code access is not allowed"),
|
||||
(r"\.f_builtins", "frame builtins access is not allowed"),
|
||||
(r"\bcurrentframe\b", "currentframe() is not allowed"),
|
||||
(r"\b_getframe\b", "sys._getframe() is not allowed"),
|
||||
(r"\.co_consts", "code object constants access is not allowed"),
|
||||
(r"\.co_names", "code object names access is not allowed"),
|
||||
# OS/system access
|
||||
(r"\bos\.", "os module access is not allowed"),
|
||||
(r"\bsys\.", "sys module access is not allowed"),
|
||||
|
|
|
|||
|
|
@ -1494,7 +1494,13 @@ async def test_model_connection(
|
|||
Returns:
|
||||
dict: A dictionary containing the health check result with either success information or error details.
|
||||
"""
|
||||
from litellm.proxy._types import CommonProxyErrors
|
||||
from litellm.proxy._types import CommonProxyErrors, LitellmUserRoles
|
||||
|
||||
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="Admin access required to test model connections",
|
||||
)
|
||||
from litellm.proxy.management_endpoints.model_management_endpoints import (
|
||||
ModelManagementAuthChecks,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -16,6 +16,39 @@ from litellm.secret_managers.secret_manager_handler import get_secret_from_manag
|
|||
|
||||
oidc_cache = DualCache()
|
||||
|
||||
# Directories blocked from OIDC file reads for security
|
||||
_OIDC_BLOCKED_PATH_PREFIXES = ["/etc/", "/root/", "/proc/", "/sys/", "/dev/"]
|
||||
|
||||
|
||||
def _validate_oidc_file_path(file_path: str) -> None:
|
||||
"""Validate that an OIDC token file path is safe to read."""
|
||||
# Normalize the path to resolve .. components
|
||||
normalized = os.path.normpath(file_path)
|
||||
resolved = os.path.realpath(file_path)
|
||||
|
||||
# Check both normalized and resolved paths against blocked prefixes
|
||||
# (resolved handles symlinks, normalized handles cases where file doesn't exist yet)
|
||||
for prefix in _OIDC_BLOCKED_PATH_PREFIXES:
|
||||
if normalized.startswith(prefix) or resolved.startswith(prefix):
|
||||
raise ValueError(
|
||||
f"OIDC file path not allowed: reading from {prefix} is blocked for security"
|
||||
)
|
||||
# Also check /private variants (macOS)
|
||||
private_prefix = f"/private{prefix}"
|
||||
if resolved.startswith(private_prefix):
|
||||
raise ValueError(
|
||||
f"OIDC file path not allowed: reading from {prefix} is blocked for security"
|
||||
)
|
||||
|
||||
# If LITELLM_ALLOWED_OIDC_DIRS env var is set, enforce allowlist
|
||||
allowed_dirs_env = os.getenv("LITELLM_ALLOWED_OIDC_DIRS")
|
||||
if allowed_dirs_env:
|
||||
allowed = [os.path.realpath(d.strip()) for d in allowed_dirs_env.split(",")]
|
||||
if not any(resolved.startswith(d) for d in allowed):
|
||||
raise ValueError(
|
||||
f"OIDC file path {resolved} not in allowed directories"
|
||||
)
|
||||
|
||||
|
||||
def _get_oidc_http_handler(timeout: Optional[httpx.Timeout] = None) -> HTTPHandler:
|
||||
"""
|
||||
|
|
@ -197,6 +230,7 @@ def get_secret( # noqa: PLR0915
|
|||
return oidc_token
|
||||
elif oidc_provider == "file":
|
||||
# Load token from a file
|
||||
_validate_oidc_file_path(oidc_aud)
|
||||
with open(oidc_aud, "r") as f:
|
||||
oidc_token = f.read()
|
||||
return oidc_token
|
||||
|
|
@ -211,6 +245,7 @@ def get_secret( # noqa: PLR0915
|
|||
token_file_path = os.getenv(oidc_aud)
|
||||
if token_file_path is None:
|
||||
raise ValueError(f"Environment variable {oidc_aud} not found")
|
||||
_validate_oidc_file_path(token_file_path)
|
||||
with open(token_file_path, "r") as f:
|
||||
oidc_token = f.read()
|
||||
return oidc_token
|
||||
|
|
|
|||
|
|
@ -90,5 +90,46 @@ async def test_custom_code_guardrail_apply():
|
|||
assert result["texts"][0] == "test"
|
||||
|
||||
|
||||
# The RBAC endpoint tests are harder to write right here, but the core security
|
||||
# validations are fully covered by the simple tests above.
|
||||
# Phase 4.3: Test frame traversal patterns are blocked
|
||||
|
||||
|
||||
def test_validate_custom_code_cr_frame():
|
||||
code = "async def apply_guardrail(i, r, t):\n x = some_coro.cr_frame\n return allow()"
|
||||
with pytest.raises(CustomCodeValidationError, match="coroutine frame access"):
|
||||
validate_custom_code(code)
|
||||
|
||||
|
||||
def test_validate_custom_code_gi_frame():
|
||||
code = "def apply_guardrail(i, r, t):\n x = gen.gi_frame\n return allow()"
|
||||
with pytest.raises(CustomCodeValidationError, match="generator frame access"):
|
||||
validate_custom_code(code)
|
||||
|
||||
|
||||
def test_validate_custom_code_f_back():
|
||||
code = "def apply_guardrail(i, r, t):\n frame.f_back.f_globals\n return allow()"
|
||||
with pytest.raises(CustomCodeValidationError, match="frame traversal via f_back"):
|
||||
validate_custom_code(code)
|
||||
|
||||
|
||||
def test_validate_custom_code_f_globals():
|
||||
code = "def apply_guardrail(i, r, t):\n x = frame.f_globals\n return allow()"
|
||||
with pytest.raises(CustomCodeValidationError, match="frame globals access"):
|
||||
validate_custom_code(code)
|
||||
|
||||
|
||||
def test_validate_custom_code_currentframe():
|
||||
code = "def apply_guardrail(i, r, t):\n f = currentframe()\n return allow()"
|
||||
with pytest.raises(CustomCodeValidationError, match="currentframe"):
|
||||
validate_custom_code(code)
|
||||
|
||||
|
||||
def test_validate_custom_code_getframe():
|
||||
code = "def apply_guardrail(i, r, t):\n f = _getframe(0)\n return allow()"
|
||||
with pytest.raises(CustomCodeValidationError, match="_getframe"):
|
||||
validate_custom_code(code)
|
||||
|
||||
|
||||
def test_validate_custom_code_co_consts():
|
||||
code = "def apply_guardrail(i, r, t):\n x = func.co_consts\n return allow()"
|
||||
with pytest.raises(CustomCodeValidationError, match="code object constants"):
|
||||
validate_custom_code(code)
|
||||
|
|
|
|||
71
tests/llm_translation/test_skills_security.py
Normal file
71
tests/llm_translation/test_skills_security.py
Normal file
|
|
@ -0,0 +1,71 @@
|
|||
"""
|
||||
Security tests for Skills ZIP extraction path traversal vulnerability.
|
||||
"""
|
||||
|
||||
import os
|
||||
import zipfile
|
||||
from io import BytesIO
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.llms.litellm_proxy.skills.prompt_injection import (
|
||||
SkillPromptInjectionHandler,
|
||||
)
|
||||
|
||||
|
||||
def _make_skill_with_zip(entries: dict) -> MagicMock:
|
||||
"""Create a mock skill with a ZIP containing the given {name: content} entries."""
|
||||
buf = BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
for name, content in entries.items():
|
||||
zf.writestr(name, content)
|
||||
skill = MagicMock()
|
||||
skill.file_content = buf.getvalue()
|
||||
skill.skill_id = "test-skill"
|
||||
return skill
|
||||
|
||||
|
||||
class TestExtractAllFilesPathTraversal:
|
||||
handler = SkillPromptInjectionHandler()
|
||||
|
||||
def test_normal_extraction(self):
|
||||
"""Normal ZIP entries should extract fine."""
|
||||
skill = _make_skill_with_zip({
|
||||
"my-skill/main.py": "print('hello')",
|
||||
"my-skill/lib/utils.py": "x = 1",
|
||||
})
|
||||
files = self.handler.extract_all_files(skill)
|
||||
assert "main.py" in files
|
||||
assert "lib/utils.py" in files
|
||||
|
||||
def test_rejects_dotdot_traversal(self):
|
||||
"""ZIP entries with ../ should be skipped."""
|
||||
skill = _make_skill_with_zip({
|
||||
"my-skill/../../../etc/cron.d/evil": "malicious",
|
||||
"my-skill/legit.py": "safe = True",
|
||||
})
|
||||
files = self.handler.extract_all_files(skill)
|
||||
# Malicious entry should be skipped
|
||||
assert not any("etc" in k for k in files)
|
||||
assert not any(".." in k for k in files)
|
||||
# Legit file should still be extracted
|
||||
assert "legit.py" in files
|
||||
|
||||
def test_rejects_absolute_path(self):
|
||||
"""ZIP entries with absolute paths should be skipped."""
|
||||
skill = _make_skill_with_zip({
|
||||
"my-skill//etc/passwd": "root:x:0:0",
|
||||
"my-skill/ok.py": "ok = True",
|
||||
})
|
||||
files = self.handler.extract_all_files(skill)
|
||||
assert not any("etc" in k for k in files)
|
||||
assert "ok.py" in files
|
||||
|
||||
def test_rejects_deep_traversal(self):
|
||||
"""ZIP entries with deeply nested traversal should be skipped."""
|
||||
skill = _make_skill_with_zip({
|
||||
"skill/../../../../tmp/pwn.sh": "#!/bin/bash\nrm -rf /",
|
||||
})
|
||||
files = self.handler.extract_all_files(skill)
|
||||
assert len(files) == 0
|
||||
|
|
@ -234,3 +234,66 @@ def test_unsupported_oidc_provider():
|
|||
|
||||
with pytest.raises(ValueError, match="Unsupported OIDC provider"):
|
||||
get_secret(secret_name)
|
||||
|
||||
|
||||
# Security tests: OIDC file path validation
|
||||
|
||||
|
||||
def test_oidc_file_blocks_etc_passwd():
|
||||
"""Ensure oidc/file/ cannot read sensitive system files."""
|
||||
secret_name = "oidc/file//etc/passwd"
|
||||
with pytest.raises(ValueError, match="OIDC file path not allowed"):
|
||||
get_secret(secret_name)
|
||||
|
||||
|
||||
def test_oidc_file_blocks_etc_shadow():
|
||||
"""Ensure oidc/file/ blocks /etc/shadow."""
|
||||
secret_name = "oidc/file//etc/shadow"
|
||||
with pytest.raises(ValueError, match="OIDC file path not allowed"):
|
||||
get_secret(secret_name)
|
||||
|
||||
|
||||
def test_oidc_file_blocks_root():
|
||||
"""Ensure oidc/file/ cannot read from /root/."""
|
||||
secret_name = "oidc/file//root/.ssh/id_rsa"
|
||||
with pytest.raises(ValueError, match="OIDC file path not allowed"):
|
||||
get_secret(secret_name)
|
||||
|
||||
|
||||
def test_oidc_file_blocks_proc():
|
||||
"""Ensure oidc/file/ cannot read /proc."""
|
||||
secret_name = "oidc/file//proc/self/environ"
|
||||
with pytest.raises(ValueError, match="OIDC file path not allowed"):
|
||||
get_secret(secret_name)
|
||||
|
||||
|
||||
def test_oidc_file_valid_path_still_works(tmp_path):
|
||||
"""Ensure legitimate OIDC file reads still work."""
|
||||
token_file = tmp_path / "oidc_token.txt"
|
||||
token_file.write_text("valid_token_123")
|
||||
|
||||
secret_name = f"oidc/file/{token_file}"
|
||||
result = get_secret(secret_name)
|
||||
assert result == "valid_token_123"
|
||||
|
||||
|
||||
def test_oidc_env_path_blocks_sensitive_files(mock_env):
|
||||
"""Ensure oidc/env_path/ validates the resolved file path."""
|
||||
mock_env["EVIL_PATH"] = "/etc/passwd"
|
||||
secret_name = "oidc/env_path/EVIL_PATH"
|
||||
with pytest.raises(ValueError, match="OIDC file path not allowed"):
|
||||
get_secret(secret_name)
|
||||
|
||||
|
||||
def test_oidc_file_blocks_sys():
|
||||
"""Ensure oidc/file/ cannot read from /sys/."""
|
||||
secret_name = "oidc/file//sys/kernel/version"
|
||||
with pytest.raises(ValueError, match="OIDC file path not allowed"):
|
||||
get_secret(secret_name)
|
||||
|
||||
|
||||
def test_oidc_file_blocks_dev():
|
||||
"""Ensure oidc/file/ cannot read from /dev/."""
|
||||
secret_name = "oidc/file//dev/null"
|
||||
with pytest.raises(ValueError, match="OIDC file path not allowed"):
|
||||
get_secret(secret_name)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue