6337c030-fd5d-4165-ba69-027771511104

Committed 06 May 2024 03:08PM UTC coverage: 84.36% (-0.2%) from 84.572%

Build # 6337c030-fd5d-4165-ba69-027771511104

Build Type

Pull #4664

circleci

Committed by

jwhitlock

Commit Message

Redo types for authentication tests

Pull Request Pull Request #4664: Enable `mypy --disallow-incomplete-defs`

Run Details

3461 of 4518 branches covered (76.6%)

Branch coverage included in aggregate %.

158 of 158 new or added lines in 27 files covered. (100.0%)

115 existing lines in 5 files now uncovered.

14538 of 16818 relevant lines covered (86.44%)

10.98 hits per line

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

87.64

/api/views/privaterelay.py

from logging import getLogger

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

import requests
from allauth.account.adapter import get_adapter as get_account_adapter
from allauth.socialaccount.adapter import get_adapter as get_social_adapter
from allauth.socialaccount.helpers import complete_social_login
from allauth.socialaccount.models import SocialAccount
from django_filters.rest_framework import FilterSet
from drf_spectacular.utils import OpenApiResponse, extend_schema
from rest_framework.authentication import get_authorization_header
from rest_framework.decorators import (
    api_view,
    authentication_classes,
    permission_classes,
)
from rest_framework.exceptions import AuthenticationFailed, ErrorDetail, ParseError
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.status import HTTP_201_CREATED, HTTP_400_BAD_REQUEST
from rest_framework.viewsets import ModelViewSet
from waffle import get_waffle_flag_model
from waffle.models import Sample, Switch

from emails.models import Profile
from emails.utils import incr_if_enabled
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 get_fxa_uid_from_oauth_token
from ..permissions import CanManageFlags, IsOwner
from ..serializers.privaterelay import (
    FlagSerializer,
    ProfileSerializer,
    UserSerializer,
    WebcompatIssueSerializer,
)

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


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


class FlagViewSet(ModelViewSet):
    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


class ProfileViewSet(ModelViewSet):
    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()


class UserViewSet(ModelViewSet):
    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(methods=["POST"], request=WebcompatIssueSerializer)
@api_view(["POST"])
def report_webcompat_issue(request):
    serializer = WebcompatIssueSerializer(data=request.data)
    if serializer.is_valid():
        info_logger.info("webcompat_issue", extra=serializer.data)
        incr_if_enabled("webcompat_issue", 1)
        for k, v in serializer.data.items():
            if v and k != "issue_on_domain":
                incr_if_enabled(f"webcompat_issue_{k}", 1)
        return Response(status=HTTP_201_CREATED)
    return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)


@api_view()
@permission_classes([AllowAny])
def runtime_data(request):
    flags = get_waffle_flag_model().get_all()
    flag_values = [(f.name, f.is_active(request)) for f in flags]
    switches = Switch.get_all()
    switch_values = [(s.name, s.is_active()) for s in switches]
    samples = Sample.get_all()
    sample_values = [(s.name, s.is_active()) for s in samples]
    return Response(
        {
            "FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
            "PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
            "GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
            "BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
            "PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
            "PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
                request, get_premium_country_language_mapping()
            ),
            "PHONE_PLANS": get_countries_info_from_request_and_mapping(
                request, get_phone_country_language_mapping()
            ),
            "BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
                request, get_bundle_country_language_mapping()
            ),
            "BASKET_ORIGIN": settings.BASKET_ORIGIN,
            "WAFFLE_FLAGS": flag_values,
            "WAFFLE_SWITCHES": switch_values,
            "WAFFLE_SAMPLES": sample_values,
            "MAX_MINUTES_TO_VERIFY_REAL_PHONE": (
                settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE
            ),
        }
    )


@extend_schema(
    responses={
        201: OpenApiResponse(description="Created; returned when user is created."),
        202: OpenApiResponse(
            description="Accepted; returned when user already exists."
        ),
        400: OpenApiResponse(
            description=(
                "Bad request; returned when request is missing Authorization: Bearer"
                " header or token value."
            )
        ),
        401: OpenApiResponse(
            description=(
                "Unauthorized; returned when the FXA token is invalid or expired."
            )
        ),
    },
)
@api_view(["POST"])
@permission_classes([AllowAny])
@authentication_classes([])
def terms_accepted_user(request):
    """
    Create a Relay user from an FXA token.

    See [API Auth doc][api-auth-doc] for details.

    [api-auth-doc]: https://github.com/mozilla/fx-private-relay/blob/main/docs/api_auth.md#firefox-oauth-token-authentication-and-accept-terms-of-service
    """  # noqa: E501
    # Setting authentication_classes to empty due to
    # authentication still happening despite permissions being set to allowany
    # https://forum.djangoproject.com/t/solved-allowany-override-does-not-work-on-apiview/9754
    authorization = get_authorization_header(request).decode()
    if not authorization or not authorization.startswith("Bearer "):
        raise ParseError("Missing Bearer header.")

    token = authorization.split(" ")[1]
    if token == "":
        raise ParseError("Missing FXA Token after 'Bearer'.")

    try:
        fxa_uid = get_fxa_uid_from_oauth_token(token, use_cache=False)
    except AuthenticationFailed as e:
        # AuthenticationFailed exception returns 403 instead of 401 because we are not
        # using the proper config that comes with the authentication_classes. See:
        # https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
        if isinstance(e.detail, ErrorDetail):
            return Response(data={"detail": e.detail.title()}, status=e.status_code)
        else:
            return Response(data={"detail": e.get_full_details()}, status=e.status_code)
    status_code = 201

    try:
        sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")
        status_code = 202
    except SocialAccount.DoesNotExist:
        # User does not exist, create a new Relay user
        fxa_profile_resp = requests.get(
            FXA_PROFILE_URL, headers={"Authorization": f"Bearer {token}"}
        )
        if not (fxa_profile_resp.ok and fxa_profile_resp.content):
            logger.error(
                "terms_accepted_user: bad account profile response",
                extra={
                    "status_code": fxa_profile_resp.status_code,
                    "content": fxa_profile_resp.content,
                },
            )
            return Response(
                data={"detail": "Did not receive a 200 response for account profile."},
                status=500,
            )

        # This is not exactly the request object that FirefoxAccountsProvider expects,
        # but it has all of the necessary attributes to initialize the Provider
        provider = get_social_adapter().get_provider(request, "fxa")
        # This may not save the new user that was created
        # https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/base/provider.py#L44
        social_login = provider.sociallogin_from_response(
            request, fxa_profile_resp.json()
        )
        # Complete social login is called by callback, see
        # https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/oauth/views.py#L118
        # for what we are mimicking to create new SocialAccount, User, and Profile for
        # the new Relay user from Firefox Since this is a Resource Provider/Server flow
        # and are NOT a Relying Party (RP) of FXA No social token information is stored
        # (no Social Token object created).
        try:
            complete_social_login(request, social_login)
            # complete_social_login writes ['account_verified_email', 'user_created',
            # '_auth_user_id', '_auth_user_backend', '_auth_user_hash'] on
            # request.session which sets the cookie because complete_social_login does
            # the "login" The user did not actually log in, logout to clear the session
            if request.user.is_authenticated:
                get_account_adapter(request).logout(request)
        except NoReverseMatch as e:
            # TODO: use this logging to fix the underlying issue
            # https://mozilla-hub.atlassian.net/browse/MPP-3473
            if "socialaccount_signup" in e.args[0]:
                logger.error(
                    "socialaccount_signup_error",
                    extra={
                        "exception": str(e),
                        "fxa_uid": fxa_uid,
                        "social_login_state": social_login.state,
                    },
                )
                return Response(status=500)
            raise e
        sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")
        # Indicate profile was created from the resource flow
        profile = sa.user.profile
        profile.created_by = "firefox_resource"
        profile.save()
    info_logger.info(
        "terms_accepted_user",
        extra={"social_account": sa.uid, "status_code": status_code},
    )
    return Response(status=status_code)

1	from logging import getLogger	1✔
2
3	from django.conf import settings	1✔
4	from django.contrib.auth.models import User	1✔
5	from django.db.models.query import QuerySet	1✔
6	from django.urls.exceptions import NoReverseMatch	1✔
7
8	import requests	1✔
9	from allauth.account.adapter import get_adapter as get_account_adapter	1✔
10	from allauth.socialaccount.adapter import get_adapter as get_social_adapter	1✔
11	from allauth.socialaccount.helpers import complete_social_login	1✔
12	from allauth.socialaccount.models import SocialAccount	1✔
13	from django_filters.rest_framework import FilterSet	1✔
14	from drf_spectacular.utils import OpenApiResponse, extend_schema	1✔
15	from rest_framework.authentication import get_authorization_header	1✔
16	from rest_framework.decorators import (	1✔
17	api_view,
18	authentication_classes,
19	permission_classes,
20	)
21	from rest_framework.exceptions import AuthenticationFailed, ErrorDetail, ParseError	1✔
22	from rest_framework.permissions import AllowAny, IsAuthenticated	1✔
23	from rest_framework.response import Response	1✔
24	from rest_framework.status import HTTP_201_CREATED, HTTP_400_BAD_REQUEST	1✔
25	from rest_framework.viewsets import ModelViewSet	1✔
26	from waffle import get_waffle_flag_model	1✔
27	from waffle.models import Sample, Switch	1✔
28
29	from emails.models import Profile	1✔
30	from emails.utils import incr_if_enabled	1✔
31	from privaterelay.plans import (	1✔
32	get_bundle_country_language_mapping,
33	get_phone_country_language_mapping,
34	get_premium_country_language_mapping,
35	)
36	from privaterelay.utils import get_countries_info_from_request_and_mapping	1✔
37
38	from ..authentication import get_fxa_uid_from_oauth_token	1✔
39	from ..permissions import CanManageFlags, IsOwner	1✔
40	from ..serializers.privaterelay import (	1✔
41	FlagSerializer,
42	ProfileSerializer,
43	UserSerializer,
44	WebcompatIssueSerializer,
45	)
46
47	logger = getLogger("events")	1✔
48	info_logger = getLogger("eventsinfo")	1✔
49	FXA_PROFILE_URL = (	1✔
50	f"{settings.SOCIALACCOUNT_PROVIDERS['fxa']['PROFILE_ENDPOINT']}/profile"
51	)
52
53
54	class FlagFilter(FilterSet):	1✔
55	class Meta:	1✔
56	model = get_waffle_flag_model()	1✔
57	fields = [	1✔
58	"name",
59	"everyone",
60	# "users",
61	# read-only
62	"id",
63	]
64
65
66	class FlagViewSet(ModelViewSet):	1✔
67	serializer_class = FlagSerializer	1✔
68	permission_classes = [IsAuthenticated, CanManageFlags]	1✔
69	filterset_class = FlagFilter	1✔
70	http_method_names = ["get", "post", "head", "patch"]	1✔
71
72	def get_queryset(self):	1✔
73	flags = get_waffle_flag_model().objects	1✔
74	return flags	1✔
75
76
77	class ProfileViewSet(ModelViewSet):	1✔
78	serializer_class = ProfileSerializer	1✔
79	permission_classes = [IsAuthenticated, IsOwner]	1✔
80	http_method_names = ["get", "post", "head", "put", "patch"]	1✔
81
82	def get_queryset(self) -> QuerySet[Profile]:	1✔
83	if isinstance(self.request.user, User):	1✔
84	return Profile.objects.filter(user=self.request.user)	1✔
85	return Profile.objects.none()	1✔
86
87
88	class UserViewSet(ModelViewSet):	1✔
89	serializer_class = UserSerializer	1✔
90	permission_classes = [IsAuthenticated, IsOwner]	1✔
91	http_method_names = ["get", "head"]	1✔
92
93	def get_queryset(self) -> QuerySet[User]:	1✔
94	if isinstance(self.request.user, User):	1!
UNCOV 95	return User.objects.filter(id=self.request.user.id)	×
96	return User.objects.none()	1✔
97
98
99	@permission_classes([IsAuthenticated])	1✔
100	@extend_schema(methods=["POST"], request=WebcompatIssueSerializer)	1✔
101	@api_view(["POST"])	1✔
102	def report_webcompat_issue(request):	1✔
UNCOV 103	serializer = WebcompatIssueSerializer(data=request.data)	×
UNCOV 104	if serializer.is_valid():	×
UNCOV 105	info_logger.info("webcompat_issue", extra=serializer.data)	×
UNCOV 106	incr_if_enabled("webcompat_issue", 1)	×
UNCOV 107	for k, v in serializer.data.items():	×
UNCOV 108	if v and k != "issue_on_domain":	×
UNCOV 109	incr_if_enabled(f"webcompat_issue_{k}", 1)	×
UNCOV 110	return Response(status=HTTP_201_CREATED)	×
111	return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)	×
112
113
114	@api_view()	1✔
115	@permission_classes([AllowAny])	1✔
116	def runtime_data(request):	1✔
117	flags = get_waffle_flag_model().get_all()	1✔
118	flag_values = [(f.name, f.is_active(request)) for f in flags]	1✔
119	switches = Switch.get_all()	1✔
120	switch_values = [(s.name, s.is_active()) for s in switches]	1✔
121	samples = Sample.get_all()	1✔
122	sample_values = [(s.name, s.is_active()) for s in samples]	1✔
123	return Response(	1✔
124	{
125	"FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
126	"PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
127	"GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
128	"BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
129	"PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
130	"PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
131	request, get_premium_country_language_mapping()
132	),
133	"PHONE_PLANS": get_countries_info_from_request_and_mapping(
134	request, get_phone_country_language_mapping()
135	),
136	"BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
137	request, get_bundle_country_language_mapping()
138	),
139	"BASKET_ORIGIN": settings.BASKET_ORIGIN,
140	"WAFFLE_FLAGS": flag_values,
141	"WAFFLE_SWITCHES": switch_values,
142	"WAFFLE_SAMPLES": sample_values,
143	"MAX_MINUTES_TO_VERIFY_REAL_PHONE": (
144	settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE
145	),
146	}
147	)
148
149
150	@extend_schema(	1✔
151	responses={
152	201: OpenApiResponse(description="Created; returned when user is created."),
153	202: OpenApiResponse(
154	description="Accepted; returned when user already exists."
155	),
156	400: OpenApiResponse(
157	description=(
158	"Bad request; returned when request is missing Authorization: Bearer"
159	" header or token value."
160	)
161	),
162	401: OpenApiResponse(
163	description=(
164	"Unauthorized; returned when the FXA token is invalid or expired."
165	)
166	),
167	},
168	)
169	@api_view(["POST"])	1✔
170	@permission_classes([AllowAny])	1✔
171	@authentication_classes([])	1✔
172	def terms_accepted_user(request):	1✔
173	"""
174	Create a Relay user from an FXA token.
175
176	See [API Auth doc][api-auth-doc] for details.
177
178	[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
179	""" # noqa: E501
180	# Setting authentication_classes to empty due to
181	# authentication still happening despite permissions being set to allowany
182	# https://forum.djangoproject.com/t/solved-allowany-override-does-not-work-on-apiview/9754
183	authorization = get_authorization_header(request).decode()	1✔
184	if not authorization or not authorization.startswith("Bearer "):	1✔
185	raise ParseError("Missing Bearer header.")	1✔
186
187	token = authorization.split(" ")[1]	1✔
188	if token == "":	1✔
189	raise ParseError("Missing FXA Token after 'Bearer'.")	1✔
190
191	try:	1✔
192	fxa_uid = get_fxa_uid_from_oauth_token(token, use_cache=False)	1✔
193	except AuthenticationFailed as e:	1✔
194	# AuthenticationFailed exception returns 403 instead of 401 because we are not
195	# using the proper config that comes with the authentication_classes. See:
196	# https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
197	if isinstance(e.detail, ErrorDetail):	1!
198	return Response(data={"detail": e.detail.title()}, status=e.status_code)	1✔
199	else:
UNCOV 200	return Response(data={"detail": e.get_full_details()}, status=e.status_code)	×
201	status_code = 201	1✔
202
203	try:	1✔
204	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
205	status_code = 202	1✔
206	except SocialAccount.DoesNotExist:	1✔
207	# User does not exist, create a new Relay user
208	fxa_profile_resp = requests.get(	1✔
209	FXA_PROFILE_URL, headers={"Authorization": f"Bearer {token}"}
210	)
211	if not (fxa_profile_resp.ok and fxa_profile_resp.content):	1✔
212	logger.error(	1✔
213	"terms_accepted_user: bad account profile response",
214	extra={
215	"status_code": fxa_profile_resp.status_code,
216	"content": fxa_profile_resp.content,
217	},
218	)
219	return Response(	1✔
220	data={"detail": "Did not receive a 200 response for account profile."},
221	status=500,
222	)
223
224	# This is not exactly the request object that FirefoxAccountsProvider expects,
225	# but it has all of the necessary attributes to initialize the Provider
226	provider = get_social_adapter().get_provider(request, "fxa")	1✔
227	# This may not save the new user that was created
228	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/base/provider.py#L44
229	social_login = provider.sociallogin_from_response(	1✔
230	request, fxa_profile_resp.json()
231	)
232	# Complete social login is called by callback, see
233	# https://github.com/pennersr/django-allauth/blob/77368a84903d32283f07a260819893ec15df78fb/allauth/socialaccount/providers/oauth/views.py#L118
234	# for what we are mimicking to create new SocialAccount, User, and Profile for
235	# the new Relay user from Firefox Since this is a Resource Provider/Server flow
236	# and are NOT a Relying Party (RP) of FXA No social token information is stored
237	# (no Social Token object created).
238	try:	1✔
239	complete_social_login(request, social_login)	1✔
240	# complete_social_login writes ['account_verified_email', 'user_created',
241	# '_auth_user_id', '_auth_user_backend', '_auth_user_hash'] on
242	# request.session which sets the cookie because complete_social_login does
243	# the "login" The user did not actually log in, logout to clear the session
244	if request.user.is_authenticated:	1!
245	get_account_adapter(request).logout(request)	1✔
246	except NoReverseMatch as e:	1✔
247	# TODO: use this logging to fix the underlying issue
248	# https://mozilla-hub.atlassian.net/browse/MPP-3473
249	if "socialaccount_signup" in e.args[0]:	1!
250	logger.error(	1✔
251	"socialaccount_signup_error",
252	extra={
253	"exception": str(e),
254	"fxa_uid": fxa_uid,
255	"social_login_state": social_login.state,
256	},
257	)
258	return Response(status=500)	1✔
UNCOV 259	raise e	×
260	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
261	# Indicate profile was created from the resource flow
262	profile = sa.user.profile	1✔
263	profile.created_by = "firefox_resource"	1✔
264	profile.save()	1✔
265	info_logger.info(	1✔
266	"terms_accepted_user",
267	extra={"social_account": sa.uid, "status_code": status_code},
268	)
269	return Response(status=status_code)	1✔

mozilla / fx-private-relay / 6337c030-fd5d-4165-ba69-027771511104

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