CVE-2023-47116: Server-Side Request Forgery (SSRF) (CWE-918)
value: Label Studio is a popular open source data labeling tool. The vulnerability affects all versions of Label Studio prior to 1.11.0 and was tested on version 1.8.2. Label Studio's SSRF protections that can be enabled by setting the `SSRF_PROTECTION_ENABLED` environment variable can be bypassed to access internal web servers. This is because the current SSRF validation is done by executing a single DNS lookup to verify that the IP address is not in an excluded subnet range. This protection can be bypassed by either using HTTP redirection or performing a DNS rebinding attack.
CWE description: The web server receives a URL or similar request from an upstream component and retrieves the contents of this URL, but it does not sufficiently ensure that the request is being sent to the expected destination.
CVSS v3 base score: 5.3
Published: 2024-01-31T17:15Z
Repository: https://github.com/HumanSignal/label-studio
Language: Python
Files changed: label_studio/core/settings/base.py, label_studio/core/utils/io.py, label_studio/tests/data_import/test_uploader.py
Commit date: 2024-01-22 16:44:30 -0500
Commit message: fix: LEAP-396: More exhaustative IP validation for SSRF defenses, plus user configurability (#5316)

* fix: LEAP-396: More exhaustative IP validation for SSRF, plus user configurability

* unbotch

* fix test

* more testing

* fix false positive finding from Bandit
Diff: diff --git a/label_studio/core/settings/base.py b/label_studio/core/settings/base.py
index 6d88089a7..c3ad75f98 100644
--- a/label_studio/core/settings/base.py
+++ b/label_studio/core/settings/base.py
@@ -401,10 +401,12 @@ INACTIVITY_SESSION_TIMEOUT_ENABLED = bool(int(get_env('INACTIVITY_SESSION_TIMEOU
 MAX_SESSION_AGE = int(get_env('MAX_SESSION_AGE', timedelta(days=14).total_seconds()))
 # The most time that can elapse between activity with the server before the user is logged out
 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')
 os.makedirs(MEDIA_ROOT, exist_ok=True)
 MEDIA_URL = '/data/'
diff --git a/label_studio/core/utils/io.py b/label_studio/core/utils/io.py
index 3d30643a7..311400199 100644
--- a/label_studio/core/utils/io.py
+++ b/label_studio/core/utils/io.py
@@ -195,28 +195,65 @@ def validate_upload_url(url, block_local_urls=True):
     if block_local_urls:
         validate_ip(ip)
 
 
 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):
     validate_upload_url(url, block_local_urls=settings.SSRF_PROTECTION_ENABLED)
     # Reason for #nosec: url has been validated as SSRF safe by the
diff --git a/label_studio/tests/data_import/test_uploader.py b/label_studio/tests/data_import/test_uploader.py
index 9946b995c..7c5c9946d 100644
--- a/label_studio/tests/data_import/test_uploader.py
+++ b/label_studio/tests/data_import/test_uploader.py
@@ -68,11 +68,58 @@ class TestUploader:
             # 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 '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:
     @pytest.mark.parametrize('value', (0, settings.TASKS_MAX_FILE_SIZE - 1))
     def test_check_tasks_max_file_size_does_not_raise_for_correct_value(self, value):

Vulnerable code:     """Checks if an IP is local/private.
    if ip == '0.0.0.0':  # nosec
        raise InvalidUploadUrlError
    local_subnets = [
        '127.0.0.0/8',
        '10.0.0.0/8',
        '172.16.0.0/12',
        '192.168.0.0/16',
    for subnet in local_subnets:
            raise InvalidUploadUrlError
            assert 'The provided URL was not valid.' in str(e.value)
Fixed code: USE_DEFAULT_BANNED_SUBNETS = get_bool_env('USE_DEFAULT_BANNED_SUBNETS', True)
USER_ADDITIONAL_BANNED_SUBNETS = get_env_list('USER_ADDITIONAL_BANNED_SUBNETS', default=[])
    """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.
    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
    ]
    banned_subnets = [
        *(default_banned_subnets if settings.USE_DEFAULT_BANNED_SUBNETS else []),
        *(settings.USER_ADDITIONAL_BANNED_SUBNETS or []),
    for subnet in banned_subnets:
            raise InvalidUploadUrlError(f'URL resolves to a reserved network address (block: {subnet})')
            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