Python cryptography.hazmat.primitives.asymmetric.rsa 模块，RSAPublicNumbers() 实例源码

def _test_private_jwk(key):
        """
        Attempt to read in the key into a private key object
        """
        keys = json.loads(key.decode())
        public_key_numbers = rsa.RSAPublicNumbers(
            long_from_bytes(keys['keys'][0]['e']),
            long_from_bytes(keys['keys'][0]['n'])
        )
        private_key_numbers = rsa.RSAPrivateNumbers(
            long_from_bytes(keys['keys'][0]['p']),
            long_from_bytes(keys['keys'][0]['q']),
            long_from_bytes(keys['keys'][0]['d']),
            long_from_bytes(keys['keys'][0]['dp']),
            long_from_bytes(keys['keys'][0]['dq']),
            long_from_bytes(keys['keys'][0]['qi']),
            public_key_numbers
        )
        return private_key_numbers.private_key(default_backend())

def __init__(self, msg=None, data=None, filename=None, password=None,
                 key=None, file_obj=None):
        self.key = None
        if file_obj is not None:
            self._from_private_key(file_obj, password)
            return
        if filename is not None:
            self._from_private_key_file(filename, password)
            return
        if (msg is None) and (data is not None):
            msg = Message(data)
        if key is not None:
            self.key = key
        else:
            if msg is None:
                raise SSHException('Key object may not be empty')
            if msg.get_text() != 'ssh-rsa':
                raise SSHException('Invalid key')
            self.key = rsa.RSAPublicNumbers(
                e=msg.get_mpint(), n=msg.get_mpint()
            ).public_key(default_backend())

def calculate_cost(self):
        """Calculate the cost of fulfilling self condition.

        The cost of the RSA condition is the size of the modulus
        squared, divided By 64.

        Returns:
            int: Expected maximum cost to fulfill self condition.

        """
        if self.modulus is None:
            raise MissingDataError('Requires a public modulus')

        public_numbers = RSAPublicNumbers(
            PUBLIC_EXPONENT,
            int.from_bytes(self.modulus, byteorder='big'),
        )
        public_key = public_numbers.public_key(default_backend())
        modulus_bit_length = public_key.key_size
        # TODO watch out >> in Python is not the sane as JS >>>, may need to be
        # corrected. For instance see:
        # http://grokbase.com/t/python/python-list/0454t3tgaw/zero-fill-shift
        return int(math.pow(modulus_bit_length, 2)) >> RsaSha256.COST_RIGHT_SHIFT

def __init__(self, msg=None, data=None, filename=None, password=None,
                 key=None, file_obj=None):
        self.key = None
        if file_obj is not None:
            self._from_private_key(file_obj, password)
            return
        if filename is not None:
            self._from_private_key_file(filename, password)
            return
        if (msg is None) and (data is not None):
            msg = Message(data)
        if key is not None:
            self.key = key
        else:
            if msg is None:
                raise SSHException('Key object may not be empty')
            if msg.get_text() != 'ssh-rsa':
                raise SSHException('Invalid key')
            self.key = rsa.RSAPublicNumbers(
                e=msg.get_mpint(), n=msg.get_mpint()
            ).public_key(default_backend())

def __init__(self, msg=None, data=None, filename=None, password=None, key=None, file_obj=None):
        self.key = None
        if file_obj is not None:
            self._from_private_key(file_obj, password)
            return
        if filename is not None:
            self._from_private_key_file(filename, password)
            return
        if (msg is None) and (data is not None):
            msg = Message(data)
        if key is not None:
            self.key = key
        else:
            if msg is None:
                raise SSHException('Key object may not be empty')
            if msg.get_text() != 'ssh-rsa':
                raise SSHException('Invalid key')
            self.key = rsa.RSAPublicNumbers(
                e=msg.get_mpint(), n=msg.get_mpint()
            ).public_key(default_backend())

def __init__(self, msg=None, data=None, filename=None, password=None, key=None, file_obj=None):
        self.key = None
        if file_obj is not None:
            self._from_private_key(file_obj, password)
            return
        if filename is not None:
            self._from_private_key_file(filename, password)
            return
        if (msg is None) and (data is not None):
            msg = Message(data)
        if key is not None:
            self.key = key
        else:
            if msg is None:
                raise SSHException('Key object may not be empty')
            if msg.get_text() != 'ssh-rsa':
                raise SSHException('Invalid key')
            self.key = rsa.RSAPublicNumbers(
                e=msg.get_mpint(), n=msg.get_mpint()
            ).public_key(default_backend())

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

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 _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _ssh_read_next_mpint(decoded_data)
    n, rest = _ssh_read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def rsa_public_key_from_map(jwk):
    nb64url = jwk['n'].encode('utf-8')
    eb64url = jwk['e'].encode('utf-8')

    n = util.parse_rsa_modules_params(nb64url)
    e = util.parse_rsa_public_exponent_param(eb64url)
    public_key = rsa.RSAPublicNumbers(e, n).public_key(default_backend())
    return RSAPublicKey(public_key)

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 _test_public_jwk(key):
        """
        Attempt to read in the key into a key object
        """
        keys = json.loads(key.decode())
        public_key_numbers = rsa.RSAPublicNumbers(
            long_from_bytes(keys['keys'][0]['e']),
            long_from_bytes(keys['keys'][0]['n'])
        )
        return public_key_numbers.public_key(default_backend())

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def compute_rsa_private_key(cls, p, q, e, n, d):
        """
        Computes RSA private key based on provided RSA semi-primes
            and returns cryptography lib instance.

        @developer: tnanoba

        :param p: int
        :param q: int
        :param e: int
        :param n: int
        :param d: int
        :return: object
        """

        # Computes: d % (p - 1)
        dmp1 = rsa.rsa_crt_dmp1(d, p)

        # Computes: d % (q - 1)
        dmq1 = rsa.rsa_crt_dmq1(d, q)

        # Modular inverse q of p, (q ^ -1 mod p)
        iqmp = rsa.rsa_crt_iqmp(p, q)

        public_numbers = rsa.RSAPublicNumbers(e, n)

        return rsa.RSAPrivateNumbers(p, q, d, dmp1, dmq1, iqmp, public_numbers).private_key(default_backend())

def compute_rsa_public_key(cls, e, n):
        """
        Computes RSA public key based on provided RSA semi-primes (e, n)
                and returns cryptography lib instance.
        :return: object
        """

        public_numbers = rsa.RSAPublicNumbers(e, n)

        return public_numbers.public_key(default_backend())

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _ssh_read_next_mpint(decoded_data)
    n, rest = _ssh_read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def validate(self, message):
        """Verify the signature of self RSA fulfillment.

        The signature of self RSA fulfillment is verified against the
        provided message and the condition's public modulus.

        Args:
            message (bytes): Message to verify.

        Returns:
            bool: Whether self fulfillment is valid.

        """
        if not isinstance(message, bytes):
            raise Exception(
                'Message must be provided as bytes, was: ' + message)

        public_numbers = RSAPublicNumbers(
            PUBLIC_EXPONENT,
            int.from_bytes(self.modulus, byteorder='big'),
        )
        public_key = public_numbers.public_key(default_backend())
        verifier = public_key.verifier(
            self.signature,
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=SALT_LENGTH,
            ),
            hashes.SHA256()
        )
        verifier.update(message)
        try:
            verifier.verify()
        except InvalidSignature as exc:
            raise ValidationError('Invalid RSA signature') from exc

        return True

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _ssh_read_next_mpint(decoded_data)
    n, rest = _ssh_read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _ssh_read_next_mpint(decoded_data)
    n, rest = _ssh_read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_ssh_rsa_public_key(key_type, decoded_data, backend):
    e, rest = _read_next_mpint(decoded_data)
    n, rest = _read_next_mpint(rest)

    if rest:
        raise ValueError('Key body contains extra bytes.')

    return rsa.RSAPublicNumbers(e, n).public_key(backend)

def _load_public_key(self, e, n):
        def to_int(x):
            bs = base64.urlsafe_b64decode(x + "==")
            return int.from_bytes(bs, byteorder="big")

        ei = to_int(e)
        ni = to_int(n)
        numbers = RSAPublicNumbers(ei, ni)
        public_key = numbers.public_key(backend=default_backend())
        pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )
        return pem

def jwk_to_pubkey(jwk):
    if jwk['kty'] == 'oct':
        return jwk_to_bytes(jwk['k'])
    elif jwk['kty'] == 'EC':
        numbers = ec.EllipticCurvePublicNumbers(jwk_to_uint(jwk['x']),
                                                jwk_to_uint(jwk['y']),
                                                jwk_to_curve(jwk['crv']))
        return numbers.public_key(backend)
    elif jwk['kty'] == 'RSA':
        numbers = rsa.RSAPublicNumbers(jwk_to_uint(jwk['e']),
                                       jwk_to_uint(jwk['n']))
        return numbers.public_key(backend)

def __init__(self, ssh_public_key):
        """
        Extracts the useful RSA Public Key information from an SSH Public Key file.
        :param ssh_public_key: SSH Public Key file contents. (i.e. 'ssh-rsa AAAAB3NzaC1yc2E..').
        """
        super(RSAPublicKey, self).__init__()

        self.type = SSHPublicKeyType.RSA

        split_ssh_public_key = ssh_public_key.split(' ')
        split_key_len = len(split_ssh_public_key)

        # is there a key comment at the end?
        if split_key_len > 2:
            self.key_comment = ' '.join(split_ssh_public_key[2:])
        else:
            self.key_comment = ''

        public_key = serialization.load_ssh_public_key(ssh_public_key, default_backend())
        ca_pub_numbers = public_key.public_numbers()
        if not isinstance(ca_pub_numbers, RSAPublicNumbers):
            raise TypeError("Public Key is not the correct type or format")

        self.key_size = public_key.key_size
        self.e = ca_pub_numbers.e
        self.n = ca_pub_numbers.n

        key_bytes = base64.b64decode(split_ssh_public_key[1])
        fingerprint = hashlib.md5(key_bytes).hexdigest()

        self.fingerprint = 'RSA ' + ':'.join(
            fingerprint[i:i + 2] for i in range(0, len(fingerprint), 2))

def _fromString_BLOB(cls, blob):
        """
        Return a public key object corresponding to this public key blob.
        The format of a RSA public key blob is::
            string 'ssh-rsa'
            integer e
            integer n

        The format of a DSA public key blob is::
            string 'ssh-dss'
            integer p
            integer q
            integer g
            integer y

        The format of ECDSA-SHA2-* public key blob is::
            string 'ecdsa-sha2-[identifier]'
            integer x
            integer y

            identifier is the standard NIST curve name.

        @type blob: L{bytes}
        @param blob: The key data.

        @return: A new key.
        @rtype: L{twisted.conch.ssh.keys.Key}
        @raises BadKeyError: if the key type (the first string) is unknown.
        """
        keyType, rest = common.getNS(blob)
        if keyType == b'ssh-rsa':
            e, n, rest = common.getMP(rest, 2)
            return cls(
                rsa.RSAPublicNumbers(e, n).public_key(default_backend()))
        elif keyType == b'ssh-dss':
            p, q, g, y, rest = common.getMP(rest, 4)
            return cls(
                dsa.DSAPublicNumbers(
                    y=y,
                    parameter_numbers=dsa.DSAParameterNumbers(
                        p=p,
                        q=q,
                        g=g
                    )
                ).public_key(default_backend())
            )
        elif keyType in _curveTable:
            # First we have to make an EllipticCuvePublicNumbers from the
            # provided curve and points,
            # then turn it into a public key object.
            return cls(
                ec.EllipticCurvePublicNumbers.from_encoded_point(
                      _curveTable[keyType],
                       common.getNS(rest, 2)[1]).public_key(default_backend()))
        else:
            raise BadKeyError('unknown blob type: %s' % (keyType,))

def _fromRSAComponents(cls, n, e, d=None, p=None, q=None, u=None):
        """
        Build a key from RSA numerical components.

        @type n: L{int}
        @param n: The 'n' RSA variable.

        @type e: L{int}
        @param e: The 'e' RSA variable.

        @type d: L{int} or L{None}
        @param d: The 'd' RSA variable (optional for a public key).

        @type p: L{int} or L{None}
        @param p: The 'p' RSA variable (optional for a public key).

        @type q: L{int} or L{None}
        @param q: The 'q' RSA variable (optional for a public key).

        @type u: L{int} or L{None}
        @param u: The 'u' RSA variable. Ignored, as its value is determined by
        p and q.

        @rtype: L{Key}
        @return: An RSA key constructed from the values as given.
        """
        publicNumbers = rsa.RSAPublicNumbers(e=e, n=n)
        if d is None:
            # We have public components.
            keyObject = publicNumbers.public_key(default_backend())
        else:
            privateNumbers = rsa.RSAPrivateNumbers(
                p=p,
                q=q,
                d=d,
                dmp1=rsa.rsa_crt_dmp1(d, p),
                dmq1=rsa.rsa_crt_dmq1(d, q),
                iqmp=rsa.rsa_crt_iqmp(p, q),
                public_numbers=publicNumbers,
            )
            keyObject = privateNumbers.private_key(default_backend())

        return cls(keyObject)