# Zip Slip Path Traversal via zipfile.extractall

Language: Python
Severity: Critical
CWE: CWE-22

## Source
7

## Flow
7-8-9-10-11-12

## Sink
12

## Vulnerable Code
```python
import zipfile
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_zip = request.files['firmware']
    deployment_path = f'/var/iot/devices/{device_id}/firmware'
    os.makedirs(deployment_path, exist_ok=True)
    zip_path = os.path.join(deployment_path, 'package.zip')
    firmware_zip.save(zip_path)
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        zip_ref.extractall(deployment_path)
    return jsonify({'status': 'deployed', 'device': device_id, 'path': deployment_path})
```

## Explanation

The code uses zipfile.extractall() without validating the paths of archive members, allowing Zip Slip attacks. A malicious ZIP file can contain entries with path traversal sequences (e.g., '../../../../etc/cron.d/malicious') that escape the intended deployment_path directory and write files to arbitrary locations on the filesystem.

## Remediation

The fix validates each ZIP archive member's resolved path to ensure it stays within the intended deployment directory before extraction. It uses os.path.realpath to resolve any path traversal sequences (like '../') and checks that the resulting absolute path starts with the target directory prefix. Additionally, the device_id input is validated to prevent directory traversal through the device identifier itself.

## Secure Code
```python
import zipfile
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')
    if not device_id or '..' in device_id or '/' in device_id or '\\' in device_id:
        return jsonify({'error': 'Invalid device_id'}), 400
    firmware_zip = request.files['firmware']
    deployment_path = f'/var/iot/devices/{device_id}/firmware'
    os.makedirs(deployment_path, exist_ok=True)
    zip_path = os.path.join(deployment_path, 'package.zip')
    firmware_zip.save(zip_path)
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        for member in zip_ref.namelist():
            member_path = os.path.realpath(os.path.join(deployment_path, member))
            if not member_path.startswith(os.path.realpath(deployment_path) + os.sep) and member_path != os.path.realpath(deployment_path):
                return jsonify({'error': f'Illegal path in archive: {member}'}), 400
        zip_ref.extractall(deployment_path)
    return jsonify({'status': 'deployed', 'device': device_id, 'path': deployment_path})
```