b821b0f9-15b7-4e53-b172-d7fffcb2f6b6

Committed 15 Oct 2024 02:41PM UTC coverage: 84.424% (+0.004%) from 84.42%

Build # b821b0f9-15b7-4e53-b172-d7fffcb2f6b6

Build Type

Pull #5108

circleci

Committed by

jwhitlock

Commit Message

Update metrics documentation

Pull Request Pull Request #5108: MPP-3831: Send metrics_event data to influxdb, logs instead of UA

Run Details

2357 of 3497 branches covered (67.4%)

Branch coverage included in aggregate %.

32 of 32 new or added lines in 2 files covered. (100.0%)

19 existing lines in 1 file now uncovered.

16331 of 18639 relevant lines covered (87.62%)

10.23 hits per line

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

73.45

/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 markus.utils import generate_tag
from oauthlib.oauth2.rfc6749.errors import CustomOAuth2Error
from rest_framework.decorators import api_view, schema

from emails.models import DomainAddress, RelayAddress
from emails.utils import incr_if_enabled

from .apps import PrivateRelayConfig
from .exceptions import CannotMakeSubdomainException
from .fxa_utils import NoSocialToken, _get_oauth2_session
from .validators import valid_available_subdomain

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)


@csrf_exempt
@require_http_methods(["POST"])
def metrics_event(request: HttpRequest) -> JsonResponse:
    """
    Handle metrics events from the Relay extension.

    This used to forward data to Google Analytics, but was not updated for GA4.

    Now it logs the information and updates statsd counters.
    """
    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)

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

mozilla / fx-private-relay / b821b0f9-15b7-4e53-b172-d7fffcb2f6b6

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