Implement Audience and Issuer claims

Author: jpadilla

README.md

@@ -94,6 +94,8 @@ used. PyJWT supports these reserved claim names:
 
  - "exp" (Expiration Time) Claim
  - "nbf" (Not Before Time) Claim
+ - "iss" (Issuer) Claim
+ - "aud" (Audience) Claim
 
 ### Expiration Time Claim
 
@@ -156,7 +158,51 @@ time.sleep(32)
 jwt.decode(jwt_payload, 'secret', leeway=10)
 ```
 
-PyJWT also supports not-before validation via the [`nbf` claim](https://tools.ietf.org/html/draft-ietf-oauth-json-web-token-27#section-4.1.5) in a similar fashion.
+### Not Before Time Claim
+
+> The nbf (not before) claim identifies the time before which the JWT MUST NOT be accepted for processing. The processing of the nbf claim requires that the current date/time MUST be after or equal to the not-before date/time listed in the nbf claim. Implementers MAY provide for some small leeway, usually no more than a few minutes, to account for clock skew. Its value MUST be a number containing a NumericDate value. Use of this claim is OPTIONAL.
+
+The `nbf` claim works similarly to the `exp` claim above.
+
+```python
+jwt.encode({'nbf': 1371720939}, 'secret')
+
+jwt.encode({'nbf': datetime.utcnow()}, 'secret')
+```
+
+### Issuer Claim
+
+> The iss (issuer) claim identifies the principal that issued the JWT. The processing of this claim is generally application specific. The iss value is a case-sensitive string containing a StringOrURI value. Use of this claim is OPTIONAL.
+
+```python
+import jwt
+
+
+payload = {
+    'some': 'payload',
+    'iss': 'urn:foo'
+}
+
+token = jwt.encode(payload, 'secret')
+decoded = jwt.decode(token, 'secret', issuer='urn:foo')
+```
+
+### Audience Claim
+
+> The aud (audience) claim identifies the recipients that the JWT is intended for. Each principal intended to process the JWT MUST identify itself with a value in the audience claim. If the principal processing the claim does not identify itself with a value in the aud claim when this claim is present, then the JWT MUST be rejected. In the general case, the aud value is an array of case-sensitive strings, each containing a StringOrURI value. In the special case when the JWT has one audience, the aud value MAY be a single case-sensitive string containing a StringOrURI value. The interpretation of audience values is generally application specific. Use of this claim is OPTIONAL.
+
+```python
+import jwt
+
+
+payload = {
+    'some': 'payload',
+    'aud': 'urn:foo'
+}
+
+token = jwt.encode(payload, 'secret')
+decoded = jwt.decode(token, 'secret', audience='urn:foo')
+```
 
 ## License
 

jwt/__init__.py

@@ -37,6 +37,14 @@ class ExpiredSignature(Exception):
     pass
 
 
+class InvalidAudience(Exception):
+    pass
+
+
+class InvalidIssuer(Exception):
+    pass
+
+
 signing_methods = {
     'none': lambda msg, key: b'',
     'HS256': lambda msg, key: hmac.new(key, msg, hashlib.sha256).digest(),
@@ -246,12 +254,15 @@ def encode(payload, key, algorithm='HS256', headers=None):
     return b'.'.join(segments)
 
 
-def decode(jwt, key='', verify=True, verify_expiration=True, leeway=0):
+def decode(jwt, key='', verify=True, **kwargs):
     payload, signing_input, header, signature = load(jwt)
 
     if verify:
+        verify_expiration = kwargs.pop('verify_expiration', True)
+        leeway = kwargs.pop('leeway', 0)
+
         verify_signature(payload, signing_input, header, signature, key,
-                         verify_expiration, leeway)
+                         verify_expiration, leeway, **kwargs)
 
     return payload
 
@@ -292,7 +303,7 @@ def load(jwt):
 
 
 def verify_signature(payload, signing_input, header, signature, key='',
-                     verify_expiration=True, leeway=0):
+                     verify_expiration=True, leeway=0, **kwargs):
     try:
         algorithm = header['alg'].upper()
         key = prepare_key_methods[algorithm](key)
@@ -310,6 +321,7 @@ def verify_signature(payload, signing_input, header, signature, key='',
 
     if 'nbf' in payload and verify_expiration:
         utc_timestamp = timegm(datetime.utcnow().utctimetuple())
+
         if payload['nbf'] > (utc_timestamp + leeway):
             raise ExpiredSignature('Signature not yet valid')
 
@@ -318,3 +330,20 @@ def verify_signature(payload, signing_input, header, signature, key='',
 
         if payload['exp'] < (utc_timestamp - leeway):
             raise ExpiredSignature('Signature has expired')
+
+    audience = kwargs.get('audience')
+
+    if audience:
+        if isinstance(audience, list):
+            audiences = audience
+        else:
+            audiences = [audience]
+
+        if payload.get('aud') not in audiences:
+            raise InvalidAudience('Invalid audience')
+
+    issuer = kwargs.get('issuer')
+
+    if issuer:
+        if payload.get('iss') != issuer:
+            raise InvalidIssuer('Invalid issuer')

tests/test_jwt.py

@@ -691,6 +691,126 @@ def test_ecdsa_related_key_preparation_methods(self):
             self.assertFalse('ES384' in jwt.prepare_key_methods)
             self.assertFalse('ES512' in jwt.prepare_key_methods)
 
+    def test_check_audience(self):
+        audience = 'urn:foo'
+
+        payload = {
+            'some': 'payload',
+            'aud': 'urn:foo'
+        }
+
+        token = jwt.encode(payload, 'secret')
+        decoded = jwt.decode(token, 'secret', audience=audience)
+
+        self.assertEqual(decoded, payload)
+
+    def test_check_audience_in_array(self):
+        audience = ['urn:foo', 'urn:other']
+
+        payload = {
+            'some': 'payload',
+            'aud': 'urn:foo'
+        }
+
+        token = jwt.encode(payload, 'secret')
+        decoded = jwt.decode(token, 'secret', audience=audience)
+
+        self.assertEqual(decoded, payload)
+
+    def test_raise_exception_invalid_audience(self):
+        audience = 'urn:wrong'
+
+        payload = {
+            'some': 'payload',
+            'aud': 'urn:foo'
+        }
+
+        token = jwt.encode(payload, 'secret')
+
+        self.assertRaises(
+            jwt.InvalidAudience,
+            lambda: jwt.decode(token, 'secret', audience=audience))
+
+    def test_raise_exception_invalid_audience_in_array(self):
+        audience = ['urn:wrong', 'urn:morewrong']
+
+        payload = {
+            'some': 'payload',
+            'aud': 'urn:foo'
+        }
+
+        token = jwt.encode(payload, 'secret')
+
+        self.assertRaises(
+            jwt.InvalidAudience,
+            lambda: jwt.decode(token, 'secret', audience=audience))
+
+    def test_raise_exception_token_without_audience(self):
+        audience = 'urn:wrong'
+
+        payload = {
+            'some': 'payload',
+        }
+
+        token = jwt.encode(payload, 'secret')
+
+        self.assertRaises(
+            jwt.InvalidAudience,
+            lambda: jwt.decode(token, 'secret', audience=audience))
+
+    def test_raise_exception_token_without_audience_in_array(self):
+        audience = ['urn:wrong', 'urn:morewrong']
+
+        payload = {
+            'some': 'payload',
+        }
+
+        token = jwt.encode(payload, 'secret')
+
+        self.assertRaises(
+            jwt.InvalidAudience,
+            lambda: jwt.decode(token, 'secret', audience=audience))
+
+    def test_check_issuer(self):
+        issuer = 'urn:foo'
+
+        payload = {
+            'some': 'payload',
+            'iss': 'urn:foo'
+        }
+
+        token = jwt.encode(payload, 'secret')
+        decoded = jwt.decode(token, 'secret', issuer=issuer)
+
+        self.assertEqual(decoded, payload)
+
+    def test_raise_exception_invalid_issuer(self):
+        issuer = 'urn:wrong'
+
+        payload = {
+            'some': 'payload',
+            'iss': 'urn:foo'
+        }
+
+        token = jwt.encode(payload, 'secret')
+
+        self.assertRaises(
+            jwt.InvalidIssuer,
+            lambda: jwt.decode(token, 'secret', issuer=issuer))
+
+    def test_raise_exception_token_without_issuer(self):
+        issuer = 'urn:wrong'
+
+        payload = {
+            'some': 'payload',
+        }
+
+        token = jwt.encode(payload, 'secret')
+
+        self.assertRaises(
+            jwt.InvalidIssuer,
+            lambda: jwt.decode(token, 'secret', issuer=issuer))
+
 
 if __name__ == '__main__':
     unittest.main()