identity claim -> standard claim 'sub'

Author: IndominusByte

fastapi_jwt_auth/auth_jwt.py

@@ -120,7 +120,7 @@ def _get_secret_key(self, algorithm: str, process: str) -> str:
 
     def _create_token(
         self,
-        identity: Union[str,int],
+        subject: Union[str,int],
         type_token: str,
         exp_time: Optional[int],
         fresh: Optional[bool] = False,
@@ -132,7 +132,7 @@ def _create_token(
         """
         Create token for access_token and refresh_token (utf-8)
 
-        :param identity: Identifier for who this token is for example id or username from database.
+        :param subject: Identifier for who this token is for example id or username from database.
         :param type_token: indicate token is access_token or refresh_token
         :param exp_time: Set the duration of the JWT
         :param fresh: Optional when token is access_token this param required
@@ -144,8 +144,8 @@ def _create_token(
         :return: Encoded token
         """
         # Validation type data
-        if not isinstance(identity, (str,int)):
-            raise TypeError("identity must be a string or integer")
+        if not isinstance(subject, (str,int)):
+            raise TypeError("subject must be a string or integer")
         if not isinstance(fresh, (bool)):
             raise TypeError("fresh must be a boolean")
         if audience and not isinstance(audience, (str, list, tuple, set, frozenset, GeneratorType)):
@@ -155,15 +155,13 @@ def _create_token(
 
         # Data section
         reserved_claims = {
+            "sub": subject,
             "iat": self._get_int_from_datetime(datetime.now(timezone.utc)),
             "nbf": self._get_int_from_datetime(datetime.now(timezone.utc)),
-            "jti": self._get_jwt_identifier(),
+            "jti": self._get_jwt_identifier()
         }
 
-        custom_claims = {
-            "identity": identity,
-            "type": type_token
-        }
+        custom_claims = {"type": type_token}
 
         # for access_token only fresh needed
         if type_token == 'access':
@@ -253,7 +251,7 @@ def _get_expired_time(
 
     def create_access_token(
         self,
-        identity: Union[str,int],
+        subject: Union[str,int],
         fresh: Optional[bool] = False,
         algorithm: Optional[str] = None,
         headers: Optional[Dict] = None,
@@ -267,7 +265,7 @@ def create_access_token(
         :return: hash token
         """
         return self._create_token(
-            identity=identity,
+            subject=subject,
             type_token="access",
             exp_time=self._get_expired_time("access",expires_time),
             fresh=fresh,
@@ -279,7 +277,7 @@ def create_access_token(
 
     def create_refresh_token(
         self,
-        identity: Union[str,int],
+        subject: Union[str,int],
         algorithm: Optional[str] = None,
         headers: Optional[Dict] = None,
         expires_time: Optional[Union[timedelta,int,bool]] = None,
@@ -292,7 +290,7 @@ def create_refresh_token(
         :return: hash token
         """
         return self._create_token(
-            identity=identity,
+            subject=subject,
             type_token="refresh",
             exp_time=self._get_expired_time("refresh",expires_time),
             algorithm=algorithm,
@@ -634,9 +632,9 @@ def jwt_required(self) -> None:
 
     def jwt_optional(self) -> None:
         """
-        If an access token in present in the request you can get data from get_raw_jwt() or get_jwt_identity(),
+        If an access token in present in the request you can get data from get_raw_jwt() or get_jwt_subject(),
         If no access token is present in the request, this endpoint will still be called, but
-        get_raw_jwt() or get_jwt_identity() will return None
+        get_raw_jwt() or get_jwt_subject() will return None
         """
         if len(self._token_location) == 2:
             if self._token and self.jwt_in_headers:
@@ -698,15 +696,15 @@ def get_jti(self,encoded_token: str) -> str:
         """
         return self._verified_token(encoded_token)['jti']
 
-    def get_jwt_identity(self) -> Optional[Union[str,int]]:
+    def get_jwt_subject(self) -> Optional[Union[str,int]]:
         """
-        this will return the identity of the JWT that is accessing this endpoint.
+        this will return the subject of the JWT that is accessing this endpoint.
         If no JWT is present, `None` is returned instead.
 
-        :return: identity of JWT
+        :return: sub of JWT
         """
         if self._token:
-            return self._verified_token(self._token)['identity']
+            return self._verified_token(self._token)['sub']
         return None
 
     def get_unverified_jwt_headers(self,encoded_token: Optional[str] = None) -> dict:

tests/test_config.py

@@ -71,20 +71,20 @@ class TokenFalse(BaseSettings):
     def get_expired_false():
         return TokenFalse()
 
-    access_token = Authorize.create_access_token(identity=1)
+    access_token = Authorize.create_access_token(subject=1)
     assert 'exp' not in jwt.decode(access_token,"testing",algorithms="HS256")
 
-    refresh_token = Authorize.create_refresh_token(identity=1)
+    refresh_token = Authorize.create_refresh_token(subject=1)
     assert 'exp' not in jwt.decode(refresh_token,"testing",algorithms="HS256")
 
 def test_secret_key_not_exist(client,Authorize):
     AuthJWT._secret_key = None
 
     with pytest.raises(RuntimeError,match=r"AUTHJWT_SECRET_KEY"):
-        Authorize.create_access_token(identity='test')
+        Authorize.create_access_token(subject='test')
 
     Authorize._secret_key = "secret"
-    token = Authorize.create_access_token(identity=1)
+    token = Authorize.create_access_token(subject=1)
     Authorize._secret_key = None
 
     with pytest.raises(RuntimeError,match=r"AUTHJWT_SECRET_KEY"):
@@ -103,7 +103,7 @@ def get_settings_one():
 
     Authorize = AuthJWT()
 
-    token = Authorize.create_access_token(identity='test')
+    token = Authorize.create_access_token(subject='test')
 
     response = client.get('/protected',headers={"Authorization": f"Bearer {token}"})
     assert response.status_code == 200

tests/test_create_token.py

@@ -16,80 +16,80 @@ def get_settings():
     with pytest.raises(TypeError,match=r"missing 1 required positional argument"):
         Authorize.create_access_token()
 
-    with pytest.raises(TypeError,match=r"identity"):
-        Authorize.create_access_token(identity=0.123)
+    with pytest.raises(TypeError,match=r"subject"):
+        Authorize.create_access_token(subject=0.123)
 
     with pytest.raises(TypeError,match=r"fresh"):
-        Authorize.create_access_token(identity="test",fresh="lol")
+        Authorize.create_access_token(subject="test",fresh="lol")
 
     with pytest.raises(ValueError,match=r"dictionary update sequence element"):
-        Authorize.create_access_token(identity=1,headers="test")
+        Authorize.create_access_token(subject=1,headers="test")
 
 def test_create_refresh_token(Authorize):
     with pytest.raises(TypeError,match=r"missing 1 required positional argument"):
         Authorize.create_refresh_token()
 
-    with pytest.raises(TypeError,match=r"identity"):
-        Authorize.create_refresh_token(identity=0.123)
+    with pytest.raises(TypeError,match=r"subject"):
+        Authorize.create_refresh_token(subject=0.123)
 
     with pytest.raises(ValueError,match=r"dictionary update sequence element"):
-        Authorize.create_refresh_token(identity=1,headers="test")
+        Authorize.create_refresh_token(subject=1,headers="test")
 
 def test_create_dynamic_access_token_expires(Authorize):
     expires_time = int(datetime.now(timezone.utc).timestamp()) + 90
-    token = Authorize.create_access_token(identity=1,expires_time=90)
+    token = Authorize.create_access_token(subject=1,expires_time=90)
     assert jwt.decode(token,"testing",algorithms="HS256")['exp'] == expires_time
 
     expires_time = int(datetime.now(timezone.utc).timestamp()) + 86400
-    token = Authorize.create_access_token(identity=1,expires_time=timedelta(days=1))
+    token = Authorize.create_access_token(subject=1,expires_time=timedelta(days=1))
     assert jwt.decode(token,"testing",algorithms="HS256")['exp'] == expires_time
 
     expires_time = int(datetime.now(timezone.utc).timestamp()) + 2
-    token = Authorize.create_access_token(identity=1,expires_time=True)
+    token = Authorize.create_access_token(subject=1,expires_time=True)
     assert jwt.decode(token,"testing",algorithms="HS256")['exp'] == expires_time
 
-    token = Authorize.create_access_token(identity=1,expires_time=False)
+    token = Authorize.create_access_token(subject=1,expires_time=False)
     assert 'exp' not in jwt.decode(token,"testing",algorithms="HS256")
 
     with pytest.raises(TypeError,match=r"expires_time"):
-        Authorize.create_access_token(identity=1,expires_time="test")
+        Authorize.create_access_token(subject=1,expires_time="test")
 
 def test_create_dynamic_refresh_token_expires(Authorize):
     expires_time = int(datetime.now(timezone.utc).timestamp()) + 90
-    token = Authorize.create_refresh_token(identity=1,expires_time=90)
+    token = Authorize.create_refresh_token(subject=1,expires_time=90)
     assert jwt.decode(token,"testing",algorithms="HS256")['exp'] == expires_time
 
     expires_time = int(datetime.now(timezone.utc).timestamp()) + 86400
-    token = Authorize.create_refresh_token(identity=1,expires_time=timedelta(days=1))
+    token = Authorize.create_refresh_token(subject=1,expires_time=timedelta(days=1))
     assert jwt.decode(token,"testing",algorithms="HS256")['exp'] == expires_time
 
     expires_time = int(datetime.now(timezone.utc).timestamp()) + 4
-    token = Authorize.create_refresh_token(identity=1,expires_time=True)
+    token = Authorize.create_refresh_token(subject=1,expires_time=True)
     assert jwt.decode(token,"testing",algorithms="HS256")['exp'] == expires_time
 
-    token = Authorize.create_refresh_token(identity=1,expires_time=False)
+    token = Authorize.create_refresh_token(subject=1,expires_time=False)
     assert 'exp' not in jwt.decode(token,"testing",algorithms="HS256")
 
     with pytest.raises(TypeError,match=r"expires_time"):
-        Authorize.create_refresh_token(identity=1,expires_time="test")
+        Authorize.create_refresh_token(subject=1,expires_time="test")
 
 def test_create_token_invalid_type_data_audience(Authorize):
     with pytest.raises(TypeError,match=r"audience"):
-        Authorize.create_access_token(identity=1,audience=1)
+        Authorize.create_access_token(subject=1,audience=1)
 
     with pytest.raises(TypeError,match=r"audience"):
-        Authorize.create_refresh_token(identity=1,audience=1)
+        Authorize.create_refresh_token(subject=1,audience=1)
 
 def test_create_token_invalid_algorithm(Authorize):
     with pytest.raises(ValueError,match=r"Algorithm"):
-        Authorize.create_access_token(identity=1,algorithm="test")
+        Authorize.create_access_token(subject=1,algorithm="test")
 
     with pytest.raises(ValueError,match=r"Algorithm"):
-        Authorize.create_refresh_token(identity=1,algorithm="test")
+        Authorize.create_refresh_token(subject=1,algorithm="test")
 
 def test_create_token_invalid_type_data_algorithm(Authorize):
     with pytest.raises(TypeError,match=r"algorithm"):
-        Authorize.create_access_token(identity=1,algorithm=1)
+        Authorize.create_access_token(subject=1,algorithm=1)
 
     with pytest.raises(TypeError,match=r"algorithm"):
-        Authorize.create_refresh_token(identity=1,algorithm=1)
+        Authorize.create_refresh_token(subject=1,algorithm=1)