9865d2a1-d912-4dff-bfd6-a9b7da612c0e

Committed 20 Dec 2024 07:30PM UTC coverage: 85.344% (+0.3%) from 85.088%

Build # 9865d2a1-d912-4dff-bfd6-a9b7da612c0e

Build Type

Pull #5272

circleci

Committed by

jwhitlock

Commit Message

Fix assert_called_once_with, again

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

Run Details

2493 of 3619 branches covered (68.89%)

Branch coverage included in aggregate %.

764 of 765 new or added lines in 5 files covered. (99.87%)

2 existing lines in 2 files now uncovered.

17277 of 19546 relevant lines covered (88.39%)

9.8 hits per line

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

88.83

/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.authentication import get_authorization_header
from rest_framework.decorators import (
    api_view,
    authentication_classes,
    permission_classes,
)
from rest_framework.exceptions import ParseError
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, 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([AllowAny])
@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

    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'.")

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

    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.authentication import get_authorization_header	1✔
24	from rest_framework.decorators import (	1✔
25	api_view,
26	authentication_classes,
27	permission_classes,
28	)
29	from rest_framework.exceptions import ParseError	1✔
30	from rest_framework.permissions import AllowAny, IsAuthenticated	1✔
31	from rest_framework.request import Request	1✔
32	from rest_framework.response import Response	1✔
33	from rest_framework.status import HTTP_201_CREATED, HTTP_400_BAD_REQUEST	1✔
34	from rest_framework.viewsets import ModelViewSet	1✔
35	from waffle import get_waffle_flag_model	1✔
36	from waffle.models import Sample, Switch	1✔
37
38	from emails.utils import incr_if_enabled	1✔
39	from privaterelay.models import Profile	1✔
40	from privaterelay.plans import (	1✔
41	get_bundle_country_language_mapping,
42	get_phone_country_language_mapping,
43	get_premium_country_language_mapping,
44	)
45	from privaterelay.utils import get_countries_info_from_request_and_mapping	1✔
46
47	from ..authentication import (	1✔
48	FxaTokenAuthenticationRelayUserOptional,
49	IntrospectionResponse,
50	)
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	return Response(	1✔
240	{
241	"FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
242	"PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
243	"GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
244	"GA4_MEASUREMENT_ID": settings.GA4_MEASUREMENT_ID,
245	"BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
246	"PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
247	"PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
248	request, get_premium_country_language_mapping()
249	),
250	"PHONE_PLANS": get_countries_info_from_request_and_mapping(
251	request, get_phone_country_language_mapping()
252	),
253	"BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
254	request, get_bundle_country_language_mapping()
255	),
256	"BASKET_ORIGIN": settings.BASKET_ORIGIN,
257	"WAFFLE_FLAGS": flag_values,
258	"WAFFLE_SWITCHES": switch_values,
259	"WAFFLE_SAMPLES": sample_values,
260	"MAX_MINUTES_TO_VERIFY_REAL_PHONE": (
261	settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE
262	),
263	}
264	)
265
266
267	@extend_schema(	1✔
268	tags=["privaterelay"],
269	parameters=[
270	OpenApiParameter(
271	name="Authorization",
272	required=True,
273	location="header",
274	examples=[OpenApiExample("bearer", "Bearer XXXX-ZZZZ")],
275	description="FXA Bearer Token. Can not be set in browsable API.",
276	)
277	],
278	request=OpenApiRequest(),
279	responses={
280	201: OpenApiResponse(description="Created; returned when user is created."),
281	202: OpenApiResponse(
282	description="Accepted; returned when user already exists."
283	),
284	400: OpenApiResponse(
285	description=(
286	"Bad request; returned when request is missing Authorization: Bearer"
287	" header or token value."
288	)
289	),
290	401: OpenApiResponse(
291	description=(
292	"Unauthorized; returned when the FXA token is invalid or expired."
293	)
294	),
295	404: OpenApiResponse(description="FXA did not return a user."),
296	500: OpenApiResponse(description="No response from FXA server."),
297	},
298	)
299	@api_view(["POST"])	1✔
300	@permission_classes([AllowAny])	1✔
301	@authentication_classes([FxaTokenAuthenticationRelayUserOptional])	1✔
302	def terms_accepted_user(request: Request) -> Response:	1✔
303	"""
304	Create a Relay user from an FXA token.
305
306	See [API Auth doc][api-auth-doc] for details.
307
308	[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
309	""" # noqa: E501
310
311	authorization = get_authorization_header(request).decode()	1✔
312	if not authorization or not authorization.startswith("Bearer "):	1✔
313	raise ParseError("Missing Bearer header.")	1✔
314
315	token = authorization.split(" ")[1]	1✔
316	if token == "":	1!
UNCOV 317	raise ParseError("Missing FXA Token after 'Bearer'.")	×
318
319	user = request.user	1✔
320	introspect_response = request.auth	1✔
321	if not isinstance(introspect_response, IntrospectionResponse):	1!
NEW 322	raise ValueError(	×
323	"Expected request.auth to be IntrospectionResponse,"
324	f" got {type(IntrospectionResponse)}"
325	)
326	fxa_uid = introspect_response.fxa_id	1✔
327
328	existing_sa = False	1✔
329	action: str \| None = None	1✔
330	if isinstance(user, User):	1✔
331	socialaccount = SocialAccount.objects.get(user=user, provider="fxa")	1✔
332	existing_sa = True	1✔
333	action = "found_existing"	1✔
334
335	if not existing_sa:	1✔
336	# Get the user's profile
337	fxa_profile, response = _get_fxa_profile_from_bearer_token(token)	1✔
338	if response:	1✔
339	return response	1✔
340
341	# Since this takes time, see if another request created the SocialAccount
342	try:	1✔
343	socialaccount = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
344	existing_sa = True	1✔
345	action = "created_during_profile_fetch"	1✔
346	except SocialAccount.DoesNotExist:	1✔
347	pass	1✔
348
349	if not existing_sa:	1✔
350	# mypy type narrowing, but should always be true
351	if not isinstance(fxa_profile, dict): # pragma: no cover
352	raise ValueError(f"fxa_profile is {type(fxa_profile)}, not dict")
353
354	# Still no SocialAccount, attempt to create it
355	socialaccount, response = _create_socialaccount_from_bearer_token(	1✔
356	request, token, fxa_uid, fxa_profile
357	)
358	if response:	1✔
359	return response	1✔
360	action = "created"	1✔
361
362	status_code = 202 if existing_sa else 201	1✔
363	info_logger.info(	1✔
364	"terms_accepted_user",
365	extra={
366	"social_account": socialaccount.uid,
367	"status_code": status_code,
368	"action": action,
369	},
370	)
371	return Response(status=status_code)	1✔
372
373
374	def _create_socialaccount_from_bearer_token(	1✔
375	request: Request, token: str, fxa_uid: str, fxa_profile: dict[str, Any]
376	) -> tuple[SocialAccount, None] \| tuple[None, Response]:
377	"""Create a new Relay user with a SocialAccount from an FXA Bearer Token."""
378
379	# This is not exactly the request object that FirefoxAccountsProvider expects,
380	# but it has all of the necessary attributes to initialize the Provider
381	provider = get_social_adapter().get_provider(request, "fxa")	1✔
382
383	# This may not save the new user that was created
384	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/base/provider.py#L44
385	social_login = provider.sociallogin_from_response(request, fxa_profile)	1✔
386
387	# Complete social login is called by callback, see
388	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/oauth/views.py#L118
389	# for what we are mimicking to create new SocialAccount, User, and Profile for
390	# the new Relay user from Firefox Since this is a Resource Provider/Server flow
391	# and are NOT a Relying Party (RP) of FXA No social token information is stored
392	# (no Social Token object created).
393	try:	1✔
394	complete_social_login(request, social_login)	1✔
395	except (IntegrityError, NoReverseMatch) as e:	1✔
396	# TODO: use this logging to fix the underlying issue
397	# MPP-3473: NoReverseMatch socialaccount_signup
398	# Another user has the same email?
399	log_extra = {"exception": str(e), "social_login_state": social_login.state}	1✔
400	if fxa_profile.get("metricsEnabled", False):	1✔
401	log_extra["fxa_uid"] = fxa_uid	1✔
402	logger.error("socialaccount_signup_error", extra=log_extra)	1✔
403	return None, Response(status=500)	1✔
404
405	# complete_social_login writes ['account_verified_email', 'user_created',
406	# '_auth_user_id', '_auth_user_backend', '_auth_user_hash'] on
407	# request.session which sets the cookie because complete_social_login does
408	# the "login" The user did not actually log in, logout to clear the session
409	get_account_adapter(request).logout(request)	1✔
410
411	sa: SocialAccount = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
412
413	# Indicate profile was created from the resource flow
414	profile = sa.user.profile	1✔
415	profile.created_by = "firefox_resource"	1✔
416	profile.save()	1✔
417	return sa, None	1✔
418
419
420	def _get_fxa_profile_from_bearer_token(	1✔
421	token: str,
422	) -> tuple[dict[str, Any], None] \| tuple[None, Response]:
423	"""Use a bearer token to get the Mozilla Account user's profile data"""
424	# Use the bearer token
425	try:	1✔
426	fxa_profile_resp = requests.get(	1✔
427	FXA_PROFILE_URL,
428	headers={"Authorization": f"Bearer {token}"},
429	timeout=settings.FXA_REQUESTS_TIMEOUT_SECONDS,
430	)
431	except requests.Timeout:	1✔
432	return None, Response(	1✔
433	"Account profile request timeout, try again later.",
434	status=503,
435	)
436
437	if not (fxa_profile_resp.ok and fxa_profile_resp.content):	1✔
438	logger.error(	1✔
439	"terms_accepted_user: bad account profile response",
440	extra={
441	"status_code": fxa_profile_resp.status_code,
442	"content": fxa_profile_resp.content,
443	},
444	)
445	return None, Response(	1✔
446	data={"detail": "Did not receive a 200 response for account profile."},
447	status=500,
448	)
449	return fxa_profile_resp.json(), None	1✔

mozilla / fx-private-relay / 9865d2a1-d912-4dff-bfd6-a9b7da612c0e

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