381 lines
14 KiB
Python
381 lines
14 KiB
Python
import os
|
|
import datetime
|
|
import ipaddress
|
|
from typing import Tuple, List, Optional
|
|
from cryptography import x509
|
|
from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
|
|
|
|
def calculate_cert_fingerprint(cert_pem: str) -> str:
|
|
"""Computes SHA-256 fingerprint for a PEM-encoded X.509 certificate."""
|
|
cert = x509.load_pem_x509_certificate(cert_pem.encode("utf-8"))
|
|
return cert.fingerprint(hashes.SHA256()).hex().upper()
|
|
|
|
|
|
def is_cert_expiring_soon(cert_pem: str, threshold_days: int = 30) -> bool:
|
|
"""
|
|
Checks if a PEM-encoded X.509 certificate expires within `threshold_days` (or is already expired).
|
|
Returns True if expiring soon or expired, False otherwise.
|
|
"""
|
|
try:
|
|
cert = x509.load_pem_x509_certificate(cert_pem.encode("utf-8"))
|
|
expiry = getattr(cert, "not_valid_after_utc", None)
|
|
if expiry is None:
|
|
expiry = cert.not_valid_after.replace(tzinfo=datetime.timezone.utc)
|
|
now = datetime.datetime.now(datetime.timezone.utc)
|
|
return expiry <= (now + datetime.timedelta(days=threshold_days))
|
|
except Exception:
|
|
return True
|
|
|
|
|
|
def generate_ca_if_needed(
|
|
cert_dir: str = "certs",
|
|
common_name: str = "LOGAR-Root-CA",
|
|
force_renew: bool = False,
|
|
threshold_days: int = 30
|
|
) -> Tuple[x509.Certificate, rsa.RSAPrivateKey, str, str]:
|
|
"""
|
|
Loads an existing Root CA or generates a self-signed Root CA certificate and private key.
|
|
If existing CA cert is expiring within threshold_days (or force_renew is True), regenerates it.
|
|
Returns (ca_cert_obj, ca_key_obj, ca_cert_pem, ca_key_pem).
|
|
"""
|
|
os.makedirs(cert_dir, exist_ok=True)
|
|
ca_cert_path = os.path.join(cert_dir, "ca.crt")
|
|
ca_key_path = os.path.join(cert_dir, "ca.key")
|
|
|
|
if not force_renew and os.path.exists(ca_cert_path) and os.path.exists(ca_key_path):
|
|
with open(ca_cert_path, "r", encoding="utf-8") as f:
|
|
ca_cert_pem = f.read()
|
|
with open(ca_key_path, "r", encoding="utf-8") as f:
|
|
ca_key_pem = f.read()
|
|
try:
|
|
ca_cert = x509.load_pem_x509_certificate(ca_cert_pem.encode("utf-8"))
|
|
ca_key = serialization.load_pem_private_key(ca_key_pem.encode("utf-8"), password=None)
|
|
if not is_cert_expiring_soon(ca_cert_pem, threshold_days=threshold_days):
|
|
return ca_cert, ca_key, ca_cert_pem, ca_key_pem
|
|
except Exception:
|
|
pass
|
|
|
|
# Create timestamped backup of previous CA if present
|
|
if os.path.exists(ca_cert_path):
|
|
try:
|
|
timestamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S")
|
|
os.replace(ca_cert_path, f"{ca_cert_path}.{timestamp}.bak")
|
|
if os.path.exists(ca_key_path):
|
|
os.replace(ca_key_path, f"{ca_key_path}.{timestamp}.bak")
|
|
except Exception:
|
|
pass
|
|
|
|
# Generate RSA 4096 private key for Root CA
|
|
ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
|
|
subject = issuer = x509.Name([
|
|
x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"),
|
|
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"),
|
|
x509.NameAttribute(NameOID.COMMON_NAME, common_name),
|
|
])
|
|
|
|
now = datetime.datetime.now(datetime.timezone.utc)
|
|
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(now - datetime.timedelta(minutes=5))
|
|
.not_valid_after(now + datetime.timedelta(days=3650))
|
|
.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
|
|
.add_extension(
|
|
x509.KeyUsage(
|
|
digital_signature=True,
|
|
key_encipherment=False,
|
|
key_cert_sign=True,
|
|
crl_sign=True,
|
|
content_commitment=False,
|
|
data_encipherment=False,
|
|
key_agreement=False,
|
|
encipher_only=False,
|
|
decipher_only=False
|
|
),
|
|
critical=True
|
|
)
|
|
.add_extension(
|
|
x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()),
|
|
critical=False
|
|
)
|
|
.sign(ca_key, hashes.SHA256())
|
|
)
|
|
|
|
ca_cert_pem = ca_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
|
|
ca_key_pem = ca_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
|
encryption_algorithm=serialization.NoEncryption()
|
|
).decode("utf-8")
|
|
|
|
with open(ca_cert_path, "w", encoding="utf-8") as f:
|
|
f.write(ca_cert_pem)
|
|
with open(ca_key_path, "w", encoding="utf-8") as f:
|
|
f.write(ca_key_pem)
|
|
|
|
try:
|
|
os.chmod(ca_key_path, 0o600)
|
|
except Exception:
|
|
pass
|
|
|
|
return ca_cert, ca_key, ca_cert_pem, ca_key_pem
|
|
|
|
|
|
def generate_server_cert_if_needed(
|
|
ca_cert: x509.Certificate,
|
|
ca_key: rsa.RSAPrivateKey,
|
|
hostnames: Optional[List[str]] = None,
|
|
cert_dir: str = "certs",
|
|
days_valid: int = 825,
|
|
force_renew: bool = False,
|
|
threshold_days: int = 30
|
|
) -> Tuple[x509.Certificate, rsa.RSAPrivateKey, str, str]:
|
|
"""
|
|
Loads an existing server certificate or generates a new server TLS certificate signed by the Root CA.
|
|
If existing server cert is expiring within threshold_days (or force_renew is True), regenerates it.
|
|
Includes SANs for localhost, 127.0.0.1, and specified hostnames.
|
|
"""
|
|
os.makedirs(cert_dir, exist_ok=True)
|
|
server_cert_path = os.path.join(cert_dir, "server.crt")
|
|
server_key_path = os.path.join(cert_dir, "server.key")
|
|
|
|
if not force_renew and os.path.exists(server_cert_path) and os.path.exists(server_key_path):
|
|
with open(server_cert_path, "r", encoding="utf-8") as f:
|
|
server_cert_pem = f.read()
|
|
with open(server_key_path, "r", encoding="utf-8") as f:
|
|
server_key_pem = f.read()
|
|
try:
|
|
srv_cert = x509.load_pem_x509_certificate(server_cert_pem.encode("utf-8"))
|
|
srv_key = serialization.load_pem_private_key(server_key_pem.encode("utf-8"), password=None)
|
|
if not is_cert_expiring_soon(server_cert_pem, threshold_days=threshold_days):
|
|
return srv_cert, srv_key, server_cert_pem, server_key_pem
|
|
except Exception:
|
|
pass
|
|
|
|
# Create timestamped backup of previous server cert if present
|
|
if os.path.exists(server_cert_path):
|
|
try:
|
|
timestamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S")
|
|
os.replace(server_cert_path, f"{server_cert_path}.{timestamp}.bak")
|
|
if os.path.exists(server_key_path):
|
|
os.replace(server_key_path, f"{server_key_path}.{timestamp}.bak")
|
|
except Exception:
|
|
pass
|
|
|
|
server_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
subject = x509.Name([
|
|
x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"),
|
|
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"),
|
|
x509.NameAttribute(NameOID.COMMON_NAME, "LOGAR-Server-Hub"),
|
|
])
|
|
|
|
san_list = [
|
|
x509.DNSName("localhost"),
|
|
x509.DNSName("LOGAR-Server-Hub"),
|
|
x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")),
|
|
x509.IPAddress(ipaddress.IPv6Address("::1")),
|
|
]
|
|
|
|
if hostnames:
|
|
for host in hostnames:
|
|
if not host:
|
|
continue
|
|
try:
|
|
ip_obj = ipaddress.ip_address(host)
|
|
san_list.append(x509.IPAddress(ip_obj))
|
|
except ValueError:
|
|
san_list.append(x509.DNSName(host))
|
|
|
|
now = datetime.datetime.now(datetime.timezone.utc)
|
|
server_cert = (
|
|
x509.CertificateBuilder()
|
|
.subject_name(subject)
|
|
.issuer_name(ca_cert.subject)
|
|
.public_key(server_key.public_key())
|
|
.serial_number(x509.random_serial_number())
|
|
.not_valid_before(now - datetime.timedelta(minutes=5))
|
|
.not_valid_after(now + datetime.timedelta(days=days_valid))
|
|
.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.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]),
|
|
critical=False
|
|
)
|
|
.add_extension(
|
|
x509.SubjectKeyIdentifier.from_public_key(server_key.public_key()),
|
|
critical=False
|
|
)
|
|
.add_extension(
|
|
x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()),
|
|
critical=False
|
|
)
|
|
.add_extension(x509.SubjectAlternativeName(san_list), critical=False)
|
|
.sign(ca_key, hashes.SHA256())
|
|
)
|
|
|
|
server_cert_pem = server_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
|
|
server_key_pem = server_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
|
encryption_algorithm=serialization.NoEncryption()
|
|
).decode("utf-8")
|
|
|
|
with open(server_cert_path, "w", encoding="utf-8") as f:
|
|
f.write(server_cert_pem)
|
|
with open(server_key_path, "w", encoding="utf-8") as f:
|
|
f.write(server_key_pem)
|
|
|
|
try:
|
|
os.chmod(server_key_path, 0o600)
|
|
except Exception:
|
|
pass
|
|
|
|
return server_cert, server_key, server_cert_pem, server_key_pem
|
|
|
|
|
|
def issue_client_cert(
|
|
client_id: str,
|
|
ca_cert: x509.Certificate,
|
|
ca_key: rsa.RSAPrivateKey,
|
|
days_valid: int = 365
|
|
) -> Tuple[str, str]:
|
|
"""
|
|
Generates a 2048-bit RSA private key and signs an X.509 client certificate
|
|
with Common Name set to client_id.
|
|
Returns (cert_pem, key_pem).
|
|
"""
|
|
client_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
|
|
subject = x509.Name([
|
|
x509.NameAttribute(NameOID.COUNTRY_NAME, "AT"),
|
|
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LOGAR"),
|
|
x509.NameAttribute(NameOID.COMMON_NAME, client_id),
|
|
])
|
|
|
|
now = datetime.datetime.now(datetime.timezone.utc)
|
|
cert = (
|
|
x509.CertificateBuilder()
|
|
.subject_name(subject)
|
|
.issuer_name(ca_cert.subject)
|
|
.public_key(client_key.public_key())
|
|
.serial_number(x509.random_serial_number())
|
|
.not_valid_before(now - datetime.timedelta(minutes=5))
|
|
.not_valid_after(now + datetime.timedelta(days=days_valid))
|
|
.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.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]),
|
|
critical=False
|
|
)
|
|
.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())
|
|
)
|
|
|
|
cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
|
|
key_pem = client_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
|
encryption_algorithm=serialization.NoEncryption()
|
|
).decode("utf-8")
|
|
|
|
return cert_pem, key_pem
|
|
|
|
|
|
def check_and_renew_hub_pki(
|
|
cert_dir: str = "certs",
|
|
hostnames: Optional[List[str]] = None,
|
|
threshold_days: int = 30
|
|
) -> Tuple[bool, bool]:
|
|
"""
|
|
Evaluates expiration status of Root CA and Server TLS certificates.
|
|
If CA certificate is expiring within threshold_days (or missing):
|
|
- Regenerates Root CA.
|
|
- Automatically regenerates Server TLS certificate (since CA issuer changed).
|
|
- Returns (ca_renewed=True, server_renewed=True)
|
|
Else if Server TLS certificate is expiring within threshold_days (or missing):
|
|
- Regenerates Server TLS certificate signed by existing Root CA.
|
|
- Returns (ca_renewed=False, server_renewed=True)
|
|
Otherwise:
|
|
- Returns (False, False)
|
|
"""
|
|
os.makedirs(cert_dir, exist_ok=True)
|
|
ca_cert_path = os.path.join(cert_dir, "ca.crt")
|
|
server_cert_path = os.path.join(cert_dir, "server.crt")
|
|
|
|
renew_ca = False
|
|
renew_server = False
|
|
|
|
if not os.path.exists(ca_cert_path):
|
|
renew_ca = True
|
|
else:
|
|
try:
|
|
with open(ca_cert_path, "r", encoding="utf-8") as f:
|
|
ca_pem = f.read()
|
|
if is_cert_expiring_soon(ca_pem, threshold_days=threshold_days):
|
|
renew_ca = True
|
|
except Exception:
|
|
renew_ca = True
|
|
|
|
if renew_ca:
|
|
ca_cert, ca_key, _, _ = generate_ca_if_needed(cert_dir=cert_dir, force_renew=True)
|
|
generate_server_cert_if_needed(ca_cert, ca_key, hostnames=hostnames, cert_dir=cert_dir, force_renew=True)
|
|
return True, True
|
|
|
|
if not os.path.exists(server_cert_path):
|
|
renew_server = True
|
|
else:
|
|
try:
|
|
with open(server_cert_path, "r", encoding="utf-8") as f:
|
|
srv_pem = f.read()
|
|
if is_cert_expiring_soon(srv_pem, threshold_days=threshold_days):
|
|
renew_server = True
|
|
except Exception:
|
|
renew_server = True
|
|
|
|
if renew_server:
|
|
ca_cert, ca_key, _, _ = generate_ca_if_needed(cert_dir=cert_dir, force_renew=False)
|
|
generate_server_cert_if_needed(ca_cert, ca_key, hostnames=hostnames, cert_dir=cert_dir, force_renew=True)
|
|
return False, True
|
|
|
|
return False, False
|
|
|