Merge pull request #51 from jschlyter/unknown_issuer

Better exception handling for missing keys and issuers

Author: jschlyter

src/cryptojwt/exception.py

@@ -40,7 +40,15 @@ class BadType(Invalid):
 
 
 class MissingKey(JWKESTException):
-    """ No usable key """
+    """No usable key"""
+
+
+class KeyNotFound(KeyError):
+    """Key not found"""
+
+
+class IssuerNotFound(KeyError):
+    """Issuer not found"""
 
 
 class KeyIOError(Exception):

src/cryptojwt/key_jar.py

@@ -5,6 +5,7 @@
 
 from requests import request
 
+from .exception import UnknownKeyType, KeyIOError, UpdateFailed, IssuerNotFound
 from .jwe.jwe import alg2keytype as jwe_alg2keytype
 from .jws.utils import alg2keytype as jws_alg2keytype
 from .key_bundle import KeyBundle
@@ -20,18 +21,6 @@
 logger = logging.getLogger(__name__)
 
 
-class KeyIOError(Exception):
-    pass
-
-
-class UnknownKeyType(KeyIOError):
-    pass
-
-
-class UpdateFailed(KeyIOError):
-    pass
-
-
 class KeyJar(object):
     """ A keyjar contains a number of KeyBundles sorted by owner/issuer """
 
@@ -252,7 +241,7 @@ def get_issuer_keys(self, issuer_id):
         """
         _issuer = self._get_issuer(issuer_id)
         if _issuer is None:
-            raise KeyError(issuer_id)
+            raise IssuerNotFound(issuer_id)
         return _issuer.all_keys()
 
     @deprecated_alias(issuer='issuer_id', owner='issuer_id')
@@ -273,7 +262,7 @@ def __getitem__(self, issuer_id=''):
         """
         _issuer = self._get_issuer(issuer_id)
         if _issuer is None:
-            raise KeyError(issuer_id)
+            raise IssuerNotFound(issuer_id)
         return _issuer
 
     @deprecated_alias(issuer='issuer_id', owner='issuer_id')
@@ -478,7 +467,7 @@ def _add_key(self, keys, issuer_id, use, key_type='', kid='',
         _issuer = self._get_issuer(issuer_id)
         if _issuer is None:
             logger.error('Issuer "{}" not in keyjar'.format(issuer_id))
-            return keys
+            raise IssuerNotFound(issuer_id)
 
         logger.debug('Key summary for {}: {}'.format(issuer_id, _issuer.key_summary()))
 
@@ -678,7 +667,7 @@ def key_summary(self, issuer_id):
         if _issuer is not None:
             return _issuer.key_summary()
 
-        raise KeyError('Unknown Issuer ID: "{}"'.format(issuer_id))
+        raise IssuerNotFound(issuer_id)
 
     def update(self):
         """

tests/test_04_key_jar.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-from cryptojwt.exception import JWKESTException
+from cryptojwt.exception import JWKESTException, IssuerNotFound
 from cryptojwt.jwe.jwenc import JWEnc
 from cryptojwt.jws.jws import JWS
 from cryptojwt.jws.jws import factory
@@ -799,8 +799,8 @@ def test_get_decrypt_keys():
     keys = kj.get_jwt_decrypt_keys(jwt)
     assert keys
 
-    keys = kj.get_jwt_decrypt_keys(jwt, aud='Bob')
-    assert keys
+    with pytest.raises(IssuerNotFound):
+        keys = kj.get_jwt_decrypt_keys(jwt, aud='Bob')
 
 
 def test_update_keyjar():

tests/test_09_jwt.py

@@ -1,5 +1,9 @@
 import os
 
+import pytest
+
+from cryptojwt.exception import JWKESTException, IssuerNotFound
+from cryptojwt.jws.exception import NoSuitableSigningKeys
 from cryptojwt.jwt import JWT
 from cryptojwt.jwt import pick_key
 from cryptojwt.key_bundle import KeyBundle
@@ -64,6 +68,29 @@ def test_jwt_pack_and_unpack():
     assert set(info.keys()) == {'iat', 'iss', 'sub'}
 
 
+def test_jwt_pack_and_unpack_unknown_issuer():
+    alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, sign_alg='RS256')
+    payload = {'sub': 'sub'}
+    _jwt = alice.pack(payload=payload)
+
+    kj = KeyJar()
+    bob = JWT(key_jar=kj, iss=BOB, allowed_sign_algs=["RS256"])
+    with pytest.raises(IssuerNotFound):
+        info = bob.unpack(_jwt)
+
+
+def test_jwt_pack_and_unpack_unknown_key():
+    alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, sign_alg='RS256')
+    payload = {'sub': 'sub'}
+    _jwt = alice.pack(payload=payload)
+
+    kj = KeyJar()
+    kj.add_kb(ALICE, KeyBundle())
+    bob = JWT(key_jar=kj, iss=BOB, allowed_sign_algs=["RS256"])
+    with pytest.raises(NoSuitableSigningKeys):
+        info = bob.unpack(_jwt)
+
+
 def test_jwt_pack_and_unpack_with_lifetime():
     alice = JWT(key_jar=ALICE_KEY_JAR, iss=ALICE, lifetime=600)
     payload = {'sub': 'sub'}