Introduce claims constants

Author: lnagel

README.rst

@@ -31,13 +31,15 @@ Authentication class
 In order to get-or-create User accounts automatically within your microservice apps,
 you can use the following DRF Authentication class template::
 
+    from rest_framework_sso import claims
+    
     class Authentication(rest_framework_sso.authentication.JWTAuthentication):
         def authenticate_credentials(self, payload):
             user_model = get_user_model()
 
             user, created = user_model.objects.get_or_create(
-                service=payload.get('iss'),
-                external_id=payload.get('uid'),
+                service=payload.get(claims.ISSUER),
+                external_id=payload.get(claims.USER_ID),
             )
 
             if not user.is_active:

rest_framework_sso/claims.py

@@ -0,0 +1,19 @@
+# coding: utf-8
+from __future__ import absolute_import, unicode_literals
+
+ISSUER = 'iss'
+SUBJECT = 'sub'
+AUDIENCE = 'aud'
+EXPIRATION_TIME = 'exp'
+NOT_BEFORE = 'nbf'
+ISSUED_AT = 'iat'
+JWT_ID = 'jti'
+
+TOKEN = 'token'
+SESSION_ID = 'sid'
+USER_ID = 'uid'
+EMAIL = 'email'
+SCOPES = 'scopes'
+
+TOKEN_SESSION = 'session'
+TOKEN_AUTHORIZATION = 'auth'

rest_framework_sso/utils.py

@@ -11,58 +11,59 @@
 from jwt.exceptions import MissingRequiredClaimError, InvalidIssuerError
 from rest_framework import exceptions
 
+from rest_framework_sso import claims
 from rest_framework_sso.settings import api_settings
 
 
 def create_session_payload(session_token, user, **kwargs):
     return {
-        'type': 'session',
-        'sid': session_token.pk,
-        'uid': user.pk,
-        'email': user.email,
+        claims.TOKEN: claims.TOKEN_SESSION,
+        claims.SESSION_ID: session_token.pk,
+        claims.USER_ID: user.pk,
+        claims.EMAIL: user.email,
     }
 
 
 def create_authorization_payload(session_token, user, **kwargs):
     return {
-        'type': 'auth',
-        'sid': session_token.pk,
-        'uid': user.pk,
-        'email': user.email,
-        'scope': [],
+        claims.TOKEN: claims.TOKEN_AUTHORIZATION,
+        claims.SESSION_ID: session_token.pk,
+        claims.USER_ID: user.pk,
+        claims.EMAIL: user.email,
+        claims.SCOPES: [],
     }
 
 
 def encode_jwt_token(payload):
-    if payload.get('type') not in ('session', 'auth'):
+    if payload.get(claims.TOKEN) not in (claims.TOKEN_SESSION, claims.TOKEN_AUTHORIZATION):
         raise RuntimeError('Unknown token type')
 
-    if not payload.get('iss'):
+    if not payload.get(claims.ISSUER):
         if api_settings.IDENTITY is not None:
-            payload['iss'] = api_settings.IDENTITY
+            payload[claims.ISSUER] = api_settings.IDENTITY
         else:
             raise RuntimeError('IDENTITY must be specified in settings')
 
-    if not payload.get('aud'):
-        if payload.get('type') == 'session' and api_settings.SESSION_AUDIENCE is not None:
-            payload['aud'] = api_settings.SESSION_AUDIENCE
-        elif payload.get('type') == 'auth' and api_settings.AUTHORIZATION_AUDIENCE is not None:
-            payload['aud'] = api_settings.AUTHORIZATION_AUDIENCE
+    if not payload.get(claims.AUDIENCE):
+        if payload.get(claims.TOKEN) == claims.TOKEN_SESSION and api_settings.SESSION_AUDIENCE is not None:
+            payload[claims.AUDIENCE] = api_settings.SESSION_AUDIENCE
+        elif payload.get(claims.TOKEN) == claims.TOKEN_AUTHORIZATION and api_settings.AUTHORIZATION_AUDIENCE is not None:
+            payload[claims.AUDIENCE] = api_settings.AUTHORIZATION_AUDIENCE
         elif api_settings.IDENTITY is not None:
-            payload['aud'] = [api_settings.IDENTITY]
+            payload[claims.AUDIENCE] = [api_settings.IDENTITY]
         else:
             raise RuntimeError('SESSION_AUDIENCE must be specified in settings')
 
-    if not payload.get('exp'):
-        if payload.get('type') == 'session' and api_settings.SESSION_EXPIRATION is not None:
-            payload['exp'] = datetime.utcnow() + api_settings.SESSION_EXPIRATION
-        elif payload.get('type') == 'auth' and api_settings.AUTHORIZATION_EXPIRATION is not None:
-            payload['exp'] = datetime.utcnow() + api_settings.AUTHORIZATION_EXPIRATION
+    if not payload.get(claims.EXPIRATION_TIME):
+        if payload.get(claims.TOKEN) == claims.TOKEN_SESSION and api_settings.SESSION_EXPIRATION is not None:
+            payload[claims.EXPIRATION_TIME] = datetime.utcnow() + api_settings.SESSION_EXPIRATION
+        elif payload.get(claims.TOKEN) == claims.TOKEN_AUTHORIZATION and api_settings.AUTHORIZATION_EXPIRATION is not None:
+            payload[claims.EXPIRATION_TIME] = datetime.utcnow() + api_settings.AUTHORIZATION_EXPIRATION
 
-    if payload['iss'] not in api_settings.PRIVATE_KEYS:
+    if payload[claims.ISSUER] not in api_settings.PRIVATE_KEYS:
         raise RuntimeError('Private key for specified issuer was not found in settings')
 
-    private_key = open(api_settings.PRIVATE_KEYS.get(payload['iss']), 'rt').read()
+    private_key = open(api_settings.PRIVATE_KEYS.get(payload[claims.ISSUER]), 'rt').read()
 
     return jwt.encode(
         payload=payload,
@@ -75,10 +76,10 @@ def encode_jwt_token(payload):
 def decode_jwt_token(token):
     unverified_claims = jwt.decode(token, verify=False)
 
-    if 'iss' not in unverified_claims:
-        raise MissingRequiredClaimError('iss')
+    if claims.ISSUER not in unverified_claims:
+        raise MissingRequiredClaimError(claims.ISSUER)
 
-    unverified_issuer = six.text_type(unverified_claims['iss'])
+    unverified_issuer = six.text_type(unverified_claims[claims.ISSUER])
 
     if api_settings.ACCEPTED_ISSUERS is not None and unverified_issuer not in api_settings.ACCEPTED_ISSUERS:
         raise InvalidIssuerError('Invalid issuer')
@@ -115,13 +116,13 @@ def authenticate_payload(payload):
             session_token = SessionToken.objects.\
                 active().\
                 select_related('user').\
-                get(pk=payload.get('sid'), user_id=payload.get('uid'))
+                get(pk=payload.get(claims.SESSION_ID), user_id=payload.get(claims.USER_ID))
             user = session_token.user
         except SessionToken.DoesNotExist:
             raise exceptions.AuthenticationFailed(_('Invalid token.'))
     else:
         try:
-            user = user_model.objects.get(pk=payload.get('uid'))
+            user = user_model.objects.get(pk=payload.get(claims.USER_ID))
         except user_model.DoesNotExist:
             raise exceptions.AuthenticationFailed(_('Invalid token.'))
 

rest_framework_sso/views.py

@@ -6,6 +6,7 @@
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
+from rest_framework_sso import claims
 from rest_framework_sso.models import SessionToken
 from rest_framework_sso.serializers import SessionTokenSerializer, AuthorizationTokenSerializer
 from rest_framework_sso.settings import api_settings
@@ -88,10 +89,10 @@ def post(self, request, *args, **kwargs):
         serializer = self.get_serializer(data=request.data)
         serializer.is_valid(raise_exception=True)
 
-        if hasattr(request.auth, 'get') and request.auth.get('sid'):
+        if hasattr(request.auth, 'get') and request.auth.get(claims.SESSION_ID):
             try:
                 session_token = SessionToken.objects.active().\
-                    get(pk=request.auth.get('sid'), user=request.user)
+                    get(pk=request.auth.get(claims.SESSION_ID), user=request.user)
             except SessionToken.DoesNotExist:
                 return Response({'detail': 'Invalid token.'}, status=status.HTTP_401_UNAUTHORIZED)
         else: