Security fix

Author: Mehdi-Bl

claude_code_api/api/projects.py

@@ -1,7 +1,6 @@
 """Projects API endpoint - Extension to OpenAI API."""
 
 import math
-import os
 import uuid
 
 import structlog
@@ -15,7 +14,7 @@
 )
 from claude_code_api.core.config import settings
 from claude_code_api.core.database import db_manager
-from claude_code_api.core.security import validate_path
+from claude_code_api.core.security import ensure_directory_within_base
 from claude_code_api.models.openai import (
     CreateProjectRequest,
     PaginatedResponse,
@@ -74,10 +73,9 @@ async def create_project(
 
     # Create project directory
     if project_request.path:
-        # Validate path
-        project_path = validate_path(project_request.path, settings.project_root)
-
-        os.makedirs(project_path, exist_ok=True)
+        project_path = ensure_directory_within_base(
+            project_request.path, settings.project_root
+        )
     else:
         project_path = create_project_directory(project_id)
 

claude_code_api/core/claude_manager.py

@@ -11,6 +11,7 @@
 from claude_code_api.models.claude import get_default_model
 
 from .config import settings
+from .security import ensure_directory_within_base
 
 logger = structlog.get_logger()
 
@@ -396,9 +397,7 @@ def _cleanup_process(self, process: ClaudeProcess):
 # Utility functions for project management
 def create_project_directory(project_id: str) -> str:
     """Create project directory."""
-    project_path = os.path.join(settings.project_root, project_id)
-    os.makedirs(project_path, exist_ok=True)
-    return project_path
+    return ensure_directory_within_base(project_id, settings.project_root)
 
 
 def cleanup_project_directory(project_path: str):

claude_code_api/core/security.py

@@ -1,20 +1,21 @@
 """Security utilities."""
 
 import os
+from pathlib import Path
 
 import structlog
 from fastapi import HTTPException, status
 
 logger = structlog.get_logger()
 
 
-def validate_path(path: str, base_path: str) -> str:
+def resolve_path_within_base(path: str, base_path: str) -> str:
     """
-    Validate that a path is safe and within the base path.
-    Prevents directory traversal attacks.
+    Resolve a user-provided path within a base directory.
+    Prevents directory traversal and symlink escapes.
 
     Args:
-        path: The path to validate (can be absolute or relative)
+        path: The path to resolve (can be absolute or relative)
         base_path: The allowed base directory
 
     Returns:
@@ -24,36 +25,71 @@ def validate_path(path: str, base_path: str) -> str:
         HTTPException: If path is invalid or outside base_path
     """
     try:
-        # Normalize base path to absolute path
-        abs_base_path = os.path.abspath(base_path)
-
-        # Handle relative paths by joining with base_path
-        if not os.path.isabs(path):
-            abs_path = os.path.abspath(os.path.join(abs_base_path, path))
-        else:
-            abs_path = os.path.abspath(path)
-
-        # Check if path is within base_path
-        # os.path.commonpath returns the longest common sub-path
-        # If valid, commonpath should be equal to base_path
-        if os.path.commonpath([abs_base_path, abs_path]) != abs_base_path:
+        if path is None or not str(path).strip():
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid path: Path is required",
+            )
+        if "\x00" in str(path):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid path: Null byte detected",
+            )
+
+        abs_base_path = Path(base_path).resolve()
+        candidate_path = Path(path)
+        if not candidate_path.is_absolute():
+            candidate_path = abs_base_path / candidate_path
+
+        resolved_path = candidate_path.resolve(strict=False)
+
+        if not resolved_path.is_relative_to(abs_base_path):
             logger.warning(
                 "Path traversal attempt detected",
                 path=path,
-                resolved_path=abs_path,
-                base_path=abs_base_path,
+                resolved_path=str(resolved_path),
+                base_path=str(abs_base_path),
             )
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail="Invalid path: Path traversal detected",
             )
 
-        return abs_path
+        return str(resolved_path)
 
     except HTTPException:
         raise
     except Exception as e:
         logger.error("Path validation error", error=str(e), path=path)
         raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid path: {str(e)}"
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid path: Path validation failed",
         )
+
+
+def ensure_directory_within_base(
+    path: str, base_path: str, *, allow_subpaths: bool = True
+) -> str:
+    """Validate a path within base_path and create the directory."""
+    path_value = os.fspath(path)
+    if not allow_subpaths:
+        if os.path.isabs(path_value):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid path: Absolute paths are not allowed",
+            )
+        for sep in (os.path.sep, os.path.altsep):
+            if sep and sep in path_value:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="Invalid path: Path separators are not allowed",
+                )
+
+    resolved_path = resolve_path_within_base(path_value, base_path)
+    os.makedirs(resolved_path, exist_ok=True)
+    return resolved_path
+
+
+def validate_path(path: str, base_path: str) -> str:
+    """Backward-compatible wrapper for path resolution."""
+    return resolve_path_within_base(path, base_path)

claude_code_api/utils/streaming.py

@@ -150,8 +150,8 @@ async def convert_stream(
             yield SSEFormatter.format_completion("")
 
         except Exception as e:
-            logger.error("Error in stream conversion", error=str(e))
-            yield SSEFormatter.format_error(f"Stream error: {str(e)}")
+            logger.error("Error in stream conversion", error=str(e), exc_info=True)
+            yield SSEFormatter.format_error("Stream error")
 
     def get_final_response(self) -> Dict[str, Any]:
         """Get complete response in OpenAI format."""
@@ -198,8 +198,10 @@ async def create_stream(
             heartbeat_task.cancel()
 
         except Exception as e:
-            logger.error("Streaming error", session_id=session_id, error=str(e))
-            yield SSEFormatter.format_error(f"Streaming failed: {str(e)}")
+            logger.error(
+                "Streaming error", session_id=session_id, error=str(e), exc_info=True
+            )
+            yield SSEFormatter.format_error("Streaming failed")
         finally:
             # Cleanup
             if session_id in self.active_streams:
@@ -305,10 +307,16 @@ async def create_sse_response(
     session_id: str, model: str, claude_process: ClaudeProcess
 ) -> AsyncGenerator[str, None]:
     """Create SSE response for Claude Code output."""
-    async for chunk in streaming_manager.create_stream(
-        session_id, model, claude_process
-    ):
-        yield chunk
+    try:
+        async for chunk in streaming_manager.create_stream(
+            session_id, model, claude_process
+        ):
+            yield chunk
+    except Exception as e:
+        logger.error(
+            "SSE response error", session_id=session_id, error=str(e), exc_info=True
+        )
+        yield SSEFormatter.format_error("Stream error")
 
 
 def _extract_assistant_payload(