fix: Guard against ssrf attacks when creating image input (#4068)

# Description

This pull request add a check to guard against server-side request
forgery (ssrf) attacks
when the executor service invokes create_image_from_url

see: aka.ms/antissrf

# All Promptflow Contribution checklist:
- [ ] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **I have read the [contribution
guidelines](https://github.com/microsoft/promptflow/blob/main/CONTRIBUTING.md).**
- [ ] **I confirm that all new dependencies are compatible with the MIT
license.**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [ ] Title of the pull request is clear and informative.
- [ ] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [ ] Pull request includes test coverage for the included changes.

Author: kdestin

src/promptflow-core/promptflow/_utils/anti_ssrf/__init__.py

@@ -0,0 +1,7 @@
+# Export the main classes for easy import
+from .anti_ssrf import AntiSSRF
+from .anti_ssrf_policy import AntiSSRFPolicy
+from .exceptions import AntiSSRFException
+
+# Make classes available for import
+__all__ = ["AntiSSRF", "AntiSSRFPolicy", "AntiSSRFException"]

src/promptflow-core/promptflow/_utils/anti_ssrf/anti_ssrf.py

@@ -0,0 +1,76 @@
+# Export the main classes for easy import
+from typing import List, Optional
+
+from .anti_ssrf_policy import AntiSSRFPolicy
+from .exceptions import AntiSSRFException
+
+
+# Simple wrapper class that provides validate_url method
+class AntiSSRF:
+    """Anti-SSRF protection class that validates URLs and network connections."""
+
+    def __init__(self, policy: Optional[AntiSSRFPolicy] = None):
+        """Initialize AntiSSRF with an optional custom policy."""
+        self.policy: AntiSSRFPolicy = policy if policy is not None else AntiSSRFPolicy(use_defaults=True)
+
+    def validate_url(self, url: str, headers={}) -> None:
+        """
+        Validate a URL against the Anti-SSRF policy.
+
+        Args:
+            url: The URL to validate
+
+        Raises:
+            AntiSSRFException: If the URL is not allowed by the policy
+        """
+        if not url:
+            return
+
+        from urllib.parse import urlparse
+
+        # Parse the URL
+        try:
+            parsed_url = urlparse(url)
+        except Exception as e:
+            raise AntiSSRFException(f"Invalid URL format: {e}")
+
+        if not parsed_url.hostname:
+            raise AntiSSRFException("URL must have a hostname")
+
+        # Resolve DNS and check network connections
+        if parsed_url.hostname != "registries" and parsed_url.hostname != "location.api.azureml.ms":
+            dns_resolved_ips = self._resolve_hostname(parsed_url.hostname)
+
+            if not self.policy.is_network_connection_allowed(dns_resolved_ips):
+                raise AntiSSRFException(f"Network connection to '{parsed_url.hostname}' is not allowed")
+
+        # Check HTTP scheme
+        if not self.policy.is_http_request_allowed(parsed_url.scheme, headers):
+            raise AntiSSRFException(f"HTTP scheme '{parsed_url.scheme}' is not allowed")
+
+    def _resolve_hostname(self, hostname: str) -> List[str]:
+        """Resolve hostname to IP addresses."""
+        import ipaddress
+        import socket
+
+        # Handle localhost explicitly
+        if hostname.lower() == "localhost":
+            return ["127.0.0.1"]
+
+        # Try to parse as IP address first
+        try:
+            ip_address = ipaddress.ip_address(hostname)
+            return [str(ip_address)]
+        except ValueError:
+            pass  # Not an IP address, continue with DNS resolution
+
+        # Perform DNS resolution
+        try:
+            _, _, ip_addresses = socket.gethostbyname_ex(hostname)
+            if not ip_addresses:
+                raise AntiSSRFException(f"No IP addresses resolved for hostname: {hostname}")
+            return ip_addresses
+        except socket.gaierror as e:
+            raise AntiSSRFException(f"DNS resolution failed for hostname '{hostname}': {e}")
+        except Exception as e:
+            raise AntiSSRFException(f"Error resolving hostname '{hostname}': {e}")

src/promptflow-core/promptflow/_utils/anti_ssrf/anti_ssrf_policy.py

@@ -0,0 +1,173 @@
+import ipaddress
+from typing import List, Optional
+
+from .cidr_helpers import IPNetwork, try_parse_cidr_string
+from .exceptions import AntiSSRFException
+
+
+class AntiSSRFPolicy:
+    def __init__(self, use_defaults: bool = True):
+        self.AllowedAddresses: List[IPNetwork] = []
+        self.DeniedAddresses: List[IPNetwork] = []
+        self.DeniedHeaders: List[str] = []
+        self.RequiredHeaders: List[str] = []
+        self.AllowPlainTextHttp: bool = False
+        self.AddXFFHeader: bool = True
+        self.DenyAllUnspecifiedIPs: bool = False
+
+        if use_defaults:
+            self._set_defaults()
+
+    def add_allowed_addresses(self, networks: List[str]) -> bool:
+        for network in networks:
+            outnet = try_parse_cidr_string(network)
+            self.AllowedAddresses.append(outnet)
+        return True
+
+    def add_denied_addresses(self, networks: List[str]) -> bool:
+        if self.DenyAllUnspecifiedIPs:
+            raise AntiSSRFException("Can't add denied networks when * is already supplied")
+        if not networks:
+            raise AntiSSRFException("Bad networks parameter")
+        if len(networks) == 1 and networks[0] == "*":
+            if len(self.DeniedAddresses) > 0:
+                raise AntiSSRFException("Can't add * when deny list already has entries")
+            self.DenyAllUnspecifiedIPs = True
+            return True
+        else:
+            for network in networks:
+                outnet = try_parse_cidr_string(network)
+                self.DeniedAddresses.append(outnet)
+            return True
+
+    def add_denied_headers(self, denied_headers: Optional[List[str]]) -> None:
+        if denied_headers:
+            self.DeniedHeaders.extend(denied_headers)
+
+    def add_required_headers(self, required_headers: Optional[List[str]]) -> None:
+        if required_headers:
+            self.RequiredHeaders.extend(required_headers)
+
+    def set_allow_plain_text_http(self, allow_plain_text_http: bool = False) -> None:
+        self.AllowPlainTextHttp = allow_plain_text_http
+
+    def add_xff(self, add_xff: bool = True) -> None:
+        self.AddXFFHeader = add_xff
+
+    # IP Addresses in Deny List can be IPv4, IPv6 or IPv4 mapped to IPv6
+    # Accordingly, to check if an input address from DNS resolution is to be denied, we should:
+    # 1. Check if the input IP is an IPv4 address, and then check if it is present in deny list
+    #    as a pure IPv4 or an IPv4 mapped to IPv6 format
+    # 2. Check if the input IP is an IPv6 address, and then check if it is present in deny list
+    #    as a pure IPv6 address. This includes addresses in IPv4 mapped to IPv6 format
+    # 3. Check if the input IP is an IPv4 mapped to IPv6, then check if it is present in the
+    #    deny list as an IPv4 mapped IPv6 address, then convert it to IPv4 and check if it is
+    #    present in the deny list as a pure IPv4 address
+    #
+    # For example, 169.254.169.254, if present in the deny list, should deny DNS resolved
+    # addresses 169.254.169.254 and ::ffff:a9fe:a9fe
+    # Likewise ::ffff:a9fe:a9fe, if present in the deny list, should deny DNS resolved
+    # addresses ::ffff:a9fe:a9fe and 169.254.169.254
+    #
+    # Such case-by-case comparisons leads to a lot of branches in code leading to
+    # sphagettification and also makes code difficult to follow and maintain
+    # Furthermore, the complexity gets compounded if one adds an allow list to the mix
+    #
+    # To make things easier and efficient, we convert every IPv4 address to IPv6 across the
+    # deny list, allow list and also the input DNS resolved addresses
+    # The CIDR helper class is accordingly written
+    #
+    # As IPv6 is the future anyway, this also makes the code future proof
+    def is_network_connection_allowed(self, dns_resolved_ip_addresses: List[str]) -> bool:
+        for ip_str in dns_resolved_ip_addresses:
+            ip_address = ipaddress.ip_address(ip_str)
+            ipv6_address = (
+                ip_address
+                if isinstance(ip_address, ipaddress.IPv6Address)
+                else ipaddress.IPv6Address(f"::ffff:{ip_address}")
+            )
+
+            if self.DenyAllUnspecifiedIPs:
+                # If the address is not in an allow list, it's not allowed.
+                if not self._networks_contain_address(self.AllowedAddresses, ipv6_address):
+                    return False
+            elif self.DeniedAddresses:
+                # If address is in deny list and not in allow list, it's not allowed.
+                if self._networks_contain_address(
+                    self.DeniedAddresses, ipv6_address
+                ) and not self._networks_contain_address(self.AllowedAddresses, ipv6_address):
+                    return False
+        # No IP addresses returned by DNS resolution were denied
+        return True
+
+    @staticmethod
+    def _networks_contain_address(networks: List[IPNetwork], address: ipaddress.IPv6Address) -> bool:
+        for network in networks:
+            if network.contains(address):
+                return True
+        return False
+
+    def is_http_request_allowed(self, scheme: str, headers: dict) -> bool:
+        if scheme.lower() == "http" and not self.AllowPlainTextHttp:
+            return False
+
+        if self.AddXFFHeader:
+            if "X-Forwarded-For" not in headers:
+                headers["X-Forwarded-For"] = "true"
+
+        if self.DeniedHeaders:
+            for header in self.DeniedHeaders:
+                if header in headers:
+                    return False
+
+        if self.RequiredHeaders:
+            for header in self.RequiredHeaders:
+                if header not in headers:
+                    return False
+
+        return True
+
+    def _set_defaults(self):
+        self.AllowedAddresses = []
+        self.DeniedAddresses = []
+        self.RequiredHeaders = []
+        self.DeniedHeaders = []
+        self.AllowPlainTextHttp = False
+        self.DenyAllUnspecifiedIPs = False
+        self.AddXFFHeader = True
+
+        self.add_denied_addresses(
+            [
+                # ==== IPv4 ==== #
+                "255.255.255.255/32",
+                "168.63.129.16/32",  # Not nonroutable,
+                # but this is the WireServer IP we should block.
+                "192.0.0.0/24",
+                "192.0.2.0/24",
+                "192.88.99.0/24",
+                "198.51.100.0/24",
+                "203.0.113.0/24",
+                "169.254.0.0/16",
+                "192.168.0.0/16",
+                "198.18.0.0/15",
+                "172.16.0.0/12",
+                "100.64.0.0/10",  # IANA-Reserved
+                "0.0.0.0/8",
+                "10.0.0.0/8",
+                "127.0.0.0/8",
+                "25.0.0.0/8",  # GNS Core
+                "224.0.0.0/4",
+                "240.0.0.0/4",
+                # ==== IPv6 ==== #
+                "::1/128",  # Localhost
+                "FC00::/7",  # Unique-local
+                "fe80::/10",  # Link-local
+                "fec0::/10",  # Site-local
+                "2001::/32",  # Teredo
+            ]
+        )
+        self.DenyAllUnspecifiedIPs = False
+
+    # Deprecated method, for backward compatibility only
+    def set_defaults(self):
+        self._set_defaults()

src/promptflow-core/promptflow/_utils/anti_ssrf/cidr_helpers.py

@@ -0,0 +1,75 @@
+import ipaddress
+from typing import Union
+
+from .exceptions import AntiSSRFException
+
+
+class IPNetwork:
+    def __init__(self, ip: Union[str, ipaddress.IPv4Address, ipaddress.IPv6Address], prefix: int) -> None:
+        self._base_address = ipaddress.ip_network(f"{ip}/{prefix}", strict=False)
+        self._prefix_length = prefix
+
+    def contains(self, ip: Union[str, ipaddress.IPv4Address, ipaddress.IPv6Address]) -> bool:
+        ip_obj = ip
+        if isinstance(ip, str):
+            ip_obj = ipaddress.ip_address(ip)
+        # Convert IPv4 to IPv6-mapped if base is IPv6
+        if self._base_address.version == 6 and ip_obj.version == 4:
+            ip_obj = ipaddress.IPv6Address(f"::ffff:{ip_obj}")
+        return ip_obj in self._base_address
+
+
+def _parse_ip_address(ip_string: str) -> Union[ipaddress.IPv4Address, ipaddress.IPv6Address]:
+    """Parse IP address from string, raising AntiSSRFException on error."""
+    try:
+        return ipaddress.ip_address(ip_string)
+    except ValueError as e:
+        raise AntiSSRFException("Bad CIDR", e)
+
+
+def _parse_prefix_length(prefix_string: str) -> int:
+    """Parse prefix length from string, raising AntiSSRFException on error."""
+    try:
+        return int(prefix_string)
+    except ValueError as e:
+        raise AntiSSRFException("Bad CIDR", e)
+
+
+def _create_single_ip_network(ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]) -> IPNetwork:
+    """Create network for single IP address (no prefix specified)."""
+    if ip.version == 4:
+        # IPv4 mapped to IPv6, /128
+        return IPNetwork(f"::ffff:{ip}", 128)
+    elif ip.version == 6:
+        return IPNetwork(ip, 128)
+    else:
+        raise AntiSSRFException("Bad CIDR")
+
+
+def _create_prefixed_network(ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address], prefix_length: int) -> IPNetwork:
+    """Create network for IP address with prefix."""
+    if ip.version == 4 and 0 <= prefix_length <= 32:
+        # IPv4 mapped to IPv6, prefix + 96
+        return IPNetwork(f"::ffff:{ip}", prefix_length + 96)
+    elif ip.version == 6 and 0 <= prefix_length <= 128:
+        return IPNetwork(ip, prefix_length)
+    else:
+        raise AntiSSRFException("Bad CIDR")
+
+
+# Try parse CIDR string
+# Returns an IPNetwork object if everything went fine, or throws an exception
+# For easy computation of allow/deny, every IP Address is converted into an IPv6 address
+def try_parse_cidr_string(cidr_string: str) -> IPNetwork:
+    parts = cidr_string.split("/")
+    ip = _parse_ip_address(parts[0])
+
+    if len(parts) == 1:
+        # e.g. "127.0.0.1" or "::ffff:909:909"
+        return _create_single_ip_network(ip)
+    elif len(parts) == 2:
+        # Cases such as "127.0.0.1/2" or "::ffff:909:909/80"
+        prefix_length = _parse_prefix_length(parts[1])
+        return _create_prefixed_network(ip, prefix_length)
+    else:
+        raise AntiSSRFException("Bad CIDR")

src/promptflow-core/promptflow/_utils/anti_ssrf/exceptions.py

@@ -0,0 +1,8 @@
+class AntiSSRFException(Exception):
+    def __init__(self, message=None, inner=None):
+        if inner is not None:
+            super().__init__(message, inner)
+        elif message is not None:
+            super().__init__(message)
+        else:
+            super().__init__()

src/promptflow-core/promptflow/_utils/multimedia_utils.py

@@ -12,6 +12,7 @@
 
 from promptflow._constants import MessageFormatType
 from promptflow._utils._errors import InvalidImageInput, InvalidMessageFormatType, LoadMultimediaDataError
+from promptflow._utils.anti_ssrf import AntiSSRF, AntiSSRFException
 from promptflow._utils.yaml_utils import load_yaml
 from promptflow.contracts.flow import FlowInputDefinition
 from promptflow.contracts.multimedia import Image, PFBytes, Text
@@ -93,8 +94,30 @@ def create_image_from_base64(base64_str: str, mime_type: str = None):
         return Image(image_bytes, mime_type=mime_type)
 
     @staticmethod
-    def create_image_from_url(url: str, mime_type: str = None):
-        response = requests.get(url)
+    def create_image_from_url(url: str, mime_type: str = None) -> Image:
+        anti_ssrf = AntiSSRF()
+        anti_ssrf.policy.set_allow_plain_text_http(True)
+
+        def block_redirect_if_ssrf(response: requests.Response, *args, **kwargs) -> None:
+            if not response.is_redirect:
+                return
+
+            anti_ssrf.validate_url(response.headers["Location"])
+
+        try:
+            anti_ssrf.validate_url(url)
+
+            # Use the requests "response" hook to allow us to inspect each response
+            # in a redirect chain.
+            # See: https://requests.readthedocs.io/en/latest/user/advanced/#event-hooks
+            response = requests.get(url, hooks={"response": block_redirect_if_ssrf})
+        except AntiSSRFException as e:
+            raise InvalidImageInput(
+                message_format="Failed to fetch image from URL: {url}.",
+                target=ErrorTarget.EXECUTOR,
+                url=url,
+            ) from e
+
         if response.status_code == 200:
             return Image(response.content, mime_type=mime_type, source_url=url)
         else: