check nbf/exp and add allowed_max_lifetime

Author: jschlyter

src/cryptojwt/jwt.py

@@ -1,6 +1,7 @@
 """Basic JSON Web Token implementation."""
 import json
 import logging
+import time
 import uuid
 from datetime import datetime
 from datetime import timezone
@@ -95,6 +96,7 @@ def __init__(
         allowed_sign_algs=None,
         allowed_enc_algs=None,
         allowed_enc_encs=None,
+        allowed_max_lifetime=None,
         zip="",
     ):
         self.key_jar = key_jar  # KeyJar instance
@@ -115,6 +117,7 @@ def __init__(
         self.allowed_sign_algs = allowed_sign_algs
         self.allowed_enc_algs = allowed_enc_algs
         self.allowed_enc_encs = allowed_enc_encs
+        self.allowed_max_lifetime = allowed_max_lifetime
         self.zip = zip
 
     def receiver_keys(self, recv, use):
@@ -304,11 +307,12 @@ def verify_profile(msg_cls, info, **kwargs):
             raise VerificationError()
         return _msg
 
-    def unpack(self, token):
+    def unpack(self, token, timestamp=None):
         """
         Unpack a received signed or signed and encrypted Json Web Token
 
         :param token: The Json Web Token
+        :param t: Time for evaluation (default now)
         :return: If decryption and signature verification work the payload
             will be returned as a Message instance if possible.
         """
@@ -378,6 +382,26 @@ def unpack(self, token):
             except KeyError:
                 _msg_cls = None
 
+        timestamp = timestamp or time.time()
+
+        if "nbf" in _info:
+            nbf = int(_info["nbf"])
+            if timestamp < nbf - self.skew:
+                raise VerificationError("Token not yet valid")
+
+        if "exp" in _info:
+            exp = int(_info["exp"])
+            if timestamp >= exp + self.skew:
+                raise VerificationError("Token expired")
+        else:
+            exp = None
+
+        if "iat" in _info:
+            iat = int(_info["iat"])
+            if self.allowed_max_lifetime and exp:
+                if abs(exp - iat) > self.allowed_max_lifetime:
+                    raise VerificationError("Token lifetime exceeded")
+
         if _msg_cls:
             vp_args = {"skew": self.skew}
             if self.iss:

tests/test_09_jwt.py

@@ -5,6 +5,7 @@
 from cryptojwt.exception import IssuerNotFound
 from cryptojwt.jws.exception import NoSuitableSigningKeys
 from cryptojwt.jwt import JWT
+from cryptojwt.jwt import VerificationError, utc_time_sans_frac
 from cryptojwt.jwt import pick_key
 from cryptojwt.key_bundle import KeyBundle
 from cryptojwt.key_jar import KeyJar
@@ -81,6 +82,59 @@ def test_jwt_pack_and_unpack():
     assert set(info.keys()) == {"iat", "iss", "sub"}
 
 
+def test_jwt_pack_and_unpack_valid():
+    alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, sign_alg="RS256")
+    t = utc_time_sans_frac()
+    payload = {"sub": "sub", "nbf": t, "exp": t + 3600}
+    _jwt = alice.pack(payload=payload)
+
+    bob = JWT(key_jar=BOB_KEY_JAR, iss=BOB, allowed_sign_algs=["RS256"])
+    info = bob.unpack(_jwt)
+
+    assert set(info.keys()) == {"iat", "iss", "sub", "nbf", "exp"}
+
+
+def test_jwt_pack_and_unpack_not_yet_valid():
+    lifetime = 3600
+    skew = 15
+    alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, sign_alg="RS256", lifetime=lifetime)
+    timestamp = utc_time_sans_frac()
+    payload = {"sub": "sub", "nbf": timestamp}
+    _jwt = alice.pack(payload=payload)
+
+    bob = JWT(key_jar=BOB_KEY_JAR, iss=BOB, allowed_sign_algs=["RS256"], skew=skew)
+    _ = bob.unpack(_jwt, timestamp=timestamp - skew)
+    with pytest.raises(VerificationError):
+        _ = bob.unpack(_jwt, timestamp=timestamp - skew - 1)
+
+
+def test_jwt_pack_and_unpack_expired():
+    lifetime = 3600
+    skew = 15
+    alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, sign_alg="RS256", lifetime=lifetime)
+    payload = {"sub": "sub"}
+    _jwt = alice.pack(payload=payload)
+
+    bob = JWT(key_jar=BOB_KEY_JAR, iss=BOB, allowed_sign_algs=["RS256"], skew=skew)
+    iat = bob.unpack(_jwt)["iat"]
+    _ = bob.unpack(_jwt, timestamp=iat + lifetime + skew - 1)
+    with pytest.raises(VerificationError):
+        _ = bob.unpack(_jwt, timestamp=iat + lifetime + skew)
+
+
+def test_jwt_pack_and_unpack_max_lifetime_exceeded():
+    lifetime = 3600
+    alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, sign_alg="RS256", lifetime=lifetime)
+    payload = {"sub": "sub"}
+    _jwt = alice.pack(payload=payload)
+
+    bob = JWT(
+        key_jar=BOB_KEY_JAR, iss=BOB, allowed_sign_algs=["RS256"], allowed_max_lifetime=lifetime - 1
+    )
+    with pytest.raises(VerificationError):
+        _ = bob.unpack(_jwt)
+
+
 def test_jwt_pack_and_unpack_unknown_issuer():
     alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, sign_alg="RS256")
     payload = {"sub": "sub"}
@@ -261,4 +315,4 @@ def test_eddsa_jwt():
     kj = KeyJar()
     kj.add_kb(ISSUER, KeyBundle(JWKS_DICT))
     jwt = JWT(key_jar=kj)
-    _ = jwt.unpack(JWT_TEST)
+    _ = jwt.unpack(JWT_TEST, timestamp=1655278809)