#!/usr/bin/env python3 """ Fleet CA and Agent Certificate Generator — #806 Generates a Fleet CA and per-agent TLS certificates for mutual TLS authentication between fleet agents. Usage: # Generate Fleet CA python scripts/generate_fleet_ca.py --ca-dir ./fleet-ca # Generate agent cert python scripts/generate_fleet_ca.py --ca-dir ./fleet-ca --agent timmy python scripts/generate_fleet_ca.py --ca-dir ./fleet-ca --agent allegro python scripts/generate_fleet_ca.py --ca-dir ./fleet-ca --agent ezra # Generate all fleet certs python scripts/generate_fleet_ca.py --ca-dir ./fleet-ca --all """ import argparse import datetime import os import sys from pathlib import Path try: from cryptography import x509 from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa HAS_CRYPTO = True except ImportError: HAS_CRYPTO = False FLEET_AGENTS = ["timmy", "allegro", "ezra", "bezalel"] CA_VALIDITY_DAYS = 3650 # 10 years CERT_VALIDITY_DAYS = 365 # 1 year KEY_SIZE = 2048 def generate_ca(ca_dir: Path) -> tuple: """Generate Fleet CA key and certificate.""" ca_dir.mkdir(parents=True, exist_ok=True) # Generate CA key ca_key = rsa.generate_private_key(public_exponent=65537, key_size=KEY_SIZE) # Generate CA cert subject = issuer = x509.Name([ x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Timmy Foundation"), x509.NameAttribute(NameOID.COMMON_NAME, "Fleet CA"), ]) ca_cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(issuer) .public_key(ca_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.datetime.utcnow()) .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=CA_VALIDITY_DAYS)) .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) .add_extension( x509.KeyUsage( digital_signature=True, key_cert_sign=True, crl_sign=True, content_commitment=False, key_encipherment=False, data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False, ), critical=True, ) .sign(ca_key, hashes.SHA256()) ) # Save ca_key_path = ca_dir / "fleet-ca.key" ca_cert_path = ca_dir / "fleet-ca.crt" ca_key_path.write_bytes(ca_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), )) ca_cert_path.write_bytes(ca_cert.public_bytes(serialization.Encoding.PEM)) # Secure permissions os.chmod(ca_key_path, 0o600) os.chmod(ca_cert_path, 0o644) print(f"CA key: {ca_key_path}") print(f"CA cert: {ca_cert_path}") return ca_key, ca_cert def generate_agent_cert(ca_dir: Path, agent_name: str, ca_key=None, ca_cert=None) -> tuple: """Generate TLS certificate for an agent.""" agent_dir = ca_dir / agent_name agent_dir.mkdir(parents=True, exist_ok=True) # Load CA if not provided if ca_key is None or ca_cert is None: ca_key_path = ca_dir / "fleet-ca.key" ca_cert_path = ca_dir / "fleet-ca.crt" if not ca_key_path.exists() or not ca_cert_path.exists(): print(f"Error: CA not found in {ca_dir}. Run --ca first.") return None, None with open(ca_key_path, "rb") as f: ca_key = serialization.load_pem_private_key(f.read(), password=None) with open(ca_cert_path, "rb") as f: ca_cert = x509.load_pem_x509_certificate(f.read()) # Generate agent key agent_key = rsa.generate_private_key(public_exponent=65537, key_size=KEY_SIZE) # Generate agent cert subject = x509.Name([ x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Timmy Foundation"), x509.NameAttribute(NameOID.COMMON_NAME, f"agent-{agent_name}"), ]) agent_cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(ca_cert.subject) .public_key(agent_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.datetime.utcnow()) .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=CERT_VALIDITY_DAYS)) .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) .add_extension( x509.SubjectAlternativeName([ x509.DNSName(f"{agent_name}.fleet.local"), x509.DNSName(f"{agent_name}"), x509.DNSName("localhost"), ]), critical=False, ) .sign(ca_key, hashes.SHA256()) ) # Save key_path = agent_dir / f"{agent_name}.key" cert_path = agent_dir / f"{agent_name}.crt" key_path.write_bytes(agent_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), )) cert_path.write_bytes(agent_cert.public_bytes(serialization.Encoding.PEM)) # Copy CA cert to agent dir ca_copy = agent_dir / "fleet-ca.crt" ca_copy.write_bytes(ca_cert.public_bytes(serialization.Encoding.PEM)) # Secure permissions os.chmod(key_path, 0o600) os.chmod(cert_path, 0o644) print(f"Agent {agent_name}:") print(f" Key: {key_path}") print(f" Cert: {cert_path}") print(f" CA: {ca_copy}") return agent_key, agent_cert def main(): parser = argparse.ArgumentParser(description="Fleet CA and Agent Certificate Generator") parser.add_argument("--ca-dir", type=Path, default=Path("./fleet-ca"), help="CA directory") parser.add_argument("--ca", action="store_true", help="Generate Fleet CA") parser.add_argument("--agent", type=str, help="Generate cert for agent") parser.add_argument("--all", action="store_true", help="Generate certs for all fleet agents") args = parser.parse_args() if not HAS_CRYPTO: print("Error: cryptography package required. pip install cryptography") sys.exit(1) if args.ca: generate_ca(args.ca_dir) if args.agent: generate_agent_cert(args.ca_dir, args.agent) if args.all: # Generate CA first if not exists ca_key_path = args.ca_dir / "fleet-ca.key" if not ca_key_path.exists(): ca_key, ca_cert = generate_ca(args.ca_dir) else: ca_key, ca_cert = None, None for agent in FLEET_AGENTS: generate_agent_cert(args.ca_dir, agent, ca_key, ca_cert) if not args.ca and not args.agent and not args.all: parser.print_help() if __name__ == "__main__": main()