# Zip Slip via tarfile.extractall() on Untrusted Archives

Language: Python
Severity: Critical
CWE: CWE-22

## Source
10

## Flow
10-13-14-16

## Sink
16

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

app = Flask(__name__)

@app.route('/api/cloud/deploy-model', methods=['POST'])
def deploy_ml_model():
    model_archive = request.files['model_package']
    deployment_id = request.form.get('deployment_id', 'default')
    extract_path = f'/opt/ml_models/{deployment_id}'
    os.makedirs(extract_path, exist_ok=True)
    archive_path = f'/tmp/{model_archive.filename}'
    model_archive.save(archive_path)
    with tarfile.open(archive_path, 'r:gz') as tar:
        tar.extractall(path=extract_path)
    return jsonify({'status': 'deployed', 'model_path': extract_path})
```

## Explanation

The application accepts an untrusted tar.gz archive via file upload and extracts it using tarfile.extractall() without validating archive member paths. A malicious archive can contain entries with path traversal sequences (e.g., '../../etc/cron.d/malicious') that escape the intended extraction directory, allowing arbitrary file writes anywhere on the filesystem where the application has permissions.

## Remediation

The fix validates every tar archive member before extraction by resolving its real path and ensuring it stays within the intended extraction directory. It also checks symbolic and hard links to prevent link-based escapes, and sanitizes the deployment_id and filename inputs to prevent path traversal through those parameters as well.

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

app = Flask(__name__)

def is_safe_tar_member(member, extract_path):
    """Validate that a tar member extracts within the intended directory."""
    member_path = os.path.realpath(os.path.join(extract_path, member.name))
    extract_path_real = os.path.realpath(extract_path)
    if not member_path.startswith(extract_path_real + os.sep) and member_path != extract_path_real:
        return False
    if member.issym() or member.islnk():
        link_target = os.path.realpath(os.path.join(extract_path, member.linkname))
        if not link_target.startswith(extract_path_real + os.sep):
            return False
    return True

@app.route('/api/cloud/deploy-model', methods=['POST'])
def deploy_ml_model():
    model_archive = request.files['model_package']
    deployment_id = request.form.get('deployment_id', 'default')
    safe_deployment_id = os.path.basename(deployment_id)
    if not safe_deployment_id or safe_deployment_id in ('.', '..'):
        return jsonify({'error': 'Invalid deployment_id'}), 400
    extract_path = f'/opt/ml_models/{safe_deployment_id}'
    os.makedirs(extract_path, exist_ok=True)
    archive_path = f'/tmp/{os.path.basename(model_archive.filename)}'
    model_archive.save(archive_path)
    try:
        with tarfile.open(archive_path, 'r:gz') as tar:
            for member in tar.getmembers():
                if not is_safe_tar_member(member, extract_path):
                    os.remove(archive_path)
                    return jsonify({'error': f'Unsafe path detected in archive: {member.name}'}), 400
            tar.extractall(path=extract_path)
    finally:
        if os.path.exists(archive_path):
            os.remove(archive_path)
    return jsonify({'status': 'deployed', 'model_path': extract_path})
```