feat(pki): add certificate expiration check and auto-renewal in server_enrollment.py

2026-09-04 22:49:54 +02:00
parent cfc633c398
commit 2cff629e23
1 changed files with 123 additions and 10 deletions
@@ -14,23 +14,59 @@ def calculate_cert_fingerprint(cert_pem: str) -> str:
    return cert.fingerprint(hashes.SHA256()).hex().upper()


-def generate_ca_if_needed(cert_dir: str = "certs", common_name: str = "LOGAR-Root-CA") -> Tuple[x509.Certificate, rsa.RSAPrivateKey, str, str]:
+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 os.path.exists(ca_cert_path) and os.path.exists(ca_key_path):
+    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)
@@ -96,24 +132,41 @@ def generate_server_cert_if_needed(
    ca_key: rsa.RSAPrivateKey,
    hostnames: Optional[List[str]] = None,
    cert_dir: str = "certs",
-    days_valid: int = 825
+    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 os.path.exists(server_cert_path) and os.path.exists(server_key_path):
+    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([
@@ -265,3 +318,63 @@ def issue_client_cert(
    ).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
+