Refactor Server.py into modular domain classes (models, auth, storage, queue, licensing, PKI, UI)

src/pki_manager.py

@@ -0,0 +1,141 @@
+"""
+LOGAR - Dynamic PKI & mTLS Context Manager
+Provides modular classes for TLS 1.3 socket contexts, in-flight certificate rotation,
+and automated Sub-CA renewal.
+"""
+
+import os
+import ssl
+import sqlite3
+from typing import Optional, List, Dict, Any
+
+try:
+    from src import server_enrollment as enrollment
+except ImportError:
+    import server_enrollment as enrollment
+
+
+class PKIManager:
+    """Manages server TLS certificates, mTLS SSL contexts, and hot certificate reloading."""
+
+    @staticmethod
+    def init_mtls_server_context(cert_dir: str = "certs") -> ssl.SSLContext:
+        """Initializes TLS 1.3 server SSLContext with client certificate requirement (mTLS)."""
+        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")
+
+        ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        ctx.load_cert_chain(certfile=srv_cert, keyfile=srv_key)
+        ctx.load_verify_locations(cafile=ca_file)
+        upstream_file = os.path.join(cert_dir, "upstream_ca.crt")
+        if os.path.exists(upstream_file):
+            try:
+                ctx.load_verify_locations(cafile=upstream_file)
+            except Exception:
+                pass
+        ctx.verify_mode = ssl.CERT_REQUIRED
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+        return ctx
+
+    @staticmethod
+    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)
+        upstream_file = os.path.join(cert_dir, "upstream_ca.crt")
+        if os.path.exists(upstream_file):
+            try:
+                ssl_ctx.load_verify_locations(cafile=upstream_file)
+            except Exception:
+                pass
+
+    @staticmethod
+    def check_and_rotate_server_certs(
+        cert_dir: str = "certs",
+        hostnames: Optional[List[str]] = None,
+        threshold_days: int = 30,
+        server_state: Optional[Dict[str, Any]] = None
+    ) -> bool:
+        """
+        Checks if Root/Subordinate CA or server TLS certificate are expiring within threshold_days.
+        If a licensed Subordinate CA is expiring, automatically connects over mTLS to logar.eibl.tech
+        to request an in-flight renewal, then reloads the active SSLContext without socket interruption.
+        Returns True if renewed/reloaded, False otherwise.
+        """
+        state = server_state or {}
+        config = state.get("config", {})
+        license_server = config.get("license_server_url", "https://logar.eibl.tech")
+        ca_cert_path = os.path.join(cert_dir, "ca.crt")
+
+        # 1. Check if CA certificate is expiring and attempt upstream mTLS renewal
+        if os.path.exists(ca_cert_path):
+            try:
+                with open(ca_cert_path, "r", encoding="utf-8") as f:
+                    ca_pem = f.read()
+                if enrollment.is_cert_expiring_soon(ca_pem, threshold_days=threshold_days):
+                    info = enrollment.extract_license_info_from_cert(ca_pem)
+                    if not info.get("is_self_signed", True) or info.get("license_id"):
+                        print("[*] Subordinate CA expiring soon. Initiating automated mTLS renewal with logar.eibl.tech...")
+                        success, msg, new_pem = enrollment.renew_ca_cert_mtls(
+                            cert_dir=cert_dir,
+                            license_server_url=license_server
+                        )
+                        if success and new_pem:
+                            print(f"[+] Automated mTLS renewal successful: {msg}")
+                            new_info = enrollment.extract_license_info_from_cert(new_pem)
+                            if new_info.get("seats"):
+                                config["max_seats"] = new_info["seats"]
+                                db_path = config.get("db_path", "logar_state.db")
+                                conn = sqlite3.connect(db_path)
+                                conn.execute("UPDATE license_config SET max_seats = ? WHERE id = 1", (new_info["seats"],))
+                                conn.commit()
+                                conn.close()
+
+                            ca_cert, ca_key, ca_pem, ca_key_pem = enrollment.generate_ca_if_needed(cert_dir=cert_dir, force_renew=False)
+                            state["ca_cert"] = ca_cert
+                            state["ca_key"] = ca_key
+                            state["ca_cert_pem"] = ca_pem
+                            enrollment.generate_server_cert_if_needed(ca_cert, ca_key, hostnames=hostnames, cert_dir=cert_dir, force_renew=True)
+                            ssl_ctx = state.get("ssl_ctx")
+                            if ssl_ctx is not None:
+                                try:
+                                    PKIManager.reload_mtls_context(ssl_ctx, cert_dir=cert_dir)
+                                    print("[+] In-flight mTLS SSLContext successfully reloaded with renewed CA.")
+                                except Exception as e:
+                                    print(f"[!] Failed to reload in-flight SSLContext: {e}")
+                            return True
+                        else:
+                            print(f"[!] Automated mTLS renewal attempt failed: {msg}")
+            except Exception as e:
+                print(f"[!] Error evaluating CA certificate for mTLS renewal: {e}")
+
+        # 2. Local self-healing / regeneration check
+        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)
+            state["ca_cert"] = ca_cert
+            state["ca_key"] = ca_key
+            state["ca_cert_pem"] = ca_pem
+
+            ssl_ctx = state.get("ssl_ctx")
+            if ssl_ctx is not None:
+                try:
+                    PKIManager.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