3412a3a7-bfd9-41bc-a32b-a965e1ab1a7c

Committed 30 Apr 2024 03:03PM UTC coverage: 84.304% (+0.03%) from 84.273%

Build # 3412a3a7-bfd9-41bc-a32b-a965e1ab1a7c

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #4657 from mozilla/move-api-stuff-mpp-3799

MPP-3799: Split API views by emails and privaterelay

Run Details

3446 of 4504 branches covered (76.51%)

Branch coverage included in aggregate %.

814 of 852 new or added lines in 12 files covered. (95.54%)

1 existing line in 1 file now uncovered.

14488 of 16769 relevant lines covered (86.4%)

11.01 hits per line

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

75.0

/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 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)
        assert serializer.instance
        glean_logger().log_email_mask_created(
            request=self.request,
            mask=serializer.instance,
            created_by_api=True,
        )

    def perform_update(self, serializer: BaseSerializer[_Address]) -> None:
        assert serializer.instance is not None
        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,
        )


class RelayAddressViewSet(AddressViewSet[RelayAddress]):
    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",
        ]


class DomainAddressViewSet(AddressViewSet[DomainAddress]):
    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(methods=["POST"], request=FirstForwardedEmailSerializer)
@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")
    assert isinstance(app_config, EmailsConfig)
    ses_client = app_config.ses_client
    assert ses_client
    assert settings.RELAY_FROM_ADDRESS
    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 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	assert serializer.instance	1✔
78	glean_logger().log_email_mask_created(	1✔
79	request=self.request,
80	mask=serializer.instance,
81	created_by_api=True,
82	)
83
84	def perform_update(self, serializer: BaseSerializer[_Address]) -> None:	1✔
85	assert serializer.instance is not None	1✔
86	old_description = serializer.instance.description	1✔
87	super().perform_update(serializer)	1✔
88	new_description = serializer.instance.description	1✔
89	if old_description != new_description:	1✔
90	glean_logger().log_email_mask_label_updated(	1✔
91	request=self.request, mask=serializer.instance
92	)
93
94	def perform_destroy(self, instance: _Address) -> None:	1✔
95	user = instance.user	1✔
96	is_random_mask = isinstance(instance, RelayAddress)	1✔
97	super().perform_destroy(instance)	1✔
98	glean_logger().log_email_mask_deleted(	1✔
99	request=self.request,
100	user=user,
101	is_random_mask=is_random_mask,
102	)
103
104
105	class RelayAddressViewSet(AddressViewSet[RelayAddress]):	1✔
106	serializer_class = RelayAddressSerializer	1✔
107	permission_classes = [IsAuthenticated, IsOwner]	1✔
108	filterset_class = RelayAddressFilter	1✔
109
110	def get_queryset(self) -> QuerySet[RelayAddress]:	1✔
111	if isinstance(self.request.user, User):	1✔
112	return RelayAddress.objects.filter(user=self.request.user)	1✔
113	return RelayAddress.objects.none()	1✔
114
115
116	class DomainAddressFilter(filters.FilterSet):	1✔
117	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
118
119	class Meta:	1✔
120	model = DomainAddress	1✔
121	fields = [	1✔
122	"enabled",
123	"description",
124	"block_list_emails",
125	"used_on",
126	# read-only
127	"id",
128	"address",
129	"domain",
130	"created_at",
131	"last_modified_at",
132	"last_used_at",
133	"num_forwarded",
134	"num_blocked",
135	"num_spam",
136	]
137
138
139	class DomainAddressViewSet(AddressViewSet[DomainAddress]):	1✔
140	serializer_class = DomainAddressSerializer	1✔
141	permission_classes = [IsAuthenticated, IsOwner]	1✔
142	filterset_class = DomainAddressFilter	1✔
143
144	def get_queryset(self) -> QuerySet[DomainAddress]:	1✔
145	if isinstance(self.request.user, User):	1✔
146	return DomainAddress.objects.filter(user=self.request.user)	1✔
147	return DomainAddress.objects.none()	1✔
148
149
150	class FirstForwardedEmailRateThrottle(UserRateThrottle):	1✔
151	rate = settings.FIRST_EMAIL_RATE_LIMIT	1✔
152
153
154	@permission_classes([IsAuthenticated])	1✔
155	@extend_schema(methods=["POST"], request=FirstForwardedEmailSerializer)	1✔
156	@api_view(["POST"])	1✔
157	@throttle_classes([FirstForwardedEmailRateThrottle])	1✔
158	def first_forwarded_email(request):	1✔
159	"""
160	Requires `free_user_onboarding` flag to be active for the user.
161
162	Send the `first_forwarded_email.html` email to the user via a mask.
163	See [/emails/first_forwarded_email](/emails/first_forwarded_email).
164
165	Note: `mask` value must be a `RelayAddress` that belongs to the authenticated user.
166	A `DomainAddress` will not work.
167	"""
NEW 168	if not flag_is_active(request, "free_user_onboarding"):	×
169	# Return Permission Denied error
NEW 170	return Response(	×
171	{"detail": "Requires free_user_onboarding waffle flag."}, status=403
172	)
173
NEW 174	serializer = FirstForwardedEmailSerializer(data=request.data)	×
NEW 175	if not serializer.is_valid():	×
NEW 176	return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)	×
177
NEW 178	mask = str(serializer.data.get("mask"))	×
NEW 179	user = request.user	×
NEW 180	try:	×
NEW 181	address = _get_address(mask)	×
NEW 182	RelayAddress.objects.get(user=user, address=address)	×
NEW 183	except ObjectDoesNotExist:	×
NEW 184	return Response(f"{mask} does not exist for user.", status=HTTP_404_NOT_FOUND)	×
NEW 185	profile = user.profile	×
NEW 186	app_config = apps.get_app_config("emails")	×
NEW 187	assert isinstance(app_config, EmailsConfig)	×
NEW 188	ses_client = app_config.ses_client	×
NEW 189	assert ses_client	×
NEW 190	assert settings.RELAY_FROM_ADDRESS	×
NEW 191	with django_ftl.override(profile.language):	×
NEW 192	translated_subject = ftl_bundle.format("forwarded-email-hero-header")	×
NEW 193	first_forwarded_email_html = render_to_string(	×
194	"emails/first_forwarded_email.html",
195	{
196	"SITE_ORIGIN": settings.SITE_ORIGIN,
197	},
198	)
NEW 199	from_address = generate_from_header(settings.RELAY_FROM_ADDRESS, mask)	×
NEW 200	wrapped_email = wrap_html_email(	×
201	first_forwarded_email_html,
202	profile.language,
203	profile.has_premium,
204	from_address,
205	0,
206	)
NEW 207	ses_client.send_email(	×
208	Destination={
209	"ToAddresses": [user.email],
210	},
211	Source=from_address,
212	Message={
213	"Subject": ses_message_props(translated_subject),
214	"Body": {
215	"Html": ses_message_props(wrapped_email),
216	},
217	},
218	)
NEW 219	logger.info(f"Sent first_forwarded_email to user ID: {user.id}")	×
NEW 220	return Response(status=HTTP_201_CREATED)	×

mozilla / fx-private-relay / 3412a3a7-bfd9-41bc-a32b-a965e1ab1a7c

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