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

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')