lauri/certidude
1
0
Fork 0
mirror of https://github.com/laurivosandi/certidude synced 2024-12-23 00:25:18 +00:00
Code Issues Activity

Add AKID and SKID

Browse Source
This commit is contained in:
Lauri Võsandi 2016-03-29 08:47:43 +03:00
parent ff71ca42d7
commit acc0e29109
2 changed files with 98 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
certidude/cli.py
View File

@ -863,7 +863,15 @@ def certidude_setup_authority(parent, country, state, locality, organization, or
crypto.X509Extension( crypto.X509Extension(
b"subjectAltName", b"subjectAltName",
False, False,
"DNS: %s, email: %s" % (common_name.encode("ascii"), email_address.encode("ascii"))) (u"DNS: %s, email: %s" % (common_name, email_address)).encode("ascii"))
])
ca.add_extensions([
crypto.X509Extension(
b"authorityKeyIdentifier",
False,
b"keyid:always",
issuer = ca)
]) ])
if ocsp_responder_url: if ocsp_responder_url:

157
certidude/signer.py
View File

@ -7,25 +7,21 @@ import os
import asyncore import asyncore
import asynchat import asynchat
from certidude import constants, config from certidude import constants, config
from datetime import datetime
from OpenSSL import crypto from OpenSSL import crypto
""" from cryptography import x509
Signer processes are spawned per private key. from cryptography.hazmat.backends import default_backend
Private key should only be readable by root. from cryptography.hazmat.primitives import hashes, serialization
Signer process starts up as root, reads private key, from cryptography.hazmat.primitives.serialization import Encoding
drops privileges and awaits for opcodes (sign-request, export-crl) at UNIX domain socket from datetime import datetime, timedelta
under /run/certidude/signer/ from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
The main motivation behind the concept is to mitigate private key leaks import random
by confining it to a separate process.
Note that signer process uses basicConstraints, keyUsage and extendedKeyUsage DN_WHITELIST = NameOID.COMMON_NAME, NameOID.GIVEN_NAME, NameOID.SURNAME, \
attributes from openssl.cnf via CertificateAuthority wrapper class. NameOID.EMAIL_ADDRESS
Hence it's possible only to sign such certificates via the signer process,
making it hard to take advantage of hacked Certidude server, eg. being able to sign SERIAL_MIN = 0x1000000000000000000000000000000000000000
certificate authoirty (basicConstraints=CA:TRUE) or SERIAL_MAX = 0xffffffffffffffffffffffffffffffffffffffff
TLS server certificates (extendedKeyUsage=serverAuth).
"""
def raw_sign(private_key, ca_cert, request, basic_constraints, lifetime, key_usage=None, extended_key_usage=None): def raw_sign(private_key, ca_cert, request, basic_constraints, lifetime, key_usage=None, extended_key_usage=None):
""" """
@ -42,15 +38,20 @@ def raw_sign(private_key, ca_cert, request, basic_constraints, lifetime, key_usa
# Set issuer # Set issuer
cert.set_issuer(ca_cert.get_subject()) cert.set_issuer(ca_cert.get_subject())
# Copy attributes from CA # Set SKID and AKID extensions
if ca_cert.get_subject().C: cert.add_extensions([
cert.get_subject().C = ca_cert.get_subject().C crypto.X509Extension(
if ca_cert.get_subject().ST: b"subjectKeyIdentifier",
cert.get_subject().ST = ca_cert.get_subject().ST False,
if ca_cert.get_subject().L: b"hash",
cert.get_subject().L = ca_cert.get_subject().L subject = cert),
if ca_cert.get_subject().O: crypto.X509Extension(
cert.get_subject().O = ca_cert.get_subject().O b"authorityKeyIdentifier",
False,
b"keyid:always",
issuer = ca_cert)
])
# Copy attributes from request # Copy attributes from request
cert.get_subject().CN = request.get_subject().CN cert.get_subject().CN = request.get_subject().CN
@ -98,10 +99,8 @@ def raw_sign(private_key, ca_cert, request, basic_constraints, lifetime, key_usa
cert.gmtime_adj_notBefore(-3600) cert.gmtime_adj_notBefore(-3600)
cert.gmtime_adj_notAfter(lifetime * 24 * 60 * 60) cert.gmtime_adj_notAfter(lifetime * 24 * 60 * 60)
# Generate serial from 0x10000000000000000000 to 0xffffffffffffffffffff # Generate random serial
cert.set_serial_number(random.randint( cert.set_serial_number(random.randint(SERIAL_MIN, SERIAL_MAX))
0x1000000000000000000000000000000000000000,
0xffffffffffffffffffffffffffffffffffffffff))
cert.sign(private_key, 'sha256') cert.sign(private_key, 'sha256')
return cert return cert
@ -114,57 +113,82 @@ class SignHandler(asynchat.async_chat):
self.server = server self.server = server
def parse_command(self, cmd, body=""): def parse_command(self, cmd, body=""):
now = datetime.utcnow()
if cmd == "export-crl": if cmd == "export-crl":
""" """
Generate CRL object based on certificate serial number and revocation timestamp Generate CRL object based on certificate serial number and revocation timestamp
""" """
crl = crypto.CRL()
builder = x509.CertificateRevocationListBuilder(
).last_update(now
).next_update(now + timedelta(days=1)
).issuer_name(self.server.certificate.issuer
).add_extension(
x509.AuthorityKeyIdentifier.from_issuer_public_key(
self.server.certificate.public_key()), False)
if body: if body:
for line in body.split("\n"): for line in body.split("\n"):
serial_number, timestamp = line.split(":") serial_number, timestamp = line.split(":")
# TODO: Assert serial against regex revocation = x509.RevokedCertificateBuilder(
revocation = crypto.Revoked() ).serial_number(int(serial_number, 16)
revocation.set_rev_date(datetime.utcfromtimestamp(int(timestamp)).strftime("%Y%m%d%H%M%SZ").encode("ascii")) ).revocation_date(datetime.utcfromtimestamp(int(timestamp))
revocation.set_reason(b"keyCompromise") ).add_extension(x509.CRLReason(x509.ReasonFlags.key_compromise), False
revocation.set_serial(serial_number.encode("ascii")) ).build(default_backend())
crl.add_revoked(revocation) builder = builder.add_revoked_certificate(revocation)
self.send(crl.export( crl = builder.sign(
self.server.certificate,
self.server.private_key, self.server.private_key,
crypto.FILETYPE_PEM, hashes.SHA512(),
config.REVOCATION_LIST_LIFETIME)) default_backend())
self.send(crl.public_bytes(Encoding.PEM))
elif cmd == "ocsp-request": elif cmd == "ocsp-request":
NotImplemented # TODO: Implement OCSP NotImplemented # TODO: Implement OCSP
elif cmd == "sign-request": elif cmd == "sign-request":
request = crypto.load_certificate_request(crypto.FILETYPE_PEM, body) request = x509.load_pem_x509_csr(body, default_backend())
subject = x509.Name([n for n in request.subject if n.oid in DN_WHITELIST])
for e in request.get_extensions(): cert = x509.CertificateBuilder(
key = e.get_short_name().decode("ascii") ).subject_name(subject
if key not in constants.EXTENSION_WHITELIST: ).serial_number(random.randint(SERIAL_MIN, SERIAL_MAX)
raise ValueError("Certificte Signing Request contains extension '%s' which is not whitelisted" % key) ).issuer_name(self.server.certificate.issuer
).public_key(request.public_key()
).not_valid_before(now - timedelta(hours=1)
).not_valid_after(now + timedelta(days=config.CERTIFICATE_LIFETIME)
).add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True,
).add_extension(x509.KeyUsage(
digital_signature=True,
key_encipherment=True,
content_commitment=False,
data_encipherment=False,
key_agreement=False,
key_cert_sign=False,
crl_sign=False,
encipher_only=False,
decipher_only=False), critical=True,
).add_extension(x509.ExtendedKeyUsage(
[ExtendedKeyUsageOID.CLIENT_AUTH]
), critical=True,
).add_extension(
x509.SubjectKeyIdentifier.from_public_key(request.public_key()),
critical=False
).add_extension(
x509.AuthorityKeyIdentifier.from_issuer_public_key(
self.server.certificate.public_key()),
critical=False
).sign(self.server.private_key, hashes.SHA512(), default_backend())
# TODO: Potential exploits during PEM parsing? self.send(cert.public_bytes(serialization.Encoding.PEM))
cert = raw_sign(
self.server.private_key,
self.server.certificate,
request,
basic_constraints=config.CERTIFICATE_BASIC_CONSTRAINTS,
key_usage=config.CERTIFICATE_KEY_USAGE_FLAGS,
extended_key_usage=config.CERTIFICATE_EXTENDED_KEY_USAGE_FLAGS,
lifetime=config.CERTIFICATE_LIFETIME)
self.send(crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
else: else:
raise NotImplementedError("Unknown command: %s" % cmd) raise NotImplementedError("Unknown command: %s" % cmd)
self.close_when_done() self.close_when_done()
def found_terminator(self): def found_terminator(self):
args = (b"".join(self.buffer)).decode("ascii").split("\n", 1) args = (b"".join(self.buffer)).split("\n", 1)
self.parse_command(*args) self.parse_command(*args)
self.buffer = [] self.buffer = []
@ -184,20 +208,17 @@ class SignServer(asyncore.dispatcher):
self.bind(config.SIGNER_SOCKET_PATH) self.bind(config.SIGNER_SOCKET_PATH)
self.listen(5) self.listen(5)
# Load CA private key and certificate # Load CA private key and certificate
self.private_key = crypto.load_privatekey(crypto.FILETYPE_PEM, self.private_key = serialization.load_pem_private_key(
open(config.AUTHORITY_PRIVATE_KEY_PATH).read()) open(config.AUTHORITY_PRIVATE_KEY_PATH).read(),
self.certificate = crypto.load_certificate(crypto.FILETYPE_PEM, password=None, # TODO: Ask password for private key?
open(config.AUTHORITY_CERTIFICATE_PATH).read()) backend=default_backend())
self.certificate = x509.load_pem_x509_certificate(
open(config.AUTHORITY_CERTIFICATE_PATH).read(),
backend=default_backend())
# Perhaps perform chroot as well, currently results in # Drop privileges
# (<class 'OpenSSL.crypto.Error'>:[('random number generator', 'SSLEAY_RAND_BYTES', 'PRNG not seeded')
# probably needs partially populated /dev in chroot
# Dropping privileges
_, _, uid, gid, gecos, root, shell = pwd.getpwnam("nobody") _, _, uid, gid, gecos, root, shell = pwd.getpwnam("nobody")
#os.chroot("/run/certidude/signer/jail")
os.setgid(gid) os.setgid(gid)
os.setuid(uid) os.setuid(uid)