# Insecure Multipart Form File Upload Leading to Polyglot File Execution via MIME Sniffing Bypass

Language: Python
Severity: Critical
CWE: CWE-434

## Source
9

## Flow
9-11

## Sink
11

## Vulnerable Code
```python
from flask import Flask, request, send_from_directory
import os
app = Flask(__name__)
UPLOAD_ZONE = '/var/www/drone_telemetry/uploads'
def store_telemetry_artifact(req_file):
    declared_mime = req_file.content_type
    raw_hdr = req_file.stream.read(16)
    req_file.stream.seek(0)
    artifact_label = req_file.filename
    dest_path = os.path.join(UPLOAD_ZONE, artifact_label)
    req_file.save(dest_path)
    return send_from_directory(UPLOAD_ZONE, artifact_label, mimetype=declared_mime)
@app.route('/telemetry/upload', methods=['POST'])
def ingest_payload():
    pkg = request.files.get('artifact')
    return store_telemetry_artifact(pkg)
```

## Explanation

The code accepts an attacker-controlled filename (line 8, req_file.filename) and uses it unsanitized in os.path.join (line 10), allowing path traversal (e.g., ../../shell.php). The file is then saved (line 11) to the web-accessible upload directory without any extension or content validation. Additionally, the client-supplied Content-Type header is trusted directly as declared_mime and passed into send_from_directory, while the raw header bytes (raw_hdr) are read but never actually validated or enforced — serving as a false-security decoy. This allows an attacker to upload a polyglot file (e.g., JPEG magic bytes + PHP shell) named shell.php that gets stored and later executed by the server or MIME-sniffed by browsers.

## Remediation

The fix addresses all identified vulnerabilities: (1) the filename is sanitized with secure_filename() and a UUID-based rename eliminates path traversal and attacker-controlled filenames; (2) a strict extension allowlist replaces the unrestricted upload, rejecting executable file types like .php; (3) magic byte validation ensures file content matches its declared type, defeating polyglot attacks; (4) the server determines the MIME type from the allowlist instead of trusting the client-supplied Content-Type; (5) X-Content-Type-Options: nosniff and Content-Disposition: attachment headers prevent browser MIME sniffing and force download; (6) os.path.realpath canonicalization confirms the final path stays within UPLOAD_ZONE.

## Secure Code
```python
from flask import Flask, request, send_from_directory, abort
import os
import uuid
from werkzeug.utils import secure_filename

app = Flask(__name__)
UPLOAD_ZONE = '/var/www/drone_telemetry/uploads'

# Allowed extensions and their corresponding safe MIME types
ALLOWED_EXTENSIONS = {
    '.log': 'text/plain',
    '.txt': 'text/plain',
    '.csv': 'text/csv',
    '.json': 'application/json',
    '.bin': 'application/octet-stream',
    '.hex': 'application/octet-stream',
}

# Magic byte signatures for additional validation
MAGIC_BYTES = {
    '.json': [b'{', b'['],
    '.csv': None,
    '.log': None,
    '.txt': None,
    '.bin': None,
    '.hex': None,
}


def _validate_extension(filename):
    """Return the lowercased extension if it is in the allowlist, else None."""
    _, ext = os.path.splitext(filename)
    ext = ext.lower()
    if ext in ALLOWED_EXTENSIONS:
        return ext
    return None


def _validate_magic_bytes(stream, ext):
    """Check file magic bytes against known signatures for the extension."""
    expected = MAGIC_BYTES.get(ext)
    if expected is None:
        return True
    header = stream.read(16)
    stream.seek(0)
    return any(header.lstrip().startswith(sig) for sig in expected)


def store_telemetry_artifact(req_file):
    # 1. Sanitize the filename to prevent path traversal
    safe_name = secure_filename(req_file.filename)
    if not safe_name:
        abort(400, description='Invalid filename.')

    # 2. Validate extension against allowlist
    ext = _validate_extension(safe_name)
    if ext is None:
        abort(400, description=f'File type not allowed. Accepted: {list(ALLOWED_EXTENSIONS.keys())}')

    # 3. Validate magic bytes match the declared extension
    if not _validate_magic_bytes(req_file.stream, ext):
        abort(400, description='File content does not match its declared type.')

    # 4. Generate a unique filename to avoid overwrites and name-based attacks
    unique_name = f"{uuid.uuid4().hex}{ext}"
    dest_path = os.path.join(UPLOAD_ZONE, unique_name)

    # 5. Resolve the final path and ensure it is still within UPLOAD_ZONE
    real_dest = os.path.realpath(dest_path)
    real_upload_zone = os.path.realpath(UPLOAD_ZONE)
    if not real_dest.startswith(real_upload_zone + os.sep):
        abort(400, description='Invalid file path.')

    # 6. Save the file
    req_file.save(real_dest)

    # 7. Serve back with a safe, server-determined MIME type and
    #    X-Content-Type-Options: nosniff to prevent browser MIME sniffing
    safe_mime = ALLOWED_EXTENSIONS[ext]
    response = send_from_directory(real_upload_zone, unique_name, mimetype=safe_mime)
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['Content-Disposition'] = f'attachment; filename="{safe_name}"'
    return response


@app.route('/telemetry/upload', methods=['POST'])
def ingest_payload():
    pkg = request.files.get('artifact')
    if pkg is None:
        abort(400, description='No artifact file provided.')
    return store_telemetry_artifact(pkg)
```