From 3fbb59574ca3bceb42baa0847d146f471145ab31 Mon Sep 17 00:00:00 2001 From: Michael Krotscheck Date: Wed, 5 Aug 2015 16:03:28 -0700 Subject: [PATCH] Added latent properties to CORS middleware. Latent properties allow a consumer of this middleware to declare system-required headers and methods options. For instance, if an API exposes version-negotiation headers, these may be hard coded when the middleware is attached. This only works when the middleware is explicitly used. It does not work in paste configuration. Change-Id: Ic55b1af23603a0d83a32d20054c18e50367be8fb --- doc/source/cors.rst | 11 +++ oslo_middleware/cors.py | 60 +++++++++++++--- oslo_middleware/tests/test_cors.py | 110 +++++++++++++++++++++++++++++ 3 files changed, 173 insertions(+), 8 deletions(-) diff --git a/doc/source/cors.rst b/doc/source/cors.rst index 5763099..09f5d77 100644 --- a/doc/source/cors.rst +++ b/doc/source/cors.rst @@ -78,6 +78,17 @@ legibility, we recommend using a reasonable human-readable string:: allowed_origin=* allow_methods=GET +If your software requires specific headers or methods for proper operation, you +may include these as latent properties. These will be evaluated in addition +to any found in configuration:: + + from oslo_middleware import cors + + app = cors.CORS(your_wsgi_application) + app.set_latent(allow_headers=['X-System-Header'], + expose_headers=['X-System-Header'], + allow_methods=['GET','PATCH']) + Configuration for pastedeploy ----------------------------- diff --git a/oslo_middleware/cors.py b/oslo_middleware/cors.py index ed33393..1b22178 100644 --- a/oslo_middleware/cors.py +++ b/oslo_middleware/cors.py @@ -98,6 +98,13 @@ class CORS(base.Middleware): def _init_conf(self): '''Initialize this middleware from an oslo.config instance.''' + # Set up a location for our latent configuration options + self.latent_configuration = { + 'allow_headers': [], + 'expose_headers': [], + 'methods': [] + } + # First, check the configuration and register global options. self.oslo_conf.register_opts(CORS_OPTS, 'cors') @@ -165,6 +172,39 @@ class CORS(base.Middleware): 'allow_headers': allow_headers } + def set_latent(self, allow_headers=None, allow_methods=None, + expose_headers=None): + '''Add a new latent property for this middleware. + + Latent properties are those values which a system requires for + operation. API-specific headers, for example, may be added by an + engineer so that they ship with the codebase, and thus do not require + extra documentation or passing of institutional knowledge. + + :param allow_headers: HTTP headers permitted in client requests. + :param allow_methods: HTTP methods permitted in client requests. + :param expose_headers: HTTP Headers exposed to clients. + ''' + + if allow_headers: + if isinstance(allow_headers, list): + self.latent_configuration['allow_headers'] = allow_headers + else: + raise TypeError("allow_headers must be a list or None.") + + if expose_headers: + if isinstance(expose_headers, list): + self.latent_configuration['expose_headers'] = expose_headers + else: + raise TypeError("expose_headers must be a list or None.") + + if allow_methods: + if isinstance(allow_methods, list): + self.latent_configuration['methods'] = allow_methods + else: + raise TypeError("allow_methods parameter must be a list or" + " None.") + def process_response(self, response, request=None): '''Check for CORS headers, and decorate if necessary. @@ -249,19 +289,23 @@ class CORS(base.Middleware): return response # Compare request method to permitted methods (Section 6.2.5) - if request_method not in cors_config['allow_methods']: + permitted_methods = ( + cors_config['allow_methods'] + self.latent_configuration['methods'] + ) + if request_method not in permitted_methods: LOG.debug('Request method \'%s\' not in permitted list: %s' - % (request_method, cors_config['allow_methods'])) + % (request_method, permitted_methods)) return response # Compare request headers to permitted headers, case-insensitively. # (Section 6.2.6) + permitted_headers = [header.upper() for header in + (cors_config['allow_headers'] + + self.simple_headers + + self.latent_configuration['allow_headers'])] for requested_header in request_headers: upper_header = requested_header.upper() - permitted_headers = (cors_config['allow_headers'] + - self.simple_headers) - if upper_header not in (header.upper() for header in - permitted_headers): + if upper_header not in permitted_headers: LOG.debug('Request header \'%s\' not in permitted list: %s' % (requested_header, permitted_headers)) return response @@ -319,8 +363,8 @@ class CORS(base.Middleware): # Attach the exposed headers and exit. (Section 6.1.4) if cors_config['expose_headers']: response.headers['Access-Control-Expose-Headers'] = \ - ','.join(cors_config['expose_headers']) - + ','.join(cors_config['expose_headers'] + + self.latent_configuration['expose_headers']) # NOTE(sileht): Shortcut for backwards compatibility filter_factory = CORS.factory diff --git a/oslo_middleware/tests/test_cors.py b/oslo_middleware/tests/test_cors.py index 86bbc79..f1fa150 100644 --- a/oslo_middleware/tests/test_cors.py +++ b/oslo_middleware/tests/test_cors.py @@ -999,3 +999,113 @@ class CORSTestWildcard(CORSTestBase): allow_headers='', allow_credentials='true', expose_headers=None) + + +class CORSTestLatentProperties(CORSTestBase): + """Test the CORS wildcard specification.""" + + def setUp(self): + super(CORSTestLatentProperties, self).setUp() + + # Set up the config fixture. + config = self.useFixture(fixture.Config(cfg.CONF)) + + config.load_raw_values(group='cors', + allowed_origin='http://default.example.com', + allow_credentials='True', + max_age='', + expose_headers='X-Configured', + allow_methods='GET', + allow_headers='X-Configured') + + # Now that the config is set up, create our application. + self.application = cors.CORS(test_application, cfg.CONF) + + def test_latent_methods(self): + """Assert that latent HTTP methods are permitted.""" + + self.application.set_latent(allow_headers=None, + expose_headers=None, + allow_methods=['POST']) + + request = webob.Request.blank('/') + request.method = "OPTIONS" + request.headers['Origin'] = 'http://default.example.com' + request.headers['Access-Control-Request-Method'] = 'POST' + response = request.get_response(self.application) + self.assertCORSResponse(response, + status='200 OK', + allow_origin='http://default.example.com', + max_age=None, + allow_methods='POST', + allow_headers='', + allow_credentials='true', + expose_headers=None) + + def test_invalid_latent_methods(self): + """Assert that passing a non-list is caught.""" + + self.assertRaises(TypeError, + self.application.set_latent, + allow_methods='POST') + + def test_latent_allow_headers(self): + """Assert that latent HTTP headers are permitted.""" + + self.application.set_latent(allow_headers=['X-Latent'], + expose_headers=None, + allow_methods=None) + + request = webob.Request.blank('/') + request.method = "OPTIONS" + request.headers['Origin'] = 'http://default.example.com' + request.headers['Access-Control-Request-Method'] = 'GET' + request.headers[ + 'Access-Control-Request-Headers'] = 'X-Latent,X-Configured' + response = request.get_response(self.application) + self.assertCORSResponse(response, + status='200 OK', + allow_origin='http://default.example.com', + max_age=None, + allow_methods='GET', + allow_headers='X-Latent,X-Configured', + allow_credentials='true', + expose_headers=None) + + def test_invalid_latent_allow_headers(self): + """Assert that passing a non-list is caught in allow headers.""" + + self.assertRaises(TypeError, + self.application.set_latent, + allow_headers='X-Latent') + + def test_latent_expose_headers(self): + """Assert that latent HTTP headers are exposed.""" + + self.application.set_latent(allow_headers=None, + expose_headers=[ + 'X-Server-Generated-Response'], + allow_methods=None) + + request = webob.Request.blank('/') + request.method = "GET" + request.headers['Origin'] = 'http://default.example.com' + response = request.get_response(self.application) + self.assertCORSResponse(response, + status='200 OK', + allow_origin='http://default.example.com', + max_age=None, + allow_methods=None, + allow_headers=None, + allow_credentials='true', + expose_headers='X-Configured,' + 'X-Server-Generated-Response') + + def test_invalid_latent_expose_headers(self): + """Assert that passing a non-list is caught in expose headers.""" + + # Add headers to the application. + + self.assertRaises(TypeError, + self.application.set_latent, + expose_headers='X-Latent')