mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
fix(proxy): add shared path utilities, prevent directory traversal
Add safe_join() and safe_filename() in proxy/common_utils/path_utils.py for constructing filesystem paths from user-controlled inputs. Apply to guardrail category YAML endpoint and dotprompt file converter.
This commit is contained in:
parent
93faf321df
commit
9691649606
3 changed files with 68 additions and 4 deletions
57
litellm/proxy/common_utils/path_utils.py
Normal file
57
litellm/proxy/common_utils/path_utils.py
Normal file
|
|
@ -0,0 +1,57 @@
|
|||
"""
|
||||
Safe filesystem path construction for user-controlled inputs.
|
||||
|
||||
Use safe_join() instead of os.path.join() whenever a path component
|
||||
comes from user input (request parameters, uploaded filenames, etc.)
|
||||
to prevent directory traversal attacks.
|
||||
"""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def safe_join(base_dir: str, *parts: str) -> str:
|
||||
"""
|
||||
Join path components and verify the result stays within base_dir.
|
||||
|
||||
Resolves symlinks and '..' sequences, then checks the final path
|
||||
is a descendant of base_dir. Raises ValueError if traversal is
|
||||
detected.
|
||||
|
||||
Args:
|
||||
base_dir: The trusted base directory.
|
||||
*parts: User-controlled path components to append.
|
||||
|
||||
Returns:
|
||||
The resolved absolute path as a string.
|
||||
|
||||
Raises:
|
||||
ValueError: If the resolved path escapes base_dir.
|
||||
"""
|
||||
base = os.path.realpath(base_dir)
|
||||
resolved = os.path.realpath(os.path.join(base, *parts))
|
||||
if not (resolved.startswith(base + os.sep) or resolved == base):
|
||||
raise ValueError(f"Path escapes base directory")
|
||||
return resolved
|
||||
|
||||
|
||||
def safe_filename(filename: str) -> str:
|
||||
"""
|
||||
Extract a safe filename from a user-supplied path.
|
||||
|
||||
Strips all directory components, returning only the final name.
|
||||
Use this for uploaded file names before writing to disk.
|
||||
|
||||
Args:
|
||||
filename: User-supplied filename (may contain path separators).
|
||||
|
||||
Returns:
|
||||
The basename only, with no directory components.
|
||||
|
||||
Raises:
|
||||
ValueError: If the resulting filename is empty.
|
||||
"""
|
||||
name = Path(filename).name
|
||||
if not name:
|
||||
raise ValueError("Empty filename")
|
||||
return name
|
||||
|
|
@ -1358,9 +1358,14 @@ async def get_category_yaml(category_name: str):
|
|||
"categories",
|
||||
)
|
||||
|
||||
from litellm.proxy.common_utils.path_utils import safe_join
|
||||
|
||||
# Try to find the file with either .yaml or .json extension
|
||||
yaml_path = os.path.join(categories_dir, f"{category_name}.yaml")
|
||||
json_path = os.path.join(categories_dir, f"{category_name}.json")
|
||||
try:
|
||||
yaml_path = safe_join(categories_dir, f"{category_name}.yaml")
|
||||
json_path = safe_join(categories_dir, f"{category_name}.json")
|
||||
except ValueError:
|
||||
raise HTTPException(status_code=400, detail="Invalid category name")
|
||||
|
||||
category_file_path = None
|
||||
file_type = None
|
||||
|
|
|
|||
|
|
@ -1356,8 +1356,10 @@ async def convert_prompt_file_to_json(
|
|||
# Read file content
|
||||
file_content = await file.read()
|
||||
|
||||
# Create temporary file
|
||||
temp_file_path = Path(tempfile.mkdtemp()) / file.filename
|
||||
from litellm.proxy.common_utils.path_utils import safe_filename
|
||||
|
||||
# Create temporary file — use safe_filename to prevent path traversal
|
||||
temp_file_path = Path(tempfile.mkdtemp()) / safe_filename(file.filename)
|
||||
temp_file_path.write_bytes(file_content)
|
||||
|
||||
# Create a PromptManager instance just for conversion
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue