Python cryptography.hazmat.primitives.asymmetric.padding 模块，OAEP 实例源码

def decrypt_pk(priv_key, ciphertext):
    """
    Decrypt a b64encoded ciphertext string with the RSA private key priv_key,
    using CryptoHash() as the OAEP/MGF1 padding hash.
    Returns the plaintext.
    Decryption failures result in an exception being raised.
    """
    try:
        plaintext = priv_key.decrypt(
            b64decode(ciphertext),
            padding.OAEP(
                mgf=padding.MGF1(algorithm=CryptoHash()),
                algorithm=CryptoHash(),
                label=None
                )
            )
    except UnsupportedAlgorithm as e:
        # a failure to dencrypt someone else's data is not typically a fatal
        # error, but in this particular case, the most likely cause of this
        # error is an old cryptography library
        logging.error("Fatal error: encryption hash {} unsupported, try upgrading to cryptography >= 1.4. Exception: {}".format(
                          CryptoHash, e))
        # re-raise the exception for the caller to handle
        raise e
    return plaintext

def encrypt(cls, message, public_key, algorithm='sha1'):
        """Public key encrypt."""
        if not isinstance(message, bytes):
            message = message.encode()

        algorithm = cls.ALGORITHM_DICT.get(algorithm)

        ciphertext = public_key.encrypt(
            message,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=algorithm),
                algorithm=algorithm,
                label=None
            )
        )
        return ciphertext

def create_decryptor(private_location, public_location):
    try:
        with open(private_location, "rb") as key_file:
            private_key = serialization.load_pem_private_key(
                key_file.read(),
                password=None,
                backend=default_backend()
            )
    except FileNotFoundError:
        with open(private_location, "wb") as key_file:
            private_key = rsa.generate_private_key(
                public_exponent=65537,
                key_size=2048,
                backend=default_backend()
            )
            key_file.write(private_key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()
            ))

    with open(public_location, "wb") as public_file:
        public_key = private_key.public_key()
        pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )
        public_file.write(pem)

    def decrypt(ciphertext):
        return private_key.decrypt(
            ciphertext,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA1()),
                algorithm=hashes.SHA1(),
                label=None
            )
        )

    return decrypt

def encrypt_key_with_public_key(secret_key, public_encryption_key):
    """
    Encrypts the given secret key with the public key.

    :param bytes secret_key: the key to encrypt
    :param public_encryption_key: the public encryption key
    :type public_encryption_key: :class:`~rsa.RSAPublicKey`
    :return: the encrypted key
    :rtype: bytes
    """
    return public_encryption_key.encrypt(
        secret_key,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None))

def decrypt_with_private_key(secret_key, private_encryption_key):
    """
    Decrypts the given secret key with the private key.

    :param bytes secret_key: the secret key to decrypt
    :param private_encryption_key: the private encryption key
    :type private_encryption_key: :class:`~rsa.RSAPrivateKey`
    :return: the decrypted key
    :rtype: bytes
    """
    return private_encryption_key.decrypt(
        secret_key,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None))

def _mgf1_sha256_supported():
    wk = serialization.load_pem_private_key(
        data=VALUES['raw'][b'asym1'][EncryptionKeyType.PRIVATE],
        password=None,
        backend=default_backend()
    )
    try:
        wk.public_key().encrypt(
            plaintext=b'aosdjfoiajfoiaj;foijae;rogijaerg',
            padding=padding.OAEP(
                mgf=padding.MGF1(hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )
    except cryptography.exceptions.UnsupportedAlgorithm:
        return False
    return True

def __init__(self, encryption_type, algorithm, padding_type, padding_algorithm, padding_mgf):
        """Prepares new WrappingAlgorithm."""
        self.encryption_type = encryption_type
        self.algorithm = algorithm
        if padding_type == padding.OAEP:
            padding_args = {
                'mgf': padding_mgf(
                    algorithm=padding_algorithm()
                ),
                'algorithm': padding_algorithm(),
                'label': None
            }
        else:
            padding_args = {}
        if padding_type is not None:
            padding_type = padding_type(**padding_args)
        self.padding = padding_type

def encrypt_pk(pub_key, plaintext):
    """
    Encrypt plaintext with the RSA public key pub_key, using CryptoHash()
    as the OAEP/MGF1 padding hash.
    plaintext is limited to the size of the RSA key, minus the padding, or a
    few hundred bytes.
    Returns a b64encoded ciphertext string.
    Encryption failures result in an exception being raised.
    """
    try:
        ciphertext = pub_key.encrypt(
            plaintext,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=CryptoHash()),
                algorithm=CryptoHash(),
                label=None
                )
            )
    except UnsupportedAlgorithm as e:
        # a failure to encrypt our own data is a fatal error
        # the most likely cause of this error is an old cryptography library
        # although some newer binary cryptography libraries are linked with
        # old OpenSSL versions, to fix, check 'openssl version' >= 1.0.2, then:
        # pip install -I --no-binary cryptography cryptography
        logging.error("Fatal error: encryption hash {} unsupported, try upgrading to cryptography >= 1.4 compiled with OpenSSL >= 1.0.2. Exception: {}".format(
                          CryptoHash, e))
        # re-raise the exception for the caller to handle
        raise e
    return b64encode(ciphertext)

def encrypt_rsa(text, key):
    private_key = load_key(key)
    public_key = private_key.public_key()
    return public_key.encrypt(
        text,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA1()),
            algorithm=hashes.SHA1(),
            label=None
        )
    )

def decrypt_rsa(text, key):
    private_key = load_key(key)
    return private_key.decrypt(
        text,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA1()),
            algorithm=hashes.SHA1(),
            label=None
        )
    )

def __init__(self, key_pair):
        self.key_pair = key_pair
        self._padding = asym_padding.OAEP(
            mgf=asym_padding.MGF1(algorithm=hashes.SHA1()),
            algorithm=hashes.SHA1(),
            label=None
        )

def decrypt(cls, ciphertext, private_key, algorithm='sha1'):
        """Private key descrption."""
        algorithm = cls.ALGORITHM_DICT.get(algorithm)

        plaintext = private_key.decrypt(
            ciphertext,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=algorithm),
                algorithm=algorithm,
                label=None
            )
        )
        return plaintext.decode()

def encrypt(self, message: bytes):
        symkey = generate_sym_key()
        ciphertext = symkey.encrypt(message)
        ciphersymkey = self._hazmat_public_key.encrypt(
            symkey.key,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )
        return struct.pack(">I", len(ciphersymkey)) + ciphersymkey + ciphertext

def decrypt(self, ciphertext: bytes):
        lenciphersymkey, = struct.unpack(">I", ciphertext[:4])
        ciphersymkey = ciphertext[4:4 + lenciphersymkey]
        ciphertext = ciphertext[4 + lenciphersymkey:]

        symkey = load_sym_key(self._hazmat_private_key.decrypt(
            ciphersymkey,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        ))
        return symkey.decrypt(ciphertext)

def test_unsupported_oaep_label(self):
        private_key = RSA_KEY_512.private_key(backend)
        with pytest.raises(ValueError):
            private_key.decrypt(
                b"0" * 64,
                padding.OAEP(
                    mgf=padding.MGF1(algorithm=hashes.SHA1()),
                    algorithm=hashes.SHA1(),
                    label=b"label"
                )
            )

def test_unsupported_mgf1_hash_algorithm(self):
        private_key = RSA_KEY_512.private_key(backend)
        with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_HASH):
            private_key.decrypt(
                b"0" * 64,
                padding.OAEP(
                    mgf=padding.MGF1(algorithm=hashes.SHA256()),
                    algorithm=hashes.SHA1(),
                    label=None
                )
            )

def test_unsupported_oaep_hash_algorithm(self):
        private_key = RSA_KEY_512.private_key(backend)
        with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_HASH):
            private_key.decrypt(
                b"0" * 64,
                padding.OAEP(
                    mgf=padding.MGF1(algorithm=hashes.SHA1()),
                    algorithm=hashes.SHA256(),
                    label=None
                )
            )

def decrypt(privkey, ciphertext):
    plaintext = privkey.decrypt(
        ciphertext,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    return plaintext

def encrypt(pubkey, plaintext):
    ciphertext= pubkey.encrypt(plaintext=plaintext,
        padding=padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    return ciphertext

def wrap_rsa_key(public_key, private_key_bytes):
    # Use the Google public key to encrypt the customer private key.
    # This means that only the Google private key is capable of decrypting
    # the customer private key.
    wrapped_key = public_key.encrypt(
        private_key_bytes,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA1()),
            algorithm=hashes.SHA1(),
            label=None))
    encoded_wrapped_key = base64.b64encode(wrapped_key)
    return encoded_wrapped_key

def __encrypt_with_rsa(self, content, recipient_pk):
        """ Encrypt content with RSAES-OAEP scheme

        @developer: vsmysle

        This method handles an encryption of a *single* RSA block with a
        specified above scheme. It does not handle splitting of a header into
        several blocks. It has to be done by other method that would use this
        one only for single block encryption purpose.

        TODO: what is a maximum size of a content that can be padded and
        encrypted given a particular size of RSA key?

        :param content:         bytes content to encrypt (probably a part of
                                ASN.1 DER-encoded MPHeader block)
        :param recipient_pk:    instance of cryptography.hazmat.primitives.rsa
                                .RSAPublicKey to use for a content encryption

        :return: string encryption of an input content

        """

        # TODO: add exceptions
        self.logger.debug("rsa encryption")

        ciphertext = recipient_pk.encrypt(
            content, asym_padding.OAEP(
                mgf=asym_padding.MGF1(algorithm=SHA1()),
                algorithm=SHA1(),
                label=None
            )
        )
        self.logger.info("encrypted")
        return ciphertext

def __decrypt_with_rsa(self, content, user_sk):
        """ Decrypt RSAES-OAEP encrypted content (single block)

        @developer: vsmysle

        This method decrypts a single RSA ciphertext block only

        :param content: bytes content to decrypt
        :param user_sk: instance of cryptography.hazmat.primitives.rsa
                        .RSAPrivateKey to use for a decryption

        :return: string decryption of an input content

        """
        # TODO: add exceptions

        self.logger.debug("rsa decryption")
        try:
            plaintext = user_sk.decrypt(
                content, asym_padding.OAEP(
                    mgf=asym_padding.MGF1(algorithm=SHA1()),
                    algorithm=SHA1(),
                    label=None
                )
            )
        except InvalidKey:
            self.logger.warning("Invalid key!")
            return
        return plaintext

def encrypt(self, message):
        if self._value is None:
            raise ValueError("Can't Encrypt with empty key.")

        try:
            return self._value.encrypt(
                    message,
                    padding.OAEP(
                            mgf=padding.MGF1(algorithm=hashes.SHA1()),
                            algorithm=hashes.SHA1(),
                            label=None))
        except ValueError as e:
            raise CipherError(e)

def decrypt(self, message):
        if self._value is None:
            raise ValueError("Can't Decrypt with empty key.")

        try:
            return self._value.decrypt(
                    message,
                    padding.OAEP(
                            mgf=padding.MGF1(algorithm=hashes.SHA1()),
                            algorithm=hashes.SHA1(),
                            label=None))
        except ValueError as e:
            raise CipherError(e)

def create_decryptor(private_location, public_location):
    try:
        with open(private_location, "rb") as key_file:
            private_key = serialization.load_pem_private_key(
                key_file.read(),
                password=None,
                backend=default_backend()
            )
    except FileNotFoundError:
        with open(private_location, "wb") as key_file:
            private_key = rsa.generate_private_key(
                public_exponent=65537,
                key_size=2048,
                backend=default_backend()
            )
            key_file.write(private_key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()
            ))

    with open(public_location, "wb") as public_file:
        public_key = private_key.public_key()
        pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )
        public_file.write(pem)

    def decrypt(ciphertext):
        return private_key.decrypt(
            ciphertext,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA1()),
                algorithm=hashes.SHA1(),
                label=None
            )
        )

    return decrypt

def decrypt(self, C, t=None, h=None, mgf=None, L=None):
        """
        Decrypt ciphertext 'C' using 't' decryption scheme where 't' can be:

        - None: the ciphertext 'C' is directly applied the RSADP decryption
                primitive, as described in PKCS#1 v2.1, i.e. RFC 3447
                Sect 5.1.2. Simply, put the message undergo a modular
                exponentiation using the private key. Additionnal method
                parameters are just ignored.

        - 'pkcs': the ciphertext 'C' is applied RSAES-PKCS1-V1_5-DECRYPT
                decryption scheme as described in section 7.2.2 of RFC 3447.
                In that context, other parameters ('h', 'mgf', 'l') are not
                used.

        - 'oaep': the ciphertext 'C' is applied the RSAES-OAEP-DECRYPT decryption
                scheme, as described in PKCS#1 v2.1, i.e. RFC 3447 Sect
                7.1.2. In that context,

                o 'h' parameter provides the name of the hash method to use.
                  Possible values are "md2", "md4", "md5", "sha1", "tls",
                  "sha224", "sha256", "sha384" and "sha512". if none is provided,
                  sha1 is used by default.

                o 'mgf' is the mask generation function. By default, mgf
                  is derived from the provided hash function using the
                  generic MGF1 (see pkcs_mgf1() for details).

                o 'L' is the optional label to be associated with the
                  message. If not provided, the default value is used, i.e
                  the empty string. No check is done on the input limitation
                  of the hash function regarding the size of 'L' (for
                  instance, 2^61 - 1 for SHA-1). You have been warned.        
        """
        if t is None:
            C = pkcs_os2ip(C)
            c = self._rsadp(C)
            l = int(math.ceil(math.log(c, 2) / 8.)) # Hack
            return pkcs_i2osp(c, l)

        elif t == "pkcs":
            return self._rsaes_pkcs1_v1_5_decrypt(C)

        elif t == "oaep":
            return self._rsaes_oaep_decrypt(C, h, mgf, L)

        else:
            warning("Key.decrypt(): Unknown decryption type (%s) provided" % t)
            return None

    ### Below are signature related methods. Verification ones are inherited from
    ### PubKey