905e1d01-ee7f-42fa-9224-6c6d0cd712cd

Committed 24 Apr 2024 08:16PM CUT coverage: 75.618% (-0.1%) from 75.723%

Build # 905e1d01-ee7f-42fa-9224-6c6d0cd712cd

Build Type

Pull #4612

circleci

Committed by

rafeerahman

Commit Message

Added e2e-tests path for prettier linting, updated e2e readme

Pull Request Pull Request #4612: MPP-3779: E2E test fixes and additions

Run Details

2463 of 3426 branches covered (71.89%)

Branch coverage included in aggregate %.

6813 of 8841 relevant lines covered (77.06%)

19.99 hits per line

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

72.54

/privaterelay/views.py

import json
import logging
from collections.abc import Iterable
from datetime import UTC, datetime
from functools import cache
from hashlib import sha256
from typing import Any, TypedDict

from django.apps import apps
from django.conf import settings
from django.db import IntegrityError, transaction
from django.http import HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import redirect
from django.urls import reverse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods

import jwt
import sentry_sdk
from allauth.socialaccount.models import SocialAccount, SocialApp
from allauth.socialaccount.providers.fxa.views import FirefoxAccountsOAuth2Adapter
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
from google_measurement_protocol import event, report
from oauthlib.oauth2.rfc6749.errors import CustomOAuth2Error
from rest_framework.decorators import api_view, schema

# from silk.profiling.profiler import silk_profile
from emails.models import (
    CannotMakeSubdomainException,
    DomainAddress,
    RelayAddress,
    valid_available_subdomain,
)
from emails.utils import incr_if_enabled

from .apps import PrivateRelayConfig
from .fxa_utils import NoSocialToken, _get_oauth2_session

FXA_PROFILE_CHANGE_EVENT = "https://schemas.accounts.firefox.com/event/profile-change"
FXA_SUBSCRIPTION_CHANGE_EVENT = (
    "https://schemas.accounts.firefox.com/event/subscription-state-change"
)
FXA_DELETE_EVENT = "https://schemas.accounts.firefox.com/event/delete-user"
PROFILE_EVENTS = [FXA_PROFILE_CHANGE_EVENT, FXA_SUBSCRIPTION_CHANGE_EVENT]

logger = logging.getLogger("events")
info_logger = logging.getLogger("eventsinfo")


@cache
def _get_fxa(request):
    return request.user.socialaccount_set.filter(provider="fxa").first()


@api_view()
@schema(None)
@require_http_methods(["GET"])
def profile_refresh(request):
    if not request.user or request.user.is_anonymous:
        return redirect(reverse("fxa_login"))
    profile = request.user.profile

    fxa = _get_fxa(request)
    update_fxa(fxa)
    if "clicked-purchase" in request.COOKIES and profile.has_premium:
        event = "user_purchased_premium"
        incr_if_enabled(event, 1)

    return JsonResponse({})


@api_view(["POST", "GET"])
@schema(None)
@require_http_methods(["POST", "GET"])
def profile_subdomain(request):
    if not request.user or request.user.is_anonymous:
        return redirect(reverse("fxa_login"))
    profile = request.user.profile
    if not profile.has_premium:
        raise CannotMakeSubdomainException("error-premium-check-subdomain")
    try:
        if request.method == "GET":
            subdomain = request.GET.get("subdomain", None)
            available = valid_available_subdomain(subdomain)
            return JsonResponse({"available": available})
        else:
            subdomain = request.POST.get("subdomain", None)
            profile.add_subdomain(subdomain)
            return JsonResponse(
                {"status": "Accepted", "message": "success-subdomain-registered"},
                status=202,
            )
    except CannotMakeSubdomainException as e:
        return JsonResponse({"message": e.message, "subdomain": subdomain}, status=400)


@csrf_exempt
@require_http_methods(["POST"])
def metrics_event(request):
    try:
        request_data = json.loads(request.body)
    except json.JSONDecodeError:
        return JsonResponse({"msg": "Could not decode JSON"}, status=415)
    if "ga_uuid" not in request_data:
        return JsonResponse({"msg": "No GA uuid found"}, status=404)
    # "dimension5" is a Google Analytics-specific variable to track a custom dimension,
    # used to determine which browser vendor the add-on is using: Firefox or Chrome
    # "dimension7" is a Google Analytics-specific variable to track a custom dimension,
    # used to determine where the ping is coming from: website (default), add-on or app
    event_data = event(
        request_data.get("category", None),
        request_data.get("action", None),
        request_data.get("label", None),
        request_data.get("value", None),
        dimension5=request_data.get("dimension5", None),
        dimension7=request_data.get("dimension7", "website"),
    )
    try:
        report(settings.GOOGLE_ANALYTICS_ID, request_data.get("ga_uuid"), event_data)
    except Exception as e:
        logger.error("metrics_event", extra={"error": e})
        return JsonResponse({"msg": "Unable to report metrics event."}, status=500)
    return JsonResponse({"msg": "OK"}, status=200)


@csrf_exempt
def fxa_rp_events(request: HttpRequest) -> HttpResponse:
    req_jwt = _parse_jwt_from_request(request)
    authentic_jwt = _authenticate_fxa_jwt(req_jwt)
    event_keys = _get_event_keys_from_jwt(authentic_jwt)
    try:
        social_account = _get_account_from_jwt(authentic_jwt)
    except SocialAccount.DoesNotExist as e:
        # capture an exception in sentry, but don't error, or FXA will retry
        sentry_sdk.capture_exception(e)
        return HttpResponse("202 Accepted", status=202)

    for event_key in event_keys:
        if event_key in PROFILE_EVENTS:
            if settings.DEBUG:
                info_logger.info(
                    "fxa_profile_update",
                    extra={
                        "jwt": authentic_jwt,
                        "event_key": event_key,
                    },
                )
            update_fxa(social_account, authentic_jwt, event_key)
        if event_key == FXA_DELETE_EVENT:
            _handle_fxa_delete(authentic_jwt, social_account, event_key)
    return HttpResponse("200 OK", status=200)


def _parse_jwt_from_request(request: HttpRequest) -> str:
    request_auth = request.headers["Authorization"]
    return request_auth.split("Bearer ")[1]


def fxa_verifying_keys(reload: bool = False) -> list[dict[str, Any]]:
    """Get list of FxA verifying (public) keys."""
    private_relay_config = apps.get_app_config("privaterelay")
    assert isinstance(private_relay_config, PrivateRelayConfig)
    if reload:
        private_relay_config.ready()
    return private_relay_config.fxa_verifying_keys


class FxAEvent(TypedDict):
    """
    FxA Security Event Token (SET) payload, sent to relying parties.

    See:
    https://github.com/mozilla/fxa/tree/main/packages/fxa-event-broker
    https://www.rfc-editor.org/rfc/rfc8417 (Security Event Token)
    """

    iss: str  # Issuer, https://accounts.firefox.com/
    sub: str  # Subject, FxA user ID
    aud: str  # Audience, Relay's client ID
    iat: int  # Creation time, timestamp
    jti: str  # JWT ID, unique for this SET
    events: dict[str, dict[str, Any]]  # Event data


def _authenticate_fxa_jwt(req_jwt: str) -> FxAEvent:
    authentic_jwt = _verify_jwt_with_fxa_key(req_jwt, fxa_verifying_keys())

    if not authentic_jwt:
        # FXA key may be old? re-fetch FXA keys and try again
        authentic_jwt = _verify_jwt_with_fxa_key(
            req_jwt, fxa_verifying_keys(reload=True)
        )
        if not authentic_jwt:
            raise Exception("Could not authenticate JWT with FXA key.")

    return authentic_jwt


def _verify_jwt_with_fxa_key(
    req_jwt: str, verifying_keys: list[dict[str, Any]]
) -> FxAEvent | None:
    if not verifying_keys:
        raise Exception("FXA verifying keys are not available.")
    social_app = SocialApp.objects.get(provider="fxa")
    for verifying_key in verifying_keys:
        if verifying_key["alg"] == "RS256":
            public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(verifying_key))
            assert isinstance(public_key, RSAPublicKey)
            try:
                security_event = jwt.decode(
                    req_jwt,
                    public_key,
                    audience=social_app.client_id,
                    algorithms=["RS256"],
                    leeway=5,  # allow iat to be slightly in future, for clock skew
                )
            except jwt.ImmatureSignatureError:
                # Issue 2738: Log age of iat, if present
                claims = jwt.decode(
                    req_jwt,
                    public_key,
                    algorithms=["RS256"],
                    options={"verify_signature": False},
                )
                iat = claims.get("iat")
                iat_age = None
                if iat:
                    iat_age = round(datetime.now(tz=UTC).timestamp() - iat, 3)
                info_logger.warning(
                    "fxa_rp_event.future_iat", extra={"iat": iat, "iat_age_s": iat_age}
                )
                raise
            return FxAEvent(
                iss=security_event["iss"],
                sub=security_event["sub"],
                aud=security_event["aud"],
                iat=security_event["iat"],
                jti=security_event["jti"],
                events=security_event["events"],
            )
    return None


def _get_account_from_jwt(authentic_jwt: FxAEvent) -> SocialAccount:
    social_account_uid = authentic_jwt["sub"]
    return SocialAccount.objects.get(uid=social_account_uid, provider="fxa")


def _get_event_keys_from_jwt(authentic_jwt: FxAEvent) -> Iterable[str]:
    return authentic_jwt["events"].keys()


def update_fxa(
    social_account: SocialAccount,
    authentic_jwt: FxAEvent | None = None,
    event_key: str | None = None,
) -> HttpResponse:
    try:
        client = _get_oauth2_session(social_account)
    except NoSocialToken as e:
        sentry_sdk.capture_exception(e)
        return HttpResponse("202 Accepted", status=202)

    # TODO: more graceful handling of profile fetch failures
    try:
        resp = client.get(FirefoxAccountsOAuth2Adapter.profile_url)
    except CustomOAuth2Error as e:
        sentry_sdk.capture_exception(e)
        return HttpResponse("202 Accepted", status=202)

    extra_data = resp.json()

    try:
        new_email = extra_data["email"]
    except KeyError as e:
        sentry_sdk.capture_exception(e)
        return HttpResponse("202 Accepted", status=202)

    if authentic_jwt and event_key:
        info_logger.info(
            "fxa_rp_event",
            extra={
                "fxa_uid": authentic_jwt["sub"],
                "event_key": event_key,
                "real_address": sha256(new_email.encode("utf-8")).hexdigest(),
            },
        )

    return _update_all_data(social_account, extra_data, new_email)


def _update_all_data(
    social_account: SocialAccount, extra_data: dict[str, Any], new_email: str
) -> HttpResponse:
    try:
        profile = social_account.user.profile
        had_premium = profile.has_premium
        had_phone = profile.has_phone
        with transaction.atomic():
            social_account.extra_data = extra_data
            social_account.save()
            profile = social_account.user.profile
            now_has_premium = profile.has_premium
            newly_premium = not had_premium and now_has_premium
            no_longer_premium = had_premium and not now_has_premium
            if newly_premium:
                incr_if_enabled("user_purchased_premium", 1)
                profile.date_subscribed = datetime.now(UTC)
                profile.save()
            if no_longer_premium:
                incr_if_enabled("user_has_downgraded", 1)
            now_has_phone = profile.has_phone
            newly_phone = not had_phone and now_has_phone
            no_longer_phone = had_phone and not now_has_phone
            if newly_phone:
                incr_if_enabled("user_purchased_phone", 1)
                profile.date_subscribed_phone = datetime.now(UTC)
                profile.date_phone_subscription_reset = datetime.now(UTC)
                profile.save()
            if no_longer_phone:
                incr_if_enabled("user_has_dropped_phone", 1)
            social_account.user.email = new_email
            social_account.user.save()
            email_address_record = social_account.user.emailaddress_set.first()
            if email_address_record:
                email_address_record.email = new_email
                email_address_record.save()
            else:
                social_account.user.emailaddress_set.create(email=new_email)
            return HttpResponse("202 Accepted", status=202)
    except IntegrityError as e:
        sentry_sdk.capture_exception(e)
        return HttpResponse("Conflict", status=409)


def _handle_fxa_delete(
    authentic_jwt: FxAEvent, social_account: SocialAccount, event_key: str
) -> None:
    # Using for loops here because QuerySet.delete() does a bulk delete which does
    # not call the model delete() methods that create DeletedAddress records
    for relay_address in RelayAddress.objects.filter(user=social_account.user):
        relay_address.delete()
    for domain_address in DomainAddress.objects.filter(user=social_account.user):
        domain_address.delete()

    social_account.user.delete()
    info_logger.info(
        "fxa_rp_event",
        extra={
            "fxa_uid": authentic_jwt["sub"],
            "event_key": event_key,
        },
    )

1	import json	1✔
2	import logging	1✔
3	from collections.abc import Iterable	1✔
4	from datetime import UTC, datetime	1✔
5	from functools import cache	1✔
6	from hashlib import sha256	1✔
7	from typing import Any, TypedDict	1✔
8
9	from django.apps import apps	1✔
10	from django.conf import settings	1✔
11	from django.db import IntegrityError, transaction	1✔
12	from django.http import HttpRequest, HttpResponse, JsonResponse	1✔
13	from django.shortcuts import redirect	1✔
14	from django.urls import reverse	1✔
15	from django.views.decorators.csrf import csrf_exempt	1✔
16	from django.views.decorators.http import require_http_methods	1✔
17
18	import jwt	1✔
19	import sentry_sdk	1✔
20	from allauth.socialaccount.models import SocialAccount, SocialApp	1✔
21	from allauth.socialaccount.providers.fxa.views import FirefoxAccountsOAuth2Adapter	1✔
22	from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey	1✔
23	from google_measurement_protocol import event, report	1✔
24	from oauthlib.oauth2.rfc6749.errors import CustomOAuth2Error	1✔
25	from rest_framework.decorators import api_view, schema	1✔
26
27	# from silk.profiling.profiler import silk_profile
28	from emails.models import (	1✔
29	CannotMakeSubdomainException,
30	DomainAddress,
31	RelayAddress,
32	valid_available_subdomain,
33	)
34	from emails.utils import incr_if_enabled	1✔
35
36	from .apps import PrivateRelayConfig	1✔
37	from .fxa_utils import NoSocialToken, _get_oauth2_session	1✔
38
39	FXA_PROFILE_CHANGE_EVENT = "https://schemas.accounts.firefox.com/event/profile-change"	1✔
40	FXA_SUBSCRIPTION_CHANGE_EVENT = (	1✔
41	"https://schemas.accounts.firefox.com/event/subscription-state-change"
42	)
43	FXA_DELETE_EVENT = "https://schemas.accounts.firefox.com/event/delete-user"	1✔
44	PROFILE_EVENTS = [FXA_PROFILE_CHANGE_EVENT, FXA_SUBSCRIPTION_CHANGE_EVENT]	1✔
45
46	logger = logging.getLogger("events")	1✔
47	info_logger = logging.getLogger("eventsinfo")	1✔
48
49
50	@cache	1✔
51	def _get_fxa(request):	1✔
52	return request.user.socialaccount_set.filter(provider="fxa").first()	×
53
54
55	@api_view()	1✔
56	@schema(None)	1✔
57	@require_http_methods(["GET"])	1✔
58	def profile_refresh(request):	1✔
59	if not request.user or request.user.is_anonymous:	×
60	return redirect(reverse("fxa_login"))	×
61	profile = request.user.profile	×
62
63	fxa = _get_fxa(request)	×
64	update_fxa(fxa)	×
65	if "clicked-purchase" in request.COOKIES and profile.has_premium:	×
66	event = "user_purchased_premium"	×
67	incr_if_enabled(event, 1)	×
68
69	return JsonResponse({})	×
70
71
72	@api_view(["POST", "GET"])	1✔
73	@schema(None)	1✔
74	@require_http_methods(["POST", "GET"])	1✔
75	def profile_subdomain(request):	1✔
76	if not request.user or request.user.is_anonymous:	×
77	return redirect(reverse("fxa_login"))	×
78	profile = request.user.profile	×
79	if not profile.has_premium:	×
80	raise CannotMakeSubdomainException("error-premium-check-subdomain")	×
81	try:	×
82	if request.method == "GET":	×
83	subdomain = request.GET.get("subdomain", None)	×
84	available = valid_available_subdomain(subdomain)	×
85	return JsonResponse({"available": available})	×
86	else:
87	subdomain = request.POST.get("subdomain", None)	×
88	profile.add_subdomain(subdomain)	×
89	return JsonResponse(	×
90	{"status": "Accepted", "message": "success-subdomain-registered"},
91	status=202,
92	)
93	except CannotMakeSubdomainException as e:	×
94	return JsonResponse({"message": e.message, "subdomain": subdomain}, status=400)	×
95
96
97	@csrf_exempt	1✔
98	@require_http_methods(["POST"])	1✔
99	def metrics_event(request):	1✔
100	try:	×
101	request_data = json.loads(request.body)	×
102	except json.JSONDecodeError:	×
103	return JsonResponse({"msg": "Could not decode JSON"}, status=415)	×
104	if "ga_uuid" not in request_data:	×
105	return JsonResponse({"msg": "No GA uuid found"}, status=404)	×
106	# "dimension5" is a Google Analytics-specific variable to track a custom dimension,
107	# used to determine which browser vendor the add-on is using: Firefox or Chrome
108	# "dimension7" is a Google Analytics-specific variable to track a custom dimension,
109	# used to determine where the ping is coming from: website (default), add-on or app
110	event_data = event(	×
111	request_data.get("category", None),
112	request_data.get("action", None),
113	request_data.get("label", None),
114	request_data.get("value", None),
115	dimension5=request_data.get("dimension5", None),
116	dimension7=request_data.get("dimension7", "website"),
117	)
118	try:	×
119	report(settings.GOOGLE_ANALYTICS_ID, request_data.get("ga_uuid"), event_data)	×
120	except Exception as e:	×
121	logger.error("metrics_event", extra={"error": e})	×
122	return JsonResponse({"msg": "Unable to report metrics event."}, status=500)	×
123	return JsonResponse({"msg": "OK"}, status=200)	×
124
125
126	@csrf_exempt	1✔
127	def fxa_rp_events(request: HttpRequest) -> HttpResponse:	1✔
128	req_jwt = _parse_jwt_from_request(request)	1✔
129	authentic_jwt = _authenticate_fxa_jwt(req_jwt)	1✔
130	event_keys = _get_event_keys_from_jwt(authentic_jwt)	1✔
131	try:	1✔
132	social_account = _get_account_from_jwt(authentic_jwt)	1✔
133	except SocialAccount.DoesNotExist as e:	×
134	# capture an exception in sentry, but don't error, or FXA will retry
135	sentry_sdk.capture_exception(e)	×
136	return HttpResponse("202 Accepted", status=202)	×
137
138	for event_key in event_keys:	1✔
139	if event_key in PROFILE_EVENTS:	1✔
140	if settings.DEBUG:	1!
141	info_logger.info(	×
142	"fxa_profile_update",
143	extra={
144	"jwt": authentic_jwt,
145	"event_key": event_key,
146	},
147	)
148	update_fxa(social_account, authentic_jwt, event_key)	1✔
149	if event_key == FXA_DELETE_EVENT:	1✔
150	_handle_fxa_delete(authentic_jwt, social_account, event_key)	1✔
151	return HttpResponse("200 OK", status=200)	1✔
152
153
154	def _parse_jwt_from_request(request: HttpRequest) -> str:	1✔
155	request_auth = request.headers["Authorization"]	1✔
156	return request_auth.split("Bearer ")[1]	1✔
157
158
159	def fxa_verifying_keys(reload: bool = False) -> list[dict[str, Any]]:	1✔
160	"""Get list of FxA verifying (public) keys."""
161	private_relay_config = apps.get_app_config("privaterelay")	1✔
162	assert isinstance(private_relay_config, PrivateRelayConfig)	1✔
163	if reload:	1✔
164	private_relay_config.ready()	1✔
165	return private_relay_config.fxa_verifying_keys	1✔
166
167
168	class FxAEvent(TypedDict):	1✔
169	"""
170	FxA Security Event Token (SET) payload, sent to relying parties.
171
172	See:
173	https://github.com/mozilla/fxa/tree/main/packages/fxa-event-broker
174	https://www.rfc-editor.org/rfc/rfc8417 (Security Event Token)
175	"""
176
177	iss: str # Issuer, https://accounts.firefox.com/	1✔
178	sub: str # Subject, FxA user ID	1✔
179	aud: str # Audience, Relay's client ID	1✔
180	iat: int # Creation time, timestamp	1✔
181	jti: str # JWT ID, unique for this SET	1✔
182	events: dict[str, dict[str, Any]] # Event data	1✔
183
184
185	def _authenticate_fxa_jwt(req_jwt: str) -> FxAEvent:	1✔
186	authentic_jwt = _verify_jwt_with_fxa_key(req_jwt, fxa_verifying_keys())	1✔
187
188	if not authentic_jwt:	1!
189	# FXA key may be old? re-fetch FXA keys and try again
190	authentic_jwt = _verify_jwt_with_fxa_key(	×
191	req_jwt, fxa_verifying_keys(reload=True)
192	)
193	if not authentic_jwt:	×
194	raise Exception("Could not authenticate JWT with FXA key.")	×
195
196	return authentic_jwt	1✔
197
198
199	def _verify_jwt_with_fxa_key(	1✔
200	req_jwt: str, verifying_keys: list[dict[str, Any]]
201	) -> FxAEvent \| None:
202	if not verifying_keys:	1!
203	raise Exception("FXA verifying keys are not available.")	×
204	social_app = SocialApp.objects.get(provider="fxa")	1✔
205	for verifying_key in verifying_keys:	1!
206	if verifying_key["alg"] == "RS256":	1!
207	public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(verifying_key))	1✔
208	assert isinstance(public_key, RSAPublicKey)	1✔
209	try:	1✔
210	security_event = jwt.decode(	1✔
211	req_jwt,
212	public_key,
213	audience=social_app.client_id,
214	algorithms=["RS256"],
215	leeway=5, # allow iat to be slightly in future, for clock skew
216	)
217	except jwt.ImmatureSignatureError:	1✔
218	# Issue 2738: Log age of iat, if present
219	claims = jwt.decode(	1✔
220	req_jwt,
221	public_key,
222	algorithms=["RS256"],
223	options={"verify_signature": False},
224	)
225	iat = claims.get("iat")	1✔
226	iat_age = None	1✔
227	if iat:	1!
228	iat_age = round(datetime.now(tz=UTC).timestamp() - iat, 3)	1✔
229	info_logger.warning(	1✔
230	"fxa_rp_event.future_iat", extra={"iat": iat, "iat_age_s": iat_age}
231	)
232	raise	1✔
233	return FxAEvent(	1✔
234	iss=security_event["iss"],
235	sub=security_event["sub"],
236	aud=security_event["aud"],
237	iat=security_event["iat"],
238	jti=security_event["jti"],
239	events=security_event["events"],
240	)
241	return None	×
242
243
244	def _get_account_from_jwt(authentic_jwt: FxAEvent) -> SocialAccount:	1✔
245	social_account_uid = authentic_jwt["sub"]	1✔
246	return SocialAccount.objects.get(uid=social_account_uid, provider="fxa")	1✔
247
248
249	def _get_event_keys_from_jwt(authentic_jwt: FxAEvent) -> Iterable[str]:	1✔
250	return authentic_jwt["events"].keys()	1✔
251
252
253	def update_fxa(	1✔
254	social_account: SocialAccount,
255	authentic_jwt: FxAEvent \| None = None,
256	event_key: str \| None = None,
257	) -> HttpResponse:
258	try:	1✔
259	client = _get_oauth2_session(social_account)	1✔
260	except NoSocialToken as e:	×
261	sentry_sdk.capture_exception(e)	×
262	return HttpResponse("202 Accepted", status=202)	×
263
264	# TODO: more graceful handling of profile fetch failures
265	try:	1✔
266	resp = client.get(FirefoxAccountsOAuth2Adapter.profile_url)	1✔
267	except CustomOAuth2Error as e:	×
268	sentry_sdk.capture_exception(e)	×
269	return HttpResponse("202 Accepted", status=202)	×
270
271	extra_data = resp.json()	1✔
272
273	try:	1✔
274	new_email = extra_data["email"]	1✔
275	except KeyError as e:	×
276	sentry_sdk.capture_exception(e)	×
277	return HttpResponse("202 Accepted", status=202)	×
278
279	if authentic_jwt and event_key:	1!
280	info_logger.info(	1✔
281	"fxa_rp_event",
282	extra={
283	"fxa_uid": authentic_jwt["sub"],
284	"event_key": event_key,
285	"real_address": sha256(new_email.encode("utf-8")).hexdigest(),
286	},
287	)
288
289	return _update_all_data(social_account, extra_data, new_email)	1✔
290
291
292	def _update_all_data(	1✔
293	social_account: SocialAccount, extra_data: dict[str, Any], new_email: str
294	) -> HttpResponse:
295	try:	1✔
296	profile = social_account.user.profile	1✔
297	had_premium = profile.has_premium	1✔
298	had_phone = profile.has_phone	1✔
299	with transaction.atomic():	1✔
300	social_account.extra_data = extra_data	1✔
301	social_account.save()	1✔
302	profile = social_account.user.profile	1✔
303	now_has_premium = profile.has_premium	1✔
304	newly_premium = not had_premium and now_has_premium	1✔
305	no_longer_premium = had_premium and not now_has_premium	1✔
306	if newly_premium:	1✔
307	incr_if_enabled("user_purchased_premium", 1)	1✔
308	profile.date_subscribed = datetime.now(UTC)	1✔
309	profile.save()	1✔
310	if no_longer_premium:	1!
311	incr_if_enabled("user_has_downgraded", 1)	×
312	now_has_phone = profile.has_phone	1✔
313	newly_phone = not had_phone and now_has_phone	1✔
314	no_longer_phone = had_phone and not now_has_phone	1✔
315	if newly_phone:	1✔
316	incr_if_enabled("user_purchased_phone", 1)	1✔
317	profile.date_subscribed_phone = datetime.now(UTC)	1✔
318	profile.date_phone_subscription_reset = datetime.now(UTC)	1✔
319	profile.save()	1✔
320	if no_longer_phone:	1!
321	incr_if_enabled("user_has_dropped_phone", 1)	×
322	social_account.user.email = new_email	1✔
323	social_account.user.save()	1✔
324	email_address_record = social_account.user.emailaddress_set.first()	1✔
325	if email_address_record:	1✔
326	email_address_record.email = new_email	1✔
327	email_address_record.save()	1✔
328	else:
329	social_account.user.emailaddress_set.create(email=new_email)	1✔
330	return HttpResponse("202 Accepted", status=202)	1✔
331	except IntegrityError as e:	1✔
332	sentry_sdk.capture_exception(e)	1✔
333	return HttpResponse("Conflict", status=409)	1✔
334
335
336	def _handle_fxa_delete(	1✔
337	authentic_jwt: FxAEvent, social_account: SocialAccount, event_key: str
338	) -> None:
339	# Using for loops here because QuerySet.delete() does a bulk delete which does
340	# not call the model delete() methods that create DeletedAddress records
341	for relay_address in RelayAddress.objects.filter(user=social_account.user):	1✔
342	relay_address.delete()	1✔
343	for domain_address in DomainAddress.objects.filter(user=social_account.user):	1✔
344	domain_address.delete()	1✔
345
346	social_account.user.delete()	1✔
347	info_logger.info(	1✔
348	"fxa_rp_event",
349	extra={
350	"fxa_uid": authentic_jwt["sub"],
351	"event_key": event_key,
352	},
353	)

mozilla / fx-private-relay / 905e1d01-ee7f-42fa-9224-6c6d0cd712cd

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