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

Language: Python
Severity: Critical
CWE: CWE-22

## Source
7

## Flow
7-10-11

## Sink
11

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

@app.route('/iot/firmware/deploy', methods=['POST'])
def deploy_firmware_package():
    firmware_archive = request.files['firmware_pkg']
    device_id = request.form.get('device_id')
    deployment_path = f'/opt/iot_devices/{device_id}/firmware'
    os.makedirs(deployment_path, exist_ok=True)
    temp_tar = f'/tmp/{device_id}_firmware.tar.gz'
    firmware_archive.save(temp_tar)
    with tarfile.open(temp_tar, 'r:gz') as tar:
        tar.extractall(path=deployment_path)
    return jsonify({'status': 'deployed', 'device': device_id, 'path': deployment_path})
```

## Explanation

The application accepts an untrusted tar.gz archive uploaded by a user and extracts it using tarfile.extractall() without validating the file paths contained within the archive. A malicious tar archive can contain path traversal sequences (e.g., ../../etc/cron.d/malicious) that allow files to be written outside the intended deployment_path directory, enabling arbitrary file write vulnerabilities.

## Remediation

The fix adds a safe extraction function that validates every member's resolved path is within the intended deployment directory before extraction, blocking path traversal attacks via relative paths (../) or absolute paths. It also validates symlinks to prevent indirect traversal, sanitizes the device_id input to prevent injection into the base path, cleans up the temporary file after processing, and returns a 400 status on malicious archive detection.

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

def _is_within_directory(directory, target):
    """Check if the target path is within the intended directory."""
    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, path):
    """Safely extract tar members after validating all paths are within the target directory."""
    for member in tar.getmembers():
        member_path = os.path.join(path, member.name)
        if not _is_within_directory(path, member_path):
            raise Exception(f"Attempted path traversal in tar file: {member.name}")
        if member.issym() or member.islnk():
            link_target = os.path.join(path, os.path.dirname(member.name), member.linkname)
            if not _is_within_directory(path, link_target):
                raise Exception(f"Attempted symlink path traversal in tar file: {member.name} -> {member.linkname}")
    tar.extractall(path=path)

@app.route('/iot/firmware/deploy', methods=['POST'])
def deploy_firmware_package():
    firmware_archive = request.files['firmware_pkg']
    device_id = request.form.get('device_id')
    if not device_id or '..' in device_id or '/' in device_id or '\\' in device_id:
        return jsonify({'status': 'error', 'message': 'Invalid device_id'}), 400
    deployment_path = f'/opt/iot_devices/{device_id}/firmware'
    os.makedirs(deployment_path, exist_ok=True)
    temp_tar = f'/tmp/{device_id}_firmware.tar.gz'
    firmware_archive.save(temp_tar)
    try:
        with tarfile.open(temp_tar, 'r:gz') as tar:
            _safe_extract(tar, deployment_path)
    except Exception as e:
        os.remove(temp_tar)
        return jsonify({'status': 'error', 'message': str(e)}), 400
    finally:
        if os.path.exists(temp_tar):
            os.remove(temp_tar)
    return jsonify({'status': 'deployed', 'device': device_id, 'path': deployment_path})
```