CKAN - Stored Cross-Site Scripting (XSS) via SVG File Upload

@@ -13,7 +13,7 @@
 import ckan.lib.munge as munge
 import ckan.logic as logic
 import ckan.plugins as plugins
-from ckan.common import config
+from ckan.common import config, aslist
 
 ALLOWED_UPLOAD_TYPES = (cgi.FieldStorage, FlaskFileStorage)
 MB = 1 << 20
@@ -192,6 +192,26 @@ def upload(self, max_size=2):
             except OSError:
                 pass
 
+    def verify_type(self):
+        if not self.filename:
+            return
+
+        actual = magic.from_buffer(self.upload_file.read(1024), mime=True)
+        self.upload_file.seek(0, os.SEEK_SET)
+
+        err = {self.file_field: [f"Unsupported upload type: {actual}"]}
+
+        mimetypes = aslist(
+            config.get(f"ckan.upload.{self.object_type}.mimetypes"))
+        if mimetypes and actual not in mimetypes:
+            raise logic.ValidationError(err)
+
+        type_ = actual.split("/")[0]
+        types = aslist(
+            config.get(f"ckan.upload.{self.object_type}.types"))
+        if types and type_ not in types:
+            raise logic.ValidationError(err)
+
 
 class ResourceUpload(object):
     def __init__(self, resource):

@@ -762,6 +762,9 @@ def _group_or_org_create(context, data_dict, is_org=False):
     }
     logic.get_action('activity_create')(activity_create_context, activity_dict)
 
+    if hasattr(upload, "verify_type"):
+        upload.verify_type()
+
     upload.upload(uploader.get_max_image_size())
 
     if not context.get('defer_commit'):
@@ -1012,6 +1015,9 @@ def user_create(context, data_dict):
     }
     logic.get_action('activity_create')(activity_create_context, activity_dict)
 
+    if hasattr(upload, "verify_type"):
+        upload.verify_type()
+
     upload.upload(uploader.get_max_image_size())
 
     if not context.get('defer_commit'):

@@ -1666,7 +1666,7 @@ def test_ignored_on_create_if_non_sysadmin(self):
 
 @pytest.mark.usefixtures("clean_db")
 class TestUserImageUrl(object):
-    def test_upload_picture(self):
+    def test_external_picture(self):
 
         params = {
             "name": "test_user",
@@ -1682,6 +1682,56 @@ def test_upload_picture(self):
             user_dict["image_display_url"] == "https://example.com/mypic.png"
         )
 
+    def test_upload_non_picture_works_without_extra_config(
+            self, create_with_upload, faker):
+        params = {
+            "name": faker.user_name(),
+            "email": faker.email(),
+            "password": "12345678",
+            "action": "user_create",
+            "upload_field_name": "image_upload",
+        }
+        assert create_with_upload("hello world", "file.txt", **params)
+
+    @pytest.mark.ckan_config("ckan.upload.user.types", "image")
+    def test_upload_non_picture(self, create_with_upload, faker):
+        params = {
+            "name": faker.user_name(),
+            "email": faker.email(),
+            "password": "12345678",
+            "action": "user_create",
+            "upload_field_name": "image_upload",
+        }
+        with pytest.raises(
+                logic.ValidationError, match="Unsupported upload type"):
+            create_with_upload("hello world", "file.txt", **params)
+
+    @pytest.mark.ckan_config("ckan.upload.user.types", "image")
+    def test_upload_non_picture_with_png_extension(
+            self, create_with_upload, faker):
+        params = {
+            "name": faker.user_name(),
+            "email": faker.email(),
+            "password": "12345678",
+            "action": "user_create",
+            "upload_field_name": "image_upload",
+        }
+        with pytest.raises(
+                logic.ValidationError, match="Unsupported upload type"):
+            create_with_upload("hello world", "file.png", **params)
+
+    @pytest.mark.ckan_config("ckan.upload.user.types", "image")
+    def test_upload_picture(self, create_with_upload, faker):
+        params = {
+            "name": faker.user_name(),
+            "email": faker.email(),
+            "password": "12345678",
+            "action": "user_create",
+            "upload_field_name": "image_upload",
+        }
+        assert create_with_upload(faker.image(), "file.png", **params)
+
+
 
 class TestVocabularyCreate(object):
     @pytest.mark.usefixtures("clean_db")

@@ -4,12 +4,14 @@ beautifulsoup4==4.9.3
 cookiecutter==1.7.3
 coveralls   #Let Unpinned - Requires latest coveralls
 docutils==0.16
+Faker==9.3.1
 factory-boy==3.2.0
 flask-debugtoolbar==0.11.0
 freezegun==1.1.0
 ipdb==0.13.7
 pip-tools==5.1.2
 pycodestyle==2.5.0
+Pillow==8.4.0
 responses==0.13.3
 sphinx-rtd-theme==0.4.3
 sphinx==1.8.5