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

Language: Python
Severity: Critical
CWE: CWE-22

## Source
8

## Flow
8-10-11-12

## Sink
12

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

@app.route('/api/iot/firmware/deploy', methods=['POST'])
def deploy_firmware_package():
    device_id = request.form.get('device_id')
    firmware_tar = request.files['firmware_package']
    deployment_path = f'/opt/iot/devices/{device_id}/firmware'
    os.makedirs(deployment_path, exist_ok=True)
    tar_location = f'/tmp/{firmware_tar.filename}'
    firmware_tar.save(tar_location)
    with tarfile.open(tar_location, 'r:gz') as tar_archive:
        tar_archive.extractall(path=deployment_path)
    return jsonify({'status': 'deployed', 'device': device_id, 'path': deployment_path})
```

## Explanation

The vulnerability occurs when an untrusted tar archive uploaded via the Flask endpoint is extracted without validating member paths. A malicious archive can contain entries with path traversal sequences (e.g., '../../../etc/cron.d/malicious') that escape the intended deployment directory during extractall(), allowing arbitrary file writes on the system.

## Remediation

The fix validates every tar archive member's resolved path to ensure it remains within the intended deployment directory before extraction. It checks both regular file paths and symbolic/hard link targets for path traversal attempts. Additionally, the device_id is validated to be alphanumeric to prevent path injection, and the temporary filename is sanitized using os.path.basename().

## 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_archive, path):
    for member in tar_archive.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 member: {member.name}")
        if member.issym() or member.islnk():
            link_target = os.path.join(path, os.path.normpath(member.linkname)) if not os.path.isabs(member.linkname) else member.linkname
            if not _is_within_directory(path, link_target):
                raise Exception(f"Attempted path traversal via symlink in tar member: {member.name}")
    tar_archive.extractall(path=path)

@app.route('/api/iot/firmware/deploy', methods=['POST'])
def deploy_firmware_package():
    device_id = request.form.get('device_id')
    if not device_id or not device_id.isalnum():
        return jsonify({'status': 'error', 'message': 'Invalid device_id'}), 400
    firmware_tar = request.files['firmware_package']
    deployment_path = f'/opt/iot/devices/{device_id}/firmware'
    os.makedirs(deployment_path, exist_ok=True)
    safe_filename = os.path.basename(firmware_tar.filename)
    tar_location = f'/tmp/{safe_filename}'
    firmware_tar.save(tar_location)
    try:
        with tarfile.open(tar_location, 'r:gz') as tar_archive:
            _safe_extract(tar_archive, deployment_path)
    except Exception as e:
        os.remove(tar_location)
        return jsonify({'status': 'error', 'message': str(e)}), 400
    finally:
        if os.path.exists(tar_location):
            os.remove(tar_location)
    return jsonify({'status': 'deployed', 'device': device_id, 'path': deployment_path})
```