certauth/api/signing.py

import subprocess
import os
import logging
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, BestAvailableEncryption
from cryptography.x509.oid import NameOID, ExtensionOID
from config import *

logger = logging.getLogger(__name__)

TMP_DIR = tempfile.mkdtemp(prefix="certauth_")
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 sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
    with tempfile.NamedTemporaryFile(suffix=".der", delete=False, dir=TMP_DIR) as tbs_file:
        tbs_file.write(tbs_bytes)
        tbs_path = tbs_file.name
    sig_path = os.path.join(TMP_DIR, f"sig_{os.path.basename(tbs_path)}")
    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_path,
                "--output-file", sig_path,
            ],
            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()
        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):
    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.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,
            )
    tbs_bytes = builder.signature_algorithm_oid
    cert_bytes = builder.sign(leaf_key, hashes.SHA384())
    serial = x509.load_der_x509_certificate(cert_bytes).serial_number
    key_pem = leaf_key.private_bytes(
        encoding=Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=NoEncryption(),
    ).decode()
    return (
        {
            "serial": hex(serial),
            "cert_pem": cert_bytes.public_bytes(Encoding.PEM).decode(),
            "key_pem": key_pem,
            "expires_at": (datetime.now(timezone.utc) + timedelta(days=days)).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)