5e9994ac-40d1-44e7-a22f-8565e529b07d

Committed 22 Apr 2024 08:02PM CUT coverage: 75.618% (+0.007%) from 75.611%

Build # 5e9994ac-40d1-44e7-a22f-8565e529b07d

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #4607 from mozilla/add-ruff-again-mpp-79

MPP-79: Add `ruff` Python linter

Run Details

2463 of 3426 branches covered (71.89%)

Branch coverage included in aggregate %.

184 of 194 new or added lines in 36 files covered. (94.85%)

1 existing line in 1 file now uncovered.

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

69.03

/privaterelay/fxa_utils.py

import logging
from datetime import UTC, datetime, timedelta
from typing import Any, cast

from django.conf import settings

import sentry_sdk
from allauth.socialaccount.models import SocialAccount, SocialToken
from allauth.socialaccount.providers.fxa.views import FirefoxAccountsOAuth2Adapter
from oauthlib.oauth2.rfc6749.errors import CustomOAuth2Error, TokenExpiredError
from requests_oauthlib import OAuth2Session

from .utils import flag_is_active_in_task

logger = logging.getLogger("events")


class NoSocialToken(Exception):
    """The SocialAccount has no SocialToken"""

    def __init__(self, uid: str, *args, **kwargs):
        self.uid = uid
        super().__init__(*args, **kwargs)

    def __str__(self) -> str:
        return f'NoSocialToken: The SocialAccount "{self.uid}" has no token.'

    def __repr__(self) -> str:
        return f'{self.__class__.__name__}("{self.uid}")'


def update_social_token(
    existing_social_token: SocialToken, new_oauth2_token: dict[str, Any]
) -> None:
    existing_social_token.token = new_oauth2_token["access_token"]
    existing_social_token.token_secret = new_oauth2_token["refresh_token"]
    existing_social_token.expires_at = datetime.now(UTC) + timedelta(
        seconds=int(new_oauth2_token["expires_in"])
    )
    existing_social_token.save()


# use "raw" requests_oauthlib to automatically refresh the access token
# https://github.com/pennersr/django-allauth/issues/420#issuecomment-301805706
def _get_oauth2_session(social_account: SocialAccount) -> OAuth2Session:
    refresh_token_url = FirefoxAccountsOAuth2Adapter.access_token_url
    social_token = social_account.socialtoken_set.first()
    if social_token is None:
        raise NoSocialToken(uid=social_account.uid)

    def _token_updater(new_token):
        update_social_token(social_token, new_token)

    client_id = social_token.app.client_id
    client_secret = social_token.app.secret

    extra = {
        "client_id": client_id,
        "client_secret": client_secret,
    }

    expires_in = (social_token.expires_at - datetime.now(UTC)).total_seconds()
    token = {
        "access_token": social_token.token,
        "refresh_token": social_token.token_secret,
        "token_type": "Bearer",
        "expires_in": expires_in,
    }

    # TODO: find out why the auto_refresh and token_updater is not working
    # and instead we are manually refreshing the token at get_subscription_data_from_fxa
    client = OAuth2Session(
        client_id,
        scope=settings.SOCIALACCOUNT_PROVIDERS["fxa"]["SCOPE"],
        token=token,
        auto_refresh_url=refresh_token_url,
        auto_refresh_kwargs=extra,
        token_updater=_token_updater,
    )
    return client


def _refresh_token(client, social_account):
    social_token = SocialToken.objects.get(account=social_account)
    # refresh user token to expand the scope to get accounts subscription data
    new_token = client.refresh_token(FirefoxAccountsOAuth2Adapter.access_token_url)
    update_social_token(social_token, new_token)
    return {"social_token": new_token, "refreshed": True}


def get_subscription_data_from_fxa(social_account: SocialAccount) -> dict[str, Any]:
    accounts_subscription_url = (
        settings.FXA_ACCOUNTS_ENDPOINT
        + "/oauth/mozilla-subscriptions/customer/billing-and-subscriptions"
    )

    try:
        client = _get_oauth2_session(social_account)
    except NoSocialToken as e:
        sentry_sdk.capture_exception(e)
        return {}

    try:
        # get detailed subscription data from FxA
        resp = client.get(accounts_subscription_url)
        json_resp = cast(dict[str, Any], resp.json())

        if "Requested scopes are not allowed" in json_resp.get("message", ""):
            logger.error("accounts_subscription_scope_failed")
            json_resp = _refresh_token(client, social_account)
    except TokenExpiredError as e:
        sentry_sdk.capture_exception(e)
        json_resp = _refresh_token(client, social_account)
    except CustomOAuth2Error as e:
        sentry_sdk.capture_exception(e)
        json_resp = {}
    return json_resp


def get_phone_subscription_dates(social_account):
    subscription_data = get_subscription_data_from_fxa(social_account)
    if "refreshed" in subscription_data.keys():
        # user token refreshed for expanded scope
        social_account.refresh_from_db()
        # retry getting detailed subscription data
        subscription_data = get_subscription_data_from_fxa(social_account)
        if "refreshed" in subscription_data.keys():
            return None, None, None
    if "subscriptions" not in subscription_data.keys():
        # failed to get subscriptions data which may mean user never had subscription
        # and/or there is data mismatch with FxA
        if not flag_is_active_in_task("free_phones", social_account.user):
            # User who was flagged for having phone subscriptions
            # did not actually have phone subscriptions
            logger.error(
                "accounts_subscription_endpoint_failed",
                extra={"fxa_message": subscription_data.get("message", "")},
            )
        return None, None, None

    date_subscribed_phone = start_date = end_date = None
    product_w_phone_capabilites = [settings.PHONE_PROD_ID, settings.BUNDLE_PROD_ID]
    for sub in subscription_data.get("subscriptions", []):
        # Even if a user upgrade subscription e.g. from monthly to yearly
        # or from phone to VPN bundle use the last subscription subscription dates
        # Later, when the subscription details only show one valid subsription
        # this information can be updated
        subscription_created_timestamp = None
        subscription_start_timestamp = None
        subscription_end_timestamp = None
        if sub.get("product_id") in product_w_phone_capabilites:
            subscription_created_timestamp = sub.get("created")
            subscription_start_timestamp = sub.get("current_period_start")
            subscription_end_timestamp = sub.get("current_period_end")
        else:
            # not a product id for phone subscription, continue
            continue

        subscription_date_none = (
            subscription_created_timestamp
            and subscription_start_timestamp
            and subscription_end_timestamp
        ) is None
        if subscription_date_none:
            # subscription dates are required fields according to FxA documentation:
            # https://mozilla.github.io/ecosystem-platform/api#tag/Subscriptions/operation/getOauthMozillasubscriptionsCustomerBillingandsubscriptions
            logger.error(
                "accounts_subscription_subscription_date_invalid",
                extra={"subscription": sub},
            )
            return None, None, None

        date_subscribed_phone = datetime.fromtimestamp(
            subscription_created_timestamp, tz=UTC
        )
        start_date = datetime.fromtimestamp(subscription_start_timestamp, tz=UTC)
        end_date = datetime.fromtimestamp(subscription_end_timestamp, tz=UTC)
    return date_subscribed_phone, start_date, end_date

1	import logging	1✔
2	from datetime import UTC, datetime, timedelta	1✔
3	from typing import Any, cast	1✔
4
5	from django.conf import settings	1✔
6
7	import sentry_sdk	1✔
8	from allauth.socialaccount.models import SocialAccount, SocialToken	1✔
9	from allauth.socialaccount.providers.fxa.views import FirefoxAccountsOAuth2Adapter	1✔
10	from oauthlib.oauth2.rfc6749.errors import CustomOAuth2Error, TokenExpiredError	1✔
11	from requests_oauthlib import OAuth2Session	1✔
12
13	from .utils import flag_is_active_in_task	1✔
14
15	logger = logging.getLogger("events")	1✔
16
17
18	class NoSocialToken(Exception):	1✔
19	"""The SocialAccount has no SocialToken"""
20
21	def __init__(self, uid: str, args, *kwargs):	1✔
22	self.uid = uid	1✔
23	super().__init__(args, *kwargs)	1✔
24
25	def __str__(self) -> str:	1✔
26	return f'NoSocialToken: The SocialAccount "{self.uid}" has no token.'	1✔
27
28	def __repr__(self) -> str:	1✔
29	return f'{self.__class__.__name__}("{self.uid}")'	1✔
30
31
32	def update_social_token(	1✔
33	existing_social_token: SocialToken, new_oauth2_token: dict[str, Any]
34	) -> None:
35	existing_social_token.token = new_oauth2_token["access_token"]	×
36	existing_social_token.token_secret = new_oauth2_token["refresh_token"]	×
NEW 37	existing_social_token.expires_at = datetime.now(UTC) + timedelta(	×
38	seconds=int(new_oauth2_token["expires_in"])
39	)
40	existing_social_token.save()	×
41
42
43	# use "raw" requests_oauthlib to automatically refresh the access token
44	# https://github.com/pennersr/django-allauth/issues/420#issuecomment-301805706
45	def _get_oauth2_session(social_account: SocialAccount) -> OAuth2Session:	1✔
46	refresh_token_url = FirefoxAccountsOAuth2Adapter.access_token_url	1✔
47	social_token = social_account.socialtoken_set.first()	1✔
48	if social_token is None:	1!
49	raise NoSocialToken(uid=social_account.uid)	×
50
51	def _token_updater(new_token):	1✔
52	update_social_token(social_token, new_token)	×
53
54	client_id = social_token.app.client_id	1✔
55	client_secret = social_token.app.secret	1✔
56
57	extra = {	1✔
58	"client_id": client_id,
59	"client_secret": client_secret,
60	}
61
62	expires_in = (social_token.expires_at - datetime.now(UTC)).total_seconds()	1✔
63	token = {	1✔
64	"access_token": social_token.token,
65	"refresh_token": social_token.token_secret,
66	"token_type": "Bearer",
67	"expires_in": expires_in,
68	}
69
70	# TODO: find out why the auto_refresh and token_updater is not working
71	# and instead we are manually refreshing the token at get_subscription_data_from_fxa
72	client = OAuth2Session(	1✔
73	client_id,
74	scope=settings.SOCIALACCOUNT_PROVIDERS["fxa"]["SCOPE"],
75	token=token,
76	auto_refresh_url=refresh_token_url,
77	auto_refresh_kwargs=extra,
78	token_updater=_token_updater,
79	)
80	return client	1✔
81
82
83	def _refresh_token(client, social_account):	1✔
84	social_token = SocialToken.objects.get(account=social_account)	×
85	# refresh user token to expand the scope to get accounts subscription data
86	new_token = client.refresh_token(FirefoxAccountsOAuth2Adapter.access_token_url)	×
87	update_social_token(social_token, new_token)	×
88	return {"social_token": new_token, "refreshed": True}	×
89
90
91	def get_subscription_data_from_fxa(social_account: SocialAccount) -> dict[str, Any]:	1✔
92	accounts_subscription_url = (	×
93	settings.FXA_ACCOUNTS_ENDPOINT
94	+ "/oauth/mozilla-subscriptions/customer/billing-and-subscriptions"
95	)
96
97	try:	×
98	client = _get_oauth2_session(social_account)	×
99	except NoSocialToken as e:	×
100	sentry_sdk.capture_exception(e)	×
101	return {}	×
102
103	try:	×
104	# get detailed subscription data from FxA
105	resp = client.get(accounts_subscription_url)	×
106	json_resp = cast(dict[str, Any], resp.json())	×
107
108	if "Requested scopes are not allowed" in json_resp.get("message", ""):	×
109	logger.error("accounts_subscription_scope_failed")	×
110	json_resp = _refresh_token(client, social_account)	×
111	except TokenExpiredError as e:	×
112	sentry_sdk.capture_exception(e)	×
113	json_resp = _refresh_token(client, social_account)	×
114	except CustomOAuth2Error as e:	×
115	sentry_sdk.capture_exception(e)	×
116	json_resp = {}	×
117	return json_resp	×
118
119
120	def get_phone_subscription_dates(social_account):	1✔
121	subscription_data = get_subscription_data_from_fxa(social_account)	1✔
122	if "refreshed" in subscription_data.keys():	1✔
123	# user token refreshed for expanded scope
124	social_account.refresh_from_db()	1✔
125	# retry getting detailed subscription data
126	subscription_data = get_subscription_data_from_fxa(social_account)	1✔
127	if "refreshed" in subscription_data.keys():	1!
128	return None, None, None	1✔
129	if "subscriptions" not in subscription_data.keys():	1✔
130	# failed to get subscriptions data which may mean user never had subscription
131	# and/or there is data mismatch with FxA
132	if not flag_is_active_in_task("free_phones", social_account.user):	1✔
133	# User who was flagged for having phone subscriptions
134	# did not actually have phone subscriptions
135	logger.error(	1✔
136	"accounts_subscription_endpoint_failed",
137	extra={"fxa_message": subscription_data.get("message", "")},
138	)
139	return None, None, None	1✔
140
141	date_subscribed_phone = start_date = end_date = None	1✔
142	product_w_phone_capabilites = [settings.PHONE_PROD_ID, settings.BUNDLE_PROD_ID]	1✔
143	for sub in subscription_data.get("subscriptions", []):	1✔
144	# Even if a user upgrade subscription e.g. from monthly to yearly
145	# or from phone to VPN bundle use the last subscription subscription dates
146	# Later, when the subscription details only show one valid subsription
147	# this information can be updated
148	subscription_created_timestamp = None	1✔
149	subscription_start_timestamp = None	1✔
150	subscription_end_timestamp = None	1✔
151	if sub.get("product_id") in product_w_phone_capabilites:	1✔
152	subscription_created_timestamp = sub.get("created")	1✔
153	subscription_start_timestamp = sub.get("current_period_start")	1✔
154	subscription_end_timestamp = sub.get("current_period_end")	1✔
155	else:
156	# not a product id for phone subscription, continue
157	continue	1✔
158
159	subscription_date_none = (	1✔
160	subscription_created_timestamp
161	and subscription_start_timestamp
162	and subscription_end_timestamp
163	) is None
164	if subscription_date_none:	1✔
165	# subscription dates are required fields according to FxA documentation:
166	# https://mozilla.github.io/ecosystem-platform/api#tag/Subscriptions/operation/getOauthMozillasubscriptionsCustomerBillingandsubscriptions
167	logger.error(	1✔
168	"accounts_subscription_subscription_date_invalid",
169	extra={"subscription": sub},
170	)
171	return None, None, None	1✔
172
173	date_subscribed_phone = datetime.fromtimestamp(	1✔
174	subscription_created_timestamp, tz=UTC
175	)
176	start_date = datetime.fromtimestamp(subscription_start_timestamp, tz=UTC)	1✔
177	end_date = datetime.fromtimestamp(subscription_end_timestamp, tz=UTC)	1✔
178	return date_subscribed_phone, start_date, end_date	1✔

mozilla / fx-private-relay / 5e9994ac-40d1-44e7-a22f-8565e529b07d

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