Harden admin deployment and local acceptance

Enforce mTLS and least-privilege runtime boundaries while adding repeatable MySQL 8.4 and Docker smoke checks that require no production secrets.
2026-08-17 15:20:46 +08:00
parent 1a9c518f96
commit 405a2cfc0f
17 changed files with 1480 additions and 37 deletions
@@ -0,0 +1,160 @@
+#!/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()