diff --git a/src/Server.py b/src/Server.py index 977ee35..2effcf2 100644 --- a/src/Server.py +++ b/src/Server.py @@ -1,781 +1,1743 @@ -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.1") - -# 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() \ No newline at end of file +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, Request +from fastapi.responses import HTMLResponse, JSONResponse, Response +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.1") + +# Global context holding server state +SERVER_STATE: Dict[str, Any] = {} + + +class ClientEnrollRequest(BaseModel): + client_id: str + hostname: str + os: str + enrollment_secret: str + + +class ClientActionRequest(BaseModel): + client_id: str + + +class LicenseImportRequest(BaseModel): + ca_cert: str + upstream_ca: Optional[str] = None + + + +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 + if "license_server_url" not in config: + config["license_server_url"] = "https://logar.eibl.tech" + 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, + "license_server_url": "https://logar.eibl.tech", + "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, + license_id TEXT, + license_server_url TEXT, + issued_at TEXT, + expires_at TEXT, + license_status TEXT DEFAULT 'active' + ) + """) + # Migration helper for existing databases + for col, col_def in [ + ("license_id", "TEXT"), + ("license_server_url", "TEXT"), + ("issued_at", "TEXT"), + ("expires_at", "TEXT"), + ("license_status", "TEXT DEFAULT 'active'") + ]: + try: + conn.execute(f"ALTER TABLE license_config ADD COLUMN {col} {col_def}") + except sqlite3.OperationalError: + pass + + 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 export_license_request_lic(config: Dict[str, Any]) -> Tuple[Dict[str, Any], str]: + """Generates CA CSR and packages into exportable .lic dictionary and JSON string.""" + cert_dir = config.get("cert_dir", "certs") + lic_server = config.get("license_server_url", "https://logar.eibl.tech") + hostname = socket.gethostname() + seats = config.get("max_seats", 10) + fp = config.get("server_fingerprint", "") + return enrollment.create_license_request_lic( + cert_dir=cert_dir, + license_server=lic_server, + server_fingerprint=fp, + hostname=hostname, + requested_seats=seats + ) + + + +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/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. + """ + config = SERVER_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", DEFAULT_DB_FILE) + 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) + SERVER_STATE["ca_cert"] = ca_cert + SERVER_STATE["ca_key"] = ca_key + SERVER_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 = 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 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) + 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 + ] + + +def get_license_summary() -> Dict[str, Any]: + """Inspects certs/ca.crt, database quotas, and returns comprehensive licensing status.""" + config = SERVER_STATE.get("config", {}) + cert_dir = config.get("cert_dir", "certs") + ca_path = os.path.join(cert_dir, "ca.crt") + db_path = config.get("db_path", DEFAULT_DB_FILE) + + max_seats = config.get("max_seats", 10) + try: + conn = sqlite3.connect(db_path) + c = conn.cursor() + c.execute("SELECT max_seats FROM license_config WHERE id = 1") + row = c.fetchone() + if row and row[0]: + max_seats = row[0] + c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'") + active_seats = c.fetchone()[0] + c.execute("SELECT COUNT(*) FROM clients WHERE status = 'revoked'") + revoked_seats = c.fetchone()[0] + conn.close() + except Exception: + active_seats = 0 + revoked_seats = 0 + + csr_path = os.path.join(cert_dir, "ca.csr") + csr_exists = os.path.exists(csr_path) + + if os.path.exists(ca_path): + try: + with open(ca_path, "r", encoding="utf-8") as f: + ca_pem = f.read() + info = enrollment.extract_license_info_from_cert(ca_pem) + if info.get("seats"): + max_seats = info["seats"] + except Exception: + info = {"valid": False} + else: + info = {"valid": False, "is_self_signed": False, "is_expired": False, "days_remaining": 0} + + license_id = info.get("license_id") or "UNLICENSED-LOCAL" + issuer_url = info.get("issuer_url") or config.get("license_server_url", "https://logar.eibl.tech") + status_label = "ACTIVE" + if info.get("is_expired"): + status_label = "EXPIRED" + elif not info.get("seats") and info.get("is_self_signed"): + status_label = "SELF-SIGNED" + + pct = int((active_seats / max_seats * 100)) if max_seats > 0 else 0 + return { + "status": status_label, + "active_seats": active_seats, + "revoked_seats": revoked_seats, + "max_seats": max_seats, + "available_seats": max(0, max_seats - active_seats), + "percentage": min(100, pct), + "license_id": license_id, + "issuer_url": issuer_url, + "issuer_cn": info.get("issuer_cn", "N/A"), + "subject_cn": info.get("subject_cn", "N/A"), + "is_self_signed": info.get("is_self_signed", False), + "not_valid_before": info.get("not_valid_before", "N/A"), + "not_valid_after": info.get("not_valid_after", "N/A"), + "days_remaining": info.get("days_remaining", 0), + "fingerprint": info.get("fingerprint", "N/A"), + "csr_exists": csr_exists, + "server_fingerprint": config.get("server_fingerprint", "N/A") + } + + +@app.get("/api/license/status") +def api_license_status(): + """Returns license details, seat quotas, and certificate validity.""" + return get_license_summary() + + +@app.get("/api/license/export-lic") +def export_lic_file(): + """Generates and downloads the logar_server.lic file containing CA CSR and authority coordinates.""" + config = SERVER_STATE.get("config", {}) + lic_dict, lic_json = export_license_request_lic(config) + return Response( + content=lic_json, + media_type="application/json", + headers={"Content-Disposition": 'attachment; filename="logar_server.lic"'} + ) + + +@app.get("/api/license/csr") +def get_ca_csr(): + """Returns raw PEM Certificate Signing Request (CSR) for subordinate CA.""" + config = SERVER_STATE.get("config", {}) + cert_dir = config.get("cert_dir", "certs") + csr_path = os.path.join(cert_dir, "ca.csr") + if not os.path.exists(csr_path): + csr_pem, _ = enrollment.generate_ca_csr(cert_dir=cert_dir) + else: + with open(csr_path, "r", encoding="utf-8") as f: + csr_pem = f.read() + return {"csr": csr_pem} + + +@app.post("/api/license/import") +async def import_signed_license(req: Request): + """ + Imports and activates a signed CA certificate issued by logar.eibl.tech. + Extracts the seat quota from enterprise X.509 OIDs, stores in SQLite, + regenerates Server TLS certificate, and dynamically reloads in-flight mTLS context. + """ + try: + body = await req.json() + except Exception: + raise HTTPException(status_code=400, detail="Invalid JSON body") + + ca_cert_pem = body.get("ca_cert") + if not ca_cert_pem and "lic" in body and isinstance(body["lic"], dict): + ca_cert_pem = body["lic"].get("ca_cert") + if not ca_cert_pem: + raw_text = body.get("raw", "") + if "-----BEGIN CERTIFICATE-----" in raw_text: + ca_cert_pem = raw_text + + if not ca_cert_pem: + raise HTTPException(status_code=400, detail="Missing 'ca_cert' PEM certificate in request") + + info = enrollment.extract_license_info_from_cert(ca_cert_pem) + if not info.get("valid"): + raise HTTPException(status_code=400, detail=f"Invalid CA certificate: {info.get('error')}") + + config = SERVER_STATE.get("config", {}) + cert_dir = config.get("cert_dir", "certs") + ca_path = os.path.join(cert_dir, "ca.crt") + + # Backup existing CA if present + if os.path.exists(ca_path): + ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + try: + os.replace(ca_path, f"{ca_path}.{ts}.bak") + except Exception: + pass + + with open(ca_path, "w", encoding="utf-8") as f: + f.write(ca_cert_pem) + + new_seats = info.get("seats") or config.get("max_seats", 10) + config["max_seats"] = new_seats + db_path = config.get("db_path", DEFAULT_DB_FILE) + conn = sqlite3.connect(db_path) + conn.execute(""" + UPDATE license_config + SET max_seats = ?, license_id = ?, issued_at = ?, expires_at = ?, license_status = 'active' + WHERE id = 1 + """, (new_seats, info.get("license_id"), info.get("not_valid_before"), info.get("not_valid_after"))) + conn.commit() + conn.close() + + # Regenerate server TLS certificate signed by new CA + ca_cert, ca_key, _, _ = 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_cert_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, + force_renew=True + ) + + # In-flight SSLContext hot reload + ssl_ctx = SERVER_STATE.get("ssl_ctx") + if ssl_ctx: + reload_mtls_context(ssl_ctx, cert_dir=cert_dir) + print("[+] In-flight mTLS SSLContext successfully reloaded with newly imported CA.") + + return { + "status": "success", + "message": f"License successfully activated for {new_seats} seats.", + "license_id": info.get("license_id"), + "seats": new_seats, + "expires_at": info.get("not_valid_after"), + "days_remaining": info.get("days_remaining") + } + + +@app.post("/api/license/renew") +def trigger_license_renewal(): + """Explicitly triggers automated mTLS renewal with logar.eibl.tech.""" + config = SERVER_STATE.get("config", {}) + cert_dir = config.get("cert_dir", "certs") + lic_server = config.get("license_server_url", "https://logar.eibl.tech") + success, msg, new_pem = enrollment.renew_ca_cert_mtls( + cert_dir=cert_dir, + license_server_url=lic_server + ) + if not success or not new_pem: + raise HTTPException(status_code=400, detail=msg) + + info = enrollment.extract_license_info_from_cert(new_pem) + if info.get("seats"): + config["max_seats"] = info["seats"] + db_path = config.get("db_path", DEFAULT_DB_FILE) + conn = sqlite3.connect(db_path) + conn.execute("UPDATE license_config SET max_seats = ? WHERE id = 1", (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) + SERVER_STATE["ca_cert"] = ca_cert + SERVER_STATE["ca_key"] = ca_key + SERVER_STATE["ca_cert_pem"] = ca_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, + force_renew=True + ) + ssl_ctx = SERVER_STATE.get("ssl_ctx") + if ssl_ctx: + reload_mtls_context(ssl_ctx, cert_dir=cert_dir) + print("[+] In-flight mTLS SSLContext successfully reloaded after manual renewal.") + + return { + "status": "success", + "message": msg, + "seats": info.get("seats"), + "expires_at": info.get("not_valid_after"), + "days_remaining": info.get("days_remaining") + } + + +@app.post("/api/clients/revoke") +def revoke_client(action: ClientActionRequest): + """Revokes a client certificate/license seat. Revoked clients are immediately rejected during mTLS handshakes.""" + db_path = SERVER_STATE["config"]["db_path"] + conn = sqlite3.connect(db_path) + c = conn.cursor() + c.execute("UPDATE clients SET status = 'revoked' WHERE client_id = ?", (action.client_id,)) + affected = c.rowcount + conn.commit() + conn.close() + if affected == 0: + raise HTTPException(status_code=404, detail="Client ID not found") + print(f"[!] Revoked client license seat: {action.client_id}") + return {"status": "revoked", "client_id": action.client_id} + + +@app.post("/api/clients/reactivate") +def reactivate_client(action: ClientActionRequest): + """Reactivates a revoked client certificate if within the active seat quota.""" + 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") + row = c.fetchone() + max_seats = row[0] if row else 10 + + 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=f"Cannot reactivate client: maximum seat quota ({max_seats}) reached.") + + c.execute("UPDATE clients SET status = 'active' WHERE client_id = ?", (action.client_id,)) + affected = c.rowcount + conn.commit() + conn.close() + if affected == 0: + raise HTTPException(status_code=404, detail="Client ID not found") + print(f"[+] Reactivated client license seat: {action.client_id}") + return {"status": "active", "client_id": action.client_id} + + +@app.get("/", response_class=HTMLResponse) +@app.get("/webui", response_class=HTMLResponse) +def serve_webui(): + """Renders the comprehensive LOGAR Server Hub & Fleet Licensing WebUI.""" + summary = get_license_summary() + config = SERVER_STATE.get("config", {}) + + # Fetch clients + db_path = config.get("db_path", DEFAULT_DB_FILE) + conn = sqlite3.connect(db_path) + c = conn.cursor() + c.execute("SELECT client_id, hostname, os_type, cert_fingerprint, status, first_seen, last_seen FROM clients ORDER BY last_seen DESC") + clients = c.fetchall() + + # Fetch verified anomalies + c.execute("SELECT fingerprint, server, severity, message, run_count, last_seen FROM active_issues WHERE status = 'VERIFIED' ORDER BY last_seen DESC LIMIT 50") + anomalies = c.fetchall() + conn.close() + + # Build client rows + client_rows = [] + for cl in clients: + cid, host, ost, fp, st, first, last = cl + badge_cls = "badge-active" if st == "active" else "badge-revoked" + btn_action = ( + f'' + if st == "active" else + f'' + ) + client_rows.append(f""" + + {st.upper()} + {host} + {cid} + {ost.upper()} + {fp[:16]}... + {last} + {btn_action} + + """) + clients_html = "".join(client_rows) if client_rows else 'No edge clients registered yet.' + + # Build anomaly rows + anomaly_rows = [] + for an in anomalies: + afp, srv, sev, msg, rc, ls = an + sev_cls = "badge-error" if sev in ("ERROR", "CRITICAL", "FATAL") else "badge-warning" + short_msg = msg[:90] + "..." if len(msg) > 90 else msg + anomaly_rows.append(f""" + + {sev} + {srv} + {rc} runs + {short_msg} + {ls} + + """) + anomalies_html = "".join(anomaly_rows) if anomaly_rows else 'No verified anomalies currently reported.' + + # Progress bar color logic + pct = summary["percentage"] + bar_color = "#10b981" if pct < 70 else ("#f59e0b" if pct < 90 else "#ef4444") + badge_status_cls = "badge-active" if summary["status"] == "ACTIVE" else ("badge-warning" if summary["status"] == "SELF-SIGNED" else "badge-revoked") + + html_template = """ + + + + + LOGAR Cloud Hub & Fleet Manager + + + +
+
+
+ 🛰️ +
+

LOGAR Fleet Hub & License Management

+
Cloud Hub: __SERVER_NAME__ | Evaluator: 12h Rolling Window & 4-Run Rule
+
+
+
+ mTLS Socket: Port __TCP_PORT__ + Hermes API: Port __HERMES_PORT__ + OPERATIONAL +
+
+ +
+ +
+
+
+ Seat Quota Usage + __STATUS__ +
+
__ACTIVE_SEATS__ / __MAX_SEATS__ Seats
+
__AVAILABLE_SEATS__ seats available for auto-enrollment
+
+
+
+
+ +
+
+ License & CA Identity + X.509 SUB-CA +
+
__LICENSE_ID__
+
Expires: __EXPIRES_AT__ (__DAYS_REMAINING__ days remaining)
+
Fingerprint: __CA_FP__
+
+ +
+
+ Upstream Authority + mTLS 1.3 AUTO-RENEW +
+
__ISSUER_URL__
+
Watchdog: In-flight auto-renewal triggers < 30d
+
Sub-CA Subject: __SUBJECT_CN__
+
+ +
+
+ Fleet Deployment + __TOTAL_NODES__ NODES +
+
__ACTIVE_SEATS__ Active Forwarders
+
__REVOKED_SEATS__ revoked forwarders blocked at socket
+
+
+ +
+ Licensing Controls: + + 📥 Export .lic Request + + + + +
+ +
+
+ Enrolled Edge Forwarders & Allocated Seats + __TOTAL_NODES__ Registered +
+ + + + + + + + + + + + + + __CLIENTS_HTML__ + +
StatusHost NodeClient IdentifierOS TypeCert FingerprintLast SeenSeat Actions
+
+ +
+
+ Hermes Verified Anomalies (4-Run Rule & Errors) + Active Window: 12h +
+ + + + + + + + + + + + __ANOMALIES_HTML__ + +
SeveritySource NodePersistenceVerified Anomaly MessageLast Seen
+
+
+ + + + + + + + + +""" + + total_nodes = summary["active_seats"] + summary["revoked_seats"] + rendered = ( + html_template + .replace("__SERVER_NAME__", config.get("server_name", "LOGAR-Cloud-Hub")) + .replace("__TCP_PORT__", str(config.get("tcp_port", 9443))) + .replace("__HERMES_PORT__", str(config.get("hermes_port", 8443))) + .replace("__STATUS__", summary["status"]) + .replace("__BADGE_STATUS_CLS__", badge_status_cls) + .replace("__ACTIVE_SEATS__", str(summary["active_seats"])) + .replace("__REVOKED_SEATS__", str(summary["revoked_seats"])) + .replace("__MAX_SEATS__", str(summary["max_seats"])) + .replace("__AVAILABLE_SEATS__", str(summary["available_seats"])) + .replace("__PERCENT__", str(summary["percentage"])) + .replace("__BAR_COLOR__", bar_color) + .replace("__LICENSE_ID__", summary["license_id"]) + .replace("__EXPIRES_AT__", summary["not_valid_after"][:10] if summary["not_valid_after"] != "N/A" else "N/A") + .replace("__DAYS_REMAINING__", str(summary["days_remaining"])) + .replace("__CA_FP__", summary["fingerprint"][:16] + "...") + .replace("__ISSUER_URL__", summary["issuer_url"]) + .replace("__SUBJECT_CN__", summary["subject_cn"]) + .replace("__TOTAL_NODES__", str(total_nodes)) + .replace("__CLIENTS_HTML__", clients_html) + .replace("__ANOMALIES_HTML__", anomalies_html) + ) + return HTMLResponse(content=rendered) + + +@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") + parser.add_argument("--export-lic", type=str, help="Export a .lic license request file containing CA CSR and coordinates") + parser.add_argument("--import-ca", type=str, help="Import and activate a signed CA certificate or activation .lic file") + 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) + ) + + cert_dir = config.get("cert_dir", "certs") + + # Handle standalone CLI license export + if args.export_lic: + lic_dict, lic_json = export_license_request_lic(config) + with open(args.export_lic, "w", encoding="utf-8") as f: + f.write(lic_json) + print(f"[+] License request file successfully exported to: {os.path.abspath(args.export_lic)}") + sys.exit(0) + + # Handle standalone CLI CA import + if args.import_ca: + with open(args.import_ca, "r", encoding="utf-8") as f: + content = f.read() + ca_pem = content + if content.strip().startswith("{"): + data = json.loads(content) + ca_pem = data.get("ca_cert") or data.get("cert_pem") or content + + info = enrollment.extract_license_info_from_cert(ca_pem) + if not info.get("valid"): + print(f"[-] Invalid certificate in {args.import_ca}: {info.get('error')}") + sys.exit(1) + + ca_path = os.path.join(cert_dir, "ca.crt") + if os.path.exists(ca_path): + ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + try: + os.replace(ca_path, f"{ca_path}.{ts}.bak") + except Exception: + pass + with open(ca_path, "w", encoding="utf-8") as f: + f.write(ca_pem) + + new_seats = info.get("seats") or config.get("max_seats", 10) + config["max_seats"] = new_seats + conn = sqlite3.connect(config["db_path"]) + conn.execute("UPDATE license_config SET max_seats = ?, license_id = ? WHERE id = 1", (new_seats, info.get("license_id"))) + conn.commit() + conn.close() + print(f"[+] Successfully imported and activated signed CA certificate!") + print(f" - Granted Seats: {new_seats}") + print(f" - License ID: {info.get('license_id')}") + print(f" - Expires: {info.get('not_valid_after')}") + sys.exit(0) + + # Initialize dynamic PKI (Subordinate CA and Server TLS Certificate) + ca_cert, ca_key, ca_pem, ca_key_pem = enrollment.generate_ca_if_needed(cert_dir=cert_dir) + + # Auto-extract seats if ca.crt contains custom enterprise OID + ca_info = enrollment.extract_license_info_from_cert(ca_pem) + if ca_info.get("seats"): + config["max_seats"] = ca_info["seats"] + conn = sqlite3.connect(config["db_path"]) + conn.execute("UPDATE license_config SET max_seats = ? WHERE id = 1", (ca_info["seats"],)) + conn.commit() + conn.close() + + # Ensure CA CSR and exportable .lic exist for convenience + csr_path = os.path.join(cert_dir, "ca.csr") + if not os.path.exists(csr_path): + enrollment.generate_ca_csr(cert_dir=cert_dir) + export_license_request_lic(config) + + 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) + + hermes_host = config.get("hermes_host", "0.0.0.0") + hermes_port = config.get("hermes_port", 8443) + display_host = "localhost" if hermes_host in ("0.0.0.0", "127.0.0.1") else hermes_host + + print("=" * 64) + 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 ({ca_info.get('license_id') or 'Local Sub-CA'})") + print(f" Upstream Authority: {config.get('license_server_url', 'https://logar.eibl.tech')}") + print(f" Server Encryption Fingerprint: {config['server_fingerprint']}") + print(f" WebUI & Fleet Dashboard: http://{display_host}:{hermes_port}/webui") + print(f" Hermes Reporting API: http://{display_host}:{hermes_port}/api/hermes/report") + print("=" * 64) + + try: + asyncio.run(run_server()) + except KeyboardInterrupt: + print("\n[!] Server shutting down.") + + +if __name__ == "__main__": + main() \ No newline at end of file