Merge pull request jpadilla#166 from mark-adams/opts-for-requiring-claims

Added new options for requiring exp, iat, and nbf claims.

Author: mark-adams

CHANGELOG.md

@@ -9,6 +9,13 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 ### Fixed
 - Exclude Python cache files from PyPI releases.
 
+### Added
+- Added new options to require certain claims
+  (require_nbf, require_iat, require_exp) and raise `MissingRequiredClaimError`
+  if they are not present.
+- If `audience=` or `issuer=` is specified but the claim is not present,
+  `MissingRequiredClaimError` is now raised instead of `InvalidAudienceError`
+  and `InvalidIssuerError`
 
 [v1.3][1.3.0]
 -------------------------------------------------------------------------

README.md

@@ -55,10 +55,12 @@ You can still get the payload by setting the `verify` argument to `False`.
 {u'some': u'payload'}
 ```
 
-The `decode()` function can raise other exceptions, e.g. for invalid issuer or
-audience (see below). All exceptions that signify that the token is invalid
-extend from the base `InvalidTokenError` exception class, so applications can
-use this approach to catch any issues relating to invalid tokens:
+## Validation
+Exceptions can be raised during `decode()` for other errors besides an
+invalid signature (e.g. for invalid issuer or audience (see below). All
+exceptions that signify that the token is invalid extend from the base
+`InvalidTokenError` exception class, so applications can use this approach to
+catch any issues relating to invalid tokens:
 
 ```python
 try:
@@ -67,8 +69,9 @@ except jwt.InvalidTokenError:
      pass  # do something sensible here, e.g. return HTTP 403 status code
 ```
 
-You may also override exception checking via an `options` dictionary.  The default
-options are as follows:
+### Skipping Claim Verification
+You may also override claim verification via the `options` dictionary.  The
+default options are:
 
 ```python
 options = {
@@ -77,24 +80,49 @@ options = {
    'verify_nbf': True,
    'verify_iat': True,
    'verify_aud': True
+   'require_exp': False,
+   'require_iat': False,
+   'require_nbf': False
 }
 ```
 
-You can skip individual checks by passing an `options` dictionary with certain keys set to `False`.
-For example, if you want to verify the signature of a JWT that has already expired.
+You can skip validation of individual claims by passing an `options` dictionary
+with the "verify_<claim_name>" key set to `False` when you call `jwt.decode()`.
+For example, if you want to verify the signature of a JWT that has already
+expired, you could do so by setting `verify_exp` to `False`.
 
 ```python
 >>> options = {
->>>    'verify_exp': True,
+>>>    'verify_exp': False,
 >>> }
 
+>>> encoded = '...' # JWT with an expired exp claim
 >>> jwt.decode(encoded, 'secret', options=options)
 {u'some': u'payload'}
 ```
 
-**NOTE**: *Changing the default behavior is done at your own risk, and almost certainly will make your
-application less secure.  Doing so should only be done with a very clear understanding of what you
-are doing.*
+**NOTE**: *Changing the default behavior is done at your own risk, and almost
+certainly will make your application less secure.  Doing so should only be done
+with a very clear understanding of what you are doing.*
+
+### Requiring Optional Claims
+In addition to skipping certain validations, you may also specify that certain
+optional claims are required by setting the appropriate `require_<claim_name>`
+option to True. If the claim is not present, PyJWT will raise a
+`jwt.exceptions.MissingRequiredClaimError`.
+
+For instance, the following code would require that the token has a 'exp'
+claim and raise an error if it is not present:
+
+```python
+>>> options = {
+>>>     'require_exp': True
+>>> }
+
+>>> encoded =  '...' # JWT without an exp claim
+>>> jwt.decode(encoded, 'secret', options=options)
+jwt.exceptions.MissingRequiredClaimError: Token is missing the "exp" claim
+```
 
 ## Tests
 

jwt/__init__.py

@@ -24,5 +24,6 @@
 from .exceptions import (
     InvalidTokenError, DecodeError, InvalidAudienceError,
     ExpiredSignatureError, ImmatureSignatureError, InvalidIssuedAtError,
-    InvalidIssuerError, ExpiredSignature, InvalidAudience, InvalidIssuer
+    InvalidIssuerError, ExpiredSignature, InvalidAudience, InvalidIssuer,
+    MissingRequiredClaimError
 )

jwt/api_jwt.py

@@ -11,7 +11,7 @@
 from .exceptions import (
     DecodeError, ExpiredSignatureError, ImmatureSignatureError,
     InvalidAudienceError, InvalidIssuedAtError,
-    InvalidIssuerError
+    InvalidIssuerError, MissingRequiredClaimError
 )
 from .utils import merge_dict
 
@@ -27,7 +27,10 @@ def _get_default_options():
             'verify_nbf': True,
             'verify_iat': True,
             'verify_aud': True,
-            'verify_iss': True
+            'verify_iss': True,
+            'require_exp': False,
+            'require_iat': False,
+            'require_nbf': False
         }
 
     def encode(self, payload, key, algorithm='HS256', headers=None,
@@ -87,6 +90,8 @@ def _validate_claims(self, payload, options, audience=None, issuer=None,
         if not isinstance(audience, (string_types, type(None))):
             raise TypeError('audience must be a string or None')
 
+        self._validate_required_claims(payload, options)
+
         now = timegm(datetime.utcnow().utctimetuple())
 
         if 'iat' in payload and options.get('verify_iat'):
@@ -104,6 +109,16 @@ def _validate_claims(self, payload, options, audience=None, issuer=None,
         if options.get('verify_aud'):
             self._validate_aud(payload, audience)
 
+    def _validate_required_claims(self, payload, options):
+        if options.get('require_exp') and payload.get('exp') is None:
+            raise MissingRequiredClaimError('exp')
+
+        if options.get('require_iat') and payload.get('iat') is None:
+            raise MissingRequiredClaimError('iat')
+
+        if options.get('require_nbf') and payload.get('nbf') is None:
+            raise MissingRequiredClaimError('nbf')
+
     def _validate_iat(self, payload, now, leeway):
         try:
             iat = int(payload['iat'])
@@ -140,7 +155,7 @@ def _validate_aud(self, payload, audience):
         if audience is not None and 'aud' not in payload:
             # Application specified an audience, but it could not be
             # verified since the token does not contain a claim.
-            raise InvalidAudienceError('No audience claim in token')
+            raise MissingRequiredClaimError('aud')
 
         audience_claims = payload['aud']
 
@@ -158,7 +173,7 @@ def _validate_iss(self, payload, issuer):
             return
 
         if 'iss' not in payload:
-            raise InvalidIssuerError('Token does not contain an iss claim')
+            raise MissingRequiredClaimError('iss')
 
         if payload['iss'] != issuer:
             raise InvalidIssuerError('Invalid issuer')

jwt/exceptions.py

@@ -34,6 +34,14 @@ class InvalidAlgorithmError(InvalidTokenError):
     pass
 
 
+class MissingRequiredClaimError(InvalidTokenError):
+    def __init__(self, claim):
+        self.claim = claim
+
+    def __str__(self):
+        return 'Token is missing the "%s" claim' % self.claim
+
+
 # Compatibility aliases (deprecated)
 ExpiredSignature = ExpiredSignatureError
 InvalidAudience = InvalidAudienceError

tests/test_api_jwt.py

@@ -9,7 +9,8 @@
 from jwt.api_jwt import PyJWT
 from jwt.exceptions import (
     DecodeError, ExpiredSignatureError, ImmatureSignatureError,
-    InvalidAudienceError, InvalidIssuedAtError, InvalidIssuerError
+    InvalidAudienceError, InvalidIssuedAtError, InvalidIssuerError,
+    MissingRequiredClaimError
 )
 
 import pytest
@@ -317,15 +318,31 @@ def test_raise_exception_invalid_audience_in_array(self, jwt):
         with pytest.raises(InvalidAudienceError):
             jwt.decode(token, 'secret', audience='urn:me')
 
+    def test_raise_exception_token_without_issuer(self, jwt):
+        issuer = 'urn:wrong'
+
+        payload = {
+            'some': 'payload'
+        }
+
+        token = jwt.encode(payload, 'secret')
+
+        with pytest.raises(MissingRequiredClaimError) as exc:
+            jwt.decode(token, 'secret', issuer=issuer)
+
+        assert exc.value.claim == 'iss'
+
     def test_raise_exception_token_without_audience(self, jwt):
         payload = {
             'some': 'payload',
         }
         token = jwt.encode(payload, 'secret')
 
-        with pytest.raises(InvalidAudienceError):
+        with pytest.raises(MissingRequiredClaimError) as exc:
             jwt.decode(token, 'secret', audience='urn:me')
 
+        assert exc.value.claim == 'aud'
+
     def test_check_issuer_when_valid(self, jwt):
         issuer = 'urn:foo'
         payload = {
@@ -348,33 +365,57 @@ def test_raise_exception_invalid_issuer(self, jwt):
         with pytest.raises(InvalidIssuerError):
             jwt.decode(token, 'secret', issuer=issuer)
 
-    def test_raise_exception_token_without_issuer(self, jwt):
-        issuer = 'urn:wrong'
+    def test_skip_check_audience(self, jwt):
+        payload = {
+            'some': 'payload',
+            'aud': 'urn:me',
+        }
+        token = jwt.encode(payload, 'secret')
+        jwt.decode(token, 'secret', options={'verify_aud': False})
 
+    def test_skip_check_exp(self, jwt):
         payload = {
             'some': 'payload',
+            'exp': datetime.utcnow() - timedelta(days=1)
         }
+        token = jwt.encode(payload, 'secret')
+        jwt.decode(token, 'secret', options={'verify_exp': False})
 
+    def test_decode_should_raise_error_if_exp_required_but_not_present(self, jwt):
+        payload = {
+            'some': 'payload',
+            # exp not present
+        }
         token = jwt.encode(payload, 'secret')
 
-        with pytest.raises(InvalidIssuerError):
-            jwt.decode(token, 'secret', issuer=issuer)
+        with pytest.raises(MissingRequiredClaimError) as exc:
+            jwt.decode(token, 'secret', options={'require_exp': True})
 
-    def test_skip_check_audience(self, jwt):
+        assert exc.value.claim == 'exp'
+
+    def test_decode_should_raise_error_if_iat_required_but_not_present(self, jwt):
         payload = {
             'some': 'payload',
-            'aud': 'urn:me',
+            # iat not present
         }
         token = jwt.encode(payload, 'secret')
-        jwt.decode(token, 'secret', options={'verify_aud': False})
 
-    def test_skip_check_exp(self, jwt):
+        with pytest.raises(MissingRequiredClaimError) as exc:
+            jwt.decode(token, 'secret', options={'require_iat': True})
+
+        assert exc.value.claim == 'iat'
+
+    def test_decode_should_raise_error_if_nbf_required_but_not_present(self, jwt):
         payload = {
             'some': 'payload',
-            'exp': datetime.utcnow() - timedelta(days=1)
+            # nbf not present
         }
         token = jwt.encode(payload, 'secret')
-        jwt.decode(token, 'secret', options={'verify_exp': False})
+
+        with pytest.raises(MissingRequiredClaimError) as exc:
+            jwt.decode(token, 'secret', options={'require_nbf': True})
+
+        assert exc.value.claim == 'nbf'
 
     def test_skip_check_signature(self, jwt):
         token = ("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"

tests/test_exceptions.py

@@ -0,0 +1,7 @@
+from jwt.exceptions import MissingRequiredClaimError
+
+
+def test_missing_required_claim_error_has_proper_str():
+    exc = MissingRequiredClaimError('abc')
+
+    assert str(exc) == 'Token is missing the "abc" claim'