Block non-image uploads

- Added `IAuthFunctions` to `plugin.py`
- Added chained auth functions for `user_update`, `organization_update` and `group_update` to check `request.files` mimetype

ckanext/fortify/logic/auth/update.py

@@ -0,0 +1,38 @@
+import ckan.plugins.toolkit as toolkit
+import logging
+
+from ckan.common import _, request
+
+log = logging.getLogger(__name__)
+
+
+def disallow_non_image_uploads(next_auth, context, data_dict):
+    try:
+        if request.files:
+            files_dict = dict(request.files)
+            image_upload = files_dict.get('image_upload')
+            if image_upload and image_upload.mimetype and 'image' not in image_upload.mimetype:
+                log.error('User {0} upload attempt blocked - file: {1}'.format(
+                    context['user'],
+                    image_upload
+                ))
+                return {'success': False, 'msg': _('Invalid filetype')}
+    except Exception as e:
+        log.error(str(e))
+
+    return next_auth(context, data_dict)
+
+
+@toolkit.chained_auth_function
+def user_update(next_auth, context, data_dict):
+    return disallow_non_image_uploads(next_auth, context, data_dict)
+
+
+@toolkit.chained_auth_function
+def organization_update(next_auth, context, data_dict):
+    return disallow_non_image_uploads(next_auth, context, data_dict)
+
+
+@toolkit.chained_auth_function
+def group_update(next_auth, context, data_dict):
+    return disallow_non_image_uploads(next_auth, context, data_dict)

ckanext/fortify/plugin.py

@@ -4,6 +4,7 @@
 import logging
 
 from ckanext.fortify import helpers, validators, blueprint
+from ckanext.fortify.logic.auth import update as auth_update
 from ckan.lib.uploader import ALLOWED_UPLOAD_TYPES
 
 config = toolkit.config
@@ -37,6 +38,8 @@ def create(self, entity):
 
     if asbool(config.get('ckan.fortify.block_html_resource_uploads', False)):
         plugins.implements(plugins.IUploader, inherit=True)
+        plugins.implements(plugins.IAuthFunctions)
+
         # IUploader
 
         def get_resource_uploader(self, data_dict):
@@ -47,6 +50,17 @@ def get_resource_uploader(self, data_dict):
                 # Returning None will make sure it uses the CKAN default uploader ResourceUpload
                 return None
 
+        # IAuthFunctions
+
+        def get_auth_functions(self):
+            return {
+                'user_update': auth_update.user_update,
+                'organization_update': auth_update.organization_update,
+                'group_update': auth_update.group_update,
+            }
+
+
+
     if asbool(config.get('ckan.fortify.enable_anti_csrf_tokens', False)) \
             or asbool(config.get('ckan.fortify.enable_password_policy', False)) \
             or asbool(config.get('ckan.fortify.force_html_resource_downloads', False)):