From 4a40b4d39579065eb9f899231bbf8eabdde52c8e Mon Sep 17 00:00:00 2001 From: max Date: Sun, 27 Sep 2026 15:00:26 +0200 Subject: [PATCH] windows installer fixes --- src/Server.py | 371 ++++++++++++++++++++++++++++++++++++++++--- tests/test_server.py | 37 ++++- 2 files changed, 384 insertions(+), 24 deletions(-) diff --git a/src/Server.py b/src/Server.py index 4a09381..c541052 100644 --- a/src/Server.py +++ b/src/Server.py @@ -37,11 +37,17 @@ from pgpy.constants import ( CompressionAlgorithm ) import ssl +import subprocess from pydantic import BaseModel from fastapi import FastAPI, HTTPException, Request, Depends from fastapi.responses import HTMLResponse, JSONResponse, Response, RedirectResponse import uvicorn +try: + import win32evtlog +except ImportError: + win32evtlog = None + try: from src import server_enrollment as enrollment except ImportError: @@ -909,6 +915,311 @@ def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: i } +# --- Integrated Local Hub Host Log Collector --- + +def get_machine_identifier() -> str: + """Returns the hostname / FQDN of the local server hub host.""" + fqdn = socket.getfqdn() + if fqdn and "." in fqdn and not fqdn.startswith("localhost"): + return fqdn + + hostname = socket.gethostname() + user_dns_domain = os.environ.get("USERDNSDOMAIN") + if user_dns_domain and user_dns_domain.lower() != hostname.lower(): + return f"{hostname}.{user_dns_domain.lower()}" + + try: + host_ip = socket.gethostbyname(hostname) + canonical_name = socket.gethostbyaddr(host_ip)[0] + if canonical_name and "." in canonical_name and not canonical_name.startswith("localhost"): + return canonical_name + except Exception: + pass + + return hostname + + +def get_local_windows_logs(hours: int = 24, state: Optional[dict] = None) -> list: + """Collects local Windows Application and System event logs.""" + if win32evtlog is None: + return [] + + logs = [] + cutoff_time = datetime.now() - timedelta(hours=hours) + machine_id = get_machine_identifier() + + last_record_number = 0 + sent_record_ids = set() + if state: + last_record_number = int(state.get("last_record_number", 0)) + sent_record_ids = set(state.get("sent_record_ids", [])) + + sev_map = { + 1: "ERROR", + 2: "WARNING", + 4: "INFO" + } + + newest_record_number = 0 + collected_record_ids = [] + + for log_type in ["Application", "System"]: + flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ + try: + hand = win32evtlog.OpenEventLog("localhost", log_type) + except Exception: + continue + + while True: + try: + events = win32evtlog.ReadEventLog(hand, flags, 0) + except Exception: + break + if not events: + break + + for event in events: + rec_num = int(event.RecordNumber) + if newest_record_number == 0: + newest_record_number = rec_num + + if event.TimeGenerated < cutoff_time: + break + + if last_record_number > 0 and newest_record_number >= last_record_number: + if rec_num <= last_record_number: + break + + rec_id = f"{log_type}:{rec_num}:{event.TimeGenerated.isoformat()}" + if rec_id in sent_record_ids: + continue + + if event.EventType in sev_map: + msg = " ".join(event.StringInserts) if event.StringInserts else "Event Log Entry" + logs.append({ + "server": machine_id, + "os_type": "windows", + "signature": f"{log_type}/{event.SourceName or 'Windows-Event'}", + "severity": sev_map[event.EventType], + "message": msg[:2048] + }) + collected_record_ids.append(rec_id) + + if events[-1].TimeGenerated < cutoff_time: + break + if last_record_number > 0 and newest_record_number >= last_record_number and events[-1].RecordNumber <= last_record_number: + break + + try: + win32evtlog.CloseEventLog(hand) + except Exception: + pass + + if state is not None: + target_rec = max(newest_record_number, last_record_number) + state["last_record_number"] = target_rec + state["sent_record_ids"] = (list(sent_record_ids) + collected_record_ids)[-1000:] + state["last_run_timestamp"] = datetime.now(timezone.utc).isoformat() + + return logs + + +def get_local_linux_logs(hours: int = 24, state: Optional[dict] = None) -> list: + """Collects local Linux journalctl events.""" + last_cursor = None + last_timestamp_us = 0.0 + sent_cursors = set() + if state: + last_cursor = state.get("last_cursor") + try: + last_timestamp_us = float(state.get("last_timestamp_us", 0)) + except (ValueError, TypeError): + last_timestamp_us = 0.0 + sent_cursors = set(state.get("sent_cursors", [])) + + cmd = ["journalctl", "--since", f"{hours} hours ago", "-p", "info", "--output=json"] + result = None + + if last_cursor: + cmd_with_cursor = ["journalctl", "--since", f"{hours} hours ago", "--after-cursor", str(last_cursor), "-p", "info", "--output=json"] + try: + res = subprocess.run(cmd_with_cursor, capture_output=True, text=True, check=False) + if res.returncode == 0: + result = res + except Exception: + pass + + if result is None: + try: + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + except Exception: + return [] + + if not result or result.returncode != 0: + return [] + + logs = [] + machine_id = get_machine_identifier() + cutoff_epoch_us = (datetime.now(timezone.utc) - timedelta(hours=hours)).timestamp() * 1_000_000 + + newest_cursor = None + newest_timestamp_us = last_timestamp_us + collected_cursors = [] + + for line in result.stdout.splitlines(): + line_str = line.strip() + if not line_str: + continue + try: + entry = json.loads(line_str) + entry_cursor = entry.get("__CURSOR") + entry_ts_us_raw = entry.get("__REALTIME_TIMESTAMP") + + entry_ts_us = 0.0 + if entry_ts_us_raw: + try: + entry_ts_us = float(entry_ts_us_raw) + except (ValueError, TypeError): + pass + + if entry_ts_us and entry_ts_us < cutoff_epoch_us: + continue + + if entry_cursor and (entry_cursor in sent_cursors or entry_cursor == last_cursor): + continue + if last_timestamp_us > 0 and entry_ts_us > 0 and entry_ts_us < last_timestamp_us: + continue + + if entry_cursor: + newest_cursor = entry_cursor + collected_cursors.append(entry_cursor) + if entry_ts_us > newest_timestamp_us: + newest_timestamp_us = entry_ts_us + + priority = int(entry.get("PRIORITY", "6")) + if priority > 6: + continue + + if priority <= 3: + sev = "ERROR" + elif priority in (4, 5): + sev = "WARNING" + else: + sev = "INFO" + + logs.append({ + "server": machine_id, + "os_type": "linux", + "signature": entry.get("SYSLOG_IDENTIFIER", "unknown"), + "severity": sev, + "message": entry.get("MESSAGE", "")[:2048] + }) + except Exception: + continue + + if state is not None: + state["last_cursor"] = newest_cursor or last_cursor + state["last_timestamp_us"] = max(newest_timestamp_us, last_timestamp_us) + state["sent_cursors"] = (list(sent_cursors) + collected_cursors)[-1000:] + state["last_run_timestamp"] = datetime.now(timezone.utc).isoformat() + + return logs + + +def get_local_system_logs(hours: int = 24, state: Optional[dict] = None) -> list: + """Platform-agnostic local host log collector.""" + if sys.platform == "win32": + return get_local_windows_logs(hours=hours, state=state) + elif sys.platform.startswith("linux"): + return get_local_linux_logs(hours=hours, state=state) + return [] + + +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. + Does not require a license seat, certificate enrollment, or mTLS socket connection. + """ + 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") + + # 1. Ensure local node is registered in clients table with LOCAL-HOST-BUILTIN fingerprint + 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}") + + # 2. Collect local logs + logs = get_local_system_logs(hours=window_hours, state=state) + if not logs: + return 0 + + # 3. Direct ingestion without network overhead or seat checks + 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 = {} + + # Run initial cycle immediately + 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}") + + def init_mtls_server_context(cert_dir: str = "certs") -> ssl.SSLContext: """Initializes TLS 1.3 server SSLContext with client certificate requirement (mTLS).""" ca_file = os.path.join(cert_dir, "ca.crt") @@ -1043,7 +1354,7 @@ def report_seat_quota_to_authority_sync(): if os.path.exists(db_path): try: conn = sqlite3.connect(db_path) - row = conn.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'").fetchone() + row = conn.execute("SELECT COUNT(*) FROM clients WHERE status = 'active' AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'").fetchone() if row: active_seats = row[0] conn.close() @@ -1406,7 +1717,7 @@ def enroll_client(req: ClientEnrollRequest, request: Request = None): } # 3. New client: check seat limits - c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'") + 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() @@ -1491,7 +1802,7 @@ def list_clients(): } for r in rows ] - active_count = sum(1 for cl in clients if cl["status"] == "active") + 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, @@ -1510,7 +1821,7 @@ def approve_client(action: ClientActionRequest): row = c.fetchone() max_seats = row[0] if row else 10 - c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'") + 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() @@ -1535,10 +1846,12 @@ reactivate_client = approve_client @app.post("/api/clients/revoke", dependencies=[Depends(require_admin_auth)]) def revoke_client(action: ClientActionRequest): """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 = ?", (action.client_id,)) + 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() @@ -1553,10 +1866,12 @@ def revoke_client(action: ClientActionRequest): @app.post("/api/clients/remove", dependencies=[Depends(require_admin_auth)]) def delete_client(action: ClientActionRequest): """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 = ?", (action.client_id,)) + c.execute("DELETE FROM clients WHERE client_id = ? AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'", (action.client_id,)) affected = c.rowcount conn.commit() conn.close() @@ -1698,9 +2013,9 @@ def get_license_summary() -> Dict[str, Any]: max_seats = row[0] if len(row) > 1 and row[1]: license_status_db = row[1] - c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'") + c.execute("SELECT COUNT(*) FROM clients WHERE status = 'active' AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'") active_seats = c.fetchone()[0] - c.execute("SELECT COUNT(*) FROM clients WHERE status = 'revoked'") + c.execute("SELECT COUNT(*) FROM clients WHERE status = 'revoked' AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'") revoked_seats = c.fetchone()[0] conn.close() except Exception: @@ -1888,7 +2203,7 @@ async def trigger_license_renewal(request: Request): try: conn_cnt = sqlite3.connect(db_path) c_cnt = conn_cnt.cursor() - c_cnt.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'") + c_cnt.execute("SELECT COUNT(*) FROM clients WHERE status = 'active' AND cert_fingerprint != 'LOCAL-HOST-BUILTIN'") row = c_cnt.fetchone() if row: active_seats = row[0] @@ -1989,12 +2304,20 @@ def serve_webui(request: Request = None): rel_time = format_relative_time(last) last_display = f"
{html.escape(str(last[:19]))}
{html.escape(rel_time)}
" - btn_action = ( - f'' - if st == "active" else - f'' - ) - btn_del = f'' + if fp == "LOCAL-HOST-BUILTIN": + esc_st = "LOCAL HUB HOST" + badge_cls = "badge-active" + fp_display = '⚡ Built-in Log Collector' + btn_action = 'Integrated (0 Seats)' + btn_del = '' + else: + fp_display = f"{esc_fp[:16]}..." + btn_action = ( + f'' + if st == "active" else + f'' + ) + btn_del = f'' client_rows.append(f""" @@ -2002,7 +2325,7 @@ def serve_webui(request: Request = None): {esc_host} {esc_cid} {esc_ost.upper()} - {esc_fp[:16]}... + {fp_display} {last_display}
@@ -2852,19 +3175,23 @@ async def run_server(): 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)) try: await asyncio.gather( tcp_server.serve_forever(), uv_server.serve(), - watchdog_task + watchdog_task, + collector_task ) finally: watchdog_task.cancel() - try: - await watchdog_task - except asyncio.CancelledError: - pass + collector_task.cancel() + for t in [watchdog_task, collector_task]: + try: + await t + except asyncio.CancelledError: + pass def main(): @@ -2951,7 +3278,7 @@ def main(): if os.path.exists(config.get("db_path", "server_logs.db")): try: conn_tmp = sqlite3.connect(config.get("db_path", "server_logs.db")) - row = conn_tmp.execute("SELECT COUNT(*) FROM clients WHERE status = 'active'").fetchone() + 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() diff --git a/tests/test_server.py b/tests/test_server.py index 28ce0bd..3b9a45a 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -435,10 +435,43 @@ class TestServerComponent(unittest.TestCase): self.assertEqual(Server.format_relative_time(None), "Never") self.assertEqual(Server.format_relative_time(""), "Never") self.assertEqual(Server.format_relative_time("invalid-date-string"), "invalid-date-str") - just_now = now.strftime("%Y-%m-%d %H:%M:%S") - self.assertIn(Server.format_relative_time(just_now), ("0s ago", "1s ago", "Just now")) + now_fresh = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + rel = Server.format_relative_time(now_fresh) + self.assertTrue(rel.endswith("s ago") or rel == "Just now", f"Unexpected relative time: {rel}") + + def test_integrated_local_log_collector_zero_seats(self): + """Tests that local hub log collection runs without license, registering a 0-seat built-in client.""" + Server.init_db(self.test_db, enrollment_secret="test-sec-456", max_seats=0) + Server.SERVER_STATE["config"] = { + "db_path": self.test_db, + "evaluation_window_hours": 12, + "min_persistence_runs": 1 + } + + # Run cycle + state = {} + processed = Server.run_local_log_collection_cycle(self.test_db, window_hours=12, min_runs=1, state=state) + + # Verify local hub client entry exists in clients table + machine_id = Server.get_machine_identifier() + conn = sqlite3.connect(self.test_db) + c = conn.cursor() + c.execute("SELECT client_id, cert_fingerprint, status FROM clients WHERE client_id = ?", (machine_id,)) + row = c.fetchone() + conn.close() + + self.assertIsNotNone(row) + self.assertEqual(row[0], machine_id) + self.assertEqual(row[1], "LOCAL-HOST-BUILTIN") + self.assertEqual(row[2], "active") + + # Verify active seat count is STILL 0 (exempt from seat accounting) + client_data = Server.list_clients() + self.assertEqual(client_data["active_seats"], 0) + self.assertEqual(len(client_data["clients"]), 1) if __name__ == "__main__": unittest.main() +