🛡️ Sentinel: [CRITICAL] Fix SSRF via DNS Rebinding

.jules/sentinel.md

@@ -0,0 +1,4 @@
+## 2025-02-18 - [Preventing SSRF via DNS Rebinding with Post-Connect Verification]
+**Vulnerability:** Validation of URLs checks the resolved IP, but `httpx` re-resolves the domain during connection, allowing a TOCTOU (Time-of-Check Time-of-Use) attack where the IP changes to a private one (DNS Rebinding).
+**Learning:** Standard URL validation is insufficient against sophisticated attackers controlling DNS. Checking the IP *after* the connection is established (using `stream.get_extra_info("server_addr")`) is a robust defense because it verifies the actual endpoint used.
+**Prevention:** Always verify the peer/server address after connection establishment when making requests to untrusted or user-controlled URLs, especially when preventing access to internal resources.

.python-version

@@ -1 +1 @@
-3.13
+3.13

main.py

@@ -364,7 +364,7 @@
                     # res is (family, type, proto, canonname, sockaddr)
                     # sockaddr is (address, port) for AF_INET/AF_INET6
                     ip_str = res[4][0]
                     ip = ipaddress.ip_address(ip_str)
                     if not ip.is_global or ip.is_multicast:
                         log.warning(
                             f"Skipping unsafe URL (domain {hostname} resolves to non-global/multicast IP {ip}): {sanitize_for_log(url)}"
@@ -526,17 +526,36 @@
             time.sleep(wait_time)
 
 
 def _gh_get(url: str) -> Dict:
     # First check: Quick check without holding lock for long
     with _cache_lock:
         if url in _cache:
             return _cache[url]
 
     # Fetch data if not cached
     # Explicitly let HTTPError propagate (no need to catch just to re-raise)
     with _gh.stream("GET", url) as r:
         r.raise_for_status()
 
+        # SECURITY FIX: Post-connect SSRF verification (DNS Rebinding protection)
+        # Verify the actual IP address we connected to, to prevent DNS rebinding attacks.
+        stream = r.extensions.get("network_stream")
+        if stream:
+            # "server_addr" returns (ip, port) for TCP connections
+            server_addr = stream.get_extra_info("server_addr")
+            if server_addr and len(server_addr) >= 1:
+                ip_str = server_addr[0]
+                try:
+                    ip = ipaddress.ip_address(ip_str)
+                except ValueError:
+                    # server_addr[0] might not be an IP string (unlikely for TCP)
+                    ip = None
+
+                if ip and (not ip.is_global or ip.is_multicast):
+                    raise ValueError(
+                        f"Security Alert: Domain resolved to private IP {ip_str} (DNS Rebinding protection)"
+                    )
+
         # 1. Check Content-Length header if present
         cl = r.headers.get("Content-Length")
         if cl:

tests/test_ssrf_fix.py

@@ -0,0 +1,70 @@
+import pytest
+import httpx
+from unittest.mock import MagicMock
+import main
+
+@pytest.fixture(autouse=True)
+def clear_cache():
+    main._cache.clear()
+
+def test_gh_get_blocks_private_ip_after_connect(monkeypatch):
+    """
+    Test that _gh_get raises ValueError if the connection was established to a private IP.
+    This simulates a DNS Rebinding attack where the initial check passes but the connection goes to private IP.
+    """
+
+    # Mock response stream with private IP
+    mock_stream = MagicMock()
+    mock_stream.get_extra_info.return_value = ('127.0.0.1', 443)
+
+    mock_response = MagicMock(spec=httpx.Response)
+    mock_response.extensions = {"network_stream": mock_stream}
+    mock_response.headers = {}
+    mock_response.iter_bytes.return_value = [b'{}']
+    mock_response.raise_for_status.return_value = None
+
+    # Context manager mock for stream()
+    mock_context = MagicMock()
+    mock_context.__enter__.return_value = mock_response
+    mock_context.__exit__.return_value = None
+
+    # Mock _gh.stream
+    mock_gh = MagicMock()
+    mock_gh.stream.return_value = mock_context
+
+    monkeypatch.setattr(main, "_gh", mock_gh)
+
+    # We expect ValueError because of our security fix
+    # Before the fix, this test will FAIL (it won't raise)
+    with pytest.raises(ValueError, match="Security Alert: Domain resolved to private IP"):
+        main._gh_get("https://example.com/config.json")
+
+def test_gh_get_allows_public_ip_after_connect(monkeypatch):
+    """
+    Test that _gh_get allows connection if established to a public IP.
+    """
+
+    # Mock response stream with public IP
+    mock_stream = MagicMock()
+    mock_stream.get_extra_info.return_value = ('8.8.8.8', 443)
+
+    mock_response = MagicMock(spec=httpx.Response)
+    mock_response.extensions = {"network_stream": mock_stream}
+    mock_response.headers = {}
+    mock_response.iter_bytes.return_value = [b'{"valid": "json"}']
+    mock_response.raise_for_status.return_value = None
+
+    # Context manager mock
+    mock_context = MagicMock()
+    mock_context.__enter__.return_value = mock_response
+    mock_context.__exit__.return_value = None
+
+    # Mock _gh.stream
+    mock_gh = MagicMock()
+    mock_gh.stream.return_value = mock_context
+
+    monkeypatch.setattr(main, "_gh", mock_gh)
+
+    # Should not raise
+    result = main._gh_get("https://example.com/config.json")
+    assert result == {"valid": "json"}