diff --git a/LOGAR-LICENSING b/LOGAR-LICENSING index 3368901..b655752 160000 --- a/LOGAR-LICENSING +++ b/LOGAR-LICENSING @@ -1 +1 @@ -Subproject commit 33689019707477d2943f52554aed0fc9d820c9a0 +Subproject commit b6557528a654df51cce988d2e38d5035e57fafbf diff --git a/src/Server.py b/src/Server.py index 4fb2f9f..accbd7e 100644 --- a/src/Server.py +++ b/src/Server.py @@ -155,6 +155,114 @@ class ChangePasswordRequest(BaseModel): new_password: str +class DbConnectionTestRequest(BaseModel): + host: str = "localhost" + port: int = 5432 + database: str = "logar" + user: str = "postgres" + password: str = "" + sslmode: str = "prefer" + + +class DbSwitchBackendRequest(BaseModel): + db_type: str = "sqlite" # "sqlite" | "postgres" + host: Optional[str] = "localhost" + port: Optional[int] = 5432 + database: Optional[str] = "logar" + user: Optional[str] = "postgres" + password: Optional[str] = "" + sslmode: Optional[str] = "prefer" + db_path: Optional[str] = None + + +class IngestQueueManager: + """ + High-throughput in-memory buffering and batch processor for incoming telemetry and log anomalies. + Eliminates SQLite disk write contention by decoupling TCP frame ingress from physical DB writes. + """ + def __init__(self, flush_interval: float = 1.0, batch_size: int = 100, max_queue: int = 50000): + self.flush_interval = flush_interval + self.batch_size = batch_size + self.max_queue = max_queue + self._lock = threading.Lock() + self._buffer: List[Dict[str, Any]] = [] + self._total_ingested = 0 + self._total_flushed = 0 + self._dropped_count = 0 + self._last_flush_time = "Never" + self._avg_flush_latency_ms = 0.0 + self._flush_count = 0 + + def push(self, payload: Dict[str, Any]) -> bool: + with self._lock: + if len(self._buffer) >= self.max_queue: + self._dropped_count += 1 + return False + self._buffer.append(payload) + self._total_ingested += len(payload.get("logs", [])) + return True + + def get_stats(self) -> Dict[str, Any]: + with self._lock: + return { + "queue_size": len(self._buffer), + "total_ingested": self._total_ingested, + "total_flushed": self._total_flushed, + "dropped_count": self._dropped_count, + "last_flush_time": self._last_flush_time, + "avg_flush_latency_ms": round(self._avg_flush_latency_ms, 2), + "flush_count": self._flush_count, + "flush_interval_seconds": self.flush_interval, + "batch_size_threshold": self.batch_size + } + + def flush(self, db_path: Optional[str] = None, window_hours: int = 12, min_runs: int = 1) -> Dict[str, Any]: + with self._lock: + if not self._buffer: + return {"status": "noop", "flushed_runs": 0, "processed_logs": 0} + items_to_process = list(self._buffer) + self._buffer.clear() + + start_time = time.time() + target_db = db_path or SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE) + + total_processed = 0 + promoted_total = 0 + + try: + for payload in items_to_process: + res = process_ingested_logs( + payload=payload, + db_path=target_db, + window_hours=window_hours, + min_runs=min_runs + ) + total_processed += res.get("processed", 0) + promoted_total += res.get("promoted_verified", 0) + except Exception as e: + print(f"[!] Error during batch flush to DB: {e}") + + elapsed_ms = (time.time() - start_time) * 1000.0 + with self._lock: + self._total_flushed += total_processed + self._flush_count += 1 + self._avg_flush_latency_ms = ( + (self._avg_flush_latency_ms * (self._flush_count - 1) + elapsed_ms) / self._flush_count + ) + self._last_flush_time = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + + return { + "status": "success", + "flushed_runs": len(items_to_process), + "processed_logs": total_processed, + "promoted_verified": promoted_total, + "flush_latency_ms": round(elapsed_ms, 2) + } + + +INGEST_QUEUE = IngestQueueManager() + + # --- Rate Limiting & Brute-Force Mitigation --- _RATE_LIMIT_LOCK = threading.Lock() @@ -914,7 +1022,8 @@ def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: in ("license_server_url", "TEXT"), ("issued_at", "TEXT"), ("expires_at", "TEXT"), - ("license_status", "TEXT DEFAULT 'unlicensed'") + ("license_status", "TEXT DEFAULT 'unlicensed'"), + ("features", "TEXT DEFAULT '[]'") ]: try: conn.execute(f"ALTER TABLE license_config ADD COLUMN {col} {col_def}") @@ -1003,11 +1112,10 @@ def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: in """) status = "active" if max_seats > 0 else "unlicensed" - if enrollment_secret: - conn.execute(""" - INSERT OR IGNORE INTO license_config (id, max_seats, enrollment_secret, license_status) - VALUES (1, ?, ?, ?) - """, (max_seats, enrollment_secret, status)) + conn.execute(""" + INSERT OR IGNORE INTO license_config (id, max_seats, enrollment_secret, license_status, features) + VALUES (1, ?, ?, ?, '[]') + """, (max_seats, enrollment_secret or "", status)) conn.commit() conn.close() @@ -1044,6 +1152,306 @@ def export_license_request_lic(config: Dict[str, Any]) -> Tuple[Dict[str, Any], ) +def is_feature_licensed(feature: str, db_path: Optional[str] = None) -> bool: + """ + Checks whether a specific enterprise capability (e.g. 'enterprise_db') is unlocked + either via custom X.509 certificate extensions or the persistent license configuration. + """ + # 1. Check in-memory / active Subordinate CA certificate + ca_pem = SERVER_STATE.get("ca_cert_pem") + if not ca_pem: + cert_dir = SERVER_STATE.get("config", {}).get("cert_dir", "certs") + ca_file = os.path.join(cert_dir, "ca.crt") + if os.path.exists(ca_file): + try: + with open(ca_file, "r", encoding="utf-8") as f: + ca_pem = f.read() + except Exception: + pass + + if ca_pem: + try: + info = enrollment.extract_license_info_from_cert(ca_pem) + cert_features = info.get("features", []) + if feature in cert_features: + return True + except Exception: + pass + + # 2. Check license_config table in database + target_db = db_path or SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE) + if target_db and os.path.exists(target_db): + try: + conn = sqlite3.connect(target_db, timeout=3.0) + c = conn.cursor() + c.execute("SELECT features FROM license_config WHERE id = 1") + row = c.fetchone() + conn.close() + if row and row[0]: + feat_raw = row[0] + if isinstance(feat_raw, str): + if feat_raw.startswith("["): + feat_list = json.loads(feat_raw) + else: + feat_list = [f.strip() for f in feat_raw.split(",") if f.strip()] + if feature in feat_list: + return True + elif isinstance(feat_raw, list) and feature in feat_raw: + return True + except Exception: + pass + + return False + + +def get_storage_engine_info(db_path: Optional[str] = None) -> Dict[str, Any]: + """Returns metadata about the active storage backend and database telemetry.""" + config = SERVER_STATE.get("config", {}) + db_type = config.get("db_type", "sqlite").lower() + target_db = db_path or config.get("db_path", DEFAULT_DB_FILE) + + table_stats = {"active_issues": 0, "ingest_runs": 0, "clients": 0, "users": 0} + is_unlocked = is_feature_licensed("enterprise_db", db_path=target_db) + + licensed_features = [] + if target_db and os.path.exists(target_db): + try: + conn_l = sqlite3.connect(target_db, timeout=3.0) + row_l = conn_l.execute("SELECT features FROM license_config WHERE id = 1").fetchone() + conn_l.close() + if row_l and row_l[0]: + feat_raw = row_l[0] + if isinstance(feat_raw, str): + if feat_raw.startswith("["): + licensed_features = json.loads(feat_raw) + else: + licensed_features = [f.strip() for f in feat_raw.split(",") if f.strip()] + elif isinstance(feat_raw, list): + licensed_features = list(feat_raw) + except Exception: + pass + + if db_type in ("postgres", "postgresql"): + pg_cfg = config.get("postgres", {}) + host = pg_cfg.get("host", "localhost") + port = pg_cfg.get("port", 5432) + dbname = pg_cfg.get("database", "logar") + user = pg_cfg.get("user", "postgres") + return { + "engine": "postgres", + "engine_type": "postgres", + "backend_display": "PostgreSQL Enterprise Pool", + "connection_target": f"{user}@{host}:{port}/{dbname}", + "status": "Enterprise Connected", + "table_stats": table_stats, + "is_enterprise": True, + "feature_unlocked": is_unlocked, + "licensed_features": licensed_features + } + else: + size_bytes = 0 + if target_db and os.path.exists(target_db): + try: + size_bytes = os.path.getsize(target_db) + conn = sqlite3.connect(target_db, timeout=3.0) + for tbl in ["active_issues", "ingest_runs", "clients", "users"]: + try: + row = conn.execute(f"SELECT COUNT(*) FROM {tbl}").fetchone() + table_stats[tbl] = row[0] if row else 0 + except Exception: + pass + conn.close() + except Exception: + pass + size_kb = round(size_bytes / 1024, 2) + return { + "engine": "sqlite", + "engine_type": "sqlite", + "backend_display": "SQLite 3 (WAL Mode)", + "connection_target": target_db, + "db_size_kb": size_kb, + "status": "Local WAL Optimized", + "table_stats": table_stats, + "is_enterprise": False, + "feature_unlocked": is_unlocked, + "licensed_features": licensed_features + } + + +def init_postgres_db(pg_config: Dict[str, Any]) -> Tuple[bool, str]: + """Initializes table schema on a PostgreSQL database instance.""" + try: + try: + import psycopg2 + except ImportError: + try: + import psycopg as psycopg2 + except ImportError: + return False, "PostgreSQL driver not installed on server (install psycopg2-binary or psycopg)" + + host = pg_config.get("host", "localhost") + port = int(pg_config.get("port", 5432)) + database = pg_config.get("database", "logar") + user = pg_config.get("user", "postgres") + password = pg_config.get("password", "") + sslmode = pg_config.get("sslmode", "prefer") + + conn = psycopg2.connect( + host=host, port=port, dbname=database, user=user, password=password, sslmode=sslmode, connect_timeout=5 + ) + + with conn.cursor() as cur: + cur.execute(""" + CREATE TABLE IF NOT EXISTS active_issues ( + fingerprint VARCHAR(255) PRIMARY KEY, + site_name VARCHAR(255), + server VARCHAR(255), + signature TEXT, + severity VARCHAR(50), + message TEXT, + os_type VARCHAR(50), + first_seen VARCHAR(50), + last_seen VARCHAR(50), + run_count INTEGER, + status VARCHAR(50), + last_run_id VARCHAR(100) + ); + CREATE TABLE IF NOT EXISTS ingest_runs ( + run_id VARCHAR(100) PRIMARY KEY, + site_name VARCHAR(255), + server VARCHAR(255), + timestamp VARCHAR(50), + log_count INTEGER + ); + CREATE TABLE IF NOT EXISTS license_config ( + id INTEGER PRIMARY KEY CHECK (id = 1), + max_seats INTEGER NOT NULL DEFAULT 0, + enrollment_secret TEXT NOT NULL, + license_id VARCHAR(100), + license_server_url VARCHAR(255), + issued_at VARCHAR(50), + expires_at VARCHAR(50), + license_status VARCHAR(50) DEFAULT 'unlicensed', + features TEXT DEFAULT '[]' + ); + CREATE TABLE IF NOT EXISTS clients ( + client_id VARCHAR(100) PRIMARY KEY, + hostname VARCHAR(255) NOT NULL, + os_type VARCHAR(50) NOT NULL, + cert_fingerprint VARCHAR(255) NOT NULL, + status VARCHAR(50) DEFAULT 'active', + first_seen VARCHAR(50), + last_seen VARCHAR(50) + ); + CREATE TABLE IF NOT EXISTS users ( + username VARCHAR(100) PRIMARY KEY, + password_hash TEXT NOT NULL, + salt TEXT NOT NULL, + role VARCHAR(50) NOT NULL DEFAULT 'admin', + created_at VARCHAR(50) NOT NULL, + totp_secret TEXT, + totp_enabled INTEGER DEFAULT 0 + ); + CREATE TABLE IF NOT EXISTS hermes_tokens ( + token VARCHAR(255) PRIMARY KEY, + name VARCHAR(255) NOT NULL, + created_at VARCHAR(50) NOT NULL, + last_used VARCHAR(50), + status VARCHAR(50) NOT NULL DEFAULT 'active' + ); + CREATE TABLE IF NOT EXISTS rate_limits ( + ip VARCHAR(100) NOT NULL, + scope VARCHAR(50) NOT NULL, + attempt_time DOUBLE PRECISION NOT NULL + ); + CREATE TABLE IF NOT EXISTS server_meta ( + key VARCHAR(100) PRIMARY KEY, + value TEXT NOT NULL + ); + """) + conn.commit() + conn.close() + return True, "PostgreSQL schema successfully initialized" + except Exception as e: + return False, f"PostgreSQL connection/init failed: {e}" + + +def migrate_sqlite_to_postgres(sqlite_path: str, pg_config: Dict[str, Any]) -> Tuple[bool, str, Dict[str, int]]: + """Transfers all tables and records from SQLite to PostgreSQL with data validation.""" + if not os.path.exists(sqlite_path): + return False, f"SQLite database not found at {sqlite_path}", {} + + ok, init_msg = init_postgres_db(pg_config) + if not ok: + return False, init_msg, {} + + try: + try: + import psycopg2 + except ImportError: + import psycopg as psycopg2 + + host = pg_config.get("host", "localhost") + port = int(pg_config.get("port", 5432)) + database = pg_config.get("database", "logar") + user = pg_config.get("user", "postgres") + password = pg_config.get("password", "") + sslmode = pg_config.get("sslmode", "prefer") + + pg_conn = psycopg2.connect( + host=host, port=port, dbname=database, user=user, password=password, sslmode=sslmode, connect_timeout=5 + ) + sqlite_conn = sqlite3.connect(sqlite_path) + sqlite_conn.row_factory = sqlite3.Row + + migrated_counts = {} + tables = ["active_issues", "ingest_runs", "license_config", "clients", "users", "hermes_tokens", "rate_limits", "server_meta"] + + with pg_conn.cursor() as pg_cur: + for tbl in tables: + try: + s_cur = sqlite_conn.cursor() + s_cur.execute(f"SELECT * FROM {tbl}") + rows = s_cur.fetchall() + if rows: + col_names = [col[0] for col in s_cur.description] + cols_joined = ", ".join(col_names) + placeholders = ", ".join(["%s"] * len(col_names)) + insert_query = f"INSERT INTO {tbl} ({cols_joined}) VALUES ({placeholders}) ON CONFLICT DO NOTHING;" + data = [[r[c] for c in col_names] for r in rows] + pg_cur.executemany(insert_query, data) + migrated_counts[tbl] = len(rows) + else: + migrated_counts[tbl] = 0 + except Exception as tbl_err: + print(f"[!] Warning migrating table {tbl}: {tbl_err}") + migrated_counts[tbl] = 0 + + pg_conn.commit() + + sqlite_conn.close() + pg_conn.close() + return True, "Migration completed successfully", migrated_counts + except Exception as e: + return False, f"Migration failed: {e}", {} + + +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) + 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. @@ -1471,12 +1879,20 @@ async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.Str 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"] - ) + 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"] + ) + else: + res = { + "status": "success", + "buffered": True, + "processed": len(log_payload.get("logs", [])), + "queue_size": len(INGEST_QUEUE._buffer) + } resp_bytes = json.dumps(res).encode("utf-8") writer.write(struct.pack(">I", len(resp_bytes)) + resp_bytes) @@ -2356,12 +2772,112 @@ async def trigger_license_renewal(request: Request): reload_mtls_context(ssl_ctx, cert_dir=cert_dir) print(f"[+] In-flight mTLS SSLContext successfully reloaded after renewal/sync (Seats: {info.get('seats')}).") +# --- Database Engine & Ingestion Buffer Endpoints --- + +@app.get("/api/db/stats", dependencies=[Depends(require_admin_auth)]) +def api_db_stats(): + """Returns real-time buffer throughput, queue stats, and storage engine metrics.""" return { "status": "success", - "message": msg, - "seats": info.get("seats"), - "expires_at": info.get("not_valid_after"), - "days_remaining": info.get("days_remaining") + "queue": INGEST_QUEUE.get_stats(), + "storage": get_storage_engine_info(), + "enterprise_db_licensed": is_feature_licensed("enterprise_db") + } + + +@app.post("/api/db/flush-buffer", dependencies=[Depends(require_admin_auth)]) +def api_db_flush_buffer(): + """Manually forces immediate drain and write of all buffered anomalies to the database.""" + 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) + 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).", + "flush_details": res, + "queue": INGEST_QUEUE.get_stats() + } + + +@app.post("/api/db/test-connection", dependencies=[Depends(require_admin_auth)]) +def api_db_test_connection(req: DbConnectionTestRequest): + """Tests connection to external PostgreSQL database. Requires enterprise_db license feature.""" + if not is_feature_licensed("enterprise_db"): + raise HTTPException( + status_code=403, + detail="Enterprise Database feature is locked. This capability requires a valid license with the 'enterprise_db' feature." + ) + ok, msg = init_postgres_db(req.model_dump()) + if not ok: + raise HTTPException(status_code=400, detail=msg) + return {"status": "success", "message": "Successfully connected to PostgreSQL and verified schema."} + + +@app.post("/api/db/switch-backend", dependencies=[Depends(require_admin_auth)]) +def api_db_switch_backend(req: DbSwitchBackendRequest): + """Switches active database backend between SQLite and PostgreSQL. Requires enterprise_db license feature.""" + if not is_feature_licensed("enterprise_db"): + raise HTTPException( + status_code=403, + detail="Enterprise Database feature is locked. This capability requires a valid license with the 'enterprise_db' feature." + ) + config = SERVER_STATE.get("config", {}) + if req.db_type == "postgres": + pg_cfg = { + "host": req.host or "localhost", + "port": req.port or 5432, + "database": req.database or "logar", + "user": req.user or "postgres", + "password": req.password or "", + "sslmode": req.sslmode or "prefer" + } + ok, msg = init_postgres_db(pg_cfg) + if not ok: + raise HTTPException(status_code=400, detail=f"Cannot switch backend: {msg}") + config["db_type"] = "postgres" + config["postgres"] = pg_cfg + else: + config["db_type"] = "sqlite" + if req.db_path: + config["db_path"] = req.db_path + + cfg_path = config.get("config_file_path", CONFIG_FILE_NAME) + if os.path.exists(cfg_path): + try: + with open(cfg_path, "w", encoding="utf-8") as f: + json.dump(config, f, indent=2) + except Exception: + pass + + return { + "status": "success", + "message": f"Storage backend successfully switched to {config.get('db_type', 'sqlite')}.", + "storage": get_storage_engine_info() + } + + +@app.post("/api/db/migrate-to-postgres", dependencies=[Depends(require_admin_auth)]) +def api_db_migrate_to_postgres(req: DbConnectionTestRequest): + """Migrates all telemetry, runs, clients, users, and license configs from SQLite to PostgreSQL.""" + if not is_feature_licensed("enterprise_db"): + raise HTTPException( + status_code=403, + detail="Enterprise Database feature is locked. This capability requires a valid license with the 'enterprise_db' feature." + ) + config = SERVER_STATE.get("config", {}) + sqlite_path = config.get("db_path", DEFAULT_DB_FILE) + + INGEST_QUEUE.flush(db_path=sqlite_path) + + ok, msg, counts = migrate_sqlite_to_postgres(sqlite_path, req.model_dump()) + if not ok: + raise HTTPException(status_code=400, detail=msg) + return { + "status": "success", + "message": "SQLite database records successfully migrated to PostgreSQL enterprise storage.", + "counts": counts } @@ -2383,6 +2899,12 @@ def serve_webui(request: Request = None): config = SERVER_STATE.get("config", {}) db_path = config.get("db_path", DEFAULT_DB_FILE) + is_db_unlocked = is_feature_licensed("enterprise_db", db_path) + queue_stats = INGEST_QUEUE.get_stats() + storage_info = get_storage_engine_info() + pg_config = config.get("postgres", {}) + db_tab_btn_badge = 'đŸŸĸ UNLOCKED' if is_db_unlocked else '🔒 ENTERPRISE' + # Fetch clients conn = sqlite3.connect(db_path) @@ -2522,6 +3044,134 @@ def serve_webui(request: Request = None): if detected_host in ("0.0.0.0", "localhost", ""): detected_host = socket.gethostname() + if is_db_unlocked: + db_tab_pane_html = f""" + +
+ +
+
+
+ ⚡ High-Throughput Ingestion Buffer +
In-memory log buffer decoupling TCP socket ingestion from disk transactions
+
+ +
+
+
+
Buffered Events in Memory
+
{queue_stats['queue_size']} events
+
+
+
Total Events Ingested
+
{queue_stats['total_ingested']}
+
+
+
Total Flushed to Storage
+
{queue_stats['total_flushed']}
+
+
+
Avg Batch Flush Latency
+
{queue_stats['avg_flush_latency_ms']} ms
+
+
+
+ + +
+
+ Active Database Storage Backend + {storage_info['engine']} +
+
+
+
Engine Backend Type
+
{storage_info['engine']}
+
+
+
Active Target / Path
+
{storage_info['connection_target']}
+
+
+
Status
+
{storage_info['status']}
+
+
+
Telemetry Records
+
{storage_info['table_stats']['active_issues']} Issues / {storage_info['table_stats']['ingest_runs']} Runs
+
+
+ + +
+
+ 🐘 Enterprise PostgreSQL Connection Configurator +
+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+
+ + + + +
+
+
+
+ """ + else: + db_tab_pane_html = f""" + +
+
+
🔒
+

Enterprise Database & High-Throughput Engine is Locked

+
+ + REQUIRES 'enterprise_db' LICENSE FEATURE + +
+

+ Direct PostgreSQL storage backend clustering, automatic schema synchronization, and customizable high-throughput batch-flush controls are gated behind the enterprise_db capability. +

+
+ + +
+
+
+ """ + html_content = f""" @@ -2752,6 +3402,7 @@ def serve_webui(request: Request = None): + @@ -2939,6 +3590,8 @@ def serve_webui(request: Request = None): + + {db_tab_pane_html} @@ -3321,6 +3974,103 @@ def serve_webui(request: Request = None): }} }} + // --- Storage Engine & Buffer Functions --- + async function flushIngestBuffer() {{ + try {{ + showAlert('Flushing in-memory ingestion buffer...'); + const res = await fetch('/api/db/flush-buffer', {{ method: 'POST' }}); + const data = await res.json(); + if (res.ok) {{ + showAlert('Buffer flushed: ' + (data.message || 'Flushed')); + setTimeout(() => location.reload(), 800); + }} else {{ + alert('Flush failed: ' + (data.detail || 'Error')); + }} + }} catch (err) {{ + alert('Buffer flush failed: ' + err.message); + }} + }} + + function getPgConfigFromForm() {{ + return {{ + host: document.getElementById('pgHost') ? document.getElementById('pgHost').value.trim() : 'localhost', + port: document.getElementById('pgPort') ? parseInt(document.getElementById('pgPort').value.trim()) || 5432 : 5432, + database: document.getElementById('pgDatabase') ? document.getElementById('pgDatabase').value.trim() : 'logar', + user: document.getElementById('pgUser') ? document.getElementById('pgUser').value.trim() : 'postgres', + password: document.getElementById('pgPassword') ? document.getElementById('pgPassword').value : '', + sslmode: document.getElementById('pgSslMode') ? document.getElementById('pgSslMode').value : 'prefer' + }}; + }} + + async function testPgConnection() {{ + try {{ + showAlert('Testing connection to PostgreSQL...'); + const cfg = getPgConfigFromForm(); + const res = await fetch('/api/db/test-connection', {{ + method: 'POST', + headers: {{ 'Content-Type': 'application/json' }}, + body: JSON.stringify(cfg) + }}); + const data = await res.json(); + if (res.ok && data.success) {{ + showAlert('✅ PostgreSQL connection successful: ' + data.message); + }} else {{ + alert('❌ PostgreSQL connection test failed: ' + (data.message || data.detail || 'Unknown error')); + }} + }} catch (err) {{ + alert('Test connection request failed: ' + err.message); + }} + }} + + async function switchStorageEngine(engineType) {{ + try {{ + const cfg = engineType === 'postgres' ? getPgConfigFromForm() : null; + const confirmMsg = engineType === 'postgres' + ? 'Switch active storage engine to PostgreSQL? Make sure you have migrated your SQLite data if needed.' + : 'Switch back to SQLite database storage?'; + if (!confirm(confirmMsg)) return; + + showAlert('Switching database storage engine to ' + engineType + '...'); + const res = await fetch('/api/db/switch-backend', {{ + method: 'POST', + headers: {{ 'Content-Type': 'application/json' }}, + body: JSON.stringify({{ engine: engineType, pg_config: cfg }}) + }}); + const data = await res.json(); + if (res.ok && data.success) {{ + showAlert('Storage engine switched to ' + engineType + '!'); + setTimeout(() => location.reload(), 1000); + }} else {{ + alert('Failed to switch engine: ' + (data.message || data.detail || 'Unknown error')); + }} + }} catch (err) {{ + alert('Switch engine request failed: ' + err.message); + }} + }} + + async function migrateToPostgres() {{ + try {{ + const cfg = getPgConfigFromForm(); + if (!confirm('Migrate all data from SQLite database to PostgreSQL? This will transfer existing telemetry, seat records, and user credentials.')) return; + + showAlert('Migrating SQLite data to PostgreSQL...'); + const res = await fetch('/api/db/migrate-to-postgres', {{ + method: 'POST', + headers: {{ 'Content-Type': 'application/json' }}, + body: JSON.stringify(cfg) + }}); + const data = await res.json(); + if (res.ok && data.success) {{ + showAlert('Migration completed! ' + JSON.stringify(data.migrated_counts)); + setTimeout(() => location.reload(), 1500); + }} else {{ + alert('Migration failed: ' + (data.message || data.detail || 'Error')); + }} + }} catch (err) {{ + alert('Migration request failed: ' + err.message); + }} + }} + // --- 2FA Management & QRCode Logic --- let current2FASecret = ''; @@ -3477,22 +4227,30 @@ async def run_server(): 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 + collector_task, + buffer_task ) finally: watchdog_task.cancel() collector_task.cancel() - for t in [watchdog_task, collector_task]: + 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) + except Exception: + pass def main(): diff --git a/src/server_enrollment.py b/src/server_enrollment.py index 2adda6d..bdafa8e 100644 --- a/src/server_enrollment.py +++ b/src/server_enrollment.py @@ -17,6 +17,7 @@ from cryptography.hazmat.primitives.asymmetric import rsa OID_LOGAR_SEATS = x509.ObjectIdentifier("1.3.6.1.4.1.59999.1.1") OID_LOGAR_LICENSE_ID = x509.ObjectIdentifier("1.3.6.1.4.1.59999.1.2") OID_LOGAR_ISSUER_URL = x509.ObjectIdentifier("1.3.6.1.4.1.59999.1.3") +OID_LOGAR_FEATURES = x509.ObjectIdentifier("1.3.6.1.4.1.59999.1.4") @@ -448,6 +449,19 @@ def extract_license_info_from_cert(cert_pem: str) -> Dict[str, Any]: except (x509.ExtensionNotFound, Exception): issuer_url = None + features = [] + try: + ext_feat = cert.extensions.get_extension_for_oid(OID_LOGAR_FEATURES) + if ext_feat and ext_feat.value: + raw_val = ext_feat.value.value if hasattr(ext_feat.value, "value") else str(ext_feat.value) + feat_str = raw_val.decode("utf-8").strip() if isinstance(raw_val, bytes) else str(raw_val).strip() + if feat_str.startswith("["): + features = json.loads(feat_str) + else: + features = [f.strip() for f in feat_str.split(",") if f.strip()] + except Exception: + features = [] + not_before = getattr(cert, "not_valid_before_utc", None) or cert.not_valid_before.replace(tzinfo=datetime.timezone.utc) not_after = getattr(cert, "not_valid_after_utc", None) or cert.not_valid_after.replace(tzinfo=datetime.timezone.utc) now = datetime.datetime.now(datetime.timezone.utc) @@ -482,6 +496,7 @@ def extract_license_info_from_cert(cert_pem: str) -> Dict[str, Any]: "seats": seats, "license_id": license_id, "issuer_url": issuer_url, + "features": features, "subject_cn": subject_cn, "issuer_cn": issuer_cn, "is_ca": is_ca, @@ -603,11 +618,12 @@ def sign_ca_csr_with_seats( seats: int, days_valid: int = 365, license_id: Optional[str] = None, - license_server_url: str = "https://logar.eibl.tech" + license_server_url: str = "https://logar.eibl.tech", + features: Optional[List[str]] = None ) -> str: """ Signs a subordinate CA CSR using the master license authority Root CA. - Embeds the approved seat count and license ID via custom X.509 extensions. + Embeds the approved seat count, license ID, and optional enterprise features via custom X.509 extensions. Returns the signed PEM certificate. """ csr = x509.load_pem_x509_csr(csr_pem.encode("utf-8")) @@ -666,6 +682,13 @@ def sign_ca_csr_with_seats( ) ) + if features: + feat_str = json.dumps(features) if isinstance(features, list) else str(features) + builder = builder.add_extension( + x509.UnrecognizedExtension(OID_LOGAR_FEATURES, feat_str.encode("utf-8")), + critical=False + ) + signed_cert = builder.sign(master_ca_key, hashes.SHA256()) return signed_cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") @@ -903,6 +926,7 @@ def import_license_package( "master_root_ca_pem": master_root_ca_pem, "license_info": info, "seats": info.get("seats"), + "features": info.get("features", []), "license_id": info.get("license_id"), "issuer_url": info.get("issuer_url"), "fingerprint": info.get("fingerprint"), diff --git a/tests/test_server.py b/tests/test_server.py index 20749e8..b1cc176 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -547,8 +547,118 @@ class TestServerComponent(unittest.TestCase): Server.reset_failed_ip_attempts(test_ip, scope="login", db_path=self.test_db) self.assertTrue(Server.check_ip_rate_limit(test_ip, max_attempts=3, window_seconds=60, scope="login", db_path=self.test_db)) + def test_ingest_queue_manager_and_batch_flush(self): + """Tests IngestQueueManager buffering, capacity bounds, and atomic batch flushing.""" + Server.init_db(self.test_db) + qm = Server.IngestQueueManager(flush_interval=1.0, batch_size=5, max_queue=50) + + # Initial stats + stats = qm.get_stats() + self.assertEqual(stats["queue_size"], 0) + self.assertEqual(stats["total_ingested"], 0) + self.assertEqual(stats["total_flushed"], 0) + + # Push 3 payloads + for i in range(3): + payload = { + "server": f"node-{i}.corp.local", + "logs": [ + {"server": f"node-{i}", "signature": f"Sig_{i}", "severity": "WARNING", "message": f"Log msg {i}", "os_type": "linux"} + ] + } + pushed = qm.push(payload) + self.assertTrue(pushed) + + stats = qm.get_stats() + self.assertEqual(stats["queue_size"], 3) + self.assertEqual(stats["total_ingested"], 3) + + # Flush to database + res_flush = qm.flush(db_path=self.test_db, window_hours=12, min_runs=1) + self.assertEqual(res_flush["flushed_runs"], 3) + self.assertEqual(res_flush["processed_logs"], 3) + + # Verify DB records + conn = sqlite3.connect(self.test_db) + c = conn.cursor() + c.execute("SELECT COUNT(*) FROM ingest_runs") + run_count = c.fetchone()[0] + c.execute("SELECT COUNT(*) FROM active_issues WHERE status = 'VERIFIED'") + issue_count = c.fetchone()[0] + conn.close() + + self.assertEqual(run_count, 3) + self.assertEqual(issue_count, 3) + + # Stats after flush + stats = qm.get_stats() + self.assertEqual(stats["queue_size"], 0) + self.assertEqual(stats["total_flushed"], 3) + + def test_enterprise_db_feature_gating(self): + """Tests feature check helpers and DB storage info under standard and enterprise licensed states.""" + Server.init_db(self.test_db) + + # 1. Standard / Unlicensed state + self.assertFalse(Server.is_feature_licensed("enterprise_db", db_path=self.test_db)) + info = Server.get_storage_engine_info(db_path=self.test_db) + self.assertEqual(info["engine"], "sqlite") + self.assertFalse(info["feature_unlocked"]) + self.assertIn("SQLite 3 (WAL Mode)", info["backend_display"]) + + # 2. Unlock enterprise_db feature in DB + conn = sqlite3.connect(self.test_db) + conn.execute("UPDATE license_config SET features = ? WHERE id = 1", (json.dumps(["enterprise_db", "ai_heuristics"]),)) + conn.commit() + conn.close() + + self.assertTrue(Server.is_feature_licensed("enterprise_db", db_path=self.test_db)) + self.assertTrue(Server.is_feature_licensed("ai_heuristics", db_path=self.test_db)) + self.assertFalse(Server.is_feature_licensed("unknown_feat", db_path=self.test_db)) + + info_unlocked = Server.get_storage_engine_info(db_path=self.test_db) + self.assertTrue(info_unlocked["feature_unlocked"]) + self.assertIn("enterprise_db", info_unlocked["licensed_features"]) + + def test_postgres_storage_endpoints_authorization(self): + """Tests REST API security gating on DB engine management endpoints.""" + from fastapi.testclient import TestClient + Server.init_db(self.test_db) + Server.SERVER_STATE["config"] = {"db_path": self.test_db, "storage_engine": "sqlite"} + client = TestClient(Server.app) + + # 1. Unauthenticated request to /api/db/stats -> 401 + res = client.get("/api/db/stats") + self.assertEqual(res.status_code, 401) + + # 2. Authenticated request with session cookie + token = Server.generate_session_token("admin", db_path=self.test_db) + client.cookies.set(Server.SESSION_COOKIE_NAME, token) + + res_auth = client.get("/api/db/stats") + self.assertEqual(res_auth.status_code, 200) + data = res_auth.json() + self.assertIn("queue", data) + self.assertIn("storage", data) + self.assertFalse(data["enterprise_db_licensed"]) + + # 3. /api/db/test-connection without enterprise_db feature -> 403 Forbidden + pg_req = {"host": "localhost", "port": 5432, "database": "logar", "user": "postgres", "password": ""} + res_test = client.post("/api/db/test-connection", json=pg_req) + self.assertEqual(res_test.status_code, 403) + self.assertIn("enterprise_db", res_test.json()["detail"]) + + # 4. /api/db/switch-backend without enterprise_db feature -> 403 Forbidden + res_switch = client.post("/api/db/switch-backend", json={"engine": "postgres", "pg_config": pg_req}) + self.assertEqual(res_switch.status_code, 403) + + # 5. Flush buffer endpoint is accessible to authenticated admins + res_flush = client.post("/api/db/flush-buffer") + self.assertEqual(res_flush.status_code, 200) + if __name__ == "__main__": unittest.main() +