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

def _cert_builder_common(subject, issuer, public_key):
    return (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(issuer)
        .public_key(public_key)
        .not_valid_before(datetime.datetime(2000, 1, 1))
        # OpenSSL on Windows freaks out if you try to give it a date after
        # ~3001-01-19
        # https://github.com/pyca/cryptography/issues/3194
        .not_valid_after(datetime.datetime(3000, 1, 1))
        .serial_number(x509.random_serial_number())
        .add_extension(
            x509.SubjectKeyIdentifier.from_public_key(public_key),
            critical=False,
        )
    )

def prepareCert(cert_request):
    #copy data from cert_request or certificate ,add default data and save to certificates database
    #used for both New requests and renewals

    cert_data=Certificate()
    keys={'country_name','state_or_province_name','locality_name','organization_name','organization_unit_name','email_address','user_id','dns_name','common_name','dn_qualifier'}

    for key in keys:
        #copyattrorgetdefault only for text attributes
        copyAttrOrGetDefault(cert_data,key,cert_request)

    cert_data.is_ca=cert_request.is_ca
    cert_data.not_valid_before=timezone.now()-datetime.timedelta(1,0,0)
    cert_data.not_valid_after=cert_request.not_valid_after   
    cert_data.serial_number=x509.random_serial_number()
    cert_data.issuer_serial_number=get_issuer(cert_data).serial_number
    logger.info("random serial %s",cert_data.serial_number)
    cert_data.save()

    return cert_data

def _create_self_signed_certificate(self, private_key):
        issuer = x509.Name([
            x509.NameAttribute(NameOID.COUNTRY_NAME, u"US"),
            x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, u"CA"),
            x509.NameAttribute(NameOID.LOCALITY_NAME, u"San Francisco"),
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, u"My Company"),
            x509.NameAttribute(NameOID.COMMON_NAME, u"Test Certificate"),
        ])
        cert_builder = x509.CertificateBuilder(
            issuer_name=issuer, subject_name=issuer,
            public_key=private_key.public_key(),
            serial_number=x509.random_serial_number(),
            not_valid_before=datetime.utcnow(),
            not_valid_after=datetime.utcnow() + timedelta(days=10)
        )
        cert = cert_builder.sign(private_key,
                                 hashes.SHA256(),
                                 default_backend())
        return cert

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 gen_certificate(key: rsa.RSAPrivateKey,
                    common_name: str,
                    *,
                    issuer: Optional[str]=None,
                    sign_key: Optional[rsa.RSAPrivateKey]=None) -> x509.Certificate:
    now = datetime.datetime.utcnow()
    return x509.CertificateBuilder().subject_name(
        x509.Name([
            x509.NameAttribute(NameOID.COMMON_NAME, common_name),
        ])
    ).issuer_name(
        x509.Name([
            x509.NameAttribute(
                NameOID.COMMON_NAME,
                issuer or common_name
            )
        ])
    ).not_valid_before(
        now
    ).not_valid_after(
        now + datetime.timedelta(seconds=86400)
    ).serial_number(
        x509.random_serial_number()
    ).public_key(
        key.public_key()
    ).add_extension(
        x509.BasicConstraints(ca=True, path_length=0), critical=True
    ).sign(
        private_key=sign_key or key,
        algorithm=hashes.SHA256(),
        backend=BACKEND
    )

def random_serial_number():
        return utils.int_from_bytes(os.urandom(20), "big") >> 1

def issue_certificate(cn, ca_cert, ca_key,
                      organizations=(),
                      san_dns=(),
                      san_ips=(),
                      key_size=2048,
                      certify_days=365,
                      is_web_server=False,
                      is_web_client=False):
    ca_cert = x509.load_pem_x509_certificate(ca_cert, default_backend())
    ca_key = serialization.load_pem_private_key(ca_key, password=None, backend=default_backend())
    ca_key_id = x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key())

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

    subject_name_attributes = [x509.NameAttribute(NameOID.COMMON_NAME, cn)]
    subject_name_attributes += [x509.NameAttribute(NameOID.ORGANIZATION_NAME, org) for org in organizations]
    subject = x509.Name(subject_name_attributes)

    now = datetime.datetime.utcnow()
    cert = x509.CertificateBuilder() \
        .subject_name(subject) \
        .issuer_name(ca_cert.issuer) \
        .public_key(key.public_key()) \
        .serial_number(x509.random_serial_number()) \
        .not_valid_before(now) \
        .not_valid_after(now + datetime.timedelta(days=certify_days)) \
        .add_extension(x509.AuthorityKeyIdentifier(ca_key_id.digest,
                                                   [x509.DirectoryName(ca_cert.issuer)],
                                                   ca_cert.serial_number),
                       critical=False) \
        .add_extension(x509.KeyUsage(digital_signature=True,
                                     content_commitment=False,
                                     key_encipherment=True,
                                     data_encipherment=False,
                                     key_agreement=False,
                                     key_cert_sign=False,
                                     crl_sign=False,
                                     encipher_only=False,
                                     decipher_only=False),
                       critical=True)

    extended_usages = []
    if is_web_server:
        extended_usages.append(ExtendedKeyUsageOID.SERVER_AUTH)
    if is_web_client:
        extended_usages.append(ExtendedKeyUsageOID.CLIENT_AUTH)
    if extended_usages:
        cert = cert.add_extension(x509.ExtendedKeyUsage(extended_usages), critical=False)

    sans = [x509.DNSName(name) for name in san_dns]
    sans += [x509.IPAddress(ipaddress.ip_address(ip)) for ip in san_ips]
    if sans:
        cert = cert.add_extension(x509.SubjectAlternativeName(sans), critical=False)

    cert = cert.sign(ca_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