4276dcd9-8c30-4786-8a41-f9c1cdae7f05

Committed 05 Mar 2024 07:15PM CUT coverage: 74.734% (+0.6%) from 74.139%

Build # 4276dcd9-8c30-4786-8a41-f9c1cdae7f05

Build Type

Pull #4452

circleci

Committed by

jwhitlock

Commit Message

Pass user to create_expected_glean_event

Pass the related user to the test helper create_expected_glean_event, so
that the user-specific values such as fxa_id and date_joined_relay can
be extracted in the helper rather than each test function.

Pull Request Pull Request #4452: MPP-3352: Add first Glean metrics to measure email mask usage

Run Details

2084 of 3047 branches covered (68.4%)

Branch coverage included in aggregate %.

251 of 256 new or added lines in 7 files covered. (98.05%)

79 existing lines in 3 files now uncovered.

6763 of 8791 relevant lines covered (76.93%)

20.12 hits per line

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

80.0

/api/views/__init__.py

"""
API views for emails and accounts

TODO: Move these functions to mirror the Django apps

Email stuff should be in api/views/emails.py
Runtime data should be in api/views/privaterelay.py
Profile stuff is strange - model is in emails, but probably should be in privaterelay.
"""

import logging
from django.core.exceptions import ObjectDoesNotExist
from django.template.loader import render_to_string
from django.urls.exceptions import NoReverseMatch
import requests
from typing import Any

from django.apps import apps
from django.conf import settings
from django.contrib.auth.models import User

import django_ftl
from drf_spectacular.utils import OpenApiResponse, extend_schema
from rest_framework.authentication import get_authorization_header
from rest_framework.exceptions import (
    AuthenticationFailed,
    ErrorDetail,
    ParseError,
)
from rest_framework.response import Response
from rest_framework.serializers import BaseSerializer
from rest_framework.views import exception_handler

from allauth.account.adapter import get_adapter as get_account_adapter
from allauth.socialaccount.adapter import get_adapter as get_social_adapter
from allauth.socialaccount.models import SocialAccount
from allauth.socialaccount.helpers import complete_social_login
from django_filters import rest_framework as filters
from waffle import flag_is_active, get_waffle_flag_model
from waffle.models import Switch, Sample
from rest_framework import (
    decorators,
    permissions,
    response,
    status,
    throttling,
    viewsets,
)
from emails.apps import EmailsConfig
from emails.utils import generate_from_header, incr_if_enabled, ses_message_props
from emails.views import wrap_html_email, _get_address

from privaterelay.plans import (
    get_bundle_country_language_mapping,
    get_premium_country_language_mapping,
    get_phone_country_language_mapping,
)
from privaterelay.utils import get_countries_info_from_request_and_mapping, glean_logger

from emails.models import (
    DomainAddress,
    Profile,
    RelayAddress,
)

from ..authentication import get_fxa_uid_from_oauth_token
from ..exceptions import RelayAPIException
from ..permissions import IsOwner, CanManageFlags
from ..serializers import (
    DomainAddressSerializer,
    FirstForwardedEmailSerializer,
    ProfileSerializer,
    RelayAddressSerializer,
    UserSerializer,
    FlagSerializer,
    WebcompatIssueSerializer,
)

from privaterelay.ftl_bundles import main as ftl_bundle

logger = logging.getLogger("events")
info_logger = logging.getLogger("eventsinfo")
FXA_PROFILE_URL = (
    f"{settings.SOCIALACCOUNT_PROVIDERS['fxa']['PROFILE_ENDPOINT']}/profile"
)


class SaveToRequestUser:
    def perform_create(self, serializer):
        assert hasattr(self, "request")
        assert hasattr(self.request, "user")
        serializer.save(user=self.request.user)


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",
        ]


class RelayAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):
    serializer_class = RelayAddressSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    filterset_class = RelayAddressFilter

    def get_queryset(self):
        assert isinstance(self.request.user, User)
        return RelayAddress.objects.filter(user=self.request.user)

    def perform_create(self, serializer: BaseSerializer[RelayAddress]) -> 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_destroy(self, instance: RelayAddress) -> None:
        mask_id = instance.metrics_id
        user = instance.user
        super().perform_destroy(instance)
        glean_logger().log_email_mask_deleted(
            request=self.request, user=user, mask_id=mask_id, is_random_mask=True
        )

    def perform_update(self, serializer: BaseSerializer[RelayAddress]) -> 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
            )


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(SaveToRequestUser, viewsets.ModelViewSet):
    serializer_class = DomainAddressSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    filterset_class = DomainAddressFilter

    def get_queryset(self):
        assert isinstance(self.request.user, User)
        return DomainAddress.objects.filter(user=self.request.user)

    def perform_create(self, serializer: BaseSerializer[DomainAddress]) -> None:
        super().perform_create(serializer)
        assert serializer.instance is not None
        glean_logger().log_email_mask_created(
            request=self.request,
            mask=serializer.instance,
            created_by_api=True,
        )

    def perform_destroy(self, instance: DomainAddress) -> None:
        mask_id = instance.metrics_id
        user = instance.user
        super().perform_destroy(instance)
        glean_logger().log_email_mask_deleted(
            request=self.request, user=user, mask_id=mask_id, is_random_mask=False
        )

    def perform_update(self, serializer: BaseSerializer[DomainAddress]) -> 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
            )


class ProfileViewSet(viewsets.ModelViewSet):
    serializer_class = ProfileSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    http_method_names = ["get", "post", "head", "put", "patch"]

    def get_queryset(self):
        assert isinstance(self.request.user, User)
        return Profile.objects.filter(user=self.request.user)


class UserViewSet(viewsets.ModelViewSet):
    serializer_class = UserSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    http_method_names = ["get", "head"]

    def get_queryset(self):
        assert isinstance(self.request.user, User)
        return User.objects.filter(id=self.request.user.id)


@extend_schema(
    responses={
        201: OpenApiResponse(description="Created; returned when user is created."),
        202: OpenApiResponse(
            description="Accepted; returned when user already exists."
        ),
        400: OpenApiResponse(
            description=(
                "Bad request; returned when request is missing Authorization: Bearer"
                " header or token value."
            )
        ),
        401: OpenApiResponse(
            description=(
                "Unauthorized; returned when the FXA token is invalid or expired."
            )
        ),
    },
)
@decorators.api_view(["POST"])
@decorators.permission_classes([permissions.AllowAny])
@decorators.authentication_classes([])
def terms_accepted_user(request):
    """
    Create a Relay user from an FXA token.

    See [API Auth doc][api-auth-doc] for details.

    [api-auth-doc]: https://github.com/mozilla/fx-private-relay/blob/main/docs/api_auth.md#firefox-oauth-token-authentication-and-accept-terms-of-service
    """  # noqa: E501
    # Setting authentication_classes to empty due to
    # authentication still happening despite permissions being set to allowany
    # https://forum.djangoproject.com/t/solved-allowany-override-does-not-work-on-apiview/9754
    authorization = get_authorization_header(request).decode()
    if not authorization or not authorization.startswith("Bearer "):
        raise ParseError("Missing Bearer header.")

    token = authorization.split(" ")[1]
    if token == "":
        raise ParseError("Missing FXA Token after 'Bearer'.")

    try:
        fxa_uid = get_fxa_uid_from_oauth_token(token, use_cache=False)
    except AuthenticationFailed as e:
        # AuthenticationFailed exception returns 403 instead of 401 because we are not
        # using the proper config that comes with the authentication_classes. See:
        # https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
        if isinstance(e.detail, ErrorDetail):
            return response.Response(
                data={"detail": e.detail.title()}, status=e.status_code
            )
        else:
            return response.Response(
                data={"detail": e.get_full_details()}, status=e.status_code
            )
    status_code = 201

    try:
        sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")
        status_code = 202
    except SocialAccount.DoesNotExist:
        # User does not exist, create a new Relay user
        fxa_profile_resp = requests.get(
            FXA_PROFILE_URL, headers={"Authorization": f"Bearer {token}"}
        )
        if not (fxa_profile_resp.ok and fxa_profile_resp.content):
            logger.error(
                "terms_accepted_user: bad account profile response",
                extra={
                    "status_code": fxa_profile_resp.status_code,
                    "content": fxa_profile_resp.content,
                },
            )
            return response.Response(
                data={"detail": "Did not receive a 200 response for account profile."},
                status=500,
            )

        # This is not exactly the request object that FirefoxAccountsProvider expects,
        # but it has all of the necessary attributes to initialize the Provider
        provider = get_social_adapter().get_provider(request, "fxa")
        # This may not save the new user that was created
        # https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/base/provider.py#L44
        social_login = provider.sociallogin_from_response(
            request, fxa_profile_resp.json()
        )
        # Complete social login is called by callback, see
        # https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/oauth/views.py#L118
        # for what we are mimicking to create new SocialAccount, User, and Profile for
        # the new Relay user from Firefox Since this is a Resource Provider/Server flow
        # and are NOT a Relying Party (RP) of FXA No social token information is stored
        # (no Social Token object created).
        try:
            complete_social_login(request, social_login)
            # complete_social_login writes ['account_verified_email', 'user_created',
            # '_auth_user_id', '_auth_user_backend', '_auth_user_hash'] on
            # request.session which sets the cookie because complete_social_login does
            # the "login" The user did not actually log in, logout to clear the session
            if request.user.is_authenticated:
                get_account_adapter(request).logout(request)
        except NoReverseMatch as e:
            # TODO: use this logging to fix the underlying issue
            # https://mozilla-hub.atlassian.net/browse/MPP-3473
            if "socialaccount_signup" in e.args[0]:
                logger.error(
                    "socialaccount_signup_error",
                    extra={
                        "exception": str(e),
                        "fxa_uid": fxa_uid,
                        "social_login_state": social_login.state,
                    },
                )
                return response.Response(status=500)
            raise e
        sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")
        # Indicate profile was created from the resource flow
        profile = sa.user.profile
        profile.created_by = "firefox_resource"
        profile.save()
    info_logger.info(
        "terms_accepted_user",
        extra={"social_account": sa.uid, "status_code": status_code},
    )
    return response.Response(status=status_code)


@decorators.api_view()
@decorators.permission_classes([permissions.AllowAny])
def runtime_data(request):
    flags = get_waffle_flag_model().get_all()
    flag_values = [(f.name, f.is_active(request)) for f in flags]
    switches = Switch.get_all()
    switch_values = [(s.name, s.is_active()) for s in switches]
    samples = Sample.get_all()
    sample_values = [(s.name, s.is_active()) for s in samples]
    return response.Response(
        {
            "FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
            "PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
            "GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
            "BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
            "PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
            "PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
                request, get_premium_country_language_mapping()
            ),
            "PHONE_PLANS": get_countries_info_from_request_and_mapping(
                request, get_phone_country_language_mapping()
            ),
            "BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
                request, get_bundle_country_language_mapping()
            ),
            "BASKET_ORIGIN": settings.BASKET_ORIGIN,
            "WAFFLE_FLAGS": flag_values,
            "WAFFLE_SWITCHES": switch_values,
            "WAFFLE_SAMPLES": sample_values,
            "MAX_MINUTES_TO_VERIFY_REAL_PHONE": (
                settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE
            ),
        }
    )


class FlagFilter(filters.FilterSet):
    class Meta:
        model = get_waffle_flag_model()
        fields = [
            "name",
            "everyone",
            # "users",
            # read-only
            "id",
        ]


class FlagViewSet(viewsets.ModelViewSet):
    serializer_class = FlagSerializer
    permission_classes = [permissions.IsAuthenticated, CanManageFlags]
    filterset_class = FlagFilter
    http_method_names = ["get", "post", "head", "patch"]

    def get_queryset(self):
        flags = get_waffle_flag_model().objects
        return flags


@decorators.permission_classes([permissions.IsAuthenticated])
@extend_schema(methods=["POST"], request=WebcompatIssueSerializer)
@decorators.api_view(["POST"])
def report_webcompat_issue(request):
    serializer = WebcompatIssueSerializer(data=request.data)
    if serializer.is_valid():
        info_logger.info("webcompat_issue", extra=serializer.data)
        incr_if_enabled("webcompat_issue", 1)
        for k, v in serializer.data.items():
            if v and k != "issue_on_domain":
                incr_if_enabled(f"webcompat_issue_{k}", 1)
        return response.Response(status=status.HTTP_201_CREATED)
    return response.Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)


class FirstForwardedEmailRateThrottle(throttling.UserRateThrottle):
    rate = settings.FIRST_EMAIL_RATE_LIMIT


@decorators.permission_classes([permissions.IsAuthenticated])
@extend_schema(methods=["POST"], request=FirstForwardedEmailSerializer)
@decorators.api_view(["POST"])
@decorators.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.Response(
            {"detail": "Requires free_user_onboarding waffle flag."}, status=403
        )

    serializer = FirstForwardedEmailSerializer(data=request.data)
    if not serializer.is_valid():
        return response.Response(serializer.errors, status=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.Response(
            f"{mask} does not exist for user.", status=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.Response(status=status.HTTP_201_CREATED)


def relay_exception_handler(exc: Exception, context: dict[str, Any]) -> Response | None:
    """
    Add error information to response data.

    When the error is a RelayAPIException, fields may be changed or added:

    detail - Translated to the best match from the request's Accept-Language header.
    error_code - A string identifying the error, for client-side translation.
    error_context - Additional data needed for client-side translation, if non-empty
    """
    response = exception_handler(exc, context)
    if response and isinstance(exc, RelayAPIException):
        response.data.update(exc.error_data())
    return response

1	"""
2	API views for emails and accounts
3
4	TODO: Move these functions to mirror the Django apps
5
6	Email stuff should be in api/views/emails.py
7	Runtime data should be in api/views/privaterelay.py
8	Profile stuff is strange - model is in emails, but probably should be in privaterelay.
9	"""
10
11	import logging	1✔
12	from django.core.exceptions import ObjectDoesNotExist	1✔
13	from django.template.loader import render_to_string	1✔
14	from django.urls.exceptions import NoReverseMatch	1✔
15	import requests	1✔
16	from typing import Any	1✔
17
18	from django.apps import apps	1✔
19	from django.conf import settings	1✔
20	from django.contrib.auth.models import User	1✔
21
22	import django_ftl	1✔
23	from drf_spectacular.utils import OpenApiResponse, extend_schema	1✔
24	from rest_framework.authentication import get_authorization_header	1✔
25	from rest_framework.exceptions import (	1✔
26	AuthenticationFailed,
27	ErrorDetail,
28	ParseError,
29	)
30	from rest_framework.response import Response	1✔
31	from rest_framework.serializers import BaseSerializer	1✔
32	from rest_framework.views import exception_handler	1✔
33
34	from allauth.account.adapter import get_adapter as get_account_adapter	1✔
35	from allauth.socialaccount.adapter import get_adapter as get_social_adapter	1✔
36	from allauth.socialaccount.models import SocialAccount	1✔
37	from allauth.socialaccount.helpers import complete_social_login	1✔
38	from django_filters import rest_framework as filters	1✔
39	from waffle import flag_is_active, get_waffle_flag_model	1✔
40	from waffle.models import Switch, Sample	1✔
41	from rest_framework import (	1✔
42	decorators,
43	permissions,
44	response,
45	status,
46	throttling,
47	viewsets,
48	)
49	from emails.apps import EmailsConfig	1✔
50	from emails.utils import generate_from_header, incr_if_enabled, ses_message_props	1✔
51	from emails.views import wrap_html_email, _get_address	1✔
52
53	from privaterelay.plans import (	1✔
54	get_bundle_country_language_mapping,
55	get_premium_country_language_mapping,
56	get_phone_country_language_mapping,
57	)
58	from privaterelay.utils import get_countries_info_from_request_and_mapping, glean_logger	1✔
59
60	from emails.models import (	1✔
61	DomainAddress,
62	Profile,
63	RelayAddress,
64	)
65
66	from ..authentication import get_fxa_uid_from_oauth_token	1✔
67	from ..exceptions import RelayAPIException	1✔
68	from ..permissions import IsOwner, CanManageFlags	1✔
69	from ..serializers import (	1✔
70	DomainAddressSerializer,
71	FirstForwardedEmailSerializer,
72	ProfileSerializer,
73	RelayAddressSerializer,
74	UserSerializer,
75	FlagSerializer,
76	WebcompatIssueSerializer,
77	)
78
79	from privaterelay.ftl_bundles import main as ftl_bundle	1✔
80
81	logger = logging.getLogger("events")	1✔
82	info_logger = logging.getLogger("eventsinfo")	1✔
83	FXA_PROFILE_URL = (	1✔
84	f"{settings.SOCIALACCOUNT_PROVIDERS['fxa']['PROFILE_ENDPOINT']}/profile"
85	)
86
87
88	class SaveToRequestUser:	1✔
89	def perform_create(self, serializer):	1✔
90	assert hasattr(self, "request")	1✔
91	assert hasattr(self.request, "user")	1✔
92	serializer.save(user=self.request.user)	1✔
93
94
95	class RelayAddressFilter(filters.FilterSet):	1✔
96	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
97
98	class Meta:	1✔
99	model = RelayAddress	1✔
100	fields = [	1✔
101	"enabled",
102	"description",
103	"generated_for",
104	"block_list_emails",
105	"used_on",
106	# read-only
107	"id",
108	"address",
109	"domain",
110	"created_at",
111	"last_modified_at",
112	"last_used_at",
113	"num_forwarded",
114	"num_blocked",
115	"num_spam",
116	]
117
118
119	class RelayAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):	1✔
120	serializer_class = RelayAddressSerializer	1✔
121	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
122	filterset_class = RelayAddressFilter	1✔
123
124	def get_queryset(self):	1✔
125	assert isinstance(self.request.user, User)	1✔
126	return RelayAddress.objects.filter(user=self.request.user)	1✔
127
128	def perform_create(self, serializer: BaseSerializer[RelayAddress]) -> None:	1✔
129	super().perform_create(serializer)	1✔
130	assert serializer.instance	1✔
131	glean_logger().log_email_mask_created(	1✔
132	request=self.request,
133	mask=serializer.instance,
134	created_by_api=True,
135	)
136
137	def perform_destroy(self, instance: RelayAddress) -> None:	1✔
138	mask_id = instance.metrics_id	1✔
139	user = instance.user	1✔
140	super().perform_destroy(instance)	1✔
141	glean_logger().log_email_mask_deleted(	1✔
142	request=self.request, user=user, mask_id=mask_id, is_random_mask=True
143	)
144
145	def perform_update(self, serializer: BaseSerializer[RelayAddress]) -> None:	1✔
146	assert serializer.instance is not None	1✔
147	old_description = serializer.instance.description	1✔
148	super().perform_update(serializer)	1✔
149	new_description = serializer.instance.description	1✔
150	if old_description != new_description:	1✔
151	glean_logger().log_email_mask_label_updated(	1✔
152	request=self.request, mask=serializer.instance
153	)
154
155
156	class DomainAddressFilter(filters.FilterSet):	1✔
157	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
158
159	class Meta:	1✔
160	model = DomainAddress	1✔
161	fields = [	1✔
162	"enabled",
163	"description",
164	"block_list_emails",
165	"used_on",
166	# read-only
167	"id",
168	"address",
169	"domain",
170	"created_at",
171	"last_modified_at",
172	"last_used_at",
173	"num_forwarded",
174	"num_blocked",
175	"num_spam",
176	]
177
178
179	class DomainAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):	1✔
180	serializer_class = DomainAddressSerializer	1✔
181	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
182	filterset_class = DomainAddressFilter	1✔
183
184	def get_queryset(self):	1✔
185	assert isinstance(self.request.user, User)	1✔
186	return DomainAddress.objects.filter(user=self.request.user)	1✔
187
188	def perform_create(self, serializer: BaseSerializer[DomainAddress]) -> None:	1✔
189	super().perform_create(serializer)	1✔
190	assert serializer.instance is not None	1✔
191	glean_logger().log_email_mask_created(	1✔
192	request=self.request,
193	mask=serializer.instance,
194	created_by_api=True,
195	)
196
197	def perform_destroy(self, instance: DomainAddress) -> None:	1✔
198	mask_id = instance.metrics_id	1✔
199	user = instance.user	1✔
200	super().perform_destroy(instance)	1✔
201	glean_logger().log_email_mask_deleted(	1✔
202	request=self.request, user=user, mask_id=mask_id, is_random_mask=False
203	)
204
205	def perform_update(self, serializer: BaseSerializer[DomainAddress]) -> None:	1✔
206	assert serializer.instance is not None	1✔
207	old_description = serializer.instance.description	1✔
208	super().perform_update(serializer)	1✔
209	new_description = serializer.instance.description	1✔
210	if old_description != new_description:	1✔
211	glean_logger().log_email_mask_label_updated(	1✔
212	request=self.request, mask=serializer.instance
213	)
214
215
216	class ProfileViewSet(viewsets.ModelViewSet):	1✔
217	serializer_class = ProfileSerializer	1✔
218	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
219	http_method_names = ["get", "post", "head", "put", "patch"]	1✔
220
221	def get_queryset(self):	1✔
222	assert isinstance(self.request.user, User)	1✔
223	return Profile.objects.filter(user=self.request.user)	1✔
224
225
226	class UserViewSet(viewsets.ModelViewSet):	1✔
227	serializer_class = UserSerializer	1✔
228	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
229	http_method_names = ["get", "head"]	1✔
230
231	def get_queryset(self):	1✔
232	assert isinstance(self.request.user, User)	1✔
UNCOV 233	return User.objects.filter(id=self.request.user.id)	×
234
235
236	@extend_schema(	1✔
237	responses={
238	201: OpenApiResponse(description="Created; returned when user is created."),
239	202: OpenApiResponse(
240	description="Accepted; returned when user already exists."
241	),
242	400: OpenApiResponse(
243	description=(
244	"Bad request; returned when request is missing Authorization: Bearer"
245	" header or token value."
246	)
247	),
248	401: OpenApiResponse(
249	description=(
250	"Unauthorized; returned when the FXA token is invalid or expired."
251	)
252	),
253	},
254	)
255	@decorators.api_view(["POST"])	1✔
256	@decorators.permission_classes([permissions.AllowAny])	1✔
257	@decorators.authentication_classes([])	1✔
258	def terms_accepted_user(request):	1✔
259	"""
260	Create a Relay user from an FXA token.
261
262	See [API Auth doc][api-auth-doc] for details.
263
264	[api-auth-doc]: https://github.com/mozilla/fx-private-relay/blob/main/docs/api_auth.md#firefox-oauth-token-authentication-and-accept-terms-of-service
265	""" # noqa: E501
266	# Setting authentication_classes to empty due to
267	# authentication still happening despite permissions being set to allowany
268	# https://forum.djangoproject.com/t/solved-allowany-override-does-not-work-on-apiview/9754
269	authorization = get_authorization_header(request).decode()	1✔
270	if not authorization or not authorization.startswith("Bearer "):	1✔
271	raise ParseError("Missing Bearer header.")	1✔
272
273	token = authorization.split(" ")[1]	1✔
274	if token == "":	1✔
275	raise ParseError("Missing FXA Token after 'Bearer'.")	1✔
276
277	try:	1✔
278	fxa_uid = get_fxa_uid_from_oauth_token(token, use_cache=False)	1✔
279	except AuthenticationFailed as e:	1✔
280	# AuthenticationFailed exception returns 403 instead of 401 because we are not
281	# using the proper config that comes with the authentication_classes. See:
282	# https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
283	if isinstance(e.detail, ErrorDetail):	1!
284	return response.Response(	1✔
285	data={"detail": e.detail.title()}, status=e.status_code
286	)
287	else:
UNCOV 288	return response.Response(	×
289	data={"detail": e.get_full_details()}, status=e.status_code
290	)
291	status_code = 201	1✔
292
293	try:	1✔
294	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
295	status_code = 202	1✔
296	except SocialAccount.DoesNotExist:	1✔
297	# User does not exist, create a new Relay user
298	fxa_profile_resp = requests.get(	1✔
299	FXA_PROFILE_URL, headers={"Authorization": f"Bearer {token}"}
300	)
301	if not (fxa_profile_resp.ok and fxa_profile_resp.content):	1✔
302	logger.error(	1✔
303	"terms_accepted_user: bad account profile response",
304	extra={
305	"status_code": fxa_profile_resp.status_code,
306	"content": fxa_profile_resp.content,
307	},
308	)
309	return response.Response(	1✔
310	data={"detail": "Did not receive a 200 response for account profile."},
311	status=500,
312	)
313
314	# This is not exactly the request object that FirefoxAccountsProvider expects,
315	# but it has all of the necessary attributes to initialize the Provider
316	provider = get_social_adapter().get_provider(request, "fxa")	1✔
317	# This may not save the new user that was created
318	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/base/provider.py#L44
319	social_login = provider.sociallogin_from_response(	1✔
320	request, fxa_profile_resp.json()
321	)
322	# Complete social login is called by callback, see
323	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/oauth/views.py#L118
324	# for what we are mimicking to create new SocialAccount, User, and Profile for
325	# the new Relay user from Firefox Since this is a Resource Provider/Server flow
326	# and are NOT a Relying Party (RP) of FXA No social token information is stored
327	# (no Social Token object created).
328	try:	1✔
329	complete_social_login(request, social_login)	1✔
330	# complete_social_login writes ['account_verified_email', 'user_created',
331	# '_auth_user_id', '_auth_user_backend', '_auth_user_hash'] on
332	# request.session which sets the cookie because complete_social_login does
333	# the "login" The user did not actually log in, logout to clear the session
334	if request.user.is_authenticated:	1!
335	get_account_adapter(request).logout(request)	1✔
336	except NoReverseMatch as e:	1✔
337	# TODO: use this logging to fix the underlying issue
338	# https://mozilla-hub.atlassian.net/browse/MPP-3473
339	if "socialaccount_signup" in e.args[0]:	1!
340	logger.error(	1✔
341	"socialaccount_signup_error",
342	extra={
343	"exception": str(e),
344	"fxa_uid": fxa_uid,
345	"social_login_state": social_login.state,
346	},
347	)
348	return response.Response(status=500)	1✔
UNCOV 349	raise e	×
350	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
351	# Indicate profile was created from the resource flow
352	profile = sa.user.profile	1✔
353	profile.created_by = "firefox_resource"	1✔
354	profile.save()	1✔
355	info_logger.info(	1✔
356	"terms_accepted_user",
357	extra={"social_account": sa.uid, "status_code": status_code},
358	)
359	return response.Response(status=status_code)	1✔
360
361
362	@decorators.api_view()	1✔
363	@decorators.permission_classes([permissions.AllowAny])	1✔
364	def runtime_data(request):	1✔
365	flags = get_waffle_flag_model().get_all()	1✔
366	flag_values = [(f.name, f.is_active(request)) for f in flags]	1✔
367	switches = Switch.get_all()	1✔
368	switch_values = [(s.name, s.is_active()) for s in switches]	1✔
369	samples = Sample.get_all()	1✔
370	sample_values = [(s.name, s.is_active()) for s in samples]	1✔
371	return response.Response(	1✔
372	{
373	"FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
374	"PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
375	"GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
376	"BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
377	"PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
378	"PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
379	request, get_premium_country_language_mapping()
380	),
381	"PHONE_PLANS": get_countries_info_from_request_and_mapping(
382	request, get_phone_country_language_mapping()
383	),
384	"BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
385	request, get_bundle_country_language_mapping()
386	),
387	"BASKET_ORIGIN": settings.BASKET_ORIGIN,
388	"WAFFLE_FLAGS": flag_values,
389	"WAFFLE_SWITCHES": switch_values,
390	"WAFFLE_SAMPLES": sample_values,
391	"MAX_MINUTES_TO_VERIFY_REAL_PHONE": (
392	settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE
393	),
394	}
395	)
396
397
398	class FlagFilter(filters.FilterSet):	1✔
399	class Meta:	1✔
400	model = get_waffle_flag_model()	1✔
401	fields = [	1✔
402	"name",
403	"everyone",
404	# "users",
405	# read-only
406	"id",
407	]
408
409
410	class FlagViewSet(viewsets.ModelViewSet):	1✔
411	serializer_class = FlagSerializer	1✔
412	permission_classes = [permissions.IsAuthenticated, CanManageFlags]	1✔
413	filterset_class = FlagFilter	1✔
414	http_method_names = ["get", "post", "head", "patch"]	1✔
415
416	def get_queryset(self):	1✔
417	flags = get_waffle_flag_model().objects	1✔
418	return flags	1✔
419
420
421	@decorators.permission_classes([permissions.IsAuthenticated])	1✔
422	@extend_schema(methods=["POST"], request=WebcompatIssueSerializer)	1✔
423	@decorators.api_view(["POST"])	1✔
424	def report_webcompat_issue(request):	1✔
UNCOV 425	serializer = WebcompatIssueSerializer(data=request.data)	×
UNCOV 426	if serializer.is_valid():	×
UNCOV 427	info_logger.info("webcompat_issue", extra=serializer.data)	×
UNCOV 428	incr_if_enabled("webcompat_issue", 1)	×
UNCOV 429	for k, v in serializer.data.items():	×
UNCOV 430	if v and k != "issue_on_domain":	×
UNCOV 431	incr_if_enabled(f"webcompat_issue_{k}", 1)	×
UNCOV 432	return response.Response(status=status.HTTP_201_CREATED)	×
UNCOV 433	return response.Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)	×
434
435
436	class FirstForwardedEmailRateThrottle(throttling.UserRateThrottle):	1✔
437	rate = settings.FIRST_EMAIL_RATE_LIMIT	1✔
438
439
440	@decorators.permission_classes([permissions.IsAuthenticated])	1✔
441	@extend_schema(methods=["POST"], request=FirstForwardedEmailSerializer)	1✔
442	@decorators.api_view(["POST"])	1✔
443	@decorators.throttle_classes([FirstForwardedEmailRateThrottle])	1✔
444	def first_forwarded_email(request):	1✔
445	"""
446	Requires `free_user_onboarding` flag to be active for the user.
447
448	Send the `first_forwarded_email.html` email to the user via a mask.
449	See [/emails/first_forwarded_email](/emails/first_forwarded_email).
450
451	Note: `mask` value must be a `RelayAddress` that belongs to the authenticated user.
452	A `DomainAddress` will not work.
453	"""
UNCOV 454	if not flag_is_active(request, "free_user_onboarding"):	×
455	# Return Permission Denied error
456	return response.Response(	×
457	{"detail": "Requires free_user_onboarding waffle flag."}, status=403
458	)
459
460	serializer = FirstForwardedEmailSerializer(data=request.data)	×
461	if not serializer.is_valid():	×
462	return response.Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)	×
463
UNCOV 464	mask = str(serializer.data.get("mask"))	×
UNCOV 465	user = request.user	×
UNCOV 466	try:	×
UNCOV 467	address = _get_address(mask)	×
UNCOV 468	RelayAddress.objects.get(user=user, address=address)	×
469	except ObjectDoesNotExist:	×
470	return response.Response(	×
471	f"{mask} does not exist for user.", status=status.HTTP_404_NOT_FOUND
472	)
UNCOV 473	profile = user.profile	×
UNCOV 474	app_config = apps.get_app_config("emails")	×
UNCOV 475	assert isinstance(app_config, EmailsConfig)	×
UNCOV 476	ses_client = app_config.ses_client	×
477	assert ses_client	×
UNCOV 478	assert settings.RELAY_FROM_ADDRESS	×
UNCOV 479	with django_ftl.override(profile.language):	×
UNCOV 480	translated_subject = ftl_bundle.format("forwarded-email-hero-header")	×
UNCOV 481	first_forwarded_email_html = render_to_string(	×
482	"emails/first_forwarded_email.html",
483	{
484	"SITE_ORIGIN": settings.SITE_ORIGIN,
485	},
486	)
UNCOV 487	from_address = generate_from_header(settings.RELAY_FROM_ADDRESS, mask)	×
UNCOV 488	wrapped_email = wrap_html_email(	×
489	first_forwarded_email_html,
490	profile.language,
491	profile.has_premium,
492	from_address,
493	0,
494	)
UNCOV 495	ses_client.send_email(	×
496	Destination={
497	"ToAddresses": [user.email],
498	},
499	Source=from_address,
500	Message={
501	"Subject": ses_message_props(translated_subject),
502	"Body": {
503	"Html": ses_message_props(wrapped_email),
504	},
505	},
506	)
UNCOV 507	logger.info(f"Sent first_forwarded_email to user ID: {user.id}")	×
UNCOV 508	return response.Response(status=status.HTTP_201_CREATED)	×
509
510
511	def relay_exception_handler(exc: Exception, context: dict[str, Any]) -> Response \| None:	1✔
512	"""
513	Add error information to response data.
514
515	When the error is a RelayAPIException, fields may be changed or added:
516
517	detail - Translated to the best match from the request's Accept-Language header.
518	error_code - A string identifying the error, for client-side translation.
519	error_context - Additional data needed for client-side translation, if non-empty
520	"""
521	response = exception_handler(exc, context)	1✔
522	if response and isinstance(exc, RelayAPIException):	1✔
523	response.data.update(exc.error_data())	1✔
524	return response	1✔

mozilla / fx-private-relay / 4276dcd9-8c30-4786-8a41-f9c1cdae7f05

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