Andal.LND/edx-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: add create_jwt_token_dict (#30485)

Browse Source 
Moves the functionality from views._get_jwt_dict_from_access_token_dict
to a new method jwt.create_jwt_token_dict, which create a JWT version
of the passed token_dict. Also updates create_jwt_from_token to wrap
this call and return the "access_token" from the dict.

This will hopefully make it less likely that the token
dict attributes could get out of sync with the claims
inside the JWT.
This commit is contained in:
Robert Raposa
2022-05-25 17:02:19 -04:00
committed by GitHub
parent d828f155a6
commit 432cde31a5
3 changed files with 66 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
openedx/core/djangoapps/oauth_dispatch/jwt.py
View File

@@ -43,13 +43,14 @@ def create_jwt_for_user(user, secret=None, aud=None, additional_claims=None, sco
)
def create_jwt_from_token(token_dict, oauth_adapter, use_asymmetric_key=None):
def create_jwt_token_dict(token_dict, oauth_adapter, use_asymmetric_key=None):
"""
Returns a JWT created from the given access token.
Returns a JWT access token dict based on the provided access token.
Arguments:
token_dict (dict): An access token structure as returned from an
underlying OAuth provider.
underlying OAuth provider. Dict includes "access_token",
"expires_in", "token_type", and "scope".
Deprecated Arguments (to be removed):
oauth_adapter (DOPAdapter|DOTAdapter): An OAuth adapter that will
@@ -61,18 +62,39 @@ def create_jwt_from_token(token_dict, oauth_adapter, use_asymmetric_key=None):
access_token = oauth_adapter.get_access_token(token_dict['access_token'])
client = oauth_adapter.get_client_for_token(access_token)
# TODO (ARCH-204) put access_token as a JWT ID claim (jti)
return _create_jwt(
jwt_expires_in = _get_jwt_access_token_expire_seconds()
jwt_access_token = _create_jwt(
access_token.user,
scopes=token_dict['scope'].split(' '),
expires_in=get_jwt_access_token_expire_seconds(),
expires_in=jwt_expires_in,
use_asymmetric_key=use_asymmetric_key,
is_restricted=oauth_adapter.is_client_restricted(client),
filters=oauth_adapter.get_authorization_filters(client),
)
jwt_token_dict = token_dict.copy()
# Note: only "scope" is not overwritten at this point.
jwt_token_dict.update({
"access_token": jwt_access_token,
"token_type": "JWT",
"expires_in": jwt_expires_in,
})
return jwt_token_dict
def get_jwt_access_token_expire_seconds():
def create_jwt_from_token(token_dict, oauth_adapter, use_asymmetric_key=None):
"""
Returns a JWT created from the provided access token dict.
Note: if you need the token dict, and not just the JWT, use
create_jwt_token_dict instead. See its docs for more details.
"""
jwt_token_dict = create_jwt_token_dict(token_dict, oauth_adapter, use_asymmetric_key)
return jwt_token_dict["access_token"]
def _get_jwt_access_token_expire_seconds():
"""
Returns the number of seconds before a JWT access token expires.

40
openedx/core/djangoapps/oauth_dispatch/tests/test_jwt.py
View File

@@ -38,10 +38,8 @@ class TestCreateJWTs(AccessTokenMixin, TestCase):
RestrictedApplication.objects.create(application=client)
return client
def _create_jwt_for_token(
self, oauth_adapter, use_asymmetric_key, client_restricted=False,
):
""" Creates and returns the jwt returned by jwt_api.create_jwt_from_token. """
def _get_token_dict(self, client_restricted, oauth_adapter):
""" Creates and returns an (opaque) access token dict """
client = self._create_client(oauth_adapter, client_restricted)
expires_in = 60 * 60
expires = now() + timedelta(seconds=expires_in)
@@ -50,6 +48,13 @@ class TestCreateJWTs(AccessTokenMixin, TestCase):
expires_in=expires_in,
scope=' '.join(self.default_scopes)
)
return token_dict
def _create_jwt_for_token(
self, oauth_adapter, use_asymmetric_key, client_restricted=False,
):
""" Creates and returns the jwt returned by jwt_api.create_jwt_from_token. """
token_dict = self._get_token_dict(client_restricted, oauth_adapter)
return jwt_api.create_jwt_from_token(token_dict, oauth_adapter, use_asymmetric_key=use_asymmetric_key)
def _assert_jwt_is_valid(self, jwt_token, should_be_asymmetric_key):
@@ -84,8 +89,33 @@ class TestCreateJWTs(AccessTokenMixin, TestCase):
jwt_token, self.user, self.default_scopes, expires_in=expected_expires_in,
)
def test_create_jwt_token_dict_for_default_expire_seconds(self):
oauth_adapter = DOTAdapter()
token_dict = self._get_token_dict(client_restricted=False, oauth_adapter=oauth_adapter)
jwt_token_dict = jwt_api.create_jwt_token_dict(token_dict, oauth_adapter, use_asymmetric_key=False)
expected_expires_in = 60 * 60
self.assert_valid_jwt_access_token(
jwt_token_dict["access_token"], self.user, self.default_scopes, expires_in=expected_expires_in,
)
assert jwt_token_dict["token_type"] == "JWT"
assert jwt_token_dict["expires_in"] == expected_expires_in
assert jwt_token_dict["scope"] == token_dict["scope"]
def test_create_jwt_token_dict_for_overridden_expire_seconds(self):
oauth_adapter = DOTAdapter()
expected_expires_in = 60
with override_settings(JWT_ACCESS_TOKEN_EXPIRE_SECONDS=expected_expires_in):
token_dict = self._get_token_dict(client_restricted=False, oauth_adapter=oauth_adapter)
jwt_token_dict = jwt_api.create_jwt_token_dict(token_dict, oauth_adapter, use_asymmetric_key=False)
self.assert_valid_jwt_access_token(
jwt_token_dict["access_token"], self.user, self.default_scopes, expires_in=expected_expires_in,
)
assert jwt_token_dict["token_type"] == "JWT"
assert jwt_token_dict["expires_in"] == expected_expires_in
assert jwt_token_dict["scope"] == token_dict["scope"]
@ddt.data((True, False))
def test_dot_create_jwt_for_token(self, client_restricted):
def test_create_jwt_for_client_restricted(self, client_restricted):
jwt_token = self._create_jwt_for_token(
DOTAdapter(),
use_asymmetric_key=None,

26
openedx/core/djangoapps/oauth_dispatch/views.py
View File

@@ -17,7 +17,7 @@ from ratelimit.decorators import ratelimit
from openedx.core.djangoapps.auth_exchange import views as auth_exchange_views
from openedx.core.djangoapps.oauth_dispatch import adapters
from openedx.core.djangoapps.oauth_dispatch.dot_overrides import views as dot_overrides_views
from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_from_token, get_jwt_access_token_expire_seconds
from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_token_dict
class _DispatchingView(View):
@@ -88,26 +88,6 @@ def _get_token_type(request):
return token_type
def _get_jwt_dict_from_access_token_dict(token_dict, oauth_adapter):
"""
Returns a JWT token dict from the provided original (opaque) access token dict.
Creates the new JWT, and then overrides various values in a copy of the
token dict with the JWT specific values.
"""
jwt_dict = token_dict.copy()
# TODO: It would be safer if create_jwt_from_token returned this
# dict directly, so it would not be possible for the dict and JWT
# to get out of sync, but that is a larger refactor to think through.
jwt = create_jwt_from_token(jwt_dict, oauth_adapter)
jwt_dict.update({
'access_token': jwt,
'token_type': 'JWT',
'expires_in': get_jwt_access_token_expire_seconds(),
})
return jwt_dict
@method_decorator(
ratelimit(
key='openedx.core.djangoapps.util.ratelimit.real_ip', rate=settings.RATELIMIT_RATE,
@@ -137,9 +117,7 @@ class AccessTokenView(_DispatchingView):
Includes the JWT token and token type in the response.
"""
opaque_token_dict = json.loads(response.content.decode('utf-8'))
jwt_token_dict = _get_jwt_dict_from_access_token_dict(
opaque_token_dict, self.get_adapter(request)
)
jwt_token_dict = create_jwt_token_dict(opaque_token_dict, self.get_adapter(request))
return json.dumps(jwt_token_dict)