Jarian Cottingham
c759597ad0
--pin-source stdin prevents PIN visibility in `ps` output. Use tempfile.mkstemp for all temp files (unpredictable names, 0600 perms). Clean up temp files in finally block. Add tests for PIN not in args and mkstemp usage.
159 lines
6.3 KiB
Python
159 lines
6.3 KiB
Python
|
|
import subprocess, datetime, os, hashlib, ipaddress, re, tempfile
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.x509.oid import NameOID
|
|
from config import *
|
|
|
|
os.makedirs(TMP_DIR, exist_ok=True)
|
|
|
|
def get_root_pub_key():
|
|
with open(YK_PUB_ROOT, "rb") as f:
|
|
return serialization.load_pem_public_key(f.read())
|
|
|
|
def get_int_pub_key():
|
|
with open(YK_PUB_INT, "rb") as f:
|
|
return serialization.load_pem_public_key(f.read())
|
|
|
|
def get_root_ca_cert():
|
|
with open(ROOT_CA_PATH, "rb") as f:
|
|
return x509.load_pem_x509_certificate(f.read())
|
|
|
|
def der_len(n):
|
|
if n < 0x80: return bytes([n])
|
|
elif n < 0x100: return bytes([0x81, n])
|
|
return bytes([0x82, n>>8, n&0xff])
|
|
|
|
def _make_temp_file(prefix: str, data: bytes = None):
|
|
"""Create temp file with unpredictable name in TMP_DIR."""
|
|
fd, path = tempfile.mkstemp(prefix=prefix, dir=TMP_DIR)
|
|
try:
|
|
if data is not None:
|
|
os.write(fd, data)
|
|
finally:
|
|
os.close(fd)
|
|
os.chmod(path, 0o600)
|
|
return path
|
|
|
|
|
|
def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
|
|
tbs_file = _make_temp_file("tbs_", tbs_bytes)
|
|
sig_file = _make_temp_file("sig_")
|
|
os.unlink(sig_file)
|
|
try:
|
|
r = subprocess.run([
|
|
"sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
|
|
"--login", "--pin-source", "stdin",
|
|
"--sign", "--mechanism", "ECDSA-SHA384",
|
|
"--token-label", token_label,
|
|
"--label", "SIGN key",
|
|
"--input-file", tbs_file,
|
|
"--output-file", sig_file
|
|
], input=yk_pin, capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
return None, r.stderr
|
|
with open(sig_file, "rb") as f:
|
|
raw = f.read()
|
|
finally:
|
|
for f in (tbs_file, sig_file):
|
|
if os.path.exists(f):
|
|
os.unlink(f)
|
|
rb = raw[:48].lstrip(b"\x00") or b"\x00"
|
|
sb = raw[48:].lstrip(b"\x00") or b"\x00"
|
|
if rb[0] & 0x80: rb = b"\x00" + rb
|
|
if sb[0] & 0x80: sb = b"\x00" + sb
|
|
r_der = b"\x02" + bytes([len(rb)]) + rb
|
|
s_der = b"\x02" + bytes([len(sb)]) + sb
|
|
seq = b"\x30" + bytes([len(r_der+s_der)]) + r_der + s_der
|
|
bs = b"\x00" + seq
|
|
return b"\x03" + bytes([len(bs)]) + bs, None
|
|
|
|
def build_leaf_cert(cn, sans, days=365):
|
|
root_cert = get_root_ca_cert()
|
|
int_pub = get_int_pub_key()
|
|
root_pub = get_root_pub_key()
|
|
leaf_key = ec.generate_private_key(ec.SECP384R1())
|
|
subject = x509.Name([
|
|
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
|
|
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"),
|
|
x509.NameAttribute(NameOID.COMMON_NAME, cn),
|
|
])
|
|
issuer = x509.Name([
|
|
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
|
|
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"),
|
|
x509.NameAttribute(NameOID.COMMON_NAME, "certauth Intermediate CA"),
|
|
])
|
|
builder = (x509.CertificateBuilder()
|
|
.subject_name(subject).issuer_name(issuer)
|
|
.public_key(leaf_key.public_key())
|
|
.serial_number(x509.random_serial_number())
|
|
.not_valid_before(datetime.datetime.now(datetime.timezone.utc))
|
|
.not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=days))
|
|
.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
|
|
.add_extension(x509.KeyUsage(
|
|
digital_signature=True, key_encipherment=True,
|
|
key_cert_sign=False, crl_sign=False,
|
|
content_commitment=False, data_encipherment=False,
|
|
key_agreement=False, encipher_only=False, decipher_only=False), critical=True)
|
|
.add_extension(x509.ExtendedKeyUsage([
|
|
x509.oid.ExtendedKeyUsageOID.SERVER_AUTH,
|
|
x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH,
|
|
]), critical=False)
|
|
.add_extension(x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()), critical=False)
|
|
.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(int_pub), critical=False))
|
|
if sans:
|
|
san_list = []
|
|
for s in sans.split(","):
|
|
s = s.strip()
|
|
# Check if it's an IP address
|
|
if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", s):
|
|
san_list.append(x509.IPAddress(ipaddress.ip_address(s)))
|
|
else:
|
|
san_list.append(x509.DNSName(s))
|
|
builder = builder.add_extension(x509.SubjectAlternativeName(san_list), critical=False)
|
|
tmp = ec.generate_private_key(ec.SECP384R1())
|
|
temp = builder.sign(tmp, hashes.SHA384())
|
|
td = temp.public_bytes(serialization.Encoding.DER)
|
|
o = 1
|
|
if td[o] & 0x80: n = td[o] & 0x7f; o += 1 + n
|
|
else: o += 1
|
|
tbs_start = o
|
|
o += 1
|
|
if td[o] & 0x80: n = td[o] & 0x7f; tl = int.from_bytes(td[o+1:o+1+n], "big"); o += 1 + n
|
|
else: tl = td[o]; o += 1
|
|
tbs_end = o + tl
|
|
tbs_full = td[tbs_start:tbs_end]
|
|
alg_start = tbs_end
|
|
o2 = alg_start + 1
|
|
if td[o2] & 0x80: n = td[o2] & 0x7f; al = int.from_bytes(td[o2+1:o2+1+n], "big"); o2 += 1 + n
|
|
else: al = td[o2]; o2 += 1
|
|
alg_full = td[alg_start:o2+al]
|
|
new_sig, err = sign_tbs_with_yk(tbs_full, YK_INT_PIN)
|
|
if new_sig is None: return None, err
|
|
content = tbs_full + alg_full + new_sig
|
|
cl = len(content)
|
|
final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content
|
|
der_file = _make_temp_file("leaf_", final)
|
|
pem_file = _make_temp_file("leaf_pem_")
|
|
os.unlink(pem_file)
|
|
try:
|
|
r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
|
|
"-in", der_file, "-out", pem_file],
|
|
capture_output=True, text=True)
|
|
if r.returncode != 0: return None, r.stderr
|
|
with open(pem_file) as f: leaf_pem = f.read()
|
|
finally:
|
|
for f in (der_file, pem_file):
|
|
if os.path.exists(f):
|
|
os.unlink(f)
|
|
key_pem = leaf_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.PKCS8,
|
|
encryption_algorithm=serialization.NoEncryption()
|
|
).decode()
|
|
cert = x509.load_pem_x509_certificate(leaf_pem.encode())
|
|
serial = format(cert.serial_number, 'x')
|
|
return {"cert_pem": leaf_pem, "key_pem": key_pem, "serial": serial,
|
|
"expires_at": cert.not_valid_after.isoformat()}, None
|