0bec9e2c-8c96-4752-b48e-8149870c8d44

Committed 10 Jan 2025 11:19PM CUT coverage: 85.356% (+0.3%) from 85.088%

Build # 0bec9e2c-8c96-4752-b48e-8149870c8d44

Build Type

Pull #5272

circleci

Committed by

jwhitlock

Commit Message

Remove permission checks from auth

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

Run Details

2487 of 3611 branches covered (68.87%)

Branch coverage included in aggregate %.

787 of 788 new or added lines in 6 files covered. (99.87%)

1 existing line in 1 file now uncovered.

17284 of 19552 relevant lines covered (88.4%)

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

mozilla / fx-private-relay / 0bec9e2c-8c96-4752-b48e-8149870c8d44

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