| 201 | |
| 202 | |
| 203 | class CryptographyRSAKey(Key): |
| 204 | SHA256 = hashes.SHA256 |
| 205 | SHA384 = hashes.SHA384 |
| 206 | SHA512 = hashes.SHA512 |
| 207 | |
| 208 | RSA1_5 = padding.PKCS1v15() |
| 209 | RSA_OAEP = padding.OAEP(padding.MGF1(hashes.SHA1()), hashes.SHA1(), None) |
| 210 | RSA_OAEP_256 = padding.OAEP(padding.MGF1(hashes.SHA256()), hashes.SHA256(), None) |
| 211 | |
| 212 | def __init__(self, key, algorithm, cryptography_backend=default_backend): |
| 213 | if algorithm not in ALGORITHMS.RSA: |
| 214 | raise JWKError("hash_alg: %s is not a valid hash algorithm" % algorithm) |
| 215 | |
| 216 | self.hash_alg = { |
| 217 | ALGORITHMS.RS256: self.SHA256, |
| 218 | ALGORITHMS.RS384: self.SHA384, |
| 219 | ALGORITHMS.RS512: self.SHA512, |
| 220 | }.get(algorithm) |
| 221 | self._algorithm = algorithm |
| 222 | |
| 223 | self.padding = { |
| 224 | ALGORITHMS.RSA1_5: self.RSA1_5, |
| 225 | ALGORITHMS.RSA_OAEP: self.RSA_OAEP, |
| 226 | ALGORITHMS.RSA_OAEP_256: self.RSA_OAEP_256, |
| 227 | }.get(algorithm) |
| 228 | |
| 229 | self.cryptography_backend = cryptography_backend |
| 230 | |
| 231 | # if it conforms to RSAPublicKey or RSAPrivateKey interface |
| 232 | if (hasattr(key, "public_bytes") and hasattr(key, "public_numbers")) or hasattr(key, "private_bytes"): |
| 233 | self.prepared_key = key |
| 234 | return |
| 235 | |
| 236 | if isinstance(key, dict): |
| 237 | self.prepared_key = self._process_jwk(key) |
| 238 | return |
| 239 | |
| 240 | if isinstance(key, str): |
| 241 | key = key.encode("utf-8") |
| 242 | |
| 243 | if isinstance(key, bytes): |
| 244 | try: |
| 245 | if key.startswith(b"-----BEGIN CERTIFICATE-----"): |
| 246 | self._process_cert(key) |
| 247 | return |
| 248 | |
| 249 | try: |
| 250 | self.prepared_key = load_pem_public_key(key, self.cryptography_backend()) |
| 251 | except ValueError: |
| 252 | self.prepared_key = load_pem_private_key(key, password=None, backend=self.cryptography_backend()) |
| 253 | except Exception as e: |
| 254 | raise JWKError(e) |
| 255 | return |
| 256 | |
| 257 | raise JWKError("Unable to parse an RSA_JWK from key: %s" % key) |
| 258 | |
| 259 | def _process_jwk(self, jwk_dict): |
| 260 | if not jwk_dict.get("kty") == "RSA": |
no outgoing calls
searching dependent graphs…