Patched code sample

import ipaddress
import socket
from urllib.parse import urlparse
import logging

# A logger to report issues, similar to the real application.
LOGGER = logging.getLogger(__name__)


def is_url_safe(url: str) -> bool:
    """
    This function represents the core of the security fix.
    It checks if a URL's hostname resolves to a public, globally routable
    IP address to prevent Server-Side Request Forgery (SSRF) attacks
    against internal network services.

    :param url: The URL to be validated.
    :returns: `bool` of whether the URL is considered safe.
    """
    try:
        parsed_url = urlparse(url)

        # Ensure the URL has a valid scheme and hostname
        if not all([parsed_url.scheme, parsed_url.hostname]):
            LOGGER.warning(f"URL missing scheme or hostname: {url}")
            return False

        # Resolve the hostname to an IP address
        hostname = parsed_url.hostname
        ip_addr = socket.gethostbyname(hostname)
        ip_obj = ipaddress.ip_address(ip_addr)

        # THE FIX: Check if the resolved IP address is a public address.
        # This will return False for:
        # - Private networks (e.g., 10.x.x.x, 192.168.x.x)
        # - Loopback addresses (e.g., 127.0.0.1)
        # - Link-local addresses (e.g., 169.254.x.x used by cloud metadata services)
        if not ip_obj.is_global:
            LOGGER.error(f"Unsafe URL detected: {url} resolves to non-global IP {ip_addr}")
            return False

    except (ValueError, socket.gaierror) as err:
        LOGGER.error(f"Could not validate URL {url}: {err}")
        return False

    return True


def fixed_process_execution(data: dict):
    """
    A simplified function demonstrating how the `is_url_safe` check
    is used to patch the vulnerability.
    """
    subscriber = data.get('subscriber', {})
    url = subscriber.get('url')

    if url:
        # Applying the fix: The user-provided URL is validated before use.
        # If the URL is unsafe, the function will not proceed to make a request.
        if not is_url_safe(url):
            # In a real application, this would return an HTTP 400 Bad Request.
            msg = f"Error: Subscriber URL '{url}' is not safe and was rejected."
            LOGGER.error(msg)
            # Abort the operation
            return

        # If the check passes, the code can safely proceed.
        LOGGER.info(f"URL '{url}' is safe. Proceeding with notification.")
        # ... code to make the HTTP POST request would follow here ...
        # e.g., requests.post(url, json={"status": "complete"})