# Zip Slip via unsafe tarfile.extractall() extraction

Language: Python
Severity: Critical
CWE: CWE-22

## Source
9

## Flow
9-11-12-13-14

## Sink
14

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

app = Flask(__name__)

@app.route('/api/cloud/restore-backup', methods=['POST'])
def restore_cloud_backup():
    backup_file = request.files['backup']
    tenant_id = request.form.get('tenant_id')
    restore_path = f'/var/cloud/tenants/{tenant_id}/data'
    os.makedirs(restore_path, exist_ok=True)
    backup_location = f'/tmp/{backup_file.filename}'
    backup_file.save(backup_location)
    with tarfile.open(backup_location, 'r:gz') as tar_archive:
        tar_archive.extractall(path=restore_path)
    return jsonify({'status': 'restored', 'tenant': tenant_id, 'files': len(os.listdir(restore_path))})
```

## Explanation

The code uses tarfile.extractall() without validating archive member paths, allowing Zip Slip attacks. A malicious tar.gz archive containing paths like '../../../etc/cron.d/malicious' can traverse outside the intended restore_path directory and overwrite arbitrary files on the server.

## Remediation

The fix validates every tar archive member's resolved path before extraction, ensuring no file can be written outside the intended restore directory. It also sanitizes the tenant_id input to prevent path traversal through the directory construction, sanitizes the uploaded filename, and cleans up temporary files after processing.

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

app = Flask(__name__)


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


def safe_extract(tar_archive, extract_path):
    """Extract tar members only if all paths are safe."""
    for member in tar_archive.getmembers():
        if not is_safe_tar_member(member, extract_path):
            raise ValueError(f"Unsafe path detected in archive: {member.name}")
    tar_archive.extractall(path=extract_path)


@app.route('/api/cloud/restore-backup', methods=['POST'])
def restore_cloud_backup():
    backup_file = request.files['backup']
    tenant_id = request.form.get('tenant_id')
    # Sanitize tenant_id to prevent path traversal via tenant_id itself
    if not tenant_id or '..' in tenant_id or '/' in tenant_id or '\\' in tenant_id:
        return jsonify({'error': 'Invalid tenant_id'}), 400
    restore_path = f'/var/cloud/tenants/{tenant_id}/data'
    os.makedirs(restore_path, exist_ok=True)
    # Sanitize the filename to prevent path traversal in temp storage
    safe_filename = os.path.basename(backup_file.filename)
    backup_location = f'/tmp/{safe_filename}'
    backup_file.save(backup_location)
    try:
        with tarfile.open(backup_location, 'r:gz') as tar_archive:
            safe_extract(tar_archive, restore_path)
    except (ValueError, tarfile.TarError) as e:
        os.remove(backup_location)
        return jsonify({'error': f'Invalid backup archive: {str(e)}'}), 400
    finally:
        if os.path.exists(backup_location):
            os.remove(backup_location)
    return jsonify({'status': 'restored', 'tenant': tenant_id, 'files': len(os.listdir(restore_path))})
```