OSGAccountServer/deploy/mtls/generate_admin_certificates.py

#!/usr/bin/env python3
"""Generate a dedicated admin client CA and one short-lived client certificate."""

from __future__ import annotations

import argparse
import base64
import datetime as dt
import os
from pathlib import Path

from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID


def private_write(path: Path, value: bytes) -> None:
    path.write_bytes(value)
    path.chmod(0o600)


def public_write(path: Path, value: bytes) -> None:
    path.write_bytes(value)
    path.chmod(0o644)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("output_directory", type=Path)
    parser.add_argument("--client-name", default="osg-admin-owner")
    args = parser.parse_args()

    output = args.output_directory.expanduser().resolve()
    if output.exists():
        raise SystemExit("output_directory_already_exists")
    output.mkdir(mode=0o700, parents=False)

    now = dt.datetime.now(dt.timezone.utc)
    ca_key = ec.generate_private_key(ec.SECP384R1())
    ca_subject = x509.Name(
        [
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, "OSG Admin"),
            x509.NameAttribute(NameOID.COMMON_NAME, "OSG Admin Client CA"),
        ]
    )
    ca_certificate = (
        x509.CertificateBuilder()
        .subject_name(ca_subject)
        .issuer_name(ca_subject)
        .public_key(ca_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - dt.timedelta(minutes=5))
        .not_valid_after(now + dt.timedelta(days=3650))
        .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
        .add_extension(
            x509.KeyUsage(
                digital_signature=False,
                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,
        )
        .add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), False)
        .sign(ca_key, hashes.SHA384())
    )

    client_key = ec.generate_private_key(ec.SECP256R1())
    client_subject = x509.Name(
        [
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, "OSG Admin"),
            x509.NameAttribute(NameOID.COMMON_NAME, args.client_name),
        ]
    )
    client_certificate = (
        x509.CertificateBuilder()
        .subject_name(client_subject)
        .issuer_name(ca_certificate.subject)
        .public_key(client_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - dt.timedelta(minutes=5))
        .not_valid_after(now + dt.timedelta(days=180))
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
        .add_extension(
            x509.KeyUsage(
                digital_signature=True,
                content_commitment=False,
                key_encipherment=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(client_key.public_key()),
            critical=False,
        )
        .add_extension(
            x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()),
            critical=False,
        )
        .sign(ca_key, hashes.SHA256())
    )

    password = base64.urlsafe_b64encode(os.urandom(24)).rstrip(b"=")
    private_write(
        output / "admin-client-ca-key.pem",
        ca_key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        ),
    )
    public_write(
        output / "admin-client-ca.pem",
        ca_certificate.public_bytes(serialization.Encoding.PEM),
    )
    private_write(
        output / "admin-client-key.pem",
        client_key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        ),
    )
    public_write(
        output / "admin-client.pem",
        client_certificate.public_bytes(serialization.Encoding.PEM),
    )
    private_write(
        output / "admin-client.p12",
        pkcs12.serialize_key_and_certificates(
            args.client_name.encode(),
            client_key,
            client_certificate,
            [ca_certificate],
            serialization.BestAvailableEncryption(password),
        ),
    )
    private_write(output / "admin-client-p12-password.txt", password + b"\n")
    print("certificates_created=true")


if __name__ == "__main__":
    main()