check presence among claims

jose/jwt.py

@@ -97,6 +97,14 @@ def decode(token, key, algorithms=None, options=None, audience=None,
                 'verify_sub': True,
                 'verify_jti': True,
                 'verify_at_hash': True,
+                'require_aud': False,
+                'require_iat': False,
+                'require_exp': False,
+                'require_nbf': False,
+                'require_iss': False,
+                'require_sub': False,
+                'require_jti': False,
+                'require_at_hash': False,
                 'leeway': 0,
             }
 
@@ -126,6 +134,14 @@ def decode(token, key, algorithms=None, options=None, audience=None,
         'verify_sub': True,
         'verify_jti': True,
         'verify_at_hash': True,
+        'require_aud': False,
+        'require_iat': False,
+        'require_exp': False,
+        'require_nbf': False,
+        'require_iss': False,
+        'require_sub': False,
+        'require_jti': False,
+        'require_at_hash': False,
         'leeway': 0,
     }
 
@@ -455,6 +471,14 @@ def _validate_claims(claims, audience=None, issuer=None, subject=None,
     if isinstance(leeway, timedelta):
         leeway = timedelta_total_seconds(leeway)
 
+    for require_claim in [
+        e[len("require_"):] for e in options.keys() if e.startswith("require_") and options[e]
+    ]:
+        if require_claim not in claims:
+            raise JWTError('missing required key "%s" among claims' % require_claim)
+        else:
+            options['verify_' + require_claim] = True  # override verify when required
+
     if not isinstance(audience, (string_types, type(None))):
         raise JWTError('audience must be a string or None')
 

tests/test_jwt.py

@@ -579,3 +579,26 @@ def test_unverified_claims_list(self):
     def test_unverified_claims_object(self, claims, key):
         token = jwt.encode(claims, key)
         assert jwt.get_unverified_claims(token) == claims
+
+    @pytest.mark.parametrize(
+        "claim,value", [
+            ("aud", "aud"),
+            ("ait", "ait"),
+            ("exp", datetime.utcnow() + timedelta(seconds=5)),
+            ("nbf", datetime.utcnow() - timedelta(seconds=5)),
+            ("iss", "iss"),
+            ("sub", "sub"),
+            ("jti", "jti"),
+        ]
+    )
+    def test_require(self, claims, key, claim, value):
+        options = {"require_" + claim: True, "verify_" + claim: False}
+
+        token = jwt.encode(claims, key)
+        with pytest.raises(JWTError):
+            jwt.decode(token, key, options=options, audience=str(value))
+
+        new_claims = dict(claims)
+        new_claims[claim] = value
+        token = jwt.encode(new_claims, key)
+        jwt.decode(token, key, options=options, audience=str(value))