2b84d1ac-a8ac-471e-b557-7783076ddcf9

Committed 10 Jan 2025 10:22PM CUT coverage: 85.372% (+0.3%) from 85.088%

Build # 2b84d1ac-a8ac-471e-b557-7783076ddcf9

Build Type

Pull #5272

circleci

Committed by

jwhitlock

Commit Message

Add HasValidFxaToken permission

Pull Request Pull Request #5272: WIP MPP-3505: RewriteAccount Bearer Token Authentication

Run Details

2492 of 3615 branches covered (68.93%)

Branch coverage included in aggregate %.

791 of 792 new or added lines in 6 files covered. (99.87%)

4 existing lines in 1 file now uncovered.

17293 of 19560 relevant lines covered (88.41%)

9.79 hits per line

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

89.27

/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 import IntegrityError
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.decorators import (
    api_view,
    authentication_classes,
    permission_classes,
)
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.request import Request
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 (
    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

from ..authentication import (
    FxaTokenAuthenticationRelayUserOptional,
    IntrospectionResponse,
)
from ..permissions import CanManageFlags, HasValidFxaToken, 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]
    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, 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
            ),
        }
    )


@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([HasValidFxaToken])
@authentication_classes([FxaTokenAuthenticationRelayUserOptional])
def terms_accepted_user(request: Request) -> Response:
    """
    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

    user = request.user
    introspect_response = request.auth
    if not isinstance(introspect_response, IntrospectionResponse):
        raise ValueError(
            "Expected request.auth to be IntrospectionResponse,"
            f" got {type(introspect_response)}"
        )
    fxa_uid = introspect_response.fxa_id
    token = introspect_response.token

    existing_sa = False
    action: str | None = None
    if isinstance(user, User):
        socialaccount = SocialAccount.objects.get(user=user, provider="fxa")
        existing_sa = True
        action = "found_existing"

    if not existing_sa:
        # Get the user's profile
        fxa_profile, response = _get_fxa_profile_from_bearer_token(token)
        if response:
            return response

        # Since this takes time, see if another request created the SocialAccount
        try:
            socialaccount = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")
            existing_sa = True
            action = "created_during_profile_fetch"
        except SocialAccount.DoesNotExist:
            pass

    if not existing_sa:
        # mypy type narrowing, but should always be true
        if not isinstance(fxa_profile, dict):  # pragma: no cover
            raise ValueError(f"fxa_profile is {type(fxa_profile)}, not dict")

        # Still no SocialAccount, attempt to create it
        socialaccount, response = _create_socialaccount_from_bearer_token(
            request, token, fxa_uid, fxa_profile
        )
        if response:
            return response
        action = "created"

    status_code = 202 if existing_sa else 201
    info_logger.info(
        "terms_accepted_user",
        extra={
            "social_account": socialaccount.uid,
            "status_code": status_code,
            "action": action,
        },
    )
    return Response(status=status_code)


def _create_socialaccount_from_bearer_token(
    request: Request, token: str, fxa_uid: str, fxa_profile: dict[str, Any]
) -> tuple[SocialAccount, None] | tuple[None, Response]:
    """Create a new Relay user with a SocialAccount from an FXA Bearer Token."""

    # 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)

    # 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)
    except (IntegrityError, NoReverseMatch) as e:
        # TODO: use this logging to fix the underlying issue
        # MPP-3473: NoReverseMatch socialaccount_signup
        #  Another user has the same email?
        log_extra = {"exception": str(e), "social_login_state": social_login.state}
        if fxa_profile.get("metricsEnabled", False):
            log_extra["fxa_uid"] = fxa_uid
        logger.error("socialaccount_signup_error", extra=log_extra)
        return None, Response(status=500)

    # 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
    get_account_adapter(request).logout(request)

    sa: SocialAccount = 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()
    return sa, None


def _get_fxa_profile_from_bearer_token(
    token: str,
) -> tuple[dict[str, Any], None] | tuple[None, Response]:
    """Use a bearer token to get the Mozilla Account user's profile data"""
    # Use the bearer token
    try:
        fxa_profile_resp = requests.get(
            FXA_PROFILE_URL,
            headers={"Authorization": f"Bearer {token}"},
            timeout=settings.FXA_REQUESTS_TIMEOUT_SECONDS,
        )
    except requests.Timeout:
        return None, Response(
            "Account profile request timeout, try again later.",
            status=503,
        )

    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 None, Response(
            data={"detail": "Did not receive a 200 response for account profile."},
            status=500,
        )
    return fxa_profile_resp.json(), None

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

mozilla / fx-private-relay / 2b84d1ac-a8ac-471e-b557-7783076ddcf9

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