docs: add certificate auto-renewal details to out/linux_client/README.md

README.md

@@ -114,6 +114,15 @@ graph TB
   - If `active_seats >= max_seats`, the hub rejects registration with `HTTP 403 (License seat limit reached)`.
   - Existing registered clients can re-enroll / renew seamlessly without consuming additional seats.
 
+### 4. In-Flight Certificate Watchdog & Automated Self-Healing Renewal
+- **Continuous Hub PKI Watchdog**:
+  - The server hub runs a continuous background watchdog coroutine (`cert_validity_watchdog`, running every 12 hours) alongside startup checks.
+  - The hub automatically inspects expiration dates of both the Root CA (`ca.crt`) and the Server TLS certificate (`server.crt`).
+  - If either certificate is within 30 days of expiration, the server regenerates certificates (backing up previous keys as `ca.crt.<timestamp>.bak`) and dynamically reloads its active `ssl.SSLContext` in memory without dropping socket connections or restarting the service.
+- **Client Proactive Check & Reactive Self-Healing**:
+  - **Proactive Renewal**: Edge clients inspect `client.crt` before every run cycle. If the certificate expires in less than 30 days, it automatically contacts `/api/client/enroll` to renew its certificate.
+  - **Reactive Self-Healing**: If the server hub Root CA rotates or a handshake fails with `ssl.SSLError` / `SSLCertVerificationError`, edge clients catch the verification exception, re-bootstrap certificate enrollment against the hub, and re-establish the connection cleanly without human intervention.
+
 ---
 
 ## Cloud-Side Temporal Persistence & 4-Run Rule

out/linux_client/README.md

@@ -11,6 +11,7 @@ Standalone compiled binary and automated systemd service distribution for Linux
 - **Pre-compiled & Dependency-Free**: Ships as a standalone native binary (`Linux_Client.bin`). No Python environment, pip packages, or GnuPG binaries are required on the host.
 - **Mutual TLS 1.3 (mTLS) Ingestion**: Streams directly over hardware-authenticated TLS 1.3 sockets with machine-bound client certificates.
 - **Automated Client Enrollment**: On first run with an `enrollment_secret`, the client automatically calls `POST /api/client/enroll` on the hub, saves its certificates into `/etc/logar/certs/`, and establishes secure mTLS streaming.
+- **Proactive Expiry Check & Reactive Self-Healing**: Before each run, the client evaluates `client.crt` validity. If within 30 days of expiry, it automatically contacts the hub to renew certificates. If the server Root CA rotates or a TLS handshake error occurs, the client catch-heals by re-enrolling immediately and re-establishing connection without human intervention.
 - **Source-Level Filtering**: Retains events spanning `INFO`, `WARNING`, and `ERROR` (`journalctl -p warning`). Drops debug noise and skips events older than 24 hours.
 - **State Tracking & Deduplication**: Maintains persistent client state in `client_state.json` (tracking systemd journalctl cursors and microsecond timestamps) so every log record is forwarded exactly once without duplicates.
 - **Fail-Safe State Commit**: State is committed only when the server returns a verified `success` response. In the event of a network outage, state remains unchanged and unsent events are retried automatically on the next run.

out/linux_server/README.md

@@ -12,6 +12,7 @@ Standalone compiled binary and automated systemd service distribution for Linux
 - **Pre-compiled & Dependency-Free**: Ships as a standalone native Linux ELF binary (`Server.bin`). No Python runtime, pip dependencies, or GnuPG binaries are required on the host system.
 - **Mutual TLS 1.3 (mTLS) Ingestion (Port 9443)**: Enforces mutual TLS 1.3 authentication for all incoming edge connections. Validates client certificates against an internal Root CA and verifies active licensing in SQLite.
 - **Dynamic PKI & License Accounting**: Built-in Root CA generates server TLS certificates with SANs and dynamically signs client certificates via `POST /api/client/enroll` while enforcing seat limits (`max_seats`).
+- **In-Flight Certificate Watchdog & Dynamic Reloading**: Continuously monitors Root CA (`ca.crt`) and Server TLS certificate (`server.crt`) validity in the background (every 12 hours). When nearing expiration (< 30 days), certificates are automatically regenerated with timestamped backups, and active `ssl.SSLContext` structures are reloaded dynamically without dropping socket connections or restarting the systemd service.
 - **Warning Persistence & Immediate Error Routing**: High-severity `ERROR`, `CRITICAL`, and `FATAL` events are promoted to `VERIFIED` immediately on their first occurrence. Operational `WARNING` and `INFO` events require persistence across at least 4 distinct client transmission cycles within a rolling 12-hour evaluation window.
 - **Embedded Hermes Reporting & Management API (Port 8443)**: Integrated REST API exposing `/api/hermes/report`, `/api/clients`, and `/api/client/enroll`.
 - **State Database**: Stores issue lifecycle records, client telemetry, and licensing quotas in a local SQLite database (`logar_state.db`).

out/win_client/README.md

@@ -11,6 +11,7 @@ Standalone compiled executable and installer distribution for Windows Server and
 - **Pre-compiled & Dependency-Free**: Ships as a standalone native Windows executable (`Win_Client.exe`) or full installer (`LOGAR-Client-Setup.exe`). No Python installation, pip packages, or GnuPG binaries are required on the host.
 - **Mutual TLS 1.3 (mTLS) Ingestion**: Streams directly over hardware-authenticated TLS 1.3 sockets with hardware/machine-bound client certificates.
 - **Automated Client Enrollment**: On first run with an `enrollment_secret`, the client automatically calls `POST /api/client/enroll` on the hub, saves its certificates into `certs/`, and establishes secure mTLS streaming.
+- **Proactive Expiry Check & Reactive Self-Healing**: Before each run, the client evaluates `client.crt` validity. If within 30 days of expiry, it automatically contacts the hub to renew certificates. If the server Root CA rotates or a TLS handshake error occurs, the client catch-heals by re-enrolling immediately and re-establishing connection without human intervention.
 - **Source-Level Filtering**: Retains events spanning `INFO`, `WARNING`, and `ERROR`. Strips audit events and debug noise, skipping events older than 24 hours.
 - **State Tracking & Deduplication**: Maintains persistent client state in `client_state.json` (tracking event record numbers and timestamp signatures) so every log record is forwarded exactly once without duplicates.
 - **Fail-Safe State Commit**: State is committed only when the server returns a verified `success` response. In the event of a network outage, state remains unchanged and unsent events are retried automatically on the next run.

out/win_server/README.md

@@ -12,6 +12,7 @@ Standalone compiled executable and installer distribution for Windows Server env
 - **Pre-compiled & Dependency-Free**: Ships as a standalone Windows executable (`Server.exe`) or full installer (`LOGAR-Server-Setup.exe`). No Python installation, pip packages, or GnuPG binaries are required on Windows Server.
 - **Mutual TLS 1.3 (mTLS) Ingestion (Port 9443)**: Enforces mutual TLS 1.3 authentication for all incoming edge connections. Validates client certificates against an internal Root CA and verifies active licensing in SQLite.
 - **Dynamic PKI & License Accounting**: Built-in Root CA generates server TLS certificates with SANs and dynamically signs client certificates via `POST /api/client/enroll` while enforcing seat limits (`max_seats`).
+- **In-Flight Certificate Watchdog & Dynamic Reloading**: Continuously monitors Root CA (`ca.crt`) and Server TLS certificate (`server.crt`) validity in the background (every 12 hours). When nearing expiration (< 30 days), certificates are automatically regenerated with timestamped backups, and active `ssl.SSLContext` structures are reloaded dynamically without dropping socket connections or restarting the Windows service.
 - **Warning Persistence & Immediate Error Routing**: High-severity `ERROR`, `CRITICAL`, and `FATAL` events are promoted to `VERIFIED` immediately on their first occurrence. Operational `WARNING` and `INFO` events require persistence across at least 4 distinct transmission cycles within a rolling 12-hour evaluation window.
 - **Embedded Hermes Reporting & Management API (Port 8443)**: Integrated REST API exposing `/api/hermes/report`, `/api/clients`, and `/api/client/enroll`.
 - **State Database**: Stores issue lifecycle records, client telemetry, and licensing quotas in a local SQLite database (`logar_state.db`).

src/Linux_Client.py

@@ -20,17 +20,46 @@ CONFIG_FILE_NAME = "client_config.json"
 STATE_FILE_NAME = "client_state.json"
 
 
-def enroll_client_if_needed(hub_url: str, enrollment_secret: str, cert_dir: str, client_id: str, hostname: str, os_type: str = "linux"):
-    """Bootstraps client enrollment if certificates are missing."""
+def is_cert_expiring_soon(cert_path: str, threshold_days: int = 30) -> bool:
+    """Checks if client certificate at cert_path is expiring within threshold_days."""
+    if not os.path.exists(cert_path):
+        return True
+    try:
+        from cryptography import x509
+        with open(cert_path, "r", encoding="utf-8") as f:
+            cert = x509.load_pem_x509_certificate(f.read().encode("utf-8"))
+        expiry = getattr(cert, "not_valid_after_utc", None)
+        if expiry is None:
+            expiry = cert.not_valid_after.replace(tzinfo=timezone.utc)
+        now = datetime.now(timezone.utc)
+        return expiry <= (now + timedelta(days=threshold_days))
+    except Exception:
+        return True
+
+
+def enroll_client_if_needed(
+    hub_url: str,
+    enrollment_secret: str,
+    cert_dir: str,
+    client_id: str,
+    hostname: str,
+    os_type: str = "linux",
+    force_renew: bool = False,
+    threshold_days: int = 30
+):
+    """Bootstraps client enrollment if certificates are missing or expiring soon."""
     os.makedirs(cert_dir, exist_ok=True)
     ca_path = os.path.join(cert_dir, "ca.crt")
     cert_path = os.path.join(cert_dir, "client.crt")
     key_path = os.path.join(cert_dir, "client.key")
 
-    if os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path):
-        return True
+    if not force_renew and os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path):
+        if not is_cert_expiring_soon(cert_path, threshold_days=threshold_days):
+            return True
+        print(f"[*] Client certificate at {cert_path} is expiring within {threshold_days} days. Auto-renewing...")
 
-    print(f"[*] Bootstrapping client enrollment with LOGAR Hub at {hub_url}...")
+    action_name = "re-enrolling" if os.path.exists(cert_path) else "enrolling"
+    print(f"[*] Bootstrapping client {action_name} with LOGAR Hub at {hub_url}...")
     enroll_endpoint = f"{hub_url.rstrip('/')}/api/client/enroll"
     payload = {
         "client_id": client_id,
@@ -60,7 +89,7 @@ def enroll_client_if_needed(hub_url: str, enrollment_secret: str, cert_dir: str,
     except Exception:
         pass
 
-    print(f"[+] Client enrolled successfully! Certificates saved to {os.path.abspath(cert_dir)}")
+    print(f"[+] Client certificates updated successfully in {os.path.abspath(cert_dir)}")
     return True
 
 
@@ -302,6 +331,7 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
     machine_id = get_machine_identifier()
 
     # Attempt automatic enrollment bootstrap if certs are missing and secret is provided
+    hub_url = None
     if enrollment_secret:
         hermes_host = config.get("hermes_host", server_host)
         hermes_port = config.get("hermes_port", 8443)
@@ -318,7 +348,22 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
 
     if has_mtls_certs:
         print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over mTLS (TLS 1.3)...")
-        with get_tls_socket(server_host, server_port, cert_dir) as sock:
+        sock = None
+        try:
+            sock = get_tls_socket(server_host, server_port, cert_dir)
+        except (ssl.SSLError, ssl.CertificateError, ConnectionResetError) as tls_err:
+            if enrollment_secret and hub_url:
+                print(f"[!] TLS handshake error ({tls_err}). Re-enrolling with LOGAR Hub...")
+                try:
+                    enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="linux", force_renew=True)
+                    sock = get_tls_socket(server_host, server_port, cert_dir)
+                except Exception as retry_err:
+                    print(f"[!] Re-enrollment or reconnection retry failed: {retry_err}")
+                    raise
+            else:
+                raise
+
+        with sock:
             payload = {
                 "server": machine_id,
                 "timestamp": datetime.now(timezone.utc).isoformat(),

src/Server.py

@@ -322,6 +322,72 @@ def init_mtls_server_context(cert_dir: str = "certs") -> ssl.SSLContext:
     return ctx
 
 
+def reload_mtls_context(ssl_ctx: ssl.SSLContext, cert_dir: str = "certs"):
+    """
+    Dynamically reloads server certificate chain and Root CA in an active SSLContext.
+    Allows in-flight TLS certificate rotation without dropping the listening socket.
+    """
+    ca_file = os.path.join(cert_dir, "ca.crt")
+    srv_cert = os.path.join(cert_dir, "server.crt")
+    srv_key = os.path.join(cert_dir, "server.key")
+
+    ssl_ctx.load_cert_chain(certfile=srv_cert, keyfile=srv_key)
+    ssl_ctx.load_verify_locations(cafile=ca_file)
+
+
+def check_and_rotate_server_certs(
+    cert_dir: str = "certs",
+    hostnames: Optional[List[str]] = None,
+    threshold_days: int = 30
+) -> bool:
+    """
+    Checks if Root CA or server TLS certificate are expiring within threshold_days.
+    If so, regenerates them, dynamically reloads the active SSLContext in-place,
+    and updates the server's in-memory CA reference so future enrollments use the new CA.
+    Returns True if renewed/reloaded, False otherwise.
+    """
+    ca_renewed, srv_renewed = enrollment.check_and_renew_hub_pki(
+        cert_dir=cert_dir,
+        hostnames=hostnames,
+        threshold_days=threshold_days
+    )
+    if ca_renewed or srv_renewed:
+        print(f"[!] Server Hub PKI certificates renewed (CA renewed: {ca_renewed}, Server cert renewed: {srv_renewed}).")
+        ca_cert, ca_key, ca_pem, ca_key_pem = enrollment.generate_ca_if_needed(cert_dir=cert_dir, force_renew=False)
+        SERVER_STATE["ca_cert"] = ca_cert
+        SERVER_STATE["ca_key"] = ca_key
+        SERVER_STATE["ca_cert_pem"] = ca_pem
+
+        ssl_ctx = SERVER_STATE.get("ssl_ctx")
+        if ssl_ctx is not None:
+            try:
+                reload_mtls_context(ssl_ctx, cert_dir=cert_dir)
+                print("[+] In-flight mTLS SSLContext successfully reloaded with updated certificates.")
+            except Exception as e:
+                print(f"[!] Failed to reload in-flight SSLContext: {e}")
+        return True
+    return False
+
+
+async def cert_validity_watchdog(interval_seconds: int = 43200, threshold_days: int = 30):
+    """
+    Periodically checks the validity of Hub Root CA and Server TLS certificates (default every 12 hours).
+    Triggers in-flight renewal and dynamic context reloading if expiration is within threshold_days.
+    """
+    config = SERVER_STATE.get("config", {})
+    cert_dir = config.get("cert_dir", "certs")
+    hostnames = [config.get("tcp_host", "0.0.0.0"), "127.0.0.1", "localhost"]
+    while True:
+        try:
+            await asyncio.sleep(interval_seconds)
+            check_and_rotate_server_certs(cert_dir=cert_dir, hostnames=hostnames, threshold_days=threshold_days)
+        except asyncio.CancelledError:
+            break
+        except Exception as e:
+            print(f"[!] Exception in cert_validity_watchdog: {e}")
+
+
+
 async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
     """
     mTLS TCP socket handler.
@@ -631,10 +697,20 @@ async def run_server():
     print(f"[*] Hermes Reporting API available at http://{hermes_host}:{hermes_port}/api/hermes/report")
     print(f"[*] Client Enrollment API available at http://{hermes_host}:{hermes_port}/api/client/enroll")
 
-    await asyncio.gather(
-        tcp_server.serve_forever(),
-        uv_server.serve()
-    )
+    watchdog_task = asyncio.create_task(cert_validity_watchdog())
+
+    try:
+        await asyncio.gather(
+            tcp_server.serve_forever(),
+            uv_server.serve(),
+            watchdog_task
+        )
+    finally:
+        watchdog_task.cancel()
+        try:
+            await watchdog_task
+        except asyncio.CancelledError:
+            pass
 
 
 def main():

src/Win_Client.py

@@ -24,17 +24,46 @@ CONFIG_FILE_NAME = "client_config.json"
 STATE_FILE_NAME = "client_state.json"
 
 
-def enroll_client_if_needed(hub_url: str, enrollment_secret: str, cert_dir: str, client_id: str, hostname: str, os_type: str = "windows"):
-    """Bootstraps client enrollment if certificates are missing."""
+def is_cert_expiring_soon(cert_path: str, threshold_days: int = 30) -> bool:
+    """Checks if client certificate at cert_path is expiring within threshold_days."""
+    if not os.path.exists(cert_path):
+        return True
+    try:
+        from cryptography import x509
+        with open(cert_path, "r", encoding="utf-8") as f:
+            cert = x509.load_pem_x509_certificate(f.read().encode("utf-8"))
+        expiry = getattr(cert, "not_valid_after_utc", None)
+        if expiry is None:
+            expiry = cert.not_valid_after.replace(tzinfo=timezone.utc)
+        now = datetime.now(timezone.utc)
+        return expiry <= (now + timedelta(days=threshold_days))
+    except Exception:
+        return True
+
+
+def enroll_client_if_needed(
+    hub_url: str,
+    enrollment_secret: str,
+    cert_dir: str,
+    client_id: str,
+    hostname: str,
+    os_type: str = "windows",
+    force_renew: bool = False,
+    threshold_days: int = 30
+):
+    """Bootstraps client enrollment if certificates are missing or expiring soon."""
     os.makedirs(cert_dir, exist_ok=True)
     ca_path = os.path.join(cert_dir, "ca.crt")
     cert_path = os.path.join(cert_dir, "client.crt")
     key_path = os.path.join(cert_dir, "client.key")
 
-    if os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path):
-        return True
+    if not force_renew and os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path):
+        if not is_cert_expiring_soon(cert_path, threshold_days=threshold_days):
+            return True
+        print(f"[*] Client certificate at {cert_path} is expiring within {threshold_days} days. Auto-renewing...")
 
-    print(f"[*] Bootstrapping client enrollment with LOGAR Hub at {hub_url}...")
+    action_name = "re-enrolling" if os.path.exists(cert_path) else "enrolling"
+    print(f"[*] Bootstrapping client {action_name} with LOGAR Hub at {hub_url}...")
     enroll_endpoint = f"{hub_url.rstrip('/')}/api/client/enroll"
     payload = {
         "client_id": client_id,
@@ -64,7 +93,7 @@ def enroll_client_if_needed(hub_url: str, enrollment_secret: str, cert_dir: str,
     except Exception:
         pass
 
-    print(f"[+] Client enrolled successfully! Certificates saved to {os.path.abspath(cert_dir)}")
+    print(f"[+] Client certificates updated successfully in {os.path.abspath(cert_dir)}")
     return True
 
 
@@ -269,6 +298,7 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
     machine_id = get_machine_identifier()
 
     # Attempt automatic enrollment bootstrap if certs are missing and secret is provided
+    hub_url = None
     if enrollment_secret:
         hermes_host = config.get("hermes_host", server_host)
         hermes_port = config.get("hermes_port", 8443)
@@ -285,7 +315,22 @@ def send_encrypted_logs_over_socket(config: dict, logs: list):
 
     if has_mtls_certs:
         print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over mTLS (TLS 1.3)...")
-        with get_tls_socket(server_host, server_port, cert_dir) as sock:
+        sock = None
+        try:
+            sock = get_tls_socket(server_host, server_port, cert_dir)
+        except (ssl.SSLError, ssl.CertificateError, ConnectionResetError) as tls_err:
+            if enrollment_secret and hub_url:
+                print(f"[!] TLS handshake error ({tls_err}). Re-enrolling with LOGAR Hub...")
+                try:
+                    enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="windows", force_renew=True)
+                    sock = get_tls_socket(server_host, server_port, cert_dir)
+                except Exception as retry_err:
+                    print(f"[!] Re-enrollment or reconnection retry failed: {retry_err}")
+                    raise
+            else:
+                raise
+
+        with sock:
             payload = {
                 "server": machine_id,
                 "timestamp": datetime.now(timezone.utc).isoformat(),

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
+

tests/test_linux_client.py

@@ -223,6 +223,27 @@ class TestLinuxClientComponent(unittest.TestCase):
         finally:
             shutil.rmtree(test_dir, ignore_errors=True)
 
+    def test_client_certificate_validity_and_proactive_check(self):
+        import shutil
+        test_dir = "test_linux_client_validity"
+        os.makedirs(test_dir, exist_ok=True)
+        try:
+            from src import server_enrollment as se
+            ca_cert, ca_key, _, _ = se.generate_ca_if_needed(test_dir)
+            client_cert_pem, client_key_pem = se.issue_client_cert("linux-validity-test", ca_cert, ca_key, days_valid=365)
+            cert_path = os.path.join(test_dir, "client.crt")
+            with open(cert_path, "w", encoding="utf-8") as f:
+                f.write(client_cert_pem)
+
+            # Newly issued cert (365 days) is not expiring soon at 30 days
+            self.assertFalse(Linux_Client.is_cert_expiring_soon(cert_path, threshold_days=30))
+            # Large threshold (500 days) reports expiring soon
+            self.assertTrue(Linux_Client.is_cert_expiring_soon(cert_path, threshold_days=500))
+            # Non-existent file reports expiring / missing
+            self.assertTrue(Linux_Client.is_cert_expiring_soon(os.path.join(test_dir, "missing.crt")))
+        finally:
+            shutil.rmtree(test_dir, ignore_errors=True)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_server.py

@@ -291,6 +291,43 @@ class TestServerComponent(unittest.TestCase):
             if os.path.exists(test_cert_dir):
                 shutil.rmtree(test_cert_dir, ignore_errors=True)
 
+    def test_cert_validity_and_hub_pki_renewal(self):
+        test_cert_dir = "test_certs_renew"
+        try:
+            ca_cert, ca_key, ca_pem, _ = Server.enrollment.generate_ca_if_needed(cert_dir=test_cert_dir)
+            srv_cert, srv_key, srv_pem, _ = Server.enrollment.generate_server_cert_if_needed(
+                ca_cert, ca_key, hostnames=["127.0.0.1"], cert_dir=test_cert_dir
+            )
+
+            # 1. Freshly generated certificates should NOT be expiring soon with standard 30-day threshold
+            self.assertFalse(Server.enrollment.is_cert_expiring_soon(ca_pem, threshold_days=30))
+            self.assertFalse(Server.enrollment.is_cert_expiring_soon(srv_pem, threshold_days=30))
+
+            # 2. Huge threshold (e.g. 5000 days) should flag expiration
+            self.assertTrue(Server.enrollment.is_cert_expiring_soon(srv_pem, threshold_days=5000))
+
+            # 3. check_and_renew_hub_pki with standard threshold should report no renewal needed
+            ca_renewed, srv_renewed = Server.enrollment.check_and_renew_hub_pki(cert_dir=test_cert_dir, threshold_days=30)
+            self.assertFalse(ca_renewed)
+            self.assertFalse(srv_renewed)
+
+            # 4. In-flight reload of SSLContext
+            ssl_ctx = Server.init_mtls_server_context(cert_dir=test_cert_dir)
+            Server.SERVER_STATE["ssl_ctx"] = ssl_ctx
+            Server.SERVER_STATE["config"] = {"db_path": self.test_db, "cert_dir": test_cert_dir, "tcp_host": "127.0.0.1"}
+
+            # Trigger rotation using high threshold
+            rotated = Server.check_and_rotate_server_certs(cert_dir=test_cert_dir, hostnames=["127.0.0.1"], threshold_days=5000)
+            self.assertTrue(rotated)
+
+            # Check that backup files were generated
+            bak_files = [f for f in os.listdir(test_cert_dir) if f.endswith(".bak")]
+            self.assertGreater(len(bak_files), 0)
+        finally:
+            import shutil
+            if os.path.exists(test_cert_dir):
+                shutil.rmtree(test_cert_dir, ignore_errors=True)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_win_client.py

@@ -209,6 +209,27 @@ class TestWinClientComponent(unittest.TestCase):
         finally:
             shutil.rmtree(test_dir, ignore_errors=True)
 
+    def test_client_certificate_validity_and_proactive_check(self):
+        import shutil
+        test_dir = "test_win_client_validity"
+        os.makedirs(test_dir, exist_ok=True)
+        try:
+            from src import server_enrollment as se
+            ca_cert, ca_key, _, _ = se.generate_ca_if_needed(test_dir)
+            client_cert_pem, client_key_pem = se.issue_client_cert("win-validity-test", ca_cert, ca_key, days_valid=365)
+            cert_path = os.path.join(test_dir, "client.crt")
+            with open(cert_path, "w", encoding="utf-8") as f:
+                f.write(client_cert_pem)
+
+            # Newly issued cert (365 days) is not expiring soon at 30 days
+            self.assertFalse(Win_Client.is_cert_expiring_soon(cert_path, threshold_days=30))
+            # Large threshold (500 days) reports expiring soon
+            self.assertTrue(Win_Client.is_cert_expiring_soon(cert_path, threshold_days=500))
+            # Non-existent file reports expiring / missing
+            self.assertTrue(Win_Client.is_cert_expiring_soon(os.path.join(test_dir, "missing.crt")))
+        finally:
+            shutil.rmtree(test_dir, ignore_errors=True)
+
 
 if __name__ == "__main__":
     unittest.main()