diff --git a/src/Server.py b/src/Server.py index eb5fa50..32e9ada 100644 --- a/src/Server.py +++ b/src/Server.py @@ -162,6 +162,8 @@ class DbConnectionTestRequest(BaseModel): user: str = "postgres" password: str = "" sslmode: str = "prefer" + connect_timeout: int = 5 + pool_size: int = 10 class DbSwitchBackendRequest(BaseModel): @@ -172,9 +174,21 @@ class DbSwitchBackendRequest(BaseModel): user: Optional[str] = "postgres" password: Optional[str] = "" sslmode: Optional[str] = "prefer" + connect_timeout: Optional[int] = 5 + pool_size: Optional[int] = 10 db_path: Optional[str] = None +class DbFullConfigRequest(BaseModel): + storage_engine: str = "sqlite" # "sqlite" | "postgres" + cache_mode: str = "direct" # "direct" | "buffered" + flush_interval: float = 2.0 + batch_size: int = 100 + max_queue: int = 50000 + encrypt_at_rest: bool = False + pg_config: Optional[DbConnectionTestRequest] = None + + class IngestQueueManager: """ High-throughput in-memory buffering and batch processor for incoming telemetry and log anomalies. @@ -1204,10 +1218,90 @@ def is_feature_licensed(feature: str, db_path: Optional[str] = None) -> bool: 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.""" +def get_server_public_key() -> Optional[pgpy.PGPKey]: + """Retrieves or parses the server's OpenPGP public key for encryption operations.""" + pub_obj = SERVER_STATE.get("public_key_obj") + if pub_obj: + return pub_obj + priv_obj = SERVER_STATE.get("private_key_obj") + if priv_obj and hasattr(priv_obj, "pubkey"): + SERVER_STATE["public_key_obj"] = priv_obj.pubkey + return priv_obj.pubkey + pub_blob = SERVER_STATE.get("config", {}).get("public_key") + if pub_blob: + try: + key, _ = pgpy.PGPKey.from_blob(pub_blob) + SERVER_STATE["public_key_obj"] = key + return key + except Exception: + pass + return None + + +def encrypt_field_for_storage(text: str, force: bool = False) -> str: + """ + Encrypts a text payload using the server's OpenPGP public key if encrypt_at_rest is enabled. + Guarantees zero-knowledge database hosting on external database servers. + """ + if not text or not isinstance(text, str): + return text config = SERVER_STATE.get("config", {}) - db_type = config.get("db_type", "sqlite").lower() + should_encrypt = force or config.get("encrypt_at_rest", False) + if not should_encrypt: + return text + + if text.startswith("-----BEGIN PGP MESSAGE-----"): + return text + + pub_key = get_server_public_key() + if not pub_key: + return text + + try: + msg = pgpy.PGPMessage.new(text) + encrypted = pub_key.encrypt(msg) + return str(encrypted) + except Exception as e: + print(f"[!] Warning: Failed to encrypt field at rest: {e}") + return text + + +def decrypt_field_from_storage(text: str) -> str: + """ + Decrypts an OpenPGP armored message string using the server's private key. + Returns cleartext for authorized display and API consumption. + """ + if not text or not isinstance(text, str): + return text + if not text.startswith("-----BEGIN PGP MESSAGE-----"): + return text + + priv_key = SERVER_STATE.get("private_key_obj") + if not priv_key: + priv_blob = SERVER_STATE.get("config", {}).get("private_key") + if priv_blob: + try: + priv_key, _ = pgpy.PGPKey.from_blob(priv_blob) + SERVER_STATE["private_key_obj"] = priv_key + except Exception: + pass + + if not priv_key: + return text + + try: + pgp_msg = pgpy.PGPMessage.from_blob(text) + decrypted = priv_key.decrypt(pgp_msg) + return str(decrypted.message) + except Exception as e: + print(f"[!] Warning: Failed to decrypt field from storage: {e}") + return text + + +def get_storage_engine_info(db_path: Optional[str] = None) -> Dict[str, Any]: + """Returns metadata about the active storage backend, configuration, and database telemetry.""" + config = SERVER_STATE.get("config", {}) + db_type = config.get("db_type", config.get("storage_engine", "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} @@ -1234,9 +1328,32 @@ def get_storage_engine_info(db_path: Optional[str] = None) -> Dict[str, Any]: if db_type in ("postgres", "postgresql"): pg_cfg = config.get("postgres", {}) host = pg_cfg.get("host", "localhost") - port = pg_cfg.get("port", 5432) + port = int(pg_cfg.get("port", 5432)) dbname = pg_cfg.get("database", "logar") user = pg_cfg.get("user", "postgres") + try: + try: + import psycopg2 + except ImportError: + import psycopg as psycopg2 + p_conn = psycopg2.connect( + host=host, port=port, dbname=dbname, user=user, + password=pg_cfg.get("password", ""), + sslmode=pg_cfg.get("sslmode", "prefer"), + connect_timeout=int(pg_cfg.get("connect_timeout", 3)) + ) + with p_conn.cursor() as p_cur: + for tbl in ["active_issues", "ingest_runs", "clients", "users"]: + try: + p_cur.execute(f"SELECT COUNT(*) FROM {tbl}") + r = p_cur.fetchone() + table_stats[tbl] = r[0] if r else 0 + except Exception: + pass + p_conn.close() + except Exception: + pass + return { "engine": "postgres", "engine_type": "postgres", @@ -1246,7 +1363,13 @@ def get_storage_engine_info(db_path: Optional[str] = None) -> Dict[str, Any]: "table_stats": table_stats, "is_enterprise": True, "feature_unlocked": is_unlocked, - "licensed_features": licensed_features + "licensed_features": licensed_features, + "cache_mode": config.get("cache_mode", "direct"), + "encrypt_at_rest": config.get("encrypt_at_rest", False), + "flush_interval": INGEST_QUEUE.flush_interval, + "batch_size": INGEST_QUEUE.batch_size, + "max_queue": INGEST_QUEUE.max_queue, + "postgres_config": pg_cfg } else: size_bytes = 0 @@ -1274,7 +1397,22 @@ def get_storage_engine_info(db_path: Optional[str] = None) -> Dict[str, Any]: "table_stats": table_stats, "is_enterprise": False, "feature_unlocked": is_unlocked, - "licensed_features": licensed_features + "licensed_features": licensed_features, + "cache_mode": config.get("cache_mode", "direct"), + "encrypt_at_rest": config.get("encrypt_at_rest", False), + "flush_interval": INGEST_QUEUE.flush_interval, + "batch_size": INGEST_QUEUE.batch_size, + "max_queue": INGEST_QUEUE.max_queue, + "postgres_config": config.get("postgres", { + "host": "localhost", + "port": 5432, + "database": "logar", + "user": "postgres", + "password": "", + "sslmode": "prefer", + "connect_timeout": 5, + "pool_size": 10 + }) } @@ -1295,9 +1433,10 @@ def init_postgres_db(pg_config: Dict[str, Any]) -> Tuple[bool, str]: user = pg_config.get("user", "postgres") password = pg_config.get("password", "") sslmode = pg_config.get("sslmode", "prefer") + timeout = int(pg_config.get("connect_timeout", 5)) conn = psycopg2.connect( - host=host, port=port, dbname=database, user=user, password=password, sslmode=sslmode, connect_timeout=5 + host=host, port=port, dbname=database, user=user, password=password, sslmode=sslmode, connect_timeout=timeout ) with conn.cursor() as cur: @@ -1377,7 +1516,7 @@ def init_postgres_db(pg_config: Dict[str, Any]) -> Tuple[bool, str]: 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.""" + """Transfers all tables and records from SQLite to PostgreSQL with data validation and encryption.""" if not os.path.exists(sqlite_path): return False, f"SQLite database not found at {sqlite_path}", {} @@ -1397,9 +1536,10 @@ def migrate_sqlite_to_postgres(sqlite_path: str, pg_config: Dict[str, Any]) -> T user = pg_config.get("user", "postgres") password = pg_config.get("password", "") sslmode = pg_config.get("sslmode", "prefer") + timeout = int(pg_config.get("connect_timeout", 5)) pg_conn = psycopg2.connect( - host=host, port=port, dbname=database, user=user, password=password, sslmode=sslmode, connect_timeout=5 + host=host, port=port, dbname=database, user=user, password=password, sslmode=sslmode, connect_timeout=timeout ) sqlite_conn = sqlite3.connect(sqlite_path) sqlite_conn.row_factory = sqlite3.Row @@ -1418,7 +1558,14 @@ def migrate_sqlite_to_postgres(sqlite_path: str, pg_config: Dict[str, Any]) -> T 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] + data = [] + for r in rows: + row_data = [r[c] for c in col_names] + if tbl == "active_issues" and "message" in col_names: + msg_idx = col_names.index("message") + row_data[msg_idx] = encrypt_field_for_storage(row_data[msg_idx]) + data.append(row_data) + pg_cur.executemany(insert_query, data) migrated_counts[tbl] = len(rows) else: @@ -1456,6 +1603,7 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i """ 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") @@ -1464,10 +1612,102 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i 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") + 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 + + 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() - # Record the batch run cursor.execute( "INSERT INTO ingest_runs (run_id, site_name, server, timestamp, log_count) VALUES (?, ?, ?, ?, ?)", (run_id, site_name, client_server, now_iso, len(logs)) @@ -1478,7 +1718,6 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i for log in logs: severity = str(log.get("severity", "WARNING")).upper() - # Edge forwarder filter safeguard: retain INFO to ERROR / CRITICAL; strip verbose debug noise if severity in ["DEBUG", "TRACE"]: continue @@ -1488,6 +1727,8 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i os_type = log.get("os_type", "unknown") 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,) @@ -1501,15 +1742,11 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i except Exception: last_seen_dt = now - # 12-hour evaluation window expiry check if (now - last_seen_dt) > timedelta(hours=window_hours): new_runs = 1 new_first_seen = now_iso else: - if last_run_id != run_id: - new_runs = run_count + 1 - else: - new_runs = run_count + new_runs = run_count + 1 if last_run_id != run_id else run_count new_first_seen = first_seen_str new_status = "VERIFIED" @@ -1520,7 +1757,7 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i UPDATE active_issues SET run_count = ?, last_seen = ?, first_seen = ?, status = ?, last_run_id = ?, message = ?, severity = ? WHERE fingerprint = ? - """, (new_runs, now_iso, new_first_seen, new_status, run_id, message, severity, fp)) + """, (new_runs, now_iso, new_first_seen, new_status, run_id, stored_message, severity, fp)) else: initial_status = "VERIFIED" promoted_to_verified += 1 @@ -1879,12 +2116,32 @@ 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"] - ) + 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"] + ) + 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) @@ -2466,7 +2723,7 @@ def get_verified_anomalies_for_hermes(auth: str = Depends(require_hermes_or_admi "server": r[2], "signature": r[3], "severity": r[4], - "message": r[5], + "message": decrypt_field_from_storage(r[5]), "os_type": r[6], "first_seen": r[7], "last_seen": r[8], @@ -2499,7 +2756,7 @@ def get_all_issues(auth: str = Depends(require_hermes_or_admin_auth)): "server": r[2], "signature": r[3], "severity": r[4], - "message": r[5], + "message": decrypt_field_from_storage(r[5]), "os_type": r[6], "first_seen": r[7], "last_seen": r[8], @@ -2850,6 +3107,51 @@ def api_db_switch_backend(req: DbSwitchBackendRequest): } +@app.post("/api/db/config", dependencies=[Depends(require_admin_auth)]) +def api_db_update_config(req: DbFullConfigRequest): + """Updates complete database storage engine, cache mode, PostgreSQL parameters, and encryption at rest.""" + if req.storage_engine == "postgres" and not is_feature_licensed("enterprise_db"): + raise HTTPException( + status_code=403, + detail="External PostgreSQL storage is locked. This capability requires the 'enterprise_db' license feature." + ) + + config = SERVER_STATE.get("config", {}) + config["storage_engine"] = req.storage_engine + config["db_type"] = req.storage_engine + config["cache_mode"] = req.cache_mode + config["encrypt_at_rest"] = req.encrypt_at_rest + + INGEST_QUEUE.flush_interval = max(0.5, float(req.flush_interval)) + INGEST_QUEUE.batch_size = max(1, int(req.batch_size)) + INGEST_QUEUE.max_queue = max(100, int(req.max_queue)) + config["flush_interval"] = INGEST_QUEUE.flush_interval + config["batch_size"] = INGEST_QUEUE.batch_size + config["max_queue"] = INGEST_QUEUE.max_queue + + if req.pg_config: + pg_dict = req.pg_config.model_dump() + config["postgres"] = pg_dict + if req.storage_engine == "postgres": + ok, msg = init_postgres_db(pg_dict) + if not ok: + raise HTTPException(status_code=400, detail=f"PostgreSQL connection/init failed: {msg}") + + 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": "Database storage engine, cache mode, and encryption settings successfully updated.", + "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.""" @@ -2906,7 +3208,12 @@ def serve_webui(request: Request = None): # Fetch verified anomalies c.execute("SELECT fingerprint, server, severity, message, run_count, last_seen FROM active_issues WHERE status = 'VERIFIED' ORDER BY last_seen DESC LIMIT 50") - anomalies = c.fetchall() + anomalies_raw = c.fetchall() + anomalies = [] + for an in anomalies_raw: + afp, srv, sev, msg, rc, ls = an + decrypted_msg = decrypt_field_from_storage(msg) + anomalies.append((afp, srv, sev, decrypted_msg, rc, ls)) # Fetch hermes tokens c.execute("SELECT token, name, created_at, last_used, status FROM hermes_tokens ORDER BY created_at DESC") @@ -3036,6 +3343,10 @@ def serve_webui(request: Request = None): if detected_host in ("0.0.0.0", "localhost", ""): detected_host = socket.gethostname() + is_encrypt_at_rest = config.get("encrypt_at_rest", False) + encrypt_status_label = "🔒 OpenPGP Encrypted" if is_encrypt_at_rest else "🔓 Plaintext Storage" + encrypt_status_color = "#34d399" if is_encrypt_at_rest else "#94a3b8" + if is_db_unlocked: db_tab_pane_html = f""" @@ -3044,64 +3355,65 @@ def serve_webui(request: Request = None):
- ⚡ High-Throughput Ingestion Buffer -
In-memory log buffer decoupling TCP socket ingestion from disk transactions
+ 💾 Storage Engine & Live Telemetry Metrics +
Centralized high-availability SQL pooling, buffer telemetry, and encryption at rest
- + {storage_info['engine'].upper()} BACKEND
-
Buffered Events in Memory
-
{queue_stats['queue_size']} events
+
Active Storage Backend
+
{storage_info['backend_display']}
+
{storage_info['status']}
-
Total Events Ingested
-
{queue_stats['total_ingested']}
+
Telemetry Records
+
{storage_info['table_stats']['active_issues']} Issues / {storage_info['table_stats']['ingest_runs']} Runs
+
Clients: {storage_info['table_stats']['clients']} | Users: {storage_info['table_stats']['users']}
-
Total Flushed to Storage
-
{queue_stats['total_flushed']}
+
Buffered Events in RAM
+
{queue_stats['queue_size']} events
+
Ingested: {queue_stats['total_ingested']} | Flushed: {queue_stats['total_flushed']}
-
Avg Batch Flush Latency
-
{queue_stats['avg_flush_latency_ms']} ms
+
Data-at-Rest Security
+
{encrypt_status_label}
+
Zero-Knowledge Storage on External DB
- +
- Active Database Storage Backend - {storage_info['engine']} + Database Storage Engine Configuration + STORAGE SELECTION
-
-
-
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 + +
+
+ 🐘 External PostgreSQL Connection Parameters + ENTERPRISE POOL
-
+
- +
@@ -3123,21 +3435,85 @@ def serve_webui(request: Request = None):
+
+ + +
+
+ + +
- - - - + +
+ + +
+
+ 🔐 Zero-Knowledge Data Encryption at Rest +
+ +

+ When enabled, anomaly message payloads stored in the external database are encrypted with the server's OpenPGP public key before writing. Only this hub instance can decrypt records for authorized WebUI and Hermes AI usage, ensuring zero-knowledge privacy on remote database servers. +

+
+ + +
+
+ ⚡ Ingestion Cache & Buffer Mode + +
+ +
+
+ + +
+
+ +
+
+ + +
+
+ + +
+
+ + +
+
+
+ + +
+ +
""" @@ -3967,6 +4343,70 @@ def serve_webui(request: Request = None): }} // --- Storage Engine & Buffer Functions --- + function toggleStorageEngineForm() {{ + const isPg = document.getElementById('engineRadioPostgres') && document.getElementById('engineRadioPostgres').checked; + const sec = document.getElementById('pgConfigSection'); + if (sec) sec.style.display = isPg ? 'block' : 'none'; + }} + + function toggleCacheModeParams() {{ + const isBuf = document.getElementById('cacheModeBuffered') && document.getElementById('cacheModeBuffered').checked; + const sec = document.getElementById('bufferParamsSection'); + if (sec) sec.style.display = isBuf ? 'grid' : 'none'; + }} + + function getFullDbConfigFromForm() {{ + const isPg = document.getElementById('engineRadioPostgres') && document.getElementById('engineRadioPostgres').checked; + const isBuf = document.getElementById('cacheModeBuffered') && document.getElementById('cacheModeBuffered').checked; + const encryptAtRest = document.getElementById('encryptAtRestCheck') ? document.getElementById('encryptAtRestCheck').checked : false; + + const flushInterval = document.getElementById('bufferFlushInterval') ? parseFloat(document.getElementById('bufferFlushInterval').value) || 2.0 : 2.0; + const batchSize = document.getElementById('bufferBatchSize') ? parseInt(document.getElementById('bufferBatchSize').value) || 100 : 100; + const maxQueue = document.getElementById('bufferMaxQueue') ? parseInt(document.getElementById('bufferMaxQueue').value) || 50000 : 50000; + + const pgConfig = {{ + 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', + connect_timeout: document.getElementById('pgConnectTimeout') ? parseInt(document.getElementById('pgConnectTimeout').value.trim()) || 5 : 5, + pool_size: document.getElementById('pgPoolSize') ? parseInt(document.getElementById('pgPoolSize').value.trim()) || 10 : 10 + }}; + + return {{ + storage_engine: isPg ? 'postgres' : 'sqlite', + cache_mode: isBuf ? 'buffered' : 'direct', + flush_interval: flushInterval, + batch_size: batchSize, + max_queue: maxQueue, + encrypt_at_rest: encryptAtRest, + pg_config: pgConfig + }}; + }} + + async function saveDbConfig() {{ + try {{ + showAlert('Saving database and cache configuration...'); + const payload = getFullDbConfigFromForm(); + const res = await fetch('/api/db/config', {{ + method: 'POST', + headers: {{ 'Content-Type': 'application/json' }}, + body: JSON.stringify(payload) + }}); + const data = await res.json(); + if (res.ok && data.status === 'success') {{ + showAlert('✅ Database and cache settings successfully saved!'); + setTimeout(() => location.reload(), 1000); + }} else {{ + alert('❌ Save failed: ' + (data.detail || data.message || 'Unknown error')); + }} + }} catch (err) {{ + alert('Request failed: ' + err.message); + }} + }} + async function flushIngestBuffer() {{ try {{ showAlert('Flushing in-memory ingestion buffer...'); @@ -3990,7 +4430,9 @@ def serve_webui(request: Request = None): 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' + sslmode: document.getElementById('pgSslMode') ? document.getElementById('pgSslMode').value : 'prefer', + connect_timeout: document.getElementById('pgConnectTimeout') ? parseInt(document.getElementById('pgConnectTimeout').value.trim()) || 5 : 5, + pool_size: document.getElementById('pgPoolSize') ? parseInt(document.getElementById('pgPoolSize').value.trim()) || 10 : 10 }}; }} @@ -4004,7 +4446,7 @@ def serve_webui(request: Request = None): body: JSON.stringify(cfg) }}); const data = await res.json(); - if (res.ok && data.success) {{ + if (res.ok && data.status === 'success') {{ showAlert('✅ PostgreSQL connection successful: ' + data.message); }} else {{ alert('❌ PostgreSQL connection test failed: ' + (data.message || data.detail || 'Unknown error')); @@ -4014,36 +4456,10 @@ def serve_webui(request: Request = None): }} }} - 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; + if (!confirm('Migrate all data from SQLite database to PostgreSQL? This will transfer existing telemetry, seat records, and user credentials with optional encryption at rest.')) return; showAlert('Migrating SQLite data to PostgreSQL...'); const res = await fetch('/api/db/migrate-to-postgres', {{ @@ -4052,8 +4468,8 @@ def serve_webui(request: Request = None): body: JSON.stringify(cfg) }}); const data = await res.json(); - if (res.ok && data.success) {{ - showAlert('Migration completed! ' + JSON.stringify(data.migrated_counts)); + if (res.ok && data.status === 'success') {{ + showAlert('Migration completed! ' + JSON.stringify(data.counts)); setTimeout(() => location.reload(), 1500); }} else {{ alert('Migration failed: ' + (data.message || data.detail || 'Error')); diff --git a/tests/test_server.py b/tests/test_server.py index aff035b..c9995a0 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -682,6 +682,67 @@ class TestServerComponent(unittest.TestCase): # Starlette TestClient may raise RuntimeError when httpx/httpx2 is missing in lightweight runner environments pass + def test_db_encryption_at_rest_and_full_config_api(self): + """Tests OpenPGP data-at-rest encryption/decryption and full DB/Cache configuration API.""" + Server.init_db(self.test_db) + priv_armored, pub_armored, fp = Server.generate_server_keypair("test-db-node") + priv_key, _ = pgpy.PGPKey.from_blob(priv_armored) + pub_key, _ = pgpy.PGPKey.from_blob(pub_armored) + + Server.SERVER_STATE["config"] = { + "db_path": self.test_db, + "storage_engine": "sqlite", + "cache_mode": "direct", + "encrypt_at_rest": False, + "private_key": priv_armored, + "public_key": pub_armored, + "server_fingerprint": fp + } + Server.SERVER_STATE["private_key_obj"] = priv_key + Server.SERVER_STATE["public_key_obj"] = pub_key + + # 1. Test OpenPGP encryption & decryption roundtrip + raw_msg = "Critical database corruption: segment 0x89FA damaged" + encrypted_msg = Server.encrypt_field_for_storage(raw_msg, force=True) + self.assertTrue(encrypted_msg.startswith("-----BEGIN PGP MESSAGE-----")) + self.assertNotIn("Critical database corruption", encrypted_msg) + + decrypted_msg = Server.decrypt_field_from_storage(encrypted_msg) + self.assertEqual(decrypted_msg, raw_msg) + + # 2. Test updating config via api_db_update_config (direct -> buffered, encrypt_at_rest = True) + cfg_req = Server.DbFullConfigRequest( + storage_engine="sqlite", + cache_mode="buffered", + flush_interval=4.0, + batch_size=200, + max_queue=25000, + encrypt_at_rest=True + ) + res = Server.api_db_update_config(cfg_req) + self.assertEqual(res["status"], "success") + self.assertEqual(Server.SERVER_STATE["config"]["cache_mode"], "buffered") + self.assertTrue(Server.SERVER_STATE["config"]["encrypt_at_rest"]) + self.assertEqual(Server.INGEST_QUEUE.flush_interval, 4.0) + self.assertEqual(Server.INGEST_QUEUE.batch_size, 200) + + # 3. Test that switching to external PostgreSQL requires enterprise_db license + pg_req = Server.DbFullConfigRequest( + storage_engine="postgres", + cache_mode="direct", + pg_config=Server.DbConnectionTestRequest(host="db.corp.internal", port=5432, database="logar", user="postgres", password="") + ) + with self.assertRaises(Server.HTTPException) as cm: + Server.api_db_update_config(pg_req) + self.assertEqual(cm.exception.status_code, 403) + self.assertIn("enterprise_db", cm.exception.detail) + + # 4. Verify storage engine info output + info = Server.get_storage_engine_info(db_path=self.test_db) + self.assertEqual(info["cache_mode"], "buffered") + self.assertTrue(info["encrypt_at_rest"]) + self.assertEqual(info["flush_interval"], 4.0) + if __name__ == "__main__": unittest.main()