"""Certificate signing via the YubiKey-held Intermediate CA key. The private key never leaves the YubiKey. Signing works in three steps: 1. Build the TBS (to-be-signed) certificate bytes. 2. Hand the TBS to ``pkcs11-tool`` (OpenSC PKCS#11) which signs it on the YubiKey with ECDSA-SHA384. 3. Reassemble the certificate DER from TBS + signature algorithm + YubiKey signature, then verify it against the Intermediate CA public key before trusting the result. """ import logging import os import subprocess import tempfile from datetime import datetime, timedelta, timezone from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption from cryptography.x509.oid import NameOID from config import ( CA_COUNTRY, CA_ORG, INT_CA_PATH, PKCS11_MODULE, PKCS11_TOKEN_LABEL, YK_INT_PIN, ) from models import get_db logger = logging.getLogger(__name__) # Scratch dir for pkcs11-tool input/output files (per-process, unlinked after). PKCS11_TMP = tempfile.mkdtemp(prefix="certauth_") def _der_read_len(data: bytes, i: int) -> tuple: """Read a DER length field at offset i -> (length, offset_after_field).""" b0 = data[i] if b0 < 0x80: return b0, i + 1 n = b0 & 0x7F return int.from_bytes(data[i + 1:i + 1 + n], "big"), i + 1 + n def _der_seq(body: bytes) -> bytes: if len(body) < 0x80: return b"\x30" + bytes([len(body)]) + body lb = len(body).to_bytes((len(body).bit_length() + 7) // 8, "big") return b"\x30" + bytes([0x80 | len(lb)]) + lb + body def _split_cert_der(der: bytes): """Split a certificate DER into (tbs, signature_algorithm, signature).""" if der[0] != 0x30: raise ValueError("not a DER certificate") _, body_start = _der_read_len(der, 1) # tbsCertificate SEQUENCE if der[body_start] != 0x30: raise ValueError("bad tbsCertificate tag") tbs_len, tbs_len_end = _der_read_len(der, body_start + 1) tbs_end = tbs_len_end + tbs_len tbs = der[body_start:tbs_end] # signatureAlgorithm SEQUENCE if der[tbs_end] != 0x30: raise ValueError("bad signatureAlgorithm tag") alg_len, alg_len_end = _der_read_len(der, tbs_end + 1) alg_end = alg_len_end + alg_len alg = der[tbs_end:alg_end] return tbs, alg, der[alg_end:] def get_int_ca_cert(): with open(INT_CA_PATH, "rb") as f: return x509.load_pem_x509_certificate(f.read()) def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label=None): """Sign DER bytes with the YubiKey SIGN key via pkcs11-tool. Returns (signature_BIT_STRING, None) on success or (None, error). """ with tempfile.NamedTemporaryFile(suffix=".der", delete=False, dir=PKCS11_TMP) as tbs_file: tbs_file.write(tbs_bytes) tbs_path = tbs_file.name sig_path = os.path.join(PKCS11_TMP, f"sig_{os.path.basename(tbs_path)}") cmd = [ "sudo", "pkcs11-tool", "--module", PKCS11_MODULE, "--login", "--pin-source", "stdin", "--sign", "--mechanism", "ECDSA-SHA384", "--label", "SIGN key", "--input-file", tbs_path, "--output-file", sig_path, ] label = token_label if token_label is not None else PKCS11_TOKEN_LABEL if label: cmd[3:3] = ["--token-label", label] try: r = subprocess.run( cmd, input=yk_pin.encode(), capture_output=True, text=True, timeout=30, ) if r.returncode != 0: logger.error("YubiKey signing failed: %s", r.stderr[:200]) return None, "Signing failed" with open(sig_path, "rb") as f: raw = f.read() # pkcs11-tool returns raw r||s (48 bytes each for P-384); wrap in a # BIT STRING carrying the DER ECDSA-Sig-Value. 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 finally: for f in [tbs_path, sig_path]: try: os.unlink(f) except OSError: pass def build_leaf_cert(cn, sans, days=365): """Issue a leaf certificate for cn, signed by the Intermediate CA. Returns ({serial, cert_pem, key_pem, expires_at}, None) or (None, error). The private key stays on the server (leaf keys are not hardware-backed); only the CA signing key requires the YubiKey. """ int_cert = get_int_ca_cert() leaf_key = ec.generate_private_key(ec.SECP384R1()) subject = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, CA_COUNTRY), x509.NameAttribute(NameOID.ORGANIZATION_NAME, CA_ORG), x509.NameAttribute(NameOID.COMMON_NAME, cn), ]) builder = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(int_cert.subject) .public_key(leaf_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.now(timezone.utc)) .not_valid_after(datetime.now(timezone.utc) + timedelta(days=days)) .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([ x509.OID_SERVER_AUTH, x509.OID_CLIENT_AUTH, ]), critical=False, ) ) if sans: san_names = [x509.DNSName(s.strip()) for s in sans.split(",") if s.strip()] if san_names: builder = builder.add_extension( x509.SubjectAlternativeName(san_names), critical=False, ) # 1) TBS bytes: sign with a throwaway key of identical parameters — # the TBS block is independent of the signing key. dummy_key = ec.generate_private_key(ec.SECP384R1()) tbs, sig_alg, _ = _split_cert_der(builder.sign(dummy_key, hashes.SHA384())) # 2) Real signature from the YubiKey. yk_sig, err = sign_tbs_with_yk(tbs, YK_INT_PIN) if err: return None, err # 3) Reassemble and verify before trusting. cert = x509.load_der_x509_certificate(_der_seq(tbs + sig_alg + yk_sig)) cert.verify(int_cert.public_key()) key_pem = leaf_key.private_bytes( encoding=Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=NoEncryption(), ).decode() return ( { "serial": hex(cert.serial_number), "cert_pem": cert.public_bytes(Encoding.PEM).decode(), "key_pem": key_pem, "expires_at": cert.not_valid_after_utc.isoformat(), }, None, ) def revoke_certificate(serial_hex: str, reason: str = "key_compromise"): conn = get_db() conn.execute( "INSERT OR REPLACE INTO crl (serial, revoked_at, reason) VALUES (?, ?, ?)", (serial_hex, datetime.now(timezone.utc).isoformat(), reason), ) conn.commit() conn.close() logger.info("Certificate %s revoked: %s", serial_hex, reason)