eab89bcb-99db-4601-8163-24bb36d79cb1

Committed 04 Jul 2024 03:31PM UTC coverage: 85.425% (+0.009%) from 85.416%

Build # eab89bcb-99db-4601-8163-24bb36d79cb1

Build Type

push

circleci

Committed by

groovecoder

Commit Message

MPP-3838: restore safer CSP

Use a new EagerNonceCSPMiddleware to add nonce to the CSP and update the
React app to include it in dynamic scripts.

Run Details

4082 of 5229 branches covered (78.06%)

Branch coverage included in aggregate %.

17 of 19 new or added lines in 3 files covered. (89.47%)

1 existing line in 1 file now uncovered.

15916 of 18181 relevant lines covered (87.54%)

10.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.37

/privaterelay/middleware.py

import binascii
import os
import re
import time
from collections.abc import Callable
from datetime import UTC, datetime

from django.conf import settings
from django.http import HttpRequest, HttpResponse
from django.shortcuts import redirect

import markus
from csp.middleware import CSPMiddleware
from whitenoise.middleware import WhiteNoiseMiddleware

metrics = markus.get_metrics("fx-private-relay")


CSP_NONCE_COOKIE_URLS = ["/", "/premium", "/faq", "/accounts/profile/"]


class EagerNonceCSPMiddleware(CSPMiddleware):
    # We need a nonce to use Google Tag Manager with a safe CSP:
    # https://developers.google.com/tag-platform/security/guides/csp
    # django-csp only includes the nonce value in the CSP header if the csp_nonce
    # attribute is accessed:
    # https://django-csp.readthedocs.io/en/latest/nonce.html
    # That works for urls served by Django views that access the attribute but it
    # doesn't work for urls that are served by views which don't access the attribute.
    # (e.g., Whitenoise)
    # So, to ensure django-csp includes the nonce value in the CSP header of every
    # response, we override the default CSPMiddleware with this middleware. If the
    # request is for one of the HTML urls, this middleware sets the request.csp_nonce
    # attribute and adds a cookie for the React app to get the nonce value for scripts.
    def process_request(self, request):
        if request.path not in CSP_NONCE_COOKIE_URLS:
            pass
        request_nonce = binascii.hexlify(os.urandom(16)).decode("ascii")
        request.csp_nonce = request_nonce

    def process_response(self, request, response):
        response = super().process_response(request, response)
        if request.path not in CSP_NONCE_COOKIE_URLS:
            return response
        response.set_cookie(
            "csp_nonce", request.csp_nonce, secure=True, samesite="Strict"
        )
        return response


class RedirectRootIfLoggedIn:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # To prevent showing a flash of the landing page when a user is logged
        # in, use a server-side redirect to send them to the dashboard,
        # rather than handling that on the client-side:
        if request.path == "/" and settings.SESSION_COOKIE_NAME in request.COOKIES:
            query_string = (
                "?" + request.META["QUERY_STRING"]
                if request.META["QUERY_STRING"]
                else ""
            )
            return redirect("accounts/profile/" + query_string)

        response = self.get_response(request)
        return response


class AddDetectedCountryToRequestAndResponseHeaders:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        region_key = "X-Client-Region"
        region_dict = None
        if region_key in request.headers:
            region_dict = request.headers
        if region_key in request.GET:
            region_dict = request.GET
        if not region_dict:
            return self.get_response(request)

        country = region_dict.get(region_key)
        request.country = country
        response = self.get_response(request)
        response.country = country
        return response


class ResponseMetrics:

    re_dockerflow = re.compile(r"/__(version|heartbeat|lbheartbeat)__/?$")

    def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
        self.get_response = get_response
        self.middleware = RelayStaticFilesMiddleware()

    def __call__(self, request: HttpRequest) -> HttpResponse:
        if not settings.STATSD_ENABLED:
            return self.get_response(request)

        start_time = time.time()
        response = self.get_response(request)
        delta = time.time() - start_time
        view_name = self._get_metric_view_name(request)
        metrics.timing(
            "response",
            value=delta * 1000.0,
            tags=[
                f"status:{response.status_code}",
                f"view:{view_name}",
                f"method:{request.method}",
            ],
        )
        return response

    def _get_metric_view_name(self, request: HttpRequest) -> str:
        if request.resolver_match:
            view = request.resolver_match.func
            if hasattr(view, "view_class"):
                # Wrapped with rest_framework.decorators.api_view
                return f"{view.__module__}.{view.view_class.__name__}"
            return f"{view.__module__}.{view.__name__}"
        if match := self.re_dockerflow.match(request.path_info):
            return f"dockerflow.django.views.{match[1]}"
        if self.middleware.is_staticfile(request.path_info):
            return "<static_file>"
        return "<unknown_view>"


class StoreFirstVisit:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)
        first_visit = request.COOKIES.get("first_visit")
        if first_visit is None and not request.user.is_anonymous:
            response.set_cookie("first_visit", datetime.now(UTC))
        return response


class RelayStaticFilesMiddleware(WhiteNoiseMiddleware):
    """Customize WhiteNoiseMiddleware for Relay.

    The WhiteNoiseMiddleware serves static files and sets headers. In
    production, the files are read from staticfiles/staticfiles.json,
    and files with hashes in the name are treated as immutable with
    10-year cache timeouts.

    This class also treats Next.js output files (already hashed) as immutable.
    """

    def immutable_file_test(self, path, url):
        """
        Determine whether given URL represents an immutable file (i.e. a
        file with a hash of its contents as part of its name) which can
        therefore be cached forever.

        All files outputed by next.js are hashed and immutable
        """
        if not url.startswith(self.static_prefix):
            return False
        name = url[len(self.static_prefix) :]
        if name.startswith("_next/static/"):
            return True
        else:
            return super().immutable_file_test(path, url)

    def is_staticfile(self, path_info: str) -> bool:
        """
        Returns True if this file is served by the middleware.

        This uses the logic from whitenoise.middleware.WhiteNoiseMiddleware.__call__:
        https://github.com/evansd/whitenoise/blob/220a98894495d407424e80d85d49227a5cf97e1b/src/whitenoise/middleware.py#L117-L124
        """
        if self.autorefresh:
            static_file = self.find_file(path_info)
        else:
            static_file = self.files.get(path_info)
        return static_file is not None

1	import binascii	1✔
2	import os	1✔
3	import re	1✔
4	import time	1✔
5	from collections.abc import Callable	1✔
6	from datetime import UTC, datetime	1✔
7
8	from django.conf import settings	1✔
9	from django.http import HttpRequest, HttpResponse	1✔
10	from django.shortcuts import redirect	1✔
11
12	import markus	1✔
13	from csp.middleware import CSPMiddleware	1✔
14	from whitenoise.middleware import WhiteNoiseMiddleware	1✔
15
16	metrics = markus.get_metrics("fx-private-relay")	1✔
17
18
19	CSP_NONCE_COOKIE_URLS = ["/", "/premium", "/faq", "/accounts/profile/"]	1✔
20
21
22	class EagerNonceCSPMiddleware(CSPMiddleware):	1✔
23	# We need a nonce to use Google Tag Manager with a safe CSP:
24	# https://developers.google.com/tag-platform/security/guides/csp
25	# django-csp only includes the nonce value in the CSP header if the csp_nonce
26	# attribute is accessed:
27	# https://django-csp.readthedocs.io/en/latest/nonce.html
28	# That works for urls served by Django views that access the attribute but it
29	# doesn't work for urls that are served by views which don't access the attribute.
30	# (e.g., Whitenoise)
31	# So, to ensure django-csp includes the nonce value in the CSP header of every
32	# response, we override the default CSPMiddleware with this middleware. If the
33	# request is for one of the HTML urls, this middleware sets the request.csp_nonce
34	# attribute and adds a cookie for the React app to get the nonce value for scripts.
35	def process_request(self, request):	1✔
36	if request.path not in CSP_NONCE_COOKIE_URLS:	1!
37	pass	1✔
38	request_nonce = binascii.hexlify(os.urandom(16)).decode("ascii")	1✔
39	request.csp_nonce = request_nonce	1✔
40
41	def process_response(self, request, response):	1✔
42	response = super().process_response(request, response)	1✔
43	if request.path not in CSP_NONCE_COOKIE_URLS:	1!
44	return response	1✔
NEW 45	response.set_cookie(	×
46	"csp_nonce", request.csp_nonce, secure=True, samesite="Strict"
47	)
NEW 48	return response	×
49
50
51	class RedirectRootIfLoggedIn:	1✔
52	def __init__(self, get_response):	1✔
53	self.get_response = get_response	1✔
54
55	def __call__(self, request):	1✔
56	# To prevent showing a flash of the landing page when a user is logged
57	# in, use a server-side redirect to send them to the dashboard,
58	# rather than handling that on the client-side:
59	if request.path == "/" and settings.SESSION_COOKIE_NAME in request.COOKIES:	1!
60	query_string = (	×
61	"?" + request.META["QUERY_STRING"]
62	if request.META["QUERY_STRING"]
63	else ""
64	)
65	return redirect("accounts/profile/" + query_string)	×
66
67	response = self.get_response(request)	1✔
68	return response	1✔
69
70
71	class AddDetectedCountryToRequestAndResponseHeaders:	1✔
72	def __init__(self, get_response):	1✔
73	self.get_response = get_response	1✔
74
75	def __call__(self, request):	1✔
76	region_key = "X-Client-Region"	1✔
77	region_dict = None	1✔
78	if region_key in request.headers:	1✔
79	region_dict = request.headers	1✔
80	if region_key in request.GET:	1!
81	region_dict = request.GET	×
82	if not region_dict:	1✔
83	return self.get_response(request)	1✔
84
85	country = region_dict.get(region_key)	1✔
86	request.country = country	1✔
87	response = self.get_response(request)	1✔
88	response.country = country	1✔
89	return response	1✔
90
91
92	class ResponseMetrics:	1✔
93
94	re_dockerflow = re.compile(r"/__(version\|heartbeat\|lbheartbeat)__/?$")	1✔
95
96	def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:	1✔
97	self.get_response = get_response	1✔
98	self.middleware = RelayStaticFilesMiddleware()	1✔
99
100	def __call__(self, request: HttpRequest) -> HttpResponse:	1✔
101	if not settings.STATSD_ENABLED:	1✔
102	return self.get_response(request)	1✔
103
104	start_time = time.time()	1✔
105	response = self.get_response(request)	1✔
106	delta = time.time() - start_time	1✔
107	view_name = self._get_metric_view_name(request)	1✔
108	metrics.timing(	1✔
109	"response",
110	value=delta * 1000.0,
111	tags=[
112	f"status:{response.status_code}",
113	f"view:{view_name}",
114	f"method:{request.method}",
115	],
116	)
117	return response	1✔
118
119	def _get_metric_view_name(self, request: HttpRequest) -> str:	1✔
120	if request.resolver_match:	1✔
121	view = request.resolver_match.func	1✔
122	if hasattr(view, "view_class"):	1✔
123	# Wrapped with rest_framework.decorators.api_view
124	return f"{view.__module__}.{view.view_class.__name__}"	1✔
125	return f"{view.__module__}.{view.__name__}"	1✔
126	if match := self.re_dockerflow.match(request.path_info):	1✔
127	return f"dockerflow.django.views.{match[1]}"	1✔
128	if self.middleware.is_staticfile(request.path_info):	1!
129	return "<static_file>"	1✔
130	return "<unknown_view>"	×
131
132
133	class StoreFirstVisit:	1✔
134	def __init__(self, get_response):	1✔
135	self.get_response = get_response	1✔
136
137	def __call__(self, request):	1✔
138	response = self.get_response(request)	1✔
139	first_visit = request.COOKIES.get("first_visit")	1✔
140	if first_visit is None and not request.user.is_anonymous:	1✔
141	response.set_cookie("first_visit", datetime.now(UTC))	1✔
142	return response	1✔
143
144
145	class RelayStaticFilesMiddleware(WhiteNoiseMiddleware):	1✔
146	"""Customize WhiteNoiseMiddleware for Relay.
147
148	The WhiteNoiseMiddleware serves static files and sets headers. In
149	production, the files are read from staticfiles/staticfiles.json,
150	and files with hashes in the name are treated as immutable with
151	10-year cache timeouts.
152
153	This class also treats Next.js output files (already hashed) as immutable.
154	"""
155
156	def immutable_file_test(self, path, url):	1✔
157	"""
158	Determine whether given URL represents an immutable file (i.e. a
159	file with a hash of its contents as part of its name) which can
160	therefore be cached forever.
161
162	All files outputed by next.js are hashed and immutable
163	"""
164	if not url.startswith(self.static_prefix):	1!
165	return False	×
166	name = url[len(self.static_prefix) :]	1✔
167	if name.startswith("_next/static/"):	1✔
168	return True	1✔
169	else:
170	return super().immutable_file_test(path, url)	1✔
171
172	def is_staticfile(self, path_info: str) -> bool:	1✔
173	"""
174	Returns True if this file is served by the middleware.
175
176	This uses the logic from whitenoise.middleware.WhiteNoiseMiddleware.__call__:
177	https://github.com/evansd/whitenoise/blob/220a98894495d407424e80d85d49227a5cf97e1b/src/whitenoise/middleware.py#L117-L124
178	"""
179	if self.autorefresh:	1!
180	static_file = self.find_file(path_info)	×
181	else:
182	static_file = self.files.get(path_info)	1✔
183	return static_file is not None	1✔

mozilla / fx-private-relay / eab89bcb-99db-4601-8163-24bb36d79cb1

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous