refactor: harden path utils, move imports to module level

Add null byte rejection to safe_join and safe_filename. Normalize
backslash separators in safe_filename for cross-platform safety.
Include resolved path in ValueError for debugging. Move imports
to module level per project conventions.
This commit is contained in:
user
2026-04-16 03:15:04 +00:00
parent 9691649606
commit 278e3f4a6b
3 changed files with 18 additions and 11 deletions
+14 -7
View File
@@ -28,10 +28,13 @@ def safe_join(base_dir: str, *parts: str) -> str:
Raises:
ValueError: If the resolved path escapes base_dir.
"""
for part in parts:
if "\x00" in part:
raise ValueError("Path contains null byte")
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")
raise ValueError(f"Path {resolved!r} escapes base directory {base!r}")
return resolved
@@ -39,8 +42,9 @@ 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.
Strips all directory components (both Unix and Windows separators),
returning only the final name. Use this for uploaded file names
before writing to disk.
Args:
filename: User-supplied filename (may contain path separators).
@@ -49,9 +53,12 @@ def safe_filename(filename: str) -> str:
The basename only, with no directory components.
Raises:
ValueError: If the resulting filename is empty.
ValueError: If the resulting filename is empty or contains null bytes.
"""
name = Path(filename).name
if not name:
raise ValueError("Empty filename")
if "\x00" in filename:
raise ValueError("Filename contains null byte")
# Normalize backslash separators for cross-platform safety
name = filename.replace("\\", "/").rsplit("/", 1)[-1]
if not name or name in (".", ".."):
raise ValueError("Empty or unsafe filename")
return name
@@ -10,6 +10,8 @@ from typing import Any, Dict, List, Optional, Type, TypeVar, Union, cast
from urllib.parse import urlparse
from fastapi import APIRouter, Depends, HTTPException
from litellm.proxy.common_utils.path_utils import safe_join
from pydantic import BaseModel
from litellm._logging import verbose_proxy_logger
@@ -1358,8 +1360,6 @@ 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
try:
yaml_path = safe_join(categories_dir, f"{category_name}.yaml")
+2 -2
View File
@@ -6,6 +6,8 @@ import tempfile
from pathlib import Path
from typing import Any, Dict, List, Optional, cast
from litellm.proxy.common_utils.path_utils import safe_filename
from fastapi import (
APIRouter,
Depends,
@@ -1356,8 +1358,6 @@ async def convert_prompt_file_to_json(
# Read file content
file_content = await file.read()
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)