# Tar Slip via tarfile.extractall() Path Traversal

Language: Python
Severity: Critical
CWE: CWE-22

## Source
6-7

## Flow
6-7-8-11-12

## Sink
12

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

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

## Explanation

The application accepts an untrusted device_id parameter and uses it to construct the extraction path without validation. A malicious tar archive can contain files with path traversal sequences (../) in their names, allowing tarfile.extractall() to write files outside the intended directory, potentially overwriting critical system files or injecting malicious code.

## Remediation

The fix addresses both vulnerability points: (1) the device_id parameter is validated with a strict regex whitelist to prevent path traversal in the constructed directory path, and (2) a safe extraction function validates every tar member's resolved path to ensure it stays within the intended deployment directory before extraction, including checks for symlink-based traversal attacks.

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

def _is_safe_path(base_path, target_path):
    """Ensure the target path is within the base directory."""
    abs_base = os.path.realpath(base_path)
    abs_target = os.path.realpath(target_path)
    return abs_target.startswith(abs_base + os.sep) or abs_target == abs_base

def _safe_extract(tar, destination):
    """Extract tar members only if they resolve within the destination directory."""
    for member in tar.getmembers():
        member_path = os.path.join(destination, member.name)
        abs_member_path = os.path.realpath(member_path)
        if not abs_member_path.startswith(os.path.realpath(destination) + os.sep):
            raise ValueError(f"Attempted path traversal in tar member: {member.name}")
        if member.issym() or member.islnk():
            link_target = os.path.join(destination, os.path.dirname(member.name), member.linkname)
            abs_link_target = os.path.realpath(link_target)
            if not abs_link_target.startswith(os.path.realpath(destination) + os.sep):
                raise ValueError(f"Attempted path traversal via symlink in tar member: {member.name}")
    tar.extractall(path=destination)

@app.route('/iot/firmware/deploy', methods=['POST'])
def deploy_firmware_package():
    device_id = request.form.get('device_id')
    if not device_id or not re.match(r'^[a-zA-Z0-9_\-]+$', device_id):
        return jsonify({'status': 'error', 'message': 'Invalid device_id'}), 400
    firmware_archive = request.files['firmware']
    deployment_path = f'/opt/iot/devices/{device_id}/firmware'
    base_dir = '/opt/iot/devices'
    if not _is_safe_path(base_dir, deployment_path):
        return jsonify({'status': 'error', 'message': 'Invalid deployment path'}), 400
    os.makedirs(deployment_path, exist_ok=True)
    archive_path = os.path.join(deployment_path, 'package.tar.gz')
    firmware_archive.save(archive_path)
    try:
        with tarfile.open(archive_path, 'r:gz') as tar_pkg:
            _safe_extract(tar_pkg, deployment_path)
    except (ValueError, tarfile.TarError) as e:
        os.remove(archive_path)
        return jsonify({'status': 'error', 'message': f'Invalid firmware package: {str(e)}'}), 400
    return jsonify({'status': 'deployed', 'device': device_id, 'location': deployment_path})
```