b2e067fe-ce4e-4099-9bef-07b368e99782

Committed 15 Apr 2024 04:18PM CUT coverage: 75.544% (+0.002%) from 75.542%

Build # b2e067fe-ce4e-4099-9bef-07b368e99782

Build Type

push

circleci

Committed by

jwhitlock

Commit Message

Enable pyupgrade, fix issues

Run Details

2443 of 3405 branches covered (71.75%)

Branch coverage included in aggregate %.

56 of 59 new or added lines in 14 files covered. (94.92%)

234 existing lines in 24 files now uncovered.

6793 of 8821 relevant lines covered (77.01%)

20.04 hits per line

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

81.4

/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 typing import Any, 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.template.loader import render_to_string
from django.urls.exceptions import NoReverseMatch

import django_ftl
import requests
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.helpers import complete_social_login
from allauth.socialaccount.models import SocialAccount
from django_filters import rest_framework as filters
from drf_spectacular.utils import OpenApiResponse, extend_schema
from rest_framework import (
    decorators,
    permissions,
    response,
    status,
    throttling,
    viewsets,
)
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 waffle import flag_is_active, get_waffle_flag_model
from waffle.models import Sample, Switch

from emails.apps import EmailsConfig
from emails.models import DomainAddress, Profile, RelayAddress
from emails.utils import generate_from_header, incr_if_enabled, ses_message_props
from emails.views import _get_address, wrap_html_email
from privaterelay.ftl_bundles import main as ftl_bundle
from privaterelay.plans import (
    get_bundle_country_language_mapping,
    get_phone_country_language_mapping,
    get_premium_country_language_mapping,
)
from privaterelay.utils import get_countries_info_from_request_and_mapping, glean_logger

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

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


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


class AddressViewSet(Generic[_Address], SaveToRequestUser, viewsets.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:
        mask_id = instance.metrics_id
        user = instance.user
        is_random_mask = isinstance(instance, RelayAddress)
        super().perform_destroy(instance)
        glean_logger().log_email_mask_deleted(
            request=self.request,
            user=user,
            mask_id=mask_id,
            is_random_mask=is_random_mask,
        )


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


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 = [permissions.IsAuthenticated, IsOwner]
    filterset_class = DomainAddressFilter

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


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

mozilla / fx-private-relay / b2e067fe-ce4e-4099-9bef-07b368e99782

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