8354d07c-7eab-4972-926d-a2104e534166

from datetime import datetime, timezone
import logging
import shlex

import requests

from django.conf import settings
from django.core.cache import cache

from allauth.socialaccount.models import SocialAccount
from rest_framework.authentication import BaseAuthentication, get_authorization_header
from rest_framework.exceptions import (
    APIException,
    AuthenticationFailed,
    NotFound,
    ParseError,
    PermissionDenied,
)


logger = logging.getLogger("events")
INTROSPECT_TOKEN_URL = (
    "%s/introspect" % settings.SOCIALACCOUNT_PROVIDERS["fxa"]["OAUTH_ENDPOINT"]
)


def get_cache_key(token):
    return hash(token)


def introspect_token(token):
    try:
        fxa_resp = requests.post(INTROSPECT_TOKEN_URL, json={"token": token})
    except:
        logger.error(
            "Could not introspect token with FXA.",
            extra={"fxa_response": shlex.quote(fxa_resp.text)},
        )
        raise AuthenticationFailed("Could not introspect token with FXA.")

    fxa_resp_data = {"status_code": fxa_resp.status_code, "json": {}}
    try:
        fxa_resp_data["json"] = fxa_resp.json()
    except requests.exceptions.JSONDecodeError:
        logger.error(
            "JSONDecodeError from FXA introspect response.",
            extra={"fxa_response": shlex.quote(fxa_resp.text)},
        )
        raise AuthenticationFailed("JSONDecodeError from FXA introspect response")
    return fxa_resp_data


def get_fxa_uid_from_oauth_token(token):
    cache_key = get_cache_key(token)
    # set a default cache_timeout, but this will be overriden to match
    # the 'exp' time in the JWT returned by FxA
    cache_timeout = 60

    # set a default fxa_resp_data, so any error during introspection
    # will still cache for at least cache_timeout to prevent an outage
    # from causing useless run-away repetitive introspection requests
    fxa_resp_data = {"status_code": None, "json": {}}
    try:
        cached_fxa_resp_data = cache.get(cache_key)

        if cached_fxa_resp_data:
            fxa_resp_data = cached_fxa_resp_data
        else:
            # no cached data, get new
            fxa_resp_data = introspect_token(token)
    except AuthenticationFailed:
        raise
    finally:
        # Store potential valid response, errors, inactive users, etc. from FxA
        # for at least 60 seconds. Valid access_token cache extended after checking.
        cache.set(cache_key, fxa_resp_data, cache_timeout)

    if fxa_resp_data["status_code"] is None:
        raise APIException("Previous FXA call failed, wait to retry.")

    if not fxa_resp_data["status_code"] == 200:
        raise APIException("Did not receive a 200 response from FXA.")

    if not fxa_resp_data["json"].get("active"):
        raise AuthenticationFailed("FXA returned active: False for token.", code=401)

    # FxA user is active, check for the associated Relay account
    fxa_uid = fxa_resp_data.get("json", {}).get("sub")
    if not fxa_uid:
        raise NotFound("FXA did not return an FXA UID.")

    # cache valid access_token and fxa_resp_data until access_token expiration
    # TODO: revisit this since the token can expire before its time
    if type(fxa_resp_data.get("json").get("exp")) is int:
        # Note: FXA iat and exp are timestamps in *milliseconds*
        fxa_token_exp_time = int(fxa_resp_data.get("json").get("exp") / 1000)
        now_time = int(datetime.now(timezone.utc).timestamp())
        fxa_token_exp_cache_timeout = fxa_token_exp_time - now_time
        if fxa_token_exp_cache_timeout > cache_timeout:
            # cache until access_token expires (matched Relay user)
            # this handles cases where the token already expired
            cache_timeout = fxa_token_exp_cache_timeout
    cache.set(cache_key, fxa_resp_data, cache_timeout)

    return fxa_uid


class FxaTokenAuthentication(BaseAuthentication):
    def authenticate_header(self, request):
        # Note: we need to implement this function to make DRF return a 401 status code
        # when we raise AuthenticationFailed, rather than a 403.
        # See https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
        return "Bearer"

    def authenticate(self, request):
        authorization = get_authorization_header(request).decode()
        if not authorization or not authorization.startswith("Bearer "):
            # If the request has no Bearer token, return None to attempt the next
            # auth scheme in the REST_FRAMEWORK AUTHENTICATION_CLASSES list
            return None

        token = authorization.split(" ")[1]
        if token == "":
            raise ParseError("Missing FXA Token after 'Bearer'.")

        fxa_uid = get_fxa_uid_from_oauth_token(token)
        try:
            sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")
        except SocialAccount.DoesNotExist:
            raise PermissionDenied("Authenticated user does not have a Relay account.")
        user = sa.user

        if user:
            return (user, token)
        else:
            raise NotFound()

1	from datetime import datetime, timezone	1✔
2	import logging	1✔
3	import shlex	1✔
4
5	import requests	1✔
6
7	from django.conf import settings	1✔
8	from django.core.cache import cache	1✔
9
10	from allauth.socialaccount.models import SocialAccount	1✔
11	from rest_framework.authentication import BaseAuthentication, get_authorization_header	1✔
12	from rest_framework.exceptions import (	1✔
13	APIException,
14	AuthenticationFailed,
15	NotFound,
16	ParseError,
17	PermissionDenied,
18	)
19
20
21	logger = logging.getLogger("events")	1✔
22	INTROSPECT_TOKEN_URL = (	1✔
23	"%s/introspect" % settings.SOCIALACCOUNT_PROVIDERS["fxa"]["OAUTH_ENDPOINT"]
24	)
25
26
27	def get_cache_key(token):	1✔
28	return hash(token)	1✔
29
30
31	def introspect_token(token):	1✔
32	try:	1✔
33	fxa_resp = requests.post(INTROSPECT_TOKEN_URL, json={"token": token})	1✔
34	except:	×
35	logger.error(	×
36	"Could not introspect token with FXA.",
37	extra={"fxa_response": shlex.quote(fxa_resp.text)},
38	)
39	raise AuthenticationFailed("Could not introspect token with FXA.")	×
40
41	fxa_resp_data = {"status_code": fxa_resp.status_code, "json": {}}	1✔
42	try:	1✔
43	fxa_resp_data["json"] = fxa_resp.json()	1✔
44	except requests.exceptions.JSONDecodeError:	1✔
45	logger.error(	1✔
46	"JSONDecodeError from FXA introspect response.",
47	extra={"fxa_response": shlex.quote(fxa_resp.text)},
48	)
49	raise AuthenticationFailed("JSONDecodeError from FXA introspect response")	1✔
50	return fxa_resp_data	1✔
51
52
53	def get_fxa_uid_from_oauth_token(token):	1✔
54	cache_key = get_cache_key(token)	1✔
55	# set a default cache_timeout, but this will be overriden to match
56	# the 'exp' time in the JWT returned by FxA
57	cache_timeout = 60	1✔
58
59	# set a default fxa_resp_data, so any error during introspection
60	# will still cache for at least cache_timeout to prevent an outage
61	# from causing useless run-away repetitive introspection requests
62	fxa_resp_data = {"status_code": None, "json": {}}	1✔
63	try:	1✔
64	cached_fxa_resp_data = cache.get(cache_key)	1✔
65
66	if cached_fxa_resp_data:	1✔
67	fxa_resp_data = cached_fxa_resp_data	1✔
68	else:
69	# no cached data, get new
70	fxa_resp_data = introspect_token(token)	1✔
71	except AuthenticationFailed:	1✔
72	raise	1✔
73	finally:
74	# Store potential valid response, errors, inactive users, etc. from FxA
75	# for at least 60 seconds. Valid access_token cache extended after checking.
76	cache.set(cache_key, fxa_resp_data, cache_timeout)	1✔
77
78	if fxa_resp_data["status_code"] is None:	1✔
79	raise APIException("Previous FXA call failed, wait to retry.")	1✔
80
81	if not fxa_resp_data["status_code"] == 200:	1✔
82	raise APIException("Did not receive a 200 response from FXA.")	1✔
83
84	if not fxa_resp_data["json"].get("active"):	1✔
85	raise AuthenticationFailed("FXA returned active: False for token.", code=401)	1✔
86
87	# FxA user is active, check for the associated Relay account
88	fxa_uid = fxa_resp_data.get("json", {}).get("sub")	1✔
89	if not fxa_uid:	1✔
90	raise NotFound("FXA did not return an FXA UID.")	1✔
91
92	# cache valid access_token and fxa_resp_data until access_token expiration
93	# TODO: revisit this since the token can expire before its time
94	if type(fxa_resp_data.get("json").get("exp")) is int:	1✔
95	# Note: FXA iat and exp are timestamps in milliseconds
96	fxa_token_exp_time = int(fxa_resp_data.get("json").get("exp") / 1000)	1✔
97	now_time = int(datetime.now(timezone.utc).timestamp())	1✔
98	fxa_token_exp_cache_timeout = fxa_token_exp_time - now_time	1✔
99	if fxa_token_exp_cache_timeout > cache_timeout:	1!
100	# cache until access_token expires (matched Relay user)
101	# this handles cases where the token already expired
102	cache_timeout = fxa_token_exp_cache_timeout	1✔
103	cache.set(cache_key, fxa_resp_data, cache_timeout)	1✔
104
105	return fxa_uid	1✔
106
107
108	class FxaTokenAuthentication(BaseAuthentication):	1✔
109	def authenticate_header(self, request):	1✔
110	# Note: we need to implement this function to make DRF return a 401 status code
111	# when we raise AuthenticationFailed, rather than a 403.
112	# See https://www.django-rest-framework.org/api-guide/authentication/#custom-authentication
113	return "Bearer"	1✔
114
115	def authenticate(self, request):	1✔
116	authorization = get_authorization_header(request).decode()	1✔
117	if not authorization or not authorization.startswith("Bearer "):	1✔
118	# If the request has no Bearer token, return None to attempt the next
119	# auth scheme in the REST_FRAMEWORK AUTHENTICATION_CLASSES list
120	return None	1✔
121
122	token = authorization.split(" ")[1]	1✔
123	if token == "":	1✔
124	raise ParseError("Missing FXA Token after 'Bearer'.")	1✔
125
126	fxa_uid = get_fxa_uid_from_oauth_token(token)	1✔
127	try:	1✔
128	sa = SocialAccount.objects.get(uid=fxa_uid, provider="fxa")	1✔
129	except SocialAccount.DoesNotExist:	1✔
130	raise PermissionDenied("Authenticated user does not have a Relay account.")	1✔
131	user = sa.user	1✔
132
133	if user:	1!
134	return (user, token)	1✔
135	else:
136	raise NotFound()	×

mozilla / fx-private-relay / 8354d07c-7eab-4972-926d-a2104e534166

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