98c45938-8d23-4015-90f8-7dff916ce789

Committed 21 Jun 2024 01:50PM CUT coverage: 85.357% (+0.02%) from 85.337%

Build # 98c45938-8d23-4015-90f8-7dff916ce789

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #4799 from mozilla/split-exceptions-validators-mpp-3827

MPP-3827: Move non-model code out of `emails/models.py`

Run Details

4004 of 5135 branches covered (77.97%)

Branch coverage included in aggregate %.

335 of 339 new or added lines in 9 files covered. (98.82%)

1 existing line in 1 file now uncovered.

15746 of 18003 relevant lines covered (87.46%)

10.31 hits per line

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

75.62

/privaterelay/views.py

import json
import logging
import threading
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.exceptions import CannotMakeSubdomainException
from emails.models import DomainAddress, RelayAddress
from emails.utils import incr_if_enabled
from emails.validators import valid_available_subdomain

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)
            valid_available_subdomain(subdomain)
            return JsonResponse({"available": True})
        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)


def send_ga_ping(ga_id: str, ga_uuid: str, data: Any) -> None:
    try:
        report(ga_id, ga_uuid, data)
    except Exception as e:
        logger.error("metrics_event", extra={"error": e})


@csrf_exempt
@require_http_methods(["POST"])
def metrics_event(request: HttpRequest) -> JsonResponse:
    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"),
    )
    t = threading.Thread(
        target=send_ga_ping,
        args=[settings.GOOGLE_ANALYTICS_ID, request_data.get("ga_uuid"), event_data],
        daemon=True,
    )
    t.start()
    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:
        # Don't error, or FXA will retry
        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")
    if not isinstance(private_relay_config, PrivateRelayConfig):
        raise TypeError("private_relay_config must be PrivateRelayConfig")
    if reload:
        private_relay_config.ready()
    return private_relay_config.fxa_verifying_keys


def fxa_social_app(reload: bool = False) -> SocialApp:
    """Get FxA SocialApp from app config or DB."""
    private_relay_config = apps.get_app_config("privaterelay")
    if not isinstance(private_relay_config, PrivateRelayConfig):
        raise TypeError("private_relay_config must be PrivateRelayConfig")
    if reload:
        private_relay_config.ready()
    return private_relay_config.fxa_social_app


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

mozilla / fx-private-relay / 98c45938-8d23-4015-90f8-7dff916ce789

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