Repository: pallets/werkzeug
Pull request #1617 (2019-07-22 16:26:29)

Add a ContentSecurityPolicy datastructure
This should help make CSP headers easier to construct and read, by
adding structure for the directives. It is based on today's version of
https://www.w3.org/TR/CSP3/ .

davidism: Nice! Is this something [Flask-Talisman](https://github.com/GoogleCloudPlatform/flask-talisman) can take advantage of too? Is there anything in their implementation that would be useful here?
pgjones: I think it could be used by Flask-Talisman. I quite like the ability (in Flask-Talisman) to do something like `csp.default_src = ["'self'", "something"]` i.e. set a list. I thought about doing this in Werkzeug, but the complexity starts to grow very quickly - not every directive-value is a list, then 'unsafe-inline' is only valid in some directives etc... So I think this is a good place to start.
pgjones: @davidism are you happy with the API? (I plan to add the same API to Quart and hope to get Werkzeug sign-off).
davidism: Overall I'm fine with the design. Flask-Talisman has a few CSP features that I'd at least like to think about now even if they can be implemented later.

* The `Content-Security-Policy-Report-Only` header.
* Adding a per-response nonce to CSP sections.
davidism: @theacodes We're implementing a CSP header in Werkzeug, would love to hear any feedback you have.
theacodes: The API here looks fine for having a structured way of setting the CSP header.

Talisman can certainly be updated to use this, but I'd also be happy to see the need for Talisman either diminish or evaporate completely.
pgjones: @davidism I've added the Report-Only variant. This already allows a nonce to be set (although it requires some knowledge about how to format it), I think anything beyond this is best tied in with Flask (so as to inject the nonce into the HTML).

Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index e08883f0..1edbccf9 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -34,6 +34,8 @@ Unreleased
     such that ``X-Foo`` is the same as ``x-foo``. :pr:`1605`
 -   :meth:`http.dump_cookie` accepts ``'None'`` as a value for
     ``samesite``. :issue:`1549`
+-   Support the Content Security Policy header through the
+    `Response.content_security_policy` data structure. :pr:`1617`
 -   Optional request log highlighting with the development server is
     handled by Click instead of termcolor. :issue:`1235`
 -   Optional ad-hoc TLS support for the development server is handled
diff --git a/src/werkzeug/datastructures.py b/src/werkzeug/datastructures.py
index d43517e9..70a961fa 100644
--- a/src/werkzeug/datastructures.py
+++ b/src/werkzeug/datastructures.py
@@ -2012,6 +2012,97 @@ class ResponseCacheControl(_CacheControl):
 _CacheControl.cache_property = staticmethod(cache_property)
 
 
+def csp_property(key):
+    """Return a new property object for a content security policy header.
+    Useful if you want to add support for a csp extension in a
+    subclass.
+    """
+    return property(
+        lambda x: x._get_value(key),
+        lambda x, v: x._set_value(key, v),
+        lambda x: x._del_value(key),
+        "accessor for %r" % key,
+    )
+
+
+class ContentSecurityPolicy(UpdateDictMixin, dict):
+    """Subclass of a dict that stores values for a Content Security Policy
+    header. It has accessors for all the level 3 policies.
+
+    Because the csp directives in the HTTP header use dashes the
+    python descriptors use underscores for that.
+
+    To get a header of the :class:`ContentSecuirtyPolicy` object again
+    you can convert the object into a string or call the
+    :meth:`to_header` method.  If you plan to subclass it and add your
+    own items have a look at the sourcecode for that class.
+
+    .. versionadded:: 1.0.0
+       Support for Content Security Policy headers was added.
+
+    """
+
+    base_uri = csp_property("base-uri")
+    child_src = csp_property("child-src")
+    connect_src = csp_property("connect-src")
+    default_src = csp_property("default-src")
+    font_src = csp_property("font-src")
+    form_action = csp_property("form-action")
+    frame_ancestors = csp_property("frame-ancestors")
+    frame_src = csp_property("frame-src")
+    img_src = csp_property("img-src")
+    manifest_src = csp_property("manifest-src")
+    media_src = csp_property("media-src")
+    navigate_to = csp_property("navigate-to")
+    object_src = csp_property("object-src")
+    prefetch_src = csp_property("prefetch-src")
+    plugin_types = csp_property("plugin-types")
+    report_to = csp_property("report-to")
+    report_uri = csp_property("report-uri")
+    sandbox = csp_property("sandbox")
+    script_src = csp_property("script-src")
+    script_src_attr = csp_property("script-src-attr")
+    script_src_elem = csp_property("script-src-elem")
+    style_src = csp_property("style-src")
+    style_src_attr = csp_property("style-src-attr")
+    style_src_elem = csp_property("style-src-elem")
+    worker_src = csp_property("worker-src")
+
+    def __init__(self, values=(), on_update=None):
+        dict.__init__(self, values or ())
+        self.on_update = on_update
+        self.provided = values is not None
+
+    def _get_value(self, key):
+        """Used internally by the accessor properties."""
+        return self.get(key)
+
+    def _set_value(self, key, value):
+        """Used internally by the accessor properties."""
+        if value is None:
+            self.pop(key, None)
+        else:
+            self[key] = value
+
+    def _del_value(self, key):
+        """Used internally by the accessor properties."""
+        if key in self:
+            del self[key]
+
+    def to_header(self):
+        """Convert the stored values into a cache control header."""
+        return dump_csp_header(self)
+
+    def __str__(self):
+        return self.to_header()
+
+    def __repr__(self):
+        return "<%s %s>" % (
+            self.__class__.__name__,
+            " ".join("%s=%r" % (k, v) for k, v in sorted(self.items())),
+        )
+
+
 class CallbackDict(UpdateDictMixin, dict):
     """A dict that calls a function passed every time something is changed.
     The function is passed the dict instance.
@@ -2837,6 +2928,7 @@ class FileStorage(object):
 
 # circular dependencies
 from . import exceptions
+from .http import dump_csp_header
 from .http import dump_header
 from .http import dump_options_header
 from .http import generate_etag
diff --git a/src/werkzeug/http.py b/src/werkzeug/http.py
index 8b16351e..6d85ffb2 100644
--- a/src/werkzeug/http.py
+++ b/src/werkzeug/http.py
@@ -304,6 +304,19 @@ def dump_header(iterable, allow_token=True):
     return ", ".join(items)
 
 
+def dump_csp_header(header):
+    """Dump a Content Security Policy header.
+
+    These are structured into policies such as "default-src 'self';
+    script-src 'self'".
+
+    .. versionadded:: 1.0.0
+       Support for Content Security Policy headers was added.
+
+    """
+    return "; ".join("%s %s" % (key, value) for key, value in iteritems(header))
+
+
 def parse_list_header(value):
     """Parse lists as described by RFC 2068 Section 2.
 
@@ -504,6 +517,32 @@ def parse_cache_control_header(value, on_update=None, cls=None):
     return cls(parse_dict_header(value), on_update)
 
 
+def parse_csp_header(value, on_update=None, cls=None):
+    """Parse a Content Security Policy header.
+
+    .. versionadded:: 1.0.0
+       Support for Content Security Policy headers was added.
+
+    :param value: a csp header to be parsed.
+    :param on_update: an optional callable that is called every time a value
+                      on the object is changed.
+    :param cls: the class for the returned object.  By default
+                :class:`~werkzeug.datastructures.ContentSecurityPolicy` is used.
+    :return: a `cls` object.
+    """
+
+    if cls is None:
+        cls = ContentSecurityPolicy
+    items = []
+    for policy in value.split(";"):
+        policy = policy.strip()
+        # Ignore badly formatted policies (no space)
+        if " " in policy:
+            directive, value = policy.strip().split(" ", 1)
+            items.append((directive.strip(), value.strip()))
+    return cls(items, on_update)
+
+
 def parse_set_header(value, on_update=None):
     """Parse a set-like header and return a
     :class:`~werkzeug.datastructures.HeaderSet` object:
@@ -1244,6 +1283,7 @@ def is_byte_range_valid(start, stop, length):
 from .datastructures import Accept
 from .datastructures import Authorization
 from .datastructures import ContentRange
+from .datastructures import ContentSecurityPolicy
 from .datastructures import ETags
 from .datastructures import HeaderSet
 from .datastructures import IfRange
diff --git a/src/werkzeug/wrappers/common_descriptors.py b/src/werkzeug/wrappers/common_descriptors.py
index e4107ee0..f169959b 100644
--- a/src/werkzeug/wrappers/common_descriptors.py
+++ b/src/werkzeug/wrappers/common_descriptors.py
@@ -4,10 +4,12 @@ from datetime import timedelta
 from .._compat import string_types
 from ..datastructures import CallbackDict
 from ..http import dump_age
+from ..http import dump_csp_header
 from ..http import dump_header
 from ..http import dump_options_header
 from ..http import http_date
 from ..http import parse_age
+from ..http import parse_csp_header
 from ..http import parse_date
 from ..http import parse_options_header
 from ..http import parse_set_header
@@ -220,6 +222,23 @@ class CommonResponseDescriptorsMixin(object):
         modification of the entity-body in transit, but is not proof
         against malicious attacks.)""",
     )
+    content_security_policy = header_property(
+        "Content-Security-Policy",
+        None,
+        parse_csp_header,
+        dump_csp_header,
+        doc="""The Content-Security-Policy header adds an additional layer of
+        security to help detect and mitigate certain types of attacks.""",
+    )
+    content_security_policy_report_only = header_property(
+        "Content-Security-Policy-Report-Only",
+        None,
+        parse_csp_header,
+        dump_csp_header,
+        doc="""The Content-Security-Policy-Report-Only header adds a csp policy
+        that is not enforced but is reported thereby helping detect
+        certain types of attacks.""",
+    )
     date = header_property(
         "Date",
         None,


Test patch:
diff --git a/tests/test_datastructures.py b/tests/test_datastructures.py
index c930a67e..c5590348 100644
--- a/tests/test_datastructures.py
+++ b/tests/test_datastructures.py
@@ -985,6 +985,26 @@ class TestCacheControl(object):
         assert cc.no_cache is None
 
 
+class TestContentSecurityPolicy(object):
+    def test_construct(self):
+        csp = datastructures.ContentSecurityPolicy(
+            [("font-src", "'self'"), ("media-src", "*")]
+        )
+        assert csp.font_src == "'self'"
+        assert csp.media_src == "*"
+        policies = [policy.strip() for policy in csp.to_header().split(";")]
+        assert "font-src 'self'" in policies
+        assert "media-src *" in policies
+
+    def test_properties(self):
+        csp = datastructures.ContentSecurityPolicy()
+        csp.default_src = "* 'self' quart.com"
+        csp.img_src = "'none'"
+        policies = [policy.strip() for policy in csp.to_header().split(";")]
+        assert "default-src * 'self' quart.com" in policies
+        assert "img-src 'none'" in policies
+
+
 class TestAccept(object):
     storage_class = datastructures.Accept
 
diff --git a/tests/test_http.py b/tests/test_http.py
index 400fc9d7..3709daf0 100644
--- a/tests/test_http.py
+++ b/tests/test_http.py
@@ -118,6 +118,14 @@ class TestHTTPUtility(object):
         assert c.private is None
         assert c.to_header() == "no-cache"
 
+    def test_csp_header(self):
+        csp = http.parse_csp_header(
+            "default-src 'self'; script-src 'unsafe-inline' *; img-src"
+        )
+        assert csp.default_src == "'self'"
+        assert csp.script_src == "'unsafe-inline' *"
+        assert csp.img_src is None
+
     def test_authorization_header(self):
         a = http.parse_authorization_header("Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==")
         assert a.type == "basic"