windows installer fixes

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"<div><strong>{html.escape(str(last[:19]))}</strong></div><div style='font-size:11px; color:#94a3b8;'>{html.escape(rel_time)}</div>"
 
-        btn_action = (
-            f'<button class="btn btn-outline" style="padding:4px 8px; font-size:11px;" onclick="revokeClient(\'{esc_cid}\')">Revoke</button>'
-            if st == "active" else
-            f'<button class="btn btn-accent" style="padding:4px 8px; font-size:11px;" onclick="approveClient(\'{esc_cid}\')">Approve</button>'
-        )
-        btn_del = f'<button class="btn btn-danger" style="padding:4px 8px; font-size:11px;" onclick="deleteClient(\'{esc_cid}\')" title="Remove client and free seat">Remove</button>'
+        if fp == "LOCAL-HOST-BUILTIN":
+            esc_st = "LOCAL HUB HOST"
+            badge_cls = "badge-active"
+            fp_display = '<span style="font-size:11px; color:#38bdf8; font-weight:600;">⚡ Built-in Log Collector</span>'
+            btn_action = '<span style="font-size:11px; color:#94a3b8;">Integrated (0 Seats)</span>'
+            btn_del = ''
+        else:
+            fp_display = f"<code title='{esc_fp}' style='cursor:pointer;' onclick=\"copyText('{esc_fp}', 'Cert Fingerprint')\">{esc_fp[:16]}...</code>"
+            btn_action = (
+                f'<button class="btn btn-outline" style="padding:4px 8px; font-size:11px;" onclick="revokeClient(\'{esc_cid}\')">Revoke</button>'
+                if st == "active" else
+                f'<button class="btn btn-accent" style="padding:4px 8px; font-size:11px;" onclick="approveClient(\'{esc_cid}\')">Approve</button>'
+            )
+            btn_del = f'<button class="btn btn-danger" style="padding:4px 8px; font-size:11px;" onclick="deleteClient(\'{esc_cid}\')" title="Remove client and free seat">Remove</button>'
 
         client_rows.append(f"""
             <tr>
@@ -2002,7 +2325,7 @@ def serve_webui(request: Request = None):
                 <td><strong>{esc_host}</strong></td>
                 <td><code>{esc_cid}</code></td>
                 <td><span class="badge badge-os">{esc_ost.upper()}</span></td>
-                <td><code title="{esc_fp}" style="cursor:pointer;" onclick="copyText('{esc_fp}', 'Cert Fingerprint')">{esc_fp[:16]}...</code></td>
+                <td>{fp_display}</td>
                 <td>{last_display}</td>
                 <td>
                     <div style="display:flex; gap:6px; align-items:center;">
@@ -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()

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()
 
+