# TOCTOU Race Condition in File Creation via os.path.exists() Check-Then-Use

Language: Python
Severity: High
CWE: CWE-367

## Source
6

## Flow
6-7-8

## Sink
8

## Vulnerable Code
```python
import os
import json

def provision_iot_device_credentials(device_id, aws_region):
    cert_path = f"/var/iot/certs/{device_id}_{aws_region}.pem"
    if not os.path.exists(cert_path):
        device_cert = generate_aws_iot_certificate(device_id)
        with open(cert_path, 'w') as cert_file:
            cert_file.write(device_cert)
        os.chmod(cert_path, 0o600)
        return {"status": "provisioned", "cert_location": cert_path}
    return {"status": "exists", "cert_location": cert_path}

def generate_aws_iot_certificate(device_id):
    return f"-----BEGIN CERTIFICATE-----\nMIIC...{device_id}...\n-----END CERTIFICATE-----"
```

## Explanation

This is a Time-of-Check Time-of-Use (TOCTOU) race condition vulnerability. Between checking if the file exists (line 6) and creating it (line 8), an attacker can create a symlink at that path pointing to a sensitive system file, causing the application to overwrite it with certificate data and then set restrictive permissions via a separate os.chmod() call, which itself is another race window.

## Remediation

The fix eliminates the TOCTOU race condition by using os.open() with O_CREAT | O_EXCL | O_NOFOLLOW flags, which atomically creates the file only if it doesn't already exist, refuses to follow symlinks, and sets restrictive permissions (0o600) at creation time rather than after writing. This ensures no window exists between checking and creating the file where an attacker could inject a symlink.

## Secure Code
```python
import os
import json
import tempfile
import stat

def provision_iot_device_credentials(device_id, aws_region):
    cert_dir = "/var/iot/certs"
    cert_filename = f"{device_id}_{aws_region}.pem"
    cert_path = os.path.join(cert_dir, cert_filename)

    # Validate that cert_path resolves within the expected directory
    real_dir = os.path.realpath(cert_dir)
    
    try:
        # Use O_CREAT | O_EXCL to atomically create the file only if it doesn't exist
        # O_WRONLY for write-only, O_NOFOLLOW to refuse symlinks
        fd = os.open(cert_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    except FileExistsError:
        # File already exists; verify it's not a symlink before returning
        if os.path.islink(cert_path):
            raise SecurityError(f"Symlink detected at {cert_path}, refusing to use")
        # Verify the real path is within expected directory
        real_path = os.path.realpath(cert_path)
        if not real_path.startswith(real_dir + os.sep):
            raise SecurityError(f"Path traversal detected for {cert_path}")
        return {"status": "exists", "cert_location": cert_path}
    except OSError as e:
        raise RuntimeError(f"Failed to create certificate file: {e}")

    try:
        # Verify the opened file descriptor points to a regular file in expected directory
        fd_stat = os.fstat(fd)
        if not stat.S_ISREG(fd_stat.st_mode):
            os.close(fd)
            raise SecurityError(f"Unexpected file type at {cert_path}")

        device_cert = generate_aws_iot_certificate(device_id)
        with os.fdopen(fd, 'w') as cert_file:
            cert_file.write(device_cert)
        # fd is now closed by os.fdopen context manager
    except Exception:
        # Clean up on failure
        try:
            os.close(fd)
        except OSError:
            pass
        try:
            os.unlink(cert_path)
        except OSError:
            pass
        raise

    return {"status": "provisioned", "cert_location": cert_path}


class SecurityError(Exception):
    pass


def generate_aws_iot_certificate(device_id):
    return f"-----BEGIN CERTIFICATE-----\nMIIC...{device_id}...\n-----END CERTIFICATE-----"
```