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
This commit is contained in:
Michael Krotscheck
parent
3b3b30b1da
commit
3fbb59574c
|
|
@ -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
|
||||
-----------------------------
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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')
|
||||
|
|
|
|||
Loading…
Reference in New Issue