feat: fix SSRF bypass vulnerability with IPv6 support

- Update is_safe_url() to use getaddrinfo() instead of gethostbyname()
- Resolves ALL IPs (IPv4 + IPv6) to prevent dual-stack bypass
- Blocks requests if ANY resolved IP is private/loopback/reserved
- Add comprehensive SSRF tests including dual-stack scenario
- Update .gitignore with __pycache__

Author: GhostTypes

.gitignore

@@ -1 +1,3 @@
 .aider*
+__pycache__/
+*.pyc

server.py

@@ -1,7 +1,7 @@
 import cloudscraper
-import time
-import socket
 import ipaddress
+import socket
+import time
 
 from urllib.parse import unquote, urlparse
 from flask import Flask, request, Response
@@ -54,28 +54,40 @@ def generate_origin_and_ref(url, headers):
 def is_safe_url(url):
     """
     Validates URL to prevent SSRF attacks by blocking local/private IP ranges.
+    Uses getaddrinfo to resolve ALL IP addresses (IPv4 and IPv6) to prevent
+    bypass via dual-stack hostnames that resolve to both public and private IPs.
+
+    Returns:
+        tuple: (is_safe: bool, error_message: str | None)
     """
     try:
         parsed = urlparse(url)
         if parsed.scheme not in ('http', 'https'):
             return False, "Only HTTP/HTTPS protocols are allowed"
 
         hostname = parsed.hostname
+
         if not hostname:
             return False, "Invalid hostname"
 
         try:
-            # Resolve hostname to IP to check against blocklist
-            ip_str = socket.gethostbyname(hostname)
-            ip = ipaddress.ip_address(ip_str)
+            # Resolve hostname to IP(s) to check against blocklist.
+            # We use getaddrinfo to get all resolved IPs (IPv4 and IPv6) to prevent evasion
+            # where a hostname resolves to both a safe IP and a private IP.
+            addr_info = socket.getaddrinfo(hostname, None)
+
+            for family, _, _, _, sockaddr in addr_info:
+                # sockaddr is (address, port) for AF_INET and (address, port, flow info, scope id) for AF_INET6
+                ip_str = sockaddr[0]
+                ip = ipaddress.ip_address(ip_str)
+
+                if ip.is_loopback or ip.is_private or ip.is_reserved or ip.is_multicast or ip.is_unspecified:
+                    return False, "Access to private/local network is forbidden"
+
         except (socket.gaierror, ValueError):
-            # If we can't resolve it, it might be safer to block or allow if it's external.
-            # For security, fail closed if resolution fails.
+            # If we can't resolve it, fail closed.
             return False, "Could not resolve hostname or invalid IP"
 
-        if ip.is_loopback or ip.is_private or ip.is_reserved or ip.is_multicast or ip.is_unspecified:
-            return False, "Access to private/local network is forbidden"
-
         return True, None
     except Exception:
         return False, "Invalid URL format"
@@ -154,9 +166,9 @@ def get_proxy_request_headers(req, url):
     headers = get_headers()
     headers['Accept-Encoding'] = 'gzip, deflate, br'
 
-    for header in req.headers:
-        if header[0].lower() not in ['host', 'connection', 'content-length']:
-            headers[header[0]] = header[1]
+    for key, value in req.headers.items():
+        if key.lower() not in ['host', 'connection', 'content-length']:
+            headers[key] = value
     headers = generate_origin_and_ref(url, headers)
     return headers
 
@@ -167,16 +179,16 @@ def handle_proxy(url):
     if request.method == 'GET':
         full_url = get_proxy_request_url(request, url)  # parse request url
 
-        # Sentinel: SSRF Protection
+        # SSRF protection check
         is_safe, error_msg = is_safe_url(full_url)
         if not is_safe:
-            return {'error': error_msg}, 400
+            print(f"SSRF blocked: {full_url} - {error_msg}")
+            return {'error': error_msg}, 403
 
         headers = get_proxy_request_headers(request, url)  # generate headers for the request
 
         try:
             start = time.time()
-            # Sentinel: Added timeout to prevent hanging
             response = scraper.get(full_url, headers=headers, timeout=30)
             end = time.time()
             elapsed = end - start
@@ -187,7 +199,7 @@ def handle_proxy(url):
 
         except Exception as e:
             print(f"Proxy Request Error: {str(e)}")
-            # Sentinel: Don't leak stack traces or internal details
+            # Don't leak stack traces or internal details
             return {'error': "Proxy request failed. Check server logs for details."}, 500
 
 

tests/test_is_safe_url.py

@@ -0,0 +1,88 @@
+import unittest
+from unittest.mock import patch
+from server import is_safe_url
+import socket
+
+
+class TestIsSafeUrl(unittest.TestCase):
+
+    @patch('server.socket.getaddrinfo')
+    def test_public_ipv4(self, mock_getaddrinfo):
+        mock_getaddrinfo.return_value = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('8.8.8.8', 80))
+        ]
+        is_safe, msg = is_safe_url('http://example.com')
+        self.assertTrue(is_safe)
+        self.assertIsNone(msg)
+
+    @patch('server.socket.getaddrinfo')
+    def test_private_ipv4(self, mock_getaddrinfo):
+        mock_getaddrinfo.return_value = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('127.0.0.1', 80))
+        ]
+        is_safe, msg = is_safe_url('http://example.com')
+        self.assertFalse(is_safe)
+        self.assertIn("Access to private/local network is forbidden", msg)
+
+    @patch('server.socket.getaddrinfo')
+    def test_dual_stack_ipv6_private(self, mock_getaddrinfo):
+        # Scenario: DNS has public IPv4 (8.8.8.8) and private IPv6 (::1)
+        mock_getaddrinfo.return_value = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('8.8.8.8', 80)),
+            (socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('::1', 80, 0, 0))
+        ]
+
+        is_safe, msg = is_safe_url('http://example.com')
+
+        # We expect this to be False (Blocked) because of the private IPv6 address.
+        self.assertFalse(is_safe, "Should block if ANY resolved IP is private (IPv6 loopback)")
+
+    @patch('server.socket.getaddrinfo')
+    def test_private_ipv4_192_168(self, mock_getaddrinfo):
+        mock_getaddrinfo.return_value = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('192.168.1.1', 80))
+        ]
+        is_safe, msg = is_safe_url('http://example.com')
+        self.assertFalse(is_safe)
+        self.assertIn("Access to private/local network is forbidden", msg)
+
+    @patch('server.socket.getaddrinfo')
+    def test_private_ipv4_10_0_0_0(self, mock_getaddrinfo):
+        mock_getaddrinfo.return_value = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.1', 80))
+        ]
+        is_safe, msg = is_safe_url('http://example.com')
+        self.assertFalse(is_safe)
+        self.assertIn("Access to private/local network is forbidden", msg)
+
+    @patch('server.socket.getaddrinfo')
+    def test_ipv6_loopback(self, mock_getaddrinfo):
+        mock_getaddrinfo.return_value = [
+            (socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('::1', 80, 0, 0))
+        ]
+        is_safe, msg = is_safe_url('http://example.com')
+        self.assertFalse(is_safe)
+        self.assertIn("Access to private/local network is forbidden", msg)
+
+    @patch('server.socket.getaddrinfo')
+    def test_ipv6_private(self, mock_getaddrinfo):
+        mock_getaddrinfo.return_value = [
+            (socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('fe80::1', 80, 0, 0))
+        ]
+        is_safe, msg = is_safe_url('http://example.com')
+        self.assertFalse(is_safe)
+        self.assertIn("Access to private/local network is forbidden", msg)
+
+    def test_invalid_url(self):
+        is_safe, msg = is_safe_url('not-a-url')
+        self.assertFalse(is_safe)
+        self.assertIn("Only HTTP/HTTPS protocols are allowed", msg)
+
+    def test_invalid_hostname(self):
+        is_safe, msg = is_safe_url('http://')
+        self.assertFalse(is_safe)
+        self.assertIn("Invalid hostname", msg)
+
+
+if __name__ == '__main__':
+    unittest.main()