feat(hybrid-cloud): Validate Control silo IP address for IntegrationProxyClient (#61649)

src/sentry/conf/server.py

@@ -678,7 +678,7 @@ RPC_TIMEOUT = 5.0
 
 # The protocol, host and port for control silo
 # Usecases include sending requests to the Integration Proxy Endpoint and RPC requests.
-SENTRY_CONTROL_ADDRESS = os.environ.get("SENTRY_CONTROL_ADDRESS", None)
+SENTRY_CONTROL_ADDRESS: str | None = os.environ.get("SENTRY_CONTROL_ADDRESS", None)
 
 # Fallback region name for monolith deployments
 # This region name is also used by the ApiGateway to proxy org-less region

src/sentry/shared_integrations/client/proxy.py

@@ -1,15 +1,23 @@
 from __future__ import annotations
 
+import ipaddress
 import logging
+import socket
+from functools import lru_cache
 from typing import Any, Mapping
 from urllib.parse import ParseResult, urljoin, urlparse
 
+import sentry_sdk
+import urllib3
 from django.conf import settings
 from django.http import HttpResponse
+from django.utils.encoding import force_str
 from requests import PreparedRequest
 
 from sentry.db.postgres.transactions import in_test_hide_transaction_boundary
+from sentry.http import build_session
 from sentry.integrations.client import ApiClient
+from sentry.net.http import SafeSession
 from sentry.services.hybrid_cloud.integration.service import integration_service
 from sentry.services.hybrid_cloud.util import control_silo_function
 from sentry.silo.base import SiloMode
@@ -28,6 +36,36 @@ from sentry.utils.env import in_test_environment
 logger = logging.getLogger(__name__)
 
 
+@lru_cache(maxsize=1)
+def get_control_silo_ip_address() -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
+    address: str | None = settings.SENTRY_CONTROL_ADDRESS
+    if address is None:
+        return None
+
+    url = urllib3.util.parse_url(address)
+    if url.host:
+        # This is an IPv4 address.
+        # In the future we can consider adding IPv4/v6 dual stack support if and when we start using IPv6 addresses.
+        ip = socket.gethostbyname(url.host)
+        return ipaddress.ip_address(force_str(ip, strings_only=True))
+    else:
+        sentry_sdk.capture_exception(
+            Exception(f"Unable to parse hostname of control silo address: {address}")
+        )
+    return None
+
+
+def is_control_silo_ip_address(ip: str) -> bool:
+    ip_address = ipaddress.ip_address(force_str(ip, strings_only=True))
+
+    expected_address = get_control_silo_ip_address()
+    result = ip_address == expected_address
+
+    if not result:
+        sentry_sdk.capture_exception(Exception(f"Disallowed Control Silo IP address: {ip}"))
+    return result
+
+
 def infer_org_integration(
     integration_id: int, ctx_logger: logging.Logger | None = None
 ) -> int | None:
@@ -95,6 +133,17 @@ class IntegrationProxyClient(ApiClient):
             logger.info("proxy_disabled_in_test_env")
             self.proxy_url = self.base_url
 
+    def build_session(self) -> SafeSession:
+        """
+        Generates a safe Requests session for the API client to use.
+        This injects a custom is_ipaddress_permitted function to allow only connections to the IP address of the Control Silo.
+        We only validate the IP address from within the Region Silo.
+        For all other silo modes, we use the default is_ipaddress_permitted function, which tests against SENTRY_DISALLOWED_IPS.
+        """
+        if SiloMode.get_current_mode() == SiloMode.REGION:
+            return build_session(is_ipaddress_permitted=is_control_silo_ip_address)
+        return build_session()
+
     @staticmethod
     def determine_whether_should_proxy_to_control() -> bool:
         return (

tests/sentry/shared_integrations/client/test_proxy.py

@@ -1,9 +1,15 @@
-from unittest.mock import patch
+import ipaddress
+from unittest.mock import MagicMock, patch
 
 from django.test import override_settings
+from pytest import raises
 from requests import Request
 
-from sentry.shared_integrations.client.proxy import IntegrationProxyClient
+from sentry.shared_integrations.client.proxy import (
+    IntegrationProxyClient,
+    get_control_silo_ip_address,
+)
+from sentry.shared_integrations.exceptions import ApiHostError
 from sentry.silo.base import SiloMode
 from sentry.silo.util import PROXY_OI_HEADER, PROXY_SIGNATURE_HEADER
 from sentry.testutils.cases import TestCase
@@ -84,3 +90,101 @@ class IntegrationProxyClientTest(TestCase):
         assert prepared_request.url != raw_url
         for header in [PROXY_OI_HEADER, PROXY_SIGNATURE_HEADER]:
             assert header in prepared_request.headers
+
+    @override_settings(SILO_MODE=SiloMode.REGION)
+    @patch("sentry.shared_integrations.client.proxy.get_control_silo_ip_address")
+    @patch("socket.getaddrinfo")
+    def test_invalid_control_silo_ip_address(
+        self, mock_getaddrinfo, mock_get_control_silo_ip_address
+    ):
+        with patch("sentry_sdk.capture_exception") as mock_capture_exception, raises(ApiHostError):
+            mock_get_control_silo_ip_address.return_value = ipaddress.ip_address("127.0.0.1")
+            mock_getaddrinfo.return_value = [(2, 1, 6, "", ("172.31.255.255", 0))]
+            client = self.client_cls(org_integration_id=self.oi_id)
+
+            client.get(f"{self.base_url}/some/endpoint", params={"query": 1, "user": "me"})
+
+        assert mock_capture_exception.call_count == 1
+        err = mock_capture_exception.call_args.args[0]
+        assert err.args == ("Disallowed Control Silo IP address: 172.31.255.255",)
+
+    @override_settings(SILO_MODE=SiloMode.REGION)
+    @patch("sentry.shared_integrations.client.proxy.is_control_silo_ip_address")
+    @patch("sentry.shared_integrations.client.proxy.get_control_silo_ip_address")
+    @patch("socket.getaddrinfo")
+    def test_valid_control_silo_ip_address(
+        self, mock_getaddrinfo, mock_get_control_silo_ip_address, mock_is_control_silo_ip_address
+    ):
+        mock_get_control_silo_ip_address.return_value = ipaddress.ip_address("172.31.255.255")
+        mock_getaddrinfo.return_value = [(2, 1, 6, "", ("172.31.255.255", 0))]
+        client = self.client_cls(org_integration_id=self.oi_id)
+
+        class BailOut(Exception):
+            pass
+
+        def test_is_control_silo_ip_address(ip):
+            assert ip == "172.31.255.255"
+            # We can't use responses library for this unit test as it hooks Session.send. So we assert that the
+            # is_control_silo_ip_address function is properly called.
+            raise BailOut()
+
+        mock_is_control_silo_ip_address.side_effect = test_is_control_silo_ip_address
+
+        with raises(BailOut):
+            client.get(f"{self.base_url}/some/endpoint", params={"query": 1, "user": "me"})
+        assert mock_is_control_silo_ip_address.call_count == 1
+
+    @override_settings(SILO_MODE=SiloMode.CONTROL)
+    @patch("sentry.shared_integrations.client.proxy.is_control_silo_ip_address")
+    @patch("sentry.shared_integrations.client.proxy.get_control_silo_ip_address")
+    @patch("socket.getaddrinfo")
+    def test_does_not_validate_control_silo_ip_address_in_control(
+        self, mock_getaddrinfo, mock_get_control_silo_ip_address, mock_is_control_silo_ip_address
+    ):
+        mock_get_control_silo_ip_address.return_value = ipaddress.ip_address("172.31.255.255")
+        mock_getaddrinfo.return_value = [(2, 1, 6, "", ("172.31.255.255", 0))]
+        client = self.client_cls(org_integration_id=self.oi_id)
+
+        class BailOut(Exception):
+            pass
+
+        def test_socket_connection(*args, **kwargs):
+            # We can't use responses library for this unit test as it hooks Session.send. So we assert that the
+            # socket connection is being opened.
+            raise BailOut()
+
+        with patch("socket.socket") as mock_socket, raises(BailOut):
+            mock_socket.side_effect = test_socket_connection
+            client.get(f"{self.base_url}/some/endpoint", params={"query": 1, "user": "me"})
+
+        # Assert control silo ip address was not validated
+        assert mock_get_control_silo_ip_address.call_count == 0
+        assert mock_is_control_silo_ip_address.call_count == 0
+
+
+def test_get_control_silo_ip_address():
+    with override_settings(SENTRY_CONTROL_ADDRESS=None):
+        assert get_control_silo_ip_address() is None
+
+    with override_settings(SENTRY_CONTROL_ADDRESS=control_address):
+        get_control_silo_ip_address.cache_clear()
+        with patch("socket.gethostbyname") as mock_gethostbyname, patch(
+            "sentry_sdk.capture_exception"
+        ) as mock_capture_exception:
+            mock_gethostbyname.return_value = "172.31.255.255"
+            assert get_control_silo_ip_address() == ipaddress.ip_address("172.31.255.255")
+            assert mock_capture_exception.call_count == 0
+
+        get_control_silo_ip_address.cache_clear()
+        with patch("socket.gethostbyname") as mock_gethostbyname, patch(
+            "urllib3.util.parse_url"
+        ) as mock_parse_url, patch("sentry_sdk.capture_exception") as mock_capture_exception:
+            mock_parse_url.return_value = MagicMock(host=None)
+            assert get_control_silo_ip_address() is None
+            assert mock_gethostbyname.call_count == 0
+            assert mock_capture_exception.call_count == 1
+
+            err = mock_capture_exception.call_args.args[0]
+            assert err.args == (
+                f"Unable to parse hostname of control silo address: {control_address}",
+            )