Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jwt/algorithms.py: 21%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import hashlib
4import hmac
5import json
6import os
7import sys
8from abc import ABC, abstractmethod
9from typing import (
10 TYPE_CHECKING,
11 Any,
12 ClassVar,
13 Literal,
14 NoReturn,
15 Union,
16 cast,
17 get_args,
18 overload,
19)
21from .exceptions import InvalidKeyError
22from .types import HashlibHash, JWKDict
23from .utils import (
24 base64url_decode,
25 base64url_encode,
26 der_to_raw_signature,
27 force_bytes,
28 from_base64url_uint,
29 is_pem_format,
30 is_ssh_key,
31 raw_to_der_signature,
32 to_base64url_uint,
33)
35try:
36 from cryptography import x509
37 from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm
38 from cryptography.hazmat.backends import default_backend
39 from cryptography.hazmat.primitives import hashes
40 from cryptography.hazmat.primitives.asymmetric import padding
41 from cryptography.hazmat.primitives.asymmetric.ec import (
42 ECDSA,
43 SECP256K1,
44 SECP256R1,
45 SECP384R1,
46 SECP521R1,
47 EllipticCurve,
48 EllipticCurvePrivateKey,
49 EllipticCurvePrivateNumbers,
50 EllipticCurvePublicKey,
51 EllipticCurvePublicNumbers,
52 )
53 from cryptography.hazmat.primitives.asymmetric.ed448 import (
54 Ed448PrivateKey,
55 Ed448PublicKey,
56 )
57 from cryptography.hazmat.primitives.asymmetric.ed25519 import (
58 Ed25519PrivateKey,
59 Ed25519PublicKey,
60 )
61 from cryptography.hazmat.primitives.asymmetric.rsa import (
62 RSAPrivateKey,
63 RSAPrivateNumbers,
64 RSAPublicKey,
65 RSAPublicNumbers,
66 rsa_crt_dmp1,
67 rsa_crt_dmq1,
68 rsa_crt_iqmp,
69 rsa_recover_prime_factors,
70 )
71 from cryptography.hazmat.primitives.serialization import (
72 Encoding,
73 NoEncryption,
74 PrivateFormat,
75 PublicFormat,
76 load_der_public_key,
77 load_pem_private_key,
78 load_pem_public_key,
79 load_ssh_public_key,
80 )
82 if sys.version_info >= (3, 10):
83 from typing import TypeAlias
84 else:
85 # Python 3.9 and lower
86 from typing_extensions import TypeAlias
88 # Type aliases for convenience in algorithms method signatures
89 AllowedRSAKeys: TypeAlias = Union[RSAPrivateKey, RSAPublicKey]
90 AllowedECKeys: TypeAlias = Union[EllipticCurvePrivateKey, EllipticCurvePublicKey]
91 AllowedOKPKeys: TypeAlias = Union[
92 Ed25519PrivateKey, Ed25519PublicKey, Ed448PrivateKey, Ed448PublicKey
93 ]
94 AllowedKeys: TypeAlias = Union[AllowedRSAKeys, AllowedECKeys, AllowedOKPKeys]
95 #: Type alias for allowed ``cryptography`` private keys (requires ``cryptography`` to be installed)
96 AllowedPrivateKeys: TypeAlias = Union[
97 RSAPrivateKey, EllipticCurvePrivateKey, Ed25519PrivateKey, Ed448PrivateKey
98 ]
99 #: Type alias for allowed ``cryptography`` public keys (requires ``cryptography`` to be installed)
100 AllowedPublicKeys: TypeAlias = Union[
101 RSAPublicKey, EllipticCurvePublicKey, Ed25519PublicKey, Ed448PublicKey
102 ]
104 if TYPE_CHECKING or bool(os.getenv("SPHINX_BUILD", "")):
105 from cryptography.hazmat.primitives.asymmetric.types import (
106 PrivateKeyTypes,
107 PublicKeyTypes,
108 )
110 has_crypto = True
111except ModuleNotFoundError:
112 if sys.version_info >= (3, 11):
113 from typing import Never
114 else:
115 from typing_extensions import Never
117 AllowedRSAKeys = Never # type: ignore[misc]
118 AllowedECKeys = Never # type: ignore[misc]
119 AllowedOKPKeys = Never # type: ignore[misc]
120 AllowedKeys = Never # type: ignore[misc]
121 AllowedPrivateKeys = Never # type: ignore[misc]
122 AllowedPublicKeys = Never # type: ignore[misc]
123 has_crypto = False
126requires_cryptography = {
127 "RS256",
128 "RS384",
129 "RS512",
130 "ES256",
131 "ES256K",
132 "ES384",
133 "ES521",
134 "ES512",
135 "PS256",
136 "PS384",
137 "PS512",
138 "EdDSA",
139}
142def get_default_algorithms() -> dict[str, Algorithm]:
143 """
144 Returns the algorithms that are implemented by the library.
145 """
146 default_algorithms: dict[str, Algorithm] = {
147 "none": NoneAlgorithm(),
148 "HS256": HMACAlgorithm(HMACAlgorithm.SHA256),
149 "HS384": HMACAlgorithm(HMACAlgorithm.SHA384),
150 "HS512": HMACAlgorithm(HMACAlgorithm.SHA512),
151 }
153 if has_crypto:
154 default_algorithms.update(
155 {
156 "RS256": RSAAlgorithm(RSAAlgorithm.SHA256),
157 "RS384": RSAAlgorithm(RSAAlgorithm.SHA384),
158 "RS512": RSAAlgorithm(RSAAlgorithm.SHA512),
159 "ES256": ECAlgorithm(ECAlgorithm.SHA256, SECP256R1),
160 "ES256K": ECAlgorithm(ECAlgorithm.SHA256, SECP256K1),
161 "ES384": ECAlgorithm(ECAlgorithm.SHA384, SECP384R1),
162 "ES521": ECAlgorithm(ECAlgorithm.SHA512, SECP521R1),
163 "ES512": ECAlgorithm(
164 ECAlgorithm.SHA512, SECP521R1
165 ), # Backward compat for #219 fix
166 "PS256": RSAPSSAlgorithm(RSAPSSAlgorithm.SHA256),
167 "PS384": RSAPSSAlgorithm(RSAPSSAlgorithm.SHA384),
168 "PS512": RSAPSSAlgorithm(RSAPSSAlgorithm.SHA512),
169 "EdDSA": OKPAlgorithm(),
170 }
171 )
173 return default_algorithms
176class Algorithm(ABC):
177 """
178 The interface for an algorithm used to sign and verify tokens.
179 """
181 # pyjwt-964: Validate to ensure the key passed in was decoded to the correct cryptography key family
182 _crypto_key_types: tuple[type[AllowedKeys], ...] | None = None
184 def compute_hash_digest(self, bytestr: bytes) -> bytes:
185 """
186 Compute a hash digest using the specified algorithm's hash algorithm.
188 If there is no hash algorithm, raises a NotImplementedError.
189 """
190 # lookup self.hash_alg if defined in a way that mypy can understand
191 hash_alg = getattr(self, "hash_alg", None)
192 if hash_alg is None:
193 raise NotImplementedError
195 if (
196 has_crypto
197 and isinstance(hash_alg, type)
198 and issubclass(hash_alg, hashes.HashAlgorithm)
199 ):
200 digest = hashes.Hash(hash_alg(), backend=default_backend())
201 digest.update(bytestr)
202 return bytes(digest.finalize())
203 else:
204 return bytes(hash_alg(bytestr).digest())
206 def check_crypto_key_type(self, key: PublicKeyTypes | PrivateKeyTypes) -> None:
207 """Check that the key belongs to the right cryptographic family.
209 Note that this method only works when ``cryptography`` is installed.
211 :param key: Potentially a cryptography key
212 :type key: :py:data:`PublicKeyTypes <cryptography.hazmat.primitives.asymmetric.types.PublicKeyTypes>` | :py:data:`PrivateKeyTypes <cryptography.hazmat.primitives.asymmetric.types.PrivateKeyTypes>`
213 :raises ValueError: if ``cryptography`` is not installed, or this method is called by a non-cryptography algorithm
214 :raises InvalidKeyError: if the key doesn't match the expected key classes
215 """
216 if not has_crypto or self._crypto_key_types is None:
217 raise ValueError(
218 "This method requires the cryptography library, and should only be used by cryptography-based algorithms."
219 )
221 if not isinstance(key, self._crypto_key_types):
222 valid_classes = (cls.__name__ for cls in self._crypto_key_types)
223 actual_class = key.__class__.__name__
224 self_class = self.__class__.__name__
225 raise InvalidKeyError(
226 f"Expected one of {valid_classes}, got: {actual_class}. Invalid Key type for {self_class}"
227 )
229 @abstractmethod
230 def prepare_key(self, key: Any) -> Any:
231 """
232 Performs necessary validation and conversions on the key and returns
233 the key value in the proper format for sign() and verify().
234 """
236 @abstractmethod
237 def sign(self, msg: bytes, key: Any) -> bytes:
238 """
239 Returns a digital signature for the specified message
240 using the specified key value.
241 """
243 @abstractmethod
244 def verify(self, msg: bytes, key: Any, sig: bytes) -> bool:
245 """
246 Verifies that the specified digital signature is valid
247 for the specified message and key values.
248 """
250 @overload
251 @staticmethod
252 @abstractmethod
253 def to_jwk(key_obj: Any, as_dict: Literal[True]) -> JWKDict: ... # pragma: no cover
255 @overload
256 @staticmethod
257 @abstractmethod
258 def to_jwk(
259 key_obj: Any, as_dict: Literal[False] = False
260 ) -> str: ... # pragma: no cover
262 @staticmethod
263 @abstractmethod
264 def to_jwk(key_obj: Any, as_dict: bool = False) -> JWKDict | str:
265 """
266 Serializes a given key into a JWK
267 """
269 @staticmethod
270 @abstractmethod
271 def from_jwk(jwk: str | JWKDict) -> Any:
272 """
273 Deserializes a given key from JWK back into a key object
274 """
276 def check_key_length(self, key: Any) -> str | None:
277 """
278 Return a warning message if the key is below the minimum
279 recommended length for this algorithm, or None if adequate.
280 """
281 return None
284class NoneAlgorithm(Algorithm):
285 """
286 Placeholder for use when no signing or verification
287 operations are required.
288 """
290 def prepare_key(self, key: str | None) -> None:
291 if key == "":
292 key = None
294 if key is not None:
295 raise InvalidKeyError('When alg = "none", key value must be None.')
297 return key
299 def sign(self, msg: bytes, key: None) -> bytes:
300 return b""
302 def verify(self, msg: bytes, key: None, sig: bytes) -> bool:
303 return False
305 @staticmethod
306 def to_jwk(key_obj: Any, as_dict: bool = False) -> NoReturn:
307 raise NotImplementedError()
309 @staticmethod
310 def from_jwk(jwk: str | JWKDict) -> NoReturn:
311 raise NotImplementedError()
314class HMACAlgorithm(Algorithm):
315 """
316 Performs signing and verification operations using HMAC
317 and the specified hash function.
318 """
320 SHA256: ClassVar[HashlibHash] = hashlib.sha256
321 SHA384: ClassVar[HashlibHash] = hashlib.sha384
322 SHA512: ClassVar[HashlibHash] = hashlib.sha512
324 def __init__(self, hash_alg: HashlibHash) -> None:
325 self.hash_alg = hash_alg
327 @staticmethod
328 def _is_der_key(key_bytes: bytes) -> bool:
329 if not has_crypto:
330 return False
332 try:
333 load_der_public_key(key_bytes)
334 except (TypeError, ValueError, UnsupportedAlgorithm):
335 pass
336 else:
337 return True
339 try:
340 x509.load_der_x509_certificate(key_bytes)
341 except (TypeError, ValueError, UnsupportedAlgorithm):
342 return False
343 else:
344 return True
346 def prepare_key(self, key: str | bytes) -> bytes:
347 key_bytes = force_bytes(key)
349 if len(key_bytes) == 0:
350 raise InvalidKeyError("HMAC key must not be empty.")
352 if (
353 is_pem_format(key_bytes)
354 or is_ssh_key(key_bytes)
355 or self._is_der_key(key_bytes)
356 ):
357 raise InvalidKeyError(
358 "The specified key is an asymmetric key or x509 certificate and"
359 " should not be used as an HMAC secret."
360 )
362 # Defense against algorithm-confusion attacks: an attacker with
363 # control over the token header can force this code path by setting
364 # alg=HS*, and HMACAlgorithm is the only algorithm that accepts
365 # arbitrary bytes as a valid secret. Other algorithms reject
366 # non-key-shaped input naturally. Even a symmetric (kty=oct) JWK
367 # should be loaded via PyJWK / from_jwk rather than fed as raw JSON
368 # bytes (whose contents are not the secret material).
369 try:
370 jwk_obj = json.loads(key_bytes, parse_int=lambda _: 0)
371 except RecursionError:
372 try:
373 decoded_key = key_bytes.decode(
374 json.detect_encoding(key_bytes), errors="surrogatepass"
375 )
376 except UnicodeError:
377 decoded_key = ""
378 stripped_key = decoded_key.lstrip("\ufeff \t\r\n")
379 has_jwk_member = False
380 index = 0
381 while index < len(decoded_key):
382 if decoded_key[index] != '"':
383 index += 1
384 continue
385 end = index + 1
386 while end < len(decoded_key):
387 if decoded_key[end] == "\\":
388 end += 2
389 elif decoded_key[end] == '"':
390 break
391 else:
392 end += 1
393 if end >= len(decoded_key):
394 break
395 next_index = end + 1
396 while (
397 next_index < len(decoded_key)
398 and decoded_key[next_index] in " \t\r\n"
399 ):
400 next_index += 1
401 if next_index < len(decoded_key) and decoded_key[next_index] == ":":
402 try:
403 has_jwk_member = (
404 json.loads(decoded_key[index : end + 1]) == "kty"
405 )
406 except ValueError:
407 pass
408 if has_jwk_member:
409 break
410 index = end + 1
411 if stripped_key.startswith("{") or (
412 stripped_key.startswith("[") and has_jwk_member
413 ):
414 raise InvalidKeyError(
415 "The specified key looks like a JWK and should not be "
416 "used directly as an HMAC secret. Load it via "
417 "PyJWK / HMACAlgorithm.from_jwk first."
418 ) from None
419 jwk_obj = None
420 except ValueError:
421 jwk_obj = None
422 contains_jwk_member = False
423 objects_to_check = [jwk_obj]
424 while objects_to_check:
425 obj = objects_to_check.pop()
426 if isinstance(obj, dict):
427 if "kty" in obj:
428 contains_jwk_member = True
429 break
430 objects_to_check.extend(obj.values())
431 elif isinstance(obj, list):
432 objects_to_check.extend(obj)
433 if contains_jwk_member:
434 raise InvalidKeyError(
435 "The specified key looks like a JWK and should not be "
436 "used directly as an HMAC secret. Load it via "
437 "PyJWK / HMACAlgorithm.from_jwk first."
438 )
440 return key_bytes
442 @overload
443 @staticmethod
444 def to_jwk(key_obj: str | bytes, as_dict: Literal[True]) -> JWKDict: ...
446 @overload
447 @staticmethod
448 def to_jwk(key_obj: str | bytes, as_dict: Literal[False] = False) -> str: ...
450 @staticmethod
451 def to_jwk(key_obj: str | bytes, as_dict: bool = False) -> JWKDict | str:
452 jwk = {
453 "k": base64url_encode(force_bytes(key_obj)).decode(),
454 "kty": "oct",
455 }
457 if as_dict:
458 return jwk
459 else:
460 return json.dumps(jwk)
462 @staticmethod
463 def from_jwk(jwk: str | JWKDict) -> bytes:
464 try:
465 if isinstance(jwk, str):
466 obj: JWKDict = json.loads(jwk)
467 elif isinstance(jwk, dict):
468 obj = jwk
469 else:
470 raise ValueError
471 except ValueError:
472 raise InvalidKeyError("Key is not valid JSON") from None
474 if obj.get("kty") != "oct":
475 raise InvalidKeyError("Not an HMAC key")
477 key_bytes = base64url_decode(obj["k"])
478 if len(key_bytes) == 0:
479 raise InvalidKeyError("HMAC key must not be empty.")
480 return key_bytes
482 def check_key_length(self, key: bytes) -> str | None:
483 min_length = self.hash_alg().digest_size
484 if len(key) < min_length:
485 return (
486 f"The HMAC key is {len(key)} bytes long, which is below "
487 f"the minimum recommended length of {min_length} bytes for "
488 f"{self.hash_alg().name.upper()}. "
489 f"See RFC 7518 Section 3.2."
490 )
491 return None
493 def sign(self, msg: bytes, key: bytes) -> bytes:
494 return hmac.new(key, msg, self.hash_alg).digest()
496 def verify(self, msg: bytes, key: bytes, sig: bytes) -> bool:
497 return hmac.compare_digest(sig, self.sign(msg, key))
500if has_crypto:
502 class RSAAlgorithm(Algorithm):
503 """
504 Performs signing and verification operations using
505 RSASSA-PKCS-v1_5 and the specified hash function.
506 """
508 SHA256: ClassVar[type[hashes.HashAlgorithm]] = hashes.SHA256
509 SHA384: ClassVar[type[hashes.HashAlgorithm]] = hashes.SHA384
510 SHA512: ClassVar[type[hashes.HashAlgorithm]] = hashes.SHA512
512 _crypto_key_types = cast(
513 tuple[type[AllowedKeys], ...],
514 get_args(Union[RSAPrivateKey, RSAPublicKey]),
515 )
516 _MIN_KEY_SIZE: ClassVar[int] = 2048
518 def __init__(self, hash_alg: type[hashes.HashAlgorithm]) -> None:
519 self.hash_alg = hash_alg
521 def check_key_length(self, key: AllowedRSAKeys) -> str | None:
522 if key.key_size < self._MIN_KEY_SIZE:
523 return (
524 f"The RSA key is {key.key_size} bits long, which is below "
525 f"the minimum recommended size of {self._MIN_KEY_SIZE} bits. "
526 f"See NIST SP 800-131A."
527 )
528 return None
530 def prepare_key(self, key: AllowedRSAKeys | str | bytes) -> AllowedRSAKeys:
531 if isinstance(key, self._crypto_key_types):
532 # Cast is required for type narrowing on Python 3.9's mypy
533 # but redundant on newer mypy versions; suppress both
534 # diagnostics so the line works across all supported envs.
535 return cast(AllowedRSAKeys, key) # type: ignore[redundant-cast,unused-ignore]
537 if not isinstance(key, (bytes, str)):
538 raise TypeError("Expecting a PEM-formatted key.")
540 key_bytes = force_bytes(key)
542 try:
543 if key_bytes.startswith(b"ssh-rsa"):
544 public_key: PublicKeyTypes = load_ssh_public_key(key_bytes)
545 self.check_crypto_key_type(public_key)
546 return cast(RSAPublicKey, public_key)
547 else:
548 private_key: PrivateKeyTypes = load_pem_private_key(
549 key_bytes, password=None
550 )
551 self.check_crypto_key_type(private_key)
552 return cast(RSAPrivateKey, private_key)
553 except ValueError:
554 try:
555 public_key = load_pem_public_key(key_bytes)
556 self.check_crypto_key_type(public_key)
557 return cast(RSAPublicKey, public_key)
558 except (ValueError, UnsupportedAlgorithm):
559 raise InvalidKeyError(
560 "Could not parse the provided public key."
561 ) from None
563 @overload
564 @staticmethod
565 def to_jwk(key_obj: AllowedRSAKeys, as_dict: Literal[True]) -> JWKDict: ...
567 @overload
568 @staticmethod
569 def to_jwk(key_obj: AllowedRSAKeys, as_dict: Literal[False] = False) -> str: ...
571 @staticmethod
572 def to_jwk(key_obj: AllowedRSAKeys, as_dict: bool = False) -> JWKDict | str:
573 obj: dict[str, Any] | None = None
575 if hasattr(key_obj, "private_numbers"):
576 # Private key
577 numbers = key_obj.private_numbers()
579 obj = {
580 "kty": "RSA",
581 "key_ops": ["sign"],
582 "n": to_base64url_uint(numbers.public_numbers.n).decode(),
583 "e": to_base64url_uint(numbers.public_numbers.e).decode(),
584 "d": to_base64url_uint(numbers.d).decode(),
585 "p": to_base64url_uint(numbers.p).decode(),
586 "q": to_base64url_uint(numbers.q).decode(),
587 "dp": to_base64url_uint(numbers.dmp1).decode(),
588 "dq": to_base64url_uint(numbers.dmq1).decode(),
589 "qi": to_base64url_uint(numbers.iqmp).decode(),
590 }
592 elif hasattr(key_obj, "verify"):
593 # Public key
594 numbers = key_obj.public_numbers()
596 obj = {
597 "kty": "RSA",
598 "key_ops": ["verify"],
599 "n": to_base64url_uint(numbers.n).decode(),
600 "e": to_base64url_uint(numbers.e).decode(),
601 }
602 else:
603 raise InvalidKeyError("Not a public or private key")
605 if as_dict:
606 return obj
607 else:
608 return json.dumps(obj)
610 @staticmethod
611 def from_jwk(jwk: str | JWKDict) -> AllowedRSAKeys:
612 try:
613 if isinstance(jwk, str):
614 obj = json.loads(jwk)
615 elif isinstance(jwk, dict):
616 obj = jwk
617 else:
618 raise ValueError
619 except ValueError:
620 raise InvalidKeyError("Key is not valid JSON") from None
622 if obj.get("kty") != "RSA":
623 raise InvalidKeyError("Not an RSA key") from None
625 if "d" in obj and "e" in obj and "n" in obj:
626 # Private key
627 if "oth" in obj:
628 raise InvalidKeyError(
629 "Unsupported RSA private key: > 2 primes not supported"
630 )
632 other_props = ["p", "q", "dp", "dq", "qi"]
633 props_found = [prop in obj for prop in other_props]
634 any_props_found = any(props_found)
636 if any_props_found and not all(props_found):
637 raise InvalidKeyError(
638 "RSA key must include all parameters if any are present besides d"
639 ) from None
641 public_numbers = RSAPublicNumbers(
642 from_base64url_uint(obj["e"]),
643 from_base64url_uint(obj["n"]),
644 )
646 if any_props_found:
647 numbers = RSAPrivateNumbers(
648 d=from_base64url_uint(obj["d"]),
649 p=from_base64url_uint(obj["p"]),
650 q=from_base64url_uint(obj["q"]),
651 dmp1=from_base64url_uint(obj["dp"]),
652 dmq1=from_base64url_uint(obj["dq"]),
653 iqmp=from_base64url_uint(obj["qi"]),
654 public_numbers=public_numbers,
655 )
656 else:
657 d = from_base64url_uint(obj["d"])
658 p, q = rsa_recover_prime_factors(
659 public_numbers.n, d, public_numbers.e
660 )
662 numbers = RSAPrivateNumbers(
663 d=d,
664 p=p,
665 q=q,
666 dmp1=rsa_crt_dmp1(d, p),
667 dmq1=rsa_crt_dmq1(d, q),
668 iqmp=rsa_crt_iqmp(p, q),
669 public_numbers=public_numbers,
670 )
672 return numbers.private_key()
673 elif "n" in obj and "e" in obj:
674 # Public key
675 return RSAPublicNumbers(
676 from_base64url_uint(obj["e"]),
677 from_base64url_uint(obj["n"]),
678 ).public_key()
679 else:
680 raise InvalidKeyError("Not a public or private key")
682 def sign(self, msg: bytes, key: RSAPrivateKey) -> bytes:
683 signature: bytes = key.sign(msg, padding.PKCS1v15(), self.hash_alg())
684 return signature
686 def verify(self, msg: bytes, key: RSAPublicKey, sig: bytes) -> bool:
687 try:
688 key.verify(sig, msg, padding.PKCS1v15(), self.hash_alg())
689 return True
690 except InvalidSignature:
691 return False
693 class ECAlgorithm(Algorithm):
694 """
695 Performs signing and verification operations using
696 ECDSA and the specified hash function
697 """
699 SHA256: ClassVar[type[hashes.HashAlgorithm]] = hashes.SHA256
700 SHA384: ClassVar[type[hashes.HashAlgorithm]] = hashes.SHA384
701 SHA512: ClassVar[type[hashes.HashAlgorithm]] = hashes.SHA512
703 _crypto_key_types = cast(
704 tuple[type[AllowedKeys], ...],
705 get_args(Union[EllipticCurvePrivateKey, EllipticCurvePublicKey]),
706 )
708 def __init__(
709 self,
710 hash_alg: type[hashes.HashAlgorithm],
711 expected_curve: type[EllipticCurve] | None = None,
712 ) -> None:
713 self.hash_alg = hash_alg
714 self.expected_curve = expected_curve
716 def _validate_curve(self, key: AllowedECKeys) -> None:
717 """Validate that the key's curve matches the expected curve."""
718 if self.expected_curve is None:
719 return
721 if not isinstance(key.curve, self.expected_curve):
722 raise InvalidKeyError(
723 f"The key's curve '{key.curve.name}' does not match the expected "
724 f"curve '{self.expected_curve.name}' for this algorithm"
725 )
727 def prepare_key(self, key: AllowedECKeys | str | bytes) -> AllowedECKeys:
728 if isinstance(key, self._crypto_key_types):
729 # See note in RSAAlgorithm.prepare_key.
730 ec_key = cast(AllowedECKeys, key) # type: ignore[redundant-cast,unused-ignore]
731 self._validate_curve(ec_key)
732 return ec_key
734 if not isinstance(key, (bytes, str)):
735 raise TypeError("Expecting a PEM-formatted key.")
737 key_bytes = force_bytes(key)
739 # Attempt to load key. We don't know if it's
740 # a Signing Key or a Verifying Key, so we try
741 # the Verifying Key first.
742 try:
743 if key_bytes.startswith(b"ecdsa-sha2-"):
744 public_key: PublicKeyTypes = load_ssh_public_key(key_bytes)
745 else:
746 public_key = load_pem_public_key(key_bytes)
748 # Explicit check the key to prevent confusing errors from cryptography
749 self.check_crypto_key_type(public_key)
750 ec_public_key = cast(EllipticCurvePublicKey, public_key)
751 self._validate_curve(ec_public_key)
752 return ec_public_key
753 except ValueError:
754 private_key = load_pem_private_key(key_bytes, password=None)
755 self.check_crypto_key_type(private_key)
756 ec_private_key = cast(EllipticCurvePrivateKey, private_key)
757 self._validate_curve(ec_private_key)
758 return ec_private_key
760 def sign(self, msg: bytes, key: EllipticCurvePrivateKey) -> bytes:
761 der_sig = key.sign(msg, ECDSA(self.hash_alg()))
763 return der_to_raw_signature(der_sig, key.curve)
765 def verify(self, msg: bytes, key: AllowedECKeys, sig: bytes) -> bool:
766 try:
767 der_sig = raw_to_der_signature(sig, key.curve)
768 except ValueError:
769 return False
771 try:
772 public_key = (
773 key.public_key()
774 if isinstance(key, EllipticCurvePrivateKey)
775 else key
776 )
777 public_key.verify(der_sig, msg, ECDSA(self.hash_alg()))
778 return True
779 except InvalidSignature:
780 return False
782 @overload
783 @staticmethod
784 def to_jwk(key_obj: AllowedECKeys, as_dict: Literal[True]) -> JWKDict: ...
786 @overload
787 @staticmethod
788 def to_jwk(key_obj: AllowedECKeys, as_dict: Literal[False] = False) -> str: ...
790 @staticmethod
791 def to_jwk(key_obj: AllowedECKeys, as_dict: bool = False) -> JWKDict | str:
792 if isinstance(key_obj, EllipticCurvePrivateKey):
793 public_numbers = key_obj.public_key().public_numbers()
794 elif isinstance(key_obj, EllipticCurvePublicKey):
795 public_numbers = key_obj.public_numbers()
796 else:
797 raise InvalidKeyError("Not a public or private key")
799 if isinstance(key_obj.curve, SECP256R1):
800 crv = "P-256"
801 elif isinstance(key_obj.curve, SECP384R1):
802 crv = "P-384"
803 elif isinstance(key_obj.curve, SECP521R1):
804 crv = "P-521"
805 elif isinstance(key_obj.curve, SECP256K1):
806 crv = "secp256k1"
807 else:
808 raise InvalidKeyError(f"Invalid curve: {key_obj.curve}")
810 obj: dict[str, Any] = {
811 "kty": "EC",
812 "crv": crv,
813 "x": to_base64url_uint(
814 public_numbers.x,
815 bit_length=key_obj.curve.key_size,
816 ).decode(),
817 "y": to_base64url_uint(
818 public_numbers.y,
819 bit_length=key_obj.curve.key_size,
820 ).decode(),
821 }
823 if isinstance(key_obj, EllipticCurvePrivateKey):
824 obj["d"] = to_base64url_uint(
825 key_obj.private_numbers().private_value,
826 bit_length=key_obj.curve.key_size,
827 ).decode()
829 if as_dict:
830 return obj
831 else:
832 return json.dumps(obj)
834 @staticmethod
835 def from_jwk(jwk: str | JWKDict) -> AllowedECKeys:
836 try:
837 if isinstance(jwk, str):
838 obj = json.loads(jwk)
839 elif isinstance(jwk, dict):
840 obj = jwk
841 else:
842 raise ValueError
843 except ValueError:
844 raise InvalidKeyError("Key is not valid JSON") from None
846 if obj.get("kty") != "EC":
847 raise InvalidKeyError("Not an Elliptic curve key") from None
849 if "x" not in obj or "y" not in obj:
850 raise InvalidKeyError("Not an Elliptic curve key") from None
852 x = base64url_decode(obj.get("x"))
853 y = base64url_decode(obj.get("y"))
855 curve = obj.get("crv")
856 curve_obj: EllipticCurve
858 if curve == "P-256":
859 if len(x) == len(y) == 32:
860 curve_obj = SECP256R1()
861 else:
862 raise InvalidKeyError(
863 "Coords should be 32 bytes for curve P-256"
864 ) from None
865 elif curve == "P-384":
866 if len(x) == len(y) == 48:
867 curve_obj = SECP384R1()
868 else:
869 raise InvalidKeyError(
870 "Coords should be 48 bytes for curve P-384"
871 ) from None
872 elif curve == "P-521":
873 if len(x) == len(y) == 66:
874 curve_obj = SECP521R1()
875 else:
876 raise InvalidKeyError(
877 "Coords should be 66 bytes for curve P-521"
878 ) from None
879 elif curve == "secp256k1":
880 if len(x) == len(y) == 32:
881 curve_obj = SECP256K1()
882 else:
883 raise InvalidKeyError(
884 "Coords should be 32 bytes for curve secp256k1"
885 )
886 else:
887 raise InvalidKeyError(f"Invalid curve: {curve}")
889 public_numbers = EllipticCurvePublicNumbers(
890 x=int.from_bytes(x, byteorder="big"),
891 y=int.from_bytes(y, byteorder="big"),
892 curve=curve_obj,
893 )
895 if "d" not in obj:
896 return public_numbers.public_key()
898 d = base64url_decode(obj.get("d"))
899 if len(d) != len(x):
900 raise InvalidKeyError(
901 "D should be {} bytes for curve {}", len(x), curve
902 )
904 return EllipticCurvePrivateNumbers(
905 int.from_bytes(d, byteorder="big"), public_numbers
906 ).private_key()
908 class RSAPSSAlgorithm(RSAAlgorithm):
909 """
910 Performs a signature using RSASSA-PSS with MGF1
911 """
913 def sign(self, msg: bytes, key: RSAPrivateKey) -> bytes:
914 signature: bytes = key.sign(
915 msg,
916 padding.PSS(
917 mgf=padding.MGF1(self.hash_alg()),
918 salt_length=self.hash_alg().digest_size,
919 ),
920 self.hash_alg(),
921 )
922 return signature
924 def verify(self, msg: bytes, key: RSAPublicKey, sig: bytes) -> bool:
925 try:
926 key.verify(
927 sig,
928 msg,
929 padding.PSS(
930 mgf=padding.MGF1(self.hash_alg()),
931 salt_length=self.hash_alg().digest_size,
932 ),
933 self.hash_alg(),
934 )
935 return True
936 except InvalidSignature:
937 return False
939 class OKPAlgorithm(Algorithm):
940 """
941 Performs signing and verification operations using EdDSA
943 This class requires ``cryptography>=2.6`` to be installed.
944 """
946 _crypto_key_types = cast(
947 tuple[type[AllowedKeys], ...],
948 get_args(
949 Union[
950 Ed25519PrivateKey,
951 Ed25519PublicKey,
952 Ed448PrivateKey,
953 Ed448PublicKey,
954 ]
955 ),
956 )
958 def __init__(self, **kwargs: Any) -> None:
959 pass
961 def prepare_key(self, key: AllowedOKPKeys | str | bytes) -> AllowedOKPKeys:
962 if not isinstance(key, (str, bytes)):
963 self.check_crypto_key_type(key)
964 return key
966 key_str = key.decode("utf-8") if isinstance(key, bytes) else key
967 key_bytes = key.encode("utf-8") if isinstance(key, str) else key
969 loaded_key: PublicKeyTypes | PrivateKeyTypes
970 if "-----BEGIN PUBLIC" in key_str:
971 loaded_key = load_pem_public_key(key_bytes)
972 elif "-----BEGIN PRIVATE" in key_str:
973 loaded_key = load_pem_private_key(key_bytes, password=None)
974 elif key_str[0:4] == "ssh-":
975 loaded_key = load_ssh_public_key(key_bytes)
976 else:
977 raise InvalidKeyError("Not a public or private key")
979 # Explicit check the key to prevent confusing errors from cryptography
980 self.check_crypto_key_type(loaded_key)
981 return cast("AllowedOKPKeys", loaded_key)
983 def sign(
984 self, msg: str | bytes, key: Ed25519PrivateKey | Ed448PrivateKey
985 ) -> bytes:
986 """
987 Sign a message ``msg`` using the EdDSA private key ``key``
988 :param str|bytes msg: Message to sign
989 :param Ed25519PrivateKey}Ed448PrivateKey key: A :class:`.Ed25519PrivateKey`
990 or :class:`.Ed448PrivateKey` isinstance
991 :return bytes signature: The signature, as bytes
992 """
993 msg_bytes = msg.encode("utf-8") if isinstance(msg, str) else msg
994 signature: bytes = key.sign(msg_bytes)
995 return signature
997 def verify(
998 self, msg: str | bytes, key: AllowedOKPKeys, sig: str | bytes
999 ) -> bool:
1000 """
1001 Verify a given ``msg`` against a signature ``sig`` using the EdDSA key ``key``
1003 :param str|bytes sig: EdDSA signature to check ``msg`` against
1004 :param str|bytes msg: Message to sign
1005 :param Ed25519PrivateKey|Ed25519PublicKey|Ed448PrivateKey|Ed448PublicKey key:
1006 A private or public EdDSA key instance
1007 :return bool verified: True if signature is valid, False if not.
1008 """
1009 try:
1010 msg_bytes = msg.encode("utf-8") if isinstance(msg, str) else msg
1011 sig_bytes = sig.encode("utf-8") if isinstance(sig, str) else sig
1013 public_key = (
1014 key.public_key()
1015 if isinstance(key, (Ed25519PrivateKey, Ed448PrivateKey))
1016 else key
1017 )
1018 public_key.verify(sig_bytes, msg_bytes)
1019 return True # If no exception was raised, the signature is valid.
1020 except InvalidSignature:
1021 return False
1023 @overload
1024 @staticmethod
1025 def to_jwk(key: AllowedOKPKeys, as_dict: Literal[True]) -> JWKDict: ...
1027 @overload
1028 @staticmethod
1029 def to_jwk(key: AllowedOKPKeys, as_dict: Literal[False] = False) -> str: ...
1031 @staticmethod
1032 def to_jwk(key: AllowedOKPKeys, as_dict: bool = False) -> JWKDict | str:
1033 if isinstance(key, (Ed25519PublicKey, Ed448PublicKey)):
1034 x = key.public_bytes(
1035 encoding=Encoding.Raw,
1036 format=PublicFormat.Raw,
1037 )
1038 crv = "Ed25519" if isinstance(key, Ed25519PublicKey) else "Ed448"
1040 obj = {
1041 "x": base64url_encode(force_bytes(x)).decode(),
1042 "kty": "OKP",
1043 "crv": crv,
1044 }
1046 if as_dict:
1047 return obj
1048 else:
1049 return json.dumps(obj)
1051 if isinstance(key, (Ed25519PrivateKey, Ed448PrivateKey)):
1052 d = key.private_bytes(
1053 encoding=Encoding.Raw,
1054 format=PrivateFormat.Raw,
1055 encryption_algorithm=NoEncryption(),
1056 )
1058 x = key.public_key().public_bytes(
1059 encoding=Encoding.Raw,
1060 format=PublicFormat.Raw,
1061 )
1063 crv = "Ed25519" if isinstance(key, Ed25519PrivateKey) else "Ed448"
1064 obj = {
1065 "x": base64url_encode(force_bytes(x)).decode(),
1066 "d": base64url_encode(force_bytes(d)).decode(),
1067 "kty": "OKP",
1068 "crv": crv,
1069 }
1071 if as_dict:
1072 return obj
1073 else:
1074 return json.dumps(obj)
1076 raise InvalidKeyError("Not a public or private key")
1078 @staticmethod
1079 def from_jwk(jwk: str | JWKDict) -> AllowedOKPKeys:
1080 try:
1081 if isinstance(jwk, str):
1082 obj = json.loads(jwk)
1083 elif isinstance(jwk, dict):
1084 obj = jwk
1085 else:
1086 raise ValueError
1087 except ValueError:
1088 raise InvalidKeyError("Key is not valid JSON") from None
1090 if obj.get("kty") != "OKP":
1091 raise InvalidKeyError("Not an Octet Key Pair")
1093 curve = obj.get("crv")
1094 if curve != "Ed25519" and curve != "Ed448":
1095 raise InvalidKeyError(f"Invalid curve: {curve}")
1097 if "x" not in obj:
1098 raise InvalidKeyError('OKP should have "x" parameter')
1099 x = base64url_decode(obj.get("x"))
1101 try:
1102 if "d" not in obj:
1103 if curve == "Ed25519":
1104 return Ed25519PublicKey.from_public_bytes(x)
1105 return Ed448PublicKey.from_public_bytes(x)
1106 d = base64url_decode(obj.get("d"))
1107 if curve == "Ed25519":
1108 return Ed25519PrivateKey.from_private_bytes(d)
1109 return Ed448PrivateKey.from_private_bytes(d)
1110 except ValueError as err:
1111 raise InvalidKeyError("Invalid key parameter") from err