218 lines
7.9 KiB
Python
Executable File
218 lines
7.9 KiB
Python
Executable File
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 <cert_name>.key / <cert_name>.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",
|
||
) |