f6cbc6bd-9e3e-4239-9493-f713ff70bac9

Committed 22 Dec 2023 09:15PM UTC coverage: 73.652% (+0.1%) from 73.532%

Build # f6cbc6bd-9e3e-4239-9493-f713ff70bac9

Build Type

push

circleci

Committed by

rafeerahman

Commit Message

Detection for OTP within emails and an api for retrieval

Run Details

1989 of 2943 branches covered (0.0%)

Branch coverage included in aggregate %.

68 of 78 new or added lines in 2 files covered. (87.18%)

1 existing line in 1 file now uncovered.

6341 of 8367 relevant lines covered (75.79%)

19.73 hits per line

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

73.81

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

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

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,
    ParseError,
)
from rest_framework.response import Response
from rest_framework.views import exception_handler

from allauth.account.adapter import get_adapter as get_account_adapter
from allauth.socialaccount.models import SocialAccount
from allauth.socialaccount.helpers import complete_social_login
from allauth.socialaccount.providers.fxa.provider import FirefoxAccountsProvider
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

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

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

    def get_queryset(self):
        return DomainAddress.objects.filter(user=self.request.user)

    def perform_create(self, serializer):
        try:
            serializer.save(user=self.request.user)
        except IntegrityError:
            domain_address = DomainAddress.objects.filter(
                user=self.request.user, address=serializer.validated_data.get("address")
            ).first()
            raise ConflictError(
                {"id": domain_address.id, "full_address": domain_address.full_address}
            )


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

    def get_queryset(self):
        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):
        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
    """
    # 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
        # Read more: https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
        return response.Response(
            data={"detail": e.detail.title()}, 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 necssary attributes to initiatlize the Provider
        provider = FirefoxAccountsProvider(request)
        # 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)
        # which is 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 = 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]
) -> Optional[Response]:
    """
    Add error information to response data.

    When the error is a RelayAPIException, these additional fields may be present and
    the information will be translated if an Accept-Language header is added to the request:

    error_code - A string identifying the error, for client-side translation
    error_context - Additional data needed for client-side translation
    """

    response = exception_handler(exc, context)

    if response and isinstance(exc, RelayAPIException):
        error_codes = exc.get_codes()
        error_context = exc.error_context()
        if isinstance(error_codes, str):
            response.data["error_code"] = error_codes

            # Build Fluent error ID
            ftl_id_sub = "api-error-"
            ftl_id_error = error_codes.replace("_", "-")
            ftl_id = ftl_id_sub + ftl_id_error

            # Replace default message with Fluent string
            response.data["detail"] = ftl_bundle.format(ftl_id, error_context)

        if error_context:
            response.data["error_context"] = error_context

        response.data["error_code"] = error_codes

    return response


@decorators.permission_classes([permissions.IsAuthenticated])
@decorators.api_view(["GET"])
def potential_otp_code_detected(request):
    otp_data = cache.get(f"{request.user.id}_otp_code")  # Expires after 120 seconds

    if otp_data:
        potential_code = otp_data["otp_code"]
        mask = otp_data["mask"]
        cache.delete(
            f"{request.user.id}_otp_code"
        )  # Deleting to avoid duplicate notifications when polling
        return response.Response(
            data={"potential_otp_code": potential_code, "mask": mask},
            status=status.HTTP_200_OK,
        )

    return response.Response(
        data={"detail": "No data was found"}, status=status.HTTP_404_NOT_FOUND
    )

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

mozilla / fx-private-relay / f6cbc6bd-9e3e-4239-9493-f713ff70bac9

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