windows installer fixes

src/log_collector.py

@@ -0,0 +1,542 @@
+"""
+LOGAR - Unified Cross-Platform Log Collector & State Management
+Provides modular log collection classes and state management for Windows and Linux,
+reusable across Client Agents (Win_Client, Linux_Client) and Server Hubs (Windows & Linux).
+"""
+
+import os
+import sys
+import json
+import socket
+import subprocess
+import warnings
+from abc import ABC, abstractmethod
+from datetime import datetime, timezone, timedelta
+from typing import Optional, List, Dict, Any, Set
+
+# Suppress cryptography / pgpy deprecation notices
+warnings.filterwarnings("ignore")
+
+try:
+    import win32evtlog
+except ImportError:
+    win32evtlog = None
+
+CONFIG_FILE_NAME = "client_config.json"
+STATE_FILE_NAME = "client_state.json"
+
+
+def get_machine_identifier() -> str:
+    """
+    Returns the fully qualified hostname of the machine sending/collecting logs.
+    Cross-platform resolution:
+      1. FQDN (if non-local)
+      2. Windows USERDNSDOMAIN (on Windows)
+      3. Linux /etc/resolv.conf domain/search directive (on Linux)
+      4. Reverse DNS lookup
+      5. socket.gethostname() fallback
+    """
+    fqdn = socket.getfqdn()
+    if fqdn and "." in fqdn and not fqdn.startswith("localhost"):
+        return fqdn
+
+    hostname = socket.gethostname()
+
+    # Windows DNS domain
+    if sys.platform == "win32":
+        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()}"
+
+    # Linux resolv.conf search / domain
+    if sys.platform.startswith("linux") or os.path.exists("/etc/resolv.conf"):
+        try:
+            with open("/etc/resolv.conf", "r", encoding="utf-8") as f:
+                for line in f:
+                    parts = line.strip().split()
+                    if parts and parts[0] in ["domain", "search"] and len(parts) > 1:
+                        domain = parts[1]
+                        if domain and not domain.startswith("."):
+                            return f"{hostname}.{domain}"
+        except Exception:
+            pass
+
+    # Reverse DNS lookup
+    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 is_path_writable(filepath: str) -> bool:
+    """Checks if a file or directory path is actually writable by the current process."""
+    try:
+        if os.path.exists(filepath):
+            if os.path.isdir(filepath):
+                test_f = os.path.join(filepath, f".perm_{os.getpid()}")
+                with open(test_f, "w") as f:
+                    f.write("1")
+                os.remove(test_f)
+                return True
+            else:
+                with open(filepath, "a") as f:
+                    pass
+                return True
+        else:
+            parent = os.path.dirname(os.path.abspath(filepath)) or "."
+            os.makedirs(parent, exist_ok=True)
+            test_f = os.path.join(parent, f".perm_{os.getpid()}")
+            with open(test_f, "w") as f:
+                f.write("1")
+            os.remove(test_f)
+            return True
+    except Exception:
+        return False
+
+
+def get_default_data_dir(app_name: str = "LOGAR") -> str:
+    """Returns a suitable writable application data directory."""
+    if sys.platform == "win32":
+        # 1. ProgramData (C:\ProgramData\LOGAR)
+        prog_data = os.environ.get("ProgramData", r"C:\ProgramData")
+        target_prog = os.path.join(prog_data, app_name)
+        if is_path_writable(target_prog):
+            return target_prog
+
+        # 2. LocalAppData (C:\Users\<user>\AppData\Local\LOGAR)
+        local_app = os.environ.get("LOCALAPPDATA") or os.path.expanduser(r"~\AppData\Local")
+        target_local = os.path.join(local_app, app_name)
+        if is_path_writable(target_local):
+            return target_local
+
+        # 3. Temp directory (%TEMP%\LOGAR)
+        temp_dir = os.path.join(os.environ.get("TEMP", r"C:\Windows\Temp"), app_name)
+        if is_path_writable(temp_dir):
+            return temp_dir
+    else:
+        # Linux / Unix candidates: /var/lib/<app_name>, /tmp/<app_name>, ~/.local/share/<app_name>
+        var_lib = f"/var/lib/{app_name.lower()}"
+        if is_path_writable(var_lib):
+            return var_lib
+        user_share = os.path.expanduser(f"~/.local/share/{app_name.lower()}")
+        if is_path_writable(user_share):
+            return user_share
+
+    cwd = os.getcwd()
+    if is_path_writable(cwd):
+        return cwd
+    return os.path.expanduser("~")
+
+
+def resolve_writable_path(path: str, app_name: str = "LOGAR", default_filename: str = "client_state.json") -> str:
+    """Resolves a file or directory path, ensuring it is writable or falling back to a writable data directory."""
+    if not path:
+        path = default_filename
+
+    if os.path.isabs(path) and is_path_writable(path):
+        return path
+
+    if not os.path.isabs(path) and "Program Files" not in os.getcwd() and is_path_writable(os.getcwd()):
+        return os.path.abspath(path)
+
+    data_dir = get_default_data_dir(app_name)
+    fname = os.path.basename(path)
+    return os.path.join(data_dir, fname)
+
+
+def get_state_path(config_path: str, custom_state_path: Optional[str] = None) -> str:
+    if custom_state_path:
+        return custom_state_path
+    config_dir = os.path.dirname(os.path.abspath(config_path))
+    if sys.platform == "win32" and ("Program Files" in config_dir or not is_path_writable(config_dir)):
+        return os.path.join(get_default_data_dir("LOGAR"), STATE_FILE_NAME)
+    return os.path.join(config_dir, STATE_FILE_NAME)
+
+
+def load_state(state_path: str) -> dict:
+    """Loads state tracking dictionary from disk."""
+    if os.path.exists(state_path):
+        try:
+            with open(state_path, "r", encoding="utf-8") as f:
+                return json.load(f)
+        except Exception as e:
+            print(f"[!] Warning: Failed to read state file '{state_path}': {e}")
+            return {}
+    return {}
+
+
+def save_state(state_path: str, state: dict):
+    """Atomically saves state dictionary to disk."""
+    try:
+        os.makedirs(os.path.dirname(os.path.abspath(state_path)), exist_ok=True)
+        temp_path = f"{state_path}.tmp"
+        with open(temp_path, "w", encoding="utf-8") as f:
+            json.dump(state, f, indent=2)
+        os.replace(temp_path, state_path)
+    except Exception as e:
+        print(f"[!] Warning: Could not save client state to '{state_path}': {e}")
+
+
+def commit_state(state: dict, state_path: str):
+    """
+    Commits staged cursor/record state keys to persistent fields and writes to state_path.
+    Supports both Windows record numbers and Linux cursors/timestamps.
+    """
+    if "new_last_record_number" in state:
+        val = state.pop("new_last_record_number")
+        if val:
+            state["last_record_number"] = val
+    if "new_sent_record_ids" in state:
+        state["sent_record_ids"] = state.pop("new_sent_record_ids")
+    if "new_last_cursor" in state:
+        val = state.pop("new_last_cursor")
+        if val:
+            state["last_cursor"] = val
+    if "new_last_timestamp_us" in state:
+        val = state.pop("new_last_timestamp_us")
+        if val:
+            state["last_timestamp_us"] = val
+    if "new_sent_cursors" in state:
+        state["sent_cursors"] = state.pop("new_sent_cursors")
+    save_state(state_path, state)
+
+
+class BaseLogCollector(ABC):
+    """
+    Abstract base class for platform-specific edge and hub log collectors.
+    Manages state persistence, machine identifiers, lookback windows, and normalization.
+    """
+
+    def __init__(self, state: Optional[dict] = None, machine_id: Optional[str] = None):
+        self.state = state if state is not None else {}
+        self.machine_id = machine_id or get_machine_identifier()
+
+    @abstractmethod
+    def collect_logs(self, hours: int = 24) -> List[Dict[str, Any]]:
+        """
+        Collects normalized logs within the given lookback window.
+        Returns a list of dicts:
+        [
+            {
+                "server": str,
+                "os_type": "windows"|"linux",
+                "signature": str,
+                "severity": "ERROR"|"WARNING"|"INFO",
+                "message": str
+            },
+            ...
+        ]
+        """
+        pass
+
+    def commit(self, state_path: Optional[str] = None):
+        """Commits staged state keys and optionally writes to disk."""
+        if state_path:
+            commit_state(self.state, state_path)
+        else:
+            if "new_last_record_number" in self.state:
+                val = self.state.pop("new_last_record_number")
+                if val:
+                    self.state["last_record_number"] = val
+            if "new_sent_record_ids" in self.state:
+                self.state["sent_record_ids"] = self.state.pop("new_sent_record_ids")
+            if "new_last_cursor" in self.state:
+                val = self.state.pop("new_last_cursor")
+                if val:
+                    self.state["last_cursor"] = val
+            if "new_last_timestamp_us" in self.state:
+                val = self.state.pop("new_last_timestamp_us")
+                if val:
+                    self.state["last_timestamp_us"] = val
+            if "new_sent_cursors" in self.state:
+                self.state["sent_cursors"] = self.state.pop("new_sent_cursors")
+
+
+class WindowsLogCollector(BaseLogCollector):
+    """
+    Windows Event Log Collector.
+    Reads Application and System (or custom channels) backwards.
+    Edge Filtering: Retains INFO, WARNING, and ERROR. Drops Audit and Debug noise.
+    State Tracking: Skips events older than lookback window and already sent records.
+    """
+
+    def __init__(
+        self,
+        state: Optional[dict] = None,
+        machine_id: Optional[str] = None,
+        channels: Optional[List[str]] = None
+    ):
+        super().__init__(state=state, machine_id=machine_id)
+        self.channels = channels or ["Application"]
+
+    def collect_logs(self, hours: int = 24) -> List[Dict[str, Any]]:
+        if win32evtlog is None:
+            return []
+
+        logs = []
+        cutoff_time = datetime.now() - timedelta(hours=hours)
+
+        last_record_number = 0
+        sent_record_ids: Set[str] = set()
+        if self.state:
+            last_record_number = int(self.state.get("last_record_number", 0))
+            sent_record_ids = set(self.state.get("sent_record_ids", []))
+
+        # Windows Event Log EventTypes:
+        # 1: EVENTLOG_ERROR_TYPE -> ERROR
+        # 2: EVENTLOG_WARNING_TYPE -> WARNING
+        # 4: EVENTLOG_INFORMATION_TYPE -> INFO
+        sev_map = {
+            1: "ERROR",
+            2: "WARNING",
+            4: "INFO"
+        }
+
+        newest_record_number = 0
+        collected_record_ids = []
+
+        flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ
+
+        for log_type in self.channels:
+            try:
+                hand = win32evtlog.OpenEventLog("localhost", log_type)
+            except Exception as e:
+                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
+
+                    # 1. Skip entries older than lookback window
+                    if event.TimeGenerated < cutoff_time:
+                        break
+
+                    # 2. Skip already sent events
+                    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 len(self.channels) > 1 else f"{rec_num}:{event.TimeGenerated.isoformat()}"
+                    if rec_id in sent_record_ids:
+                        continue
+
+                    # Filter: upload everything from INFO to ERROR
+                    if event.EventType in sev_map:
+                        msg = " ".join(event.StringInserts) if event.StringInserts else "Event Log Entry"
+                        sig = f"{log_type}/{event.SourceName or 'Windows-Event'}" if len(self.channels) > 1 else (event.SourceName or "Windows-Event")
+                        logs.append({
+                            "server": self.machine_id,
+                            "os_type": "windows",
+                            "signature": sig,
+                            "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 self.state is not None:
+            target_rec = max(newest_record_number, last_record_number)
+            self.state["new_last_record_number"] = target_rec
+            self.state["new_sent_record_ids"] = (list(sent_record_ids) + collected_record_ids)[-1000:]
+            # Also update direct keys for immediate server-side ingestion
+            self.state["last_record_number"] = target_rec
+            self.state["sent_record_ids"] = self.state["new_sent_record_ids"]
+            self.state["last_run_timestamp"] = datetime.now(timezone.utc).isoformat()
+
+        return logs
+
+
+class LinuxLogCollector(BaseLogCollector):
+    """
+    Linux Systemd Journal / Syslog Log Collector.
+    Collects info, warnings, and errors from journalctl (or syslog fallback).
+    Edge Filtering: Retains INFO, WARNING, and ERROR. Drops DEBUG noise.
+    State Tracking: Skips events older than lookback window and already sent cursors.
+    """
+
+    def __init__(self, state: Optional[dict] = None, machine_id: Optional[str] = None):
+        super().__init__(state=state, machine_id=machine_id)
+
+    def collect_logs(self, hours: int = 24) -> List[Dict[str, Any]]:
+        last_cursor = None
+        last_timestamp_us = 0.0
+        sent_cursors: Set[str] = set()
+        if self.state:
+            last_cursor = self.state.get("last_cursor")
+            try:
+                last_timestamp_us = float(self.state.get("last_timestamp_us", 0))
+            except (ValueError, TypeError):
+                last_timestamp_us = 0.0
+            sent_cursors = set(self.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 FileNotFoundError:
+                # journalctl not present, attempt fallback to /var/log/syslog or messages
+                return self._collect_syslog_fallback(hours=hours)
+            except Exception as e:
+                print(f"[!] Error running journalctl: {e}")
+                return []
+
+        if not result or result.returncode != 0:
+            return []
+
+        logs = []
+        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
+
+                # 1. Skip entries older than lookback window
+                if entry_ts_us and entry_ts_us < cutoff_epoch_us:
+                    continue
+
+                # 2. Skip already sent events
+                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
+
+                # Advance tracking
+                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": self.machine_id,
+                    "os_type": "linux",
+                    "signature": entry.get("SYSLOG_IDENTIFIER", "unknown"),
+                    "severity": sev,
+                    "message": entry.get("MESSAGE", "")[:2048]
+                })
+            except (json.JSONDecodeError, ValueError):
+                continue
+
+        if self.state is not None:
+            self.state["new_last_cursor"] = newest_cursor or last_cursor
+            self.state["new_last_timestamp_us"] = max(newest_timestamp_us, last_timestamp_us)
+            self.state["new_sent_cursors"] = (list(sent_cursors) + collected_cursors)[-1000:]
+            # Also update direct keys for immediate server-side ingestion
+            self.state["last_cursor"] = self.state["new_last_cursor"]
+            self.state["last_timestamp_us"] = self.state["new_last_timestamp_us"]
+            self.state["sent_cursors"] = self.state["new_sent_cursors"]
+            self.state["last_run_timestamp"] = datetime.now(timezone.utc).isoformat()
+
+        return logs
+
+    def _collect_syslog_fallback(self, hours: int = 24) -> List[Dict[str, Any]]:
+        """Fallback log reader for non-systemd Linux environments."""
+        candidates = ["/var/log/syslog", "/var/log/messages"]
+        target = next((f for f in candidates if os.path.exists(f) and os.access(f, os.R_OK)), None)
+        if not target:
+            return []
+
+        logs = []
+        try:
+            with open(target, "r", encoding="utf-8", errors="ignore") as f:
+                lines = f.readlines()[-500:]  # Read recent 500 lines
+            for line in lines:
+                line_str = line.strip()
+                if not line_str:
+                    continue
+                sev = "INFO"
+                lower_l = line_str.lower()
+                if any(w in lower_l for w in ["err", "crit", "alert", "emerg", "fail"]):
+                    sev = "ERROR"
+                elif any(w in lower_l for w in ["warn", "notice"]):
+                    sev = "WARNING"
+
+                logs.append({
+                    "server": self.machine_id,
+                    "os_type": "linux",
+                    "signature": "syslog",
+                    "severity": sev,
+                    "message": line_str[:2048]
+                })
+        except Exception:
+            pass
+        return logs
+
+
+def get_log_collector(
+    os_type: Optional[str] = None,
+    state: Optional[dict] = None,
+    machine_id: Optional[str] = None,
+    channels: Optional[List[str]] = None
+) -> BaseLogCollector:
+    """
+    Factory helper that returns the appropriate BaseLogCollector instance.
+    If os_type is omitted, auto-detects based on sys.platform.
+    """
+    target_os = os_type.lower() if os_type else ("windows" if sys.platform == "win32" else "linux")
+    if target_os == "windows":
+        chans = channels or (["Application", "System"] if channels is None else ["Application"])
+        return WindowsLogCollector(state=state, machine_id=machine_id, channels=chans)
+    else:
+        return LinuxLogCollector(state=state, machine_id=machine_id)