Label Studio SSRF on Import Bypassing `SSRF_PROTECTION_ENABLED` Protections

@@ -403,6 +403,8 @@
 MAX_TIME_BETWEEN_ACTIVITY = int(get_env('MAX_TIME_BETWEEN_ACTIVITY', timedelta(days=5).total_seconds()))
 
 SSRF_PROTECTION_ENABLED = get_bool_env('SSRF_PROTECTION_ENABLED', False)
+USE_DEFAULT_BANNED_SUBNETS = get_bool_env('USE_DEFAULT_BANNED_SUBNETS', True)
+USER_ADDITIONAL_BANNED_SUBNETS = get_env_list('USER_ADDITIONAL_BANNED_SUBNETS', default=[])
 
 # user media files
 MEDIA_ROOT = os.path.join(BASE_DATA_DIR, 'media')

@@ -197,24 +197,61 @@ def validate_upload_url(url, block_local_urls=True):
 
 
 def validate_ip(ip: str) -> None:
-    """Checks if an IP is local/private.
+    """If settings.USE_DEFAULT_BANNED_SUBNETS is True, this function checks
+    if an IP is reserved for any of the reasons in
+    https://en.wikipedia.org/wiki/Reserved_IP_addresses
+    and raises an exception if so. Additionally, if settings.USER_ADDITIONAL_BANNED_SUBNETS
+    is set, it will also check against those subnets.
+
+    If settings.USE_DEFAULT_BANNED_SUBNETS is False, this function will only check
+    the IP against settings.USER_ADDITIONAL_BANNED_SUBNETS. Turning off the default
+    subnets is **risky** and should only be done if you know what you're doing.
 
     :param ip: IP address to be checked.
     """
 
-    if ip == '0.0.0.0':  # nosec
-        raise InvalidUploadUrlError
+    default_banned_subnets = [
+        '0.0.0.0/8',  # current network
+        '10.0.0.0/8',  # private network
+        '100.64.0.0/10',  # shared address space
+        '127.0.0.0/8',  # loopback
+        '169.254.0.0/16',  # link-local
+        '172.16.0.0/12',  # private network
+        '192.0.0.0/24',  # IETF protocol assignments
+        '192.0.2.0/24',  # TEST-NET-1
+        '192.88.99.0/24',  # Reserved, formerly ipv6 to ipv4 relay
+        '192.168.0.0/16',  # private network
+        '198.18.0.0/15',  # network interconnect device benchmark testing
+        '198.51.100.0/24',  # TEST-NET-2
+        '203.0.113.0/24',  # TEST-NET-3
+        '224.0.0.0/4',  # multicast
+        '233.252.0.0/24',  # MCAST-TEST-NET
+        '240.0.0.0/4',  # reserved for future use
+        '255.255.255.255/32',  # limited broadcast
+        '::/128',  # unspecified address
+        '::1/128',  # loopback
+        '::ffff:0:0/96',  # IPv4-mapped address
+        '::ffff:0:0:0/96',  # IPv4-translated address
+        '64:ff9b::/96',  # IPv4/IPv6 translation
+        '64:ff9b:1::/48',  # IPv4/IPv6 translation
+        '100::/64',  # discard prefix
+        '2001:0000::/32',  # Teredo tunneling
+        '2001:20::/28',  # ORCHIDv2
+        '2001:db8::/32',  # documentation
+        '2002::/16',  # 6to4
+        'fc00::/7',  # unique local
+        'fe80::/10',  # link-local
+        'ff00::/8',  # multicast
+    ]
 
-    local_subnets = [
-        '127.0.0.0/8',
-        '10.0.0.0/8',
-        '172.16.0.0/12',
-        '192.168.0.0/16',
+    banned_subnets = [
+        *(default_banned_subnets if settings.USE_DEFAULT_BANNED_SUBNETS else []),
+        *(settings.USER_ADDITIONAL_BANNED_SUBNETS or []),
     ]
 
-    for subnet in local_subnets:
+    for subnet in banned_subnets:
         if ipaddress.ip_address(ip) in ipaddress.ip_network(subnet):
-            raise InvalidUploadUrlError
+            raise InvalidUploadUrlError(f'URL resolves to a reserved network address (block: {subnet})')
 
 
 def ssrf_safe_get(url, *args, **kwargs):

@@ -70,7 +70,54 @@ def test_local_url_after_redirect(self, project, settings):
                 ValidationError
             ) as e:
                 load_tasks(request, project)
-            assert 'The provided URL was not valid.' in str(e.value)
+            assert 'URL resolves to a reserved network address (block: 127.0.0.0/8)' in str(e.value)
+
+        def test_user_specified_block(self, project, settings):
+            settings.SSRF_PROTECTION_ENABLED = True
+            settings.USER_ADDITIONAL_BANNED_SUBNETS = ['1.2.3.4']
+            request = MockedRequest(url='http://validurl.com')
+
+            # Mock the necessary parts of the response object
+            mock_response = Mock()
+            mock_response.raw._connection.sock.getpeername.return_value = ('1.2.3.4', 8080)
+
+            # Patch the requests.get call in the data_import.uploader module
+            with mock.patch('core.utils.io.requests.get', return_value=mock_response), pytest.raises(
+                ValidationError
+            ) as e:
+                load_tasks(request, project)
+            assert 'URL resolves to a reserved network address (block: 1.2.3.4)' in str(e.value)
+
+            mock_response.raw._connection.sock.getpeername.return_value = ('198.51.100.0', 8080)
+            with mock.patch('core.utils.io.requests.get', return_value=mock_response), pytest.raises(
+                ValidationError
+            ) as e:
+                load_tasks(request, project)
+            assert 'URL resolves to a reserved network address (block: 198.51.100.0/24)' in str(e.value)
+
+        def test_user_specified_block_without_default(self, project, settings):
+            settings.SSRF_PROTECTION_ENABLED = True
+            settings.USER_ADDITIONAL_BANNED_SUBNETS = ['1.2.3.4']
+            settings.USE_DEFAULT_BANNED_SUBNETS = False
+            request = MockedRequest(url='http://validurl.com')
+
+            # Mock the necessary parts of the response object
+            mock_response = Mock()
+            mock_response.raw._connection.sock.getpeername.return_value = ('1.2.3.4', 8080)
+
+            # Patch the requests.get call in the data_import.uploader module
+            with mock.patch('core.utils.io.requests.get', return_value=mock_response), pytest.raises(
+                ValidationError
+            ) as e:
+                load_tasks(request, project)
+            assert 'URL resolves to a reserved network address (block: 1.2.3.4)' in str(e.value)
+
+            mock_response.raw._connection.sock.getpeername.return_value = ('198.51.100.0', 8080)
+            with mock.patch('core.utils.io.requests.get', return_value=mock_response), pytest.raises(
+                ValidationError
+            ) as e:
+                load_tasks(request, project)
+            assert "'Mock' object is not subscriptable" in str(e.value)  # validate ip did not raise exception
 
 
 class TestTasksFileChecks: