# TOCTOU Race Condition in File Authorization Check

Language: Python
Severity: High
CWE: CWE-367

## Source
7

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

## Sink
12, 13-14

## Vulnerable Code
```python
import os
import time
import boto3

def download_s3_object_to_local(bucket_name, object_key, local_path, user_role):
    s3_client = boto3.client('s3')
    metadata = s3_client.head_object(Bucket=bucket_name, Key=object_key)
    required_role = metadata.get('Metadata', {}).get('required-role', 'user')
    if user_role != required_role:
        raise PermissionError(f"Access denied: requires {required_role} role")
    time.sleep(0.1)
    s3_client.download_file(bucket_name, object_key, local_path)
    with open(local_path, 'r') as f:
        return f.read()
```

## Explanation

The code checks S3 object metadata for authorization (lines 7-8) but then introduces a delay (line 11) before downloading the file (line 12) and reading it (lines 13-14). An attacker can exploit this TOCTOU (Time-Of-Check-Time-Of-Use) window by modifying the S3 object or its metadata, or by replacing the local file after authorization but before the actual download/read operation, bypassing access controls.

## Remediation

The fix eliminates the TOCTOU race condition by using s3_client.get_object() to retrieve both metadata and content in a single atomic API call, ensuring the authorization check and data retrieval happen on the exact same version of the object. The artificial delay is removed, the content is read directly from the authenticated response stream, and the local file is written atomically using a temporary file with os.replace() to prevent local file swap attacks. The already-fetched content is returned from memory rather than re-reading from disk.

## Secure Code
```python
import os
import boto3
import tempfile
import hashlib

def download_s3_object_to_local(bucket_name, object_key, local_path, user_role):
    s3_client = boto3.client('s3')
    
    # Download to a secure temporary file first, capturing metadata atomically
    # Use get_object to retrieve metadata and content in a single atomic operation
    response = s3_client.get_object(Bucket=bucket_name, Key=object_key)
    
    # Check authorization from the same response (atomic check-and-use)
    required_role = response.get('Metadata', {}).get('required-role', 'user')
    if user_role != required_role:
        # Discard the response body without saving
        response['Body'].close()
        raise PermissionError(f"Access denied: requires {required_role} role")
    
    # Read content directly from the authenticated response stream
    content = response['Body'].read()
    response['Body'].close()
    
    # Write to a secure temporary file first, then atomically move to target path
    dir_name = os.path.dirname(os.path.abspath(local_path))
    fd, tmp_path = tempfile.mkstemp(dir=dir_name)
    try:
        os.write(fd, content)
        os.fsync(fd)
        os.close(fd)
        fd = None
        # Atomically rename to prevent local file replacement attacks
        os.replace(tmp_path, local_path)
    except Exception:
        if fd is not None:
            os.close(fd)
        if os.path.exists(tmp_path):
            os.unlink(tmp_path)
        raise
    
    # Return the content we already have in memory (no second file read needed)
    return content.decode('utf-8')
```