ad546ed5-7760-4151-b12b-8748fc310dcd

Committed 12 Jul 2024 03:55PM UTC coverage: 85.419% (-0.01%) from 85.433%

Build # ad546ed5-7760-4151-b12b-8748fc310dcd

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #4854 from mozilla/ga4-csp-mpp-3838

MPP-3838: restore safer CSP

Run Details

4082 of 5229 branches covered (78.06%)

Branch coverage included in aggregate %.

14 of 17 new or added lines in 3 files covered. (82.35%)

1 existing line in 1 file now uncovered.

15912 of 18178 relevant lines covered (87.53%)

10.98 hits per line

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

87.4

/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")


# To find all the URL paths that serve HTML which need the CSP nonce:
# python manage.py collectstatic
# find staticfiles -type f -name 'index.html'
CSP_NONCE_COOKIE_PATHS = [
    "/",
    "/contains-tracker-warning/",
    "/flags/",
    "/faq/",
    "/vpn-relay/waitlist/",
    "/accounts/settings/",
    "/accounts/profile/",
    "/accounts/account_inactive/",
    "/vpn-relay-welcome/",
    "/phone/waitlist/",
    "/phone/",
    "/404/",
    "/tracker-report/",
    "/premium/waitlist/",
    "/premium/",
]


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

mozilla / fx-private-relay / ad546ed5-7760-4151-b12b-8748fc310dcd

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