Fix: Content Security Policy (CSP) Not Implemented (DataBiosphere/azu…

…l-private#6)

lambdas/service/app.py

@@ -30,6 +30,7 @@
     Response,
     UnauthorizedError,
 )
+import chevron
 from furl import (
     furl,
 )
@@ -488,10 +489,18 @@ def static_resource(file):
     cache_control='no-store'
 )
 def oauth2_redirect():
-    oauth2_redirect_html = app.load_static_resource('swagger', 'oauth2-redirect.html')
+    file_name = 'oauth2-redirect.html.template.mustache'
+    template = app.load_static_resource('swagger', file_name)
+    nonce = app.csp_nonce()
+    html = chevron.render(template, {
+        'CSP_NONCE': json.dumps(nonce)
+    })
     return Response(status_code=200,
-                    headers={'Content-Type': 'text/html'},
-                    body=oauth2_redirect_html)
+                    headers={
+                        'Content-Type': 'text/html',
+                        'Content-Security-Policy': app.content_security_policy(nonce)
+                    },
+                    body=html)
 
 
 common_specs = CommonEndpointSpecs(app_name='service')

src/azul/chalice.py

@@ -1,6 +1,7 @@
 from abc import (
     ABCMeta,
 )
+import base64
 from collections.abc import (
     Iterable,
 )
@@ -15,6 +16,8 @@
 import mimetypes
 import os
 import pathlib
+import re
+import secrets
 from typing import (
     Any,
     Iterator,
@@ -45,6 +48,9 @@
 from furl import (
     furl,
 )
+from more_itertools import (
+    one,
+)
 
 from azul import (
     config,
@@ -193,16 +199,142 @@ def _api_gateway_context_middleware(self, event, get_response):
         finally:
             config.lambda_is_handling_api_gateway_request = False
 
+    @classmethod
+    def csp_nonce(cls) -> str:
+        """
+        Return a randomly generated nonce value for use in a Content Security
+        Policy header.
+        """
+        return base64.b64encode(secrets.token_bytes(32)).decode('ascii').rstrip('=')
+
+    @classmethod
+    def content_security_policy(cls, nonce: str | None = None) -> str:
+        """
+        >>> from azul.doctests import assert_json
+        >>> assert_json(AzulChaliceApp.content_security_policy(None).split(';'))
+        [
+            "default-src 'self'",
+            "img-src 'self' data:",
+            "script-src 'self'",
+            "style-src 'self'",
+            "frame-ancestors 'none'"
+        ]
+
+        >>> assert_json(AzulChaliceApp.content_security_policy(nonce='foo').split(';'))
+        [
+            "default-src 'self'",
+            "img-src 'self' data:",
+            "script-src 'self' 'nonce-foo'",
+            "style-src 'self' 'nonce-foo'",
+            "frame-ancestors 'none'"
+        ]
+        """
+        self_ = sq('self')
+        none = sq('none')
+        nonce = [] if nonce is None else [sq('nonce-' + nonce)]
+
+        return ';'.join([
+            jw('default-src', self_),
+            jw('img-src', self_, 'data:'),
+            jw('script-src', self_, *nonce),
+            jw('style-src', self_, *nonce),
+            jw('frame-ancestors', none),
+        ])
+
+    @classmethod
+    def validate_csp(cls, csp: str, has_nonce: bool) -> Optional[str]:
+        """
+        Raises an exception if the CSP is invalid. Returns a nonce token if the
+        CSP contained one, else None.
+
+        Fails if nonce violates the RFC
+
+        >>> cls = AzulChaliceApp
+        >>> cls.validate_csp("default-src 'self';img-src 'self' data:;"
+        ...                   "script-src 'self' 'nonce-1234567890123456789012345678901234567890***';"
+        ...                   "style-src 'self' 'nonce-1234567890123456789012345678901234567890***';"
+        ...                   "frame-ancestors 'none'", has_nonce=True)
+        Traceback (most recent call last):
+        ...
+        AssertionError: 'nonce-1234567890123456789012345678901234567890***'
+
+        Fails if nonce is shorter than expected
+
+        >>> cls.validate_csp("default-src 'self';img-src 'self' data:;"
+        ...                   "script-src 'self' 'nonce-1234567890';"
+        ...                   "style-src 'self' 'nonce-1234567890';"
+        ...                   "frame-ancestors 'none'", has_nonce=True)
+        Traceback (most recent call last):
+        ...
+        AssertionError: 'nonce-1234567890'
+
+        Fails if nonce is longer than expected
+
+        >>> cls.validate_csp("default-src 'self';img-src 'self' data:;"
+        ...                   "script-src 'self' 'nonce-12345678901234567890123456789012345678901234567890';"
+        ...                   "style-src 'self' 'nonce-12345678901234567890123456789012345678901234567890';"
+        ...                   "frame-ancestors 'none'", has_nonce=True)
+        Traceback (most recent call last):
+        ...
+        AssertionError: 'nonce-12345678901234567890123456789012345678901234567890'
+        """
+        # https://www.w3.org/TR/CSP2/#policy-syntax
+        directive_re = re.compile(r'[ \t]*([a-zA-Z0-9-]+)'
+                                  # Space, tab and any visible character
+                                  # (0x21-0xFE) except for comma (0x2C) or
+                                  # semicolon (0x3B).
+                                  r'(?:[ \t]([ \t\x21-\x2B\x2D-\x3A\x3C-\xFE]*))?')
+        nonce_re = re.compile(r"'nonce-([a-zA-Z0-9+/]{43})'")
+        expected_directives = [
+            'default-src',
+            'frame-ancestors',
+            'img-src',
+            'script-src',
+            'style-src',
+        ]
+        expected_expressions = [
+            sq('none'),
+            sq('self'),
+            'data:',
+        ]
+        directives = list()
+        expressions = list()
+        nonces = dict()
+
+        for directive in csp.split(';'):
+            match = directive_re.fullmatch(directive)
+            assert match is not None
+            name, value = match.groups()
+            assert name not in directives, name
+            directives.append(name)
+            for expression in value.split(' '):
+                if expression in expected_expressions:
+                    expressions.append(expression)
+                else:
+                    match = nonce_re.fullmatch(expression)
+                    assert match is not None, expression
+                    assert name not in nonces, name
+                    nonces[name] = match.group(1)
+        if has_nonce:
+            assert ['script-src', 'style-src'] == sorted(nonces.keys()), nonces.keys()
+            nonce = one(set(nonces.values()))
+        else:
+            assert nonces == {}, nonces
+            nonce = None
+        assert expected_directives == sorted(directives), sorted(directives)
+        assert expected_expressions == sorted(set(expressions)), sorted(set(expressions))
+        return nonce
+
     @classmethod
     def security_headers(cls) -> dict[str, str]:
         """
-        Headers added to every response from the app, as well as canned 4XX and
-        5XX responses from API Gateway. Use of these headers addresses known
-        security vulnerabilities.
+        Default values for headers added to every response from the app, as well
+        as canned 4XX and 5XX responses from API Gateway. Use of these headers
+        addresses known security vulnerabilities.
         """
         hsts_max_age = 60 * 60 * 24 * 365 * 2
         return {
-            'Content-Security-Policy': jw('default-src', sq('self')),
+            'Content-Security-Policy': cls.content_security_policy(),
             'Referrer-Policy': 'strict-origin-when-cross-origin',
             'Strict-Transport-Security': jw(f'max-age={hsts_max_age};',
                                             'includeSubDomains;',
@@ -217,11 +349,10 @@ def _security_headers_middleware(self, event, get_response):
         Add headers to the response
         """
         response = get_response(event)
-        response.headers.update(self.security_headers())
-        # FIXME: Add a CSP header with a nonce value to text/html responses
-        #        https://github.com/DataBiosphere/azul-private/issues/6
-        if response.headers.get('Content-Type') == 'text/html':
-            del response.headers['Content-Security-Policy']
+        # Add security headers to the response without overwriting any headers
+        # that might have been added already (e.g. Content-Security-Policy)
+        for k, v in self.security_headers().items():
+            response.headers.setdefault(k, v)
         view_function = self.routes[event.path][event.method].view_function
         cache_control = getattr(view_function, 'cache_control')
         response.headers['Cache-Control'] = cache_control
@@ -493,11 +624,14 @@ def _controller(self, controller_cls: Type[C], **kwargs) -> C:
         return controller_cls(app=self, **kwargs)
 
     def swagger_ui(self) -> Response:
-        swagger_ui_template = self.load_static_resource('swagger', 'swagger-ui.html.template.mustache')
+        file_name = 'swagger-ui.html.template.mustache'
+        template = self.load_static_resource('swagger', file_name)
         base_url = self.base_url
         redirect_url = furl(base_url).add(path='oauth2_redirect')
         deployment_url = furl(base_url).add(path='openapi')
-        swagger_ui_html = chevron.render(swagger_ui_template, {
+        nonce = self.csp_nonce()
+        html = chevron.render(template, {
+            'CSP_NONCE': json.dumps(nonce),
             'DEPLOYMENT_PATH': json.dumps(str(deployment_url.path)),
             'OAUTH2_CLIENT_ID': json.dumps(config.google_oauth2_client_id),
             'OAUTH2_REDIRECT_URL': json.dumps(str(redirect_url)),
@@ -507,8 +641,11 @@ def swagger_ui(self) -> Response:
             ])
         })
         return Response(status_code=200,
-                        headers={'Content-Type': 'text/html'},
-                        body=swagger_ui_html)
+                        headers={
+                            'Content-Type': 'text/html',
+                            'Content-Security-Policy': self.content_security_policy(nonce)
+                        },
+                        body=html)
 
     def swagger_resource(self, file_name: str) -> Response:
         if os.sep in file_name:

swagger/oauth2-redirect.html → swagger/oauth2-redirect.html.template.mustache

@@ -2,10 +2,10 @@
 <!doctype html>
 <html lang="en-US">
 <title>Swagger UI: OAuth2 Redirect</title>
-<body onload="run()">
+<body>
 </body>
 </html>
-<script>
+<script nonce={{{CSP_NONCE}}}>
     'use strict';
     function run () {
         var oauth2 = window.opener.swaggerUIRedirectOauth2;
@@ -66,4 +66,15 @@
         }
         window.close();
     }
+    // The CSP blocks JS event handlers from inline HTML markup, so instead of
+    // calling run() from the body tag's onload property, as a workaround it
+    // is added here with an event listener.
+    //
+    // See also:
+    //
+    // https://github.com/swagger-api/swagger-ui/issues/5720
+    //
+    window.addEventListener('DOMContentLoaded', function () {
+        run();
+    });
 </script>

swagger/swagger-ui.html.template.mustache

@@ -5,7 +5,7 @@
     <meta charset="UTF-8">
     <title>Swagger UI</title>
     <link rel="stylesheet" href="static/swagger-ui.css" crossorigin="anonymous">
-    <style>
+    <style nonce={{{CSP_NONCE}}}>
       html
       {
         box-sizing: border-box;
@@ -30,7 +30,7 @@
     <div id="swagger-ui"></div>
     <script src="static/swagger-ui-bundle.js"></script>
     <script src="static/swagger-ui-standalone-preset.js"></script>
-    <script>
+    <script nonce={{{CSP_NONCE}}}>
     window.onload = function() {
       // Adapted from https://github.com/swagger-api/swagger-ui/issues/3725#issuecomment-334899276
       const DisableTryItOutPlugin = function() {

test/integration_test.py

@@ -2006,10 +2006,10 @@ def test_response_security_headers(self):
                         response = requests.get(str(endpoint / path))
                         response.raise_for_status()
                         expected = AzulChaliceApp.security_headers() | expected_headers
-                        # FIXME: Add a CSP header with a nonce value to text/html responses
-                        #        https://github.com/DataBiosphere/azul-private/issues/6
-                        if path in ['/', '/oauth2_redirect']:
-                            del expected['Content-Security-Policy']
+                        nonce = AzulChaliceApp.validate_csp(response.headers['Content-Security-Policy'],
+                                                            has_nonce=path in ['/', '/oauth2_redirect'])
+                        expected_csp = AzulChaliceApp.content_security_policy(nonce)
+                        expected['Content-Security-Policy'] = expected_csp
                         self.assertIsSubset(expected.items(), response.headers.items())
 
     def test_default_4xx_response_headers(self):

test/test_doctests.py

@@ -7,6 +7,7 @@
 import azul.bigquery
 import azul.bytes
 import azul.caching
+import azul.chalice
 import azul.collections
 import azul.docker
 import azul.doctests
@@ -67,6 +68,7 @@ def load_tests(_loader, tests, _ignore):
         azul.bigquery,
         azul.bytes,
         azul.caching,
+        azul.chalice,
         azul.collections,
         azul.doctests,
         azul.docker,