CVE-2022-31188: Server-Side Request Forgery (SSRF) (CWE-918)
value: CVAT is an opensource interactive video and image annotation tool for computer vision. Versions prior to 2.0.0 were found to be subject to a Server-side request forgery (SSRF) vulnerability. Validation has been added to urls used in the affected code path in version 2.0.0. Users are advised to upgrade. There are no known workarounds for this issue.
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: 9.8
Published: 2022-08-01T20:15Z
Repository: https://github.com/cvat-ai/cvat
Language: Python
Files changed: CHANGELOG.md, cvat/apps/engine/task.py, cvat/requirements/base.txt
Commit date: 2022-02-28 09:52:36 +0300
Commit message: Added validation for URLs which used as remote data source (#4387)


Diff: diff --git a/CHANGELOG.md b/CHANGELOG.md
index b4efddcb..51d31f72 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -79,10 +79,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Kibana requests are not proxied due to django-revproxy incompatibility with Django >3.2.x (<https://github.com/openvinotoolkit/cvat/issues/4085>)
 - Content type for getting frame with tasks/{id}/data/ endpoint (<https://github.com/openvinotoolkit/cvat/pull/4333>)
 
 ### Security
 - Updated ELK to 6.8.23 which uses log4j 2.17.1 (<https://github.com/openvinotoolkit/cvat/pull/4206>)
+- Added validation for URLs which used as remote data source (<https://github.com/openvinotoolkit/cvat/pull/4387>)
 
 ## \[1.7.0] - 2021-11-15
 
 ### Added
 
diff --git a/cvat/apps/engine/task.py b/cvat/apps/engine/task.py
index 89079929..33464493 100644
--- a/cvat/apps/engine/task.py
+++ b/cvat/apps/engine/task.py
@@ -4,18 +4,21 @@
 # SPDX-License-Identifier: MIT
 
 import itertools
 import os
 import sys
+from rest_framework.serializers import ValidationError
 import rq
 import re
 import shutil
 from distutils.dir_util import copy_tree
 from traceback import print_exception
 from urllib import parse as urlparse
 from urllib import request as urlrequest
 import requests
+import ipaddress
+import dns.resolver
 import django_rq
 
 from django.conf import settings
 from django.db import transaction
 
@@ -201,17 +204,57 @@ def _validate_manifest(manifests, root_dir):
         if is_manifest(full_manifest_path):
             return manifests[0]
         raise Exception('Invalid manifest was uploaded')
     return None
 
+def _validate_url(url):
+    def _validate_ip_address(ip_address):
+        if not ip_address.is_global:
+            raise ValidationError('Non public IP address \'{}\' is provided!'.format(ip_address))
+
+    ALLOWED_SCHEMES = ['http', 'https']
+
+    parsed_url = urlparse.urlparse(url)
+
+    if parsed_url.scheme not in ALLOWED_SCHEMES:
+        raise ValueError('Unsupported URL sheme: {}. Only http and https are supported'.format(parsed_url.scheme))
+
+    try:
+        ip_address = ipaddress.ip_address(parsed_url.hostname)
+        _validate_ip_address(ip_address)
+    except ValueError as _:
+        ip_v4_records = None
+        ip_v6_records = None
+        try:
+            ip_v4_records = dns.resolver.query(parsed_url.hostname, 'A')
+            for record in ip_v4_records:
+                _validate_ip_address(ipaddress.ip_address(record.to_text()))
+        except ValidationError:
+            raise
+        except Exception as e:
+            slogger.glob.info('Cannot get A record for domain \'{}\': {}'.format(parsed_url.hostname, e))
+
+        try:
+            ip_v6_records = dns.resolver.query(parsed_url.hostname, 'AAAA')
+            for record in ip_v6_records:
+                _validate_ip_address(ipaddress.ip_address(record.to_text()))
+        except ValidationError:
+            raise
+        except Exception as e:
+            slogger.glob.info('Cannot get AAAA record for domain \'{}\': {}'.format(parsed_url.hostname, e))
+
+        if not ip_v4_records and not ip_v6_records:
+            raise ValidationError('Cannot resolve IP address for domain \'{}\''.format(parsed_url.hostname))
+
 def _download_data(urls, upload_dir):
     job = rq.get_current_job()
     local_files = {}
     for url in urls:
         name = os.path.basename(urlrequest.url2pathname(urlparse.urlparse(url).path))
         if name in local_files:
             raise Exception("filename collision: {}".format(name))
+        _validate_url(url)
         slogger.glob.info("Downloading: {}".format(url))
         job.meta['status'] = '{} is being downloaded..'.format(url)
         job.save_meta()
 
         response = requests.get(url, stream=True)
diff --git a/cvat/requirements/base.txt b/cvat/requirements/base.txt
index d01f0f2c..e2a98b2e 100644
--- a/cvat/requirements/base.txt
+++ b/cvat/requirements/base.txt
@@ -50,5 +50,6 @@ google-cloud-storage==1.42.0
 # opencv-headless instead of regular opencv, to actually run setup script
 datumaro==0.2.0 --no-binary=datumaro
 urllib3>=1.26.5 # not directly required, pinned by Snyk to avoid a vulnerability
 natsort==8.0.0
 mistune>=2.0.1 # not directly required, pinned by Snyk to avoid a vulnerability
+dnspython==2.2.0

Fixed code: - Added validation for URLs which used as remote data source (<https://github.com/openvinotoolkit/cvat/pull/4387>)
from rest_framework.serializers import ValidationError
import ipaddress
import dns.resolver
def _validate_url(url):
    def _validate_ip_address(ip_address):
        if not ip_address.is_global:
            raise ValidationError('Non public IP address \'{}\' is provided!'.format(ip_address))

    ALLOWED_SCHEMES = ['http', 'https']

    parsed_url = urlparse.urlparse(url)

    if parsed_url.scheme not in ALLOWED_SCHEMES:
        raise ValueError('Unsupported URL sheme: {}. Only http and https are supported'.format(parsed_url.scheme))

    try:
        ip_address = ipaddress.ip_address(parsed_url.hostname)
        _validate_ip_address(ip_address)
    except ValueError as _:
        ip_v4_records = None
        ip_v6_records = None
        try:
            ip_v4_records = dns.resolver.query(parsed_url.hostname, 'A')
            for record in ip_v4_records:
                _validate_ip_address(ipaddress.ip_address(record.to_text()))
        except ValidationError:
            raise
        except Exception as e:
            slogger.glob.info('Cannot get A record for domain \'{}\': {}'.format(parsed_url.hostname, e))

        try:
            ip_v6_records = dns.resolver.query(parsed_url.hostname, 'AAAA')
            for record in ip_v6_records:
                _validate_ip_address(ipaddress.ip_address(record.to_text()))
        except ValidationError:
            raise
        except Exception as e:
            slogger.glob.info('Cannot get AAAA record for domain \'{}\': {}'.format(parsed_url.hostname, e))

        if not ip_v4_records and not ip_v6_records:
            raise ValidationError('Cannot resolve IP address for domain \'{}\''.format(parsed_url.hostname))

        _validate_url(url)
dnspython==2.2.0