feat: add SENTRY_ALLOWED_IPS to allow IP, overwrite SENTRY_DISALLOWED…

src/sentry/conf/server.py

@@ -163,6 +163,8 @@ def env(
     "ff00::/8",
 )
 
+SENTRY_ALLOWED_IPS: tuple[str, ...] = ()
+
 # When resolving DNS for external sources (source map fetching, webhooks, etc),
 # ensure that domains are fully resolved first to avoid poking internal
 # search domains.

src/sentry/net/socket.py

@@ -22,16 +22,24 @@
     ipaddress.ip_network(str(i), strict=False) for i in settings.SENTRY_DISALLOWED_IPS
 )
 
+ALLOWED_IPS = frozenset(
+    ipaddress.ip_network(str(i), strict=False) for i in settings.SENTRY_ALLOWED_IPS
+)
+
 
 @functools.lru_cache(maxsize=100)
 def is_ipaddress_allowed(ip: str) -> bool:
     """
     Test if a given IP address is allowed or not
-    based on the DISALLOWED_IPS rules.
+    based on the DISALLOWED_IPS AND ALLOWED_IPS rules.
     """
     if not DISALLOWED_IPS:
         return True
     ip_address = ipaddress.ip_address(force_str(ip, strings_only=True))
+    for ip_network in ALLOWED_IPS:
+        if ip_address in ip_network:
+            return True
+
     for ip_network in DISALLOWED_IPS:
         if ip_address in ip_network:
             return False

src/sentry/testutils/helpers/socket.py

@@ -7,7 +7,7 @@
 
 from sentry.net import socket as net_socket
 
-__all__ = ["override_blocklist"]
+__all__ = ["override_blocklist", "override_allowlist"]
 
 
 @contextlib.contextmanager
@@ -23,3 +23,18 @@ def override_blocklist(*ip_addresses: str) -> Generator[None]:
             # We end up caching these disallowed ips on this function, so
             # make sure we clear the cache as part of cleanup
             net_socket.is_ipaddress_allowed.cache_clear()
+
+
+@contextlib.contextmanager
+def override_allowlist(*ip_addresses: str) -> Generator[None]:
+    with mock.patch.object(
+        net_socket,
+        "ALLOWED_IPS",
+        frozenset(ipaddress.ip_network(ip) for ip in ip_addresses),
+    ):
+        try:
+            yield
+        finally:
+            # We end up caching these disallowed ips on this function, so
+            # make sure we clear the cache as part of cleanup
+            net_socket.is_ipaddress_allowed.cache_clear()

tests/sentry/net/test_socket.py

@@ -4,12 +4,12 @@
 
 from sentry.net.socket import ensure_fqdn, is_ipaddress_allowed, is_safe_hostname
 from sentry.testutils.cases import TestCase
-from sentry.testutils.helpers import override_blocklist
+from sentry.testutils.helpers import override_allowlist, override_blocklist
 
 
 class SocketTest(TestCase):
     @override_blocklist("10.0.0.0/8", "127.0.0.1")
-    def test_is_ipaddress_allowed(self) -> None:
+    def test_is_ipaddress_blocked(self) -> None:
         is_ipaddress_allowed.cache_clear()
         assert is_ipaddress_allowed("127.0.0.1") is False
         is_ipaddress_allowed.cache_clear()
@@ -18,7 +18,7 @@ def test_is_ipaddress_allowed(self) -> None:
         assert is_ipaddress_allowed("1.1.1.1") is True
 
     @override_blocklist("::ffff:10.0.0.0/104", "::1/128")
-    def test_is_ipaddress_allowed_ipv6(self) -> None:
+    def test_is_ipaddress_blocked_ipv6(self) -> None:
         is_ipaddress_allowed.cache_clear()
         assert is_ipaddress_allowed("::1") is False
         is_ipaddress_allowed.cache_clear()
@@ -28,6 +28,22 @@ def test_is_ipaddress_allowed_ipv6(self) -> None:
         is_ipaddress_allowed.cache_clear()
         assert is_ipaddress_allowed("2001:db8:a::123") is True
 
+    @override_blocklist("10.0.0.0/8")
+    @override_allowlist("10.0.0.1/32")
+    def test_is_ipaddress_allowed(self) -> None:
+        is_ipaddress_allowed.cache_clear()
+        assert is_ipaddress_allowed("10.0.1.1") is False
+        is_ipaddress_allowed.cache_clear()
+        assert is_ipaddress_allowed("10.0.0.1") is True
+
+    @override_blocklist("::ffff:10.0.0.0/104")
+    @override_allowlist("::ffff:10.0.0.1/128")
+    def test_is_ipaddress_allowed_ipv6(self) -> None:
+        is_ipaddress_allowed.cache_clear()
+        assert is_ipaddress_allowed("::ffff:10.0.1.2") is False
+        is_ipaddress_allowed.cache_clear()
+        assert is_ipaddress_allowed("::ffff:10.0.0.1") is True
+
     @override_blocklist("10.0.0.0/8", "127.0.0.1")
     @patch("socket.getaddrinfo")
     def test_is_safe_hostname(self, mock_getaddrinfo: MagicMock) -> None: