Server IP : 85.214.239.14 / Your IP : 3.144.21.237 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /usr/lib/python3/dist-packages/josepy/ |
Upload File : |
"""JSON Web Key.""" import abc import json import logging import math from typing import ( Any, Callable, Dict, Mapping, Optional, Sequence, Tuple, Type, Union, ) import cryptography.exceptions from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec, rsa import josepy.util from josepy import errors, json_util, util logger = logging.getLogger(__name__) class JWK(json_util.TypedJSONObjectWithFields, metaclass=abc.ABCMeta): """JSON Web Key.""" type_field_name = 'kty' TYPES: Dict[str, Type['JWK']] = {} cryptography_key_types: Tuple[Type[Any], ...] = () """Subclasses should override.""" required: Sequence[str] = NotImplemented """Required members of public key's representation as defined by JWK/JWA.""" _thumbprint_json_dumps_params: Dict[str, Union[Optional[int], Sequence[str], bool]] = { # "no whitespace or line breaks before or after any syntactic # elements" 'indent': None, 'separators': (',', ':'), # "members ordered lexicographically by the Unicode [UNICODE] # code points of the member names" 'sort_keys': True, } key: Any def thumbprint(self, hash_function: Callable[[], hashes.HashAlgorithm] = hashes.SHA256) -> bytes: """Compute JWK Thumbprint. https://tools.ietf.org/html/rfc7638 :returns: bytes """ digest = hashes.Hash(hash_function(), backend=default_backend()) digest.update(json.dumps( {k: v for k, v in self.to_json().items() if k in self.required}, **self._thumbprint_json_dumps_params).encode()) # type: ignore[arg-type] return digest.finalize() @abc.abstractmethod def public_key(self) -> 'JWK': # pragma: no cover """Generate JWK with public key. For symmetric cryptosystems, this would return ``self``. """ raise NotImplementedError() @classmethod def _load_cryptography_key(cls, data: bytes, password: Optional[bytes] = None, backend: Optional[Any] = None) -> Any: backend = default_backend() if backend is None else backend exceptions = {} # private key? loader_private: Any for loader_private in (serialization.load_pem_private_key, serialization.load_der_private_key): try: return loader_private(data, password, backend) except (ValueError, TypeError, cryptography.exceptions.UnsupportedAlgorithm) as error: exceptions[str(loader_private)] = error # public key? loader_public: Any for loader_public in (serialization.load_pem_public_key, serialization.load_der_public_key): try: return loader_public(data, backend) except (ValueError, cryptography.exceptions.UnsupportedAlgorithm) as error: exceptions[str(loader_public)] = error # no luck raise errors.Error('Unable to deserialize key: {0}'.format(exceptions)) @classmethod def load(cls, data: bytes, password: Optional[bytes] = None, backend: Optional[Any] = None) -> 'JWK': """Load serialized key as JWK. :param str data: Public or private key serialized as PEM or DER. :param str password: Optional password. :param backend: A `.PEMSerializationBackend` and `.DERSerializationBackend` provider. :raises errors.Error: if unable to deserialize, or unsupported JWK algorithm :returns: JWK of an appropriate type. :rtype: `JWK` """ try: key = cls._load_cryptography_key(data, password, backend) except errors.Error as error: logger.debug('Loading symmetric key, asymmetric failed: %s', error) return JWKOct(key=data) if cls.typ is not NotImplemented and not isinstance(key, cls.cryptography_key_types): raise errors.Error('Unable to deserialize {0} into {1}'.format( key.__class__, cls.__class__)) for jwk_cls in cls.TYPES.values(): if isinstance(key, jwk_cls.cryptography_key_types): return jwk_cls(key=key) raise errors.Error('Unsupported algorithm: {0}'.format(key.__class__)) @JWK.register class JWKOct(JWK): """Symmetric JWK.""" typ = 'oct' __slots__ = ('key',) required = ('k', JWK.type_field_name) key: bytes def fields_to_partial_json(self) -> Dict[str, str]: # TODO: An "alg" member SHOULD also be present to identify the # algorithm intended to be used with the key, unless the # application uses another means or convention to determine # the algorithm used. return {'k': json_util.encode_b64jose(self.key)} @classmethod def fields_from_json(cls, jobj: Mapping[str, Any]) -> 'JWKOct': return cls(key=json_util.decode_b64jose(jobj['k'])) def public_key(self) -> 'JWKOct': return self @JWK.register class JWKRSA(JWK): """RSA JWK. :ivar key: :class:`~cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey` or :class:`~cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey` wrapped in :class:`~josepy.util.ComparableRSAKey` """ typ = 'RSA' cryptography_key_types = (rsa.RSAPublicKey, rsa.RSAPrivateKey) __slots__ = ('key',) required = ('e', JWK.type_field_name, 'n') key: josepy.util.ComparableRSAKey def __init__(self, *args: Any, **kwargs: Any) -> None: if 'key' in kwargs and not isinstance( kwargs['key'], util.ComparableRSAKey): kwargs['key'] = util.ComparableRSAKey(kwargs['key']) super().__init__(*args, **kwargs) @classmethod def _encode_param(cls, data: int) -> str: """Encode Base64urlUInt. :type data: long :rtype: unicode """ length = max(data.bit_length(), 8) # decoding 0 length = math.ceil(length / 8) return json_util.encode_b64jose(data.to_bytes(byteorder="big", length=length)) @classmethod def _decode_param(cls, data: str) -> int: """Decode Base64urlUInt.""" try: binary = json_util.decode_b64jose(data) if not binary: raise errors.DeserializationError() return int.from_bytes(binary, byteorder="big") except ValueError: # invalid literal for long() with base 16 raise errors.DeserializationError() def public_key(self) -> 'JWKRSA': return type(self)(key=self.key.public_key()) @classmethod def fields_from_json(cls, jobj: Mapping[str, Any]) -> 'JWKRSA': n, e = (cls._decode_param(jobj[x]) for x in ('n', 'e')) public_numbers = rsa.RSAPublicNumbers(e=e, n=n) # public key if 'd' not in jobj: return cls(key=public_numbers.public_key(default_backend())) # private key d = cls._decode_param(jobj['d']) if ('p' in jobj or 'q' in jobj or 'dp' in jobj or 'dq' in jobj or 'qi' in jobj or 'oth' in jobj): # "If the producer includes any of the other private # key parameters, then all of the others MUST be # present, with the exception of "oth", which MUST # only be present when more than two prime factors # were used." p, q, dp, dq, qi, = all_params = tuple( jobj.get(x) for x in ('p', 'q', 'dp', 'dq', 'qi')) if tuple(param for param in all_params if param is None): raise errors.Error( 'Some private parameters are missing: {0}'.format( all_params)) p, q, dp, dq, qi = tuple( cls._decode_param(str(x)) for x in all_params) # TODO: check for oth else: # cryptography>=0.8 p, q = rsa.rsa_recover_prime_factors(n, e, d) dp = rsa.rsa_crt_dmp1(d, p) dq = rsa.rsa_crt_dmq1(d, q) qi = rsa.rsa_crt_iqmp(p, q) key = rsa.RSAPrivateNumbers( p, q, d, dp, dq, qi, public_numbers).private_key( default_backend()) return cls(key=key) def fields_to_partial_json(self) -> Dict[str, Any]: if isinstance(self.key._wrapped, rsa.RSAPublicKey): numbers = self.key.public_numbers() params = { 'n': numbers.n, 'e': numbers.e, } else: # rsa.RSAPrivateKey private = self.key.private_numbers() public = self.key.public_key().public_numbers() params = { 'n': public.n, 'e': public.e, 'd': private.d, 'p': private.p, 'q': private.q, 'dp': private.dmp1, 'dq': private.dmq1, 'qi': private.iqmp, } return {key: self._encode_param(value) for key, value in params.items()} @JWK.register class JWKEC(JWK): """EC JWK. :ivar key: :class:`~cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey` or :class:`~cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey` wrapped in :class:`~josepy.util.ComparableECKey` """ typ = 'EC' __slots__ = ('key',) cryptography_key_types = ( ec.EllipticCurvePublicKey, ec.EllipticCurvePrivateKey) required = ('crv', JWK.type_field_name, 'x', 'y') key: josepy.util.ComparableECKey def __init__(self, *args: Any, **kwargs: Any) -> None: if 'key' in kwargs and not isinstance( kwargs['key'], util.ComparableECKey): kwargs['key'] = util.ComparableECKey(kwargs['key']) super().__init__(*args, **kwargs) @classmethod def _encode_param(cls, data: int, length: int) -> str: """Encode Base64urlUInt. :type data: long :type key_size: long :rtype: unicode """ return json_util.encode_b64jose(data.to_bytes(byteorder="big", length=length)) @classmethod def _decode_param(cls, data: str, name: str, valid_length: int) -> int: """Decode Base64urlUInt.""" try: binary = json_util.decode_b64jose(data) if len(binary) != valid_length: raise errors.DeserializationError( f'Expected parameter "{name}" to be {valid_length} bytes ' f'after base64-decoding; got {len(binary)} bytes instead' ) return int.from_bytes(binary, byteorder="big") except ValueError: # invalid literal for long() with base 16 raise errors.DeserializationError() @classmethod def _curve_name_to_crv(cls, curve_name: str) -> str: if curve_name == 'secp256r1': return 'P-256' if curve_name == 'secp384r1': return 'P-384' if curve_name == 'secp521r1': return 'P-521' raise errors.SerializationError() @classmethod def _crv_to_curve(cls, crv: str) -> ec.EllipticCurve: # crv is case-sensitive if crv == 'P-256': return ec.SECP256R1() if crv == 'P-384': return ec.SECP384R1() if crv == 'P-521': return ec.SECP521R1() raise errors.DeserializationError() @classmethod def expected_length_for_curve(cls, curve: ec.EllipticCurve) -> int: if isinstance(curve, ec.SECP256R1): return 32 elif isinstance(curve, ec.SECP384R1): return 48 elif isinstance(curve, ec.SECP521R1): return 66 raise ValueError(f'Unexpected curve: {curve}') def fields_to_partial_json(self) -> Dict[str, Any]: params = {} if isinstance(self.key._wrapped, ec.EllipticCurvePublicKey): public = self.key.public_numbers() elif isinstance(self.key._wrapped, ec.EllipticCurvePrivateKey): private = self.key.private_numbers() public = self.key.public_key().public_numbers() params['d'] = private.private_value else: raise errors.SerializationError( 'Supplied key is neither of type EllipticCurvePublicKey ' 'nor EllipticCurvePrivateKey' ) params['x'] = public.x params['y'] = public.y params = { key: self._encode_param(value, self.expected_length_for_curve(public.curve)) for key, value in params.items() } params['crv'] = self._curve_name_to_crv(public.curve.name) return params @classmethod def fields_from_json(cls, jobj: Mapping[str, Any]) -> 'JWKEC': curve = cls._crv_to_curve(jobj['crv']) expected_length = cls.expected_length_for_curve(curve) x, y = (cls._decode_param(jobj[n], n, expected_length) for n in ('x', 'y')) public_numbers = ec.EllipticCurvePublicNumbers(x=x, y=y, curve=curve) # private key if 'd' not in jobj: return cls(key=public_numbers.public_key(default_backend())) # private key d = cls._decode_param(jobj['d'], 'd', expected_length) key = ec.EllipticCurvePrivateNumbers(d, public_numbers).private_key(default_backend()) return cls(key=key) def public_key(self) -> 'JWKEC': # Unlike RSAPrivateKey, EllipticCurvePrivateKey does not contain public_key() if hasattr(self.key, 'public_key'): key = self.key.public_key() else: key = self.key.public_numbers().public_key(default_backend()) return type(self)(key=key)