diff --git a/src/server_enrollment.py b/src/server_enrollment.py index 2ac60a0..b709698 100644 --- a/src/server_enrollment.py +++ b/src/server_enrollment.py @@ -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() - 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) - return ca_cert, ca_key, ca_cert_pem, ca_key_pem + 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() - 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) - return srv_cert, srv_key, server_cert_pem, server_key_pem + 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 +