543 lines
20 KiB
Python
543 lines
20 KiB
Python
"""
|
|
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)
|