Vulnerable code sample

import os
import subprocess
import shlex
import sys
import time

# This code is a representation of the vulnerable behavior described in CVE-2026-40159.
# It does not originate from the PraisonAI codebase but is designed to functionally
# demonstrate the core of the vulnerability.

class MCP:
    """
    A representative class simulating the vulnerable MCP component prior to the fix.
    When instantiated, it executes a command string using subprocess.Popen,
    inheriting the parent process's environment by default.
    """
    def __init__(self, command_string: str):
        print(f"[MCP CLASS] Received command: '{command_string}'")
        print("[MCP CLASS] Spawning subprocess...")
        
        # The vulnerability lies here: subprocess.Popen is called without the `env`
        # argument. By default, this means the child process inherits all
        # environment variables from the parent process.
        args = shlex.split(command_string)
        
        try:
            # For demonstration, we capture stdout and stderr to show the output.
            # In a real scenario, this might be a background server process.
            self.process = subprocess.Popen(
                args,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                # The 'env' parameter is omitted, causing the vulnerability.
                # A fixed version might look like: env={} or a sanitized dict.
            )
            print("[MCP CLASS] Subprocess created successfully.")
        except Exception as e:
            print(f"[MCP CLASS] Failed to create subprocess: {e}")
            self.process = None

    def wait_for_completion(self):
        """Helper method to wait for the subprocess and print its output."""
        if self.process:
            print("[MCP CLASS] Waiting for subprocess to complete...")
            stdout, stderr = self.process.communicate()
            if stdout:
                print("\n--- Subprocess STDOUT ---")
                print(stdout.strip())
                print("--- End Subprocess STDOUT ---\n")
            if stderr:
                print("\n--- Subprocess STDERR ---")
                print(stderr.strip())
                print("--- End Subprocess STDERR ---\n")
            print(f"[MCP CLASS] Subprocess finished with exit code: {self.process.returncode}")


def main():
    """
    Main function to demonstrate the environment variable leakage.
    """
    print("--- Starting Vulnerability Demonstration ---")

    # 1. Set a sensitive environment variable in the parent process.
    # This simulates a production environment where secrets are loaded.
    sensitive_key_name = "DATABASE_CONNECTION_STRING"
    sensitive_key_value = "postgres://user:supersecretpassword@prod-db.example.com:5432/maindb"
    os.environ[sensitive_key_name] = sensitive_key_value
    print(f"[HOST PROCESS] Sensitive environment variable '{sensitive_key_name}' is set.")
    print("-" * 40)

    # 2. Define a "malicious" command provided by a user or an untrusted source.
    # This command attempts to read the environment variable and print it.
    # In a real attack, it would exfiltrate it to a remote server.
    # We use `sys.executable` to ensure the correct python interpreter is used.
    malicious_command_string = (
        f"{sys.executable} -c \""
        f"import os; "
        f"secret = os.environ.get('{sensitive_key_name}'); "
        f"if secret: "
        f"    print(f'[MALICIOUS SCRIPT] SUCCESS: Found secret: {{secret}}'); "
        f"else: "
        f"    print('[MALICIOUS SCRIPT] FAILED: Could not find the secret.');\""
    )

    # 3. Instantiate the vulnerable MCP class with the malicious command.
    # This triggers the execution of the command in a new subprocess.
    print("[HOST PROCESS] Instantiating vulnerable MCP with a user-supplied command...")
    vulnerable_mcp_instance = MCP(malicious_command_string)
    
    # 4. Wait for the subprocess to finish to see the result.
    if vulnerable_mcp_instance.process:
        vulnerable_mcp_instance.wait_for_completion()

    print("--- Vulnerability Demonstration Finished ---")
    
    # Clean up the environment variable
    del os.environ[sensitive_key_name]

if __name__ == "__main__":
    main()