service.py 文件源码

def _issue_cert(self, client, server_name):
        """
        Issue a new cert for a particular name.
        """
        log.info(
            'Requesting a certificate for {server_name!r}.',
            server_name=server_name)
        key = self._generate_key()
        objects = [
            Key(key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption()))]

        def answer_and_poll(authzr):
            def got_challenge(stop_responding):
                return (
                    poll_until_valid(authzr, self._clock, client)
                    .addBoth(tap(lambda _: stop_responding())))
            return (
                answer_challenge(authzr, client, self._responders)
                .addCallback(got_challenge))

        def got_cert(certr):
            objects.append(
                Certificate(
                    x509.load_der_x509_certificate(
                        certr.body, default_backend())
                    .public_bytes(serialization.Encoding.PEM)))
            return certr

        def got_chain(chain):
            for certr in chain:
                got_cert(certr)
            log.info(
                'Received certificate for {server_name!r}.',
                server_name=server_name)
            return objects

        return (
            client.request_challenges(fqdn_identifier(server_name))
            .addCallback(answer_and_poll)
            .addCallback(lambda ign: client.request_issuance(
                CertificateRequest(
                    csr=csr_for_names([server_name], key))))
            .addCallback(got_cert)
            .addCallback(client.fetch_chain)
            .addCallback(got_chain)
            .addCallback(partial(self.cert_store.store, server_name)))