a3eebf4c-c61c-4db7-816c-fdac1fa36466

Committed 19 Mar 2025 02:25PM CUT coverage: 85.075% (-0.06%) from 85.137%

Build # a3eebf4c-c61c-4db7-816c-fdac1fa36466

Build Type

Pull #5456

circleci

Committed by

groovecoder

Commit Message

for MPP-4020: add sp3_plans to backend and API

Pull Request Pull Request #5456: for MPP-4020: add sp3_plans to backend and API

Run Details

2439 of 3571 branches covered (68.3%)

Branch coverage included in aggregate %.

109 of 130 new or added lines in 6 files covered. (83.85%)

2 existing lines in 1 file now uncovered.

17130 of 19431 relevant lines covered (88.16%)

9.84 hits per line

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

84.43

/api/views/privaterelay.py

from logging import getLogger
from typing import Any, Literal

from django.conf import settings
from django.contrib.auth.models import User
from django.db.models.query import QuerySet
from django.urls.exceptions import NoReverseMatch

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.rest_framework import FilterSet
from drf_spectacular.utils import (
    OpenApiExample,
    OpenApiParameter,
    OpenApiRequest,
    OpenApiResponse,
    extend_schema,
)
from rest_framework.authentication import get_authorization_header
from rest_framework.decorators import (
    api_view,
    authentication_classes,
    permission_classes,
)
from rest_framework.exceptions import AuthenticationFailed, ErrorDetail, ParseError
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.status import HTTP_201_CREATED, HTTP_400_BAD_REQUEST
from rest_framework.viewsets import ModelViewSet
from waffle import get_waffle_flag_model
from waffle.models import Sample, Switch

from emails.utils import incr_if_enabled
from privaterelay.models import Profile
from privaterelay.plans import (
    PlanCountryLangMapping,
    get_bundle_country_language_mapping,
    get_phone_country_language_mapping,
    get_premium_country_language_mapping,
)
from privaterelay.sp3_plans import (
    SP3PlanCountryLangMapping,
    get_sp3_country_language_mapping,
)
from privaterelay.utils import get_countries_info_from_request_and_mapping

from ..authentication import get_fxa_uid_from_oauth_token
from ..permissions import CanManageFlags, IsOwner
from ..serializers.privaterelay import (
    FlagSerializer,
    ProfileSerializer,
    UserSerializer,
    WebcompatIssueSerializer,
)

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


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


@extend_schema(tags=["privaterelay"])
class FlagViewSet(ModelViewSet):
    """Feature flags."""

    serializer_class = FlagSerializer
    permission_classes = [IsAuthenticated, CanManageFlags]
    filterset_class = FlagFilter
    http_method_names = ["get", "post", "head", "patch"]

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


@extend_schema(tags=["privaterelay"])
class ProfileViewSet(ModelViewSet):
    """Relay user extended profile data."""

    serializer_class = ProfileSerializer
    permission_classes = [IsAuthenticated, IsOwner]
    http_method_names = ["get", "post", "head", "put", "patch"]

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


@extend_schema(tags=["privaterelay"])
class UserViewSet(ModelViewSet):
    """Relay user data stored in Django user model."""

    serializer_class = UserSerializer
    permission_classes = [IsAuthenticated, IsOwner]
    http_method_names = ["get", "head"]

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


@permission_classes([IsAuthenticated])
@extend_schema(
    tags=["privaterelay"],
    request=WebcompatIssueSerializer,
    examples=[
        OpenApiExample(
            "mask not accepted",
            {
                "issue_on_domain": "https://accounts.firefox.com",
                "user_agent": "Firefox",
                "email_mask_not_accepted": True,
                "add_on_visual_issue": False,
                "email_not_received": False,
                "other_issue": "",
            },
        )
    ],
    responses={
        "201": OpenApiResponse(description="Report was submitted"),
        "400": OpenApiResponse(description="Report was rejected due to errors."),
        "401": OpenApiResponse(description="Authentication required."),
    },
)
@api_view(["POST"])
def report_webcompat_issue(request):
    """Report a Relay issue from an extension or integration."""

    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(status=HTTP_201_CREATED)
    return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)


def _get_example_plan(plan: Literal["premium", "phones", "bundle"]) -> dict[str, Any]:
    prices = {
        "premium": {"monthly": 1.99, "yearly": 0.99},
        "phones": {"monthly": 4.99, "yearly": 4.99},
        "bundle": {"monthly": 6.99, "yearly": 6.99},
    }
    monthly_price = {
        "id": f"price_{plan.title()}Monthlyxxxx",
        "currency": "usd",
        "price": prices[plan]["monthly"],
    }
    yearly_price = {
        "id": f"price_{plan.title()}Yearlyxxxx",
        "currency": "usd",
        "price": prices[plan]["yearly"],
    }
    return {
        "country_code": "US",
        "countries": ["CA", "US"],
        "available_in_country": True,
        "plan_country_lang_mapping": {
            "CA": {
                "*": {
                    "monthly": monthly_price,
                    "yearly": yearly_price,
                }
            },
            "US": {
                "*": {
                    "monthly": monthly_price,
                    "yearly": yearly_price,
                }
            },
        },
    }


@extend_schema(
    tags=["privaterelay"],
    responses={
        "200": OpenApiResponse(
            {"type": "object"},
            description="Site parameters",
            examples=[
                OpenApiExample(
                    "relay.firefox.com (partial)",
                    {
                        "FXA_ORIGIN": "https://accounts.firefox.com",
                        "PERIODICAL_PREMIUM_PRODUCT_ID": "prod_XXXXXXXXXXXXXX",
                        "GOOGLE_ANALYTICS_ID": "UA-########-##",
                        "GA4_MEASUREMENT_ID": "G-XXXXXXXXX",
                        "BUNDLE_PRODUCT_ID": "prod_XXXXXXXXXXXXXX",
                        "PHONE_PRODUCT_ID": "prod_XXXXXXXXXXXXXX",
                        "PERIODICAL_PREMIUM_PLANS": _get_example_plan("premium"),
                        "PHONE_PLANS": _get_example_plan("phones"),
                        "BUNDLE_PLANS": _get_example_plan("bundle"),
                        "BASKET_ORIGIN": "https://basket.mozilla.org",
                        "WAFFLE_FLAGS": [
                            ["foxfood", False],
                            ["phones", True],
                            ["bundle", True],
                        ],
                        "WAFFLE_SWITCHES": [],
                        "WAFFLE_SAMPLES": [],
                        "MAX_MINUTES_TO_VERIFY_REAL_PHONE": 5,
                    },
                )
            ],
        )
    },
)
@api_view()
@permission_classes([AllowAny])
def runtime_data(request):
    """Get data needed to present the Relay dashboard to a vistor or user."""
    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]
    premium_plans: SP3PlanCountryLangMapping | PlanCountryLangMapping
    phone_plans: SP3PlanCountryLangMapping | PlanCountryLangMapping
    bundle_plans: SP3PlanCountryLangMapping | PlanCountryLangMapping
    if settings.USE_SUBPLAT3:
        premium_plans = get_sp3_country_language_mapping("premium")
        phone_plans = get_sp3_country_language_mapping("phones")
        bundle_plans = get_sp3_country_language_mapping("bundle")
    else:
        premium_plans = get_premium_country_language_mapping()
        phone_plans = get_phone_country_language_mapping()
        bundle_plans = get_bundle_country_language_mapping()
    return Response(
        {
            "FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
            "PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
            "GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
            "GA4_MEASUREMENT_ID": settings.GA4_MEASUREMENT_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, premium_plans
            ),
            "PHONE_PLANS": get_countries_info_from_request_and_mapping(
                request, phone_plans
            ),
            "BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
                request, bundle_plans
            ),
            "USE_SUBPLAT3": settings.USE_SUBPLAT3,
            "SUBPLAT3_HOST": settings.SUBPLAT3_HOST,
            "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
            ),
        }
    )


@extend_schema(
    tags=["privaterelay"],
    parameters=[
        OpenApiParameter(
            name="Authorization",
            required=True,
            location="header",
            examples=[OpenApiExample("bearer", "Bearer XXXX-ZZZZ")],
            description="FXA Bearer Token. Can not be set in browsable API.",
        )
    ],
    request=OpenApiRequest(),
    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."
            )
        ),
        404: OpenApiResponse(description="FXA did not return a user."),
        500: OpenApiResponse(description="No response from FXA server."),
    },
)
@api_view(["POST"])
@permission_classes([AllowAny])
@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
    # TODO: Implement an FXA token authentication class
    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(data={"detail": e.detail.title()}, status=e.status_code)
        else:
            return 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}"},
            timeout=settings.FXA_REQUESTS_TIMEOUT_SECONDS,
        )
        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(
                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(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(status=status_code)

1	from logging import getLogger	1✔
2	from typing import Any, Literal	1✔
3
4	from django.conf import settings	1✔
5	from django.contrib.auth.models import User	1✔
6	from django.db.models.query import QuerySet	1✔
7	from django.urls.exceptions import NoReverseMatch	1✔
8
9	import requests	1✔
10	from allauth.account.adapter import get_adapter as get_account_adapter	1✔
11	from allauth.socialaccount.adapter import get_adapter as get_social_adapter	1✔
12	from allauth.socialaccount.helpers import complete_social_login	1✔
13	from allauth.socialaccount.models import SocialAccount	1✔
14	from django_filters.rest_framework import FilterSet	1✔
15	from drf_spectacular.utils import (	1✔
16	OpenApiExample,
17	OpenApiParameter,
18	OpenApiRequest,
19	OpenApiResponse,
20	extend_schema,
21	)
22	from rest_framework.authentication import get_authorization_header	1✔
23	from rest_framework.decorators import (	1✔
24	api_view,
25	authentication_classes,
26	permission_classes,
27	)
28	from rest_framework.exceptions import AuthenticationFailed, ErrorDetail, ParseError	1✔
29	from rest_framework.permissions import AllowAny, IsAuthenticated	1✔
30	from rest_framework.response import Response	1✔
31	from rest_framework.status import HTTP_201_CREATED, HTTP_400_BAD_REQUEST	1✔
32	from rest_framework.viewsets import ModelViewSet	1✔
33	from waffle import get_waffle_flag_model	1✔
34	from waffle.models import Sample, Switch	1✔
35
36	from emails.utils import incr_if_enabled	1✔
37	from privaterelay.models import Profile	1✔
38	from privaterelay.plans import (	1✔
39	PlanCountryLangMapping,
40	get_bundle_country_language_mapping,
41	get_phone_country_language_mapping,
42	get_premium_country_language_mapping,
43	)
44	from privaterelay.sp3_plans import (	1✔
45	SP3PlanCountryLangMapping,
46	get_sp3_country_language_mapping,
47	)
48	from privaterelay.utils import get_countries_info_from_request_and_mapping	1✔
49
50	from ..authentication import get_fxa_uid_from_oauth_token	1✔
51	from ..permissions import CanManageFlags, IsOwner	1✔
52	from ..serializers.privaterelay import (	1✔
53	FlagSerializer,
54	ProfileSerializer,
55	UserSerializer,
56	WebcompatIssueSerializer,
57	)
58
59	logger = getLogger("events")	1✔
60	info_logger = getLogger("eventsinfo")	1✔
61	FXA_PROFILE_URL = (	1✔
62	f"{settings.SOCIALACCOUNT_PROVIDERS['fxa']['PROFILE_ENDPOINT']}/profile"
63	)
64
65
66	class FlagFilter(FilterSet):	1✔
67	class Meta:	1✔
68	model = get_waffle_flag_model()	1✔
69	fields = [	1✔
70	"name",
71	"everyone",
72	# "users",
73	# read-only
74	"id",
75	]
76
77
78	@extend_schema(tags=["privaterelay"])	1✔
79	class FlagViewSet(ModelViewSet):	1✔
80	"""Feature flags."""
81
82	serializer_class = FlagSerializer	1✔
83	permission_classes = [IsAuthenticated, CanManageFlags]	1✔
84	filterset_class = FlagFilter	1✔
85	http_method_names = ["get", "post", "head", "patch"]	1✔
86
87	def get_queryset(self):	1✔
88	flags = get_waffle_flag_model().objects	1✔
89	return flags	1✔
90
91
92	@extend_schema(tags=["privaterelay"])	1✔
93	class ProfileViewSet(ModelViewSet):	1✔
94	"""Relay user extended profile data."""
95
96	serializer_class = ProfileSerializer	1✔
97	permission_classes = [IsAuthenticated, IsOwner]	1✔
98	http_method_names = ["get", "post", "head", "put", "patch"]	1✔
99
100	def get_queryset(self) -> QuerySet[Profile]:	1✔
101	if isinstance(self.request.user, User):	1✔
102	return Profile.objects.filter(user=self.request.user)	1✔
103	return Profile.objects.none()	1✔
104
105
106	@extend_schema(tags=["privaterelay"])	1✔
107	class UserViewSet(ModelViewSet):	1✔
108	"""Relay user data stored in Django user model."""
109
110	serializer_class = UserSerializer	1✔
111	permission_classes = [IsAuthenticated, IsOwner]	1✔
112	http_method_names = ["get", "head"]	1✔
113
114	def get_queryset(self) -> QuerySet[User]:	1✔
115	if isinstance(self.request.user, User):	1!
116	return User.objects.filter(id=self.request.user.id)	×
117	return User.objects.none()	1✔
118
119
120	@permission_classes([IsAuthenticated])	1✔
121	@extend_schema(	1✔
122	tags=["privaterelay"],
123	request=WebcompatIssueSerializer,
124	examples=[
125	OpenApiExample(
126	"mask not accepted",
127	{
128	"issue_on_domain": "https://accounts.firefox.com",
129	"user_agent": "Firefox",
130	"email_mask_not_accepted": True,
131	"add_on_visual_issue": False,
132	"email_not_received": False,
133	"other_issue": "",
134	},
135	)
136	],
137	responses={
138	"201": OpenApiResponse(description="Report was submitted"),
139	"400": OpenApiResponse(description="Report was rejected due to errors."),
140	"401": OpenApiResponse(description="Authentication required."),
141	},
142	)
143	@api_view(["POST"])	1✔
144	def report_webcompat_issue(request):	1✔
145	"""Report a Relay issue from an extension or integration."""
146
147	serializer = WebcompatIssueSerializer(data=request.data)	×
148	if serializer.is_valid():	×
149	info_logger.info("webcompat_issue", extra=serializer.data)	×
150	incr_if_enabled("webcompat_issue", 1)	×
151	for k, v in serializer.data.items():	×
152	if v and k != "issue_on_domain":	×
153	incr_if_enabled(f"webcompat_issue_{k}", 1)	×
154	return Response(status=HTTP_201_CREATED)	×
155	return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)	×
156
157
158	def _get_example_plan(plan: Literal["premium", "phones", "bundle"]) -> dict[str, Any]:	1✔
159	prices = {	1✔
160	"premium": {"monthly": 1.99, "yearly": 0.99},
161	"phones": {"monthly": 4.99, "yearly": 4.99},
162	"bundle": {"monthly": 6.99, "yearly": 6.99},
163	}
164	monthly_price = {	1✔
165	"id": f"price_{plan.title()}Monthlyxxxx",
166	"currency": "usd",
167	"price": prices[plan]["monthly"],
168	}
169	yearly_price = {	1✔
170	"id": f"price_{plan.title()}Yearlyxxxx",
171	"currency": "usd",
172	"price": prices[plan]["yearly"],
173	}
174	return {	1✔
175	"country_code": "US",
176	"countries": ["CA", "US"],
177	"available_in_country": True,
178	"plan_country_lang_mapping": {
179	"CA": {
180	"*": {
181	"monthly": monthly_price,
182	"yearly": yearly_price,
183	}
184	},
185	"US": {
186	"*": {
187	"monthly": monthly_price,
188	"yearly": yearly_price,
189	}
190	},
191	},
192	}
193
194
195	@extend_schema(	1✔
196	tags=["privaterelay"],
197	responses={
198	"200": OpenApiResponse(
199	{"type": "object"},
200	description="Site parameters",
201	examples=[
202	OpenApiExample(
203	"relay.firefox.com (partial)",
204	{
205	"FXA_ORIGIN": "https://accounts.firefox.com",
206	"PERIODICAL_PREMIUM_PRODUCT_ID": "prod_XXXXXXXXXXXXXX",
207	"GOOGLE_ANALYTICS_ID": "UA-########-##",
208	"GA4_MEASUREMENT_ID": "G-XXXXXXXXX",
209	"BUNDLE_PRODUCT_ID": "prod_XXXXXXXXXXXXXX",
210	"PHONE_PRODUCT_ID": "prod_XXXXXXXXXXXXXX",
211	"PERIODICAL_PREMIUM_PLANS": _get_example_plan("premium"),
212	"PHONE_PLANS": _get_example_plan("phones"),
213	"BUNDLE_PLANS": _get_example_plan("bundle"),
214	"BASKET_ORIGIN": "https://basket.mozilla.org",
215	"WAFFLE_FLAGS": [
216	["foxfood", False],
217	["phones", True],
218	["bundle", True],
219	],
220	"WAFFLE_SWITCHES": [],
221	"WAFFLE_SAMPLES": [],
222	"MAX_MINUTES_TO_VERIFY_REAL_PHONE": 5,
223	},
224	)
225	],
226	)
227	},
228	)
229	@api_view()	1✔
230	@permission_classes([AllowAny])	1✔
231	def runtime_data(request):	1✔
232	"""Get data needed to present the Relay dashboard to a vistor or user."""
233	flags = get_waffle_flag_model().get_all()	1✔
234	flag_values = [(f.name, f.is_active(request)) for f in flags]	1✔
235	switches = Switch.get_all()	1✔
236	switch_values = [(s.name, s.is_active()) for s in switches]	1✔
237	samples = Sample.get_all()	1✔
238	sample_values = [(s.name, s.is_active()) for s in samples]	1✔
239	premium_plans: SP3PlanCountryLangMapping \| PlanCountryLangMapping
240	phone_plans: SP3PlanCountryLangMapping \| PlanCountryLangMapping
241	bundle_plans: SP3PlanCountryLangMapping \| PlanCountryLangMapping
242	if settings.USE_SUBPLAT3:	1!
NEW 243	premium_plans = get_sp3_country_language_mapping("premium")	×
NEW 244	phone_plans = get_sp3_country_language_mapping("phones")	×
NEW 245	bundle_plans = get_sp3_country_language_mapping("bundle")	×
246	else:
247	premium_plans = get_premium_country_language_mapping()	1✔
248	phone_plans = get_phone_country_language_mapping()	1✔
249	bundle_plans = get_bundle_country_language_mapping()	1✔
250	return Response(	1✔
251	{
252	"FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
253	"PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
254	"GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
255	"GA4_MEASUREMENT_ID": settings.GA4_MEASUREMENT_ID,
256	"BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
257	"PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
258	"PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
259	request, premium_plans
260	),
261	"PHONE_PLANS": get_countries_info_from_request_and_mapping(
262	request, phone_plans
263	),
264	"BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
265	request, bundle_plans
266	),
267	"USE_SUBPLAT3": settings.USE_SUBPLAT3,
268	"SUBPLAT3_HOST": settings.SUBPLAT3_HOST,
269	"BASKET_ORIGIN": settings.BASKET_ORIGIN,
270	"WAFFLE_FLAGS": flag_values,
271	"WAFFLE_SWITCHES": switch_values,
272	"WAFFLE_SAMPLES": sample_values,
273	"MAX_MINUTES_TO_VERIFY_REAL_PHONE": (
274	settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE
275	),
276	}
277	)
278
279
280	@extend_schema(	1✔
281	tags=["privaterelay"],
282	parameters=[
283	OpenApiParameter(
284	name="Authorization",
285	required=True,
286	location="header",
287	examples=[OpenApiExample("bearer", "Bearer XXXX-ZZZZ")],
288	description="FXA Bearer Token. Can not be set in browsable API.",
289	)
290	],
291	request=OpenApiRequest(),
292	responses={
293	201: OpenApiResponse(description="Created; returned when user is created."),
294	202: OpenApiResponse(
295	description="Accepted; returned when user already exists."
296	),
297	400: OpenApiResponse(
298	description=(
299	"Bad request; returned when request is missing Authorization: Bearer"
300	" header or token value."
301	)
302	),
303	401: OpenApiResponse(
304	description=(
305	"Unauthorized; returned when the FXA token is invalid or expired."
306	)
307	),
308	404: OpenApiResponse(description="FXA did not return a user."),
309	500: OpenApiResponse(description="No response from FXA server."),
310	},
311	)
312	@api_view(["POST"])	1✔
313	@permission_classes([AllowAny])	1✔
314	@authentication_classes([])	1✔
315	def terms_accepted_user(request):	1✔
316	"""
317	Create a Relay user from an FXA token.
318
319	See [API Auth doc][api-auth-doc] for details.
320
321	[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
322	""" # noqa: E501
323	# Setting authentication_classes to empty due to
324	# authentication still happening despite permissions being set to allowany
325	# https://forum.djangoproject.com/t/solved-allowany-override-does-not-work-on-apiview/9754
326	# TODO: Implement an FXA token authentication class
327	authorization = get_authorization_header(request).decode()	1✔
328	if not authorization or not authorization.startswith("Bearer "):	1✔
329	raise ParseError("Missing Bearer header.")	1✔
330
331	token = authorization.split(" ")[1]	1✔
332	if token == "":	1✔
333	raise ParseError("Missing FXA Token after 'Bearer'.")	1✔
334
335	try:	1✔
336	fxa_uid = get_fxa_uid_from_oauth_token(token, use_cache=False)	1✔
337	except AuthenticationFailed as e:	1✔
338	# AuthenticationFailed exception returns 403 instead of 401 because we are not
339	# using the proper config that comes with the authentication_classes. See:
340	# https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
341	if isinstance(e.detail, ErrorDetail):	1!
342	return Response(data={"detail": e.detail.title()}, status=e.status_code)	1✔
343	else:
UNCOV 344	return Response(data={"detail": e.get_full_details()}, status=e.status_code)	×
345	status_code = 201	1✔
346
347	try:	1✔
348	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
349	status_code = 202	1✔
350	except SocialAccount.DoesNotExist:	1✔
351	# User does not exist, create a new Relay user
352	fxa_profile_resp = requests.get(	1✔
353	FXA_PROFILE_URL,
354	headers={"Authorization": f"Bearer {token}"},
355	timeout=settings.FXA_REQUESTS_TIMEOUT_SECONDS,
356	)
357	if not (fxa_profile_resp.ok and fxa_profile_resp.content):	1✔
358	logger.error(	1✔
359	"terms_accepted_user: bad account profile response",
360	extra={
361	"status_code": fxa_profile_resp.status_code,
362	"content": fxa_profile_resp.content,
363	},
364	)
365	return Response(	1✔
366	data={"detail": "Did not receive a 200 response for account profile."},
367	status=500,
368	)
369
370	# This is not exactly the request object that FirefoxAccountsProvider expects,
371	# but it has all of the necessary attributes to initialize the Provider
372	provider = get_social_adapter().get_provider(request, "fxa")	1✔
373	# This may not save the new user that was created
374	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/base/provider.py#L44
375	social_login = provider.sociallogin_from_response(	1✔
376	request, fxa_profile_resp.json()
377	)
378	# Complete social login is called by callback, see
379	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/oauth/views.py#L118
380	# for what we are mimicking to create new SocialAccount, User, and Profile for
381	# the new Relay user from Firefox Since this is a Resource Provider/Server flow
382	# and are NOT a Relying Party (RP) of FXA No social token information is stored
383	# (no Social Token object created).
384	try:	1✔
385	complete_social_login(request, social_login)	1✔
386	# complete_social_login writes ['account_verified_email', 'user_created',
387	# '_auth_user_id', '_auth_user_backend', '_auth_user_hash'] on
388	# request.session which sets the cookie because complete_social_login does
389	# the "login" The user did not actually log in, logout to clear the session
390	if request.user.is_authenticated:	1!
391	get_account_adapter(request).logout(request)	1✔
392	except NoReverseMatch as e:	1✔
393	# TODO: use this logging to fix the underlying issue
394	# https://mozilla-hub.atlassian.net/browse/MPP-3473
395	if "socialaccount_signup" in e.args[0]:	1!
396	logger.error(	1✔
397	"socialaccount_signup_error",
398	extra={
399	"exception": str(e),
400	"fxa_uid": fxa_uid,
401	"social_login_state": social_login.state,
402	},
403	)
404	return Response(status=500)	1✔
UNCOV 405	raise e	×
406	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
407	# Indicate profile was created from the resource flow
408	profile = sa.user.profile	1✔
409	profile.created_by = "firefox_resource"	1✔
410	profile.save()	1✔
411	info_logger.info(	1✔
412	"terms_accepted_user",
413	extra={"social_account": sa.uid, "status_code": status_code},
414	)
415	return Response(status=status_code)	1✔

mozilla / fx-private-relay / a3eebf4c-c61c-4db7-816c-fdac1fa36466

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