Merge pull request jpadilla#72 from wbolster/issue-72

Invalid validation of audience claim

Author: jpadilla

jwt/__init__.py

@@ -398,6 +398,9 @@ def verify_signature(payload, signing_input, header, signature, key='',
         else:
             leeway = leeway.total_seconds()
 
+    if not isinstance(audience, (basestring, type(None))):
+        raise TypeError('audience must be a string or None')
+
     try:
         algorithm = header['alg'].upper()
         key = prepare_key_methods[algorithm](key)
@@ -425,14 +428,20 @@ def verify_signature(payload, signing_input, header, signature, key='',
         if payload['exp'] < (utc_timestamp - leeway):
             raise ExpiredSignatureError('Signature has expired')
 
-    if audience is not None:
-        if isinstance(audience, list):
-            audiences = audience
-        else:
-            audiences = [audience]
-
-        if payload.get('aud') not in audiences:
+    if 'aud' in payload:
+        audience_claims = payload['aud']
+        if isinstance(audience_claims, basestring):
+            audience_claims = [audience_claims]
+        if not isinstance(audience_claims, list):
+            raise InvalidAudienceError('Invalid claim format in token')
+        if any(not isinstance(c, basestring) for c in audience_claims):
+            raise InvalidAudienceError('Invalid claim format in token')
+        if audience not in audience_claims:
             raise InvalidAudienceError('Invalid audience')
+    elif audience is not None:
+        # 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')
 
     if issuer is not None:
         if payload.get('iss') != issuer:

tests/test_jwt.py

@@ -701,93 +701,58 @@ def test_ecdsa_related_key_preparation_methods(self):
             self.assertFalse('ES512' in jwt.prepare_key_methods)
 
     def test_check_audience(self):
-        audience = 'urn:foo'
-
         payload = {
             'some': 'payload',
-            'aud': 'urn:foo'
+            'aud': 'urn:me'
         }
-
         token = jwt.encode(payload, 'secret')
-        decoded = jwt.decode(token, 'secret', audience=audience)
-
+        decoded = jwt.decode(token, 'secret', audience='urn:me')
         self.assertEqual(decoded, payload)
 
     def test_check_audience_in_array(self):
-        audience = ['urn:foo', 'urn:other']
-
         payload = {
             'some': 'payload',
-            'aud': 'urn:foo'
+            'aud': ['urn:me', 'urn:someone-else']
         }
-
         token = jwt.encode(payload, 'secret')
-        decoded = jwt.decode(token, 'secret', audience=audience)
-
+        decoded = jwt.decode(token, 'secret', audience='urn:me')
         self.assertEqual(decoded, payload)
 
     def test_raise_exception_invalid_audience(self):
-        audience = 'urn:wrong'
-
         payload = {
             'some': 'payload',
-            'aud': 'urn:foo'
+            'aud': 'urn:someone-else'
         }
-
         token = jwt.encode(payload, 'secret')
-
         self.assertRaises(
             jwt.InvalidAudienceError,
-            lambda: jwt.decode(token, 'secret', audience=audience))
+            lambda: jwt.decode(token, 'secret', audience='urn-me'))
 
     def test_raise_exception_invalid_audience_in_array(self):
-        audience = ['urn:wrong', 'urn:morewrong']
-
         payload = {
             'some': 'payload',
-            'aud': 'urn:foo'
+            'aud': ['urn:someone', 'urn:someone-else']
         }
-
         token = jwt.encode(payload, 'secret')
-
         self.assertRaises(
             jwt.InvalidAudienceError,
-            lambda: jwt.decode(token, 'secret', audience=audience))
+            lambda: jwt.decode(token, 'secret', audience='urn:me'))
 
     def test_raise_exception_token_without_audience(self):
-        audience = 'urn:wrong'
-
         payload = {
             'some': 'payload',
         }
-
         token = jwt.encode(payload, 'secret')
-
         self.assertRaises(
             jwt.InvalidAudienceError,
-            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.InvalidAudienceError,
-            lambda: jwt.decode(token, 'secret', audience=audience))
+            lambda: jwt.decode(token, 'secret', audience='urn:me'))
 
     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)