564 include leeway in validate_iat() to reject tokens that are 'issued in the future'

authlib/jose/rfc7519/claims.py

@@ -196,14 +196,19 @@ def validate_nbf(self, now, leeway):
 
     def validate_iat(self, now, leeway):
         """The "iat" (issued at) claim identifies the time at which the JWT was
-        issued.  This claim can be used to determine the age of the JWT.  Its
-        value MUST be a number containing a NumericDate value.  Use of this
-        claim is OPTIONAL.
+        issued.  This claim can be used to determine the age of the JWT.
+        Implementers MAY provide for some small leeway, usually no more
+        than a few minutes, to account for clock skew. Its value MUST be a
+        number containing a NumericDate value.  Use of this claim is OPTIONAL.
         """
         if 'iat' in self:
             iat = self['iat']
             if not _validate_numeric_time(iat):
                 raise InvalidClaimError('iat')
+            if iat > (now + leeway):
+                raise InvalidTokenError(
+                    description='The token is not valid as it was issued in the future'
+                )
 
     def validate_jti(self):
         """The "jti" (JWT ID) claim provides a unique identifier for the JWT.

tests/jose/test_jwt.py

@@ -147,6 +147,40 @@ def test_validate_nbf(self):
             claims.validate, 123
         )
 
+    def test_validate_iat_issued_in_future(self):
+        in_future = datetime.datetime.utcnow() + datetime.timedelta(seconds=10)
+        id_token = jwt.encode({'alg': 'HS256'}, {'iat': in_future}, 'k')
+        claims = jwt.decode(id_token, 'k')
+        with self.assertRaises(errors.InvalidTokenError) as error_ctx:
+            claims.validate()
+        self.assertEqual(
+            str(error_ctx.exception),
+            'invalid_token: The token is not valid as it was issued in the future'
+        )
+
+    def test_validate_iat_issued_in_future_with_insufficient_leeway(self):
+        in_future = datetime.datetime.utcnow() + datetime.timedelta(seconds=10)
+        id_token = jwt.encode({'alg': 'HS256'}, {'iat': in_future}, 'k')
+        claims = jwt.decode(id_token, 'k')
+        with self.assertRaises(errors.InvalidTokenError) as error_ctx:
+            claims.validate(leeway=5)
+        self.assertEqual(
+            str(error_ctx.exception),
+            'invalid_token: The token is not valid as it was issued in the future'
+        )
+
+    def test_validate_iat_issued_in_future_with_sufficient_leeway(self):
+        in_future = datetime.datetime.utcnow() + datetime.timedelta(seconds=10)
+        id_token = jwt.encode({'alg': 'HS256'}, {'iat': in_future}, 'k')
+        claims = jwt.decode(id_token, 'k')
+        claims.validate(leeway=20)
+
+    def test_validate_iat_issued_in_past(self):
+        in_future = datetime.datetime.utcnow() - datetime.timedelta(seconds=10)
+        id_token = jwt.encode({'alg': 'HS256'}, {'iat': in_future}, 'k')
+        claims = jwt.decode(id_token, 'k')
+        claims.validate()
+
     def test_validate_iat(self):
         id_token = jwt.encode({'alg': 'HS256'}, {'iat': 'invalid'}, 'k')
         claims = jwt.decode(id_token, 'k')