d3128616-238d-446e-82c5-ab66cd38ceaf

Committed 09 May 2024 06:22PM UTC coverage: 84.07% (-0.6%) from 84.64%

Build # d3128616-238d-446e-82c5-ab66cd38ceaf

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #4684 from mozilla/enable-flak8-bandit-checks-mpp-3802

fix MPP-3802: stop ignoring bandit security checks

Run Details

3601 of 4734 branches covered (76.07%)

Branch coverage included in aggregate %.

74 of 158 new or added lines in 24 files covered. (46.84%)

5 existing lines in 5 files now uncovered.

14686 of 17018 relevant lines covered (86.3%)

10.86 hits per line

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

93.33

/api/views/emails.py

"""API views for emails"""

from logging import getLogger
from typing import Generic, TypeVar

from django.apps import apps
from django.conf import settings
from django.contrib.auth.models import User
from django.core.exceptions import ObjectDoesNotExist
from django.db.models.query import QuerySet
from django.template.loader import render_to_string

import django_ftl
from django_filters import rest_framework as filters
from drf_spectacular.utils import OpenApiResponse, extend_schema
from rest_framework.decorators import api_view, permission_classes, throttle_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.serializers import BaseSerializer
from rest_framework.status import (
    HTTP_201_CREATED,
    HTTP_400_BAD_REQUEST,
    HTTP_404_NOT_FOUND,
)
from rest_framework.throttling import UserRateThrottle
from rest_framework.viewsets import ModelViewSet
from waffle import flag_is_active

from emails.apps import EmailsConfig
from emails.models import DomainAddress, RelayAddress
from emails.utils import generate_from_header, ses_message_props
from emails.views import _get_address, wrap_html_email
from privaterelay.ftl_bundles import main as ftl_bundle
from privaterelay.utils import glean_logger

from ..permissions import IsOwner
from ..serializers.emails import (
    DomainAddressSerializer,
    FirstForwardedEmailSerializer,
    RelayAddressSerializer,
)
from . import SaveToRequestUser

logger = getLogger("events")


class RelayAddressFilter(filters.FilterSet):
    used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")

    class Meta:
        model = RelayAddress
        fields = [
            "enabled",
            "description",
            "generated_for",
            "block_list_emails",
            "used_on",
            # read-only
            "id",
            "address",
            "domain",
            "created_at",
            "last_modified_at",
            "last_used_at",
            "num_forwarded",
            "num_blocked",
            "num_spam",
        ]


_Address = TypeVar("_Address", RelayAddress, DomainAddress)


class AddressViewSet(Generic[_Address], SaveToRequestUser, ModelViewSet):
    def perform_create(self, serializer: BaseSerializer[_Address]) -> None:
        super().perform_create(serializer)
        if not serializer.instance:
            raise ValueError("serializer.instance must be truthy value.")
        glean_logger().log_email_mask_created(
            request=self.request,
            mask=serializer.instance,
            created_by_api=True,
        )

    def perform_update(self, serializer: BaseSerializer[_Address]) -> None:
        if not serializer.instance:
            raise ValueError("serializer.instance must be truthy value.")
        old_description = serializer.instance.description
        super().perform_update(serializer)
        new_description = serializer.instance.description
        if old_description != new_description:
            glean_logger().log_email_mask_label_updated(
                request=self.request, mask=serializer.instance
            )

    def perform_destroy(self, instance: _Address) -> None:
        user = instance.user
        is_random_mask = isinstance(instance, RelayAddress)
        super().perform_destroy(instance)
        glean_logger().log_email_mask_deleted(
            request=self.request,
            user=user,
            is_random_mask=is_random_mask,
        )


@extend_schema(tags=["emails"])
class RelayAddressViewSet(AddressViewSet[RelayAddress]):
    """An email address with a random name provided by Relay."""

    serializer_class = RelayAddressSerializer
    permission_classes = [IsAuthenticated, IsOwner]
    filterset_class = RelayAddressFilter

    def get_queryset(self) -> QuerySet[RelayAddress]:
        if isinstance(self.request.user, User):
            return RelayAddress.objects.filter(user=self.request.user)
        return RelayAddress.objects.none()


class DomainAddressFilter(filters.FilterSet):
    used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")

    class Meta:
        model = DomainAddress
        fields = [
            "enabled",
            "description",
            "block_list_emails",
            "used_on",
            # read-only
            "id",
            "address",
            "domain",
            "created_at",
            "last_modified_at",
            "last_used_at",
            "num_forwarded",
            "num_blocked",
            "num_spam",
        ]


@extend_schema(tags=["emails"])
class DomainAddressViewSet(AddressViewSet[DomainAddress]):
    """An email address with subdomain chosen by a Relay user."""

    serializer_class = DomainAddressSerializer
    permission_classes = [IsAuthenticated, IsOwner]
    filterset_class = DomainAddressFilter

    def get_queryset(self) -> QuerySet[DomainAddress]:
        if isinstance(self.request.user, User):
            return DomainAddress.objects.filter(user=self.request.user)
        return DomainAddress.objects.none()


class FirstForwardedEmailRateThrottle(UserRateThrottle):
    rate = settings.FIRST_EMAIL_RATE_LIMIT


@permission_classes([IsAuthenticated])
@extend_schema(
    tags=["emails"],
    request=FirstForwardedEmailSerializer,
    responses={
        201: OpenApiResponse(description="Email sent to user."),
        400: OpenApiResponse(description="Invalid mask."),
        401: OpenApiResponse(description="Authentication required."),
        403: OpenApiResponse(description="Flag 'free_user_onboarding' is required."),
        404: OpenApiResponse(description="Unable to find the mask."),
    },
)
@api_view(["POST"])
@throttle_classes([FirstForwardedEmailRateThrottle])
def first_forwarded_email(request):
    """
    Requires `free_user_onboarding` flag to be active for the user.

    Send the `first_forwarded_email.html` email to the user via a mask.
    See [/emails/first_forwarded_email](/emails/first_forwarded_email).

    Note: `mask` value must be a `RelayAddress` that belongs to the authenticated user.
    A `DomainAddress` will not work.
    """
    if not flag_is_active(request, "free_user_onboarding"):
        # Return Permission Denied error
        return Response(
            {"detail": "Requires free_user_onboarding waffle flag."}, status=403
        )

    serializer = FirstForwardedEmailSerializer(data=request.data)
    if not serializer.is_valid():
        return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)

    mask = str(serializer.data.get("mask"))
    user = request.user
    try:
        address = _get_address(mask)
        RelayAddress.objects.get(user=user, address=address)
    except ObjectDoesNotExist:
        return Response(f"{mask} does not exist for user.", status=HTTP_404_NOT_FOUND)
    profile = user.profile
    app_config = apps.get_app_config("emails")
    if not isinstance(app_config, EmailsConfig):
        raise TypeError("app_config must be type EmailsConfig")
    ses_client = app_config.ses_client
    if not ses_client:
        raise ValueError("ses_client must be truthy value.")
    if not settings.RELAY_FROM_ADDRESS:
        raise ValueError("settings.RELAY_FROM_ADDRESS must have a value.")
    with django_ftl.override(profile.language):
        translated_subject = ftl_bundle.format("forwarded-email-hero-header")
    first_forwarded_email_html = render_to_string(
        "emails/first_forwarded_email.html",
        {
            "SITE_ORIGIN": settings.SITE_ORIGIN,
        },
    )
    from_address = generate_from_header(settings.RELAY_FROM_ADDRESS, mask)
    wrapped_email = wrap_html_email(
        first_forwarded_email_html,
        profile.language,
        profile.has_premium,
        from_address,
        0,
    )
    ses_client.send_email(
        Destination={
            "ToAddresses": [user.email],
        },
        Source=from_address,
        Message={
            "Subject": ses_message_props(translated_subject),
            "Body": {
                "Html": ses_message_props(wrapped_email),
            },
        },
    )
    logger.info(f"Sent first_forwarded_email to user ID: {user.id}")
    return Response(status=HTTP_201_CREATED)

1	"""API views for emails"""
2
3	from logging import getLogger	1✔
4	from typing import Generic, TypeVar	1✔
5
6	from django.apps import apps	1✔
7	from django.conf import settings	1✔
8	from django.contrib.auth.models import User	1✔
9	from django.core.exceptions import ObjectDoesNotExist	1✔
10	from django.db.models.query import QuerySet	1✔
11	from django.template.loader import render_to_string	1✔
12
13	import django_ftl	1✔
14	from django_filters import rest_framework as filters	1✔
15	from drf_spectacular.utils import OpenApiResponse, extend_schema	1✔
16	from rest_framework.decorators import api_view, permission_classes, throttle_classes	1✔
17	from rest_framework.permissions import IsAuthenticated	1✔
18	from rest_framework.response import Response	1✔
19	from rest_framework.serializers import BaseSerializer	1✔
20	from rest_framework.status import (	1✔
21	HTTP_201_CREATED,
22	HTTP_400_BAD_REQUEST,
23	HTTP_404_NOT_FOUND,
24	)
25	from rest_framework.throttling import UserRateThrottle	1✔
26	from rest_framework.viewsets import ModelViewSet	1✔
27	from waffle import flag_is_active	1✔
28
29	from emails.apps import EmailsConfig	1✔
30	from emails.models import DomainAddress, RelayAddress	1✔
31	from emails.utils import generate_from_header, ses_message_props	1✔
32	from emails.views import _get_address, wrap_html_email	1✔
33	from privaterelay.ftl_bundles import main as ftl_bundle	1✔
34	from privaterelay.utils import glean_logger	1✔
35
36	from ..permissions import IsOwner	1✔
37	from ..serializers.emails import (	1✔
38	DomainAddressSerializer,
39	FirstForwardedEmailSerializer,
40	RelayAddressSerializer,
41	)
42	from . import SaveToRequestUser	1✔
43
44	logger = getLogger("events")	1✔
45
46
47	class RelayAddressFilter(filters.FilterSet):	1✔
48	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
49
50	class Meta:	1✔
51	model = RelayAddress	1✔
52	fields = [	1✔
53	"enabled",
54	"description",
55	"generated_for",
56	"block_list_emails",
57	"used_on",
58	# read-only
59	"id",
60	"address",
61	"domain",
62	"created_at",
63	"last_modified_at",
64	"last_used_at",
65	"num_forwarded",
66	"num_blocked",
67	"num_spam",
68	]
69
70
71	_Address = TypeVar("_Address", RelayAddress, DomainAddress)	1✔
72
73
74	class AddressViewSet(Generic[_Address], SaveToRequestUser, ModelViewSet):	1✔
75	def perform_create(self, serializer: BaseSerializer[_Address]) -> None:	1✔
76	super().perform_create(serializer)	1✔
77	if not serializer.instance:	1!
NEW 78	raise ValueError("serializer.instance must be truthy value.")	×
79	glean_logger().log_email_mask_created(	1✔
80	request=self.request,
81	mask=serializer.instance,
82	created_by_api=True,
83	)
84
85	def perform_update(self, serializer: BaseSerializer[_Address]) -> None:	1✔
86	if not serializer.instance:	1!
NEW 87	raise ValueError("serializer.instance must be truthy value.")	×
88	old_description = serializer.instance.description	1✔
89	super().perform_update(serializer)	1✔
90	new_description = serializer.instance.description	1✔
91	if old_description != new_description:	1✔
92	glean_logger().log_email_mask_label_updated(	1✔
93	request=self.request, mask=serializer.instance
94	)
95
96	def perform_destroy(self, instance: _Address) -> None:	1✔
97	user = instance.user	1✔
98	is_random_mask = isinstance(instance, RelayAddress)	1✔
99	super().perform_destroy(instance)	1✔
100	glean_logger().log_email_mask_deleted(	1✔
101	request=self.request,
102	user=user,
103	is_random_mask=is_random_mask,
104	)
105
106
107	@extend_schema(tags=["emails"])	1✔
108	class RelayAddressViewSet(AddressViewSet[RelayAddress]):	1✔
109	"""An email address with a random name provided by Relay."""
110
111	serializer_class = RelayAddressSerializer	1✔
112	permission_classes = [IsAuthenticated, IsOwner]	1✔
113	filterset_class = RelayAddressFilter	1✔
114
115	def get_queryset(self) -> QuerySet[RelayAddress]:	1✔
116	if isinstance(self.request.user, User):	1✔
117	return RelayAddress.objects.filter(user=self.request.user)	1✔
118	return RelayAddress.objects.none()	1✔
119
120
121	class DomainAddressFilter(filters.FilterSet):	1✔
122	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
123
124	class Meta:	1✔
125	model = DomainAddress	1✔
126	fields = [	1✔
127	"enabled",
128	"description",
129	"block_list_emails",
130	"used_on",
131	# read-only
132	"id",
133	"address",
134	"domain",
135	"created_at",
136	"last_modified_at",
137	"last_used_at",
138	"num_forwarded",
139	"num_blocked",
140	"num_spam",
141	]
142
143
144	@extend_schema(tags=["emails"])	1✔
145	class DomainAddressViewSet(AddressViewSet[DomainAddress]):	1✔
146	"""An email address with subdomain chosen by a Relay user."""
147
148	serializer_class = DomainAddressSerializer	1✔
149	permission_classes = [IsAuthenticated, IsOwner]	1✔
150	filterset_class = DomainAddressFilter	1✔
151
152	def get_queryset(self) -> QuerySet[DomainAddress]:	1✔
153	if isinstance(self.request.user, User):	1✔
154	return DomainAddress.objects.filter(user=self.request.user)	1✔
155	return DomainAddress.objects.none()	1✔
156
157
158	class FirstForwardedEmailRateThrottle(UserRateThrottle):	1✔
159	rate = settings.FIRST_EMAIL_RATE_LIMIT	1✔
160
161
162	@permission_classes([IsAuthenticated])	1✔
163	@extend_schema(	1✔
164	tags=["emails"],
165	request=FirstForwardedEmailSerializer,
166	responses={
167	201: OpenApiResponse(description="Email sent to user."),
168	400: OpenApiResponse(description="Invalid mask."),
169	401: OpenApiResponse(description="Authentication required."),
170	403: OpenApiResponse(description="Flag 'free_user_onboarding' is required."),
171	404: OpenApiResponse(description="Unable to find the mask."),
172	},
173	)
174	@api_view(["POST"])	1✔
175	@throttle_classes([FirstForwardedEmailRateThrottle])	1✔
176	def first_forwarded_email(request):	1✔
177	"""
178	Requires `free_user_onboarding` flag to be active for the user.
179
180	Send the `first_forwarded_email.html` email to the user via a mask.
181	See [/emails/first_forwarded_email](/emails/first_forwarded_email).
182
183	Note: `mask` value must be a `RelayAddress` that belongs to the authenticated user.
184	A `DomainAddress` will not work.
185	"""
186	if not flag_is_active(request, "free_user_onboarding"):	1✔
187	# Return Permission Denied error
188	return Response(	1✔
189	{"detail": "Requires free_user_onboarding waffle flag."}, status=403
190	)
191
192	serializer = FirstForwardedEmailSerializer(data=request.data)	1✔
193	if not serializer.is_valid():	1✔
194	return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)	1✔
195
196	mask = str(serializer.data.get("mask"))	1✔
197	user = request.user	1✔
198	try:	1✔
199	address = _get_address(mask)	1✔
200	RelayAddress.objects.get(user=user, address=address)	1✔
201	except ObjectDoesNotExist:	1✔
202	return Response(f"{mask} does not exist for user.", status=HTTP_404_NOT_FOUND)	1✔
203	profile = user.profile	1✔
204	app_config = apps.get_app_config("emails")	1✔
205	if not isinstance(app_config, EmailsConfig):	1!
NEW 206	raise TypeError("app_config must be type EmailsConfig")	×
207	ses_client = app_config.ses_client	1✔
208	if not ses_client:	1!
NEW 209	raise ValueError("ses_client must be truthy value.")	×
210	if not settings.RELAY_FROM_ADDRESS:	1!
NEW 211	raise ValueError("settings.RELAY_FROM_ADDRESS must have a value.")	×
212	with django_ftl.override(profile.language):	1✔
213	translated_subject = ftl_bundle.format("forwarded-email-hero-header")	1✔
214	first_forwarded_email_html = render_to_string(	1✔
215	"emails/first_forwarded_email.html",
216	{
217	"SITE_ORIGIN": settings.SITE_ORIGIN,
218	},
219	)
220	from_address = generate_from_header(settings.RELAY_FROM_ADDRESS, mask)	1✔
221	wrapped_email = wrap_html_email(	1✔
222	first_forwarded_email_html,
223	profile.language,
224	profile.has_premium,
225	from_address,
226	0,
227	)
228	ses_client.send_email(	1✔
229	Destination={
230	"ToAddresses": [user.email],
231	},
232	Source=from_address,
233	Message={
234	"Subject": ses_message_props(translated_subject),
235	"Body": {
236	"Html": ses_message_props(wrapped_email),
237	},
238	},
239	)
240	logger.info(f"Sent first_forwarded_email to user ID: {user.id}")	1✔
241	return Response(status=HTTP_201_CREATED)	1✔

mozilla / fx-private-relay / d3128616-238d-446e-82c5-ab66cd38ceaf

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