sec fixes v1

This commit is contained in:
itqop 2026-01-05 17:41:31 +03:00
parent 25dcb7d17f
commit c17b42dd45
3 changed files with 65 additions and 5 deletions

View File

@ -3,7 +3,7 @@
from datetime import datetime from datetime import datetime
from typing import Optional 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 from app.domain.models import AssetStatus, AssetType
@ -82,6 +82,14 @@ class CreateUploadRequest(BaseModel):
size_bytes: int = Field(gt=0, le=21474836480) # Max 20GB size_bytes: int = Field(gt=0, le=21474836480) # Max 20GB
folder_id: Optional[str] = Field(None, max_length=36) 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): class CreateUploadResponse(BaseModel):
"""Response with upload credentials.""" """Response with upload credentials."""
@ -170,12 +178,28 @@ class FolderCreateRequest(BaseModel):
name: str = Field(min_length=1, max_length=255) name: str = Field(min_length=1, max_length=255)
parent_folder_id: Optional[str] = None 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): class FolderUpdateRequest(BaseModel):
"""Request to update a folder.""" """Request to update a folder."""
name: str = Field(min_length=1, max_length=255) 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): class BreadcrumbItem(BaseModel):
"""Breadcrumb item for folder navigation.""" """Breadcrumb item for folder navigation."""

View File

@ -1,6 +1,8 @@
"""Asset management service.""" """Asset management service."""
import os import os
import re
from pathlib import Path
from typing import AsyncIterator, Optional, Tuple from typing import AsyncIterator, Optional, Tuple
import redis import redis
@ -18,6 +20,36 @@ from app.repositories.asset_repository import AssetRepository
settings = get_settings() 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: class AssetService:
"""Service for asset management operations.""" """Service for asset management operations."""
@ -65,14 +97,17 @@ class AssetService:
Returns: Returns:
Tuple of (asset, presigned_post_data) 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) asset_type = self._get_asset_type(content_type)
_, ext = os.path.splitext(original_filename) _, ext = os.path.splitext(safe_filename)
# Create asset record # Create asset record
asset = await self.asset_repo.create( asset = await self.asset_repo.create(
user_id=user_id, user_id=user_id,
asset_type=asset_type, asset_type=asset_type,
original_filename=original_filename, original_filename=safe_filename,
content_type=content_type, content_type=content_type,
size_bytes=size_bytes, size_bytes=size_bytes,
storage_key_original="", # Will be set after upload storage_key_original="", # Will be set after upload

View File

@ -16,6 +16,7 @@ from app.domain.models import Asset
from app.infra.s3_client import S3Client from app.infra.s3_client import S3Client
from app.repositories.asset_repository import AssetRepository from app.repositories.asset_repository import AssetRepository
from app.repositories.folder_repository import FolderRepository from app.repositories.folder_repository import FolderRepository
from app.services.asset_service import sanitize_filename
@contextmanager @contextmanager
@ -219,8 +220,8 @@ class BatchOperationsService:
) )
file_data = response["Body"].read() file_data = response["Body"].read()
# Generate unique filename # Generate unique filename (sanitized to prevent path traversal)
base_name = asset.original_filename base_name = sanitize_filename(asset.original_filename)
unique_name = base_name unique_name = base_name
counter = 1 counter = 1