LOGAR/src/Server.py

import os
import sys
import json
import uuid
import struct
import socket
import sqlite3
import argparse
import asyncio
import secrets
import warnings
from datetime import datetime, timezone, timedelta
from typing import Dict, Any, List, Optional

# Suppress cryptography / pgpy deprecation notices for a clean terminal output
warnings.filterwarnings("ignore")

import pgpy
from pgpy.constants import (
    PubKeyAlgorithm,
    KeyFlags,
    HashAlgorithm,
    SymmetricKeyAlgorithm,
    CompressionAlgorithm
)
import ssl
from pydantic import BaseModel
from fastapi import FastAPI, HTTPException
import uvicorn

try:
    from src import server_enrollment as enrollment
except ImportError:
    import server_enrollment as enrollment

CONFIG_FILE_NAME = "server_config.json"
DEFAULT_DB_FILE = "logar_state.db"
EVALUATION_WINDOW_HOURS = 12
RUN_THRESHOLD = 4

app = FastAPI(title="LOGAR Cloud Ingestion & Hermes Hub", version="2.0.0")

# Global context holding server state
SERVER_STATE: Dict[str, Any] = {}


class ClientEnrollRequest(BaseModel):
    client_id: str
    hostname: str
    os: str
    enrollment_secret: str


def generate_server_keypair(server_name: str):
    """Generates an OpenPGP RSA 2048 key with encryption capability."""
    key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048)
    uid = pgpy.PGPUID.new(server_name)
    key.add_uid(
        uid,
        usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage},
        hashes=[HashAlgorithm.SHA256],
        ciphers=[SymmetricKeyAlgorithm.AES256],
        compression=[CompressionAlgorithm.Uncompressed]
    )
    private_key_armored = str(key)
    public_key_armored = str(key.pubkey)
    fingerprint = str(key.pubkey.fingerprint)
    return private_key_armored, public_key_armored, fingerprint


def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
    """Loads existing server_config.json or creates a new one on first run."""
    if os.path.exists(config_path):
        print(f"[*] Loading server configuration from: {os.path.abspath(config_path)}")
        with open(config_path, "r", encoding="utf-8") as f:
            config = json.load(f)
        if "enrollment_secret" not in config:
            config["enrollment_secret"] = secrets.token_hex(24)
        if "max_seats" not in config:
            config["max_seats"] = 10
        if "cert_dir" not in config:
            config["cert_dir"] = "certs"
        if "tls_enabled" not in config:
            config["tls_enabled"] = True
        return config

    print(f"[!] Config '{config_path}' not found. Initializing first-run configuration...")
    server_name = "LOGAR-Cloud-Hub"
    private_key, public_key, fingerprint = generate_server_keypair(server_name)
    auth_token = secrets.token_hex(24)
    enrollment_secret = secrets.token_hex(24)

    config = {
        "server_name": server_name,
        "tcp_host": "0.0.0.0",
        "tcp_port": 9443,
        "hermes_host": "0.0.0.0",
        "hermes_port": 8443,
        "auth_token": auth_token,
        "enrollment_secret": enrollment_secret,
        "max_seats": 10,
        "cert_dir": "certs",
        "tls_enabled": True,
        "db_path": DEFAULT_DB_FILE,
        "evaluation_window_hours": EVALUATION_WINDOW_HOURS,
        "min_persistence_runs": RUN_THRESHOLD,
        "server_fingerprint": fingerprint,
        "public_key": public_key,
        "private_key": private_key
    }

    with open(config_path, "w", encoding="utf-8") as f:
        json.dump(config, f, indent=2)

    print(f"[+] Successfully generated new server config and OpenPGP keypair.")
    print(f"[+] Server Encryption Fingerprint: {fingerprint}")
    print(f"[+] Saved to: {os.path.abspath(config_path)}")
    return config


def create_client_config(
    server_host: str,
    server_port: int,
    output_path: str,
    config_path: str = CONFIG_FILE_NAME,
    hermes_host: Optional[str] = None,
    hermes_port: Optional[int] = None
) -> Dict[str, Any]:
    """Creates a client configuration file containing the server address, auth token, and encryption-only key/fingerprint."""
    server_conf = load_or_init_config(config_path)

    client_conf = {
        "server_host": server_host,
        "server_port": server_port,
        "hermes_host": hermes_host or server_conf.get("hermes_host", "127.0.0.1"),
        "hermes_port": hermes_port or server_conf.get("hermes_port", 8443),
        "enrollment_secret": server_conf.get("enrollment_secret"),
        "cert_dir": "certs",
        "server_fingerprint": server_conf["server_fingerprint"],
        "server_public_key": server_conf["public_key"],
        "auth_token": server_conf["auth_token"]
    }

    out_dir = os.path.dirname(os.path.abspath(output_path))
    if out_dir and not os.path.exists(out_dir):
        os.makedirs(out_dir, exist_ok=True)

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(client_conf, f, indent=2)

    print(f"[+] Client configuration successfully written to: {os.path.abspath(output_path)}")
    print(f"    - Server Target: {server_host}:{server_port}")
    print(f"    - Encryption Fingerprint: {server_conf['server_fingerprint']}")
    return client_conf


def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: int = 10):
    """Initializes the SQLite schema for multi-run temporal tracking, client tracking, and license quota."""
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS active_issues (
            fingerprint TEXT PRIMARY KEY,
            site_name TEXT,
            server TEXT,
            signature TEXT,
            severity TEXT,
            message TEXT,
            os_type TEXT,
            first_seen TEXT,
            last_seen TEXT,
            run_count INTEGER,
            status TEXT,
            last_run_id TEXT
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS ingest_runs (
            run_id TEXT PRIMARY KEY,
            site_name TEXT,
            server TEXT,
            timestamp TEXT,
            log_count INTEGER
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS license_config (
            id INTEGER PRIMARY KEY CHECK (id = 1),
            max_seats INTEGER NOT NULL DEFAULT 10,
            enrollment_secret TEXT NOT NULL
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS clients (
            client_id TEXT PRIMARY KEY,
            hostname TEXT NOT NULL,
            os_type TEXT NOT NULL,
            cert_fingerprint TEXT NOT NULL,
            status TEXT DEFAULT 'active',
            first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    if enrollment_secret:
        conn.execute("""
            INSERT OR IGNORE INTO license_config (id, max_seats, enrollment_secret)
            VALUES (1, ?, ?)
        """, (max_seats, enrollment_secret))
    conn.commit()
    conn.close()


def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: int, min_runs: int) -> Dict[str, Any]:
    """
    Evaluates candidate issues against the 12-hour evaluation window and 4-run rule.
    Zero-state clients send raw candidate entries; this engine handles temporal state.
    """
    client_server = payload.get("server", "unknown-host")
    site_name = payload.get("site_name") or (client_server.split(".", 1)[1] if "." in client_server else "default")
    logs = payload.get("logs", [])
    run_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc)
    now_iso = now.isoformat()

    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Record the batch run
    cursor.execute(
        "INSERT INTO ingest_runs (run_id, site_name, server, timestamp, log_count) VALUES (?, ?, ?, ?, ?)",
        (run_id, site_name, client_server, now_iso, len(logs))
    )

    processed_count = 0
    promoted_to_verified = 0

    for log in logs:
        severity = str(log.get("severity", "WARNING")).upper()
        # Edge forwarder filter safeguard: retain INFO to ERROR / CRITICAL; strip verbose debug noise
        if severity in ["DEBUG", "TRACE"]:
            continue

        # Errors are always passed immediately; the 4-run rule only concerns warnings
        is_error = severity in ["ERROR", "CRITICAL", "FATAL"]

        signature = log.get("signature", "unknown")
        server = log.get("server", client_server)
        message = log.get("message", "")
        os_type = log.get("os_type", "unknown")
        fp = f"{site_name}:{server}:{signature}"

        cursor.execute(
            "SELECT run_count, first_seen, last_seen, status, last_run_id FROM active_issues WHERE fingerprint = ?",
            (fp,)
        )
        row = cursor.fetchone()

        if row:
            run_count, first_seen_str, last_seen_str, current_status, last_run_id = row
            try:
                last_seen_dt = datetime.fromisoformat(last_seen_str)
            except Exception:
                last_seen_dt = now

            # 12-hour evaluation window expiry check
            if (now - last_seen_dt) > timedelta(hours=window_hours):
                # Window elapsed: reset to new cycle
                new_runs = 1
                new_first_seen = now_iso
                new_status = "VERIFIED" if is_error else "TRANSIENT"
            else:
                # Same run guard: only increment count once per distinct run batch
                if last_run_id != run_id:
                    new_runs = run_count + 1
                else:
                    new_runs = run_count
                new_first_seen = first_seen_str
                # 4-run rule applies to warnings; errors are always passed immediately as VERIFIED
                new_status = "VERIFIED" if (is_error or new_runs >= min_runs) else "TRANSIENT"

            if new_status == "VERIFIED" and current_status != "VERIFIED":
                promoted_to_verified += 1

            cursor.execute("""
                UPDATE active_issues 
                SET run_count = ?, last_seen = ?, first_seen = ?, status = ?, last_run_id = ?, message = ?, severity = ?
                WHERE fingerprint = ?
            """, (new_runs, now_iso, new_first_seen, new_status, run_id, message, severity, fp))
        else:
            initial_status = "VERIFIED" if (is_error or 1 >= min_runs) else "TRANSIENT"
            if initial_status == "VERIFIED":
                promoted_to_verified += 1
            cursor.execute("""
                INSERT INTO active_issues 
                (fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status, last_run_id)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (fp, site_name, server, signature, severity, message, os_type, now_iso, now_iso, 1, initial_status, run_id))

        processed_count += 1

    conn.commit()
    conn.close()

    return {
        "status": "success",
        "run_id": run_id,
        "processed": processed_count,
        "promoted_verified": promoted_to_verified
    }


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)
    ctx.verify_mode = ssl.CERT_REQUIRED
    ctx.minimum_version = ssl.TLSVersion.TLSv1_3
    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.
    Extracts client CN (client_id) from the TLS handshake,
    validates active license status in SQLite, updates last_seen,
    reads 4-byte big-endian length-prefixed JSON payload,
    and ingests candidate logs into the temporal evaluation engine.
    """
    client_id = None
    ssl_obj = writer.get_extra_info("ssl_object")
    if ssl_obj:
        peercert = ssl_obj.getpeercert()
        if peercert and "subject" in peercert:
            for rdn in peercert["subject"]:
                for key, val in rdn:
                    if key == "commonName":
                        client_id = val
                        break

    # If mTLS is enforced, verify client in accounting database
    if SERVER_STATE.get("tls_enabled", False):
        if not client_id:
            writer.close()
            await writer.wait_closed()
            return

        db_path = SERVER_STATE["config"]["db_path"]
        conn = sqlite3.connect(db_path)
        c = conn.cursor()
        c.execute("SELECT status FROM clients WHERE client_id = ?", (client_id,))
        row = c.fetchone()
        if not row or row[0] != "active":
            conn.close()
            writer.close()
            await writer.wait_closed()
            return
        c.execute("UPDATE clients SET last_seen = CURRENT_TIMESTAMP WHERE client_id = ?", (client_id,))
        conn.commit()
        conn.close()

    try:
        # Read 4-byte length prefix
        length_bytes = await reader.readexactly(4)
        length = struct.unpack(">I", length_bytes)[0]
        if length <= 0 or length > 10 * 1024 * 1024:  # 10MB limit
            raise ValueError(f"Invalid frame size: {length}")

        payload_bytes = await reader.readexactly(length)
        raw_payload = json.loads(payload_bytes.decode("utf-8"))

        # Support both direct JSON payload over mTLS and legacy OpenPGP envelope
        if "encrypted_payload" in raw_payload and SERVER_STATE.get("private_key_obj"):
            pgp_msg = pgpy.PGPMessage.from_blob(raw_payload["encrypted_payload"])
            priv_key = SERVER_STATE["private_key_obj"]
            decrypted_obj = priv_key.decrypt(pgp_msg)
            log_payload = json.loads(decrypted_obj.message)
        else:
            log_payload = raw_payload

        # Attach authenticated client_id if not present
        if client_id and "server" not in log_payload:
            log_payload["server"] = client_id

        # Ingest and apply 12h window / 4-run rule
        res = process_ingested_logs(
            log_payload,
            db_path=SERVER_STATE["config"]["db_path"],
            window_hours=SERVER_STATE["config"]["evaluation_window_hours"],
            min_runs=SERVER_STATE["config"]["min_persistence_runs"]
        )

        resp_bytes = json.dumps(res).encode("utf-8")
        writer.write(struct.pack(">I", len(resp_bytes)) + resp_bytes)
        await writer.drain()

    except Exception as e:
        err = json.dumps({"status": "error", "message": str(e)}).encode("utf-8")
        try:
            writer.write(struct.pack(">I", len(err)) + err)
            await writer.drain()
        except Exception:
            pass
    finally:
        writer.close()
        try:
            await writer.wait_closed()
        except Exception:
            pass


@app.post("/api/client/enroll")
def enroll_client(req: ClientEnrollRequest):
    """
    Enrolls an edge client by validating the enrollment secret,
    checking license seat limits, issuing a signed client certificate + key,
    and recording the client in the SQLite accounting database.
    """
    db_path = SERVER_STATE["config"]["db_path"]
    conn = sqlite3.connect(db_path)
    c = conn.cursor()

    # 1. Validate enrollment secret against license_config
    c.execute("SELECT enrollment_secret, max_seats FROM license_config WHERE id = 1")
    row = c.fetchone()
    if not row:
        conn.close()
        raise HTTPException(status_code=500, detail="License configuration not initialized")

    expected_secret, max_seats = row
    if not secrets.compare_digest(str(req.enrollment_secret), str(expected_secret)):
        conn.close()
        raise HTTPException(status_code=403, detail="Invalid enrollment secret")

    ca_cert = SERVER_STATE.get("ca_cert")
    ca_key = SERVER_STATE.get("ca_key")
    ca_cert_pem = SERVER_STATE.get("ca_cert_pem")

    if not ca_cert or not ca_key:
        conn.close()
        raise HTTPException(status_code=500, detail="Root CA not loaded on server")

    # 2. Check if client_id already registered
    c.execute("SELECT status FROM clients WHERE client_id = ?", (req.client_id,))
    client_row = c.fetchone()
    if client_row:
        if client_row[0] == "revoked":
            conn.close()
            raise HTTPException(status_code=403, detail="Client certificate has been revoked")

        # Re-issue for existing active client
        client_cert_pem, client_key_pem = enrollment.issue_client_cert(req.client_id, ca_cert, ca_key)
        fp = enrollment.calculate_cert_fingerprint(client_cert_pem)
        c.execute("""
            UPDATE clients 
            SET hostname = ?, os_type = ?, cert_fingerprint = ?, last_seen = CURRENT_TIMESTAMP 
            WHERE client_id = ?
        """, (req.hostname, req.os, fp, req.client_id))
        conn.commit()
        conn.close()
        print(f"[+] Re-enrolled active client: {req.client_id} ({req.hostname})")
        return {
            "ca_cert": ca_cert_pem,
            "client_cert": client_cert_pem,
            "client_key": client_key_pem
        }

    # 3. New client: check seat limits
    c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'")
    active_count = c.fetchone()[0]
    if active_count >= max_seats:
        conn.close()
        raise HTTPException(status_code=403, detail="License seat limit reached")

    # 4. Issue signed cert + key
    client_cert_pem, client_key_pem = enrollment.issue_client_cert(req.client_id, ca_cert, ca_key)
    fp = enrollment.calculate_cert_fingerprint(client_cert_pem)
    c.execute("""
        INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status)
        VALUES (?, ?, ?, ?, 'active')
    """, (req.client_id, req.hostname, req.os, fp))
    conn.commit()
    conn.close()
    print(f"[+] Successfully enrolled new client: {req.client_id} ({req.hostname}) [Seats: {active_count + 1}/{max_seats}]")

    return {
        "ca_cert": ca_cert_pem,
        "client_cert": client_cert_pem,
        "client_key": client_key_pem
    }


@app.get("/api/clients")
def list_clients():
    """Returns all registered clients and license seat usage."""
    db_path = SERVER_STATE["config"]["db_path"]
    conn = sqlite3.connect(db_path)
    c = conn.cursor()
    c.execute("SELECT max_seats FROM license_config WHERE id = 1")
    lic_row = c.fetchone()
    max_seats = lic_row[0] if lic_row else 10

    c.execute("SELECT client_id, hostname, os_type, cert_fingerprint, status, first_seen, last_seen FROM clients")
    rows = c.fetchall()
    conn.close()

    clients = [
        {
            "client_id": r[0],
            "hostname": r[1],
            "os_type": r[2],
            "cert_fingerprint": r[3],
            "status": r[4],
            "first_seen": r[5],
            "last_seen": r[6]
        }
        for r in rows
    ]
    active_count = sum(1 for cl in clients if cl["status"] == "active")
    return {
        "active_seats": active_count,
        "max_seats": max_seats,
        "clients": clients
    }


@app.get("/api/hermes/report")
def get_verified_anomalies_for_hermes():
    """
    Ingestion endpoint for Hermes agentic workflows.
    Returns only verified anomalies that have satisfied the 4-run persistence rule
    within the active 12-hour evaluation window. Transient blips (< 4 runs) are excluded.
    """
    db_path = SERVER_STATE["config"]["db_path"]
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    cursor.execute("""
        SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status
        FROM active_issues
        WHERE status = 'VERIFIED'
        ORDER BY last_seen DESC
    """)
    rows = cursor.fetchall()
    conn.close()

    report = []
    for r in rows:
        report.append({
            "fingerprint": r[0],
            "site": r[1],
            "server": r[2],
            "signature": r[3],
            "severity": r[4],
            "message": r[5],
            "os_type": r[6],
            "first_seen": r[7],
            "last_seen": r[8],
            "consecutive_runs": r[9],
            "evaluation_window": f"{SERVER_STATE['config']['evaluation_window_hours']}h",
            "verified": True,
            "status": r[10]
        })

    return report


@app.get("/api/hermes/all")
def get_all_issues():
    """Diagnostic endpoint to inspect both transient candidate blips and verified anomalies."""
    db_path = SERVER_STATE["config"]["db_path"]
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("""
        SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status
        FROM active_issues
    """)
    rows = cursor.fetchall()
    conn.close()

    return [
        {
            "fingerprint": r[0],
            "site": r[1],
            "server": r[2],
            "signature": r[3],
            "severity": r[4],
            "message": r[5],
            "os_type": r[6],
            "first_seen": r[7],
            "last_seen": r[8],
            "run_count": r[9],
            "status": r[10]
        }
        for r in rows
    ]


@app.get("/health")
def health_check():
    return {
        "status": "healthy",
        "server_name": SERVER_STATE["config"]["server_name"],
        "fingerprint": SERVER_STATE["config"]["server_fingerprint"],
        "tcp_port": SERVER_STATE["config"]["tcp_port"],
        "hermes_port": SERVER_STATE["config"]["hermes_port"],
        "tls_enabled": SERVER_STATE.get("tls_enabled", False)
    }


async def run_server():
    """Runs the mTLS TCP socket listener and the Hermes REST API concurrently."""
    config = SERVER_STATE["config"]
    tcp_host = config["tcp_host"]
    tcp_port = int(config["tcp_port"])
    hermes_host = config["hermes_host"]
    hermes_port = int(config["hermes_port"])
    ssl_ctx = SERVER_STATE.get("ssl_ctx")

    # Start mTLS / TCP Socket Server
    tcp_server = await asyncio.start_server(handle_socket_client, tcp_host, tcp_port, ssl=ssl_ctx)
    mode_str = "mTLS TLSv1.3" if ssl_ctx else "Plain TCP"
    print(f"[*] LOGAR {mode_str} Socket Server listening on {tcp_host}:{tcp_port}")

    # Start FastAPI / Uvicorn server for Hermes & Enrollment
    uv_config = uvicorn.Config(app, host=hermes_host, port=hermes_port, log_level="warning")
    uv_server = uvicorn.Server(uv_config)
    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")

    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():
    parser = argparse.ArgumentParser(description="LOGAR Cloud Hub & TCP Socket Ingestion Server")
    parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to server_config.json")
    parser.add_argument("--create-client-config", action="store_true", help="Generate a client config with encryption-only fingerprint and server address")
    parser.add_argument("--client-out", default="client_config.json", help="Output file path for generated client config")
    parser.add_argument("--server-host", default="127.0.0.1", help="Server address to embed in client config")
    parser.add_argument("--server-port", type=int, default=None, help="TCP port to embed in client config")
    args = parser.parse_args()

    config = load_or_init_config(args.config)
    init_db(
        config["db_path"],
        enrollment_secret=config.get("enrollment_secret"),
        max_seats=config.get("max_seats", 10)
    )

    # Initialize dynamic PKI (Root CA and Server TLS Certificate)
    cert_dir = config.get("cert_dir", "certs")
    ca_cert, ca_key, ca_pem, ca_key_pem = enrollment.generate_ca_if_needed(cert_dir=cert_dir)
    srv_cert, srv_key, srv_pem, srv_key_pem = enrollment.generate_server_cert_if_needed(
        ca_cert, ca_key,
        hostnames=[config.get("tcp_host"), "127.0.0.1", "localhost"],
        cert_dir=cert_dir
    )

    # Initialize mTLS SSLContext if enabled
    ssl_ctx = None
    if config.get("tls_enabled", True):
        ssl_ctx = init_mtls_server_context(cert_dir=cert_dir)

    # Load OpenPGP private key into memory (legacy fallback)
    priv_key_obj, _ = pgpy.PGPKey.from_blob(config["private_key"])
    SERVER_STATE["config"] = config
    SERVER_STATE["private_key_obj"] = priv_key_obj
    SERVER_STATE["ca_cert"] = ca_cert
    SERVER_STATE["ca_key"] = ca_key
    SERVER_STATE["ca_cert_pem"] = ca_pem
    SERVER_STATE["ssl_ctx"] = ssl_ctx
    SERVER_STATE["tls_enabled"] = config.get("tls_enabled", True)

    if args.create_client_config:
        port = args.server_port or config["tcp_port"]
        create_client_config(
            server_host=args.server_host,
            server_port=port,
            output_path=args.client_out,
            config_path=args.config
        )
        sys.exit(0)

    print("=" * 60)
    print(f"  LOGAR Server Hub: {config['server_name']}")
    print(f"  Transport Security: {'mTLS (TLS 1.3)' if ssl_ctx else 'Plain TCP'}")
    print(f"  License Quota: {config.get('max_seats', 10)} Active Seats")
    print(f"  Server Encryption Fingerprint: {config['server_fingerprint']}")
    print(f"  Evaluation Window: {config['evaluation_window_hours']} hours | 4-Run Rule: Warnings | Immediate Pass: Errors")
    print("=" * 60)

    try:
        asyncio.run(run_server())
    except KeyboardInterrupt:
        print("\n[!] Server shutting down.")


if __name__ == "__main__":
    main()