404Forbiden

✏️ Editing: jwk.py

"""JSON Web Key."""
import abc
import json
import logging
import math

from typing import Dict, Optional, Sequence, Type, Union

import cryptography.exceptions
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric import rsa

from josepy import errors, json_util, util

logger = logging.getLogger(__name__)

class JWK(json_util.TypedJSONObjectWithFields, metaclass=abc.ABCMeta):
    # pylint: disable=too-few-public-methods
    """JSON Web Key."""
    type_field_name = 'kty'
    TYPES: Dict[str, Type] = {}
    cryptography_key_types: Sequence[Type] = ()
    """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,
    }

def thumbprint(self, hash_function=hashes.SHA256):
        """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())
        return digest.finalize()

@abc.abstractmethod
    def public_key(self):  # pragma: no cover
        """Generate JWK with public key.

For symmetric cryptosystems, this would return ``self``.

"""
        raise NotImplementedError()

@classmethod
    def _load_cryptography_key(cls, data, password=None, backend=None):
        backend = default_backend() if backend is None else backend
        exceptions = {}

# private key?
        for loader in (serialization.load_pem_private_key,
                       serialization.load_der_private_key):
            try:
                return loader(data, password, backend)
            except (ValueError, TypeError,
                    cryptography.exceptions.UnsupportedAlgorithm) as error:
                exceptions[loader] = error

# public key?
        for loader in (serialization.load_pem_public_key,
                       serialization.load_der_public_key):
            try:
                return loader(data, backend)
            except (ValueError,
                    cryptography.exceptions.UnsupportedAlgorithm) as error:
                exceptions[loader] = error

# no luck
        raise errors.Error('Unable to deserialize key: {0}'.format(exceptions))

@classmethod
    def load(cls, data, password=None, backend=None):
        """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)

def fields_to_partial_json(self):
        # 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):
        return cls(key=json_util.decode_b64jose(jobj['k']))

def public_key(self):
        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')

def __init__(self, *args, **kwargs):
        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):
        """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):
        """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):
        return type(self)(key=self.key.public_key())

@classmethod
    def fields_from_json(cls, jobj):
        # pylint: disable=invalid-name
        n, e = (cls._decode_param(jobj[x]) for x in ('n', 'e'))
        public_numbers = rsa.RSAPublicNumbers(e=e, n=n)
        if 'd' not in jobj:  # public key
            key = public_numbers.public_key(default_backend())
        else:  # 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(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):
        # pylint: disable=protected-access
        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.ComparableRSAKey`

"""
    typ = 'EC'
    __slots__ = ('key',)
    cryptography_key_types = (
        ec.EllipticCurvePublicKey, ec.EllipticCurvePrivateKey)
    required = ('crv', JWK.type_field_name, 'x', 'y')

def __init__(self, *args, **kwargs):
        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, length):
        """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, name, valid_length):
        """Decode Base64urlUInt."""
        try:
            binary = json_util.decode_b64jose(data)
            if len(binary) != valid_length:
                raise errors.DeserializationError(
                    'Expected parameter "{name}" to be {valid_lengths} bytes '
                    'after base64-decoding; got {length} bytes instead'.format(
                        name=name, valid_lengths=valid_length, length=len(binary))
                )
            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):
        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):
        # 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):
        if isinstance(curve, ec.SECP256R1):
            return 32
        elif isinstance(curve, ec.SECP384R1):
            return 48
        elif isinstance(curve, ec.SECP521R1):
            return 66

def fields_to_partial_json(self):
        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):
        # pylint: disable=invalid-name
        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)
        if 'd' not in jobj:  # public key
            key = public_numbers.public_key(default_backend())
        else:  # 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):
        # 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)

Sid Gifari From Gifari Industries - BD Cyber Security Team