import os from datetime import datetime, timedelta, timezone from cryptography import x509 from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa CRT_DIR = "crt" CA_KEY_FILE = os.path.join(CRT_DIR, "ca.key") CA_CRT_FILE = os.path.join(CRT_DIR, "ca.crt") def _ensure_crt_dir(): """Create crt/ directory if it doesn't exist.""" os.makedirs(CRT_DIR, exist_ok=True) def _save_pem(filename, data): """Write PEM data to file, converting \\n to real newlines.""" if isinstance(data, bytes): data = data.decode() with open(filename, "w") as f: for line in data.replace("\r", "").split("\\n"): f.write(line + "\n") def _build_subject(country, state, organization, locality, common_name, organizational_unit=None): """Build a cryptography X.509 Name from individual fields.""" attrs = [ x509.NameAttribute(NameOID.COUNTRY_NAME, country), x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, state), x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization), x509.NameAttribute(NameOID.LOCALITY_NAME, locality), x509.NameAttribute(NameOID.COMMON_NAME, common_name), ] if organizational_unit: attrs.append(x509.NameAttribute( NameOID.ORGANIZATIONAL_UNIT_NAME, organizational_unit)) return x509.Name(attrs) def generate_ca_cert(cert_country, cert_state, cert_organization, cert_locality, cert_common_name, valid_days=3650): """ Generate a self-signed CA certificate and save ca.key / ca.crt. Returns (cert_pem, key_pem). """ subject = _build_subject(cert_country, cert_state, cert_organization, cert_locality, cert_common_name) private_key = rsa.generate_private_key(public_exponent=65537, key_size=4096) now = datetime.now(timezone.utc) cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(subject) # self-signed .public_key(private_key.public_key()) .serial_number(1) .not_valid_before(now) .not_valid_after(now + timedelta(days=valid_days)) .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) .add_extension(x509.KeyUsage( key_cert_sign=True, crl_sign=True, digital_signature=False, content_commitment=False, key_encipherment=False, data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False, ), critical=True) .add_extension(x509.SubjectKeyIdentifier.from_public_key( private_key.public_key()), critical=False) .sign(private_key, hashes.SHA256()) ) cert_pem = cert.public_bytes(serialization.Encoding.PEM) key_pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), ) _ensure_crt_dir() _save_pem(CA_KEY_FILE, key_pem) _save_pem(CA_CRT_FILE, cert_pem) print(f"CA erstellt: {CA_KEY_FILE}, {CA_CRT_FILE}") return cert_pem, key_pem def load_ca(): """ Load CA certificate and private key from disk. Returns (ca_cert, ca_pkey) as cryptography objects, or None if missing. """ if not os.path.exists(CA_CRT_FILE) or not os.path.exists(CA_KEY_FILE): return None with open(CA_CRT_FILE, "rb") as f: ca_cert = x509.load_pem_x509_certificate(f.read()) with open(CA_KEY_FILE, "rb") as f: ca_pkey = serialization.load_pem_private_key(f.read(), password=None) return ca_cert, ca_pkey def generate_cert_signed_by_ca(ca_cert, ca_pkey, cert_country, cert_state, cert_organization, cert_locality, cert_organizational_unit, cert_common_name, valid_days, serial_number): """ Generate a certificate signed by the given CA. Returns (cert_pem, key_pem) as strings. """ subject = _build_subject(cert_country, cert_state, cert_organization, cert_locality, cert_common_name, cert_organizational_unit) private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) now = datetime.now(timezone.utc) cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(ca_cert.subject) .public_key(private_key.public_key()) .serial_number(serial_number) .not_valid_before(now) .not_valid_after(now + timedelta(days=valid_days)) .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) .add_extension(x509.KeyUsage( digital_signature=True, key_encipherment=True, key_cert_sign=False, crl_sign=False, content_commitment=False, data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False, ), critical=True) .add_extension(x509.SubjectKeyIdentifier.from_public_key( private_key.public_key()), critical=False) .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key( ca_pkey.public_key()), critical=False) .sign(ca_pkey, hashes.SHA256()) ) cert_pem = cert.public_bytes(serialization.Encoding.PEM) key_pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), ) return str(cert_pem.decode()), str(key_pem.decode()) def generate_and_save_cert(cert_country, cert_state, cert_organization, cert_locality, cert_organizational_unit, cert_common_name, valid_days=365, serial_number=100, ca_country="DE", ca_state="Sachsen", ca_organization="IchAG", ca_locality="Dresden", ca_common_name="My Root CA", cert_name="server"): """ High-level entry point: 1. If CA not present, create it. 2. Generate certificate signed by the CA. 3. Save key and cert (with optional chain) to .key / .crt. """ ca = load_ca() if ca is None: print("CA nicht gefunden – erstelle CA ...") generate_ca_cert(ca_country, ca_state, ca_organization, ca_locality, ca_common_name) ca = load_ca() if ca is None: raise RuntimeError("CA konnte nicht erstellt/geladen werden.") ca_cert, ca_pkey = ca cert_pem, key_pem = generate_cert_signed_by_ca( ca_cert, ca_pkey, cert_country, cert_state, cert_organization, cert_locality, cert_organizational_unit, cert_common_name, valid_days, serial_number, ) key_file = os.path.join(CRT_DIR, f"{cert_name}.key") crt_file = os.path.join(CRT_DIR, f"{cert_name}.crt") chain_file = os.path.join(CRT_DIR, f"{cert_name}-chain.crt") _ensure_crt_dir() _save_pem(key_file, key_pem) _save_pem(crt_file, cert_pem) # Chain file: server cert + CA cert chain = cert_pem.replace("\r", "") ca_pem = ca_cert.public_bytes(serialization.Encoding.PEM).decode() chain += "\n" + ca_pem _save_pem(chain_file, chain) print(f"Zertifikat erstellt: {key_file}, {crt_file}, {chain_file}") return cert_pem, key_pem if __name__ == "__main__": generate_and_save_cert( cert_country="DE", cert_state="Dresden", cert_organization="IchAG", cert_locality="Sachsen", cert_organizational_unit="mandan01", cert_common_name="mandan01", valid_days=365, serial_number=12, cert_name="test", )