def generate_cert(key_path, cert_out_path):
private_key = load_private_key_file(key_path)
public_key = private_key.public_key()
builder = x509.CertificateBuilder()
builder = builder.subject_name(x509.Name([
x509.NameAttribute(x509.OID_COMMON_NAME, u'PrivCount User'),
]))
builder = builder.issuer_name(x509.Name([
x509.NameAttribute(x509.OID_COMMON_NAME, u'PrivCount Authority'),
]))
builder = builder.not_valid_before(datetime.datetime.today() - datetime.timedelta(days=1))
builder = builder.not_valid_after(datetime.datetime(2020, 1, 1))
builder = builder.serial_number(int(uuid.uuid4()))
builder = builder.public_key(public_key)
builder = builder.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
certificate = builder.sign(private_key=private_key, algorithm=hashes.SHA256(), backend=default_backend())
with open(cert_out_path, 'wb') as outf:
print >>outf, certificate.public_bytes(encoding=serialization.Encoding.PEM)
python类CertificateBuilder()的实例源码
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 _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 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 generate_self_signed_cert(cert_file, key_file):
# type: (Any, Any) -> None
"""Given two file-like objects, generate an SSL key and certificate
Args:
cert_file: The certificate file you wish to write to
key_file: The key file you wish to write to
"""
one_day = timedelta(1, 0, 0)
private_key = rsa.generate_private_key(
public_exponent=65537, key_size=2048, backend=default_backend())
public_key = private_key.public_key()
builder = x509.CertificateBuilder()
builder = builder.subject_name(
x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u'cryptography.io'),
]))
builder = builder.issuer_name(
x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u'cryptography.io'),
]))
builder = builder.not_valid_before(datetime.today() - one_day)
builder = builder.not_valid_after(datetime.today() + timedelta(365 * 10))
builder = builder.serial_number(uuid4().int)
builder = builder.public_key(public_key)
builder = builder.add_extension(
x509.BasicConstraints(ca=False, path_length=None),
critical=True, )
certificate = builder.sign(
private_key=private_key,
algorithm=hashes.SHA256(),
backend=default_backend())
key_file.write(
private_key.private_bytes(
Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption()))
cert_file.write(certificate.public_bytes(Encoding.PEM))
def ca_file(tmpdir):
"""
Create a valid PEM file with CA certificates and return the path.
"""
key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
public_key = key.public_key()
builder = x509.CertificateBuilder()
builder = builder.subject_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u"pyopenssl.org"),
]))
builder = builder.issuer_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u"pyopenssl.org"),
]))
one_day = datetime.timedelta(1, 0, 0)
builder = builder.not_valid_before(datetime.datetime.today() - one_day)
builder = builder.not_valid_after(datetime.datetime.today() + one_day)
builder = builder.serial_number(int(uuid.uuid4()))
builder = builder.public_key(public_key)
builder = builder.add_extension(
x509.BasicConstraints(ca=True, path_length=None), critical=True,
)
certificate = builder.sign(
private_key=key, algorithm=hashes.SHA256(),
backend=default_backend()
)
ca_file = tmpdir.join("test.pem")
ca_file.write_binary(
certificate.public_bytes(
encoding=serialization.Encoding.PEM,
)
)
return str(ca_file).encode("ascii")
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 sign_certificate_request(csr_file, crt_file, ca_crt, ca_pkey):
with open(csr_file, 'rb') as f:
csr = x509.load_pem_x509_csr(data=f.read(), backend=default_backend())
crt = x509.CertificateBuilder().subject_name(
csr.subject
).issuer_name(
ca_crt.subject
).public_key(
csr.public_key()
).serial_number(
uuid.uuid4().int # pylint: disable=no-member
).not_valid_before(
datetime.datetime.utcnow()
).not_valid_after(
datetime.datetime.utcnow() + datetime.timedelta(days=365 * 10)
).add_extension(
extension=x509.KeyUsage(
digital_signature=True, key_encipherment=True, content_commitment=True,
data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False, key_cert_sign=False, crl_sign=False
),
critical=True
).add_extension(
extension=x509.BasicConstraints(ca=False, path_length=None),
critical=True
).add_extension(
extension=x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_pkey.public_key()),
critical=False
).sign(
private_key=ca_pkey,
algorithm=hashes.SHA256(),
backend=default_backend()
)
with open(crt_file, 'wb') as f:
f.write(crt.public_bytes(encoding=serialization.Encoding.PEM))
def makeSignedCert(cpub, ccn, cvdays, cserial, spriv, scert=None):
"""
Creates a certificate for a given public key and signs it with a given
certificate and private key. It will reuse the subject of the signing
certificate as the subject of the new certificate, only replacing the
common name with the one given as parameter, if a signing certificate is
specified, otherwise it will just use the given common name as subject
and issuer.
:param cpub: Public key for which to create a certificate.
:param ccn: Common name for the new certificate.
:param cvdays: Number of days the new certificate is valid.
:param cserial: The serial number for the new certificate as an int.
:param spriv: Private key for the signing certificate.
:param scert: Certificate used to sign the new certificate, or None if
no certificate is used.
:return: The new certificate as an object.
"""
if scert:
sname = x509.Name(
[ p for p in scert.subject if p.oid != NameOID.COMMON_NAME ]
+ [ x509.NameAttribute(NameOID.COMMON_NAME, ccn) ])
iname = scert.subject
else:
sname = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, ccn)])
iname = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, ccn)])
builder = x509.CertificateBuilder()
builder = builder.subject_name(sname)
builder = builder.issuer_name(iname)
builder = builder.not_valid_before(datetime.datetime.today())
builder = builder.not_valid_after(datetime.datetime.today() +
datetime.timedelta(cvdays, 0, 0))
builder = builder.serial_number(cserial)
builder = builder.public_key(cpub)
builder = builder.add_extension(
x509.BasicConstraints(ca=True, path_length=None),
critical=True
)
return builder.sign(private_key=spriv, algorithm=hashes.SHA256(),
backend=default_backend())
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 MakeCASignedCert(common_name,
private_key,
ca_cert,
ca_private_key,
serial_number=2,
session=None):
"""Make a cert and sign it with the CA's private key."""
public_key = private_key.public_key()
builder = x509.CertificateBuilder()
builder = builder.issuer_name(ca_cert.get_issuer())
subject = x509.Name([
x509.NameAttribute(oid.NameOID.COMMON_NAME, common_name)
])
builder = builder.subject_name(subject)
valid_from = time.time() - 60 * 60 * 24
valid_until = time.time() + 60 * 60 * 24 * 365 * 10
builder = builder.not_valid_before(datetime.datetime.fromtimestamp(
valid_from))
builder = builder.not_valid_after(datetime.datetime.fromtimestamp(
valid_until))
builder = builder.serial_number(serial_number)
builder = builder.public_key(public_key.get_raw_key())
builder = builder.add_extension(
x509.BasicConstraints(
ca=False, path_length=None), critical=True)
certificate = builder.sign(
private_key=ca_private_key.get_raw_key(),
algorithm=hashes.SHA256(),
backend=openssl.backend)
return X509Ceritifcate(session=session).from_raw_key(certificate)
def _generate_ca_cert(self):
"""
Generate a CA cert/key.
"""
if self._ca_key is None:
self._ca_key = generate_private_key(u'rsa')
self._ca_name = x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u'ACME Snake Oil CA')])
self._ca_cert = (
x509.CertificateBuilder()
.subject_name(self._ca_name)
.issuer_name(self._ca_name)
.not_valid_before(self._now() - timedelta(seconds=3600))
.not_valid_after(self._now() + timedelta(days=3650))
.public_key(self._ca_key.public_key())
.serial_number(int(uuid4()))
.add_extension(
x509.BasicConstraints(ca=True, path_length=0),
critical=True)
.add_extension(
x509.SubjectKeyIdentifier.from_public_key(
self._ca_key.public_key()),
critical=False)
.sign(
private_key=self._ca_key,
algorithm=hashes.SHA256(),
backend=default_backend()))
self._ca_aki = x509.AuthorityKeyIdentifier.from_issuer_public_key(
self._ca_key.public_key())
def request_issuance(self, csr):
csr = csr.csr
# TODO: Only in Cryptography 1.3
# assert csr.is_signature_valid
cert = (
x509.CertificateBuilder()
.subject_name(csr.subject)
.issuer_name(self._ca_name)
.not_valid_before(self._now() - timedelta(seconds=3600))
.not_valid_after(self._now() + timedelta(days=90))
.serial_number(int(uuid4()))
.public_key(csr.public_key())
.add_extension(
csr.extensions.get_extension_for_oid(
ExtensionOID.SUBJECT_ALTERNATIVE_NAME).value,
critical=False)
.add_extension(
x509.SubjectKeyIdentifier.from_public_key(csr.public_key()),
critical=False)
.add_extension(self._ca_aki, critical=False)
.sign(
private_key=self._ca_key,
algorithm=hashes.SHA256(),
backend=default_backend()))
cert_res = messages.CertificateResource(
body=cert.public_bytes(encoding=serialization.Encoding.DER))
return self._controller.issue().addCallback(lambda _: cert_res)
def generate_tls_sni_01_cert(server_name, key_type=u'rsa',
_generate_private_key=None):
"""
Generate a certificate/key pair for responding to a tls-sni-01 challenge.
:param str server_name: The SAN the certificate should have.
:param str key_type: The type of key to generate; usually not necessary.
:rtype: ``Tuple[`~cryptography.x509.Certificate`, PrivateKey]``
:return: A tuple of the certificate and private key.
"""
key = (_generate_private_key or generate_private_key)(key_type)
name = x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u'acme.invalid')])
cert = (
x509.CertificateBuilder()
.subject_name(name)
.issuer_name(name)
.not_valid_before(datetime.now() - timedelta(seconds=3600))
.not_valid_after(datetime.now() + timedelta(seconds=3600))
.serial_number(int(uuid.uuid4()))
.public_key(key.public_key())
.add_extension(
x509.SubjectAlternativeName([x509.DNSName(server_name)]),
critical=False)
.sign(
private_key=key,
algorithm=hashes.SHA256(),
backend=default_backend())
)
return (cert, key)
def keygen(pub_key, priv_key, user_id, priv_key_password=None):
"""Generate new private key and certificate RSA_SHA512_4096"""
# Generate our key
key = rsa.generate_private_key(public_exponent=65537, key_size=4096,
backend=default_backend())
if priv_key_password:
ea = serialization.BestAvailableEncryption(priv_key_password)
else:
ea = serialization.NoEncryption()
# Write our key to disk for safe keeping
with open(priv_key, "wb") as f:
f.write(key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=ea,
))
# Various details about who we are. For a self-signed certificate the
# subject and issuer are always the same.
subject = issuer = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "XX"),
x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "XX"),
x509.NameAttribute(NameOID.LOCALITY_NAME, "XX"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "I2P Anonymous Network"),
x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "I2P"),
x509.NameAttribute(NameOID.COMMON_NAME, user_id),
])
cert = x509.CertificateBuilder() \
.subject_name(subject) \
.issuer_name(issuer) \
.public_key(key.public_key()) \
.not_valid_before(datetime.datetime.utcnow()) \
.not_valid_after(
datetime.datetime.utcnow() + datetime.timedelta(days=365*10)
) \
.serial_number(random.randrange(1000000000, 2000000000)) \
.add_extension(
x509.SubjectKeyIdentifier.from_public_key(key.public_key()),
critical=False,
).sign(key, hashes.SHA512(), default_backend())
with open(pub_key, "wb") as f:
f.write(cert.public_bytes(serialization.Encoding.PEM))
def create_proxy_cert(loaded_cert, loaded_private_key,
loaded_public_key, lifetime_hours):
"""
Given cryptography objects for an issuing certificate, a public_key,
a private_key, and an int for lifetime in hours, creates a proxy
cert from the issuer and public key signed by the private key.
"""
builder = x509.CertificateBuilder()
# create a serial number for the new proxy
# Under RFC 3820 there are many ways to generate the serial number. However
# making the number unpredictable has security benefits, e.g. it can make
# this style of attack more difficult:
# http://www.win.tue.nl/hashclash/rogue-ca
serial = struct.unpack("<Q", os.urandom(8))[0]
builder = builder.serial_number(serial)
# set the new proxy as valid from now until lifetime_hours have passed
builder = builder.not_valid_before(datetime.datetime.utcnow())
builder = builder.not_valid_after(
datetime.datetime.utcnow() + datetime.timedelta(hours=lifetime_hours))
# set the public key of the new proxy to the given public key
builder = builder.public_key(loaded_public_key)
# set the issuer of the new cert to the subject of the issuing cert
builder = builder.issuer_name(loaded_cert.subject)
# set the new proxy's subject
# append a CommonName to the new proxy's subject
# with the serial as the value of the CN
new_atribute = x509.NameAttribute(
x509.oid.NameOID.COMMON_NAME, six.u(str(serial)))
subject_attributes = list(loaded_cert.subject)
subject_attributes.append(new_atribute)
builder = builder.subject_name(x509.Name(subject_attributes))
# add proxyCertInfo extension to the new proxy (We opt not to add keyUsage)
# For RFC proxies the effective usage is defined as the intersection
# of the usage of each cert in the chain. See section 4.2 of RFC 3820.
# the constants 'oid' and 'value' are gotten from
# examining output from a call to the open ssl function:
# X509V3_EXT_conf(NULL, ctx, name, value)
# ctx set by X509V3_set_nconf(&ctx, NCONF_new(NULL))
# name = "proxyCertInfo"
# value = "critical,language:Inherit all"
oid = x509.ObjectIdentifier("1.3.6.1.5.5.7.1.14")
value = b"0\x0c0\n\x06\x08+\x06\x01\x05\x05\x07\x15\x01"
extension = x509.extensions.UnrecognizedExtension(oid, value)
builder = builder.add_extension(extension, critical=True)
# sign the new proxy with the issuer's private key
new_certificate = builder.sign(
private_key=loaded_private_key, algorithm=hashes.SHA256(),
backend=default_backend())
# return the new proxy as a cryptography object
return new_certificate
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
def load_or_create_ca_certificate(crt_file, subject, pkey):
""" Load a CA certificate or create a self-signed one """
if os.path.isfile(crt_file):
with open(crt_file, 'rb') as f:
crt = x509.load_pem_x509_certificate(
data=f.read(),
backend=default_backend()
)
else:
issuer = subject
crt = x509.CertificateBuilder().subject_name(
subject
).issuer_name(
issuer
).public_key(
pkey.public_key()
).serial_number(
uuid.uuid4().int # pylint: disable=no-member
).not_valid_before(
datetime.datetime.utcnow()
).not_valid_after(
datetime.datetime.utcnow() + datetime.timedelta(days=365 * 10)
).add_extension(
extension=x509.KeyUsage(
digital_signature=True, key_encipherment=True, key_cert_sign=True, crl_sign=True, content_commitment=True,
data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False
),
critical=True
).add_extension(
extension=x509.BasicConstraints(ca=True, path_length=0),
critical=True
).add_extension(
extension=x509.SubjectKeyIdentifier.from_public_key(pkey.public_key()),
critical=True
).add_extension(
extension=x509.AuthorityKeyIdentifier.from_issuer_public_key(pkey.public_key()),
critical=True
).sign(
private_key=pkey,
algorithm=hashes.SHA256(),
backend=default_backend()
)
with open(crt_file, 'wb') as f:
f.write(crt.public_bytes(encoding=serialization.Encoding.PEM))
return crt
def MakeCACert(private_key,
common_name=u"rekall-agent-ca",
issuer_cn=u"rekall-agent-ca",
issuer_c=u"US",
session=None):
"""Generate a CA certificate.
Args:
private_key: The private key to use.
common_name: Name for cert.
issuer_cn: Name for issuer.
issuer_c: Country for issuer.
Returns:
The certificate.
"""
public_key = private_key.public_key()
builder = x509.CertificateBuilder()
issuer = x509.Name([
x509.NameAttribute(oid.NameOID.COMMON_NAME, issuer_cn),
x509.NameAttribute(oid.NameOID.COUNTRY_NAME, issuer_c)
])
subject = x509.Name([
x509.NameAttribute(oid.NameOID.COMMON_NAME, common_name)
])
builder = builder.subject_name(subject)
builder = builder.issuer_name(issuer)
valid_from = time.time() - 60 * 60 * 24
valid_until = time.time() + 60 * 60 * 24 * 365 * 10
builder = builder.not_valid_before(datetime.datetime.fromtimestamp(
valid_from))
builder = builder.not_valid_after(datetime.datetime.fromtimestamp(
valid_until))
builder = builder.serial_number(1)
builder = builder.public_key(public_key.get_raw_key())
builder = builder.add_extension(
x509.BasicConstraints(
ca=True, path_length=None), critical=True)
builder = builder.add_extension(
x509.SubjectKeyIdentifier.from_public_key(
public_key.get_raw_key()),
critical=False)
certificate = builder.sign(
private_key=private_key.get_raw_key(),
algorithm=hashes.SHA256(),
backend=openssl.backend)
return X509Ceritifcate(session=session).from_raw_key(certificate)
def _generate_cert(server_name, not_valid_before, not_valid_after,
key=RSA_KEY_512_RAW):
"""
Generate a self-signed certificate for test purposes.
:param str server_name: The SAN the certificate should have.
:param ~datetime.datetime not_valid_before: Valid from this moment.
:param ~datetime.datetime not_valid_after: Expiry time.
:param key: The private key.
:rtype: `str`
:return: The certificate in PEM format.
"""
common_name = (
u'san.too.long.invalid' if len(server_name) > 64 else server_name)
name = x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, common_name)])
cert = (
x509.CertificateBuilder()
.subject_name(name)
.issuer_name(name)
.not_valid_before(not_valid_before)
.not_valid_after(not_valid_after)
.serial_number(int(uuid.uuid4()))
.public_key(key.public_key())
.add_extension(
x509.SubjectAlternativeName([x509.DNSName(server_name)]),
critical=False)
.sign(
private_key=key,
algorithm=hashes.SHA256(),
backend=default_backend())
)
return [
Certificate(
cert.public_bytes(serialization.Encoding.PEM)),
RSAPrivateKey(
key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption())),
]
def get_certificate_authority_certificate(self):
if self.keypairs["ca"]["certificate"]:
return self.keypairs["ca"]["certificate"]
else:
if self.path:
with open(os.path.join(self.path, "ca.pem"), "rb") as fp:
certificate = x509.load_pem_x509_certificate(fp.read(), default_backend())
else:
ca_key = self.get_certificate_authority_key()
builder = x509.CertificateBuilder()
builder = builder.serial_number(int(uuid.uuid4()))
builder = builder.not_valid_before(datetime.datetime.today() - datetime.timedelta(1, 0, 0))
builder = builder.not_valid_after(datetime.datetime(2018, 8, 2))
builder = builder.public_key(ca_key.public_key())
builder = builder.subject_name(x509.Name([
x509.NameAttribute(x509.NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(x509.NameOID.STATE_OR_PROVINCE_NAME, "CO"),
x509.NameAttribute(x509.NameOID.LOCALITY_NAME, "Denver"),
x509.NameAttribute(x509.NameOID.ORGANIZATION_NAME, "Eldarion, Inc."),
x509.NameAttribute(x509.NameOID.COMMON_NAME, "eldarion.com"),
]))
builder = builder.issuer_name(x509.Name([
x509.NameAttribute(x509.NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(x509.NameOID.STATE_OR_PROVINCE_NAME, "CO"),
x509.NameAttribute(x509.NameOID.LOCALITY_NAME, "Denver"),
x509.NameAttribute(x509.NameOID.ORGANIZATION_NAME, "Eldarion, Inc."),
x509.NameAttribute(x509.NameOID.COMMON_NAME, "eldarion.com"),
]))
builder = builder.add_extension(
x509.BasicConstraints(
ca=True,
path_length=None
),
critical=False,
)
certificate = builder.sign(
private_key=ca_key,
algorithm=hashes.SHA256(),
backend=default_backend(),
)
self.keypairs["ca"]["certificate"] = certificate
return certificate
def get_certificate(self, name, opts):
if self.keypairs[name]["certificate"]:
return self.keypairs[name]["certificate"]
else:
if self.path:
with open(os.path.join(self.path, "{}.pem".format(name)), "rb") as fp:
certificate = x509.load_pem_x509_certificate(fp.read(), default_backend())
else:
ca_key = self.get_certificate_authority_key()
ca_certificate = self.get_certificate_authority_certificate()
builder = x509.CertificateBuilder()
builder = builder.serial_number(int(uuid.uuid4()))
builder = builder.not_valid_before(datetime.datetime.today() - datetime.timedelta(1, 0, 0))
builder = builder.not_valid_after(datetime.datetime(2018, 8, 2))
builder = builder.public_key(ca_key.public_key())
builder = builder.subject_name(x509.Name([
x509.NameAttribute(x509.NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(x509.NameOID.STATE_OR_PROVINCE_NAME, "CO"),
x509.NameAttribute(x509.NameOID.LOCALITY_NAME, "Denver"),
x509.NameAttribute(x509.NameOID.ORGANIZATION_NAME, "Eldarion, Inc."),
x509.NameAttribute(x509.NameOID.COMMON_NAME, "kube-{}".format(name)),
]))
builder = builder.issuer_name(ca_certificate.issuer)
if opts.get("sans"):
builder = builder.add_extension(
x509.SubjectAlternativeName(opts["sans"]),
critical=False,
)
builder = builder.add_extension(
x509.BasicConstraints(
ca=False,
path_length=None
),
critical=False,
)
certificate = builder.sign(
private_key=ca_key,
algorithm=hashes.SHA256(),
backend=default_backend(),
)
self.keypairs[name]["certificate"] = certificate
return certificate
