[#3196] CORS headers added to Flask responses.

The method to set CORS headers moved to `views` module and CKAN core
refactored to import and use it.

Original commit by @brew (77a436e)

ckan/config/middleware/flask_app.py

@@ -22,7 +22,9 @@
 import ckan.lib.app_globals as app_globals
 from ckan.plugins import PluginImplementations
 from ckan.plugins.interfaces import IBlueprint
-from ckan.views import identify_user
+from ckan.views import (identify_user,
+                        set_cors_headers_for_response,
+                        )
 
 
 import logging
@@ -91,6 +93,12 @@ def ckan_before_request():
         # Sets g.user and g.userobj
         identify_user()
 
+    @app.after_request
+    def ckan_after_request(response):
+        # Set CORS headers if necessary
+        set_cors_headers_for_response(response)
+        return response
+
     # Template context processors
     @app.context_processor
     def helper_functions():

ckan/lib/base.py

@@ -7,7 +7,6 @@
 import logging
 import time
 
-from paste.deploy.converters import asbool
 from pylons import cache, session
 from pylons.controllers import WSGIController
 from pylons.controllers.util import abort as _abort
@@ -26,7 +25,9 @@
 import ckan.plugins as p
 import ckan.model as model
 
-from ckan.views import identify_user
+from ckan.views import (identify_user,
+                        set_cors_headers_for_response,
+                        )
 
 # These imports are for legacy usages and will be removed soon these should
 # be imported directly from ckan.common for internal ckan code and via the
@@ -238,36 +239,13 @@ def __call__(self, environ, start_response):
         return res
 
     def __after__(self, action, **params):
-        # Do we have CORS settings in config?
-        if config.get('ckan.cors.origin_allow_all') \
-                and request.headers.get('Origin'):
-            self._set_cors()
+
+        set_cors_headers_for_response(response)
+
         r_time = time.time() - c.__timer
         url = request.environ['CKAN_CURRENT_URL'].split('?')[0]
         log.info(' %s render time %.3f seconds' % (url, r_time))
 
-    def _set_cors(self):
-        '''
-        Set up Access Control Allow headers if either origin_allow_all is
-        True, or the request Origin is in the origin_whitelist.
-        '''
-        cors_origin_allowed = None
-
-        if asbool(config.get('ckan.cors.origin_allow_all')):
-            cors_origin_allowed = "*"
-        elif config.get('ckan.cors.origin_whitelist') and \
-                request.headers.get('Origin') \
-                in config['ckan.cors.origin_whitelist'].split():
-            # set var to the origin to allow it.
-            cors_origin_allowed = request.headers.get('Origin')
-        if cors_origin_allowed is not None:
-            response.headers['Access-Control-Allow-Origin'] = \
-                cors_origin_allowed
-            response.headers['Access-Control-Allow-Methods'] = \
-                "POST, PUT, GET, DELETE, OPTIONS"
-            response.headers['Access-Control-Allow-Headers'] = \
-                "X-CKAN-API-KEY, Authorization, Content-Type"
-
 
 # Include the '_' function in the public names
 __all__ = [__name for __name in locals().keys() if not __name.startswith('_')

ckan/tests/lib/test_base.py

@@ -3,6 +3,7 @@
 from nose import tools as nose_tools
 
 import ckan.tests.helpers as helpers
+import ckan.plugins as p
 
 
 class TestRenderSnippet(helpers.FunctionalTestBase):
@@ -160,3 +161,153 @@ def test_cors_config_origin_allow_all_false_with_whitelist_not_containing_origin
         assert 'Access-Control-Allow-Origin' not in response_headers
         assert 'Access-Control-Allow-Methods' not in response_headers
         assert 'Access-Control-Allow-Headers' not in response_headers
+
+
+class TestCORSFlask(helpers.FunctionalTestBase):
+
+    @classmethod
+    def setup_class(cls):
+        super(TestCORSFlask, cls).setup_class()
+        cls.app = cls._get_test_app()
+        flask_app = cls.app.flask_app
+
+        if not p.plugin_loaded('test_routing_plugin'):
+            p.load('test_routing_plugin')
+            plugin = p.get_plugin('test_routing_plugin')
+            flask_app.register_blueprint(plugin.get_blueprint(),
+                                         prioritise_rules=True)
+
+    @classmethod
+    def teardown_class(cls):
+        super(TestCORSFlask, cls).teardown_class()
+        p.unload('test_routing_plugin')
+
+    def test_options(self):
+        response = self.app.options(url='/simple_flask', status=200)
+        assert len(str(response.body)) == 0, 'OPTIONS must return no content'
+
+    def test_cors_config_no_cors(self):
+        '''
+        No ckan.cors settings in config, so no Access-Control-Allow headers in
+        response.
+        '''
+        response = self.app.get('/simple_flask')
+        response_headers = dict(response.headers)
+
+        assert 'Access-Control-Allow-Origin' not in response_headers
+        assert 'Access-Control-Allow-Methods' not in response_headers
+        assert 'Access-Control-Allow-Headers' not in response_headers
+
+    def test_cors_config_no_cors_with_origin(self):
+        '''
+        No ckan.cors settings in config, so no Access-Control-Allow headers in
+        response, even with origin header in request.
+        '''
+        request_headers = {'Origin': 'http://thirdpartyrequests.org'}
+        response = self.app.get('/simple_flask', headers=request_headers)
+        response_headers = dict(response.headers)
+
+        assert 'Access-Control-Allow-Origin' not in response_headers
+        assert 'Access-Control-Allow-Methods' not in response_headers
+        assert 'Access-Control-Allow-Headers' not in response_headers
+
+    @helpers.change_config('ckan.cors.origin_allow_all', 'true')
+    def test_cors_config_origin_allow_all_true_no_origin(self):
+        '''
+        With origin_allow_all set to true, but no origin in the request
+        header, no Access-Control-Allow headers should be in the response.
+        '''
+        response = self.app.get('/simple_flask')
+        response_headers = dict(response.headers)
+
+        assert 'Access-Control-Allow-Origin' not in response_headers
+        assert 'Access-Control-Allow-Methods' not in response_headers
+        assert 'Access-Control-Allow-Headers' not in response_headers
+
+    @helpers.change_config('ckan.cors.origin_allow_all', 'true')
+    @helpers.change_config('ckan.site_url', 'http://test.ckan.org')
+    def test_cors_config_origin_allow_all_true_with_origin(self):
+        '''
+        With origin_allow_all set to true, and an origin in the request
+        header, the appropriate Access-Control-Allow headers should be in the
+        response.
+        '''
+        request_headers = {'Origin': 'http://thirdpartyrequests.org'}
+        response = self.app.get('/simple_flask', headers=request_headers)
+        response_headers = dict(response.headers)
+
+        assert 'Access-Control-Allow-Origin' in response_headers
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Origin'], '*')
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Methods'], "POST, PUT, GET, DELETE, OPTIONS")
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Headers'], "X-CKAN-API-KEY, Authorization, Content-Type")
+
+    @helpers.change_config('ckan.cors.origin_allow_all', 'false')
+    @helpers.change_config('ckan.site_url', 'http://test.ckan.org')
+    def test_cors_config_origin_allow_all_false_with_origin_without_whitelist(self):
+        '''
+        With origin_allow_all set to false, with an origin in the request
+        header, but no whitelist defined, there should be no Access-Control-
+        Allow headers in the response.
+        '''
+        request_headers = {'Origin': 'http://thirdpartyrequests.org'}
+        response = self.app.get('/simple_flask', headers=request_headers)
+        response_headers = dict(response.headers)
+
+        assert 'Access-Control-Allow-Origin' not in response_headers
+        assert 'Access-Control-Allow-Methods' not in response_headers
+        assert 'Access-Control-Allow-Headers' not in response_headers
+
+    @helpers.change_config('ckan.cors.origin_allow_all', 'false')
+    @helpers.change_config('ckan.cors.origin_whitelist', 'http://thirdpartyrequests.org')
+    @helpers.change_config('ckan.site_url', 'http://test.ckan.org')
+    def test_cors_config_origin_allow_all_false_with_whitelisted_origin(self):
+        '''
+        With origin_allow_all set to false, with an origin in the request
+        header, and a whitelist defined (containing the origin), the
+        appropriate Access-Control-Allow headers should be in the response.
+        '''
+        request_headers = {'Origin': 'http://thirdpartyrequests.org'}
+        response = self.app.get('/simple_flask', headers=request_headers)
+        response_headers = dict(response.headers)
+
+        assert 'Access-Control-Allow-Origin' in response_headers
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Origin'], 'http://thirdpartyrequests.org')
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Methods'], "POST, PUT, GET, DELETE, OPTIONS")
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Headers'], "X-CKAN-API-KEY, Authorization, Content-Type")
+
+    @helpers.change_config('ckan.cors.origin_allow_all', 'false')
+    @helpers.change_config('ckan.cors.origin_whitelist', 'http://google.com http://thirdpartyrequests.org http://yahoo.co.uk')
+    @helpers.change_config('ckan.site_url', 'http://test.ckan.org')
+    def test_cors_config_origin_allow_all_false_with_multiple_whitelisted_origins(self):
+        '''
+        With origin_allow_all set to false, with an origin in the request
+        header, and a whitelist defining multiple allowed origins (containing
+        the origin), the appropriate Access-Control-Allow headers should be in
+        the response.
+        '''
+        request_headers = {'Origin': 'http://thirdpartyrequests.org'}
+        response = self.app.get('/simple_flask', headers=request_headers)
+        response_headers = dict(response.headers)
+
+        assert 'Access-Control-Allow-Origin' in response_headers
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Origin'], 'http://thirdpartyrequests.org')
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Methods'], "POST, PUT, GET, DELETE, OPTIONS")
+        nose_tools.assert_equal(response_headers['Access-Control-Allow-Headers'], "X-CKAN-API-KEY, Authorization, Content-Type")
+
+    @helpers.change_config('ckan.cors.origin_allow_all', 'false')
+    @helpers.change_config('ckan.cors.origin_whitelist', 'http://google.com http://yahoo.co.uk')
+    @helpers.change_config('ckan.site_url', 'http://test.ckan.org')
+    def test_cors_config_origin_allow_all_false_with_whitelist_not_containing_origin(self):
+        '''
+        With origin_allow_all set to false, with an origin in the request
+        header, and a whitelist defining multiple allowed origins (but not
+        containing the requesting origin), there should be no Access-Control-
+        Allow headers in the response.
+        '''
+        request_headers = {'Origin': 'http://thirdpartyrequests.org'}
+        response = self.app.get('/simple_flask', headers=request_headers)
+        response_headers = dict(response.headers)
+
+        assert 'Access-Control-Allow-Origin' not in response_headers
+        assert 'Access-Control-Allow-Methods' not in response_headers
+        assert 'Access-Control-Allow-Headers' not in response_headers

ckan/views/__init__.py

@@ -1,5 +1,7 @@
 # encoding: utf-8
 
+from paste.deploy.converters import asbool
+
 import ckan.model as model
 from ckan.common import g, request, config
 from ckan.lib.helpers import redirect_to as redirect
@@ -12,6 +14,32 @@
 APIKEY_HEADER_NAME_DEFAULT = u'X-CKAN-API-Key'
 
 
+def set_cors_headers_for_response(response):
+    u'''
+    Set up Access Control Allow headers if either origin_allow_all is True, or
+    the request Origin is in the origin_whitelist.
+    '''
+    if config.get('ckan.cors.origin_allow_all') \
+       and request.headers.get('Origin'):
+
+        cors_origin_allowed = None
+        if asbool(config.get(u'ckan.cors.origin_allow_all')):
+            cors_origin_allowed = "*"
+        elif config.get(u'ckan.cors.origin_whitelist') and \
+                request.headers.get(u'Origin') \
+                in config[u'ckan.cors.origin_whitelist'].split(' '):
+            # set var to the origin to allow it.
+            cors_origin_allowed = request.headers.get(u'Origin')
+
+        if cors_origin_allowed is not None:
+            response.headers[u'Access-Control-Allow-Origin'] = \
+                cors_origin_allowed
+            response.headers[u'Access-Control-Allow-Methods'] = \
+                u'POST, PUT, GET, DELETE, OPTIONS'
+            response.headers[u'Access-Control-Allow-Headers'] = \
+                u'X-CKAN-API-KEY, Authorization, Content-Type'
+
+
 def identify_user():
     u'''Try to identify the user
     If the user is identified then: