Michael Krotscheck 396f430d64
Added set_defaults method to oslo_middleware.cors
This patch adds a set_defaults method to the CORS middleware. It
allows overriding the configuration defaults, both for runtime
execution and for config generation.

Change-Id: Iffc20a09c455fa73239d8fe41491daa309858da9
2016-05-04 06:53:04 -07:00

1372 lines
60 KiB
Python

# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_config import fixture
from oslotest import base as test_base
import webob
import webob.dec
import webob.exc as exc
from oslo_middleware import cors
@webob.dec.wsgify
def test_application(req):
if req.path_info == '/server_cors':
# Mirror back the origin in the request.
response = webob.Response(status=200)
response.headers['Access-Control-Allow-Origin'] = \
req.headers['Origin']
response.headers['X-Server-Generated-Response'] = '1'
return response
if req.path_info == '/server_cors_vary':
# Mirror back the origin in the request.
response = webob.Response(status=200)
response.headers['Vary'] = 'Custom-Vary'
return response
if req.path_info == '/server_no_cors':
# Send a response with no CORS headers.
response = webob.Response(status=200)
return response
if req.method == 'OPTIONS':
raise exc.HTTPNotFound()
return 'Hello World'
class CORSTestBase(test_base.BaseTestCase):
"""Base class for all CORS tests.
Sets up applications and helper methods.
"""
def setUp(self):
"""Setup the tests."""
super(CORSTestBase, self).setUp()
# Set up the config fixture.
self.config_fixture = self.useFixture(fixture.Config())
self.config = self.config_fixture.conf
def assertCORSResponse(self, response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None,
vary='Origin',
has_content_type=False):
"""Test helper for CORS response headers.
Assert all the headers in a given response. By default, we assume
the response is empty.
"""
# Assert response status.
self.assertEqual(response.status, status)
# Assert the Access-Control-Allow-Origin header.
self.assertHeader(response,
'Access-Control-Allow-Origin',
allow_origin)
# Assert the Access-Control-Max-Age header.
self.assertHeader(response,
'Access-Control-Max-Age',
max_age)
# Assert the Access-Control-Allow-Methods header.
self.assertHeader(response,
'Access-Control-Allow-Methods',
allow_methods)
# Assert the Access-Control-Allow-Headers header.
self.assertHeader(response,
'Access-Control-Allow-Headers',
allow_headers)
# Assert the Access-Control-Allow-Credentials header.
self.assertHeader(response,
'Access-Control-Allow-Credentials',
allow_credentials)
# Assert the Access-Control-Expose-Headers header.
self.assertHeader(response,
'Access-Control-Expose-Headers',
expose_headers)
# Assert no Content-Type added.
if not has_content_type:
self.assertHeader(response, 'Content-Type')
# If we're expecting an origin response, also assert that the
# Vary: Origin header is set, since this implementation of the CORS
# specification permits multiple origin domains.
if allow_origin:
self.assertHeader(response, 'Vary', vary)
def assertHeader(self, response, header, value=None):
if value:
self.assertIn(header, response.headers)
self.assertEqual(str(value),
response.headers[header])
else:
self.assertNotIn(header, response.headers)
class CORSTestDefaultOverrides(CORSTestBase):
def setUp(self):
super(CORSTestDefaultOverrides, self).setUp()
fixture = self.config_fixture # Line length accommodation
fixture.load_raw_values(group='cors',
allowed_origin='http://valid.example.com')
fixture.load_raw_values(group='cors.override_creds',
allowed_origin='http://creds.example.com',
allow_credentials='True')
fixture.load_raw_values(group='cors.override_headers',
allowed_origin='http://headers.example.com',
expose_headers='X-Header-1,X-Header-2',
allow_headers='X-Header-1,X-Header-2')
self.override_opts = {
'expose_headers': ['X-Header-1'],
'allow_headers': ['X-Header-2'],
'allow_methods': ['GET', 'DELETE'],
'allow_credentials': False,
'max_age': 10
}
def test_config_defaults(self):
"""Assert that using set_defaults overrides the appropriate values."""
cors.set_defaults(**self.override_opts)
for opt in cors.CORS_OPTS:
if opt.dest in self.override_opts:
self.assertEqual(opt.default, self.override_opts[opt.dest])
def test_invalid_default_option(self):
"""Assert that using set_defaults only permits valid options."""
self.assertRaises(AttributeError,
cors.set_defaults,
allowed_origin='test')
def test_cascading_override(self):
"""Assert that using set_defaults overrides cors.* config values."""
# set defaults
cors.set_defaults(**self.override_opts)
# Now that the config is set up, create our application.
self.application = cors.CORS(test_application, self.config)
# Check the global configuration for expected values:
gc = self.config.cors
self.assertEqual(gc.allowed_origin, ['http://valid.example.com'])
self.assertEqual(gc.allow_credentials,
self.override_opts['allow_credentials'])
self.assertEqual(gc.expose_headers,
self.override_opts['expose_headers'])
self.assertEqual(gc.max_age, 10)
self.assertEqual(gc.allow_methods,
self.override_opts['allow_methods'])
self.assertEqual(gc.allow_headers,
self.override_opts['allow_headers'])
# Check the child configuration for expected values:
cc = self.config['cors.override_creds']
self.assertEqual(cc.allowed_origin, ['http://creds.example.com'])
self.assertTrue(cc.allow_credentials)
self.assertEqual(cc.expose_headers,
self.override_opts['expose_headers'])
self.assertEqual(cc.max_age, 10)
self.assertEqual(cc.allow_methods,
self.override_opts['allow_methods'])
self.assertEqual(cc.allow_headers,
self.override_opts['allow_headers'])
# Check the other child configuration for expected values:
ec = self.config['cors.override_headers']
self.assertEqual(ec.allowed_origin, ['http://headers.example.com'])
self.assertEqual(ec.allow_credentials,
self.override_opts['allow_credentials'])
self.assertEqual(ec.expose_headers, ['X-Header-1', 'X-Header-2'])
self.assertEqual(ec.max_age, 10)
self.assertEqual(ec.allow_methods,
self.override_opts['allow_methods'])
self.assertEqual(ec.allow_headers, ['X-Header-1', 'X-Header-2'])
class CORSTestFilterFactory(CORSTestBase):
"""Test the CORS filter_factory method."""
def test_filter_factory(self):
self.config([])
# Test a valid filter.
filter = cors.filter_factory(None,
allowed_origin='http://valid.example.com',
allow_credentials='False',
max_age='',
expose_headers='',
allow_methods='GET',
allow_headers='')
application = filter(test_application)
self.assertIn('http://valid.example.com', application.allowed_origins)
config = application.allowed_origins['http://valid.example.com']
self.assertEqual(False, config['allow_credentials'])
self.assertIsNone(config['max_age'])
self.assertEqual([], config['expose_headers'])
self.assertEqual(['GET'], config['allow_methods'])
self.assertEqual([], config['allow_headers'])
def test_filter_factory_multiorigin(self):
self.config([])
# Test a valid filter.
filter = cors.filter_factory(None,
allowed_origin='http://valid.example.com,'
'http://other.example.com')
application = filter(test_application)
self.assertIn('http://valid.example.com', application.allowed_origins)
self.assertIn('http://other.example.com', application.allowed_origins)
def test_no_origin_fail(self):
'''Assert that a filter factory with no allowed_origin fails.'''
self.assertRaises(TypeError,
cors.filter_factory,
global_conf=None,
# allowed_origin=None, # Expected value.
allow_credentials='False',
max_age='',
expose_headers='',
allow_methods='GET',
allow_headers='')
def test_no_origin_but_oslo_config_project(self):
'''Assert that a filter factory with oslo_config_project succeed.'''
cors.filter_factory(global_conf=None, oslo_config_project='foobar')
def test_cor_config_sections_with_defaults(self):
'''Assert cors.* config sections with default values work.'''
# Set up the config fixture.
self.config_fixture.load_raw_values(group='cors.subdomain')
# Now that the config is set up, create our application.
self.application = cors.CORS(test_application, self.config)
def test_factory_latent_properties(self):
'''Assert latent properties in paste.ini config.
If latent_* properties are added to a paste.ini config, assert that
they are persisted in the middleware.
'''
# Spaces in config are deliberate to frobb the config parsing.
filter = cors.filter_factory(global_conf=None,
oslo_config_project='foobar',
latent_expose_headers=' X-Header-1 , X-2',
latent_allow_headers='X-Header-1 , X-2',
latent_allow_methods='GET,PUT, POST')
app = filter(test_application)
# Ensure that the properties are in latent configuration.
self.assertEqual(['X-Header-1', 'X-2'],
app._latent_configuration['expose_headers'])
self.assertEqual(['X-Header-1', 'X-2'],
app._latent_configuration['allow_headers'])
self.assertEqual(['GET', 'PUT', 'POST'],
app._latent_configuration['methods'])
class CORSRegularRequestTest(CORSTestBase):
"""CORS Specification Section 6.1
http://www.w3.org/TR/cors/#resource-requests
"""
# List of HTTP methods (other than OPTIONS) to test with.
methods = ['POST', 'PUT', 'DELETE', 'GET', 'TRACE', 'HEAD']
def setUp(self):
"""Setup the tests."""
super(CORSRegularRequestTest, self).setUp()
fixture = self.config_fixture # Line length accommodation
fixture.load_raw_values(group='cors',
allowed_origin='http://valid.example.com',
allow_credentials='False',
max_age='',
expose_headers='',
allow_methods='GET',
allow_headers='')
fixture.load_raw_values(group='cors.credentials',
allowed_origin='http://creds.example.com',
allow_credentials='True')
fixture.load_raw_values(group='cors.exposed-headers',
allowed_origin='http://headers.example.com',
expose_headers='X-Header-1,X-Header-2',
allow_headers='X-Header-1,X-Header-2')
fixture.load_raw_values(group='cors.cached',
allowed_origin='http://cached.example.com',
max_age='3600')
fixture.load_raw_values(group='cors.get-only',
allowed_origin='http://get.example.com',
allow_methods='GET')
fixture.load_raw_values(group='cors.all-methods',
allowed_origin='http://all.example.com',
allow_methods='GET,PUT,POST,DELETE,HEAD')
fixture.load_raw_values(group='cors.duplicate',
allowed_origin='http://domain1.example.com,'
'http://domain2.example.com')
# Now that the config is set up, create our application.
self.application = cors.CORS(test_application, self.config)
def test_config_overrides(self):
"""Assert that the configuration options are properly registered."""
# Confirm global configuration
gc = self.config.cors
self.assertEqual(gc.allowed_origin, ['http://valid.example.com'])
self.assertEqual(gc.allow_credentials, False)
self.assertEqual(gc.expose_headers, [])
self.assertEqual(gc.max_age, None)
self.assertEqual(gc.allow_methods, ['GET'])
self.assertEqual(gc.allow_headers, [])
# Confirm credentials overrides.
cc = self.config['cors.credentials']
self.assertEqual(cc.allowed_origin, ['http://creds.example.com'])
self.assertEqual(cc.allow_credentials, True)
self.assertEqual(cc.expose_headers, gc.expose_headers)
self.assertEqual(cc.max_age, gc.max_age)
self.assertEqual(cc.allow_methods, gc.allow_methods)
self.assertEqual(cc.allow_headers, gc.allow_headers)
# Confirm exposed-headers overrides.
ec = self.config['cors.exposed-headers']
self.assertEqual(ec.allowed_origin, ['http://headers.example.com'])
self.assertEqual(ec.allow_credentials, gc.allow_credentials)
self.assertEqual(ec.expose_headers, ['X-Header-1', 'X-Header-2'])
self.assertEqual(ec.max_age, gc.max_age)
self.assertEqual(ec.allow_methods, gc.allow_methods)
self.assertEqual(ec.allow_headers, ['X-Header-1', 'X-Header-2'])
# Confirm cached overrides.
chc = self.config['cors.cached']
self.assertEqual(chc.allowed_origin, ['http://cached.example.com'])
self.assertEqual(chc.allow_credentials, gc.allow_credentials)
self.assertEqual(chc.expose_headers, gc.expose_headers)
self.assertEqual(chc.max_age, 3600)
self.assertEqual(chc.allow_methods, gc.allow_methods)
self.assertEqual(chc.allow_headers, gc.allow_headers)
# Confirm get-only overrides.
goc = self.config['cors.get-only']
self.assertEqual(goc.allowed_origin, ['http://get.example.com'])
self.assertEqual(goc.allow_credentials, gc.allow_credentials)
self.assertEqual(goc.expose_headers, gc.expose_headers)
self.assertEqual(goc.max_age, gc.max_age)
self.assertEqual(goc.allow_methods, ['GET'])
self.assertEqual(goc.allow_headers, gc.allow_headers)
# Confirm all-methods overrides.
ac = self.config['cors.all-methods']
self.assertEqual(ac.allowed_origin, ['http://all.example.com'])
self.assertEqual(ac.allow_credentials, gc.allow_credentials)
self.assertEqual(ac.expose_headers, gc.expose_headers)
self.assertEqual(ac.max_age, gc.max_age)
self.assertEqual(ac.allow_methods,
['GET', 'PUT', 'POST', 'DELETE', 'HEAD'])
self.assertEqual(ac.allow_headers, gc.allow_headers)
# Confirm duplicate domains.
ac = self.config['cors.duplicate']
self.assertEqual(ac.allowed_origin, ['http://domain1.example.com',
'http://domain2.example.com'])
self.assertEqual(ac.allow_credentials, gc.allow_credentials)
self.assertEqual(ac.expose_headers, gc.expose_headers)
self.assertEqual(ac.max_age, gc.max_age)
self.assertEqual(ac.allow_methods, gc.allow_methods)
self.assertEqual(ac.allow_headers, gc.allow_headers)
def test_no_origin_header(self):
"""CORS Specification Section 6.1.1
If the Origin header is not present terminate this set of steps. The
request is outside the scope of this specification.
"""
for method in self.methods:
request = webob.Request.blank('/')
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None,
has_content_type=True)
def test_origin_headers(self):
"""CORS Specification Section 6.1.2
If the value of the Origin header is not a case-sensitive match for
any of the values in list of origins, do not set any additional
headers and terminate this set of steps.
"""
# Test valid origin header.
for method in self.methods:
request = webob.Request.blank('/')
request.method = method
request.headers['Origin'] = 'http://valid.example.com'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://valid.example.com',
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None,
has_content_type=True)
# Test origin header not present in configuration.
for method in self.methods:
request = webob.Request.blank('/')
request.method = method
request.headers['Origin'] = 'http://invalid.example.com'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None,
has_content_type=True)
# Test valid, but case-mismatched origin header.
for method in self.methods:
request = webob.Request.blank('/')
request.method = method
request.headers['Origin'] = 'http://VALID.EXAMPLE.COM'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None,
has_content_type=True)
# Test valid header from list of duplicates.
for method in self.methods:
request = webob.Request.blank('/')
request.method = method
request.headers['Origin'] = 'http://domain2.example.com'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://domain2.example.com',
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None,
has_content_type=True)
def test_supports_credentials(self):
"""CORS Specification Section 6.1.3
If the resource supports credentials add a single
Access-Control-Allow-Origin header, with the value of the Origin header
as value, and add a single Access-Control-Allow-Credentials header with
the case-sensitive string "true" as value.
Otherwise, add a single Access-Control-Allow-Origin header, with
either the value of the Origin header or the string "*" as value.
NOTE: We never use the "*" as origin.
"""
# Test valid origin header without credentials.
for method in self.methods:
request = webob.Request.blank('/')
request.method = method
request.headers['Origin'] = 'http://valid.example.com'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://valid.example.com',
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None,
has_content_type=True)
# Test valid origin header with credentials
for method in self.methods:
request = webob.Request.blank('/')
request.method = method
request.headers['Origin'] = 'http://creds.example.com'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://creds.example.com',
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials="true",
expose_headers=None,
has_content_type=True)
def test_expose_headers(self):
"""CORS Specification Section 6.1.4
If the list of exposed headers is not empty add one or more
Access-Control-Expose-Headers headers, with as values the header field
names given in the list of exposed headers.
"""
for method in self.methods:
request = webob.Request.blank('/')
request.method = method
request.headers['Origin'] = 'http://headers.example.com'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://headers.example.com',
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers='X-Header-1,X-Header-2',
has_content_type=True)
def test_application_options_response(self):
"""Assert that an application provided OPTIONS response is honored.
If the underlying application, via middleware or other, provides a
CORS response, its response should be honored.
"""
test_origin = 'http://creds.example.com'
request = webob.Request.blank('/server_cors')
request.method = "GET"
request.headers['Origin'] = test_origin
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
# If the regular CORS handling catches this request, it should set
# the allow credentials header. This makes sure that it doesn't.
self.assertNotIn('Access-Control-Allow-Credentials', response.headers)
self.assertEqual(response.headers['Access-Control-Allow-Origin'],
test_origin)
self.assertEqual(response.headers['X-Server-Generated-Response'],
'1')
def test_application_vary_respected(self):
"""Assert that an application's provided Vary header is persisted.
If the underlying application, via middleware or other, provides a
Vary header, its response should be honored.
"""
request = webob.Request.blank('/server_cors_vary')
request.method = "GET"
request.headers['Origin'] = 'http://valid.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://valid.example.com',
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None,
vary='Custom-Vary,Origin',
has_content_type=True)
class CORSPreflightRequestTest(CORSTestBase):
"""CORS Specification Section 6.2
http://www.w3.org/TR/cors/#resource-preflight-requests
"""
def setUp(self):
super(CORSPreflightRequestTest, self).setUp()
fixture = self.config_fixture # Line length accommodation
fixture.load_raw_values(group='cors',
allowed_origin='http://valid.example.com',
allow_credentials='False',
max_age='',
expose_headers='',
allow_methods='GET',
allow_headers='')
fixture.load_raw_values(group='cors.credentials',
allowed_origin='http://creds.example.com',
allow_credentials='True')
fixture.load_raw_values(group='cors.exposed-headers',
allowed_origin='http://headers.example.com',
expose_headers='X-Header-1,X-Header-2',
allow_headers='X-Header-1,X-Header-2')
fixture.load_raw_values(group='cors.cached',
allowed_origin='http://cached.example.com',
max_age='3600')
fixture.load_raw_values(group='cors.get-only',
allowed_origin='http://get.example.com',
allow_methods='GET')
fixture.load_raw_values(group='cors.all-methods',
allowed_origin='http://all.example.com',
allow_methods='GET,PUT,POST,DELETE,HEAD')
# Now that the config is set up, create our application.
self.application = cors.CORS(test_application, self.config)
def test_config_overrides(self):
"""Assert that the configuration options are properly registered."""
# Confirm global configuration
gc = self.config.cors
self.assertEqual(gc.allowed_origin, ['http://valid.example.com'])
self.assertEqual(gc.allow_credentials, False)
self.assertEqual(gc.expose_headers, [])
self.assertEqual(gc.max_age, None)
self.assertEqual(gc.allow_methods, ['GET'])
self.assertEqual(gc.allow_headers, [])
# Confirm credentials overrides.
cc = self.config['cors.credentials']
self.assertEqual(cc.allowed_origin, ['http://creds.example.com'])
self.assertEqual(cc.allow_credentials, True)
self.assertEqual(cc.expose_headers, gc.expose_headers)
self.assertEqual(cc.max_age, gc.max_age)
self.assertEqual(cc.allow_methods, gc.allow_methods)
self.assertEqual(cc.allow_headers, gc.allow_headers)
# Confirm exposed-headers overrides.
ec = self.config['cors.exposed-headers']
self.assertEqual(ec.allowed_origin, ['http://headers.example.com'])
self.assertEqual(ec.allow_credentials, gc.allow_credentials)
self.assertEqual(ec.expose_headers, ['X-Header-1', 'X-Header-2'])
self.assertEqual(ec.max_age, gc.max_age)
self.assertEqual(ec.allow_methods, gc.allow_methods)
self.assertEqual(ec.allow_headers, ['X-Header-1', 'X-Header-2'])
# Confirm cached overrides.
chc = self.config['cors.cached']
self.assertEqual(chc.allowed_origin, ['http://cached.example.com'])
self.assertEqual(chc.allow_credentials, gc.allow_credentials)
self.assertEqual(chc.expose_headers, gc.expose_headers)
self.assertEqual(chc.max_age, 3600)
self.assertEqual(chc.allow_methods, gc.allow_methods)
self.assertEqual(chc.allow_headers, gc.allow_headers)
# Confirm get-only overrides.
goc = self.config['cors.get-only']
self.assertEqual(goc.allowed_origin, ['http://get.example.com'])
self.assertEqual(goc.allow_credentials, gc.allow_credentials)
self.assertEqual(goc.expose_headers, gc.expose_headers)
self.assertEqual(goc.max_age, gc.max_age)
self.assertEqual(goc.allow_methods, ['GET'])
self.assertEqual(goc.allow_headers, gc.allow_headers)
# Confirm all-methods overrides.
ac = self.config['cors.all-methods']
self.assertEqual(ac.allowed_origin, ['http://all.example.com'])
self.assertEqual(ac.allow_credentials, gc.allow_credentials)
self.assertEqual(ac.expose_headers, gc.expose_headers)
self.assertEqual(ac.max_age, gc.max_age)
self.assertEqual(ac.allow_methods,
['GET', 'PUT', 'POST', 'DELETE', 'HEAD'])
self.assertEqual(ac.allow_headers, gc.allow_headers)
def test_no_origin_header(self):
"""CORS Specification Section 6.2.1
If the Origin header is not present terminate this set of steps. The
request is outside the scope of this specification.
"""
request = webob.Request.blank('/')
request.method = "OPTIONS"
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_case_sensitive_origin(self):
"""CORS Specification Section 6.2.2
If the value of the Origin header is not a case-sensitive match for
any of the values in list of origins do not set any additional headers
and terminate this set of steps.
"""
# Test valid domain
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://valid.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://valid.example.com',
max_age=None,
allow_methods='GET',
allow_headers='',
allow_credentials=None,
expose_headers=None)
# Test invalid domain
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://invalid.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
# Test case-sensitive mismatch domain
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://VALID.EXAMPLE.COM'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_simple_header_response(self):
"""CORS Specification Section 3
A header is said to be a simple header if the header field name is an
ASCII case-insensitive match for Accept, Accept-Language, or
Content-Language or if it is an ASCII case-insensitive match for
Content-Type and the header field value media type (excluding
parameters) is an ASCII case-insensitive match for
application/x-www-form-urlencoded, multipart/form-data, or text/plain.
NOTE: We are not testing the media type cases.
"""
simple_headers = ','.join([
'accept',
'accept-language',
'content-language',
'content-type'
])
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://valid.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
request.headers['Access-Control-Request-Headers'] = simple_headers
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://valid.example.com',
max_age=None,
allow_methods='GET',
allow_headers=simple_headers,
allow_credentials=None,
expose_headers=None)
def test_no_request_method(self):
"""CORS Specification Section 6.2.3
If there is no Access-Control-Request-Method header or if parsing
failed, do not set any additional headers and terminate this set of
steps. The request is outside the scope of this specification.
"""
# Test valid domain, valid method.
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://get.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://get.example.com',
max_age=None,
allow_methods='GET',
allow_headers=None,
allow_credentials=None,
expose_headers=None)
# Test valid domain, invalid HTTP method.
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://valid.example.com'
request.headers['Access-Control-Request-Method'] = 'TEAPOT'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
# Test valid domain, no HTTP method.
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://valid.example.com'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_invalid_method(self):
"""CORS Specification Section 6.2.3
If method is not a case-sensitive match for any of the values in
list of methods do not set any additional headers and terminate this
set of steps.
"""
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://get.example.com'
request.headers['Access-Control-Request-Method'] = 'get'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_no_parse_request_headers(self):
"""CORS Specification Section 6.2.4
If there are no Access-Control-Request-Headers headers let header
field-names be the empty list.
If parsing failed do not set any additional headers and terminate
this set of steps. The request is outside the scope of this
specification.
"""
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://headers.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
request.headers['Access-Control-Request-Headers'] = 'value with spaces'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_no_request_headers(self):
"""CORS Specification Section 6.2.4
If there are no Access-Control-Request-Headers headers let header
field-names be the empty list.
"""
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://headers.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
request.headers['Access-Control-Request-Headers'] = ''
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://headers.example.com',
max_age=None,
allow_methods='GET',
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_request_headers(self):
"""CORS Specification Section 6.2.4
Let header field-names be the values as result of parsing the
Access-Control-Request-Headers headers.
If there are no Access-Control-Request-Headers headers let header
field-names be the empty list.
"""
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://headers.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
request.headers['Access-Control-Request-Headers'] = 'X-Header-1,' \
'X-Header-2'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://headers.example.com',
max_age=None,
allow_methods='GET',
allow_headers='X-Header-1,X-Header-2',
allow_credentials=None,
expose_headers=None)
def test_request_headers_not_permitted(self):
"""CORS Specification Section 6.2.4, 6.2.6
If there are no Access-Control-Request-Headers headers let header
field-names be the empty list.
If any of the header field-names is not a ASCII case-insensitive
match for any of the values in list of headers do not set any
additional headers and terminate this set of steps.
"""
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://headers.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
request.headers['Access-Control-Request-Headers'] = 'X-Not-Exposed,' \
'X-Never-Exposed'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_credentials(self):
"""CORS Specification Section 6.2.7
If the resource supports credentials add a single
Access-Control-Allow-Origin header, with the value of the Origin header
as value, and add a single Access-Control-Allow-Credentials header with
the case-sensitive string "true" as value.
Otherwise, add a single Access-Control-Allow-Origin header, with either
the value of the Origin header or the string "*" as value.
NOTE: We never use the "*" as origin.
"""
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://creds.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://creds.example.com',
max_age=None,
allow_methods='GET',
allow_headers=None,
allow_credentials="true",
expose_headers=None)
def test_optional_max_age(self):
"""CORS Specification Section 6.2.8
Optionally add a single Access-Control-Max-Age header with as value
the amount of seconds the user agent is allowed to cache the result of
the request.
"""
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://cached.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://cached.example.com',
max_age=3600,
allow_methods='GET',
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_allow_methods(self):
"""CORS Specification Section 6.2.9
Add one or more Access-Control-Allow-Methods headers consisting of
(a subset of) the list of methods.
Since the list of methods can be unbounded, simply returning the method
indicated by Access-Control-Request-Method (if supported) can be
enough.
"""
for method in ['GET', 'PUT', 'POST', 'DELETE']:
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://all.example.com'
request.headers['Access-Control-Request-Method'] = method
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://all.example.com',
max_age=None,
allow_methods=method,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
for method in ['PUT', 'POST', 'DELETE']:
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://get.example.com'
request.headers['Access-Control-Request-Method'] = method
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin=None,
max_age=None,
allow_methods=None,
allow_headers=None,
allow_credentials=None,
expose_headers=None)
def test_allow_headers(self):
"""CORS Specification Section 6.2.10
Add one or more Access-Control-Allow-Headers headers consisting of
(a subset of) the list of headers.
If each of the header field-names is a simple header and none is
Content-Type, this step may be skipped.
If a header field name is a simple header and is not Content-Type, it
is not required to be listed. Content-Type is to be listed as only a
subset of its values makes it qualify as simple header.
"""
requested_headers = 'Content-Type,X-Header-1,Cache-Control,Expires,' \
'Last-Modified,Pragma'
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://headers.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
request.headers['Access-Control-Request-Headers'] = requested_headers
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://headers.example.com',
max_age=None,
allow_methods='GET',
allow_headers=requested_headers,
allow_credentials=None,
expose_headers=None)
def test_application_options_response(self):
"""Assert that an application provided OPTIONS response is honored.
If the underlying application, via middleware or other, provides a
CORS response, its response should be honored.
"""
test_origin = 'http://creds.example.com'
request = webob.Request.blank('/server_cors')
request.method = "OPTIONS"
request.headers['Origin'] = test_origin
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
# If the regular CORS handling catches this request, it should set
# the allow credentials header. This makes sure that it doesn't.
self.assertNotIn('Access-Control-Allow-Credentials', response.headers)
self.assertEqual(response.headers['Access-Control-Allow-Origin'],
test_origin)
self.assertEqual(response.headers['X-Server-Generated-Response'],
'1')
# If the application returns an OPTIONS response without CORS
# headers, assert that we apply headers.
request = webob.Request.blank('/server_no_cors')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://get.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='http://get.example.com',
max_age=None,
allow_methods='GET',
allow_headers=None,
allow_credentials=None,
expose_headers=None,
has_content_type=True)
class CORSTestWildcard(CORSTestBase):
"""Test the CORS wildcard specification."""
def setUp(self):
super(CORSTestWildcard, self).setUp()
fixture = self.config_fixture # Line length accommodation
fixture.load_raw_values(group='cors',
allowed_origin='http://default.example.com',
allow_credentials='True',
max_age='',
expose_headers='',
allow_methods='GET,PUT,POST,DELETE,HEAD',
allow_headers='')
fixture.load_raw_values(group='cors.wildcard',
allowed_origin='*',
allow_methods='GET')
# Now that the config is set up, create our application.
self.application = cors.CORS(test_application, self.config)
def test_config_overrides(self):
"""Assert that the configuration options are properly registered."""
# Confirm global configuration
gc = self.config.cors
self.assertEqual(gc.allowed_origin, ['http://default.example.com'])
self.assertEqual(gc.allow_credentials, True)
self.assertEqual(gc.expose_headers, [])
self.assertEqual(gc.max_age, None)
self.assertEqual(gc.allow_methods, ['GET', 'PUT', 'POST', 'DELETE',
'HEAD'])
self.assertEqual(gc.allow_headers, [])
# Confirm all-methods overrides.
ac = self.config['cors.wildcard']
self.assertEqual(ac.allowed_origin, ['*'])
self.assertEqual(gc.allow_credentials, True)
self.assertEqual(ac.expose_headers, gc.expose_headers)
self.assertEqual(ac.max_age, gc.max_age)
self.assertEqual(ac.allow_methods, ['GET'])
self.assertEqual(ac.allow_headers, gc.allow_headers)
def test_wildcard_domain(self):
"""CORS Specification, Wildcards
If the configuration file specifies CORS settings for the wildcard '*'
domain, it should return those for all origin domains except for the
overrides.
"""
# Test valid domain
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://default.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
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='',
allow_credentials='true',
expose_headers=None)
# Test valid domain
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_headers='',
allow_credentials='true',
expose_headers=None,
has_content_type=True)
# Test invalid domain
request = webob.Request.blank('/')
request.method = "OPTIONS"
request.headers['Origin'] = 'http://invalid.example.com'
request.headers['Access-Control-Request-Method'] = 'GET'
response = request.get_response(self.application)
self.assertCORSResponse(response,
status='200 OK',
allow_origin='*',
max_age=None,
allow_methods='GET',
allow_headers='',
allow_credentials='true',
expose_headers=None,
has_content_type=True)
class CORSTestLatentProperties(CORSTestBase):
"""Test the CORS wildcard specification."""
def setUp(self):
super(CORSTestLatentProperties, self).setUp()
fixture = self.config_fixture # Line length accommodation
fixture.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, self.config)
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',
has_content_type=True)
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')