Merge pull request jpadilla#126 from mark-adams/iat-nbf-checks

Add check for iat and new exception for nbf claims

Author: mark-adams

jwt/__init__.py

@@ -20,7 +20,7 @@
     encode, decode, register_algorithm, unregister_algorithm, PyJWT
 )
 from .exceptions import (
-    InvalidTokenError, DecodeError, ExpiredSignatureError,
-    InvalidAudienceError, InvalidIssuerError,
-    ExpiredSignature, InvalidAudience, InvalidIssuer
+    InvalidTokenError, DecodeError, InvalidAudienceError,
+    ExpiredSignatureError, ImmatureSignatureError, InvalidIssuedAtError,
+    InvalidIssuerError, ExpiredSignature, InvalidAudience, InvalidIssuer
 )

jwt/api.py

@@ -8,8 +8,9 @@
 from .algorithms import Algorithm, get_default_algorithms  # NOQA
 from .compat import string_types, text_type, timedelta_total_seconds
 from .exceptions import (
-    DecodeError, ExpiredSignatureError, InvalidAlgorithmError,
-    InvalidAudienceError, InvalidIssuerError
+    DecodeError, ExpiredSignatureError, ImmatureSignatureError,
+    InvalidAlgorithmError, InvalidAudienceError, InvalidIssuedAtError,
+    InvalidIssuerError
 )
 from .utils import base64url_decode, base64url_encode
 
@@ -184,32 +185,33 @@ def _validate_claims(self, payload, verify_expiration=True, leeway=0,
         if not isinstance(audience, (string_types, type(None))):
             raise TypeError('audience must be a string or None')
 
+        now = timegm(datetime.utcnow().utctimetuple())
+
         if 'iat' in payload:
             try:
-                int(payload['iat'])
+                iat = int(payload['iat'])
             except ValueError:
                 raise DecodeError('Issued At claim (iat) must be an integer.')
 
+            if iat > (now + leeway):
+                raise InvalidIssuedAtError('Issued At claim (iat) cannot be in the future.')
+
         if 'nbf' in payload and verify_expiration:
             try:
                 nbf = int(payload['nbf'])
             except ValueError:
                 raise DecodeError('Not Before claim (nbf) must be an integer.')
 
-            utc_timestamp = timegm(datetime.utcnow().utctimetuple())
-
-            if nbf > (utc_timestamp + leeway):
-                raise ExpiredSignatureError('Signature not yet valid')
+            if nbf > (now + leeway):
+                raise ImmatureSignatureError('The token is not yet valid (nbf)')
 
         if 'exp' in payload and verify_expiration:
             try:
                 exp = int(payload['exp'])
             except ValueError:
                 raise DecodeError('Expiration Time claim (exp) must be an integer.')
 
-            utc_timestamp = timegm(datetime.utcnow().utctimetuple())
-
-            if exp < (utc_timestamp - leeway):
+            if exp < (now - leeway):
                 raise ExpiredSignatureError('Signature has expired')
 
         if 'aud' in payload:

jwt/exceptions.py

@@ -18,6 +18,14 @@ class InvalidIssuerError(InvalidTokenError):
     pass
 
 
+class InvalidIssuedAtError(InvalidTokenError):
+    pass
+
+
+class ImmatureSignatureError(InvalidTokenError):
+    pass
+
+
 class InvalidKeyError(Exception):
     pass
 

tests/test_api.py

@@ -9,8 +9,9 @@
 from jwt.algorithms import Algorithm
 from jwt.api import PyJWT
 from jwt.exceptions import (
-    DecodeError, ExpiredSignatureError, InvalidAlgorithmError,
-    InvalidAudienceError, InvalidIssuerError
+    DecodeError, ExpiredSignatureError, ImmatureSignatureError,
+    InvalidAlgorithmError, InvalidAudienceError, InvalidIssuedAtError,
+    InvalidIssuerError
 )
 
 from .compat import text_type, unittest
@@ -210,7 +211,7 @@ def test_decode_raises_exception_if_iat_is_not_int(self):
                        'eyJpYXQiOiJub3QtYW4taW50In0.'
                        'H1GmcQgSySa5LOKYbzGm--b1OmRbHFkyk8pq811FzZM')
 
-        with self.assertRaisesRegexp(DecodeError, 'iat'):
+        with self.assertRaises(DecodeError):
             self.jwt.decode(example_jwt, 'secret')
 
     def test_decode_raises_exception_if_nbf_is_not_int(self):
@@ -219,9 +220,16 @@ def test_decode_raises_exception_if_nbf_is_not_int(self):
                        'eyJuYmYiOiJub3QtYW4taW50In0.'
                        'c25hldC8G2ZamC8uKpax9sYMTgdZo3cxrmzFHaAAluw')
 
-        with self.assertRaisesRegexp(DecodeError, 'nbf'):
+        with self.assertRaises(DecodeError):
             self.jwt.decode(example_jwt, 'secret')
 
+    def test_decode_raises_exception_if_iat_in_the_future(self):
+        now = datetime.utcnow()
+        token = self.jwt.encode({'iat': now + timedelta(days=1)}, key='secret')
+
+        with self.assertRaises(InvalidIssuedAtError):
+            self.jwt.decode(token, 'secret')
+
     def test_encode_datetime(self):
         secret = 'secret'
         current_datetime = datetime.utcnow()
@@ -451,7 +459,7 @@ def test_decode_with_notbefore(self):
         secret = 'secret'
         jwt_message = self.jwt.encode(self.payload, secret)
 
-        with self.assertRaises(ExpiredSignatureError):
+        with self.assertRaises(ImmatureSignatureError):
             self.jwt.decode(jwt_message, secret)
 
     def test_decode_skip_expiration_verification(self):
@@ -492,7 +500,7 @@ def test_decode_with_notbefore_with_leeway(self):
         # With 13 seconds leeway, should be ok
         self.jwt.decode(jwt_message, secret, leeway=13)
 
-        with self.assertRaises(ExpiredSignatureError):
+        with self.assertRaises(ImmatureSignatureError):
             self.jwt.decode(jwt_message, secret, leeway=1)
 
     def test_decode_with_algo_none_should_fail(self):