Merge pull request #61 from IdentityPython/rsa_import

Work around bad RSAKey import

Author: jschlyter

src/cryptojwt/jwe/__init__.py

@@ -22,7 +22,14 @@
         "ECDH-ES+A192KW",
         "ECDH-ES+A256KW",
     ],
-    "enc": ["A128CBC-HS256", "A192CBC-HS384", "A256CBC-HS512", "A128GCM", "A192GCM", "A256GCM",],
+    "enc": [
+        "A128CBC-HS256",
+        "A192CBC-HS384",
+        "A256CBC-HS512",
+        "A128GCM",
+        "A192GCM",
+        "A256GCM",
+    ],
 }
 
 

src/cryptojwt/jwe/aes.py

@@ -18,8 +18,7 @@
 
 
 class AES_CBCEncrypter(Encrypter):
-    """
-    """
+    """"""
 
     def __init__(self, key_len=32, key=None, msg_padding="PKCS7"):
         Encrypter.__init__(self)

src/cryptojwt/jwe/jwe_ec.py

@@ -157,7 +157,12 @@ def dec_setup(self, token, key=None, **kwargs):
                 raise Exception("Unknown key length for algorithm")
 
             self.cek = ecdh_derive_key(
-                key, epubkey.pub_key, apu, apv, str(self.headers["enc"]).encode(), dk_len,
+                key,
+                epubkey.pub_key,
+                apu,
+                apv,
+                str(self.headers["enc"]).encode(),
+                dk_len,
             )
         elif self.headers["alg"] in [
             "ECDH-ES+A128KW",

src/cryptojwt/jwe/jwe_rsa.py

@@ -85,7 +85,7 @@ def encrypt(self, key, iv="", cek="", **kwargs):
         return jwe.pack(parts=[jwe_enc_key, iv, ctxt, tag])
 
     def decrypt(self, token, key, cek=None):
-        """ Decrypts a JWT
+        """Decrypts a JWT
 
         :param token: The JWT
         :param key: A key to use for decrypting

src/cryptojwt/jwe/jwekey.py

@@ -38,7 +38,7 @@ def alg2keytype(self, alg):
         return alg2keytype(alg)
 
     def enc_setup(self, enc_alg, msg, auth_data=b"", key=None, iv=""):
-        """ Encrypt JWE content.
+        """Encrypt JWE content.
 
         :param enc_alg: The JWE "enc" value specifying the encryption algorithm
         :param msg: The plain text message
@@ -62,7 +62,7 @@ def enc_setup(self, enc_alg, msg, auth_data=b"", key=None, iv=""):
 
     @staticmethod
     def _decrypt(enc, key, ctxt, iv, tag, auth_data=b""):
-        """ Decrypt JWE content.
+        """Decrypt JWE content.
 
         :param enc: The JWE "enc" value specifying the encryption algorithm
         :param key: Key (CEK)

src/cryptojwt/jwe/rsa.py

@@ -20,7 +20,9 @@ def encrypt(self, msg, key, sign_padding="pkcs1_padding"):
         return key.encrypt(
             msg,
             _padding(
-                mgf=padding.MGF1(algorithm=_chosen_hash()), algorithm=_chosen_hash(), label=None,
+                mgf=padding.MGF1(algorithm=_chosen_hash()),
+                algorithm=_chosen_hash(),
+                label=None,
             ),
         )
 

src/cryptojwt/jwk/jwk.py

@@ -93,7 +93,9 @@ def key_from_jwk_dict(jwk_dict, private=None):
         else:
             # Ecdsa public key.
             ec_pub_numbers = ec.EllipticCurvePublicNumbers(
-                base64url_to_long(_jwk_dict["x"]), base64url_to_long(_jwk_dict["y"]), curve,
+                base64url_to_long(_jwk_dict["x"]),
+                base64url_to_long(_jwk_dict["y"]),
+                curve,
             )
             _jwk_dict["pub_key"] = ec_pub_numbers.public_key(backends.default_backend())
         return ECKey(**_jwk_dict)

src/cryptojwt/jwk/rsa.py

@@ -128,7 +128,7 @@ def rsa_eq(key1, key2):
 
 
 def x509_rsa_load(txt):
-    """ So I get the same output format as loads produces
+    """So I get the same output format as loads produces
     :param txt:
     :return:
     """
@@ -172,10 +172,10 @@ def rsa_construct_private(numbers):
     try:
         cnum["iqmp"] = numbers["di"]
     except KeyError:
-        cnum["iqmp"] = rsa.rsa_crt_iqmp(cnum["p"], cnum["p"])
+        cnum["iqmp"] = rsa.rsa_crt_iqmp(cnum["p"], cnum["q"])
     else:
         if not numbers["di"]:
-            cnum["iqmp"] = rsa.rsa_crt_iqmp(cnum["p"], cnum["p"])
+            cnum["iqmp"] = rsa.rsa_crt_iqmp(cnum["p"], cnum["q"])
 
     rpubn = rsa.RSAPublicNumbers(e=numbers["e"], n=numbers["n"])
     rprivn = rsa.RSAPrivateNumbers(public_numbers=rpubn, **cnum)

src/cryptojwt/jws/jws.py

@@ -321,7 +321,11 @@ def verify_json(self, jws, keys=None, allow_none=False, at_least_one=False):
         for _sign in _signs:
             protected_headers = _sign.get("protected", "")
             token = b".".join(
-                [protected_headers.encode(), _payload.encode(), _sign["signature"].encode(),]
+                [
+                    protected_headers.encode(),
+                    _payload.encode(),
+                    _sign["signature"].encode(),
+                ]
             )
 
             unprotected_headers = _sign.get("header", {})

src/cryptojwt/jws/pss.py

@@ -38,7 +38,8 @@ def sign(self, msg, key):
         sig = key.sign(
             digest,
             padding.PSS(
-                mgf=padding.MGF1(self.hash_algorithm()), salt_length=padding.PSS.MAX_LENGTH,
+                mgf=padding.MGF1(self.hash_algorithm()),
+                salt_length=padding.PSS.MAX_LENGTH,
             ),
             utils.Prehashed(self.hash_algorithm()),
         )
@@ -59,7 +60,8 @@ def verify(self, msg, signature, key):
                 signature,
                 msg,
                 padding.PSS(
-                    mgf=padding.MGF1(self.hash_algorithm()), salt_length=padding.PSS.MAX_LENGTH,
+                    mgf=padding.MGF1(self.hash_algorithm()),
+                    salt_length=padding.PSS.MAX_LENGTH,
                 ),
                 self.hash_algorithm(),
             )

src/cryptojwt/jws/utils.py

@@ -11,7 +11,7 @@
 
 
 def left_hash(msg, func="HS256"):
-    """ Calculate left hash as described in
+    """Calculate left hash as described in
     https://openid.net/specs/openid-connect-core-1_0.html#CodeIDToken
     for at_hash and in
     for c_hash

src/cryptojwt/key_bundle.py

@@ -402,7 +402,9 @@ def do_remote(self):
 
         else:
             LOGGER.warning(
-                "HTTP status %d reading remote JWKS from %s", _http_resp.status_code, self.source,
+                "HTTP status %d reading remote JWKS from %s",
+                _http_resp.status_code,
+                self.source,
             )
             raise UpdateFailed(REMOTE_FAILED.format(self.source, _http_resp.status_code))
         self.last_updated = time.time()

src/cryptojwt/key_jar.py

@@ -762,7 +762,12 @@ def build_keyjar(key_conf, kid_template="", keyjar=None, issuer_id="", storage=N
 
 @deprecated_alias(issuer="issuer_id", owner="issuer_id")
 def init_key_jar(
-    public_path="", private_path="", key_defs="", issuer_id="", read_only=True, storage=None,
+    public_path="",
+    private_path="",
+    key_defs="",
+    issuer_id="",
+    read_only=True,
+    storage=None,
 ):
     """
     A number of cases here:
@@ -805,7 +810,10 @@ def init_key_jar(
     """
 
     _issuer = init_key_issuer(
-        public_path=public_path, private_path=private_path, key_defs=key_defs, read_only=read_only,
+        public_path=public_path,
+        private_path=private_path,
+        key_defs=key_defs,
+        read_only=read_only,
     )
 
     if _issuer is None:

src/cryptojwt/tools/keyconv.py

@@ -115,7 +115,10 @@ def pem2jwk(
 
 
 def export_jwk(
-    jwk: JWK, private: bool = False, encrypt: bool = False, passphrase: Optional[str] = None,
+    jwk: JWK,
+    private: bool = False,
+    encrypt: bool = False,
+    passphrase: Optional[str] = None,
 ) -> bytes:
     """Export JWK as PEM/bin"""
 

src/cryptojwt/utils.py

@@ -158,8 +158,7 @@ def as_unicode(b):
 
 
 def bytes2str_conv(item):
-    """
-    """
+    """"""
     if isinstance(item, bytes):
         return item.decode("utf-8")
     elif item is None or isinstance(item, (str, int, bool)):

tests/test_01_simplejwt.py

@@ -10,7 +10,10 @@ def _eq(l1, l2):
 def test_pack_jwt():
     _jwt = SimpleJWT(**{"alg": "none", "cty": "jwt"})
     jwt = _jwt.pack(
-        parts=[{"iss": "joe", "exp": 1300819380, "http://example.com/is_root": True}, "",]
+        parts=[
+            {"iss": "joe", "exp": 1300819380, "http://example.com/is_root": True},
+            "",
+        ]
     )
 
     p = jwt.split(".")

tests/test_02_jwk.py

@@ -515,6 +515,9 @@ def test_key_from_jwk_dict_rsa():
     _key = key_from_jwk_dict(jwk)
     assert isinstance(_key, RSAKey)
     assert _key.has_private_key()
+    _key2 = RSAKey(**jwk)
+    assert isinstance(_key2, RSAKey)
+    assert _key2.has_private_key()
 
 
 def test_key_from_jwk_dict_ec():
@@ -707,7 +710,10 @@ def test_x5t_calculation():
 
 @pytest.mark.parametrize(
     "filename,key_type",
-    [("ec-public.pem", ec.EllipticCurvePublicKey), ("rsa-public.pem", rsa.RSAPublicKey),],
+    [
+        ("ec-public.pem", ec.EllipticCurvePublicKey),
+        ("rsa-public.pem", rsa.RSAPublicKey),
+    ],
 )
 def test_import_public_key_from_pem_file(filename, key_type):
     _file = full_path(filename)

tests/test_04_key_issuer.py

@@ -221,7 +221,11 @@ def test_build_keyissuer_usage():
 
 def test_build_keyissuer_missing(tmpdir):
     keys = [
-        {"type": "RSA", "key": os.path.join(tmpdir.dirname, "missing_file"), "use": ["enc", "sig"],}
+        {
+            "type": "RSA",
+            "key": os.path.join(tmpdir.dirname, "missing_file"),
+            "use": ["enc", "sig"],
+        }
     ]
 
     key_issuer = build_keyissuer(keys)
@@ -239,7 +243,11 @@ def test_build_RSA_keyissuer_from_file(tmpdir):
 
 def test_build_EC_keyissuer_missing(tmpdir):
     keys = [
-        {"type": "EC", "key": os.path.join(tmpdir.dirname, "missing_file"), "use": ["enc", "sig"],}
+        {
+            "type": "EC",
+            "key": os.path.join(tmpdir.dirname, "missing_file"),
+            "use": ["enc", "sig"],
+        }
     ]
 
     key_issuer = build_keyissuer(keys)
@@ -616,7 +624,10 @@ def test_init_key_issuer_update():
 
     # New set of keys, JWKSs with keys and public written to file
     _keyissuer_1 = init_key_issuer(
-        private_path=PRIVATE_FILE, key_defs=KEYSPEC, public_path=PUBLIC_FILE, read_only=False,
+        private_path=PRIVATE_FILE,
+        key_defs=KEYSPEC,
+        public_path=PUBLIC_FILE,
+        read_only=False,
     )
     assert len(_keyissuer_1) == 2
 
@@ -646,7 +657,10 @@ def test_init_key_issuer_update():
     assert len(_keyissuer_3.get("sig", "EC")) == 1
 
     _keyissuer_4 = init_key_issuer(
-        private_path=PRIVATE_FILE, key_defs=KEYSPEC_2, public_path=PUBLIC_FILE, read_only=False,
+        private_path=PRIVATE_FILE,
+        key_defs=KEYSPEC_2,
+        public_path=PUBLIC_FILE,
+        read_only=False,
     )
 
     # Now it should

tests/test_04_key_jar.py

@@ -229,7 +229,11 @@ def test_build_keyjar_usage():
 
 def test_build_keyjar_missing(tmpdir):
     keys = [
-        {"type": "RSA", "key": os.path.join(tmpdir.dirname, "missing_file"), "use": ["enc", "sig"],}
+        {
+            "type": "RSA",
+            "key": os.path.join(tmpdir.dirname, "missing_file"),
+            "use": ["enc", "sig"],
+        }
     ]
 
     key_jar = build_keyjar(keys)
@@ -247,7 +251,11 @@ def test_build_RSA_keyjar_from_file(tmpdir):
 
 def test_build_EC_keyjar_missing(tmpdir):
     keys = [
-        {"type": "EC", "key": os.path.join(tmpdir.dirname, "missing_file"), "use": ["enc", "sig"],}
+        {
+            "type": "EC",
+            "key": os.path.join(tmpdir.dirname, "missing_file"),
+            "use": ["enc", "sig"],
+        }
     ]
 
     key_jar = build_keyjar(keys)
@@ -303,7 +311,8 @@ def test_items(self):
             ),
         )
         ks.add_kb(
-            "http://www.example.org", keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
+            "http://www.example.org",
+            keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
         )
 
         assert len(ks.items()) == 2
@@ -329,7 +338,8 @@ def test_issuer_extra_slash(self):
             ),
         )
         ks.add_kb(
-            "http://www.example.org", keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
+            "http://www.example.org",
+            keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
         )
 
         assert ks.get("sig", "RSA", "http://www.example.org/")
@@ -355,7 +365,8 @@ def test_issuer_missing_slash(self):
             ),
         )
         ks.add_kb(
-            "http://www.example.org/", keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
+            "http://www.example.org/",
+            keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
         )
 
         assert ks.get("sig", "RSA", "http://www.example.org")
@@ -381,7 +392,8 @@ def test_get_enc(self):
             ),
         )
         ks.add_kb(
-            "http://www.example.org/", keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
+            "http://www.example.org/",
+            keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
         )
 
         assert ks.get("enc", "oct")
@@ -407,7 +419,8 @@ def test_get_enc_not_mine(self):
             ),
         )
         ks.add_kb(
-            "http://www.example.org/", keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
+            "http://www.example.org/",
+            keybundle_from_local_file(RSAKEY, "der", ["ver", "sig"]),
         )
 
         assert ks.get("enc", "oct", "http://www.example.org/")
@@ -449,7 +462,8 @@ def test_provider(self):
         kj = KeyJar()
         _url = "https://connect-op.herokuapp.com/jwks.json"
         kj.load_keys(
-            "https://connect-op.heroku.com", jwks_uri=_url,
+            "https://connect-op.heroku.com",
+            jwks_uri=_url,
         )
         iss_keys = kj.get_issuer_keys("https://connect-op.heroku.com")
         if not iss_keys:
@@ -968,7 +982,10 @@ def test_init_key_jar_update():
     assert len(_keyjar_3.get_signing_key("EC")) == 1
 
     _keyjar_4 = init_key_jar(
-        private_path=PRIVATE_FILE, key_defs=KEYSPEC_2, public_path=PUBLIC_FILE, read_only=False,
+        private_path=PRIVATE_FILE,
+        key_defs=KEYSPEC_2,
+        public_path=PUBLIC_FILE,
+        read_only=False,
     )
 
     # Now it should