# Tar Slip via tarfile.extractall() Path Traversal

Language: Python
Severity: Critical
CWE: CWE-22

## Source
8

## Flow
8-9-10-13-14-15

## Sink
15

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

app = Flask(__name__)

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

## Explanation

The code accepts a tar.gz archive from untrusted user input and extracts it using tarfile.extractall() without validating the paths within the archive. A malicious archive can contain entries with path traversal sequences (../) that escape the intended extraction directory and overwrite arbitrary files on the system.

## Remediation

The fix adds a safe_extract function that validates every tar archive member before extraction, ensuring no member's resolved path escapes the intended extraction directory. It checks both regular file paths and symbolic/hard link targets for path traversal attempts, and also sanitizes the deployment_id input to prevent directory traversal via the folder name itself.

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

app = Flask(__name__)


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


def safe_extract(tar_ref, extraction_path):
    """Extract tar members only if they are safe (no path traversal)."""
    for member in tar_ref.getmembers():
        if not is_safe_tar_member(member, extraction_path):
            raise ValueError(f"Unsafe path detected in archive: {member.name}")
    tar_ref.extractall(path=extraction_path)


@app.route('/api/v1/deploy-model', methods=['POST'])
def deploy_ml_model():
    model_archive = request.files.get('model_package')
    if not model_archive:
        return jsonify({'status': 'error', 'message': 'No model package provided'}), 400

    deployment_id = request.form.get('deployment_id', 'default')

    if not deployment_id.isalnum() and not all(c in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-' for c in deployment_id):
        return jsonify({'status': 'error', 'message': 'Invalid deployment_id'}), 400

    extraction_path = f'/var/ml_models/{deployment_id}'
    os.makedirs(extraction_path, exist_ok=True)
    archive_path = f'/tmp/model_{deployment_id}.tar.gz'
    model_archive.save(archive_path)

    try:
        with tarfile.open(archive_path, 'r:gz') as tar_ref:
            safe_extract(tar_ref, extraction_path)
    except (ValueError, tarfile.TarError) as e:
        os.remove(archive_path)
        return jsonify({'status': 'error', 'message': str(e)}), 400

    os.remove(archive_path)
    return jsonify({'status': 'deployed', 'model_path': extraction_path, 'deployment_id': deployment_id})
```