Python cryptography.x509 模块，CertificateSigningRequestBuilder() 实例源码

def create_csr(key, domains, must_staple=False):
    """
    Creates a CSR in DER format for the specified key and domain names.
    """
    assert domains
    name = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, domains[0]),
    ])
    san = x509.SubjectAlternativeName([x509.DNSName(domain) for domain in domains])
    csr = x509.CertificateSigningRequestBuilder().subject_name(name) \
        .add_extension(san, critical=False)
    if must_staple:
        ocsp_must_staple = x509.TLSFeature(features=[x509.TLSFeatureType.status_request])
        csr = csr.add_extension(ocsp_must_staple, critical=False)
    csr = csr.sign(key, hashes.SHA256(), default_backend())
    return export_csr_for_acme(csr)

def _check_or_add_cert(self, name, domain, key, authorization):
        if 'certificate' in domain:
            return domain['certificate']

        self._log('domain:{}: generating CSR...', name)
        builder = x509.CertificateSigningRequestBuilder()
        builder = builder.subject_name(x509.Name([
            x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, name),
        ]))
        csr = builder.sign(key, hashes.SHA256(), backend)
        self._log('domain:{}: done', name)

        self._log('domain:{}: requesting certificate...', name)
        certificate = self._client.new_certificate(self._key, csr)
        domain['certificate'] = certificate
        self._write_config()
        self._log('domain:{}: done: {}', name, certificate)

        return certificate

def generate_csr(self, private_key, subject_name, extensions=None):
        """Generate certificate signing request.

        Args:
            private_key: Private key
            subject_name (x509.Name): Subject name
            extensions
        Returns: x509.CertificateSigningRequest

        """
        builder = x509.CertificateSigningRequestBuilder(
            subject_name, [] if extensions is None else extensions)

        return builder.sign(
            private_key, self.sign_hash_algorithm, default_backend())

def set_csr_if_blank(self):
        if not self.csr:
            private_key = self.get_key()
            builder = x509.CertificateSigningRequestBuilder()
            builder = builder.subject_name(x509.Name([
                x509.NameAttribute(NameOID.COMMON_NAME, self.get_common_name()),
                x509.NameAttribute(NameOID.COUNTRY_NAME, u'{}'.format(self.account.country)),
                x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, u'{}'.format(self.account.state)),
                x509.NameAttribute(NameOID.LOCALITY_NAME, u'{}'.format(self.account.locality)),
                x509.NameAttribute(NameOID.ORGANIZATION_NAME, u'{}'.format(self.account.organization_name)),
                x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, u'{}'.format(self.account.organizational_unit_name)),
            ]))
            builder = builder.add_extension(x509.SubjectAlternativeName(self.get_san_entries()), critical=False)
            csr = builder.sign(private_key, hashes.SHA256(), default_backend())
            self.csr = csr.public_bytes(serialization.Encoding.PEM)

def _create_csr(common_name, key, oids=None, sans=None):
    app.logger.info('called create_csr:\n{0}'.format(pformat(locals())))
    builder = x509.CertificateSigningRequestBuilder()
    oids = _create_oids(common_name, oids if oids else {})
    subject = builder.subject_name(x509.Name(oids))
    if sans:
        _add_sans(subject, sans)
    csr = subject.sign(key, hashes.SHA256(), default_backend())
    return csr

def _create_csr(common_name, key, oids=None, sans=None):
    app.logger.info('called create_csr:\n{0}'.format(pformat(locals())))
    builder = x509.CertificateSigningRequestBuilder()
    oids = _create_oids(common_name, oids if oids else {})
    subject = builder.subject_name(x509.Name(oids))
    if sans:
        _add_sans(subject, sans)
    csr = subject.sign(key, hashes.SHA256(), default_backend())
    return csr

def _create_csr(common_name, key, oids=None, sans=None):
    app.logger.info('called create_csr:\n{0}'.format(pformat(locals())))
    builder = x509.CertificateSigningRequestBuilder()
    oids = _create_oids(common_name, oids if oids else {})
    subject = builder.subject_name(x509.Name(oids))
    if sans:
        _add_sans(subject, sans)
    csr = subject.sign(key, hashes.SHA256(), default_backend())
    return csr

def _create_csr(common_name, key, oids=None, sans=None):
    app.logger.info('called create_csr:\n{0}'.format(pformat(locals())))
    builder = x509.CertificateSigningRequestBuilder()
    oids = _create_oids(common_name, oids if oids else {})
    subject = builder.subject_name(x509.Name(oids))
    if sans:
        _add_sans(subject, sans)
    csr = subject.sign(key, hashes.SHA256(), default_backend())
    return csr

def serialize(self,
                  # password=None,
                  country=u"US",
                  state=u"CA",
                  city=u"San Francisco",
                  company=u"Lokey Examle",
                  common_name=u"example.com"):
        # This should be handled already
        # if not password:
        #     password = None
        key = serialization.load_pem_private_key(
            self.to('pem'),
            password=None,
            backend=default_backend())

        subject = x509.Name([
            x509.NameAttribute(NameOID.COUNTRY_NAME, country),
            x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, state),
            x509.NameAttribute(NameOID.LOCALITY_NAME, city),
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, company),
            x509.NameAttribute(NameOID.COMMON_NAME, common_name),
        ])
        cert = x509.CertificateSigningRequestBuilder().subject_name(
            subject
        ).sign(key, hashes.SHA256(), default_backend())
        return cert.public_bytes(serialization.Encoding.PEM)

def csr_for_names(names, key):
    """
    Generate a certificate signing request for the given names and private key.

    ..  seealso:: `acme.client.Client.request_issuance`

    ..  seealso:: `generate_private_key`

    :param ``List[str]``: One or more names (subjectAltName) for which to
        request a certificate.
    :param key: A Cryptography private key object.

    :rtype: `cryptography.x509.CertificateSigningRequest`
    :return: The certificate request message.
    """
    if len(names) == 0:
        raise ValueError('Must have at least one name')
    if len(names[0]) > 64:
        common_name = u'san.too.long.invalid'
    else:
        common_name = names[0]
    return (
        x509.CertificateSigningRequestBuilder()
        .subject_name(x509.Name([
            x509.NameAttribute(NameOID.COMMON_NAME, common_name)]))
        .add_extension(
            x509.SubjectAlternativeName(list(map(x509.DNSName, names))),
            critical=False)
        .sign(key, hashes.SHA256(), default_backend()))

def create_csr(key, domains):
    """
    Creates a CSR in DER format for the specified key and domain names.
    """
    assert domains
    name = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, domains[0]),
    ])
    san = x509.SubjectAlternativeName([x509.DNSName(domain) for domain in domains])
    csr = x509.CertificateSigningRequestBuilder().subject_name(name) \
        .add_extension(san, critical=False) \
        .sign(key, hashes.SHA256(), default_backend())
    return export_csr_for_acme(csr)

def requestauth(self):
        # Create a CSR
        subject_attrs = []
        cn_already_set = False
        for attr in self.subject:
            if attr.oid == NameOID.COMMON_NAME:
                cn_already_set = True
            subject_attrs.append(attr)
        if not cn_already_set:
            subject_attrs.append(x509.NameAttribute(NameOID.COMMON_NAME, socket.getfqdn()))
        subject = x509.Name(subject_attrs)

        csr = x509.CertificateSigningRequestBuilder().subject_name(
            subject
        ).sign(
            private_key=self.pkey,
            algorithm=hashes.SHA256(),
            backend=default_backend()
        )

        # Ask for signature
        body = {
            'csr': csr.public_bytes(serialization.Encoding.PEM).decode()
        }
        response = requests.post(
            url=self.server + '/authorize',
            json=body,
            verify=False
        )

        if response.status_code == 200 or response.status_code == 202:
            data = response.json()
            if data['status'] == 'pending':
                logger.info("Authorization requested (key fingerprint: %s).", rsa_key_fingerprint(self.pkey.public_key()))
            elif data['status'] == 'authorized':
                with open(self.certificate_file, 'w') as f:
                    f.write(data['crt'])
                logger.info("Client authorized.")
        elif response.status_code == 500:
            data = response.json()
            logger.error('An error occured on CertProxy server while processing the request: %s', data['message'])
        else:
            logger.error('CertProxy server replied with an unexpected error code: %d (%s)', response.status_code, response.reason)