# Tar Path Traversal via tarfile.extractall()

Language: Python
Severity: Critical
CWE: CWE-22

## Source
6

## Flow
6-7-10-12

## Sink
12

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

@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')
    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', 'path': extract_path})
```

## Explanation

The code extracts a user-uploaded tar archive using tarfile.extractall() without validating the paths of archive members. A malicious tar file can contain entries with absolute paths or path traversal sequences (e.g., '../../etc/crontab') that write files outside the intended extraction directory, leading to arbitrary file write vulnerabilities.

## Remediation

The fix validates every tar archive member's resolved path to ensure it stays within the intended extraction directory before calling extractall(). It also checks symlink targets to prevent indirect path traversal, sanitizes the deployment_id parameter against directory traversal, uses os.path.basename on the uploaded filename, and cleans up temporary files after extraction.

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

def _is_within_directory(directory, target):
    abs_directory = os.path.realpath(directory)
    abs_target = os.path.realpath(target)
    return abs_target.startswith(abs_directory + os.sep) or abs_target == abs_directory

def _safe_extract(tar, extract_path):
    for member in tar.getmembers():
        member_path = os.path.join(extract_path, member.name)
        if not _is_within_directory(extract_path, member_path):
            raise ValueError(f"Attempted path traversal in tar member: {member.name}")
        if member.issym() or member.islnk():
            link_target = os.path.join(extract_path, os.path.normpath(member.linkname)) if not os.path.isabs(member.linkname) else member.linkname
            if not _is_within_directory(extract_path, link_target):
                raise ValueError(f"Attempted path traversal via symlink in tar member: {member.name}")
    tar.extractall(path=extract_path)

@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')

    if not deployment_id or '..' in deployment_id or '/' in deployment_id or '\\' in deployment_id:
        return jsonify({'error': 'Invalid deployment_id'}), 400

    extract_path = f'/opt/ml_models/{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:
            _safe_extract(tar, extract_path)
    except (ValueError, tarfile.TarError) as e:
        os.remove(archive_path)
        return jsonify({'error': str(e)}), 400
    finally:
        if os.path.exists(archive_path):
            os.remove(archive_path)

    return jsonify({'status': 'deployed', 'path': extract_path})
```