diff --git a/LOGAR-LICENSING b/LOGAR-LICENSING
index 6f1af3e..30dfc9f 160000
--- a/LOGAR-LICENSING
+++ b/LOGAR-LICENSING
@@ -1 +1 @@
-Subproject commit 6f1af3ea71b2f954a3f63646f5e925de2e890e5a
+Subproject commit 30dfc9f77f8c4ba9bc232482184dcfe05e864065
diff --git a/README.md b/README.md
index 65faf08..a29b994 100644
--- a/README.md
+++ b/README.md
@@ -1,12 +1,12 @@
# LOGAR: Fleet Log Aggregation & Centralized Provisioning System
-### *Sichere Log-Zusammenführung, PKI-Flottenlizenzierung & Bereitstellung für autonome KI-Analyse*
+### *Secure Log Aggregation, PKI Fleet Licensing & Centralized Delivery for Autonomous AI Analysis*
-**LOGAR** is an enterprise-grade log aggregation, normalization, and centralized provisioning system (**"Log-Zusammenführung und Bereitstellung"**) built for heterogeneous server fleets (Windows & Linux). It pairs ultra-lightweight, zero-state edge forwarders with a secure centralized server hub that enforces mutual TLS 1.3 (**mTLS**) authentication, dynamic PKI fleet licensing with cryptographic seat leasing, 12-hour temporal persistence tracking, and direct anomaly pass-through—delivering clean, structured, verified log telemetry directly to **Hermes AI** for autonomous deep reasoning, contextual root-cause analysis, and operational decision-making.
+**LOGAR** is an enterprise-grade log aggregation, normalization, and centralized provisioning system built for heterogeneous server fleets (Windows & Linux). It pairs ultra-lightweight, zero-state edge forwarders with a secure centralized server hub that enforces mutual TLS 1.3 (**mTLS**) authentication, dynamic PKI fleet licensing with cryptographic seat leasing, 12-hour temporal persistence tracking, and direct anomaly pass-through—delivering clean, structured, verified log telemetry directly to **Hermes AI** for autonomous deep reasoning, contextual root-cause analysis, and operational decision-making.
---
## Table of Contents
-1. [Core Purpose & Philosophy (Log-Zusammenführung & Bereitstellung)](#core-purpose--philosophy)
+1. [Core Purpose & Philosophy (Log Aggregation & Provisioning)](#core-purpose--philosophy)
2. [End-to-End System Architecture](#end-to-end-system-architecture)
3. [System State Machines & Subsystem Lifecycles](#system-state-machines--subsystem-lifecycles)
- 3.1 [High-Level Overview State Machine](#31-high-level-system-state-machine)
@@ -32,7 +32,7 @@
## Core Purpose & Philosophy
-### 1. Centralized Log Aggregation & Provisioning (*Log-Zusammenführung & Bereitstellung*)
+### 1. Centralized Log Aggregation & Provisioning
The foundational objective of LOGAR is reliable, secure collection and structured delivery of system events:
- **Heterogeneous Fleet Aggregation**: Ingests Windows Event Logs (System, Application, Security) and Linux `systemd journalctl` events into a unified, normalized JSON data format.
- **Centralized Provisioning**: Serves as the authoritative clearinghouse and real-time streaming endpoint (`/api/hermes/report`), providing structured logs with temporal metadata to external consumers and automated AI agents.
diff --git a/src/Server.py b/src/Server.py
index 859f8be..2a62fd0 100644
--- a/src/Server.py
+++ b/src/Server.py
@@ -1,1809 +1,2666 @@
-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 = 1
-
-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 ingested log anomalies and updates temporal run tracking.
- All anomalies are promoted directly to VERIFIED for Hermes AI reasoning and decision-making.
- """
- 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
-
- 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):
- new_runs = 1
- new_first_seen = now_iso
- else:
- if last_run_id != run_id:
- new_runs = run_count + 1
- else:
- new_runs = run_count
- new_first_seen = first_seen_str
-
- new_status = "VERIFIED"
- if 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"
- 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 evaluate anomalies with temporal run tracking
- 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 verified anomalies within the active 12-hour evaluation window
- for Hermes AI direct reasoning and autonomous decision-making.
- """
- 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 or .lic file 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")
-
- raw_input = body.get("ca_cert") or body.get("lic") or body.get("raw")
- if not raw_input:
- raise HTTPException(status_code=400, detail="Missing 'lic' or 'ca_cert' payload in request")
-
- if isinstance(raw_input, dict):
- raw_text = json.dumps(raw_input)
- else:
- raw_text = str(raw_input)
-
- config = SERVER_STATE.get("config", {})
- cert_dir = config.get("cert_dir", "certs")
-
- try:
- res = enrollment.import_license_package(raw_text, cert_dir=cert_dir)
- except ValueError as ve:
- raise HTTPException(status_code=400, detail=str(ve))
-
- info = res["license_info"]
- ca_cert_pem = res["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")
-@app.post("/api/license/sync")
-async def trigger_license_renewal(request: Request):
- """
- Explicitly triggers automated mTLS sync and renewal with logar.eibl.tech.
- Supports picking up seat quota increases and certificate extensions in-flight.
- """
- requested_seats = None
- seat_increment = 0
- try:
- body = await request.json()
- if isinstance(body, dict):
- requested_seats = body.get("requested_seats")
- seat_increment = int(body.get("seat_increment", 0))
- except Exception:
- pass
-
- 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,
- requested_seats=requested_seats,
- seat_increment=seat_increment
- )
- 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(f"[+] In-flight mTLS SSLContext successfully reloaded after renewal/sync (Seats: {info.get('seats')}).")
-
- 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
-
-
-
-
-
-
-
-
-
-
-
-
__ACTIVE_SEATS__ / __MAX_SEATS__ Seats
-
__AVAILABLE_SEATS__ seats available for auto-enrollment
-
-
-
-
-
-
__LICENSE_ID__
-
Expires: __EXPIRES_AT__ (__DAYS_REMAINING__ days remaining)
-
Fingerprint: __CA_FP__
-
-
-
-
-
__ISSUER_URL__
-
Watchdog: In-flight auto-renewal triggers < 30d
-
Sub-CA Subject: __SUBJECT_CN__
-
-
-
-
-
__ACTIVE_SEATS__ Active Forwarders
-
__REVOKED_SEATS__ revoked forwarders blocked at socket
-
-
-
-
-
Licensing Controls:
-
- 📥 Export .lic Request
-
-
-
-
-
-
-
-
-
-
-
- | Status |
- Host Node |
- Client Identifier |
- OS Type |
- Cert Fingerprint |
- Last Seen |
- Seat Actions |
-
-
-
- __CLIENTS_HTML__
-
-
-
-
-
-
-
-
-
- | Severity |
- Source Node |
- Persistence |
- Verified Anomaly Message |
- Last Seen |
-
-
-
- __ANOMALIES_HTML__
-
-
-
-
-
-
-
-
-
-
- Submit this CSR or the exported .lic file to __ISSUER_URL__ to receive an activated .lic package embedding your seat quota.
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- Paste the activated .lic JSON package or ca.crt PEM certificate issued by __ISSUER_URL__.
-
-
-
-
-
-
-
-
-
-
-
-"""
-
- 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")
- parser.add_argument("--import-lic", type=str, help="Import and activate an activated .lic license file")
- parser.add_argument("--sync-license", action="store_true", help="Sync license with master authority and check for seat increases via mTLS")
- 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 / .lic import
- lic_file = args.import_lic or args.import_ca
- if lic_file:
- try:
- res = enrollment.import_license_package(
- lic_file,
- cert_dir=cert_dir,
- server_fingerprint=config.get("server_fingerprint", ""),
- hostname=config.get("server_name", "")
- )
- except Exception as e:
- print(f"[-] Invalid license file {lic_file}: {e}")
- sys.exit(1)
-
- info = res.get("license_info", {})
- new_seats = res.get("seats") or 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, res.get("license_id") or info.get("license_id")))
- conn.commit()
- conn.close()
-
- # Regenerate server certificate signed by new CA
- ca_cert, ca_key, _, _ = enrollment.generate_ca_if_needed(cert_dir=cert_dir, force_renew=False)
- 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
- )
-
- print(f"[+] Successfully imported and activated signed license!")
- print(f" - Granted Seats: {new_seats}")
- print(f" - License ID: {res.get('license_id') or info.get('license_id')}")
- print(f" - Expires: {info.get('not_valid_after') or res.get('expires_at') or 'N/A'}")
- sys.exit(0)
-
- # Handle standalone CLI license sync / update against master server
- if args.sync_license:
- lic_server = config.get("license_server_url", "wss://logar.eibl.tech/ws/license")
- print(f"[*] Connecting to master authority at {lic_server} via WebSocket to sync license...")
-
- ca_info = {}
- ca_path = os.path.join(cert_dir, "ca.crt")
- if os.path.exists(ca_path):
- with open(ca_path, "r", encoding="utf-8") as f:
- ca_info = enrollment.extract_license_info_from_cert(f.read())
-
- ws_url = lic_server if ("ws://" in lic_server or "wss://" in lic_server) else f"{lic_server.replace('https://', 'wss://').replace('http://', 'ws://').rstrip('/')}/ws/license"
-
- active_seats = 0
- if os.path.exists(config.get("db_path", "server_logs.db")):
- try:
- conn_tmp = sqlite3.connect(config.get("db_path", "server_logs.db"))
- row = conn_tmp.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'").fetchone()
- if row:
- active_seats = row[0]
- conn_tmp.close()
- except Exception:
- pass
-
- success, msg, res = enrollment.enroll_and_sync_license_ws(
- cert_dir=cert_dir,
- license_id=ca_info.get("license_id"),
- license_authority_ws_url=ws_url,
- server_fingerprint=config.get("server_fingerprint", ""),
- hostname=config.get("server_name", ""),
- seats_taken=active_seats
- )
-
- if not success or not res:
- # Fallback to mTLS HTTP sync
- print(f"[!] WebSocket sync returned: {msg}. Falling back to mTLS renewal...")
- http_url = lic_server.replace("wss://", "https://").replace("ws://", "http://").replace("/ws/license", "")
- success, msg, new_pem = enrollment.renew_ca_cert_mtls(cert_dir=cert_dir, license_server_url=http_url)
- if not success or not new_pem:
- print(f"[-] License sync failed: {msg}")
- sys.exit(1)
- info = enrollment.extract_license_info_from_cert(new_pem)
- new_seats = info.get("seats") or config.get("max_seats", 10)
- else:
- info = res.get("license_info", {})
- new_seats = res.get("seats", 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()
-
- ca_cert, ca_key, _, _ = enrollment.generate_ca_if_needed(cert_dir=cert_dir, force_renew=False)
- 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
- )
- print(f"[+] License successfully synchronized with master server!")
- print(f" - Active Seat Quota: {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
+import os
+import sys
+import json
+import uuid
+import struct
+import socket
+import sqlite3
+import argparse
+import asyncio
+import secrets
+import hashlib
+import hmac
+import warnings
+from datetime import datetime, timezone, timedelta
+from typing import Dict, Any, List, Optional, Tuple
+
+# 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, Depends
+from fastapi.responses import HTMLResponse, JSONResponse, Response, RedirectResponse
+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 = 1
+
+SESSION_COOKIE_NAME = "logar_server_session"
+SESSION_SECRET = os.environ.get("LOGAR_SESSION_SECRET", "logar_server_hub_session_secret_2026_!#*")
+DEFAULT_ADMIN_USER = os.environ.get("LOGAR_ADMIN_USER", "admin")
+DEFAULT_ADMIN_PASS = os.environ.get("LOGAR_ADMIN_PASS", "admin_logar_2026!")
+
+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
+
+
+class HermesTokenCreateRequest(BaseModel):
+ name: str
+
+
+class HermesTokenRevokeRequest(BaseModel):
+ token: str
+
+
+class ClientProfileExportRequest(BaseModel):
+ server_host: Optional[str] = None
+ server_port: Optional[int] = None
+ client_id: Optional[str] = None
+
+
+class ChangePasswordRequest(BaseModel):
+ old_password: str
+ new_password: str
+
+
+# --- Authentication & Password Hashing Utilities ---
+
+def hash_password(password: str, salt: Optional[str] = None) -> Tuple[str, str]:
+ """Hashes a password using PBKDF2-HMAC-SHA256 with 100,000 iterations."""
+ if not salt:
+ salt = secrets.token_hex(16)
+ pwd_hash = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), 100000).hex()
+ return pwd_hash, salt
+
+
+def verify_password(password: str, salt: str, password_hash: str) -> bool:
+ """Verifies a password against the stored salt and PBKDF2 hash."""
+ test_hash, _ = hash_password(password, salt=salt)
+ return hmac.compare_digest(test_hash, password_hash)
+
+
+def generate_session_token(username: str) -> str:
+ """Generates an HMAC-signed session cookie string."""
+ ts = int(datetime.now(timezone.utc).timestamp())
+ nonce = secrets.token_hex(8)
+ payload = f"{username}:{ts}:{nonce}"
+ sig = hmac.new(SESSION_SECRET.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest()
+ return f"{payload}:{sig}"
+
+
+def verify_session_token(token: str, max_age_hours: int = 24) -> Optional[str]:
+ """Verifies session token authenticity and age."""
+ if not token or ":" not in token:
+ return None
+ parts = token.split(":")
+ if len(parts) != 4:
+ return None
+ username, ts_str, nonce, sig = parts
+ payload = f"{username}:{ts_str}:{nonce}"
+ expected_sig = hmac.new(SESSION_SECRET.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest()
+ if not hmac.compare_digest(sig, expected_sig):
+ return None
+ try:
+ ts = int(ts_str)
+ if datetime.now(timezone.utc).timestamp() - ts > max_age_hours * 3600:
+ return None
+ except Exception:
+ return None
+ return username
+
+
+def get_current_user_optional(request: Request) -> Optional[str]:
+ """Extracts authenticated username from cookie or Bearer header if valid."""
+ cookie = request.cookies.get(SESSION_COOKIE_NAME)
+ if cookie:
+ user = verify_session_token(cookie)
+ if user:
+ return user
+ auth_header = request.headers.get("Authorization")
+ if auth_header and auth_header.startswith("Bearer "):
+ token = auth_header.split(" ", 1)[1].strip()
+ user = verify_session_token(token)
+ if user:
+ return user
+ return None
+
+
+def require_admin_auth(request: Request) -> str:
+ """Enforces administrator authentication for management endpoints."""
+ user = get_current_user_optional(request)
+ if not user:
+ raise HTTPException(status_code=401, detail="Authentication required. Please login.")
+ return user
+
+
+def verify_user_credentials(username: str, password: str, db_path: str = DEFAULT_DB_FILE) -> bool:
+ """Checks user credentials against SQLite users table."""
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute("SELECT password_hash, salt FROM users WHERE username = ?", (username,))
+ row = c.fetchone()
+ conn.close()
+ if not row:
+ return False
+ return verify_password(password, row[1], row[0])
+
+
+def change_user_password(username: str, new_password: str, db_path: str = DEFAULT_DB_FILE) -> bool:
+ """Updates a user's password."""
+ pwd_hash, salt = hash_password(new_password)
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute("UPDATE users SET password_hash = ?, salt = ? WHERE username = ?", (pwd_hash, salt, username))
+ changed = c.rowcount > 0
+ conn.commit()
+ conn.close()
+ return changed
+
+
+# --- Hermes Token Management ---
+
+def create_hermes_token(name: str, token: Optional[str] = None, db_path: str = DEFAULT_DB_FILE) -> Dict[str, Any]:
+ """Creates or saves a named Hermes API token."""
+ if not token:
+ token = f"hermes_{secrets.token_hex(20)}"
+ now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute(
+ "INSERT OR REPLACE INTO hermes_tokens (token, name, created_at, status) VALUES (?, ?, ?, 'active')",
+ (token, name, now_iso)
+ )
+ conn.commit()
+ conn.close()
+ return {"token": token, "name": name, "created_at": now_iso, "status": "active"}
+
+
+def list_hermes_tokens(db_path: str = DEFAULT_DB_FILE) -> List[Dict[str, Any]]:
+ """Lists all registered Hermes access tokens."""
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute("SELECT token, name, created_at, last_used, status FROM hermes_tokens ORDER BY created_at DESC")
+ rows = [
+ {"token": r[0], "name": r[1], "created_at": r[2], "last_used": r[3] or "Never", "status": r[4]}
+ for r in c.fetchall()
+ ]
+ conn.close()
+ return rows
+
+
+def revoke_hermes_token(token: str, db_path: str = DEFAULT_DB_FILE) -> bool:
+ """Revokes an active Hermes token."""
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute("UPDATE hermes_tokens SET status = 'revoked' WHERE token = ?", (token,))
+ changed = c.rowcount > 0
+ conn.commit()
+ conn.close()
+ return changed
+
+
+def delete_hermes_token(token: str, db_path: str = DEFAULT_DB_FILE) -> bool:
+ """Deletes a Hermes token."""
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute("DELETE FROM hermes_tokens WHERE token = ?", (token,))
+ changed = c.rowcount > 0
+ conn.commit()
+ conn.close()
+ return changed
+
+
+def verify_hermes_token(token: str, db_path: str = DEFAULT_DB_FILE) -> bool:
+ """Verifies a Hermes token and updates last_used timestamp."""
+ if not token:
+ return False
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute("SELECT token FROM hermes_tokens WHERE token = ? AND status = 'active'", (token,))
+ row = c.fetchone()
+ if row:
+ now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
+ c.execute("UPDATE hermes_tokens SET last_used = ? WHERE token = ?", (now_iso, token))
+ conn.commit()
+ conn.close()
+ return row is not None
+
+
+def require_hermes_or_admin_auth(request: Request) -> str:
+ """
+ Authorizes Hermes API access via Bearer token, query parameter,
+ active admin session cookie, or legacy auth_token.
+ """
+ # 1. Check if logged in admin user
+ user = get_current_user_optional(request)
+ if user:
+ return f"user:{user}"
+
+ # 2. Check Bearer token or token query param
+ token = None
+ auth_header = request.headers.get("Authorization")
+ if auth_header and auth_header.startswith("Bearer "):
+ token = auth_header.split(" ", 1)[1].strip()
+ if not token:
+ token = request.query_params.get("token")
+
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+
+ if token:
+ # Check against hermes_tokens table
+ if verify_hermes_token(token, db_path=db_path):
+ return f"hermes_token:{token[:8]}..."
+ # Check against config legacy auth_token
+ cfg_token = SERVER_STATE.get("config", {}).get("auth_token")
+ if cfg_token and secrets.compare_digest(str(token), str(cfg_token)):
+ return "config_auth_token"
+
+ # Backward compatibility: if no hermes tokens are defined in database yet, allow unauthenticated read
+ try:
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute("SELECT COUNT(*) FROM hermes_tokens WHERE status = 'active'")
+ active_count = c.fetchone()[0]
+ conn.close()
+ if active_count == 0:
+ return "anonymous_fallback"
+ except Exception:
+ pass
+
+ raise HTTPException(status_code=401, detail="Valid Hermes Bearer Token or Administrator Login Required")
+
+
+# --- Client Profile Export & Helper Utilities ---
+
+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"
+ if "admin_user" not in config:
+ config["admin_user"] = DEFAULT_ADMIN_USER
+ 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,
+ "admin_user": DEFAULT_ADMIN_USER,
+ "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 generate_client_profile_dict(
+ server_host: str,
+ server_port: int,
+ client_id: Optional[str] = None,
+ config_dict: Optional[Dict[str, Any]] = None
+) -> Dict[str, Any]:
+ """Generates a complete standalone client profile dictionary for export."""
+ cfg = config_dict or SERVER_STATE.get("config", {})
+ ca_pem = SERVER_STATE.get("ca_cert_pem", "")
+ if not ca_pem:
+ cert_dir = cfg.get("cert_dir", "certs")
+ ca_path = os.path.join(cert_dir, "ca.crt")
+ if os.path.exists(ca_path):
+ with open(ca_path, "r", encoding="utf-8") as f:
+ ca_pem = f.read()
+
+ return {
+ "format": "LOGAR_CLIENT_PROFILE",
+ "version": "2.0.1",
+ "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"),
+ "server_host": server_host,
+ "server_port": int(server_port),
+ "hermes_host": cfg.get("hermes_host", server_host),
+ "hermes_port": int(cfg.get("hermes_port", 8443)),
+ "enrollment_secret": cfg.get("enrollment_secret", ""),
+ "server_fingerprint": cfg.get("server_fingerprint", ""),
+ "server_public_key": cfg.get("public_key", ""),
+ "auth_token": cfg.get("auth_token", ""),
+ "ca_cert": ca_pem,
+ "client_id": client_id or ""
+ }
+
+
+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 format_relative_time(dt_str: Optional[str]) -> str:
+ """Calculates human-readable relative time from ISO or SQL timestamp string."""
+ if not dt_str or dt_str in ("N/A", "Never"):
+ return "Never"
+ try:
+ clean_str = dt_str.replace(" UTC", "")
+ if "T" in clean_str:
+ dt = datetime.fromisoformat(clean_str)
+ else:
+ dt = datetime.strptime(clean_str, "%Y-%m-%d %H:%M:%S")
+ if dt.tzinfo is None:
+ dt = dt.replace(tzinfo=timezone.utc)
+ now = datetime.now(timezone.utc)
+ diff = now - dt
+ seconds = int(diff.total_seconds())
+ if seconds < 0:
+ return "Just now"
+ if seconds < 60:
+ return f"{seconds}s ago"
+ minutes = seconds // 60
+ if minutes < 60:
+ return f"{minutes}m ago"
+ hours = minutes // 60
+ if hours < 24:
+ return f"{hours}h ago"
+ days = hours // 24
+ return f"{days}d ago"
+ except Exception:
+ return dt_str[:16] if dt_str else "N/A"
+
+
+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, users, and license quota."""
+ conn = sqlite3.connect(db_path)
+
+ # 1. Active Issues
+ 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
+ )
+ """)
+
+ # 2. Ingest Runs
+ conn.execute("""
+ CREATE TABLE IF NOT EXISTS ingest_runs (
+ run_id TEXT PRIMARY KEY,
+ site_name TEXT,
+ server TEXT,
+ timestamp TEXT,
+ log_count INTEGER
+ )
+ """)
+
+ # 3. License Config
+ 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'
+ )
+ """)
+
+ # Migrations for license_config
+ 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
+
+ # 4. Clients Table
+ 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
+ )
+ """)
+
+ # 5. Users Table
+ conn.execute("""
+ CREATE TABLE IF NOT EXISTS users (
+ username TEXT PRIMARY KEY,
+ password_hash TEXT NOT NULL,
+ salt TEXT NOT NULL,
+ role TEXT NOT NULL DEFAULT 'admin',
+ created_at TEXT NOT NULL
+ )
+ """)
+
+ # Ensure default admin user exists
+ c = conn.cursor()
+ c.execute("SELECT COUNT(*) FROM users")
+ if c.fetchone()[0] == 0:
+ now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
+ pwd_hash, salt = hash_password(DEFAULT_ADMIN_PASS)
+ conn.execute("INSERT INTO users (username, password_hash, salt, role, created_at) VALUES (?, ?, ?, 'admin', ?)",
+ (DEFAULT_ADMIN_USER, pwd_hash, salt, now_iso))
+
+ # 6. Hermes Tokens Table
+ conn.execute("""
+ CREATE TABLE IF NOT EXISTS hermes_tokens (
+ token TEXT PRIMARY KEY,
+ name TEXT NOT NULL,
+ created_at TEXT NOT NULL,
+ last_used TEXT,
+ status TEXT NOT NULL DEFAULT 'active'
+ )
+ """)
+
+ # Auto-initialize initial Hermes token if table is empty
+ c.execute("SELECT COUNT(*) FROM hermes_tokens")
+ if c.fetchone()[0] == 0:
+ initial_token = f"hermes_live_{secrets.token_hex(20)}"
+ now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
+ conn.execute("INSERT INTO hermes_tokens (token, name, created_at, status) VALUES (?, 'Default Hermes Agent', ?, 'active')",
+ (initial_token, now_iso))
+
+ 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 ingested log anomalies and updates temporal run tracking.
+ All anomalies are promoted directly to VERIFIED for Hermes AI reasoning and decision-making.
+ """
+ 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
+
+ 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):
+ new_runs = 1
+ new_first_seen = now_iso
+ else:
+ if last_run_id != run_id:
+ new_runs = run_count + 1
+ else:
+ new_runs = run_count
+ new_first_seen = first_seen_str
+
+ new_status = "VERIFIED"
+ if 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"
+ 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)."""
+ 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 length-prefixed JSON payload, and ingests candidate logs.
+ """
+ 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
+ now_sql = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+ c.execute("UPDATE clients SET last_seen = ? WHERE client_id = ?", (now_sql, client_id))
+ conn.commit()
+ conn.close()
+
+ try:
+ length_bytes = await reader.readexactly(4)
+ length = struct.unpack(">I", length_bytes)[0]
+ if length <= 0 or length > 10 * 1024 * 1024:
+ raise ValueError(f"Invalid frame size: {length}")
+
+ payload_bytes = await reader.readexactly(length)
+ raw_payload = json.loads(payload_bytes.decode("utf-8"))
+
+ 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
+
+ if client_id and "server" not in log_payload:
+ log_payload["server"] = client_id
+
+ 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
+
+
+# --- API Endpoints: Authentication, Clients, Hermes Tokens & Licensing ---
+
+@app.get("/login", response_class=HTMLResponse)
+def login_page(request: Request):
+ """Renders the dark mode login page for the aggregation server."""
+ if get_current_user_optional(request):
+ return RedirectResponse(url="/", status_code=303)
+
+ return HTMLResponse(content="""
+
+
+
+
+ LOGAR Hub - Login
+
+
+
+
+
+
+""")
+
+
+@app.post("/login")
+async def process_login(request: Request):
+ """Processes login credentials and issues session cookie."""
+ username, password = None, None
+ content_type = request.headers.get("content-type", "")
+ if "application/json" in content_type:
+ try:
+ body = await request.json()
+ username = body.get("username")
+ password = body.get("password")
+ except Exception:
+ pass
+ else:
+ form = await request.form()
+ username = form.get("username")
+ password = form.get("password")
+
+ if not username or not password:
+ raise HTTPException(status_code=400, detail="Username and password are required")
+
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ if not verify_user_credentials(username, password, db_path=db_path):
+ raise HTTPException(status_code=401, detail="Invalid username or password")
+
+ token = generate_session_token(username)
+ response = JSONResponse(content={"status": "success", "username": username, "token": token})
+ response.set_cookie(
+ key=SESSION_COOKIE_NAME,
+ value=token,
+ httponly=True,
+ max_age=86400,
+ samesite="lax"
+ )
+ return response
+
+
+@app.get("/logout")
+def process_logout():
+ """Logs out and redirects to /login."""
+ resp = RedirectResponse(url="/login", status_code=303)
+ resp.delete_cookie(SESSION_COOKIE_NAME)
+ return resp
+
+
+@app.post("/api/auth/change-password")
+async def api_change_password(req: ChangePasswordRequest, current_user: str = Depends(require_admin_auth)):
+ """Allows authenticated administrator to change password."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ if not verify_user_credentials(current_user, req.old_password, db_path=db_path):
+ raise HTTPException(status_code=403, detail="Current password incorrect")
+ if len(req.new_password) < 6:
+ raise HTTPException(status_code=400, detail="New password must be at least 6 characters")
+ change_user_password(current_user, req.new_password, db_path=db_path)
+ return {"status": "success", "message": "Password updated successfully"}
+
+
+@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.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ 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)
+ now_sql = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+ c.execute("""
+ UPDATE clients
+ SET hostname = ?, os_type = ?, cert_fingerprint = ?, last_seen = ?
+ WHERE client_id = ?
+ """, (req.hostname, req.os, fp, now_sql, 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)
+ now_sql = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+ c.execute("""
+ INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, first_seen, last_seen)
+ VALUES (?, ?, ?, ?, 'active', ?, ?)
+ """, (req.client_id, req.hostname, req.os, fp, now_sql, now_sql))
+ 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/client/export-profile")
+@app.post("/api/client/export-profile")
+def export_client_profile(
+ server_host: Optional[str] = None,
+ server_port: Optional[int] = None,
+ client_id: Optional[str] = None
+):
+ """
+ Exports a standalone client profile JSON configuration bundle
+ for zero-touch edge client bootstrapping.
+ """
+ config = SERVER_STATE.get("config", {})
+ host = server_host or config.get("tcp_host", "127.0.0.1")
+ if host in ("0.0.0.0", ""):
+ host = socket.gethostname()
+ port = int(server_port or config.get("tcp_port", 9443))
+
+ profile = generate_client_profile_dict(
+ server_host=host,
+ server_port=port,
+ client_id=client_id,
+ config_dict=config
+ )
+ profile_json = json.dumps(profile, indent=2)
+
+ return Response(
+ content=profile_json,
+ media_type="application/json",
+ headers={"Content-Disposition": 'attachment; filename="client_config.json"'}
+ )
+
+
+@app.get("/api/clients")
+def list_clients():
+ """Returns all registered clients, seat usage, and activity timestamps."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ 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 ORDER BY last_seen DESC")
+ 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],
+ "last_activity_relative": format_relative_time(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.post("/api/clients/approve")
+@app.post("/api/clients/reactivate")
+def approve_client(action: ClientActionRequest):
+ """Approves / reactivates a client certificate if within seat quota."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ 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 activate seat: maximum 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"[+] Approved/Reactivated client seat: {action.client_id}")
+ return {"status": "active", "client_id": action.client_id}
+
+
+# Backward-compatible function alias for test suite
+reactivate_client = approve_client
+
+
+
+@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.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ 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/delete")
+@app.post("/api/clients/remove")
+def delete_client(action: ClientActionRequest):
+ """Permanently removes a client forwarder record from SQLite, immediately freeing its seat."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ conn = sqlite3.connect(db_path)
+ c = conn.cursor()
+ c.execute("DELETE FROM clients 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"[+] Removed client forwarder record: {action.client_id}")
+ return {"status": "deleted", "client_id": action.client_id}
+
+
+# --- Hermes API Tokens Endpoints ---
+
+@app.get("/api/hermes/tokens", dependencies=[Depends(require_admin_auth)])
+def api_list_hermes_tokens():
+ """Lists all Hermes access tokens."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ return {"tokens": list_hermes_tokens(db_path=db_path)}
+
+
+@app.post("/api/hermes/tokens/create", dependencies=[Depends(require_admin_auth)])
+def api_create_hermes_token(req: HermesTokenCreateRequest):
+ """Generates a new Hermes API token."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ token_data = create_hermes_token(name=req.name, db_path=db_path)
+ return {"status": "success", "token": token_data}
+
+
+@app.post("/api/hermes/tokens/revoke", dependencies=[Depends(require_admin_auth)])
+def api_revoke_hermes_token(req: HermesTokenRevokeRequest):
+ """Revokes a Hermes API token."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ ok = revoke_hermes_token(req.token, db_path=db_path)
+ if not ok:
+ raise HTTPException(status_code=404, detail="Token not found")
+ return {"status": "success", "token": req.token}
+
+
+@app.post("/api/hermes/tokens/delete", dependencies=[Depends(require_admin_auth)])
+def api_delete_hermes_token(req: HermesTokenRevokeRequest):
+ """Deletes a Hermes API token."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ ok = delete_hermes_token(req.token, db_path=db_path)
+ if not ok:
+ raise HTTPException(status_code=404, detail="Token not found")
+ return {"status": "success", "token": req.token}
+
+
+# --- Hermes Reporting Endpoints ---
+
+@app.get("/api/hermes/report")
+def get_verified_anomalies_for_hermes(auth: str = Depends(require_hermes_or_admin_auth)):
+ """
+ Ingestion endpoint for Hermes agentic workflows.
+ Returns verified anomalies within the active 12-hour evaluation window
+ for Hermes AI direct reasoning and autonomous decision-making.
+ """
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ 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()
+
+ window_hours = SERVER_STATE.get("config", {}).get("evaluation_window_hours", 12)
+ 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"{window_hours}h",
+ "verified": True,
+ "status": r[10]
+ })
+
+ return report
+
+
+@app.get("/api/hermes/all")
+def get_all_issues(auth: str = Depends(require_hermes_or_admin_auth)):
+ """Diagnostic endpoint to inspect both transient candidate blips and verified anomalies."""
+ db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
+ 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 or .lic file."""
+ try:
+ body = await req.json()
+ except Exception:
+ raise HTTPException(status_code=400, detail="Invalid JSON body")
+
+ raw_input = body.get("ca_cert") or body.get("lic") or body.get("raw")
+ if not raw_input:
+ raise HTTPException(status_code=400, detail="Missing 'lic' or 'ca_cert' payload in request")
+
+ if isinstance(raw_input, dict):
+ raw_text = json.dumps(raw_input)
+ else:
+ raw_text = str(raw_input)
+
+ config = SERVER_STATE.get("config", {})
+ cert_dir = config.get("cert_dir", "certs")
+
+ try:
+ res = enrollment.import_license_package(raw_text, cert_dir=cert_dir)
+ except ValueError as ve:
+ raise HTTPException(status_code=400, detail=str(ve))
+
+ info = res["license_info"]
+ ca_cert_pem = res["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")
+@app.post("/api/license/sync")
+async def trigger_license_renewal(request: Request):
+ """Explicitly triggers automated mTLS sync and renewal with logar.eibl.tech."""
+ requested_seats = None
+ seat_increment = 0
+ try:
+ body = await request.json()
+ if isinstance(body, dict):
+ requested_seats = body.get("requested_seats")
+ seat_increment = int(body.get("seat_increment", 0))
+ except Exception:
+ pass
+
+ 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,
+ requested_seats=requested_seats,
+ seat_increment=seat_increment
+ )
+ 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(f"[+] In-flight mTLS SSLContext successfully reloaded after renewal/sync (Seats: {info.get('seats')}).")
+
+ return {
+ "status": "success",
+ "message": msg,
+ "seats": info.get("seats"),
+ "expires_at": info.get("not_valid_after"),
+ "days_remaining": info.get("days_remaining")
+ }
+
+
+# --- Comprehensive Info WebUI Dashboard ---
+
+@app.get("/", response_class=HTMLResponse)
+@app.get("/webui", response_class=HTMLResponse)
+def serve_webui(request: Request = None):
+ """Renders the comprehensive LOGAR Server Hub & Fleet Licensing WebUI."""
+ auth_user = None
+ if request is not None:
+ auth_user = get_current_user_optional(request)
+ if not auth_user:
+ return RedirectResponse(url="/login", status_code=303)
+ else:
+ auth_user = DEFAULT_ADMIN_USER
+
+ summary = get_license_summary()
+ config = SERVER_STATE.get("config", {})
+ db_path = config.get("db_path", DEFAULT_DB_FILE)
+
+
+ # Fetch clients
+ 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()
+
+ # Fetch hermes tokens
+ c.execute("SELECT token, name, created_at, last_used, status FROM hermes_tokens ORDER BY created_at DESC")
+ hermes_tokens = c.fetchall()
+ conn.close()
+
+ # Build client rows with exact + relative activity time and seat management actions
+ client_rows = []
+ for cl in clients:
+ cid, host, ost, fp, st, first, last = cl
+ badge_cls = "badge-active" if st == "active" else "badge-revoked"
+ rel_time = format_relative_time(last)
+ last_display = f"{last[:19]}
{rel_time}
"
+
+ btn_action = (
+ f''
+ if st == "active" else
+ f''
+ )
+ btn_del = f''
+
+ client_rows.append(f"""
+
+ | {st.upper()} |
+ {host} |
+ {cid} |
+ {ost.upper()} |
+ {fp[:16]}... |
+ {last_display} |
+
+
+ {btn_action}
+ {btn_del}
+
+ |
+
+ """)
+ 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[:80] + "..." if len(msg) > 80 else msg
+ anomaly_rows.append(f"""
+
+ | {sev} |
+ {srv} |
+ {rc} runs |
+ {short_msg} |
+ {ls[:19]} |
+
+ """)
+ anomalies_html = "".join(anomaly_rows) if anomaly_rows else '| No verified anomalies currently reported. |
'
+
+ # Build hermes tokens rows
+ token_rows = []
+ for tk in hermes_tokens:
+ tok, tname, tcreated, tlast, tstatus = tk
+ tbadge = "badge-active" if tstatus == "active" else "badge-revoked"
+ masked = tok[:10] + "••••••••" + tok[-4:]
+ tok_action = f'' if tstatus == "active" else f''
+ token_rows.append(f"""
+
+ | {tname} |
+ {masked} |
+ {tstatus.upper()} |
+ {tcreated[:19]} |
+ {tlast[:19] if tlast else 'Never'} |
+
+
+
+ {tok_action}
+
+ |
+
+ """)
+ tokens_html = "".join(token_rows) if token_rows else '| No Hermes API tokens created yet. |
'
+
+ # 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")
+ total_nodes = summary["active_seats"] + summary["revoked_seats"]
+
+ detected_host = (request.headers.get("host", "").split(":")[0] if request and request.headers else "") or config.get("tcp_host", "127.0.0.1")
+ if detected_host in ("0.0.0.0", "localhost", ""):
+ detected_host = socket.gethostname()
+
+ html_content = f"""
+
+
+
+
+ LOGAR Hub - Fleet Aggregation & Management
+
+
+
+
+
+
+
+
🛰️
+
+
LOGAR Fleet Hub & License Management
+
Fleet Log Aggregation & Centralized AI Provisioning
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
{summary['active_seats']} / {summary['max_seats']} Seats
+
{summary['available_seats']} seats available for automatic leasing
+
+
+
+
+
+
mTLS Port {config.get('tcp_port', 9443)}
+
Hermes Provisioning API: Port {config.get('hermes_port', 8443)}
+
Fingerprint: {summary['server_fingerprint'][:16]}...
+
+
+
+
+
{summary['license_id']}
+
Expires: {summary['not_valid_after'][:10] if summary['not_valid_after'] != 'N/A' else 'N/A'} ({summary['days_remaining']} days left)
+
Authority: {summary['issuer_url']}
+
+
+
+
+
{summary['active_seats']} Active Nodes
+
{summary['revoked_seats']} revoked nodes blocked at TLS socket
+
+
+
+
+
+
+
+
+
+
+
+ | Status |
+ Host Node |
+ Client Identifier |
+ OS |
+ Cert Fingerprint |
+ Last Activity Time |
+ Seat Actions |
+
+
+
+ {clients_html}
+
+
+
+
+
+
+
+
+
+
+
+
+ Generate Dedicated Hermes Token
+
+
+
+
+
+
+
+
+
+ | Token Name |
+ Bearer Token |
+ Status |
+ Created |
+ Last Used |
+ Action |
+
+
+
+ {tokens_html}
+
+
+
+
+
+
🚀 Hermes AI Integration Quickstart
+
Configure Hermes to query the verified log telemetry stream with Bearer authorization:
+
curl -H "Authorization: Bearer <HERMES_TOKEN>" http://{detected_host}:{config.get('hermes_port', 8443)}/api/hermes/report
+
+
+
+
+
+
+
+
+
+
+
+
+ | Severity |
+ Source Node |
+ Persistence |
+ Normalized Anomaly Message |
+ Last Ingested |
+
+
+
+ {anomalies_html}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Subordinate CA Common Name
+
{summary['subject_cn']}
+
+
+
Master Root Issuer
+
{summary['issuer_cn']}
+
+
+
Issued Seat Quota
+
{summary['max_seats']} Seats
+
+
+
Certificate Validity
+
{summary['not_valid_after'][:19] if summary['not_valid_after'] != 'N/A' else 'N/A'}
+
+
+
+
+
+
+
+
+
+
📥 Export Edge Client Profile
+
+ Download a pre-configured client_config.json profile bundle for zero-touch Windows & Linux forwarder deployment.
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Edge Client Quick Deploy Command
+
sudo ./compilation/install_linux_client.sh "http://{detected_host}:{config.get('hermes_port', 8443)}" "{config.get('enrollment_secret')}"
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Certificate Signing Request (ca.csr)
+
+
+
+
+
+
+
+
+
+
+
+
Import & Activate Signed License (.lic / ca.crt)
+
Paste the activated .lic package or PEM certificate issued by the Master Authority:
+
+
+
+
+
+
+
+
+
+
+
+
Change Password
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+"""
+ return HTMLResponse(content=html_content)
+
+
+@app.get("/health")
+def health_check():
+ return {
+ "status": "healthy",
+ "server_name": SERVER_STATE.get("config", {}).get("server_name", "LOGAR-Cloud-Hub"),
+ "fingerprint": SERVER_STATE.get("config", {}).get("server_fingerprint", ""),
+ "tcp_port": SERVER_STATE.get("config", {}).get("tcp_port", 9443),
+ "hermes_port": SERVER_STATE.get("config", {}).get("hermes_port", 8443),
+ "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")
+ parser.add_argument("--import-lic", type=str, help="Import and activate an activated .lic license file")
+ parser.add_argument("--sync-license", action="store_true", help="Sync license with master authority and check for seat increases via mTLS")
+ 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 / .lic import
+ lic_file = args.import_lic or args.import_ca
+ if lic_file:
+ try:
+ res = enrollment.import_license_package(
+ lic_file,
+ cert_dir=cert_dir,
+ server_fingerprint=config.get("server_fingerprint", ""),
+ hostname=config.get("server_name", "")
+ )
+ except Exception as e:
+ print(f"[-] Invalid license file {lic_file}: {e}")
+ sys.exit(1)
+
+ info = res.get("license_info", {})
+ new_seats = res.get("seats") or 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, res.get("license_id") or info.get("license_id")))
+ conn.commit()
+ conn.close()
+
+ # Regenerate server certificate signed by new CA
+ ca_cert, ca_key, _, _ = enrollment.generate_ca_if_needed(cert_dir=cert_dir, force_renew=False)
+ 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
+ )
+
+ print(f"[+] Successfully imported and activated signed license!")
+ print(f" - Granted Seats: {new_seats}")
+ print(f" - License ID: {res.get('license_id') or info.get('license_id')}")
+ print(f" - Expires: {info.get('not_valid_after') or res.get('expires_at') or 'N/A'}")
+ sys.exit(0)
+
+ # Handle standalone CLI license sync / update against master server
+ if args.sync_license:
+ lic_server = config.get("license_server_url", "wss://logar.eibl.tech/ws/license")
+ print(f"[*] Connecting to master authority at {lic_server} via WebSocket to sync license...")
+
+ ca_info = {}
+ ca_path = os.path.join(cert_dir, "ca.crt")
+ if os.path.exists(ca_path):
+ with open(ca_path, "r", encoding="utf-8") as f:
+ ca_info = enrollment.extract_license_info_from_cert(f.read())
+
+ ws_url = lic_server if ("ws://" in lic_server or "wss://" in lic_server) else f"{lic_server.replace('https://', 'wss://').replace('http://', 'ws://').rstrip('/')}/ws/license"
+
+ active_seats = 0
+ if os.path.exists(config.get("db_path", "server_logs.db")):
+ try:
+ conn_tmp = sqlite3.connect(config.get("db_path", "server_logs.db"))
+ row = conn_tmp.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'").fetchone()
+ if row:
+ active_seats = row[0]
+ conn_tmp.close()
+ except Exception:
+ pass
+
+ success, msg, res = enrollment.enroll_and_sync_license_ws(
+ cert_dir=cert_dir,
+ license_id=ca_info.get("license_id"),
+ license_authority_ws_url=ws_url,
+ server_fingerprint=config.get("server_fingerprint", ""),
+ hostname=config.get("server_name", ""),
+ seats_taken=active_seats
+ )
+
+ if not success or not res:
+ # Fallback to mTLS HTTP sync
+ print(f"[!] WebSocket sync returned: {msg}. Falling back to mTLS renewal...")
+ http_url = lic_server.replace("wss://", "https://").replace("ws://", "http://").replace("/ws/license", "")
+ success, msg, new_pem = enrollment.renew_ca_cert_mtls(cert_dir=cert_dir, license_server_url=http_url)
+ if not success or not new_pem:
+ print(f"[-] License sync failed: {msg}")
+ sys.exit(1)
+ info = enrollment.extract_license_info_from_cert(new_pem)
+ new_seats = info.get("seats") or config.get("max_seats", 10)
+ else:
+ info = res.get("license_info", {})
+ new_seats = res.get("seats", 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()
+
+ ca_cert, ca_key, _, _ = enrollment.generate_ca_if_needed(cert_dir=cert_dir, force_renew=False)
+ 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
+ )
+ print(f"[+] License successfully synchronized with master server!")
+ print(f" - Active Seat Quota: {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
diff --git a/tests/test_server.py b/tests/test_server.py
index e62c5f4..3b075a2 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -329,6 +329,119 @@ class TestServerComponent(unittest.TestCase):
if os.path.exists(test_cert_dir):
shutil.rmtree(test_cert_dir, ignore_errors=True)
+ def test_user_auth_and_session(self):
+ """Tests admin password hashing, verification, session tokens, and password change."""
+ Server.init_db(self.test_db)
+
+ # Verify default admin credentials
+ self.assertTrue(Server.verify_user_credentials("admin", "admin_logar_2026!", db_path=self.test_db))
+ self.assertFalse(Server.verify_user_credentials("admin", "wrong_password", db_path=self.test_db))
+ self.assertFalse(Server.verify_user_credentials("nonexistent", "password", db_path=self.test_db))
+
+ # Test session token generation & validation
+ token = Server.generate_session_token("admin")
+ self.assertEqual(Server.verify_session_token(token), "admin")
+ self.assertIsNone(Server.verify_session_token("invalid.token.structure"))
+
+ # Test changing password
+ Server.change_user_password("admin", "new_secure_pass123", db_path=self.test_db)
+ self.assertFalse(Server.verify_user_credentials("admin", "admin_logar_2026!", db_path=self.test_db))
+ self.assertTrue(Server.verify_user_credentials("admin", "new_secure_pass123", db_path=self.test_db))
+
+ def test_hermes_tokens_and_auth(self):
+ """Tests Hermes Bearer token creation, verification, revocation, and deletion."""
+ Server.init_db(self.test_db)
+
+ # Test default Hermes token was created on init
+ tokens = Server.list_hermes_tokens(db_path=self.test_db)
+ self.assertEqual(len(tokens), 1)
+ self.assertEqual(tokens[0]["status"], "active")
+ default_token = tokens[0]["token"]
+ self.assertTrue(Server.verify_hermes_token(default_token, db_path=self.test_db))
+
+ # Test creating new token
+ new_tok = Server.create_hermes_token(name="Hermes Autonomous Agent 01", db_path=self.test_db)
+ self.assertIn("token", new_tok)
+ self.assertEqual(new_tok["name"], "Hermes Autonomous Agent 01")
+ self.assertTrue(Server.verify_hermes_token(new_tok["token"], db_path=self.test_db))
+
+ # Test revoking token
+ self.assertTrue(Server.revoke_hermes_token(new_tok["token"], db_path=self.test_db))
+ self.assertFalse(Server.verify_hermes_token(new_tok["token"], db_path=self.test_db))
+
+ # Test deleting token
+ self.assertTrue(Server.delete_hermes_token(new_tok["token"], db_path=self.test_db))
+ tokens_after = Server.list_hermes_tokens(db_path=self.test_db)
+ self.assertEqual(len(tokens_after), 1)
+
+ def test_client_profile_export(self):
+ """Tests client profile dictionary generation and export formatting."""
+ cfg = Server.load_or_init_config(self.test_config)
+ profile = Server.generate_client_profile_dict(
+ server_host="logar-hub.corp.internal",
+ server_port=9443,
+ client_id="worker-node-101",
+ config_dict=cfg
+ )
+ self.assertEqual(profile["server_host"], "logar-hub.corp.internal")
+ self.assertEqual(profile["server_port"], 9443)
+ self.assertEqual(profile["client_id"], "worker-node-101")
+ self.assertEqual(profile["server_fingerprint"], cfg["server_fingerprint"])
+ self.assertEqual(profile["server_public_key"], cfg["public_key"])
+ self.assertEqual(profile["enrollment_secret"], cfg["enrollment_secret"])
+ self.assertEqual(profile["format"], "LOGAR_CLIENT_PROFILE")
+
+ def test_client_seat_management_and_relative_time(self):
+ """Tests client listing, seat approval, revocation, deletion, and relative timestamp helper."""
+ Server.init_db(self.test_db, enrollment_secret="test-sec-123", max_seats=5)
+ Server.SERVER_STATE["config"] = {"db_path": self.test_db}
+
+ # Populate test clients
+ conn = sqlite3.connect(self.test_db)
+ now = datetime.now(timezone.utc)
+ five_min_ago = (now - timedelta(minutes=5)).strftime("%Y-%m-%d %H:%M:%S")
+ two_hours_ago = (now - timedelta(hours=2)).strftime("%Y-%m-%d %H:%M:%S")
+
+ conn.execute("INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, last_seen) VALUES ('c-win', 'WIN-PROD-01', 'windows', 'FP_W1', 'active', ?)", (five_min_ago,))
+ conn.execute("INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, last_seen) VALUES ('c-lin', 'LIN-PROD-01', 'linux', 'FP_L1', 'active', ?)", (two_hours_ago,))
+ conn.commit()
+ conn.close()
+
+ # 1. Test listing
+ client_data = Server.list_clients()
+ self.assertEqual(client_data["active_seats"], 2)
+ self.assertEqual(client_data["max_seats"], 5)
+ self.assertEqual(len(client_data["clients"]), 2)
+ self.assertEqual(client_data["clients"][0]["last_activity_relative"], "5m ago")
+
+ # 2. Test revoking
+ rev_req = Server.ClientActionRequest(client_id="c-win")
+ rev_res = Server.revoke_client(rev_req)
+ self.assertEqual(rev_res["status"], "revoked")
+ client_data = Server.list_clients()
+ self.assertEqual(client_data["active_seats"], 1)
+
+ # 3. Test approving / reactivating
+ app_res = Server.approve_client(rev_req)
+ self.assertEqual(app_res["status"], "active")
+ client_data = Server.list_clients()
+ self.assertEqual(client_data["active_seats"], 2)
+
+ # 4. Test deleting client record
+ del_res = Server.delete_client(rev_req)
+ self.assertEqual(del_res["status"], "deleted")
+ client_data = Server.list_clients()
+ self.assertEqual(len(client_data["clients"]), 1)
+ self.assertEqual(client_data["active_seats"], 1)
+
+ # 5. Test format_relative_time edge cases
+ self.assertEqual(Server.format_relative_time(None), "Never")
+ self.assertEqual(Server.format_relative_time(""), "Never")
+ self.assertEqual(Server.format_relative_time("invalid-date-string"), "invalid-date-str")
+ just_now = now.strftime("%Y-%m-%d %H:%M:%S")
+ self.assertIn(Server.format_relative_time(just_now), ("0s ago", "1s ago", "Just now"))
+
if __name__ == "__main__":
unittest.main()
+