Add option to provide current time when decoding JWT

Author: ZipFile

jose/jwt.py

@@ -53,7 +53,7 @@ def encode(claims, key, algorithm=ALGORITHMS.HS256, headers=None, access_token=N
     return jws.sign(claims, key, headers=headers, algorithm=algorithm)
 
 
-def decode(token, key, algorithms=None, options=None, audience=None, issuer=None, subject=None, access_token=None):
+def decode(token, key, algorithms=None, options=None, audience=None, issuer=None, subject=None, access_token=None, now=None):
     """Verifies a JWT string's signature and validates reserved claims.
 
     Args:
@@ -73,6 +73,7 @@ def decode(token, key, algorithms=None, options=None, audience=None, issuer=None
         access_token (str): An access token string. If the "at_hash" claim is included in the
             claim set, then the access_token must be included, and it must match
             the "at_hash" claim.
+        now (datetime): Current time. If not set, defaults to current system time.
         options (dict): A dictionary of options for skipping validation steps.
 
             defaults = {
@@ -155,6 +156,7 @@ def decode(token, key, algorithms=None, options=None, audience=None, issuer=None
         raise JWTError("Invalid payload string: must be a json object")
 
     _validate_claims(
+        now or datetime.utcnow(),
         claims,
         audience=audience,
         issuer=issuer,
@@ -254,7 +256,7 @@ def _validate_iat(claims):
         raise JWTClaimsError("Issued At claim (iat) must be an integer.")
 
 
-def _validate_nbf(claims, leeway=0):
+def _validate_nbf(now, claims, leeway=0):
     """Validates that the 'nbf' claim is valid.
 
     The "nbf" (not before) claim identifies the time before which the JWT
@@ -266,6 +268,7 @@ def _validate_nbf(claims, leeway=0):
     NumericDate value.  Use of this claim is OPTIONAL.
 
     Args:
+        now (datetime): Current time.
         claims (dict): The claims dictionary to validate.
         leeway (int): The number of seconds of skew that is allowed.
     """
@@ -278,13 +281,13 @@ def _validate_nbf(claims, leeway=0):
     except ValueError:
         raise JWTClaimsError("Not Before claim (nbf) must be an integer.")
 
-    now = timegm(datetime.utcnow().utctimetuple())
+    now = timegm(now.utctimetuple())
 
     if nbf > (now + leeway):
         raise JWTClaimsError("The token is not yet valid (nbf)")
 
 
-def _validate_exp(claims, leeway=0):
+def _validate_exp(now, claims, leeway=0):
     """Validates that the 'exp' claim is valid.
 
     The "exp" (expiration time) claim identifies the expiration time on
@@ -296,6 +299,7 @@ def _validate_exp(claims, leeway=0):
     containing a NumericDate value.  Use of this claim is OPTIONAL.
 
     Args:
+        now (datetime): Current time.
         claims (dict): The claims dictionary to validate.
         leeway (int): The number of seconds of skew that is allowed.
     """
@@ -308,7 +312,7 @@ def _validate_exp(claims, leeway=0):
     except ValueError:
         raise JWTClaimsError("Expiration Time claim (exp) must be an integer.")
 
-    now = timegm(datetime.utcnow().utctimetuple())
+    now = timegm(now.utctimetuple())
 
     if exp < (now - leeway):
         raise ExpiredSignatureError("Signature has expired.")
@@ -455,7 +459,7 @@ def _validate_at_hash(claims, access_token, algorithm):
         raise JWTClaimsError("at_hash claim does not match access_token.")
 
 
-def _validate_claims(claims, audience=None, issuer=None, subject=None, algorithm=None, access_token=None, options=None):
+def _validate_claims(now, claims, audience=None, issuer=None, subject=None, algorithm=None, access_token=None, options=None):
 
     leeway = options.get("leeway", 0)
 
@@ -475,10 +479,10 @@ def _validate_claims(claims, audience=None, issuer=None, subject=None, algorithm
         _validate_iat(claims)
 
     if options.get("verify_nbf"):
-        _validate_nbf(claims, leeway=leeway)
+        _validate_nbf(now, claims, leeway=leeway)
 
     if options.get("verify_exp"):
-        _validate_exp(claims, leeway=leeway)
+        _validate_exp(now, claims, leeway=leeway)
 
     if options.get("verify_aud"):
         _validate_aud(claims, audience=audience)

tests/test_jwt.py

@@ -318,6 +318,16 @@ def test_exp_skip(self, key):
 
         jwt.decode(token, key, options=options)
 
+    def test_time_travel(self, key):
+
+        nbf = datetime(2015, 10, 21, 16, 29)
+        now = datetime(2015, 10, 21, 18, 0)
+        exp = datetime(2015, 10, 21, 19, 28)
+        claims = {"exp": exp, "nbf": nbf}
+        token = jwt.encode(claims, key)
+
+        jwt.decode(token, key, now=now)
+
     def test_aud_string(self, key):
 
         aud = "audience"