me0nline/LOGAR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity
Files
9ebfadbe89bde011fd31ed6d6503fac0086b13a7
LOGAR/src/Server.py
T
me0nline 9ebfadbe89
CI Test Suite / Run Component Tests & Pipeline Verification (push) Failing after 1m28s
Details
Optimize server startup speed, client update caching, and HTTP 304 conditional downloads
2026-10-02 22:43:18 +02:00

2456 lines
99 KiB
Python
Raw Blame History

"""
LOGAR - Central Cloud Ingestion Hub & Fleet Licensing Server
Main orchestrator and FastAPI application server.
Delegates domain responsibilities to specialized modular managers:
- server_models: Pydantic request & response data models
- auth_manager: Session, Rate Limiting, Password, 2FA TOTP, Hermes token managers
- ingest_queue: Ingestion ring buffering and batch processing
- storage_engine: Multi-engine SQL schema, migrations, and data-at-rest encryption
- license_manager: Sub-CA licensing, OID parsing, and seat synchronization
- pki_manager: TLS 1.3 mTLS context creation and dynamic certificate rotation
- web_ui: Dark-mode administrative WebUI & reactive dashboard renderer
"""
import os
import sys
import json
import uuid
import time
import struct
import socket
import sqlite3
import argparse
import asyncio
import secrets
import warnings
import threading
from datetime import datetime, timezone, timedelta
from typing import Dict, Any, List, Optional, Tuple
# Ensure script directory and repo root are in sys.path
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if SCRIPT_DIR not in sys.path:
sys.path.insert(0, SCRIPT_DIR)
REPO_ROOT = os.path.abspath(os.path.join(SCRIPT_DIR, ".."))
if REPO_ROOT not in sys.path:
sys.path.insert(0, REPO_ROOT)
# Suppress cryptography / pgpy deprecation notices for a clean terminal output
warnings.filterwarnings("ignore")
import pgpy
import ssl
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.responses import HTMLResponse, JSONResponse, Response, RedirectResponse, FileResponse
import uvicorn
try:
from src import server_enrollment as enrollment
except ImportError:
import server_enrollment as enrollment
try:
from src.version import LOGAR_VERSION, is_newer_version, compare_versions
from src.update_manager import ServerUpdateManager
except ImportError:
from version import LOGAR_VERSION, is_newer_version, compare_versions
from update_manager import ServerUpdateManager
try:
from src.log_collector import (
get_machine_identifier,
WindowsLogCollector,
LinuxLogCollector,
get_log_collector,
is_routine_service_noise
)
except ImportError:
from log_collector import (
get_machine_identifier,
WindowsLogCollector,
LinuxLogCollector,
get_log_collector,
is_routine_service_noise
)
# Import specialized modular components
from src.server_models import (
ClientEnrollRequest,
ClientActionRequest,
LicenseImportRequest,
HermesTokenCreateRequest,
HermesTokenRevokeRequest,
ClientProfileExportRequest,
ChangePasswordRequest,
DbConnectionTestRequest,
DbSwitchBackendRequest,
DbFullConfigRequest,
ClientUpdateRequest,
ServerUpdateCheckRequest
)
from src.auth_manager import (
SessionManager,
RateLimiter,
PasswordManager,
TOTPManager,
HermesTokenManager
)
from src.ingest_queue import IngestQueueManager
from src.storage_engine import StorageEngine, StorageCryptoManager
from src.license_manager import LicenseManager
from src.pki_manager import PKIManager
from src.web_ui import WebUIRenderer, format_relative_time
CONFIG_FILE_NAME = "server_config.json"
DEFAULT_DB_FILE = "logar_state.db"
EVALUATION_WINDOW_HOURS = 12
RUN_THRESHOLD = 1
SESSION_COOKIE_NAME = SessionManager.SESSION_COOKIE_NAME
DEFAULT_ADMIN_USER = StorageEngine.DEFAULT_ADMIN_USER
DEFAULT_ADMIN_PASS = StorageEngine.DEFAULT_ADMIN_PASS
app = FastAPI(title="LOGAR Cloud Ingestion & Hermes Hub", version=LOGAR_VERSION)
# Global context holding server state
UPDATE_MANAGER = ServerUpdateManager()
SERVER_STATE: Dict[str, Any] = {
"update_manager": UPDATE_MANAGER
}
# Global rate limiter & ingestion queue
RATE_LIMITER = RateLimiter()
INGEST_QUEUE = IngestQueueManager()
# =====================================================================
# Backward-Compatible Auth & Helper Functions (delegating to managers)
# =====================================================================
def get_session_secret(db_path: Optional[str] = None) -> str:
target_db = db_path or SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE
return SessionManager.get_session_secret(db_path=target_db, fallback_db=DEFAULT_DB_FILE)
SESSION_SECRET = os.environ.get("LOGAR_SESSION_SECRET") or secrets.token_hex(32)
def get_client_ip(request: Optional[Request] = None) -> str:
return RateLimiter.get_client_ip(request)
def check_ip_rate_limit(
ip: str,
store: Any = None,
max_attempts: int = 10,
window_seconds: int = 60,
scope: str = "login",
db_path: Optional[str] = None
) -> bool:
target_db = db_path or SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE
return RATE_LIMITER.check_ip_rate_limit(
ip=ip, store=store, max_attempts=max_attempts, window_seconds=window_seconds,
scope=scope, db_path=target_db, fallback_db=DEFAULT_DB_FILE
)
def record_failed_ip_attempt(
ip: str,
store: Any = None,
scope: str = "login",
db_path: Optional[str] = None
):
target_db = db_path or SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE
RATE_LIMITER.record_failed_ip_attempt(
ip=ip, store=store, scope=scope, db_path=target_db, fallback_db=DEFAULT_DB_FILE
)
def reset_failed_ip_attempts(
ip: str,
store: Any = None,
scope: str = "login",
db_path: Optional[str] = None
):
target_db = db_path or SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE
RATE_LIMITER.reset_failed_ip_attempts(
ip=ip, store=store, scope=scope, db_path=target_db, fallback_db=DEFAULT_DB_FILE
)
def hash_password(password: str, salt: Optional[str] = None) -> Tuple[str, str]:
return PasswordManager.hash_password(password, salt=salt)
def verify_password(password: str, salt: str, password_hash: str) -> bool:
return PasswordManager.verify_password(password, salt, password_hash)
def generate_session_token(username: str, db_path: Optional[str] = None) -> str:
target_db = db_path or SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE
return SessionManager.generate_session_token(username, db_path=target_db, fallback_db=DEFAULT_DB_FILE)
def verify_session_token(token: Optional[str], max_age_hours: int = 24, db_path: Optional[str] = None) -> Optional[str]:
target_db = db_path or SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE
return SessionManager.verify_session_token(token, max_age_hours=max_age_hours, db_path=target_db, fallback_db=DEFAULT_DB_FILE)
def get_current_user_optional(request: Request) -> Optional[str]:
"""Extracts authenticated username from cookie or Bearer header if valid."""
db_path = SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE
cookie = request.cookies.get(SESSION_COOKIE_NAME)
if cookie:
user = verify_session_token(cookie, db_path=db_path)
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, db_path=db_path)
if user:
return user
return None
def require_admin_auth(request: Request) -> str:
"""Enforces administrator authentication for management endpoints."""
# 1. Check if logged in admin user via session cookie or Bearer session token
user = get_current_user_optional(request)
if user:
return user
# 2. Check Bearer token or token query param matching config auth_token
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")
if 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"
raise HTTPException(status_code=401, detail="Authentication required. Please login or provide admin token.")
def verify_user_credentials(username: str, password: str, db_path: str = DEFAULT_DB_FILE) -> bool:
return PasswordManager.verify_user_credentials(username, password, db_path=db_path)
def change_user_password(username: str, new_password: str, db_path: str = DEFAULT_DB_FILE) -> bool:
return PasswordManager.change_user_password(username, new_password, db_path=db_path)
def generate_totp_secret(length: int = 20) -> str:
return TOTPManager.generate_totp_secret(length=length)
def get_totp_token(secret: str, intervals_no: Optional[int] = None) -> str:
return TOTPManager.get_totp_token(secret, intervals_no=intervals_no)
def verify_totp_code(secret: str, code: str, window: int = 1) -> bool:
return TOTPManager.verify_totp_code(secret, code, window=window)
def get_totp_uri(username: str, secret: str, issuer: str = "LOGAR Server Hub") -> str:
return TOTPManager.get_totp_uri(username, secret, issuer=issuer)
def get_user_totp_info(username: str, db_path: str = DEFAULT_DB_FILE) -> Dict[str, Any]:
return TOTPManager.get_user_totp_info(username, db_path=db_path)
def enable_user_totp(username: str, secret: str, db_path: str = DEFAULT_DB_FILE) -> bool:
return TOTPManager.enable_user_totp(username, secret, db_path=db_path)
def disable_user_totp(username: str, db_path: str = DEFAULT_DB_FILE) -> bool:
return TOTPManager.disable_user_totp(username, db_path=db_path)
def create_hermes_token(name: str, token: Optional[str] = None, db_path: str = DEFAULT_DB_FILE) -> Dict[str, Any]:
return HermesTokenManager.create_hermes_token(name, token=token, db_path=db_path)
def list_hermes_tokens(db_path: str = DEFAULT_DB_FILE) -> List[Dict[str, Any]]:
return HermesTokenManager.list_hermes_tokens(db_path=db_path)
def revoke_hermes_token(token: str, db_path: str = DEFAULT_DB_FILE) -> bool:
return HermesTokenManager.revoke_hermes_token(token, db_path=db_path)
def delete_hermes_token(token: str, db_path: str = DEFAULT_DB_FILE) -> bool:
return HermesTokenManager.delete_hermes_token(token, db_path=db_path)
def verify_hermes_token(token: str, db_path: str = DEFAULT_DB_FILE) -> bool:
return HermesTokenManager.verify_hermes_token(token, db_path=db_path)
def require_hermes_or_admin_auth(request: Request) -> str:
"""Authorizes Hermes API access via Bearer token, query parameter, admin session, or legacy auth token."""
user = get_current_user_optional(request)
if user:
return f"user:{user}"
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:
if verify_hermes_token(token, db_path=db_path):
return f"hermes_token:{token[:8]}..."
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"
# Fallback if no hermes tokens are defined in database yet
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")
# =====================================================================
# Keypair, Storage & Path Resolution Utilities
# =====================================================================
def generate_server_keypair(server_name: str) -> Tuple[str, str, str]:
return StorageCryptoManager.generate_server_keypair(server_name)
def is_path_writable(filepath: str) -> bool:
"""Checks if a file or directory path is writable by the current process."""
try:
if os.path.exists(filepath):
if os.path.isdir(filepath):
test_f = os.path.join(filepath, f".perm_{os.getpid()}")
with open(test_f, "w") as f:
f.write("1")
os.remove(test_f)
return True
else:
with open(filepath, "a") as f:
pass
return True
else:
parent = os.path.dirname(os.path.abspath(filepath)) or "."
os.makedirs(parent, exist_ok=True)
test_f = os.path.join(parent, f".perm_{os.getpid()}")
with open(test_f, "w") as f:
f.write("1")
os.remove(test_f)
return True
except Exception:
return False
def get_default_data_dir(app_name: str = "LOGAR-Server") -> str:
"""Returns a suitable writable application data directory."""
if sys.platform == "win32":
prog_data = os.environ.get("ProgramData", r"C:\ProgramData")
target_prog = os.path.join(prog_data, app_name)
if is_path_writable(target_prog):
return target_prog
local_app = os.environ.get("LOCALAPPDATA") or os.path.expanduser(r"~\AppData\Local")
target_local = os.path.join(local_app, app_name)
if is_path_writable(target_local):
return target_local
temp_dir = os.path.join(os.environ.get("TEMP", r"C:\Windows\Temp"), app_name)
if is_path_writable(temp_dir):
return temp_dir
cwd = os.getcwd()
if is_path_writable(cwd):
return cwd
return os.path.expanduser("~")
def resolve_writable_path(path: str, app_name: str = "LOGAR-Server", default_filename: str = "logar_state.db") -> str:
"""Resolves a path, ensuring it is writable or falling back to a writable data directory."""
if not path:
path = default_filename
if os.path.isabs(path) and is_path_writable(path):
return path
if not os.path.isabs(path) and "Program Files" not in os.getcwd() and is_path_writable(os.getcwd()):
return os.path.abspath(path)
data_dir = get_default_data_dir(app_name)
fname = os.path.basename(path)
return os.path.join(data_dir, fname)
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."""
actual_config_path = config_path
if not os.path.isabs(actual_config_path) and not os.path.exists(actual_config_path):
data_dir_cfg = os.path.join(get_default_data_dir("LOGAR-Server"), os.path.basename(config_path))
if os.path.exists(data_dir_cfg):
actual_config_path = data_dir_cfg
if os.path.exists(actual_config_path):
print(f"[*] Loading server configuration from: {os.path.abspath(actual_config_path)}")
with open(actual_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"] = 0
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
if "session_secret" not in config:
config["session_secret"] = secrets.token_hex(32)
if "db_path" not in config:
config["db_path"] = DEFAULT_DB_FILE
if "server_name" not in config:
config["server_name"] = "LOGAR Enterprise Hub"
if "tcp_host" not in config:
config["tcp_host"] = "0.0.0.0"
if "tcp_port" not in config:
config["tcp_port"] = 9443
if "hermes_host" not in config:
config["hermes_host"] = "0.0.0.0"
if "hermes_port" not in config:
config["hermes_port"] = 8443
if "evaluation_window_hours" not in config:
config["evaluation_window_hours"] = 12
if "min_persistence_runs" not in config:
config["min_persistence_runs"] = 2
if "storage_engine" not in config:
config["storage_engine"] = config.get("db_type", "sqlite")
if "cache_mode" not in config:
config["cache_mode"] = "direct"
if "encrypt_at_rest" not in config:
config["encrypt_at_rest"] = False
config["db_path"] = resolve_writable_path(config["db_path"], "LOGAR-Server", DEFAULT_DB_FILE)
config["cert_dir"] = resolve_writable_path(config["cert_dir"], "LOGAR-Server", "certs")
config["config_file_path"] = os.path.abspath(actual_config_path)
global SESSION_SECRET
if not os.environ.get("LOGAR_SESSION_SECRET") and "session_secret" in config:
SESSION_SECRET = config["session_secret"]
return config
if not os.path.isabs(config_path) and ("Program Files" in os.getcwd() or not is_path_writable(os.getcwd())):
config_path = os.path.join(get_default_data_dir("LOGAR-Server"), os.path.basename(config_path))
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)
session_sec = secrets.token_hex(32)
db_path = resolve_writable_path(DEFAULT_DB_FILE, "LOGAR-Server", DEFAULT_DB_FILE)
cert_dir = resolve_writable_path("certs", "LOGAR-Server", "certs")
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,
"session_secret": session_sec,
"max_seats": 0,
"license_status": "unlicensed",
"license_server_url": "https://logar.eibl.tech",
"cert_dir": cert_dir,
"tls_enabled": True,
"db_path": db_path,
"evaluation_window_hours": EVALUATION_WINDOW_HOURS,
"min_persistence_runs": RUN_THRESHOLD,
"server_fingerprint": fingerprint,
"public_key": public_key,
"private_key": private_key,
"config_file_path": os.path.abspath(config_path)
}
if not os.environ.get("LOGAR_SESSION_SECRET"):
SESSION_SECRET = session_sec
os.makedirs(os.path.dirname(os.path.abspath(config_path)), exist_ok=True)
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
print("[+] 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 rotate_enrollment_secret(config_dict: Optional[Dict[str, Any]] = None) -> str:
"""Generates a fresh enrollment secret, invalidates the previous secret, and persists change to disk & DB."""
cfg = config_dict if config_dict is not None else SERVER_STATE.get("config", {})
new_secret = secrets.token_hex(24)
cfg["enrollment_secret"] = new_secret
config_file_path = cfg.get("config_file_path") or CONFIG_FILE_NAME
try:
if os.path.exists(config_file_path):
with open(config_file_path, "r", encoding="utf-8") as f:
saved = json.load(f)
saved["enrollment_secret"] = new_secret
with open(config_file_path, "w", encoding="utf-8") as f:
json.dump(saved, f, indent=2)
except Exception as e:
print(f"[!] Warning: Could not write new enrollment secret to config file: {e}")
db_path = cfg.get("db_path", DEFAULT_DB_FILE)
try:
conn = sqlite3.connect(db_path, timeout=5.0)
c = conn.cursor()
c.execute("UPDATE license_config SET enrollment_secret = ? WHERE id = 1", (new_secret,))
conn.commit()
conn.close()
except Exception as e:
print(f"[!] Warning: Could not update database with new enrollment secret: {e}")
SERVER_STATE["config"] = cfg
return new_secret
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": "3.0.0",
"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": server_host if cfg.get("hermes_host") in ("0.0.0.0", "::", None) else 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 key."""
server_conf = load_or_init_config(config_path)
effective_hermes_host = hermes_host
if not effective_hermes_host:
cfg_hermes = server_conf.get("hermes_host")
effective_hermes_host = server_host if cfg_hermes in ("0.0.0.0", "::", None) else cfg_hermes
client_conf = {
"server_host": server_host,
"server_port": server_port,
"hermes_host": effective_hermes_host,
"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
# =====================================================================
# Database, Storage & Crypto Integrations
# =====================================================================
def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: int = 0) -> str:
return StorageEngine.init_db(db_path, enrollment_secret=enrollment_secret, max_seats=max_seats)
def init_postgres_db(pg_config: Dict[str, Any]) -> Tuple[bool, str]:
return StorageEngine.init_postgres_db(pg_config)
def migrate_sqlite_to_postgres(sqlite_path: str, pg_config: Dict[str, Any]) -> Tuple[bool, str, Dict[str, int]]:
return StorageEngine.migrate_sqlite_to_postgres(sqlite_path, pg_config, encrypt_fn=encrypt_field_for_storage)
def export_license_request_lic(config: Dict[str, Any]) -> Tuple[Dict[str, Any], str]:
return LicenseManager.export_license_request_lic(config)
def get_active_licensed_features(db_path: Optional[str] = None) -> List[str]:
return LicenseManager.get_active_licensed_features(db_path=db_path, server_state=SERVER_STATE)
def is_feature_licensed(feature: str, db_path: Optional[str] = None) -> bool:
return LicenseManager.is_feature_licensed(feature, db_path=db_path, server_state=SERVER_STATE)
def get_server_public_key() -> Optional[pgpy.PGPKey]:
return StorageCryptoManager.get_server_public_key(server_state=SERVER_STATE)
def encrypt_field_for_storage(text: str, force: bool = False) -> str:
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
is_unlocked = is_feature_licensed("enterprise_db", db_path)
return StorageCryptoManager.encrypt_field_for_storage(text, force=force, server_state=SERVER_STATE, is_unlocked=is_unlocked)
def decrypt_field_from_storage(text: str) -> str:
return StorageCryptoManager.decrypt_field_from_storage(text, server_state=SERVER_STATE)
def get_storage_engine_info(db_path: Optional[str] = None) -> Dict[str, Any]:
target_db = db_path or SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
is_unlocked = is_feature_licensed("enterprise_db", db_path=target_db)
return StorageEngine.get_storage_engine_info(
db_path=target_db, server_state=SERVER_STATE, ingest_queue=INGEST_QUEUE, is_unlocked=is_unlocked
)
def get_license_summary(db_path: Optional[str] = None) -> Dict[str, Any]:
return LicenseManager.get_license_summary(server_state=SERVER_STATE, db_path=db_path)
def report_seat_quota_to_authority_sync():
LicenseManager.report_seat_quota_to_authority_sync(server_state=SERVER_STATE)
def trigger_async_seat_report():
LicenseManager.trigger_async_seat_report(server_state=SERVER_STATE)
# =====================================================================
# PKI & mTLS Management
# =====================================================================
def init_mtls_server_context(cert_dir: str = "certs") -> ssl.SSLContext:
return PKIManager.init_mtls_server_context(cert_dir=cert_dir)
def reload_mtls_context(ssl_ctx: ssl.SSLContext, cert_dir: str = "certs"):
PKIManager.reload_mtls_context(ssl_ctx, cert_dir=cert_dir)
def check_and_rotate_server_certs(
cert_dir: str = "certs",
hostnames: Optional[List[str]] = None,
threshold_days: int = 30
) -> bool:
return PKIManager.check_and_rotate_server_certs(
cert_dir=cert_dir, hostnames=hostnames, threshold_days=threshold_days, server_state=SERVER_STATE
)
# =====================================================================
# Log Ingestion & Processing
# =====================================================================
async def ingest_buffer_worker_loop():
"""Periodically flushes the in-memory ingestion buffer to persistent database storage."""
while True:
try:
await asyncio.sleep(INGEST_QUEUE.flush_interval)
config = SERVER_STATE.get("config", {})
db_path = config.get("db_path", DEFAULT_DB_FILE)
window_hours = config.get("evaluation_window_hours", 12)
min_runs = config.get("min_persistence_runs", 1)
INGEST_QUEUE.flush(
db_path=db_path, window_hours=window_hours, min_runs=min_runs, process_callback=process_ingested_logs
)
except asyncio.CancelledError:
break
except Exception as e:
print(f"[!] Ingestion buffer worker error: {e}")
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.
Supports SQLite and external PostgreSQL with optional OpenPGP data encryption at rest.
"""
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()
config = SERVER_STATE.get("config", {})
db_type = config.get("db_type", config.get("storage_engine", "sqlite")).lower()
if db_type in ("postgres", "postgresql"):
try:
try:
import psycopg2
except ImportError:
import psycopg as psycopg2
pg_cfg = config.get("postgres", {})
conn = psycopg2.connect(
host=pg_cfg.get("host", "localhost"),
port=int(pg_cfg.get("port", 5432)),
dbname=pg_cfg.get("database", "logar"),
user=pg_cfg.get("user", "postgres"),
password=pg_cfg.get("password", ""),
sslmode=pg_cfg.get("sslmode", "prefer"),
connect_timeout=int(pg_cfg.get("connect_timeout", 5))
)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO ingest_runs (run_id, site_name, server, timestamp, log_count) VALUES (%s, %s, %s, %s, %s)",
(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()
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")
# Suppress repetitive benign routine service start/stop cycles
if is_routine_service_noise(signature, message, severity):
continue
fp = f"{site_name}:{server}:{signature}"
stored_message = encrypt_field_for_storage(message)
cursor.execute(
"SELECT run_count, first_seen, last_seen, status, last_run_id FROM active_issues WHERE fingerprint = %s",
(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
if (now - last_seen_dt) > timedelta(hours=window_hours):
new_runs = 1
new_first_seen = now_iso
else:
new_runs = run_count + 1 if last_run_id != run_id else 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 = %s, last_seen = %s, first_seen = %s, status = %s, last_run_id = %s, message = %s, severity = %s
WHERE fingerprint = %s
""", (new_runs, now_iso, new_first_seen, new_status, run_id, stored_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 (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""", (fp, site_name, server, signature, severity, stored_message, os_type, now_iso, now_iso, 1, initial_status, run_id))
processed_count += 1
# Update client node metrics if attached
metrics = payload.get("metrics")
if metrics and isinstance(metrics, dict) and client_server:
cpu_p = float(metrics.get("cpu_percent", 0.0))
mem_p = float(metrics.get("memory_percent", 0.0))
disk_p = float(metrics.get("disk_percent_free", 0.0))
try:
cursor.execute("""
UPDATE clients
SET last_seen = %s, metrics_json = %s, cpu_percent = %s, memory_percent = %s, disk_percent_free = %s
WHERE client_id = %s OR hostname = %s
""", (now_iso, json.dumps(metrics), cpu_p, mem_p, disk_p, client_server, client_server))
except Exception:
pass
conn.commit()
conn.close()
return {
"status": "success",
"run_id": run_id,
"processed": processed_count,
"promoted_verified": promoted_to_verified
}
except Exception as pg_err:
print(f"[!] Warning: PostgreSQL ingestion failed ({pg_err}), falling back to SQLite...")
# SQLite standard / fallback path
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
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()
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")
# Suppress repetitive benign routine service start/stop cycles
if is_routine_service_noise(signature, message, severity):
continue
fp = f"{site_name}:{server}:{signature}"
stored_message = encrypt_field_for_storage(message)
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
if (now - last_seen_dt) > timedelta(hours=window_hours):
new_runs = 1
new_first_seen = now_iso
else:
new_runs = run_count + 1 if last_run_id != run_id else 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, stored_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, stored_message, os_type, now_iso, now_iso, 1, initial_status, run_id))
processed_count += 1
# Update client node metrics & version if attached
metrics = payload.get("metrics")
client_ver = payload.get("version")
if metrics and isinstance(metrics, dict) and client_server:
cpu_p = float(metrics.get("cpu_percent", 0.0))
mem_p = float(metrics.get("memory_percent", 0.0))
disk_p = float(metrics.get("disk_percent_free", 0.0))
try:
if client_ver:
cursor.execute("""
UPDATE clients
SET last_seen = ?, metrics_json = ?, cpu_percent = ?, memory_percent = ?, disk_percent_free = ?, client_version = ?
WHERE client_id = ? OR hostname = ?
""", (now_iso, json.dumps(metrics), cpu_p, mem_p, disk_p, client_ver, client_server, client_server))
else:
cursor.execute("""
UPDATE clients
SET last_seen = ?, metrics_json = ?, cpu_percent = ?, memory_percent = ?, disk_percent_free = ?
WHERE client_id = ? OR hostname = ?
""", (now_iso, json.dumps(metrics), cpu_p, mem_p, disk_p, client_server, client_server))
except Exception:
pass
elif client_ver and client_server:
try:
cursor.execute("""
UPDATE clients SET last_seen = ?, client_version = ? WHERE client_id = ? OR hostname = ?
""", (now_iso, client_ver, client_server, client_server))
except Exception:
pass
conn.commit()
conn.close()
return {
"status": "success",
"run_id": run_id,
"processed": processed_count,
"promoted_verified": promoted_to_verified
}
def get_local_windows_logs(hours: int = 24, state: Optional[dict] = None) -> list:
collector = WindowsLogCollector(state=state, channels=["Application", "System"])
return collector.collect_logs(hours=hours)
def get_local_linux_logs(hours: int = 24, state: Optional[dict] = None) -> list:
collector = LinuxLogCollector(state=state)
return collector.collect_logs(hours=hours)
def get_local_system_logs(hours: int = 24, state: Optional[dict] = None) -> list:
collector = get_log_collector(state=state)
return collector.collect_logs(hours=hours)
def run_local_log_collection_cycle(db_path: str, window_hours: int = 12, min_runs: int = 1, state: Optional[dict] = None) -> int:
"""Executes one cycle of local host log collection and direct ingestion into SQLite."""
machine_id = get_machine_identifier()
os_type = "windows" if sys.platform == "win32" else "linux"
now_sql = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
try:
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute("SELECT client_id FROM clients WHERE client_id = ?", (machine_id,))
if c.fetchone():
c.execute("""
UPDATE clients
SET hostname = ?, os_type = ?, cert_fingerprint = 'LOCAL-HOST-BUILTIN', status = 'active', last_seen = ?
WHERE client_id = ?
""", (machine_id, os_type, now_sql, machine_id))
else:
c.execute("""
INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, first_seen, last_seen)
VALUES (?, ?, ?, 'LOCAL-HOST-BUILTIN', 'active', ?, ?)
""", (machine_id, machine_id, os_type, now_sql, now_sql))
conn.commit()
conn.close()
except Exception as e:
print(f"[!] Warning: Failed to update local hub client record: {e}")
logs = get_local_system_logs(hours=window_hours, state=state)
if not logs:
return 0
payload = {
"server": machine_id,
"site_name": machine_id.split(".", 1)[1] if "." in machine_id else "local-hub",
"timestamp": datetime.now(timezone.utc).isoformat(),
"logs": logs
}
res = process_ingested_logs(payload, db_path=db_path, window_hours=window_hours, min_runs=min_runs)
print(f"[+] Integrated Local Collector: Processed {len(logs)} local host events ({res.get('promoted_verified', 0)} verified anomalies).")
return len(logs)
async def local_collector_loop(interval_seconds: int = 30):
"""Background worker that periodically ingests local host logs without requiring a client license."""
config = SERVER_STATE.get("config", {})
db_path = config.get("db_path", DEFAULT_DB_FILE)
window_hours = config.get("evaluation_window_hours", 12)
min_runs = config.get("min_persistence_runs", 1)
state_file = resolve_writable_path("local_collector_state.json", "LOGAR-Server", "local_collector_state.json")
state = {}
if os.path.exists(state_file):
try:
with open(state_file, "r", encoding="utf-8") as f:
state = json.load(f)
except Exception:
state = {}
try:
run_local_log_collection_cycle(db_path, window_hours=window_hours, min_runs=min_runs, state=state)
with open(state_file, "w", encoding="utf-8") as f:
json.dump(state, f)
except Exception as e:
print(f"[!] Error in initial local log collection cycle: {e}")
while True:
try:
await asyncio.sleep(interval_seconds)
run_local_log_collection_cycle(db_path, window_hours=window_hours, min_runs=min_runs, state=state)
with open(state_file, "w", encoding="utf-8") as f:
json.dump(state, f)
except asyncio.CancelledError:
break
except Exception as e:
print(f"[!] Exception in local_collector_loop: {e}")
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 and reports seat quotas."""
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)
report_seat_quota_to_authority_sync()
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 with certificate validation, decryption, and buffering."""
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 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
cache_mode = SERVER_STATE.get("config", {}).get("cache_mode", "direct")
if cache_mode == "buffered":
INGEST_QUEUE.push(log_payload)
if len(INGEST_QUEUE._buffer) >= INGEST_QUEUE.batch_size:
res = INGEST_QUEUE.flush(
db_path=SERVER_STATE["config"]["db_path"],
window_hours=SERVER_STATE["config"]["evaluation_window_hours"],
min_runs=SERVER_STATE["config"]["min_persistence_runs"],
process_callback=process_ingested_logs
)
else:
res = {
"status": "success",
"buffered": True,
"processed": len(log_payload.get("logs", [])),
"queue_size": len(INGEST_QUEUE._buffer)
}
else:
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"]
)
with INGEST_QUEUE._lock:
INGEST_QUEUE._total_ingested += len(log_payload.get("logs", []))
INGEST_QUEUE._total_flushed += len(log_payload.get("logs", []))
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
# =====================================================================
# FastAPI Route Handlers
# =====================================================================
@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=WebUIRenderer.render_login_html())
@app.post("/login")
async def process_login(request: Request):
"""Processes login credentials and issues session cookie with 2FA TOTP enforcement."""
client_ip = get_client_ip(request)
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
if not check_ip_rate_limit(client_ip, max_attempts=10, window_seconds=60, scope="login", db_path=db_path):
raise HTTPException(status_code=429, detail="Too many failed login attempts. Please try again later.")
username, password, totp_code = None, 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")
totp_code = body.get("totp_code")
except Exception:
pass
else:
form = await request.form()
username = form.get("username")
password = form.get("password")
totp_code = form.get("totp_code")
if not username or not password:
raise HTTPException(status_code=400, detail="Username and password are required")
if not verify_user_credentials(username, password, db_path=db_path):
record_failed_ip_attempt(client_ip, scope="login", db_path=db_path)
await asyncio.sleep(0.5)
raise HTTPException(status_code=401, detail="Invalid username or password")
# Check 2FA TOTP requirement
totp_info = get_user_totp_info(username, db_path=db_path)
if totp_info.get("enabled"):
if not totp_code:
return JSONResponse(status_code=200, content={
"status": "2fa_required",
"requires_2fa": True,
"message": "Two-factor authentication code required"
})
if not verify_totp_code(totp_info["secret"], totp_code):
record_failed_ip_attempt(client_ip, scope="login", db_path=db_path)
await asyncio.sleep(0.3)
raise HTTPException(status_code=401, detail="Invalid 6-digit 2FA verification code")
reset_failed_ip_attempts(client_ip, scope="login", db_path=db_path)
token = generate_session_token(username, db_path=db_path)
is_https = request.url.scheme == "https" or request.headers.get("x-forwarded-proto") == "https"
response = JSONResponse(content={"status": "success", "username": username, "token": token})
response.set_cookie(
key=SESSION_COOKIE_NAME,
value=token,
httponly=True,
secure=is_https,
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.get("/api/user/2fa/status")
def api_user_2fa_status(auth_user: str = Depends(require_admin_auth)):
"""Returns 2FA status for currently logged in administrator."""
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
info = get_user_totp_info(auth_user, db_path=db_path)
return {"enabled": info["enabled"], "username": auth_user}
@app.get("/api/user/2fa/setup")
def api_user_2fa_setup(auth_user: str = Depends(require_admin_auth)):
"""Generates a new TOTP secret and otpauth URI for 2FA provisioning."""
secret = generate_totp_secret()
otpauth = get_totp_uri(auth_user, secret, issuer="LOGAR Server Hub")
return {"secret": secret, "otpauth_uri": otpauth, "username": auth_user}
@app.post("/api/user/2fa/enable")
async def api_user_2fa_enable(request: Request, auth_user: str = Depends(require_admin_auth)):
"""Enables TOTP for the current user after validating 6-digit code."""
body = await request.json()
secret = body.get("secret", "").strip()
code = body.get("code", "").strip()
if not secret or not code:
raise HTTPException(status_code=400, detail="Secret and 6-digit verification code required")
if not verify_totp_code(secret, code):
raise HTTPException(status_code=400, detail="Invalid 6-digit TOTP verification code. Please check your device clock.")
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
enable_user_totp(auth_user, secret, db_path=db_path)
return {"status": "success", "message": "2FA successfully activated"}
@app.post("/api/user/2fa/disable")
async def api_user_2fa_disable(request: Request, auth_user: str = Depends(require_admin_auth)):
"""Disables TOTP for the current user after verifying password."""
body = await request.json()
password = body.get("password", "")
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
if not verify_user_credentials(auth_user, password, db_path=db_path):
raise HTTPException(status_code=403, detail="Current password incorrect")
disable_user_totp(auth_user, db_path=db_path)
return {"status": "success", "message": "2FA successfully disabled"}
@app.post("/api/client/enroll")
def enroll_client(req: ClientEnrollRequest, request: Request = None):
"""Enrolls an edge client by validating the enrollment secret and issuing a signed certificate."""
client_ip = get_client_ip(request)
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
if not check_ip_rate_limit(client_ip, max_attempts=15, window_seconds=60, scope="enroll", db_path=db_path):
raise HTTPException(status_code=429, detail="Too many failed enrollment attempts. Please try again later.")
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute("SELECT enrollment_secret, max_seats, license_status 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 = row[0]
lic_summary = LicenseManager.get_license_summary(SERVER_STATE, db_path=db_path)
max_seats = lic_summary.get("max_seats", 0)
license_status = lic_summary.get("status", "UNLICENSED").lower()
if max_seats <= 0 or license_status in ("unlicensed", "revoked", "expired"):
conn.close()
raise HTTPException(status_code=403, detail="Server is unlicensed. Import an activated .lic file via the WebUI before enrolling edge clients.")
if not secrets.compare_digest(str(req.enrollment_secret), str(expected_secret)):
conn.close()
record_failed_ip_attempt(client_ip, scope="enroll", db_path=db_path)
time.sleep(0.5)
raise HTTPException(status_code=403, detail="Invalid enrollment secret")
reset_failed_ip_attempts(client_ip, scope="enroll", db_path=db_path)
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")
cert_dir = SERVER_STATE.get("config", {}).get("cert_dir", "certs")
upstream_ca_path = os.path.join(cert_dir, "upstream_ca.crt")
if os.path.exists(upstream_ca_path):
try:
with open(upstream_ca_path, "r", encoding="utf-8") as f:
upstream_pem = f.read().strip()
if upstream_pem and upstream_pem not in ca_cert_pem:
ca_cert_pem = ca_cert_pem.strip() + "\n" + upstream_pem
except Exception:
pass
c.execute("SELECT status FROM clients WHERE client_id = ?", (req.client_id,))
existing = c.fetchone()
if existing:
if existing[0] == "revoked":
conn.close()
raise HTTPException(status_code=403, detail="Client seat revoked by administrator.")
else:
c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active' AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'")
active_count = c.fetchone()[0]
if active_count >= max_seats:
conn.close()
raise HTTPException(status_code=403, detail=f"License seat limit reached ({active_count}/{max_seats} seats used).")
cert_pem, key_pem = enrollment.issue_client_cert(req.client_id, ca_cert, ca_key)
fp = enrollment.calculate_cert_fingerprint(cert_pem)
now_sql = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
client_ver = req.version or ""
if existing:
try:
c.execute("""
UPDATE clients
SET hostname = ?, os_type = ?, cert_fingerprint = ?, status = 'active', last_seen = ?, client_version = ?
WHERE client_id = ?
""", (req.hostname, req.os, fp, now_sql, client_ver, req.client_id))
except sqlite3.OperationalError:
c.execute("""
UPDATE clients
SET hostname = ?, os_type = ?, cert_fingerprint = ?, status = 'active', last_seen = ?
WHERE client_id = ?
""", (req.hostname, req.os, fp, now_sql, req.client_id))
else:
try:
c.execute("""
INSERT INTO clients (client_id, hostname, os_type, cert_fingerprint, status, first_seen, last_seen, client_version)
VALUES (?, ?, ?, ?, 'active', ?, ?, ?)
""", (req.client_id, req.hostname, req.os, fp, now_sql, now_sql, client_ver))
except sqlite3.OperationalError:
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()
trigger_async_seat_report()
return {
"status": "success",
"client_id": req.client_id,
"client_cert": cert_pem,
"client_key": key_pem,
"ca_cert": ca_cert_pem,
"server_fingerprint": SERVER_STATE.get("config", {}).get("server_fingerprint", ""),
"server_public_key": SERVER_STATE.get("config", {}).get("public_key", "")
}
@app.get("/api/client/export-profile")
def export_client_profile(
server_host: Optional[str] = None,
server_port: Optional[int] = None,
client_id: Optional[str] = None,
request: Request = None
):
"""Exports standalone client profile JSON for instant client deployment."""
rotate_enrollment_secret()
cfg = SERVER_STATE.get("config", {})
host = server_host
if not host:
host = (request.headers.get("host", "").split(":")[0] if request and request.headers else "") or cfg.get("tcp_host", "127.0.0.1")
if host in ("0.0.0.0", "localhost", ""):
host = socket.gethostname()
port = server_port or cfg.get("tcp_port", 9443)
profile = generate_client_profile_dict(server_host=host, server_port=port, client_id=client_id, config_dict=cfg)
return JSONResponse(
content=profile,
headers={"Content-Disposition": f'attachment; filename="logar_client_{host}_{port}.json"'}
)
@app.post("/api/server/reset-enrollment-secret")
def reset_server_enrollment_secret(auth_user: str = Depends(require_admin_auth)):
"""Rotates and generates a fresh server enrollment secret on demand."""
new_secret = rotate_enrollment_secret()
return {
"status": "success",
"message": "Server enrollment secret successfully rotated and updated.",
"enrollment_secret": new_secret
}
# =====================================================================
# Update & Distribution Endpoints
# =====================================================================
@app.get("/api/update/status")
@app.get("/api/server/update/status")
def get_update_status(user: str = Depends(get_current_user_optional)):
"""Returns current server update and client package distribution repository status."""
mgr: ServerUpdateManager = SERVER_STATE.get("update_manager") or UPDATE_MANAGER
return mgr.get_status()
@app.post("/api/update/check")
@app.post("/api/server/update/check")
def check_server_updates(user: str = Depends(get_current_user_optional)):
"""Triggers an immediate upstream release check."""
mgr: ServerUpdateManager = SERVER_STATE.get("update_manager") or UPDATE_MANAGER
return mgr.check_upstream_updates(force=True)
@app.get("/api/client/update/check")
def check_client_update(
client_id: str,
os: Optional[str] = "windows",
os_name: Optional[str] = None,
current_version: Optional[str] = None,
arch: str = "x86_64"
):
"""
Invoked by edge forwarders to check whether a newer client binary package is available.
"""
import os as _os
target_os = os_name or os or "windows"
mgr: ServerUpdateManager = SERVER_STATE.get("update_manager") or UPDATE_MANAGER
# Record check timestamp in database
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
if target_os:
now_sql = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
try:
conn = sqlite3.connect(db_path)
conn.execute("""
UPDATE clients SET last_update_check = ?, client_version = COALESCE(NULLIF(?, ''), client_version)
WHERE client_id = ?
""", (now_sql, current_version or "", client_id))
conn.commit()
conn.close()
except Exception:
pass
return mgr.check_client_update(client_id=client_id, os_type=target_os, current_version=current_version, arch=arch)
@app.get("/api/client/update/download")
def download_client_update_package(
request: Request,
os: Optional[str] = "windows",
os_name: Optional[str] = None
):
"""
Streams the latest client binary or installer package for the specified OS.
Supports HTTP ETag and conditional 304 Not Modified requests to minimize network bandwidth.
"""
import os as _os
import email.utils
target_os = os_name or os or "windows"
mgr: ServerUpdateManager = SERVER_STATE.get("update_manager") or UPDATE_MANAGER
pkg_tuple = mgr.get_client_package_path(os_type=target_os)
if not pkg_tuple:
raise HTTPException(status_code=404, detail=f"No client update package available for OS: {target_os}")
file_path, filename, sha256 = pkg_tuple
if not _os.path.exists(file_path):
raise HTTPException(status_code=404, detail="Client package file missing on server disk.")
etag = f'"{sha256}"'
if request is not None:
if_none_match = request.headers.get("if-none-match")
if if_none_match and (if_none_match == etag or if_none_match.strip('"') == sha256):
return Response(
status_code=304,
headers={
"ETag": etag,
"Cache-Control": "public, max-age=300, must-revalidate",
"X-Checksum-SHA256": sha256
}
)
try:
mtime = _os.path.getmtime(file_path)
last_modified = email.utils.formatdate(mtime, usegmt=True)
except Exception:
last_modified = None
if_modified_since = request.headers.get("if-modified-since")
if if_modified_since and last_modified and if_modified_since == last_modified:
return Response(
status_code=304,
headers={
"ETag": etag,
"Cache-Control": "public, max-age=300, must-revalidate",
"X-Checksum-SHA256": sha256
}
)
headers = {
"ETag": etag,
"Cache-Control": "public, max-age=300, must-revalidate",
"X-Checksum-SHA256": sha256,
"Content-Disposition": f'attachment; filename="{filename}"'
}
if last_modified:
headers["Last-Modified"] = last_modified
return FileResponse(
path=file_path,
filename=filename,
media_type="application/octet-stream",
headers=headers
)
@app.post("/api/server/client-package/upload")
async def upload_client_package(
request: Request,
auth_user: str = Depends(require_admin_auth)
):
"""
Uploads a new client executable/installer into the server's update distribution repository.
Accepts JSON payload: { "os": "windows"|"linux", "version": "...", "filename": "...", "data_base64": "..." }
or raw stream with headers 'X-Package-OS', 'X-Package-Filename', 'X-Package-Version'.
"""
import base64
import os as _os
mgr: ServerUpdateManager = SERVER_STATE.get("update_manager") or UPDATE_MANAGER
content_type = request.headers.get("content-type", "")
os_type = "windows"
version = None
filename = "client_update.bin"
file_bytes = b""
if "application/json" in content_type:
body = await request.json()
os_type = body.get("os", "windows")
version = body.get("version")
raw_name = body.get("filename") or ("LOGAR_Client_Setup.exe" if os_type == "windows" else "Linux_Client.bin")
filename = _os.path.basename(str(raw_name))
b64_data = body.get("data_base64", "")
file_bytes = base64.b64decode(b64_data)
else:
os_type = request.headers.get("X-Package-OS", "windows")
version = request.headers.get("X-Package-Version")
raw_name = request.headers.get("X-Package-Filename") or ("LOGAR_Client_Setup.exe" if os_type == "windows" else "Linux_Client.bin")
filename = _os.path.basename(str(raw_name))
file_bytes = await request.body()
if not file_bytes:
raise HTTPException(status_code=400, detail="Empty package payload.")
temp_dir = _os.path.join(mgr.update_dir, "temp")
_os.makedirs(temp_dir, exist_ok=True)
temp_path = _os.path.join(temp_dir, filename)
with open(temp_path, "wb") as f:
f.write(file_bytes)
try:
manifest = mgr.register_client_package(
os_type=os_type,
file_path=temp_path,
version=version,
filename=filename
)
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except Exception:
pass
return {"status": "success", "manifest": manifest, "sha256": manifest.get("sha256")}
except Exception as e:
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except Exception:
pass
raise HTTPException(status_code=400, detail=f"Package upload registration failed: {e}")
@app.get("/api/clients")
def list_clients(auth_user: str = Depends(require_admin_auth)):
"""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,
metrics_json, cpu_percent, memory_percent, disk_percent_free
FROM clients ORDER BY last_seen DESC
""")
rows = c.fetchall()
conn.close()
clients = []
for r in rows:
cid, host, ost, fp, st, first, last, mjson, cpu, mem, disk = r
m_dict = {}
if mjson:
try:
m_dict = json.loads(mjson)
except Exception:
pass
clients.append({
"client_id": cid,
"hostname": host,
"os_type": ost,
"cert_fingerprint": fp,
"status": st,
"first_seen": first,
"last_seen": last,
"last_activity_relative": format_relative_time(last),
"metrics": m_dict,
"cpu_percent": cpu or 0.0,
"memory_percent": mem or 0.0,
"disk_percent_free": disk or 0.0
})
active_count = sum(1 for cl in clients if cl["status"] == "active" and cl.get("cert_fingerprint") != "LOCAL-HOST-BUILTIN")
return {
"active_seats": active_count,
"max_seats": max_seats,
"clients": clients
}
@app.post("/api/client/approve")
@app.post("/api/client/reactivate")
def approve_client(action: ClientActionRequest, auth_user: str = Depends(require_admin_auth)):
"""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' AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'")
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}")
trigger_async_seat_report()
return {"status": "active", "client_id": action.client_id}
reactivate_client = approve_client
@app.post("/api/client/revoke")
def revoke_client(action: ClientActionRequest, auth_user: str = Depends(require_admin_auth)):
"""Revokes a client certificate/license seat. Revoked clients are immediately rejected during mTLS handshakes."""
if action.client_id == get_machine_identifier():
raise HTTPException(status_code=400, detail="Cannot revoke integrated local hub collector")
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 = ? AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'", (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}")
trigger_async_seat_report()
return {"status": "revoked", "client_id": action.client_id}
@app.post("/api/client/delete")
@app.post("/api/clients/delete")
@app.post("/api/clients/remove")
def delete_client(action: ClientActionRequest, auth_user: str = Depends(require_admin_auth)):
"""Permanently removes a client forwarder record from SQLite, immediately freeing its seat."""
if action.client_id == get_machine_identifier():
raise HTTPException(status_code=400, detail="Cannot remove integrated local hub collector")
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 = ? AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'", (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}")
trigger_async_seat_report()
return {"status": "deleted", "client_id": action.client_id}
@app.get("/api/hermes/tokens")
def api_list_hermes_tokens(auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
return list_hermes_tokens(db_path=db_path)
@app.post("/api/hermes/tokens/create")
def api_create_hermes_token(req: HermesTokenCreateRequest, auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
if not req.name or not req.name.strip():
raise HTTPException(status_code=400, detail="Token name is required")
return create_hermes_token(req.name.strip(), db_path=db_path)
@app.post("/api/hermes/tokens/revoke")
def api_revoke_hermes_token(req: HermesTokenRevokeRequest, auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
if not revoke_hermes_token(req.token, db_path=db_path):
raise HTTPException(status_code=404, detail="Token not found")
return {"status": "success", "message": "Token revoked"}
@app.post("/api/hermes/tokens/delete")
def api_delete_hermes_token(req: HermesTokenRevokeRequest, auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
if not delete_hermes_token(req.token, db_path=db_path):
raise HTTPException(status_code=404, detail="Token not found")
return {"status": "success", "message": "Token deleted"}
@app.get("/api/hermes/report")
def get_verified_anomalies_for_hermes(token_ctx: str = Depends(require_hermes_or_admin_auth)):
"""
Ingestion endpoint for Hermes agentic workflows.
Returns verified anomalies within the active 12-hour evaluation window,
system resource context, and silent/offline forwarder alerts for Hermes AI reasoning.
"""
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()
now = datetime.now(timezone.utc)
try:
cursor.execute("SELECT client_id, hostname, os_type, last_seen, metrics_json, cpu_percent, memory_percent, disk_percent_free FROM clients WHERE status = 'active'")
client_rows = cursor.fetchall()
except Exception:
client_rows = []
conn.close()
window_hours = SERVER_STATE.get("config", {}).get("evaluation_window_hours", 12)
report = []
client_metrics_map = {}
for cl in client_rows:
cid, chost, cost, clast, cmetrics_raw, ccpu, cmem, cdisk = cl
try:
m_dict = json.loads(cmetrics_raw) if cmetrics_raw else {}
except Exception:
m_dict = {}
if not m_dict:
m_dict = {
"cpu_percent": ccpu or 0.0,
"memory_percent": cmem or 0.0,
"disk_percent_free": cdisk or 0.0
}
client_metrics_map[cid] = m_dict
if chost:
client_metrics_map[chost] = m_dict
if clast:
try:
clast_clean = str(clast).replace("T", " ").split("+")[0].split("Z")[0].strip()
last_dt = datetime.fromisoformat(clast_clean).replace(tzinfo=timezone.utc)
if (now - last_dt) > timedelta(minutes=15):
offline_mins = int((now - last_dt).total_seconds() // 60)
report.append({
"fingerprint": f"offline-node-{cid}",
"site": "default",
"server": chost or cid,
"signature": "LOGAR/NodeSilent",
"severity": "CRITICAL",
"message": f"Forwarder node '{chost or cid}' has gone silent! Last heartbeat was {offline_mins} minutes ago ({clast}). Potential node crash, network partition, or power failure.",
"os_type": cost or "unknown",
"first_seen": clast,
"last_seen": clast,
"consecutive_runs": 1,
"evaluation_window": f"{window_hours}h",
"verified": True,
"status": "NODE_OFFLINE",
"metrics": m_dict
})
except Exception:
pass
for r in rows:
srv_name = r[2]
report.append({
"fingerprint": r[0],
"site": r[1],
"server": srv_name,
"signature": r[3],
"severity": r[4],
"message": decrypt_field_from_storage(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],
"metrics": client_metrics_map.get(srv_name, {})
})
return report
@app.get("/api/issues")
def get_all_issues(auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
conn = sqlite3.connect(db_path)
c = conn.cursor()
c.execute("SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status FROM active_issues ORDER BY last_seen DESC")
issues = [
{
"fingerprint": r[0],
"site_name": r[1],
"server": r[2],
"signature": r[3],
"severity": r[4],
"message": decrypt_field_from_storage(r[5]),
"os_type": r[6],
"first_seen": r[7],
"last_seen": r[8],
"run_count": r[9],
"status": r[10]
}
for r in c.fetchall()
]
conn.close()
return issues
@app.get("/api/license/status")
def api_license_status():
return get_license_summary()
@app.get("/api/license/export-lic")
def export_lic_file(auth_user: str = Depends(require_admin_auth)):
config = SERVER_STATE.get("config", {})
lic_dict, lic_json = export_license_request_lic(config)
hostname = socket.gethostname()
return Response(
content=lic_json,
media_type="application/json",
headers={"Content-Disposition": f'attachment; filename="logar_license_request_{hostname}.lic"'}
)
@app.get("/api/license/ca-csr")
def get_ca_csr(auth_user: str = Depends(require_admin_auth)):
cert_dir = SERVER_STATE.get("config", {}).get("cert_dir", "certs")
csr_path = os.path.join(cert_dir, "ca.csr")
if not os.path.exists(csr_path):
enrollment.generate_ca_csr(cert_dir=cert_dir)
with open(csr_path, "r", encoding="utf-8") as f:
csr_content = f.read()
return Response(
content=csr_content,
media_type="application/x-pem-file",
headers={"Content-Disposition": 'attachment; filename="ca.csr"'}
)
@app.post("/api/license/import")
def import_signed_license(req: LicenseImportRequest, auth_user: str = Depends(require_admin_auth)):
cert_dir = SERVER_STATE.get("config", {}).get("cert_dir", "certs")
ca_file = os.path.join(cert_dir, "ca.crt")
upstream_file = os.path.join(cert_dir, "upstream_ca.crt")
upstream_pem = req.upstream_ca.strip() if req.upstream_ca else None
if not upstream_pem and os.path.exists(upstream_file):
try:
with open(upstream_file, "r", encoding="utf-8") as f:
upstream_pem = f.read().strip()
except Exception:
upstream_pem = None
if upstream_pem:
if not enrollment.verify_certificate_signature(req.ca_cert, upstream_pem):
raise HTTPException(status_code=400, detail="Cryptographic verification failed: Sub-CA certificate was not signed by the upstream Master Authority.")
try:
info = enrollment.extract_license_info_from_cert(req.ca_cert, upstream_ca_pem=upstream_pem)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Failed to parse X.509 certificate: {e}")
with open(ca_file, "w", encoding="utf-8") as f:
f.write(req.ca_cert.strip() + "\n")
if req.upstream_ca:
with open(upstream_file, "w", encoding="utf-8") as f:
f.write(req.upstream_ca.strip() + "\n")
new_seats = info.get("seats")
license_id = info.get("license_id")
expires_at = info.get("not_valid_after")
issued_at = info.get("not_valid_before")
features = info.get("features", [])
config = SERVER_STATE.get("config", {})
if new_seats:
config["max_seats"] = new_seats
if license_id:
config["license_id"] = license_id
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', features = ?
WHERE id = 1
""", (new_seats or config.get("max_seats", 10), license_id or "", issued_at or "", expires_at or "", json.dumps(features)))
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 is not None:
try:
reload_mtls_context(ssl_ctx, cert_dir=cert_dir)
except Exception as e:
print(f"[!] Warning: Failed to reload in-flight SSLContext: {e}")
trigger_async_seat_report()
return {
"status": "success",
"message": "Signed Subordinate CA certificate successfully activated!",
"license_id": license_id,
"seats": new_seats,
"expires_at": expires_at,
"issued_at": issued_at,
"features": features
}
@app.get("/api/db/stats")
def api_db_stats(auth_user: str = Depends(require_admin_auth)):
return {
"status": "success",
"queue": INGEST_QUEUE.get_stats(),
"queue_stats": INGEST_QUEUE.get_stats(),
"storage": get_storage_engine_info(),
"storage_info": get_storage_engine_info(),
"enterprise_db_licensed": is_feature_licensed("enterprise_db"),
"license_summary": get_license_summary()
}
@app.post("/api/db/flush-buffer")
def api_db_flush_buffer(auth_user: str = Depends(require_admin_auth)):
config = SERVER_STATE.get("config", {})
db_path = config.get("db_path", DEFAULT_DB_FILE)
window_hours = config.get("evaluation_window_hours", 12)
min_runs = config.get("min_persistence_runs", 1)
res = INGEST_QUEUE.flush(
db_path=db_path, window_hours=window_hours, min_runs=min_runs, process_callback=process_ingested_logs
)
return {
"status": "success",
"message": f"Successfully flushed {res.get('processed_logs', 0)} events across {res.get('flushed_runs', 0)} batch runs ({res.get('flush_latency_ms', 0)}ms).",
"details": res
}
@app.post("/api/db/test-connection")
def api_db_test_connection(req: DbConnectionTestRequest, auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
if not is_feature_licensed("enterprise_db", db_path):
raise HTTPException(status_code=403, detail="Enterprise database support requires an 'enterprise_db' license capability.")
ok, msg = init_postgres_db(req.dict())
if not ok:
raise HTTPException(status_code=400, detail=f"Database connection failed: {msg}")
return {"status": "success", "message": msg}
@app.post("/api/db/switch-backend")
def api_db_switch_backend(req: DbSwitchBackendRequest, auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
config = SERVER_STATE.get("config", {})
if req.db_type.lower() in ("postgres", "postgresql"):
if not is_feature_licensed("enterprise_db", db_path):
raise HTTPException(status_code=403, detail="PostgreSQL enterprise storage requires an 'enterprise_db' license capability.")
pg_conf = {
"host": req.host,
"port": req.port,
"database": req.database,
"user": req.user,
"password": req.password,
"sslmode": req.sslmode,
"connect_timeout": req.connect_timeout,
"pool_size": req.pool_size
}
ok, msg = init_postgres_db(pg_conf)
if not ok:
raise HTTPException(status_code=400, detail=f"PostgreSQL connection/init failed: {msg}")
config["storage_engine"] = "postgres"
config["db_type"] = "postgres"
config["postgres"] = pg_conf
else:
config["storage_engine"] = "sqlite"
config["db_type"] = "sqlite"
cfg_file = config.get("config_file_path") or CONFIG_FILE_NAME
try:
with open(cfg_file, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
except Exception as e:
print(f"[!] Warning: Could not save config file on backend switch: {e}")
return {
"status": "success",
"message": f"Storage backend switched to {config['storage_engine']}.",
"storage_info": get_storage_engine_info()
}
@app.post("/api/db/update-config")
def api_db_update_config(req: DbFullConfigRequest, auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
config = SERVER_STATE.get("config", {})
is_licensed = is_feature_licensed("enterprise_db", db_path)
if (req.storage_engine.lower() in ("postgres", "postgresql") or req.cache_mode == "buffered" or req.encrypt_at_rest) and not is_licensed:
raise HTTPException(
status_code=403,
detail="Advanced database features (PostgreSQL, in-memory buffering, encryption at rest) require an 'enterprise_db' license capability."
)
if req.storage_engine.lower() in ("postgres", "postgresql"):
if req.pg_config:
pg_conf = req.pg_config.dict()
ok, msg = init_postgres_db(pg_conf)
if not ok:
raise HTTPException(status_code=400, detail=f"PostgreSQL initialization failed: {msg}")
config["postgres"] = pg_conf
config["storage_engine"] = "postgres"
config["db_type"] = "postgres"
else:
config["storage_engine"] = "sqlite"
config["db_type"] = "sqlite"
config["cache_mode"] = req.cache_mode
config["encrypt_at_rest"] = req.encrypt_at_rest
config["buffer_flush_interval_sec"] = req.flush_interval
config["buffer_batch_size"] = req.batch_size
config["buffer_max_queue"] = req.max_queue
INGEST_QUEUE.flush_interval = req.flush_interval
INGEST_QUEUE.batch_size = req.batch_size
INGEST_QUEUE.max_queue = req.max_queue
cfg_file = config.get("config_file_path") or CONFIG_FILE_NAME
try:
with open(cfg_file, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
except Exception as e:
print(f"[!] Warning: Could not save config file: {e}")
return {
"status": "success",
"message": "Storage & Ingestion parameters updated successfully.",
"storage_info": get_storage_engine_info()
}
@app.post("/api/db/migrate-to-postgres")
def api_db_migrate_to_postgres(auth_user: str = Depends(require_admin_auth)):
db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE)
config = SERVER_STATE.get("config", {})
if not is_feature_licensed("enterprise_db", db_path):
raise HTTPException(status_code=403, detail="Database migration requires an 'enterprise_db' license capability.")
pg_config = config.get("postgres")
if not pg_config:
raise HTTPException(status_code=400, detail="PostgreSQL configuration not found. Please configure PostgreSQL parameters first.")
ok, msg, counts = migrate_sqlite_to_postgres(db_path, pg_config)
if not ok:
raise HTTPException(status_code=500, detail=f"Migration failed: {msg}")
return {
"status": "success",
"message": "SQLite database records successfully migrated to PostgreSQL enterprise storage.",
"counts": counts
}
@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
html_content = WebUIRenderer.render_dashboard_html(
auth_user=auth_user,
request=request,
server_state=SERVER_STATE,
ingest_queue=INGEST_QUEUE
)
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")
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}")
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())
collector_task = asyncio.create_task(local_collector_loop(interval_seconds=30))
buffer_task = asyncio.create_task(ingest_buffer_worker_loop())
try:
await asyncio.gather(
tcp_server.serve_forever(),
uv_server.serve(),
watchdog_task,
collector_task,
buffer_task
)
finally:
watchdog_task.cancel()
collector_task.cancel()
buffer_task.cancel()
for t in [watchdog_task, collector_task, buffer_task]:
try:
await t
except asyncio.CancelledError:
pass
try:
db_p = config.get("db_path", DEFAULT_DB_FILE)
INGEST_QUEUE.flush(db_path=db_p, process_callback=process_ingested_logs)
except Exception:
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)
config["db_path"] = 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")
# 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)
# 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()
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("[+] 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)
# 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' AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'").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:
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,
seats_taken=active_seats,
active_seats=active_seats,
hostname=config.get("server_name", ""),
server_fingerprint=config.get("server_fingerprint", "")
)
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("[+] 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
ca_cert, ca_key, ca_pem, ca_key_pem = enrollment.generate_ca_if_needed(cert_dir=cert_dir)
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()
csr_path = os.path.join(cert_dir, "ca.csr")
if not os.path.exists(csr_path):
enrollment.generate_ca_csr(cert_dir=cert_dir)
srv_cert, srv_key, srv_pem, srv_key_pem = enrollment.generate_server_cert_if_needed(
ca_cert, ca_key,
hostnames=[config.get("tcp_host"), "127.0.0.1", "localhost"],
cert_dir=cert_dir
)
ssl_ctx = None
if config.get("tls_enabled", True):
ssl_ctx = init_mtls_server_context(cert_dir=cert_dir)
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.get('server_name', 'LOGAR Enterprise Hub')}")
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.get('server_fingerprint', 'N/A')}")
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()
Reference in New Issue View Git Blame Copy Permalink