Patched code sample

import os
import importlib.util
from pathlib import Path
import logging
from typing import Optional, Type

# Set up a logger for demonstration purposes
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def _resolve_pydantic_class(class_name: str, workflow_file_path: str) -> Optional[Type]:
    """
    Safely resolves a Pydantic class by name from a sibling 'tools.py' file.

    This function represents the fixed version of the vulnerable code. The fix
    involves checking an environment variable before attempting to load and
    execute code from an arbitrary file path derived from user-supplied input.
    """
    # --- FIX START ---
    # The vulnerability was that the code would unconditionally load a `tools.py`
    # file relative to the workflow file. The fix is to make this behavior
    # opt-in by checking an environment variable, thus preventing arbitrary
    # code execution by default.
    allow_local_tools = os.getenv("PRAISONAI_ALLOW_LOCAL_TOOLS", "false").lower() in ("true", "1")

    if not allow_local_tools:
        logger.warning(
            f"Attempt to dynamically load '{class_name}' from a local tools.py was blocked. "
            "To allow this, set the environment variable 'PRAISONAI_ALLOW_LOCAL_TOOLS=true'. "
            "This is a security feature to prevent arbitrary code execution."
        )
        return None
    # --- FIX END ---

    # The original, but now guarded, dynamic loading logic proceeds only if allowed.
    try:
        workflow_path = Path(workflow_file_path).parent
        tools_file_path = workflow_path / "tools.py"

        if not tools_file_path.is_file():
            logger.debug(f"Sibling tools.py not found at: {tools_file_path}")
            return None

        logger.info(f"PRAISONAI_ALLOW_LOCAL_TOOLS is true. Attempting to load from {tools_file_path}")

        # Dynamically load the module from the tools.py file
        module_name = f"praisonai.dynamic_tools.{workflow_path.name}"
        spec = importlib.util.spec_from_file_location(module_name, tools_file_path)

        if spec and spec.loader:
            module = importlib.util.module_from_spec(spec)
            # This line executes the code in tools.py
            spec.loader.exec_module(module)
            # Return the class from the now-loaded module
            return getattr(module, class_name, None)
        return None

    except Exception as e:
        logger.error(f"Error dynamically loading class '{class_name}' from '{tools_file_path}': {e}")
        return None