From c17b42dd45ea2cfd9e087fecbf2f6c6d10ced57d Mon Sep 17 00:00:00 2001
From: itqop <leonkl32@gmail.com>
Date: Mon, 5 Jan 2026 17:41:31 +0300
Subject: [PATCH] sec fixes v1

---
 backend/src/app/api/schemas.py                | 26 ++++++++++++-
 backend/src/app/services/asset_service.py     | 39 ++++++++++++++++++-
 .../app/services/batch_operations_service.py  |  5 ++-
 3 files changed, 65 insertions(+), 5 deletions(-)

diff --git a/backend/src/app/api/schemas.py b/backend/src/app/api/schemas.py
index c808fbb..1a097e1 100644
--- a/backend/src/app/api/schemas.py
+++ b/backend/src/app/api/schemas.py
@@ -3,7 +3,7 @@
 from datetime import datetime
 from typing import Optional
 
-from pydantic import BaseModel, EmailStr, Field
+from pydantic import BaseModel, EmailStr, Field, field_validator
 
 from app.domain.models import AssetStatus, AssetType
 
@@ -82,6 +82,14 @@ class CreateUploadRequest(BaseModel):
     size_bytes: int = Field(gt=0, le=21474836480)  # Max 20GB
     folder_id: Optional[str] = Field(None, max_length=36)
 
+    @field_validator("content_type")
+    @classmethod
+    def validate_content_type(cls, v: str) -> str:
+        """Validate content_type is image or video."""
+        if not (v.startswith("image/") or v.startswith("video/")):
+            raise ValueError("Only image/* and video/* content types are supported")
+        return v
+
 
 class CreateUploadResponse(BaseModel):
     """Response with upload credentials."""
@@ -170,12 +178,28 @@ class FolderCreateRequest(BaseModel):
     name: str = Field(min_length=1, max_length=255)
     parent_folder_id: Optional[str] = None
 
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, v: str) -> str:
+        """Validate folder name doesn't contain path separators."""
+        if "/" in v or "\\" in v or "\x00" in v:
+            raise ValueError("Folder name cannot contain path separators or null bytes")
+        return v.strip()
+
 
 class FolderUpdateRequest(BaseModel):
     """Request to update a folder."""
 
     name: str = Field(min_length=1, max_length=255)
 
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, v: str) -> str:
+        """Validate folder name doesn't contain path separators."""
+        if "/" in v or "\\" in v or "\x00" in v:
+            raise ValueError("Folder name cannot contain path separators or null bytes")
+        return v.strip()
+
 
 class BreadcrumbItem(BaseModel):
     """Breadcrumb item for folder navigation."""
diff --git a/backend/src/app/services/asset_service.py b/backend/src/app/services/asset_service.py
index 103dc6c..07f6473 100644
--- a/backend/src/app/services/asset_service.py
+++ b/backend/src/app/services/asset_service.py
@@ -1,6 +1,8 @@
 """Asset management service."""
 
 import os
+import re
+from pathlib import Path
 from typing import AsyncIterator, Optional, Tuple
 
 import redis
@@ -18,6 +20,36 @@ from app.repositories.asset_repository import AssetRepository
 settings = get_settings()
 
 
+def sanitize_filename(filename: str) -> str:
+    """
+    Sanitize filename to prevent path traversal attacks.
+
+    Removes path separators, null bytes, and other dangerous characters.
+    Keeps only the actual filename without any directory path.
+
+    Args:
+        filename: Original filename from user input
+
+    Returns:
+        Sanitized filename (basename only)
+    """
+    # Get only the basename (remove any directory path)
+    filename = os.path.basename(filename)
+
+    # Remove null bytes
+    filename = filename.replace('\x00', '')
+
+    # Remove path separators (just in case)
+    filename = filename.replace('/', '').replace('\\', '')
+
+    # Limit length
+    if len(filename) > 255:
+        name, ext = os.path.splitext(filename)
+        filename = name[:255 - len(ext)] + ext
+
+    return filename
+
+
 class AssetService:
     """Service for asset management operations."""
 
@@ -65,14 +97,17 @@ class AssetService:
         Returns:
             Tuple of (asset, presigned_post_data)
         """
+        # Sanitize filename to prevent path traversal
+        safe_filename = sanitize_filename(original_filename)
+
         asset_type = self._get_asset_type(content_type)
-        _, ext = os.path.splitext(original_filename)
+        _, ext = os.path.splitext(safe_filename)
 
         # Create asset record
         asset = await self.asset_repo.create(
             user_id=user_id,
             asset_type=asset_type,
-            original_filename=original_filename,
+            original_filename=safe_filename,
             content_type=content_type,
             size_bytes=size_bytes,
             storage_key_original="",  # Will be set after upload
diff --git a/backend/src/app/services/batch_operations_service.py b/backend/src/app/services/batch_operations_service.py
index 6d347af..52fd9cd 100644
--- a/backend/src/app/services/batch_operations_service.py
+++ b/backend/src/app/services/batch_operations_service.py
@@ -16,6 +16,7 @@ from app.domain.models import Asset
 from app.infra.s3_client import S3Client
 from app.repositories.asset_repository import AssetRepository
 from app.repositories.folder_repository import FolderRepository
+from app.services.asset_service import sanitize_filename
 
 
 @contextmanager
@@ -219,8 +220,8 @@ class BatchOperationsService:
                             )
                             file_data = response["Body"].read()
 
-                            # Generate unique filename
-                            base_name = asset.original_filename
+                            # Generate unique filename (sanitized to prevent path traversal)
+                            base_name = sanitize_filename(asset.original_filename)
                             unique_name = base_name
                             counter = 1