025169fe-e218-46ea-80e3-61fc1356daea

Committed 03 Jul 2024 08:48PM UTC coverage: 85.416%. Remained the same

Build # 025169fe-e218-46ea-80e3-61fc1356daea

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

4081 of 5229 branches covered (78.05%)

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.

15915 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)
        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("csp_nonce", request.csp_nonce)	×
NEW 46	return response	×
47
48
49	class RedirectRootIfLoggedIn:	1✔
50	def __init__(self, get_response):	1✔
51	self.get_response = get_response	1✔
52
53	def __call__(self, request):	1✔
54	# To prevent showing a flash of the landing page when a user is logged
55	# in, use a server-side redirect to send them to the dashboard,
56	# rather than handling that on the client-side:
57	if request.path == "/" and settings.SESSION_COOKIE_NAME in request.COOKIES:	1!
58	query_string = (	×
59	"?" + request.META["QUERY_STRING"]
60	if request.META["QUERY_STRING"]
61	else ""
62	)
63	return redirect("accounts/profile/" + query_string)	×
64
65	response = self.get_response(request)	1✔
66	return response	1✔
67
68
69	class AddDetectedCountryToRequestAndResponseHeaders:	1✔
70	def __init__(self, get_response):	1✔
71	self.get_response = get_response	1✔
72
73	def __call__(self, request):	1✔
74	region_key = "X-Client-Region"	1✔
75	region_dict = None	1✔
76	if region_key in request.headers:	1✔
77	region_dict = request.headers	1✔
78	if region_key in request.GET:	1!
79	region_dict = request.GET	×
80	if not region_dict:	1✔
81	return self.get_response(request)	1✔
82
83	country = region_dict.get(region_key)	1✔
84	request.country = country	1✔
85	response = self.get_response(request)	1✔
86	response.country = country	1✔
87	return response	1✔
88
89
90	class ResponseMetrics:	1✔
91
92	re_dockerflow = re.compile(r"/__(version\|heartbeat\|lbheartbeat)__/?$")	1✔
93
94	def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:	1✔
95	self.get_response = get_response	1✔
96	self.middleware = RelayStaticFilesMiddleware()	1✔
97
98	def __call__(self, request: HttpRequest) -> HttpResponse:	1✔
99	if not settings.STATSD_ENABLED:	1✔
100	return self.get_response(request)	1✔
101
102	start_time = time.time()	1✔
103	response = self.get_response(request)	1✔
104	delta = time.time() - start_time	1✔
105	view_name = self._get_metric_view_name(request)	1✔
106	metrics.timing(	1✔
107	"response",
108	value=delta * 1000.0,
109	tags=[
110	f"status:{response.status_code}",
111	f"view:{view_name}",
112	f"method:{request.method}",
113	],
114	)
115	return response	1✔
116
117	def _get_metric_view_name(self, request: HttpRequest) -> str:	1✔
118	if request.resolver_match:	1✔
119	view = request.resolver_match.func	1✔
120	if hasattr(view, "view_class"):	1✔
121	# Wrapped with rest_framework.decorators.api_view
122	return f"{view.__module__}.{view.view_class.__name__}"	1✔
123	return f"{view.__module__}.{view.__name__}"	1✔
124	if match := self.re_dockerflow.match(request.path_info):	1✔
125	return f"dockerflow.django.views.{match[1]}"	1✔
126	if self.middleware.is_staticfile(request.path_info):	1!
127	return "<static_file>"	1✔
128	return "<unknown_view>"	×
129
130
131	class StoreFirstVisit:	1✔
132	def __init__(self, get_response):	1✔
133	self.get_response = get_response	1✔
134
135	def __call__(self, request):	1✔
136	response = self.get_response(request)	1✔
137	first_visit = request.COOKIES.get("first_visit")	1✔
138	if first_visit is None and not request.user.is_anonymous:	1✔
139	response.set_cookie("first_visit", datetime.now(UTC))	1✔
140	return response	1✔
141
142
143	class RelayStaticFilesMiddleware(WhiteNoiseMiddleware):	1✔
144	"""Customize WhiteNoiseMiddleware for Relay.
145
146	The WhiteNoiseMiddleware serves static files and sets headers. In
147	production, the files are read from staticfiles/staticfiles.json,
148	and files with hashes in the name are treated as immutable with
149	10-year cache timeouts.
150
151	This class also treats Next.js output files (already hashed) as immutable.
152	"""
153
154	def immutable_file_test(self, path, url):	1✔
155	"""
156	Determine whether given URL represents an immutable file (i.e. a
157	file with a hash of its contents as part of its name) which can
158	therefore be cached forever.
159
160	All files outputed by next.js are hashed and immutable
161	"""
162	if not url.startswith(self.static_prefix):	1!
163	return False	×
164	name = url[len(self.static_prefix) :]	1✔
165	if name.startswith("_next/static/"):	1✔
166	return True	1✔
167	else:
168	return super().immutable_file_test(path, url)	1✔
169
170	def is_staticfile(self, path_info: str) -> bool:	1✔
171	"""
172	Returns True if this file is served by the middleware.
173
174	This uses the logic from whitenoise.middleware.WhiteNoiseMiddleware.__call__:
175	https://github.com/evansd/whitenoise/blob/220a98894495d407424e80d85d49227a5cf97e1b/src/whitenoise/middleware.py#L117-L124
176	"""
177	if self.autorefresh:	1!
178	static_file = self.find_file(path_info)	×
179	else:
180	static_file = self.files.get(path_info)	1✔
181	return static_file is not None	1✔

mozilla / fx-private-relay / 025169fe-e218-46ea-80e3-61fc1356daea

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