# Zip Bomb via Recursive Archive Expansion

Language: Python
Severity: High
CWE: CWE-409

## Source
10

## Flow
10-11-12-13

## Sink
13

## Vulnerable Code
```python
import zipfile
import os
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/cloud/extract-backup', methods=['POST'])
def restore_cloud_backup():
    backup_file = request.files['backup']
    restore_path = f"/tmp/restore_{request.form.get('tenant_id')}"
    os.makedirs(restore_path, exist_ok=True)
    backup_file.save(f"{restore_path}/backup.zip")
    with zipfile.ZipFile(f"{restore_path}/backup.zip", 'r') as archive:
        archive.extractall(restore_path)
    return jsonify({"status": "restored", "path": restore_path})
```

## Explanation

The application accepts an uploaded ZIP file from an untrusted source without validating its compressed/uncompressed size ratio or recursion depth. The extractall() method decompresses the entire archive without limits, allowing a zip bomb (highly compressed malicious archive) to exhaust disk space and system resources.

## Remediation

The fix adds comprehensive validation before extraction: it checks the total uncompressed size, individual file sizes, compression ratio (to detect zip bombs), file count limits, and rejects nested archives. Additionally, it includes path traversal protection during extraction and sanitizes the tenant_id input to prevent directory traversal attacks.

## Secure Code
```python
import zipfile
import os
from flask import Flask, request, jsonify

app = Flask(__name__)

MAX_TOTAL_EXTRACTED_SIZE = 500 * 1024 * 1024  # 500 MB max total extracted size
MAX_SINGLE_FILE_SIZE = 100 * 1024 * 1024  # 100 MB max per file
MAX_FILE_COUNT = 10000  # Maximum number of files in archive
MAX_COMPRESSION_RATIO = 100  # Maximum allowed compression ratio
MAX_NESTING_DEPTH = 0  # Do not allow nested archives

NESTED_ARCHIVE_EXTENSIONS = ('.zip', '.tar', '.gz', '.bz2', '.xz', '.7z', '.rar')


def is_nested_archive(filename):
    """Check if a file appears to be a nested archive."""
    return filename.lower().endswith(NESTED_ARCHIVE_EXTENSIONS)


def validate_zip_archive(zip_path):
    """Validate a ZIP archive for zip bomb characteristics before extraction."""
    compressed_size = os.path.getsize(zip_path)

    with zipfile.ZipFile(zip_path, 'r') as archive:
        members = archive.infolist()

        # Check file count
        if len(members) > MAX_FILE_COUNT:
            raise ValueError(f"Archive contains too many files: {len(members)} (max: {MAX_FILE_COUNT})")

        total_uncompressed = 0
        for member in members:
            # Check for nested archives
            if is_nested_archive(member.filename):
                raise ValueError(f"Nested archive detected: {member.filename}. Nested archives are not allowed.")

            # Check individual file size
            if member.file_size > MAX_SINGLE_FILE_SIZE:
                raise ValueError(f"File '{member.filename}' exceeds max size: {member.file_size} bytes (max: {MAX_SINGLE_FILE_SIZE})")

            total_uncompressed += member.file_size

        # Check total uncompressed size
        if total_uncompressed > MAX_TOTAL_EXTRACTED_SIZE:
            raise ValueError(f"Total extracted size too large: {total_uncompressed} bytes (max: {MAX_TOTAL_EXTRACTED_SIZE})")

        # Check compression ratio
        if compressed_size > 0:
            ratio = total_uncompressed / compressed_size
            if ratio > MAX_COMPRESSION_RATIO:
                raise ValueError(f"Suspicious compression ratio: {ratio:.1f}x (max: {MAX_COMPRESSION_RATIO}x)")

    return True


def safe_extract(zip_path, extract_path):
    """Safely extract a ZIP archive with path traversal protection."""
    with zipfile.ZipFile(zip_path, 'r') as archive:
        for member in archive.infolist():
            # Prevent path traversal (zip slip)
            member_path = os.path.realpath(os.path.join(extract_path, member.filename))
            if not member_path.startswith(os.path.realpath(extract_path) + os.sep) and member_path != os.path.realpath(extract_path):
                raise ValueError(f"Path traversal detected in archive member: {member.filename}")

            archive.extract(member, extract_path)


@app.route('/api/cloud/extract-backup', methods=['POST'])
def restore_cloud_backup():
    if 'backup' not in request.files:
        return jsonify({"error": "No backup file provided"}), 400

    tenant_id = request.form.get('tenant_id')
    if not tenant_id or not tenant_id.isalnum():
        return jsonify({"error": "Invalid tenant_id"}), 400

    backup_file = request.files['backup']
    restore_path = f"/tmp/restore_{tenant_id}"
    os.makedirs(restore_path, exist_ok=True)
    zip_file_path = f"{restore_path}/backup.zip"
    backup_file.save(zip_file_path)

    try:
        # Validate the archive is a proper ZIP file
        if not zipfile.is_zipfile(zip_file_path):
            os.remove(zip_file_path)
            return jsonify({"error": "Invalid ZIP file"}), 400

        # Validate archive for zip bomb characteristics
        validate_zip_archive(zip_file_path)

        # Safely extract with path traversal protection
        safe_extract(zip_file_path, restore_path)

    except ValueError as e:
        # Clean up on validation failure
        os.remove(zip_file_path)
        return jsonify({"error": str(e)}), 400
    except zipfile.BadZipFile:
        os.remove(zip_file_path)
        return jsonify({"error": "Corrupted ZIP file"}), 400
    finally:
        # Remove the uploaded zip after extraction attempt
        if os.path.exists(zip_file_path):
            os.remove(zip_file_path)

    return jsonify({"status": "restored", "path": restore_path})
```