Patched code sample

import requests
from urllib.parse import urlparse
import ipaddress


class LocalPythonExecutorFixed:
    """
    A class to execute Python code with protections against Server-Side
    Request Forgery (SSRF) vulnerabilities.

    The fix addresses the described vulnerability by not exposing the `requests`
    module directly to the executed code. Instead, it provides wrapped, safer
    versions of `get` and `post` functions that enforce security checks.
    """

    def __init__(self, allowed_domains=None):
        """
        Initializes the executor with a set of allowed domains for network requests.

        Args:
            allowed_domains (set, optional): A set of strings representing allowed
                                             hostnames. Defaults to an empty set.
        """
        self.allowed_domains = allowed_domains or set()
        
        # The 'safe_globals' dictionary defines the limited, secure environment
        # in which the user-provided code will be executed.
        # Crucially, it does NOT include the 'requests' module directly.
        self.safe_globals = {
            "__builtins__": {
                "print": print,
                "range": range,
                "len": len,
                "str": str,
                "int": int,
                "float": float,
                "list": list,
                "dict": dict,
                "set": set,
                "True": True,
                "False": False,
                "None": None,
            },
            "safe_get": self._safe_requests_get,
            "safe_post": self._safe_requests_post,
        }

    def _is_url_allowed(self, url: str) -> bool:
        """
        Validates a URL against a set of security policies to prevent SSRF.

        Checks performed:
        1. The URL scheme must be HTTP or HTTPS.
        2. The hostname must not be a private/internal or reserved IP address.
        3. The hostname must be in the predefined `allowed_domains` list.

        Args:
            url (str): The URL to validate.

        Returns:
            bool: True if the URL is allowed, False otherwise.
        """
        try:
            parsed_url = urlparse(url)

            # 1. Enforce allowed schemes
            if parsed_url.scheme not in ["http", "https"]:
                return False

            hostname = parsed_url.hostname
            if not hostname:
                return False

            # 2. Resolve hostname to IP and check if it's a private/reserved address
            # This helps prevent DNS rebinding attacks and direct calls to internal IPs.
            try:
                ip_addr = ipaddress.ip_address(requests.utils.get_host_ip_address(hostname))
                if ip_addr.is_private or ip_addr.is_loopback or ip_addr.is_reserved:
                    return False
            except Exception:
                # If hostname resolution fails, deny the request.
                return False

            # 3. Check against the domain allowlist
            if hostname not in self.allowed_domains:
                return False

            return True

        except (ValueError, AttributeError):
            # Catches malformed URLs or other parsing errors
            return False

    def _safe_requests_get(self, url, **kwargs):
        """
        A secure wrapper for `requests.get`. It enforces the URL allowlist
        and removes potentially dangerous arguments.
        """
        if self._is_url_allowed(url):
            # Sanitize kwargs to prevent security risks like proxy injection
            kwargs.pop("proxies", None)
            kwargs.setdefault("timeout", 5)  # Enforce a timeout
            return requests.get(url, **kwargs)
        else:
            raise PermissionError(f"SSRF Detected: Access to URL '{url}' is forbidden by policy.")

    def _safe_requests_post(self, url, **kwargs):
        """
        A secure wrapper for `requests.post`. It enforces the URL allowlist
        and removes potentially dangerous arguments.
        """
        if self._is_url_allowed(url):
            # Sanitize kwargs
            kwargs.pop("proxies", None)
            kwargs.setdefault("timeout", 5)
            return requests.post(url, **kwargs)
        else:
            raise PermissionError(f"SSRF Detected: Access to URL '{url}' is forbidden by policy.")

    def execute_code(self, code_string: str):
        """
        Executes a string of Python code in a sandboxed environment.

        The vulnerability is fixed by providing a restricted scope to the `exec`
        function. Malicious code attempting to use `requests.get` will raise a
        `NameError`, and code using `safe_get` with a disallowed URL will
        raise a `PermissionError`.

        Args:
            code_string (str): The Python code to execute.
        """
        try:
            # Execute the code with the safe, limited global scope.
            exec(code_string, self.safe_globals, {})
        except Exception as e:
            # In a real application, this would be logged securely.
            print(f"Code execution failed: {type(e).__name__}: {e}")