Python cryptography.hazmat.primitives.serialization 模块，NoEncryption() 实例源码

def test_key_exists(self, pem_path):
        """
        When we get the client key and the key file already exists, the file
        should be read and the existing key returned.
        """
        raw_key = generate_private_key(u'rsa')
        expected_key = JWKRSA(key=raw_key)

        pem_file = pem_path.child(u'client.key')
        pem_file.setContent(raw_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()
        ))

        actual_key = maybe_key(pem_path)
        assert_that(actual_key, Equals(expected_key))

def generate_RSA(bits=4096):
    new_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=bits,
        backend=default_backend()
    )
    private_key = new_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    public_key = new_key.public_key().public_bytes(
        encoding=serialization.Encoding.OpenSSH,
        format=serialization.PublicFormat.OpenSSH
    )
    return private_key, public_key

def generate_RSA(bits=4096):
    new_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=bits,
        backend=default_backend()
    )
    private_key = new_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    public_key = new_key.public_key().public_bytes(
        encoding=serialization.Encoding.OpenSSH,
        format=serialization.PublicFormat.OpenSSH
    )
    return private_key, public_key

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 create_rsa_key():
        # generate private key
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
            backend=default_backend()
        )

        # output PEM encoded version of private key
        privateKey = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        )

        return privateKey.decode()

def rsa_key_fingerprint(key):
    """ Return the SHA256 fingerprint of an RSA public or private key in url safe BASE64 """
    fp = hashes.Hash(algorithm=hashes.SHA256(), backend=default_backend())

    if isinstance(key, rsa.RSAPrivateKey):
        fp.update(key.private_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ))
    elif isinstance(key, rsa.RSAPublicKey):
        fp.update(key.public_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PublicFormat.PKCS1
        ))

    return urlsafe_b64encode(fp.finalize()).decode()

def dump_pem(key_or_crt):
    if isinstance(key_or_crt, rsa.RSAPrivateKey):
        return key_or_crt.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        )
    elif isinstance(key_or_crt, rsa.RSAPublicKey):
        return key_or_crt.public_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PublicFormat.PKCS1
        )
    else:
        return key_or_crt.public_bytes(
            encoding=serialization.Encoding.PEM
        )

def generate_key():
    # generate private/public key pair
    key = rsa.generate_private_key(backend=default_backend(), public_exponent=65537, key_size=2048)

    # get public key in OpenSSH format
    public_key = key.public_key().public_bytes(
        serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH
    )

    # get private key in PEM container format
    pem = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.NoEncryption()
    )

    return KeyPair(pem.decode('utf-8'), public_key.decode('utf-8'))

def store_keys(privkey, privkey_file, pubkey_file):
    privkey_serial = privkey.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    pubkey = privkey.public_key()
    pubkey_serial = pubkey.public_bytes(
        encoding=serialization.Encoding.OpenSSH,
        format=serialization.PublicFormat.OpenSSH
    )
    with os.fdopen(os.open(privkey_file, os.O_WRONLY | os.O_CREAT, 0o600), 'wb') as privkey_out:
        privkey_out.write(privkey_serial)
    try:
        with os.fdopen(os.open(pubkey_file, os.O_WRONLY | os.O_CREAT, 0o644), 'wb') as pubkey_out:
            pubkey_out.write(pubkey_serial)
    except:
        os.remove(privkey_file)
        raise

def rsa_private_key_to_pem(rsa_secret_key):
        """
        Converts and returns RSA key from cryptography lib instance into RSA key
            PEM (Privacy Enhanced Mail) format.

        @developer: tnanoba

        @link @link https://cryptography.io/en/latest/hazmat/primitives/asymmetric/rsa/

        :param rsa_secret_key: object
        :return: bytes
        """
        return rsa_secret_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        )

def generate_ssh_key():
    logging.info('NOTICE! Generating a new private/public key combination, be AWARE!')

    key = rsa.generate_private_key(
    backend=crypto_default_backend(),
    public_exponent=65537,
    key_size=2048
    )
    private_key = key.private_bytes(
        crypto_serialization.Encoding.PEM,
        crypto_serialization.PrivateFormat.PKCS8,
        crypto_serialization.NoEncryption())
    public_key = key.public_key().public_bytes(
        crypto_serialization.Encoding.OpenSSH,
        crypto_serialization.PublicFormat.OpenSSH
    )

    with open("{}/keys/private.key".format(SETTINGS['RESOURCESDIR']), 'w') as content_file:
        content_file.write(private_key)
        chmod("{}/keys/private.key".format(SETTINGS['RESOURCESDIR']), 0600)
    with open("{}/keys/public.key".format(SETTINGS['RESOURCESDIR']), 'w') as content_file:
        content_file.write(public_key)
    return public_key

def generate_account(name='default'):
    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048,
        backend=default_backend()
    )
    private_serialized = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    public_serialized = private_key.public_key().public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    public_lines = public_serialized.splitlines()
    public_shrunk = ''
    for line in range(1, len(public_lines) - 1):
        public_shrunk += public_lines[line].strip('\n')

    return {
        'name': name,
        'private_key': private_serialized,
        'public_key': public_shrunk
    }

def gen_key_pair():
    """
    This method generates the public and private key pair.
    The public key format is OpenSSH and private key format is PEM container
    :return:
    """

    private_key = rsa.generate_private_key(backend=default_backend(),
                                           public_exponent=65537,
                                           key_size=2048)

    public_key_str = private_key.public_key(). \
        public_bytes(serialization.Encoding.OpenSSH,
                     serialization.PublicFormat.OpenSSH).decode('utf-8')

    private_key_str = private_key. \
        private_bytes(encoding=serialization.Encoding.PEM,
                      format=serialization.PrivateFormat.TraditionalOpenSSL,
                      encryption_algorithm=serialization.NoEncryption()
                      ).decode('utf-8')

    return (private_key_str, public_key_str)

def create_client_cert() -> bytes:
    ca_key = gen_private_key()
    ca_cert = gen_certificate(
        ca_key,
        'certificate_authority',
    )
    client_key = gen_private_key()
    client_cert = gen_certificate(
        client_key,
        'client',
        issuer='certificate_authority',
        sign_key=ca_key,
    )
    return client_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.NoEncryption(),
    ) + client_cert.public_bytes(
        encoding=serialization.Encoding.PEM,
    )

def load_or_create_client_key(pem_path):
    """
    Load the client key from a directory, creating it if it does not exist.

    .. note:: The client key that will be created will be a 2048-bit RSA key.

    :type pem_path: ``twisted.python.filepath.FilePath``
    :param pem_path: The certificate directory
        to use, as with the endpoint.
    """
    acme_key_file = pem_path.asTextMode().child(u'client.key')
    if acme_key_file.exists():
        key = serialization.load_pem_private_key(
            acme_key_file.getContent(),
            password=None,
            backend=default_backend())
    else:
        key = generate_private_key(u'rsa')
        acme_key_file.setContent(
            key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()))
    return JWKRSA(key=key)

def _test_issue(self, name):
        def got_cert(certr):
            key_bytes = self.issued_key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption())
            FilePath('issued.crt').setContent(certr.body)
            FilePath('issued.key').setContent(key_bytes)
            return certr

        action = start_action(action_type=u'integration:issue')
        with action.context():
            self.issued_key = generate_private_key('rsa')
            csr = csr_for_names([name], self.issued_key)
            return (
                DeferredContext(
                    self.client.request_issuance(CertificateRequest(csr=csr)))
                .addCallback(got_cert)
                .addActionFinish())

def private_pem(self, password=None):
        """
        Returns the private key PEM. This is a base64 format with delimiters.

        This function returns None if the private pem information could
        not be acquired.
        """
        if not isinstance(self.private_key, RSAPrivateKey):
            return None

        if password is None:
            password = self.password

        if password:
            return self.private_key.private_bytes(
               encoding=serialization.Encoding.PEM,
               format=serialization.PrivateFormat.PKCS8,
               encryption_algorithm=serialization
                       .BestAvailableEncryption(password)
            )

        return self.private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        )

def maybe_key(pem_path):
    """
    Set up a client key if one does not exist already.

    https://gist.github.com/glyph/27867a478bb71d8b6046fbfb176e1a33#file-local-certs-py-L32-L50

    :type pem_path: twisted.python.filepath.FilePath
    :param pem_path:
        The path to the certificate directory to use.
    """
    acme_key_file = pem_path.child(u'client.key')
    if acme_key_file.exists():
        key = serialization.load_pem_private_key(
            acme_key_file.getContent(),
            password=None,
            backend=default_backend()
        )
    else:
        key = generate_private_key(u'rsa')
        acme_key_file.setContent(
            key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()
            )
        )
    return jose.JWKRSA(key=key)

def generate_wildcard_pem_bytes():
    """
    Generate a wildcard (subject name '*') self-signed certificate valid for
    10 years.

    https://cryptography.io/en/latest/x509/tutorial/#creating-a-self-signed-certificate

    :return: Bytes representation of the PEM certificate data
    """
    key = generate_private_key(u'rsa')
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, u'*')])
    cert = (
        x509.CertificateBuilder()
        .issuer_name(name)
        .subject_name(name)
        .not_valid_before(datetime.today() - timedelta(days=1))
        .not_valid_after(datetime.now() + timedelta(days=3650))
        .serial_number(int(uuid.uuid4()))
        .public_key(key.public_key())
        .sign(
            private_key=key,
            algorithm=hashes.SHA256(),
            backend=default_backend())
        )

    return b''.join((
        key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()),
        cert.public_bytes(serialization.Encoding.PEM)
    ))

def _save_state(self):
        """ Persistent user state. """
        try:
            state = {
                'name': self.name, 'org': self.org, 'roles': self.roles,
                'affiliation': self.affiliation, 'account': self.account,
                'enrollment_secret': self.enrollment_secret,
                'msp_id': self.msp_id
            }

            if self.enrollment:
                enrollment = {
                    'private_key':
                        self.enrollment.private_key.private_bytes(
                            encoding=serialization.Encoding.PEM,
                            format=serialization.PrivateFormat.PKCS8,
                            encryption_algorithm=serialization.NoEncryption()
                        ),
                    'cert': self.enrollment.cert
                }

                state['enrollment'] = enrollment

            self._state_store.set_value(
                self._state_store_key,
                binascii.hexlify(pickle.dumps(state)).decode("utf-8"))
        except Exception as e:
            raise IOError("Cannot serialize the user", e)

def generate_keypair(key_out_path):
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=4096, backend=default_backend())
    pem = private_key.private_bytes(encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption())
    with open(key_out_path, 'wb') as outf:
        print >>outf, pem

def save_key(pk, filename):
    pem = pk.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.NoEncryption()
    )
    with open(filename, 'wb') as pem_out:
        pem_out.write(pem)

def create_random_privkey():
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
            backend=default_backend()
        )

        return private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ).decode()

def create_new_key_pair(self, save_to_config=True):
        """Creates a new public/private key pair and saves them to the config file

        :return: Prints out a success message
        """
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
            backend=default_backend()
        )

        private_key_decoded = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ).decode()

        public_key_decoded = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        ).decode()

        print('New key pair was created')

        if save_to_config:
            self.config.set('key_private', private_key_decoded)
            self.config.set('key_public', public_key_decoded)
        else:
            print('\tNew Private Key: %s' % private_key_decoded)
            print('\tNew Public Key:  %s' % public_key_decoded)

        return private_key_decoded, public_key_decoded

def generate_key_pair(cls):
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=3072,
            backend=default_backend()
        )
        return AsymmetricKeyPair(
                private_key.private_bytes(encryption_algorithm=serialization.NoEncryption(),
                                          encoding=serialization.Encoding.DER,
                                          format=serialization.PrivateFormat.TraditionalOpenSSL),
                private_key.public_key().public_bytes(encoding=serialization.Encoding.DER,
                                                      format=serialization.PublicFormat.PKCS1))

def generate_RSA(bits=4096):
    # public_exponent set to 655537 is what pyCA recommends
    new_key = rsa.generate_private_key(public_exponent=65537,
                                       key_size=bits,
                                       backend=default_backend())
    # we strip trailing space for 1:1 compat with previous implementation
    private_key = new_key.private_bytes(
        encoding=Encoding.PEM,
        format=PrivateFormat.PKCS8,
        encryption_algorithm=NoEncryption())
    public_key = new_key.public_key().public_bytes(encoding=Encoding.OpenSSH,
                                                   format=PublicFormat.OpenSSH)
    return private_key, public_key

def generate_RSA(bits=4096):
    # public_exponent set to 655537 is what pyCA recommends
    new_key = rsa.generate_private_key(public_exponent=65537,
                                       key_size=bits,
                                       backend=default_backend())
    # we strip trailing space for 1:1 compat with previous implementation
    private_key = new_key.private_bytes(
        encoding=Encoding.PEM,
        format=PrivateFormat.PKCS8,
        encryption_algorithm=NoEncryption())
    public_key = new_key.public_key().public_bytes(encoding=Encoding.OpenSSH,
                                                   format=PublicFormat.OpenSSH)
    return private_key, public_key

def dump_private_key_pem(private_key):
        """Dump private key pem."""
        pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()
        )
        return pem.decode()

def pem_block(self):
        return self.key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption()
        )

def pem_block(self):
        return self.key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption()
        )

def get_unencrypted_key(self):
        return self.get_key().private_bytes(encoding=serialization.Encoding.PEM,
                                            format=serialization.PrivateFormat.TraditionalOpenSSL,
                                            encryption_algorithm=serialization.NoEncryption())

def _get_private_key_string(self):
        return self._private_key.private_bytes(
            encoding=crypto_serialization.Encoding.PEM,
            format=crypto_serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=crypto_serialization.NoEncryption(),
        )

def generate_rsa_keypair(key_size=2048):
    key = rsa.generate_private_key(public_exponent=65537, key_size=key_size, backend=default_backend())
    public_key = key.public_key()
    public = public_key.public_bytes(encoding=serialization.Encoding.PEM,
                                     format=serialization.PublicFormat.SubjectPublicKeyInfo)
    private = key.private_bytes(encoding=serialization.Encoding.PEM,
                                format=serialization.PrivateFormat.TraditionalOpenSSL,
                                encryption_algorithm=serialization.NoEncryption())
    return public, private

def create_ca_certificate(cn, key_size=4096, certify_days=365):
    key = rsa.generate_private_key(public_exponent=65537, key_size=key_size, backend=default_backend())
    key_id = x509.SubjectKeyIdentifier.from_public_key(key.public_key())

    subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])

    now = datetime.datetime.utcnow()
    serial = x509.random_serial_number()
    cert = x509.CertificateBuilder() \
        .subject_name(subject) \
        .issuer_name(issuer) \
        .public_key(key.public_key()) \
        .serial_number(serial) \
        .not_valid_before(now) \
        .not_valid_after(now + datetime.timedelta(days=certify_days)) \
        .add_extension(key_id, critical=False) \
        .add_extension(x509.AuthorityKeyIdentifier(key_id.digest,
                                                   [x509.DirectoryName(issuer)],
                                                   serial),
                       critical=False) \
        .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) \
        .add_extension(x509.KeyUsage(digital_signature=True,
                                     content_commitment=False,
                                     key_encipherment=False,
                                     data_encipherment=False,
                                     key_agreement=False,
                                     key_cert_sign=True,
                                     crl_sign=True,
                                     encipher_only=False,
                                     decipher_only=False),
                       critical=True) \
        .sign(key, hashes.SHA256(), default_backend())

    cert = cert.public_bytes(serialization.Encoding.PEM)
    key = key.private_bytes(encoding=serialization.Encoding.PEM,
                            format=serialization.PrivateFormat.TraditionalOpenSSL,
                            encryption_algorithm=serialization.NoEncryption())
    return cert, key

def create_new_key_pair(self, save_to_config=True):
        """Creates a new public/private key pair and saves them to the config file

        :return: Prints out a success message
        """
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
            backend=default_backend()
        )

        private_key_decoded = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ).decode()

        public_key_decoded = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        ).decode()

        print('New key pair was created')

        if save_to_config:
            self.config.set('key_private', private_key_decoded)
            self.config.set('key_public', public_key_decoded)
        else:
            print('\tNew Private Key: %s' % private_key_decoded)
            print('\tNew Public Key:  %s' % public_key_decoded)

        return private_key_decoded, public_key_decoded

def create_privatekey(pkey_file):
    """ Create a private key """
    pkey = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048,
        backend=default_backend()
    )
    with open(pkey_file, 'wb') as f:
        f.write(pkey.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ))
    return pkey

def save(self, path):
        content = self._key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption())
        with openfile(path, 'wb', permissions=0600) as key_file:
            key_file.write(content)

def ask_for_password_or_no_crypto(key_path):
    # we can't use prompt's "default" and "value_proc" arguments because we monkeypatch prompt in test_wile.py
    password = click.prompt('(optional) Password for %s' % key_path, default='',
                            hide_input=True, confirmation_prompt=True, show_default=False)
    if password:
        return serialization.BestAvailableEncryption(password.encode('utf-8'))
    else:
        return serialization.NoEncryption()

def key_to_pem(self):
        "Convert a key to the PEM format."

        return self.ec.private_bytes(serialization.Encoding.PEM,
                                     serialization.PrivateFormat.TraditionalOpenSSL,
                                     serialization.NoEncryption())

def create_modhash_key_and_csr(common_name, oids=None, sans=None):
    key = _create_key(common_name)
    csr = _create_csr(common_name, key)
    modhash = _create_modhash(key)
    return (
        modhash,
        key.private_bytes(
            encoding=ENCODING[CFG.key.encoding],
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()).decode('utf-8'),
        csr.public_bytes(ENCODING[CFG.csr.encoding]).decode('utf-8'))

def create_modhash_key_and_csr(common_name, oids=None, sans=None):
    key = _create_key(common_name)
    csr = _create_csr(common_name, key)
    modhash = _create_modhash(key)
    return (
        modhash,
        key.private_bytes(
            encoding=ENCODING[CFG.key.encoding],
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()).decode('utf-8'),
        csr.public_bytes(ENCODING[CFG.csr.encoding]).decode('utf-8'))

def create_modhash_key_and_csr(common_name, oids=None, sans=None):
    key = _create_key(common_name)
    csr = _create_csr(common_name, key)
    modhash = _create_modhash(key)
    return (
        modhash,
        key.private_bytes(
            encoding=ENCODING[CFG.key.encoding],
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()).decode('utf-8'),
        csr.public_bytes(ENCODING[CFG.csr.encoding]).decode('utf-8'))

def create_modhash_key_and_csr(common_name, oids=None, sans=None):
    key = _create_key(common_name)
    csr = _create_csr(common_name, key)
    modhash = _create_modhash(key)
    return (
        modhash,
        key.private_bytes(
            encoding=ENCODING[CFG.key.encoding],
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()).decode('utf-8'),
        csr.public_bytes(ENCODING[CFG.csr.encoding]).decode('utf-8'))

def serialize_private_key(key):
    '''
    Returns the private key serialization in base64
    '''
    pem = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    return b64encode(pem).decode()

def gen(bits=1024, password=None):
        """ ??SSH RSA???

        :?? bits: ?????????: 1024
        :?? password: ?????????: None (???)
        :??: RSA??, RSA??
        """

        rsaKey = rsa.generate_private_key(
            backend = default_backend(), 
            public_exponent = 65537, 
            key_size = bits
        )

        privateKey = rsaKey.private_bytes(
            encoding = serialization.Encoding.PEM, 
            format = serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm = password and serialization.BestAvailableEncryption(str(password).encode("utf-8")) or serialization.NoEncryption()
        ).decode("utf-8")

        publicKey = rsaKey.public_key().public_bytes(
            encoding = serialization.Encoding.OpenSSH, 
            format = serialization.PublicFormat.OpenSSH
        ).decode("utf-8")

        return privateKey, publicKey

def key2bytes():
    def convert(key):
        if isinstance(key, rsa.RSAPrivateKey):
            return key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.PKCS8,
                encryption_algorithm=serialization.NoEncryption())
        elif isinstance(key, rsa.RSAPublicKey):
            der = key.public_bytes(
                encoding=serialization.Encoding.DER,
                format=serialization.PublicFormat.SubjectPublicKeyInfo)
            return base64.b64encode(der).decode(encoding='utf-8')
    return convert

def _write_private_key(self, f, key, format, password=None):
        if password is None:
            encryption = serialization.NoEncryption()
        else:
            encryption = serialization.BestAvailableEncryption(b(password))

        f.write(key.private_bytes(
            serialization.Encoding.PEM,
            format,
            encryption
        ).decode())

def generate_rsa_keypair(key_size=2048):
    """Generate an RSA keypair.
    Create new RSA keypair with an exponent of 65537. Serialize the public
    key OpenSSH format that is used by providers for specifying access keys
    Serialize the private key in the PKCS#8 (RFC 3447) format.
    Args:
        bits (int): the key length in bits.
    Returns:
        (private key, public key) 2-tuple, both unicode objects holding the
        serialized keys
    """
    crypto_backend = cryptography.hazmat.backends.default_backend()

    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=key_size,
        backend=crypto_backend)

    privkey_pem = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption())

    public_key = private_key.public_key()
    pubkey_pem = public_key.public_bytes(
        encoding=serialization.Encoding.OpenSSH,
        format=serialization.PublicFormat.OpenSSH)

    return privkey_pem, pubkey_pem

def start_fake_apns_server(port=0, database=None, lag=0):
    database = {} if database is None else database
    private_key = gen_private_key()
    certificate = gen_certificate(private_key, 'server')
    with tempfile.TemporaryDirectory() as workspace:
        key_path = os.path.join(workspace, 'key.pem')
        cert_path = os.path.join(workspace, 'cert.pem')
        with open(key_path, 'wb') as fobj:
            fobj.write(
                private_key.private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.TraditionalOpenSSL,
                    encryption_algorithm=serialization.NoEncryption(),
                )
            )
        with open(cert_path, 'wb') as fobj:
            fobj.write(
                certificate.public_bytes(
                    encoding=serialization.Encoding.PEM,
                )
            )
        ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        ssl_context.load_cert_chain(certfile=cert_path, keyfile=key_path)
        ssl_context.set_alpn_protocols(["h2"])

        fake_server = FakeServer(database, lag)

        loop = asyncio.get_event_loop()
        server = await loop.create_server(
            fake_server.create_protocol,
            '127.0.0.1',
            port,
            ssl=ssl_context
        )
        fake_server.address = server.sockets[0].getsockname()
        fake_server.server = server
        try:
            yield fake_server
        finally:
            await fake_server.stop()

def _write_private_key(self, f, key, format, password=None):
        if password is None:
            encryption = serialization.NoEncryption()
        else:
            encryption = serialization.BestAvailableEncryption(b(password))

        f.write(key.private_bytes(
            serialization.Encoding.PEM,
            format,
            encryption
        ).decode())