0942fe47-f79e-43a2-ae58-e348576c1bd5

Committed 17 Jul 2024 03:07PM UTC coverage: 85.449% (+0.01%) from 85.436%

Build # 0942fe47-f79e-43a2-ae58-e348576c1bd5

Build Type

Pull #4763

circleci

Committed by

jwhitlock

Commit Message

Update imports of Profile

Pull Request Pull Request #4763: MPP-3827: Move Profile and RegisteredSubdomain models to privaterelay

Run Details

4088 of 5234 branches covered (78.1%)

Branch coverage included in aggregate %.

1000 of 1026 new or added lines in 26 files covered. (97.47%)

38 existing lines in 2 files now uncovered.

15955 of 18222 relevant lines covered (87.56%)

10.44 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 emails.models import DomainAddress, RelayAddress
from emails.utils import incr_if_enabled

from .apps import PrivateRelayConfig

# from silk.profiling.profiler import silk_profile
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)


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

mozilla / fx-private-relay / 0942fe47-f79e-43a2-ae58-e348576c1bd5

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