fix: better zip protection

application/api/user/attachments/routes.py

@@ -8,6 +8,7 @@ import uuid
 
 from flask import Response, current_app, jsonify, make_response, redirect, request
 from flask_restx import fields, Namespace, Resource
+from werkzeug.datastructures import FileStorage
 
 from application.api import api
 from application.cache import get_redis_instance
@@ -36,6 +37,11 @@ from application.stt.live_session import (
 )
 from application.stt.stt_creator import STTCreator
 from application.tts.tts_creator import TTSCreator
+from application.upload_limits import (
+    copy_upload_to_path,
+    upload_limit_message,
+    UploadTooLargeError,
+)
 from application.utils import (
     get_agent_image_content_type,
     is_external_image_url,
@@ -122,6 +128,8 @@ def _enforce_uploaded_audio_size_limit(file, filename: str) -> None:
 def _get_store_attachment_user_error(exc: Exception) -> str:
     if isinstance(exc, AudioFileTooLargeError):
         return build_stt_file_size_limit_message()
+    if isinstance(exc, UploadTooLargeError):
+        return upload_limit_message()
     return "Failed to process file"
 
 
@@ -200,7 +208,21 @@ class StoreAttachment(Resource):
                         f"{str(attachment_id)}/{original_filename}"
                     )
 
-                    metadata = storage.save_file(file, relative_path)
+                    # Validate a bounded local spool before storage reads it.
+                    # This protects local and S3 backends even without a
+                    # trustworthy Content-Length header.
+                    with tempfile.TemporaryDirectory() as temp_dir:
+                        staged_path = os.path.join(temp_dir, original_filename)
+                        copy_upload_to_path(file, staged_path)
+                        with open(staged_path, "rb") as staged_stream:
+                            staged_upload = FileStorage(
+                                stream=staged_stream,
+                                filename=original_filename,
+                                content_type=file.mimetype,
+                            )
+                            metadata = storage.save_file(
+                                staged_upload, relative_path
+                            )
                     file_info = {
                         "filename": original_filename,
                         "attachment_id": str(attachment_id),
@@ -224,15 +246,18 @@ class StoreAttachment(Resource):
                     })
             
             if not tasks:
+                size_limit_messages = {
+                    build_stt_file_size_limit_message(),
+                    upload_limit_message(),
+                }
                 if errors and all(
-                    error.get("error") == build_stt_file_size_limit_message()
-                    for error in errors
+                    error.get("error") in size_limit_messages for error in errors
                 ):
                     return make_response(
                         jsonify(
                             {
                                 "success": False,
-                                "message": build_stt_file_size_limit_message(),
+                                "message": errors[0]["error"],
                                 "errors": errors,
                             }
                         ),

application/api/user/sources/upload.py

@@ -17,6 +17,13 @@ from application.core.settings import settings
 from application.storage.db.source_ids import derive_source_id as _derive_source_id
 from application.parser.connectors.connector_creator import ConnectorCreator
 from application.parser.file.constants import SUPPORTED_SOURCE_EXTENSIONS
+from application.security.zip_archive import (
+    extract_zip_safely,
+    safe_zip_error_message,
+    ZipExtractionBudget,
+    ZipExtractionError,
+    ZipExtractionLimits,
+)
 from application.storage.db.repositories.idempotency import IdempotencyRepository
 from application.storage.db.repositories.sources import SourcesRepository
 from application.storage.db.source_config import SourceConfig
@@ -28,6 +35,11 @@ from application.stt.upload_limits import (
     enforce_audio_file_size_limit,
     is_audio_filename,
 )
+from application.upload_limits import (
+    copy_upload_to_path,
+    upload_limit_message,
+    UploadTooLargeError,
+)
 from application.utils import check_required_fields, safe_filename
 
 
@@ -145,15 +157,23 @@ def _release_claim(key):
             "Failed to release task_dedup claim for key=%s", key,
         )
 
-
-
-
 def _enforce_audio_path_size_limit(file_path: str, filename: str) -> None:
     if not is_audio_filename(filename):
         return
     enforce_audio_file_size_limit(os.path.getsize(file_path))
 
 
+def _source_archive_limits() -> ZipExtractionLimits:
+    """Return configured limits shared by all archives in one upload."""
+    return ZipExtractionLimits(
+        max_uncompressed_bytes=settings.UPLOAD_MAX_ARCHIVE_BYTES,
+        max_files=settings.UPLOAD_MAX_ARCHIVE_FILES,
+        max_compression_ratio=settings.UPLOAD_MAX_ARCHIVE_RATIO,
+        max_member_bytes=settings.UPLOAD_MAX_FILE_BYTES,
+        max_depth=settings.UPLOAD_MAX_ARCHIVE_DEPTH,
+    )
+
+
 @sources_upload_ns.route("/upload")
 class UploadFile(Resource):
     @api.expect(
@@ -222,67 +242,66 @@ class UploadFile(Resource):
         base_path = f"{settings.UPLOAD_FOLDER}/{safe_user}/{dir_name}"
         file_name_map = {}
 
+        active_upload_name = "uploaded archive"
         try:
             storage = StorageCreator.get_storage()
 
-            for file in files:
-                original_filename = os.path.basename(file.filename)
-                safe_file = safe_filename(original_filename)
-                if original_filename:
-                    file_name_map[safe_file] = original_filename
+            # Stage and validate every file before the first storage write. A
+            # later oversized file/archive cannot leave a partially accepted
+            # source behind. All ZIPs in the request share one expansion budget.
+            with tempfile.TemporaryDirectory() as temp_dir:
+                pending_files: list[tuple[str, str]] = []
+                archive_budget = ZipExtractionBudget()
+                archive_limits = _source_archive_limits()
 
-                with tempfile.TemporaryDirectory() as temp_dir:
-                    temp_file_path = os.path.join(temp_dir, safe_file)
-                    file.save(temp_file_path)
+                for index, file in enumerate(files):
+                    original_filename = os.path.basename(file.filename)
+                    safe_file = safe_filename(original_filename)
+                    active_upload_name = safe_zip_error_message(
+                        original_filename or safe_file, max_chars=200
+                    )
+                    if original_filename:
+                        file_name_map[safe_file] = original_filename
+
+                    upload_dir = os.path.join(temp_dir, str(index))
+                    os.makedirs(upload_dir, exist_ok=True)
+                    temp_file_path = os.path.join(upload_dir, safe_file)
+                    copy_upload_to_path(file, temp_file_path)
                     _enforce_audio_path_size_limit(temp_file_path, safe_file)
 
-                    # Only extract actual .zip files, not Office formats (.docx, .xlsx, .pptx)
-                    # which are technically zip archives but should be processed as-is
+                    # Office/e-book containers are ZIP-based formats but must
+                    # be parsed as documents, not expanded as user archives.
                     is_office_format = safe_file.lower().endswith(
                         (".docx", ".xlsx", ".pptx", ".odt", ".ods", ".odp", ".epub")
                     )
                     if zipfile.is_zipfile(temp_file_path) and not is_office_format:
-                        try:
-                            with zipfile.ZipFile(temp_file_path, "r") as zip_ref:
-                                zip_ref.extractall(path=temp_dir)
-
-                                # Walk through extracted files and upload them
-
-                                for root, _, files in os.walk(temp_dir):
-                                    for extracted_file in files:
-                                        if (
-                                            os.path.join(root, extracted_file)
-                                            == temp_file_path
-                                        ):
-                                            continue
-                                        rel_path = os.path.relpath(
-                                            os.path.join(root, extracted_file), temp_dir
-                                        )
-                                        storage_path = f"{base_path}/{rel_path}"
-                                        _enforce_audio_path_size_limit(
-                                            os.path.join(root, extracted_file),
-                                            extracted_file,
-                                        )
-
-                                        with open(
-                                            os.path.join(root, extracted_file), "rb"
-                                        ) as f:
-                                            storage.save_file(f, storage_path)
-                        except Exception as e:
-                            current_app.logger.error(
-                                f"Error extracting zip: {e}", exc_info=True
-                            )
-                            # If zip extraction fails, save the original zip file
-
-                            file_path = f"{base_path}/{safe_file}"
-                            with open(temp_file_path, "rb") as f:
-                                storage.save_file(f, file_path)
+                        extract_dir = os.path.join(upload_dir, "extracted")
+                        extract_zip_safely(
+                            temp_file_path,
+                            extract_dir,
+                            archive_limits,
+                            archive_budget,
+                        )
+                        for root, _, extracted_files in os.walk(extract_dir):
+                            for extracted_file in extracted_files:
+                                local_path = os.path.join(root, extracted_file)
+                                _enforce_audio_path_size_limit(
+                                    local_path, extracted_file
+                                )
+                                rel_path = os.path.relpath(
+                                    local_path, extract_dir
+                                ).replace(os.sep, "/")
+                                pending_files.append(
+                                    (local_path, f"{base_path}/{rel_path}")
+                                )
                     else:
-                        # For non-zip files, save directly
+                        pending_files.append(
+                            (temp_file_path, f"{base_path}/{safe_file}")
+                        )
 
-                        file_path = f"{base_path}/{safe_file}"
-                        with open(temp_file_path, "rb") as f:
-                            storage.save_file(f, file_path)
+                for local_path, storage_path in pending_files:
+                    with open(local_path, "rb") as staged_file:
+                        storage.save_file(staged_file, storage_path)
             # Mint the source UUID up here so the HTTP response and the
             # worker's SSE envelopes share one id. With an idempotency
             # key we reuse the deterministic uuid5 (retried task lands on
@@ -324,6 +343,33 @@ class UploadFile(Resource):
                 ),
                 413,
             )
+        except (UploadTooLargeError, ZipExtractionError) as err:
+            zip_error_detail = safe_zip_error_message(err)
+            if isinstance(err, ZipExtractionError):
+                current_app.logger.warning(
+                    "Rejected unsafe ZIP upload %s: %s",
+                    active_upload_name,
+                    zip_error_detail,
+                )
+            else:
+                current_app.logger.warning(
+                    "Rejected oversized upload %s", active_upload_name
+                )
+            if scoped_key:
+                _release_claim(scoped_key)
+            return make_response(
+                jsonify(
+                    {
+                        "success": False,
+                        "message": (
+                            f"{active_upload_name}: {zip_error_detail}"
+                            if isinstance(err, ZipExtractionError)
+                            else upload_limit_message()
+                        ),
+                    }
+                ),
+                413,
+            )
         except Exception as err:
             current_app.logger.error(f"Error uploading file: {err}", exc_info=True)
             if scoped_key:
@@ -678,21 +724,43 @@ class ManageSourceFiles(Resource):
                         cached["source_id"] = resolved_source_id
                         return make_response(jsonify(cached), 200)
 
-                added_files = []
-                map_updated = False
-
                 target_dir = source_file_path
                 if parent_dir:
                     target_dir = f"{source_file_path}/{parent_dir}"
-                for file in files:
-                    if file.filename:
+                added_files = []
+                map_updated = False
+                with tempfile.TemporaryDirectory() as temp_dir:
+                    staged_files: list[tuple[str, str, str, str]] = []
+                    for index, file in enumerate(files):
+                        if not file.filename:
+                            continue
                         original_filename = os.path.basename(file.filename)
                         safe_filename_str = safe_filename(original_filename)
-                        file_path = f"{target_dir}/{safe_filename_str}"
+                        staged_path = os.path.join(
+                            temp_dir, f"{index}-{safe_filename_str}"
+                        )
+                        copy_upload_to_path(file, staged_path)
+                        _enforce_audio_path_size_limit(
+                            staged_path, safe_filename_str
+                        )
+                        staged_files.append(
+                            (
+                                staged_path,
+                                f"{target_dir}/{safe_filename_str}",
+                                safe_filename_str,
+                                original_filename,
+                            )
+                        )
 
-                        # Save file to storage
-
-                        storage.save_file(file, file_path)
+                    # All files are bounded before the first storage mutation.
+                    for (
+                        staged_path,
+                        file_path,
+                        safe_filename_str,
+                        original_filename,
+                    ) in staged_files:
+                        with open(staged_path, "rb") as staged_file:
+                            storage.save_file(staged_file, file_path)
                         added_files.append(safe_filename_str)
                         if original_filename:
                             relative_key = (
@@ -977,6 +1045,17 @@ class ManageSourceFiles(Resource):
                     ),
                     200,
                 )
+        except (UploadTooLargeError, AudioFileTooLargeError) as err:
+            if scoped_key and not claim_transferred:
+                _release_claim(scoped_key)
+            message = (
+                build_stt_file_size_limit_message()
+                if isinstance(err, AudioFileTooLargeError)
+                else upload_limit_message()
+            )
+            return make_response(
+                jsonify({"success": False, "message": message}), 413
+            )
         except Exception as err:
             # Release the dedup claim only if it wasn't transferred to
             # a worker. Without this, a same-key retry within the 24h

application/api/user/tools/routes.py

@@ -18,6 +18,7 @@ from application.agents.tools.tool_manager import ToolManager
 from application.api import api
 from application.api.user.artifacts.authz import Principal, authorize_artifact
 from application.api.user.team_sharing import effective_write_owner, visible_with_access
+from application.core.settings import settings
 from application.core.url_validation import SSRFError, validate_url
 from application.security.encryption import decrypt_credentials, encrypt_credentials
 from application.storage.db.base_repository import looks_like_uuid
@@ -27,6 +28,11 @@ from application.storage.db.repositories.todos import TodosRepository
 from application.storage.db.repositories.user_tools import UserToolsRepository
 from application.storage.db.repositories.users import UsersRepository
 from application.storage.db.session import db_readonly, db_session
+from application.upload_limits import (
+    read_text_upload_limited,
+    upload_limit_message,
+    UploadTooLargeError,
+)
 from application.utils import check_required_fields, validate_function_name
 
 tool_config = {}
@@ -821,7 +827,21 @@ class ParseSpec(Resource):
                     jsonify({"success": False, "message": "No file selected"}), 400
                 )
             try:
-                spec_content = file.read().decode("utf-8")
+                spec_content = read_text_upload_limited(
+                    file, max_bytes=settings.PARSE_SPEC_MAX_BYTES
+                )
+            except UploadTooLargeError:
+                return make_response(
+                    jsonify(
+                        {
+                            "success": False,
+                            "message": upload_limit_message(
+                                settings.PARSE_SPEC_MAX_BYTES
+                            ),
+                        }
+                    ),
+                    413,
+                )
             except UnicodeDecodeError:
                 return make_response(
                     jsonify({"success": False, "message": "Invalid file encoding"}), 400
@@ -833,6 +853,21 @@ class ParseSpec(Resource):
             return make_response(
                 jsonify({"success": False, "message": "No spec provided"}), 400
             )
+        if (
+            not isinstance(spec_content, str)
+            or len(spec_content.encode("utf-8")) > settings.PARSE_SPEC_MAX_BYTES
+        ):
+            return make_response(
+                jsonify(
+                    {
+                        "success": False,
+                        "message": upload_limit_message(
+                            settings.PARSE_SPEC_MAX_BYTES
+                        ),
+                    }
+                ),
+                413,
+            )
         if not spec_content or not spec_content.strip():
             return make_response(
                 jsonify({"success": False, "message": "Empty spec content"}), 400

application/app.py

@@ -6,6 +6,7 @@ import uuid
 import dotenv
 from flask import Flask, Response, jsonify, redirect, request
 from jose import jwt
+from werkzeug.exceptions import RequestEntityTooLarge
 
 from application.auth import handle_auth
 
@@ -35,6 +36,10 @@ from application.stt.upload_limits import (  # noqa: E402
     build_stt_file_size_limit_message,
     should_reject_stt_request,
 )
+from application.upload_limits import (  # noqa: E402
+    is_document_upload_path,
+    upload_request_limit_message,
+)
 
 
 if platform.system() == "Windows":
@@ -85,6 +90,52 @@ app.config.update(
 celery.config_from_object("application.celeryconfig")
 api.init_app(app)
 
+
+def _upload_limit_error_payload() -> dict[str, bool | str]:
+    """Build a consistent 413 payload using the active route-specific limit."""
+    active_limit = request.max_content_length
+    if active_limit is None:
+        active_limit = settings.UPLOAD_MAX_REQUEST_BYTES
+    return {
+        "success": False,
+        "message": upload_request_limit_message(active_limit),
+    }
+
+
+@app.errorhandler(RequestEntityTooLarge)
+def handle_request_entity_too_large(_error):
+    """Return API-shaped JSON instead of Werkzeug's HTML 413 page."""
+    return jsonify(_upload_limit_error_payload()), 413
+
+
+@api.errorhandler(RequestEntityTooLarge)
+def handle_restx_request_entity_too_large(_error):
+    """Keep Flask-RESTX from replacing the configured upload-limit response."""
+    return _upload_limit_error_payload(), 413
+
+
+@app.before_request
+def enforce_document_upload_request_size_limit():
+    """Bound multipart parsing for public file-upload routes only.
+
+    Internal worker index uploads are intentionally excluded; they are trusted
+    service traffic and can legitimately exceed the end-user document limit.
+    """
+    if request.method == "OPTIONS" or not is_document_upload_path(request.path):
+        return None
+    request_limit = (
+        settings.PARSE_SPEC_MAX_BYTES
+        if request.path == "/api/parse_spec" and request.is_json
+        else settings.UPLOAD_MAX_REQUEST_BYTES
+    )
+    request.max_content_length = request_limit
+    if (
+        request.content_length is not None
+        and request.content_length > request_limit
+    ):
+        raise RequestEntityTooLarge()
+    return None
+
 # The same stable secret also signs opaque agent-image capabilities, including
 # in no-auth mode. Production replicas must receive one shared configured key;
 # only local development may use the atomic filesystem fallback.

application/core/settings.py

@@ -88,6 +88,16 @@ class Settings(BaseSettings):
         "request_limit": 500,
     }
     UPLOAD_FOLDER: str = "inputs"
+    # Public upload request cap is applied by Flask before multipart parsing.
+    # The per-file cap is also enforced while copying each controlled stream.
+    UPLOAD_MAX_REQUEST_BYTES: int = Field(default=256 * 1024 * 1024, gt=0)
+    UPLOAD_MAX_FILE_BYTES: int = Field(default=100 * 1024 * 1024, gt=0)
+    PARSE_SPEC_MAX_BYTES: int = Field(default=10 * 1024 * 1024, gt=0)
+    # ZIP limits apply cumulatively across nested archives in one extraction.
+    UPLOAD_MAX_ARCHIVE_BYTES: int = Field(default=250 * 1024 * 1024, gt=0)
+    UPLOAD_MAX_ARCHIVE_FILES: int = Field(default=10_000, gt=0)
+    UPLOAD_MAX_ARCHIVE_RATIO: int = Field(default=1000, gt=0)
+    UPLOAD_MAX_ARCHIVE_DEPTH: int = Field(default=3, ge=0)
     PARSE_PDF_AS_IMAGE: bool = False
     PARSE_IMAGE_REMOTE: bool = False
     DOCLING_OCR_ENABLED: bool = False  # Enable OCR for docling parsers (PDF, images)

application/security/zip_archive.py

@@ -0,0 +1,403 @@
+"""Safe, bounded extraction for untrusted ZIP archives."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import stat
+import tempfile
+import zipfile
+from dataclasses import dataclass
+from pathlib import PurePosixPath
+
+
+_COPY_CHUNK_BYTES = 64 * 1024
+_MAX_MEMBER_NAME_BYTES = 4096
+_MAX_MEMBER_DEPTH = 64
+
+
+class ZipExtractionError(ValueError):
+    """Raised when a ZIP is invalid or exceeds an extraction safety limit."""
+
+
+@dataclass(frozen=True)
+class ZipExtractionLimits:
+    """Resource ceilings for one extraction tree."""
+
+    max_uncompressed_bytes: int
+    max_files: int
+    max_compression_ratio: int
+    max_member_bytes: int | None = None
+    max_depth: int = 3
+
+
+@dataclass
+class ZipExtractionBudget:
+    """Cumulative resources reserved across nested archives."""
+
+    uncompressed_bytes: int = 0
+    files: int = 0
+
+
+@dataclass(frozen=True)
+class _ValidatedMember:
+    info: zipfile.ZipInfo
+    relative_path: str
+
+
+def _collision_safe_member_path(
+    relative_path: str,
+    assignments: dict[tuple[tuple[str, ...], str], str],
+    occupied_names: dict[tuple[str, ...], set[str]],
+) -> str:
+    """Map case-colliding member components to stable, portable names."""
+    original_parent: tuple[str, ...] = ()
+    assigned_parent: tuple[str, ...] = ()
+    assigned_parts: list[str] = []
+    for component in relative_path.split("/"):
+        assignment_key = (original_parent, component)
+        assigned_component = assignments.get(assignment_key)
+        if assigned_component is None:
+            used = occupied_names.setdefault(assigned_parent, set())
+            assigned_component = component
+            suffix = 2
+            stem, extension = os.path.splitext(component)
+            while assigned_component.casefold() in used:
+                assigned_component = f"{stem} ({suffix}){extension}"
+                suffix += 1
+            assignments[assignment_key] = assigned_component
+            used.add(assigned_component.casefold())
+        assigned_parts.append(assigned_component)
+        original_parent += (component,)
+        assigned_parent += (assigned_component,)
+    return "/".join(assigned_parts)
+
+
+def _safe_member_path(name: str) -> str:
+    """Normalize a ZIP member name and reject paths escaping extraction."""
+    if not name or "\x00" in name:
+        raise ZipExtractionError("ZIP contains an invalid empty member name")
+    if len(name.encode("utf-8", errors="surrogatepass")) > _MAX_MEMBER_NAME_BYTES:
+        raise ZipExtractionError("ZIP member name exceeds the safety limit")
+
+    normalized = name.replace("\\", "/")
+    path = PurePosixPath(normalized)
+    parts = path.parts
+    if (
+        path.is_absolute()
+        or normalized.startswith("//")
+        or any(part in {"", ".", ".."} for part in parts)
+        or (parts and parts[0].endswith(":"))
+    ):
+        raise ZipExtractionError(f"ZIP contains a path traversal attempt: {name}")
+    if len(parts) > _MAX_MEMBER_DEPTH:
+        raise ZipExtractionError("ZIP member path is nested too deeply")
+    return "/".join(parts)
+
+
+def _reject_special_member(info: zipfile.ZipInfo) -> None:
+    """Reject symlinks and device-like Unix entries."""
+    mode = info.external_attr >> 16
+    file_type = stat.S_IFMT(mode)
+    if file_type not in {0, stat.S_IFREG, stat.S_IFDIR}:
+        raise ZipExtractionError(f"ZIP contains a special file: {info.filename}")
+    if info.flag_bits & 0x1:
+        raise ZipExtractionError("Encrypted ZIP members are not supported")
+
+
+def _validate_members(
+    archive: zipfile.ZipFile,
+    limits: ZipExtractionLimits,
+    budget: ZipExtractionBudget,
+) -> tuple[list[_ValidatedMember], int, int]:
+    members: list[_ValidatedMember] = []
+    seen_paths: dict[str, bool] = {}
+    component_assignments: dict[tuple[tuple[str, ...], str], str] = {}
+    occupied_names: dict[tuple[str, ...], set[str]] = {}
+    added_bytes = 0
+    added_files = 0
+    added_compressed_bytes = 0
+
+    for info in archive.infolist():
+        relative_path = _safe_member_path(info.filename)
+        is_directory = info.is_dir()
+        prior_is_directory = seen_paths.get(relative_path)
+        if prior_is_directory is not None and not (
+            prior_is_directory and is_directory
+        ):
+            raise ZipExtractionError(f"ZIP contains a duplicate path: {info.filename}")
+        seen_paths[relative_path] = is_directory
+        _reject_special_member(info)
+        extraction_path = _collision_safe_member_path(
+            relative_path, component_assignments, occupied_names
+        )
+
+        if is_directory:
+            members.append(_ValidatedMember(info=info, relative_path=extraction_path))
+            continue
+
+        added_files += 1
+        added_bytes += int(info.file_size)
+        added_compressed_bytes += int(info.compress_size)
+        if limits.max_member_bytes and info.file_size > limits.max_member_bytes:
+            raise ZipExtractionError(
+                "ZIP member exceeds the "
+                f"{limits.max_member_bytes}-byte per-file limit"
+            )
+        if budget.files + added_files > limits.max_files:
+            raise ZipExtractionError(
+                f"ZIP contains too many files (>{limits.max_files})"
+            )
+        if budget.uncompressed_bytes + added_bytes > limits.max_uncompressed_bytes:
+            raise ZipExtractionError(
+                "ZIP extraction exceeds the "
+                f"{limits.max_uncompressed_bytes}-byte expansion limit"
+            )
+
+        members.append(_ValidatedMember(info=info, relative_path=extraction_path))
+
+    if added_bytes:
+        if added_compressed_bytes <= 0:
+            raise ZipExtractionError("ZIP has an invalid compression ratio")
+        ratio = added_bytes / added_compressed_bytes
+        if ratio > limits.max_compression_ratio:
+            raise ZipExtractionError(
+                "ZIP exceeds the "
+                f"{limits.max_compression_ratio}:1 compression ratio limit"
+            )
+
+    return members, added_bytes, added_files
+
+
+def validate_zip_archive(
+    zip_path: str | os.PathLike[str],
+    limits: ZipExtractionLimits,
+    budget: ZipExtractionBudget | None = None,
+) -> None:
+    """Validate central-directory metadata without extracting any content."""
+    active_budget = budget or ZipExtractionBudget()
+    try:
+        with zipfile.ZipFile(zip_path, "r") as archive:
+            _validate_members(archive, limits, active_budget)
+    except (OSError, zipfile.BadZipFile) as exc:
+        raise ZipExtractionError(f"Invalid or corrupted ZIP file: {exc}") from exc
+
+
+def _extract_archive_into(
+    zip_path: str | os.PathLike[str],
+    destination_root: str,
+    limits: ZipExtractionLimits,
+    budget: ZipExtractionBudget,
+) -> list[str]:
+    """Extract one validated archive into a private staging directory."""
+    extracted_files: list[str] = []
+    with zipfile.ZipFile(zip_path, "r") as archive:
+        members, added_bytes, added_files = _validate_members(
+            archive, limits, budget
+        )
+
+        # Reserve declared resources before decompression so nested callers
+        # cannot reset the allowance between layers.
+        budget.uncompressed_bytes += added_bytes
+        budget.files += added_files
+
+        for member in members:
+            target = os.path.realpath(
+                os.path.join(destination_root, *member.relative_path.split("/"))
+            )
+            try:
+                inside_destination = (
+                    os.path.commonpath((destination_root, target))
+                    == destination_root
+                )
+            except ValueError as exc:
+                raise ZipExtractionError(
+                    f"ZIP contains a path traversal attempt: {member.info.filename}"
+                ) from exc
+            if not inside_destination:
+                raise ZipExtractionError(
+                    "ZIP contains a path traversal attempt: "
+                    f"{member.info.filename}"
+                )
+            if member.info.is_dir():
+                os.makedirs(target, exist_ok=True)
+                continue
+            if os.path.lexists(target):
+                raise ZipExtractionError(
+                    f"ZIP member would overwrite an existing path: {member.info.filename}"
+                )
+
+            os.makedirs(os.path.dirname(target), exist_ok=True)
+            written = 0
+            with archive.open(member.info, "r") as source, open(target, "xb") as output:
+                while True:
+                    chunk = source.read(_COPY_CHUNK_BYTES)
+                    if not chunk:
+                        break
+                    written += len(chunk)
+                    if written > member.info.file_size:
+                        raise ZipExtractionError(
+                            "ZIP member expanded past its declared size: "
+                            f"{member.info.filename}"
+                        )
+                    output.write(chunk)
+            if written != member.info.file_size:
+                raise ZipExtractionError(
+                    f"ZIP member size did not match metadata: {member.info.filename}"
+                )
+            extracted_files.append(target)
+    return extracted_files
+
+
+def _extract_tree_into(
+    zip_path: str | os.PathLike[str],
+    destination_root: str,
+    limits: ZipExtractionLimits,
+    budget: ZipExtractionBudget,
+    depth: int,
+) -> None:
+    """Extract an archive and recursively expand nested ``.zip`` members."""
+    extracted_files = _extract_archive_into(
+        zip_path, destination_root, limits, budget
+    )
+    for extracted_path in extracted_files:
+        if not extracted_path.lower().endswith(".zip"):
+            continue
+        if not zipfile.is_zipfile(extracted_path):
+            continue
+        if depth >= limits.max_depth:
+            raise ZipExtractionError(
+                f"ZIP nesting exceeds the {limits.max_depth}-level limit"
+            )
+        parent = os.path.dirname(extracted_path)
+        stem = os.path.splitext(os.path.basename(extracted_path))[0] or "archive"
+        occupied_names = {name.casefold() for name in os.listdir(parent)}
+        nested_name = stem
+        suffix = 2
+        while nested_name.casefold() in occupied_names:
+            nested_name = f"{stem} ({suffix})"
+            suffix += 1
+        nested_root = os.path.join(parent, nested_name)
+        os.makedirs(nested_root)
+        _extract_tree_into(
+            extracted_path,
+            nested_root,
+            limits,
+            budget,
+            depth + 1,
+        )
+        os.remove(extracted_path)
+
+
+def _commit_staging(staging_root: str, destination_root: str) -> None:
+    """Commit a complete staging tree, rolling back an interrupted merge."""
+    if not os.path.lexists(destination_root):
+        os.replace(staging_root, destination_root)
+        return
+    if not os.path.isdir(destination_root):
+        raise ZipExtractionError("ZIP extraction destination is not a directory")
+
+    staged_directories: list[tuple[str, str]] = []
+    staged_files: list[tuple[str, str]] = []
+    for root, directories, files in os.walk(staging_root):
+        relative_root = os.path.relpath(root, staging_root)
+        destination_directory = (
+            destination_root
+            if relative_root == "."
+            else os.path.join(destination_root, relative_root)
+        )
+        for directory in directories:
+            source = os.path.join(root, directory)
+            target = os.path.join(destination_directory, directory)
+            if os.path.lexists(target) and not os.path.isdir(target):
+                raise ZipExtractionError(
+                    "ZIP member would overwrite an existing path"
+                )
+            staged_directories.append((source, target))
+        for filename in files:
+            source = os.path.join(root, filename)
+            target = os.path.join(destination_directory, filename)
+            if os.path.lexists(target):
+                raise ZipExtractionError(
+                    "ZIP member would overwrite an existing path"
+                )
+            staged_files.append((source, target))
+
+    created_directories: list[str] = []
+    moved_files: list[str] = []
+    try:
+        for _, target in staged_directories:
+            if not os.path.isdir(target):
+                os.makedirs(target)
+                created_directories.append(target)
+        for source, target in staged_files:
+            os.makedirs(os.path.dirname(target), exist_ok=True)
+            os.replace(source, target)
+            moved_files.append(target)
+    except Exception:
+        for target in reversed(moved_files):
+            try:
+                os.remove(target)
+            except OSError:
+                pass
+        for target in reversed(created_directories):
+            try:
+                os.rmdir(target)
+            except OSError:
+                pass
+        raise
+
+
+def extract_zip_safely(
+    zip_path: str | os.PathLike[str],
+    extract_to: str | os.PathLike[str],
+    limits: ZipExtractionLimits,
+    budget: ZipExtractionBudget | None = None,
+) -> ZipExtractionBudget:
+    """Transactionally extract a ZIP tree under cumulative resource limits.
+
+    Nested ``.zip`` members are expanded inside a private staging directory.
+    The destination is changed only after every layer has passed validation
+    and extraction, so a corrupt member cannot leave partial content behind.
+    """
+    active_budget = budget or ZipExtractionBudget()
+    destination_root = os.path.realpath(extract_to)
+    destination_parent = os.path.dirname(destination_root)
+    staging_root: str | None = None
+
+    try:
+        os.makedirs(destination_parent, exist_ok=True)
+        staging_root = tempfile.mkdtemp(
+            prefix=".zip-extract-", dir=destination_parent
+        )
+        _extract_tree_into(zip_path, staging_root, limits, active_budget, depth=0)
+        _commit_staging(staging_root, destination_root)
+    except ZipExtractionError:
+        raise
+    except (
+        OSError,
+        RuntimeError,
+        ValueError,
+        zipfile.BadZipFile,
+        zipfile.LargeZipFile,
+    ) as exc:
+        raise ZipExtractionError(f"Unable to extract ZIP safely: {exc}") from exc
+    finally:
+        if staging_root is not None:
+            shutil.rmtree(staging_root, ignore_errors=True)
+
+    return active_budget
+
+
+def safe_zip_error_message(error: BaseException | str, max_chars: int = 512) -> str:
+    """Return a bounded, single-line representation safe for logs and JSON."""
+    raw = str(error)
+    escaped = "".join(
+        f"\\x{ord(character):02x}"
+        if ord(character) < 32 or 127 <= ord(character) <= 159
+        else character
+        for character in raw
+    )
+    if len(escaped) > max_chars:
+        return f"{escaped[: max_chars - 3]}..."
+    return escaped

application/upload_limits.py

@@ -0,0 +1,146 @@
+"""Bound user-controlled upload streams before storage or parsing."""
+
+from __future__ import annotations
+
+import io
+import os
+from typing import BinaryIO
+
+from application.core.settings import settings
+
+
+_COPY_CHUNK_BYTES = 64 * 1024
+_DOCUMENT_UPLOAD_PATHS = frozenset(
+    {
+        "/api/upload",
+        "/api/manage_source_files",
+        "/api/store_attachment",
+        "/api/parse_spec",
+        "/api/create_agent",
+    }
+)
+
+
+class UploadTooLargeError(ValueError):
+    """Raised when one uploaded file exceeds the configured byte cap."""
+
+
+class _LimitedRawReader(io.RawIOBase):
+    """Expose a binary stream while enforcing an exact cumulative read cap."""
+
+    def __init__(self, stream: BinaryIO, max_bytes: int) -> None:
+        super().__init__()
+        self._stream = stream
+        self._max_bytes = max_bytes
+        self._bytes_read = 0
+
+    def readable(self) -> bool:
+        """Return whether this wrapper supports reads."""
+        return True
+
+    def readinto(self, buffer: bytearray) -> int:
+        """Read one bounded chunk into ``buffer``."""
+        remaining_with_probe = self._max_bytes - self._bytes_read + 1
+        chunk = self._stream.read(min(len(buffer), remaining_with_probe))
+        if not chunk:
+            return 0
+        self._bytes_read += len(chunk)
+        if self._bytes_read > self._max_bytes:
+            raise UploadTooLargeError(upload_limit_message(self._max_bytes))
+        buffer[: len(chunk)] = chunk
+        return len(chunk)
+
+
+def is_document_upload_path(path: str) -> bool:
+    """Return whether a route accepts a user-controlled document/image file."""
+    return path in _DOCUMENT_UPLOAD_PATHS or path.startswith("/api/update_agent/")
+
+
+def upload_limit_message(max_bytes: int | None = None) -> str:
+    """Return the stable client-facing file-size rejection message."""
+    limit = int(max_bytes or settings.UPLOAD_MAX_FILE_BYTES)
+    return f"File exceeds the {limit}-byte upload limit"
+
+
+def upload_request_limit_message(max_bytes: int | None = None) -> str:
+    """Return the stable client-facing request-size rejection message."""
+    limit = int(max_bytes or settings.UPLOAD_MAX_REQUEST_BYTES)
+    return f"Request exceeds the {limit}-byte upload limit"
+
+
+def copy_upload_to_path(
+    upload: BinaryIO,
+    destination: str | os.PathLike[str],
+    max_bytes: int | None = None,
+) -> int:
+    """Copy an upload to disk while enforcing a hard streaming byte limit.
+
+    The limit is checked during reads, so it does not rely on a trustworthy
+    ``Content-Length`` header or on the input stream being seekable.
+    """
+    limit = int(max_bytes or settings.UPLOAD_MAX_FILE_BYTES)
+    stream = getattr(upload, "stream", upload)
+    try:
+        stream.seek(0)
+    except (AttributeError, OSError):
+        pass
+
+    total = 0
+    with open(destination, "wb") as target:
+        while True:
+            chunk = stream.read(min(_COPY_CHUNK_BYTES, limit - total + 1))
+            if not chunk:
+                break
+            total += len(chunk)
+            if total > limit:
+                raise UploadTooLargeError(upload_limit_message(limit))
+            target.write(chunk)
+    return total
+
+
+def read_upload_limited(upload: BinaryIO, max_bytes: int | None = None) -> bytes:
+    """Read at most one configured file into memory, rejecting overflow."""
+    limit = int(max_bytes or settings.UPLOAD_MAX_FILE_BYTES)
+    stream = getattr(upload, "stream", upload)
+    try:
+        stream.seek(0)
+    except (AttributeError, OSError):
+        pass
+    data = bytearray()
+    while True:
+        chunk = stream.read(min(_COPY_CHUNK_BYTES, limit - len(data) + 1))
+        if not chunk:
+            break
+        data.extend(chunk)
+        if len(data) > limit:
+            raise UploadTooLargeError(upload_limit_message(limit))
+    return bytes(data)
+
+
+def read_text_upload_limited(
+    upload: BinaryIO,
+    max_bytes: int | None = None,
+    encoding: str = "utf-8",
+) -> str:
+    """Decode a bounded upload without first duplicating it into ``bytes``."""
+    limit = int(max_bytes or settings.UPLOAD_MAX_FILE_BYTES)
+    stream = getattr(upload, "stream", upload)
+    try:
+        stream.seek(0)
+    except (AttributeError, OSError):
+        pass
+
+    raw_reader = _LimitedRawReader(stream, limit)
+    buffered_reader = io.BufferedReader(raw_reader, buffer_size=_COPY_CHUNK_BYTES)
+    text_reader = io.TextIOWrapper(buffered_reader, encoding=encoding)
+    try:
+        return text_reader.read()
+    finally:
+        try:
+            text_reader.detach()
+        except (ValueError, OSError):
+            pass
+        try:
+            buffered_reader.detach()
+        except (ValueError, OSError):
+            pass

application/worker.py

@@ -8,7 +8,6 @@ import string
 import tempfile
 import threading
 from typing import Any, Dict
-import zipfile
 
 import uuid
 from collections import Counter
@@ -31,6 +30,14 @@ from application.parser.remote.remote_creator import (
     normalize_remote_data,
 )
 from application.parser.schema.base import Document
+from application.security.zip_archive import (
+    extract_zip_safely,
+    safe_zip_error_message,
+    validate_zip_archive,
+    ZipExtractionBudget,
+    ZipExtractionError,
+    ZipExtractionLimits,
+)
 
 from application.storage.db.base_repository import looks_like_uuid
 from application.storage.db.repositories.agents import AgentsRepository
@@ -293,15 +300,11 @@ current_dir = os.path.dirname(
     os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 )
 
-# Zip extraction security limits
-MAX_UNCOMPRESSED_SIZE = 500 * 1024 * 1024  # 500 MB max uncompressed size
-MAX_FILE_COUNT = 10000  # Maximum number of files to extract
-MAX_COMPRESSION_RATIO = 100  # Maximum compression ratio (to detect zip bombs)
-
-
-class ZipExtractionError(Exception):
-    """Raised when zip extraction fails due to security constraints."""
-    pass
+# Zip extraction security limits. Kept as module constants for backward
+# compatibility with worker callers/tests; values come from operator settings.
+MAX_UNCOMPRESSED_SIZE = settings.UPLOAD_MAX_ARCHIVE_BYTES
+MAX_FILE_COUNT = settings.UPLOAD_MAX_ARCHIVE_FILES
+MAX_COMPRESSION_RATIO = settings.UPLOAD_MAX_ARCHIVE_RATIO
 
 
 def _is_path_safe(base_path: str, target_path: str) -> bool:
@@ -337,58 +340,26 @@ def _validate_zip_safety(zip_path: str, extract_to: str) -> None:
     Raises:
         ZipExtractionError: If the zip file fails security validation.
     """
-    try:
-        with zipfile.ZipFile(zip_path, "r") as zip_ref:
-            # Get compressed size
-            compressed_size = os.path.getsize(zip_path)
-
-            # Calculate total uncompressed size and file count
-            total_uncompressed = 0
-            file_count = 0
-
-            for info in zip_ref.infolist():
-                file_count += 1
-
-                # Check file count limit
-                if file_count > MAX_FILE_COUNT:
-                    raise ZipExtractionError(
-                        f"Zip file contains too many files (>{MAX_FILE_COUNT}). "
-                        "This may be a zip bomb attack."
-                    )
-
-                # Accumulate uncompressed size
-                total_uncompressed += info.file_size
-
-                # Check total uncompressed size
-                if total_uncompressed > MAX_UNCOMPRESSED_SIZE:
-                    raise ZipExtractionError(
-                        f"Zip file uncompressed size exceeds limit "
-                        f"({total_uncompressed / (1024*1024):.1f} MB > "
-                        f"{MAX_UNCOMPRESSED_SIZE / (1024*1024):.1f} MB). "
-                        "This may be a zip bomb attack."
-                    )
-
-                # Check for path traversal (zip slip)
-                target_path = os.path.join(extract_to, info.filename)
-                if not _is_path_safe(extract_to, target_path):
-                    raise ZipExtractionError(
-                        f"Zip file contains path traversal attempt: {info.filename}"
-                    )
-
-            # Check compression ratio (only if compressed size is meaningful)
-            if compressed_size > 0 and total_uncompressed > 0:
-                compression_ratio = total_uncompressed / compressed_size
-                if compression_ratio > MAX_COMPRESSION_RATIO:
-                    raise ZipExtractionError(
-                        f"Zip file has suspicious compression ratio ({compression_ratio:.1f}:1 > "
-                        f"{MAX_COMPRESSION_RATIO}:1). This may be a zip bomb attack."
-                    )
-
-    except zipfile.BadZipFile as e:
-        raise ZipExtractionError(f"Invalid or corrupted zip file: {e}")
+    del extract_to  # Path checks are platform-independent inside the validator.
+    validate_zip_archive(
+        zip_path,
+        ZipExtractionLimits(
+            max_uncompressed_bytes=MAX_UNCOMPRESSED_SIZE,
+            max_files=MAX_FILE_COUNT,
+            max_compression_ratio=MAX_COMPRESSION_RATIO,
+            max_member_bytes=settings.UPLOAD_MAX_FILE_BYTES,
+            max_depth=RECURSION_DEPTH,
+        ),
+    )
 
 
-def extract_zip_recursive(zip_path, extract_to, current_depth=0, max_depth=5):
+def extract_zip_recursive(
+    zip_path: str,
+    extract_to: str,
+    current_depth: int = 0,
+    max_depth: int = 5,
+    _budget: ZipExtractionBudget | None = None,
+) -> None:
     """
     Recursively extract zip files with security protections.
 
@@ -410,16 +381,27 @@ def extract_zip_recursive(zip_path, extract_to, current_depth=0, max_depth=5):
         return
 
     try:
-        # Validate zip file safety before extraction
-        _validate_zip_safety(zip_path, extract_to)
-
-        # Safe to extract
-        with zipfile.ZipFile(zip_path, "r") as zip_ref:
-            zip_ref.extractall(extract_to)
+        budget = _budget or ZipExtractionBudget()
+        extract_zip_safely(
+            zip_path,
+            extract_to,
+            ZipExtractionLimits(
+                max_uncompressed_bytes=MAX_UNCOMPRESSED_SIZE,
+                max_files=MAX_FILE_COUNT,
+                max_compression_ratio=MAX_COMPRESSION_RATIO,
+                max_member_bytes=settings.UPLOAD_MAX_FILE_BYTES,
+                max_depth=max_depth - current_depth,
+            ),
+            budget,
+        )
         os.remove(zip_path)  # Remove the zip file after extracting
 
     except ZipExtractionError as e:
-        logging.error(f"Zip security validation failed for {zip_path}: {e}")
+        logging.error(
+            "Zip security validation failed for %s: %s",
+            safe_zip_error_message(zip_path),
+            safe_zip_error_message(e),
+        )
         # Remove the potentially malicious zip file
         try:
             os.remove(zip_path)
@@ -427,17 +409,14 @@ def extract_zip_recursive(zip_path, extract_to, current_depth=0, max_depth=5):
             pass
         return
     except Exception as e:
-        logging.error(f"Error extracting zip file {zip_path}: {e}", exc_info=True)
+        logging.error(
+            "Error extracting zip file %s: %s",
+            safe_zip_error_message(zip_path),
+            safe_zip_error_message(e),
+            exc_info=True,
+        )
         return
 
-    # Check for nested zip files and extract them
-    for root, dirs, files in os.walk(extract_to):
-        for file in files:
-            if file.endswith(".zip"):
-                # If a nested zip file is found, extract it recursively
-                file_path = os.path.join(root, file)
-                extract_zip_recursive(file_path, root, current_depth + 1, max_depth)
-
 
 def download_file(url, params, dest_path):
     try:
@@ -623,7 +602,7 @@ def ingest_worker(
                     f.write(file_data.read())
 
                 # Handle zip files
-                if temp_filename.endswith(".zip"):
+                if temp_filename.lower().endswith(".zip"):
                     logging.info(f"Extracting zip file: {temp_filename}")
                     extract_zip_recursive(
                         temp_file_path,