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

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)

def __init__(self):
        self._private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=_KEY_SIZE,
            backend=default_backend()
        )

        name = _name(u"Testing CA #" + random_text())
        self._certificate = (
            _cert_builder_common(name, name, self._private_key.public_key())
            .add_extension(
                x509.BasicConstraints(ca=True, path_length=9), critical=True,
            )
            .sign(
                private_key=self._private_key,
                algorithm=hashes.SHA256(),
                backend=default_backend(),
            )
        )

        self.cert_pem = Blob(self._certificate.public_bytes(Encoding.PEM))

def check_ca(cert):
        """Check if 'cert' is a proper CA. For this the BasicConstraints need to
        identify it as a CA cert and it needs to have the CertSign
        (key_cert_sign in Cryptography) KeyUsage flag. Based loosely on
        OpenSSL's check_ca()"""

        from cryptography import x509

        bconst_ca = None
        kuse_sign = None

        for e in cert.extensions:
                if isinstance(e.value, x509.BasicConstraints):
                        bconst_ca = e.value.ca
                elif isinstance(e.value, x509.KeyUsage):
                        kuse_sign = e.value.key_cert_sign

        return kuse_sign is not False and bconst_ca

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

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 _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

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 _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    if basic_constraints.pathlen == backend._ffi.NULL:
        path_length = None
    else:
        path_length = backend._asn1_integer_to_int(basic_constraints.pathlen)

    return x509.BasicConstraints(ca, path_length)

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 _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

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 test_basics():
    ca = CA()

    today = datetime.datetime.today()

    assert b"BEGIN CERTIFICATE" in ca.cert_pem.bytes()

    ca_cert = x509.load_pem_x509_certificate(
        ca.cert_pem.bytes(), default_backend())
    assert ca_cert.not_valid_before <= today <= ca_cert.not_valid_after

    assert ca_cert.issuer == ca_cert.subject
    bc = ca_cert.extensions.get_extension_for_class(x509.BasicConstraints)
    assert bc.value.ca == True
    assert bc.critical == True

    with pytest.raises(ValueError):
        ca.issue_server_cert()

    server = ca.issue_server_cert(u"test-1.example.org", u"test-2.example.org")

    assert b"PRIVATE KEY" in server.private_key_pem.bytes()
    assert b"BEGIN CERTIFICATE" in server.cert_chain_pems[0].bytes()
    assert len(server.cert_chain_pems) == 1
    assert server.private_key_pem.bytes() in server.private_key_and_cert_chain_pem.bytes()
    for blob in server.cert_chain_pems:
        assert blob.bytes() in server.private_key_and_cert_chain_pem.bytes()

    server_cert = x509.load_pem_x509_certificate(
        server.cert_chain_pems[0].bytes(), default_backend())

    assert server_cert.not_valid_before <= today <= server_cert.not_valid_after
    assert server_cert.issuer == ca_cert.subject

    san = server_cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
    hostnames = san.value.get_values_for_type(x509.DNSName)
    assert hostnames == [u"test-1.example.org", u"test-2.example.org"]

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

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 _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

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 _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

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 _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

    return x509.BasicConstraints(ca, path_length)

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 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

def __check_extensions(self, cert, usages, cur_pathlen):
                """Check whether the critical extensions in this certificate
                are supported and allow the provided use(s)."""

                try:
                        exts = cert.extensions
                except (ValueError, x509.UnsupportedExtension) as e:
                        raise api_errors.InvalidCertificateExtensions(
                            cert, e)

                for ext in exts:
                        etype = type(ext.value)
                        if etype in SUPPORTED_EXTENSIONS:
                                keys = EXTENSIONS_VALUES[etype]
                                if etype == x509.BasicConstraints:
                                        pathlen = ext.value.path_length
                                        if pathlen is not None and \
                                            cur_pathlen > pathlen:
                                                raise api_errors.PathlenTooShort(cert,
                                                    cur_pathlen, pathlen)
                                elif etype == x509.KeyUsage:
                                        keys = list(EXTENSIONS_VALUES[etype])
                                        if not getattr(ext.value,
                                            "key_agreement"):
                                                # Cryptography error:
                                                # encipher_only/decipher_only is
                                                # undefined unless key_agreement
                                                # is true
                                                keys.remove("encipher_only")
                                                keys.remove("decipher_only")
                                vs = [
                                    key
                                    for key in keys
                                    if getattr(ext.value, key)
                                ]
                                # For each use, check to see whether it's
                                # permitted by the certificate's extension
                                # values.
                                if etype not in usages:
                                        continue
                                for u in usages[etype]:
                                        if u not in vs:
                                                raise api_errors.InappropriateCertificateUse(
                                                    cert, ext, u, ", ".join(vs))
                        # If the extension name is unrecognized and critical,
                        # then the chain cannot be verified.
                        elif ext.critical:
                                raise api_errors.UnsupportedCriticalExtension(
                                    cert, ext)