Patched code sample

import subprocess
import sys
import logging

# Configure basic logging for clarity in execution
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class JobWorkflowExecutor:
    """
    Executes jobs defined in a workflow YAML file.
    This version includes security fixes for CVE-2026-40288 by requiring
    user confirmation before executing shell commands or Python code.
    """

    def __init__(self, interactive_mode=True):
        """
        Initializes the executor.

        Args:
            interactive_mode (bool): If False, disables security confirmations and
                                     refuses to run unsafe steps. This is for
                                     non-interactive CI/CD environments where safety
                                     is paramount.
        """
        self.interactive_mode = interactive_mode
        if not self.interactive_mode:
            logging.warning(
                "NON-INTERACTIVE MODE: Execution of 'run', 'script', and 'python' steps is disabled for security."
            )

    # --- START OF FIX for CVE-2026-40288 ---

    def _confirm_execution(self, execution_type: str, content: str) -> bool:
        """
        Prompts the user for confirmation before executing a potentially unsafe command or script.
        This function is the core of the fix, addressing the lack of validation or user confirmation.
        """
        if not self.interactive_mode:
            logging.error(
                f"Attempted to execute an unsafe step ('{execution_type}') in a non-interactive environment. "
                "Execution denied for security reasons."
            )
            return False

        print("-" * 60)
        print("⚠️  SECURITY WARNING: The workflow is attempting to execute a local task.")
        print(f"   Type: {execution_type}")
        print("   Content:")
        # Indent content for readability
        for line in content.strip().splitlines():
            print(f"     {line}")
        print("-" * 60)

        try:
            response = input("Do you want to allow this execution? [y/N]: ").strip().lower()
        except (EOFError, KeyboardInterrupt):
            # Handle non-interactive environments or user cancellation
            response = 'n'
            print("\nConfirmation cancelled or input stream closed. Denying execution.")

        if response in ('y', 'yes'):
            logging.info("User approved execution.")
            return True
        else:
            logging.error("User denied execution. Aborting workflow.")
            return False

    # --- END OF FIX ---

    def _exec_shell(self, command: str):
        """
        Executes a shell command only after receiving explicit user confirmation.
        """
        # FIX: The direct call to subprocess is now gated by the confirmation function.
        if self._confirm_execution("Shell Command (`run:`)", command):
            try:
                logging.info(f"Executing approved shell command: {command}")
                # The original vulnerability was calling this directly.
                subprocess.run(command, shell=True, check=True, text=True)
                logging.info("Command executed successfully.")
            except subprocess.CalledProcessError as e:
                logging.error(f"Command failed with exit code {e.returncode}")
                sys.exit(1)
            except Exception as e:
                logging.error(f"An unexpected error occurred during shell execution: {e}")
                sys.exit(1)
        else:
            sys.exit(1) # Abort workflow if user denies

    def _exec_inline_python(self, script: str):
        """
        Executes an inline Python script only after receiving explicit user confirmation.
        """
        # FIX: The direct call to exec() is now gated by the confirmation function.
        if self._confirm_execution("Inline Python Script (`script:`)", script):
            try:
                logging.info("Executing approved inline Python script.")
                # The original vulnerability was calling exec() on untrusted input directly.
                exec(script, {'__builtins__': __builtins__})
                logging.info("Inline script executed successfully.")
            except Exception as e:
                logging.error(f"Inline Python script failed: {e}")
                sys.exit(1)
        else:
            sys.exit(1) # Abort workflow if user denies

    def _exec_python_script(self, script_path: str):
        """
        Reads and executes a Python script from a file only after user confirmation.
        """
        try:
            with open(script_path, 'r') as f:
                script_content = f.read()
        except FileNotFoundError:
            logging.error(f"Python script not found at path: {script_path}")
            sys.exit(1)
        except Exception as e:
            logging.error(f"Failed to read script file '{script_path}': {e}")
            sys.exit(1)

        # FIX: Execution of the script's content is gated by user confirmation.
        if self._confirm_execution(f"Python Script File (`python: {script_path}`)", script_content):
            try:
                logging.info(f"Executing approved Python script: {script_path}")
                # The original vulnerability was executing this content without checks.
                exec(script_content, {'__builtins__': __builtins__})
                logging.info(f"Script '{script_path}' executed successfully.")
            except Exception as e:
                logging.error(f"Execution of script '{script_path}' failed: {e}")
                sys.exit(1)
        else:
            sys.exit(1) # Abort workflow if user denies

    def execute_step(self, step: dict):
        """
        Parses and executes a single step from the workflow, dispatching to
        the newly secured execution methods.
        """
        if "run" in step:
            self._exec_shell(step["run"])
        elif "script" in step:
            self._exec_inline_python(step["script"])
        elif "python" in step:
            self._exec_python_script(step["python"])
        else:
            logging.info(f"Step '{step.get('name', 'Unnamed')}' contains no executable action.")