Working on the tests

Author: rohe

src/cryptojwt/aes_key_wrap.py

@@ -145,20 +145,16 @@ def load_x509_cert(url, spec2key):
         return []
 
 
-def rsa_load(filename, passphrase):
-    """Read a PEM-encoded RSA key pair from a file."""
+def rsa_load(filename, passphrase=None):
+    """Read a PEM-encoded RSA private key from a file."""
     with open(filename, "rb") as key_file:
         pem_data = key_file.read()
         private_key = serialization.load_pem_private_key(
             pem_data,
             password=passphrase,
             backend=default_backend())
 
-        public_key = serialization.load_pem_public_key(
-            pem_data,
-            backend=default_backend()
-        )
-    return private_key, public_key
+    return private_key
 
 
 def rsa_eq(key1, key2):

src/cryptojwt/jwk.py

@@ -6,7 +6,6 @@
 from cryptojwt import jwe
 from cryptojwt import jws
 from cryptojwt.jwe import JWE
-from cryptojwt.jws import alg2keytype
 from cryptojwt.jws import JWS
 from cryptojwt.jws import NoSuitableSigningKeys
 
@@ -20,6 +19,15 @@ def utc_time_sans_frac():
 
 
 def pick_key(keys, use, alg='', key_type='', kid=''):
+    """
+
+    :param keys: List of keys
+    :param use: What the key is going to be used for
+    :param alg: crypto algorithm
+    :param key_type: Type of key
+    :param kid: Ley ID
+    :return: list of keys that match the pattern
+    """
     res = []
     if not key_type:
         if use == 'sig':
@@ -51,14 +59,17 @@ def get_jwt_keys(jwt, keys, use):
     except KeyError:
         _kid = ''
 
+    # Pick issuers keys
+
     return pick_key(keys, use, key_type=_key_type, kid=_kid)
 
 
 class JWT(object):
-    def __init__(self, keys, iss='', lifetime=0, sign_alg='RS256',
-                 encrypt=False, enc_enc="A128CBC-HS256",
+    def __init__(self, own_keys, iss='', rec_keys=None, lifetime=0,
+                 sign_alg='RS256', encrypt=False, enc_enc="A128CBC-HS256",
                  enc_alg="RSA1_5"):
-        self.keys = keys
+        self.own_keys = own_keys
+        self.rec_keys = rec_keys or {}
         self.iss = iss
         self.lifetime = lifetime
         self.sign_alg = sign_alg
@@ -67,15 +78,15 @@ def __init__(self, keys, iss='', lifetime=0, sign_alg='RS256',
         self.enc_enc = enc_enc
         self.with_jti = False
 
-    def _encrypt(self, payload, cty='JWT'):
+    def _encrypt(self, payload, recv, cty='JWT'):
         kwargs = {"alg": self.enc_alg, "enc": self.enc_enc}
 
         if cty:
             kwargs["cty"] = cty
 
         # use the clients public key for encryption
         _jwe = JWE(payload, **kwargs)
-        return _jwe.encrypt(self.keys, context="public")
+        return _jwe.encrypt(self.rec_keys[recv], context="public")
 
     def pack_init(self):
         """
@@ -96,21 +107,20 @@ def pack_key(self, owner='', kid=''):
         :param kid: Key ID
         :return: One key
         """
-        keys = pick_key(self.keys, 'sig', alg=self.sign_alg, kid=kid)
+        keys = pick_key(self.own_keys, 'sig', alg=self.sign_alg, kid=kid)
 
         if not keys:
             raise NoSuitableSigningKeys('kid={}'.format(kid))
 
         return keys[0]  # Might be more then one if kid == ''
 
-    def pack(self, payload=None, kid='', owner='', cls_instance=None, **kwargs):
+    def pack(self, payload=None, kid='', owner='', recv='', **kwargs):
         """
 
         :param payload: Information to be carried as payload in the JWT
         :param kid: Key ID
         :param owner: The owner of the the keys that are to be used for signing
-        :param cls_instance: This might be a instance of a class already
-            prepared with information
+        :param recv: The intended receiver
         :param kwargs: Extra keyword arguments
         :return: A signed or signed and encrypted JsonWebtoken
         """
@@ -144,12 +154,12 @@ def pack(self, payload=None, kid='', owner='', cls_instance=None, **kwargs):
         _sjwt = _jws.sign_compact([_key])
         #_jws = _jwt.to_jwt([_key], self.sign_alg)
         if _encrypt:
-            return self._encrypt(_sjwt)
+            return self._encrypt(_sjwt, recv)
         else:
             return _sjwt
 
     def _verify(self, rj, token):
-        keys = get_jwt_keys(rj.jwt, self.keys, 'sig')
+        keys = get_jwt_keys(rj.jwt, self.rec_keys, 'sig')
         return rj.verify_compact(token, keys)
 
     def _decrypt(self, rj, token):
@@ -160,7 +170,7 @@ def _decrypt(self, rj, token):
         :param token: The encrypted JsonWebToken
         :return: 
         """
-        keys = get_jwt_keys(rj.jwt, self.keys, 'enc')
+        keys = get_jwt_keys(rj.jwt, self.own_keys, 'enc')
         return rj.decrypt(token, keys=keys)
 
     def unpack(self, token):

src/cryptojwt/jwt.py

@@ -1,7 +1,7 @@
 import pytest
 
-from cryptojwt.jwt import bytes2str_conv
-from cryptojwt.jwt import JWT
+from cryptojwt import bytes2str_conv
+from cryptojwt import SimpleJWT
 
 __author__ = 'roland'
 
@@ -11,7 +11,7 @@ def _eq(l1, l2):
 
 
 def test_pack_jwt():
-    _jwt = JWT(**{"alg": "none", "cty": "jwt"})
+    _jwt = SimpleJWT(**{"alg": "none", "cty": "jwt"})
     jwt = _jwt.pack(parts=[{"iss": "joe", "exp": 1300819380,
                             "http://example.com/is_root": True}, ""])
 
@@ -20,22 +20,22 @@ def test_pack_jwt():
 
 
 def test_unpack_pack():
-    _jwt = JWT(**{"alg": "none"})
+    _jwt = SimpleJWT(**{"alg": "none"})
     payload = {"iss": "joe", "exp": 1300819380,
                "http://example.com/is_root": True}
     jwt = _jwt.pack(parts=[payload, ""])
-    repacked = JWT().unpack(jwt).pack()
+    repacked = SimpleJWT().unpack(jwt).pack()
 
     assert jwt == repacked
 
 
 def test_pack_unpack():
-    _jwt = JWT(**{"alg": "none"})
+    _jwt = SimpleJWT(**{"alg": "none"})
     payload = {"iss": "joe", "exp": 1300819380,
                "http://example.com/is_root": True}
     jwt = _jwt.pack(parts=[payload, ""])
 
-    _jwt2 = JWT().unpack(jwt)
+    _jwt2 = SimpleJWT().unpack(jwt)
 
     assert _jwt2
     out_payload = _jwt2.payload()
@@ -47,18 +47,18 @@ def test_pack_unpack():
 
 
 def test_pack_with_headers():
-    _jwt = JWT()
+    _jwt = SimpleJWT()
     jwt = _jwt.pack(parts=["", ""], headers={"foo": "bar"})
-    assert JWT().unpack(jwt).headers["foo"] == "bar"
+    assert SimpleJWT().unpack(jwt).headers["foo"] == "bar"
 
 
 def test_unpack_str():
-    _jwt = JWT(**{"alg": "none"})
+    _jwt = SimpleJWT(**{"alg": "none"})
     payload = {"iss": "joe", "exp": 1300819380,
                "http://example.com/is_root": True}
     jwt = _jwt.pack(parts=[payload, ""])
 
-    _jwt2 = JWT().unpack(jwt)
+    _jwt2 = SimpleJWT().unpack(jwt)
     assert _jwt2
     out_payload = _jwt2.payload()
 

tests/test_1_jwt.py renamed to tests/test_0_simplejwt.py

@@ -372,28 +372,22 @@ def test_thumbprint():
     keyl.load_dict(JWKS)
     for key in keyl:
         txt = key.thumbprint('SHA-256')
-        assert b64e(txt) in EXPECTED
+        assert txt in EXPECTED
 
 
 def test_thumbprint_7638_example():
     key = RSAKey(
         n='0vx7agoebGcQSuuPiLJXZptN9nndrQmbXEps2aiAFbWhM78LhWx4cbbfAAtVT86zwu1RK7aPFFxuhDR1L6tSoc_BJECPebWKRXjBZCiFV4n3oknjhMstn64tZ_2W-5JsGY4Hc5n9yBXArwl93lqt7_RN5w6Cf0h4QyQ5v-65YGjQR0_FDW2QvzqY368QQMicAtaSqzs8KJZgnYb9c7d0zgdAZHzu6qMQvRL5hajrn1n91CbOpbISD08qNLyrdkt-bFTWhAI4vMQFh6WeZu0fM4lFd2NcRwr3XPksINHaQ-G_xBniIqbw0Ls1jF44-csFCur-kEgU8awapJzKnqDKgw',
         e='AQAB', alg='RS256', kid='2011-04-29')
     thumbprint = key.thumbprint('SHA-256')
-    assert b64e(thumbprint) == b'NzbLsXh8uDCcd-6MNwXF4W_7noWXFZAfHkxZsRGC9Xs'
+    assert thumbprint == b'NzbLsXh8uDCcd-6MNwXF4W_7noWXFZAfHkxZsRGC9Xs'
 
 
 def test_load_jwks():
     keysl = load_jwks(json.dumps(JWKS))
     assert len(keysl) == 3
 
 
-# def test_copy():
-#     keysl = load_jwks(json.dumps(JWKS))
-#     ckl = [copy.copy(k) for k in keysl]
-#     assert len(ckl) == 3
-
-
 def test_encryption_key():
     sk = SYMKey(key='df34db91c16613deba460752522d28f6ebc8a73d0d9185836270c26b')
     _enc = sk.encryption_key(alg='A128KW')