diff --git a/src/Linux_Client.py b/src/Linux_Client.py index 2cd030d..c6528d6 100644 --- a/src/Linux_Client.py +++ b/src/Linux_Client.py @@ -1,15 +1,14 @@ +""" +LOGAR Linux Edge Log Forwarder +Collects Linux systemd journalctl / syslog entries and securely streams them +to the central LOGAR Hub via mutual TLS (mTLS 1.3) or OpenPGP envelope. +""" + import os import sys -import json import time -import socket -import ssl -import struct import argparse -import subprocess -import urllib.request import warnings -from datetime import datetime, timezone, timedelta from typing import Optional # Ensure script directory and repo root are in sys.path @@ -25,470 +24,49 @@ warnings.filterwarnings("ignore") import pgpy +from log_collector import ( + get_machine_identifier, + is_path_writable, + get_default_data_dir, + resolve_writable_path, + get_state_path, + load_state, + save_state, + commit_state, + LinuxLogCollector, + BaseLogCollector +) +from client_transport import ( + is_cert_expiring_soon, + enroll_client_if_needed, + get_tls_socket, + load_config, + ensure_enrolled as transport_ensure_enrolled, + send_encrypted_logs_over_socket as transport_send_encrypted_logs +) + CONFIG_FILE_NAME = "client_config.json" STATE_FILE_NAME = "client_state.json" -def is_cert_expiring_soon(cert_path: str, threshold_days: int = 30) -> bool: - """Checks if client certificate at cert_path is expiring within threshold_days.""" - if not os.path.exists(cert_path): - return True - try: - from cryptography import x509 - with open(cert_path, "r", encoding="utf-8") as f: - cert = x509.load_pem_x509_certificate(f.read().encode("utf-8")) - expiry = getattr(cert, "not_valid_after_utc", None) - if expiry is None: - expiry = cert.not_valid_after.replace(tzinfo=timezone.utc) - now = datetime.now(timezone.utc) - return expiry <= (now + timedelta(days=threshold_days)) - except Exception: - return True - - -def enroll_client_if_needed( - hub_url: str, - enrollment_secret: str, - cert_dir: str, - client_id: str, - hostname: str, - os_type: str = "linux", - force_renew: bool = False, - threshold_days: int = 30 -): - """Bootstraps client enrollment if certificates are missing or expiring soon.""" - os.makedirs(cert_dir, exist_ok=True) - ca_path = os.path.join(cert_dir, "ca.crt") - cert_path = os.path.join(cert_dir, "client.crt") - key_path = os.path.join(cert_dir, "client.key") - - if not force_renew and os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path): - if not is_cert_expiring_soon(cert_path, threshold_days=threshold_days): - return True - print(f"[*] Client certificate at {cert_path} is expiring within {threshold_days} days. Auto-renewing...") - - action_name = "re-enrolling" if os.path.exists(cert_path) else "enrolling" - print(f"[*] Bootstrapping client {action_name} with LOGAR Hub at {hub_url}...") - enroll_endpoint = f"{hub_url.rstrip('/')}/api/client/enroll" - payload = { - "client_id": client_id, - "hostname": hostname, - "os": os_type, - "enrollment_secret": enrollment_secret - } - req = urllib.request.Request( - enroll_endpoint, - data=json.dumps(payload).encode("utf-8"), - headers={"Content-Type": "application/json"} - ) - with urllib.request.urlopen(req, timeout=10) as resp: - if resp.status != 200: - raise RuntimeError(f"Enrollment failed with status code {resp.status}") - data = json.loads(resp.read().decode("utf-8")) - - with open(ca_path, "w", encoding="utf-8") as f: - f.write(data["ca_cert"]) - with open(cert_path, "w", encoding="utf-8") as f: - f.write(data["client_cert"]) - with open(key_path, "w", encoding="utf-8") as f: - f.write(data["client_key"]) - - try: - os.chmod(key_path, 0o600) - except Exception: - pass - - print(f"[+] Client certificates updated successfully in {os.path.abspath(cert_dir)}") - return True - - -def get_tls_socket(hub_host: str, hub_port: int, cert_dir: str): - """Establishes an mTLS connection with the LOGAR hub using client certificates.""" - ca_path = os.path.join(cert_dir, "ca.crt") - cert_path = os.path.join(cert_dir, "client.crt") - key_path = os.path.join(cert_dir, "client.key") - - if not (os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path)): - raise FileNotFoundError(f"mTLS certificates not found in '{cert_dir}'. Enroll client first.") - - ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca_path) - ctx.load_cert_chain(certfile=cert_path, keyfile=key_path) - ctx.minimum_version = ssl.TLSVersion.TLSv1_3 - ctx.check_hostname = False - - raw_sock = socket.create_connection((hub_host, hub_port), timeout=15) - return ctx.wrap_socket(raw_sock, server_hostname=hub_host) - - -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)) - return os.path.join(config_dir, STATE_FILE_NAME) - - -def load_state(state_path: str) -> dict: - 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): - try: - 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): - 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) - - -def load_config(config_path: str = CONFIG_FILE_NAME): - if not os.path.exists(config_path): - raise FileNotFoundError( - f"Client configuration file not found at: {config_path}\n" - f"Generate one from the server using: python Server.py --create-client-config --client-out {config_path}" - ) - with open(config_path, "r", encoding="utf-8") as f: - config = json.load(f) - - cfg_dir = os.path.dirname(os.path.abspath(config_path)) - raw_cert_dir = config.get("cert_dir", "certs") - if not os.path.isabs(raw_cert_dir): - config["cert_dir"] = os.path.join(cfg_dir, raw_cert_dir) - - ca_cert_pem = config.get("ca_cert") - if ca_cert_pem and isinstance(ca_cert_pem, str): - ca_file = os.path.join(config["cert_dir"], "ca.crt") - if not os.path.exists(ca_file): - try: - os.makedirs(config["cert_dir"], exist_ok=True) - with open(ca_file, "w", encoding="utf-8") as f: - f.write(ca_cert_pem) - except Exception: - pass - - return config - - -def get_machine_identifier() -> str: - """ - Returns the hostname of the machine sending the logs, - and appends the network/DNS domain if available. - """ - # 1. Try fully-qualified domain name (FQDN) - fqdn = socket.getfqdn() - if fqdn and "." in fqdn and not fqdn.startswith("localhost"): - return fqdn - - hostname = socket.gethostname() - - # 2. Check /etc/resolv.conf domain or search directive - try: - if os.path.exists("/etc/resolv.conf"): - 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 - - # 3. Try 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 get_recent_linux_logs(hours: int = 24, state: Optional[dict] = None) -> list: """ Collects info, warnings, and errors from systemd journalctl over the lookback window. - Edge Filtering: Retains INFO, WARNING, and ERROR. Strips DEBUG (priority 7) and skips events older than lookback window (default: 24h). - State Tracking: Skips events older than lookback window (default 24h) and events - that have already been sent in previous runs. + Edge Filtering: Retains INFO, WARNING, and ERROR. Strips DEBUG noise. + State Tracking: Skips events older than lookback window and already sent cursors. """ - 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 FileNotFoundError: - print("[!] journalctl command not found. Ensure this script runs on a systemd-enabled Linux system.") - return [] - except Exception: - pass - - if result is None: - try: - result = subprocess.run(cmd, capture_output=True, text=True, check=False) - except FileNotFoundError: - print("[!] journalctl command not found. Ensure this script runs on a systemd-enabled Linux system.") - return [] - except Exception as e: - print(f"[!] Error running journalctl: {e}") - 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 - - # 1. Skip entries older than lookback window (default: 24h) - 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 newest tracking for new entries - 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")) - # Priority 0: Emerg, 1: Alert, 2: Crit, 3: Err (-> ERROR) - # Priority 4: Warning, 5: Notice (-> WARNING) - # Priority 6: Info (-> INFO) - # Priority 7: Debug (skip) - 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 (json.JSONDecodeError, ValueError): - continue - - if state is not None: - state["new_last_cursor"] = newest_cursor or last_cursor - state["new_last_timestamp_us"] = max(newest_timestamp_us, last_timestamp_us) - state["new_sent_cursors"] = (list(sent_cursors) + collected_cursors)[-1000:] - state["last_run_timestamp"] = datetime.now(timezone.utc).isoformat() - - return logs + collector = LinuxLogCollector(state=state) + return collector.collect_logs(hours=hours) def send_encrypted_logs_over_socket(config: dict, logs: list): - """ - Streams logs to the LOGAR hub. - Uses mutual TLS 1.3 (mTLS) with client certificates if available, - or falls back to OpenPGP encrypted envelope over TCP. - """ - server_host = config["server_host"] - server_port = int(config["server_port"]) - cert_dir = config.get("cert_dir", "certs") - enrollment_secret = config.get("enrollment_secret") - machine_id = get_machine_identifier() - - # Attempt automatic enrollment bootstrap if certs are missing and secret is provided - hub_url = None - if enrollment_secret: - hermes_host = config.get("hermes_host", server_host) - if hermes_host in ("0.0.0.0", "::", None): - hermes_host = server_host if server_host not in ("0.0.0.0", "::", None) else "127.0.0.1" - hermes_port = config.get("hermes_port", 8443) - hub_url = f"http://{hermes_host}:{hermes_port}" - try: - enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="linux") - except Exception as e: - print(f"[!] Warning: Enrollment bootstrap failed: {e}") - - ca_path = os.path.join(cert_dir, "ca.crt") - cert_path = os.path.join(cert_dir, "client.crt") - key_path = os.path.join(cert_dir, "client.key") - has_mtls_certs = os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path) - - if has_mtls_certs: - print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over mTLS (TLS 1.3)...") - sock = None - try: - sock = get_tls_socket(server_host, server_port, cert_dir) - except (ssl.SSLError, ssl.CertificateError, ConnectionResetError) as tls_err: - if enrollment_secret and hub_url: - print(f"[!] TLS handshake error ({tls_err}). Re-enrolling with LOGAR Hub...") - try: - enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="linux", force_renew=True) - sock = get_tls_socket(server_host, server_port, cert_dir) - except Exception as retry_err: - print(f"[!] Re-enrollment or reconnection retry failed: {retry_err}") - raise - else: - raise - - with sock: - payload = { - "server": machine_id, - "timestamp": datetime.now(timezone.utc).isoformat(), - "logs": logs - } - payload_bytes = json.dumps(payload).encode("utf-8") - frame = struct.pack(">I", len(payload_bytes)) + payload_bytes - sock.sendall(frame) - - resp_len_bytes = sock.recv(4) - if not resp_len_bytes: - raise ConnectionError("Server closed mTLS connection without response.") - resp_len = struct.unpack(">I", resp_len_bytes)[0] - resp_bytes = bytearray() - while len(resp_bytes) < resp_len: - chunk = sock.recv(min(4096, resp_len - len(resp_bytes))) - if not chunk: - break - resp_bytes.extend(chunk) - - response = json.loads(resp_bytes.decode("utf-8")) - print(f"[+] Server response: {response}") - return response - - # Fallback to OpenPGP envelope over plain TCP socket - auth_token = config.get("auth_token", "") - pub_key_armored = config.get("server_public_key") - if not pub_key_armored: - raise ValueError("No server public key or mTLS certificates available for connection.") - - expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper() - pub_key, _ = pgpy.PGPKey.from_blob(pub_key_armored) - actual_fp = str(pub_key.fingerprint).replace(" ", "").upper() - if expected_fp and actual_fp != expected_fp: - raise ValueError(f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}.") - - payload = { - "server": machine_id, - "timestamp": datetime.now(timezone.utc).isoformat(), - "logs": logs - } - payload_json = json.dumps(payload) - pgp_msg = pgpy.PGPMessage.new(payload_json) - encrypted_msg = pub_key.encrypt(pgp_msg) - encrypted_armored = str(encrypted_msg) - - envelope = { - "auth_token": auth_token, - "timestamp": datetime.now(timezone.utc).isoformat(), - "encrypted_payload": encrypted_armored - } - envelope_bytes = json.dumps(envelope).encode("utf-8") - - print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over secure TCP socket...") - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(15.0) - sock.connect((server_host, server_port)) - - frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes - sock.sendall(frame) - - resp_len_bytes = sock.recv(4) - if not resp_len_bytes: - raise ConnectionError("Server closed connection without response.") - - resp_len = struct.unpack(">I", resp_len_bytes)[0] - resp_bytes = bytearray() - while len(resp_bytes) < resp_len: - chunk = sock.recv(min(4096, resp_len - len(resp_bytes))) - if not chunk: - break - resp_bytes.extend(chunk) - - response = json.loads(resp_bytes.decode("utf-8")) - print(f"[+] Server response: {response}") - return response + """Streams logs to the LOGAR hub over mTLS or encrypted TCP envelope.""" + return transport_send_encrypted_logs(config=config, logs=logs, os_type="linux") def ensure_enrolled(config: dict, machine_id: str) -> bool: """Proactively ensures client certificate enrollment with the Hub if enrollment_secret is available.""" - cert_dir = config.get("cert_dir", "certs") - enrollment_secret = config.get("enrollment_secret") - if not enrollment_secret: - return False - - server_host = config.get("server_host", "127.0.0.1") - hermes_host = config.get("hermes_host", server_host) - if hermes_host in ("0.0.0.0", "::", None, ""): - hermes_host = server_host if server_host not in ("0.0.0.0", "::", None, "") else "127.0.0.1" - hermes_port = config.get("hermes_port", 8443) - hub_url = f"http://{hermes_host}:{hermes_port}" - try: - return enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="linux") - except Exception as e: - print(f"[!] Warning: Initial client enrollment bootstrap with Hub at {hub_url} failed: {e}") - return False + return transport_ensure_enrolled(config=config, machine_id=machine_id, os_type="linux") def main(): @@ -519,12 +97,14 @@ def main(): # Proactively ensure dynamic PKI certificate enrollment with Server Hub ensure_enrolled(config, machine_id) + collector = LinuxLogCollector(state=state, machine_id=machine_id) + def run_forwarder_cycle(): # Continuously verify/renew PKI enrollment if server is licensed & online ensure_enrolled(config, machine_id) print(f"[*] Scanning Linux journalctl for unsent entries (last {args.hours} hours)...") - candidate_logs = get_recent_linux_logs(hours=args.hours, state=state) + candidate_logs = collector.collect_logs(hours=args.hours) print(f"[*] Found {len(candidate_logs)} unsent candidate entries (INFO to ERROR, entries > {args.hours}h and already-sent skipped).") if not candidate_logs: diff --git a/src/Server.py b/src/Server.py index c541052..324cb61 100644 --- a/src/Server.py +++ b/src/Server.py @@ -53,6 +53,21 @@ try: except ImportError: import server_enrollment as enrollment +try: + from src.log_collector import ( + get_machine_identifier, + WindowsLogCollector, + LinuxLogCollector, + get_log_collector + ) +except ImportError: + from log_collector import ( + get_machine_identifier, + WindowsLogCollector, + LinuxLogCollector, + get_log_collector + ) + CONFIG_FILE_NAME = "server_config.json" DEFAULT_DB_FILE = "logar_state.db" EVALUATION_WINDOW_HOURS = 12 @@ -915,224 +930,22 @@ 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 + collector = WindowsLogCollector(state=state, channels=["Application", "System"]) + return collector.collect_logs(hours=hours) 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 + """Collects local Linux journalctl / syslog events.""" + collector = LinuxLogCollector(state=state) + return collector.collect_logs(hours=hours) 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 [] + collector = get_log_collector(state=state) + return collector.collect_logs(hours=hours) def run_local_log_collection_cycle(db_path: str, window_hours: int = 12, min_runs: int = 1, state: Optional[dict] = None) -> int: diff --git a/src/Win_Client.py b/src/Win_Client.py index a2a1ce1..4821275 100644 --- a/src/Win_Client.py +++ b/src/Win_Client.py @@ -1,618 +1,144 @@ -import os -import sys -import json -import time -import socket -import ssl -import struct -import argparse -import urllib.request -import warnings -from datetime import datetime, timezone, timedelta -from typing import Optional - -# Ensure script directory and repo root are in sys.path -SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) -if SCRIPT_DIR not in sys.path: - sys.path.insert(0, SCRIPT_DIR) -REPO_ROOT = os.path.abspath(os.path.join(SCRIPT_DIR, "..")) -if REPO_ROOT not in sys.path: - sys.path.insert(0, REPO_ROOT) - -# Suppress cryptography / pgpy deprecation notices -warnings.filterwarnings("ignore") - -import pgpy - -try: - import win32evtlog -except ImportError: - win32evtlog = None - -CONFIG_FILE_NAME = "client_config.json" -STATE_FILE_NAME = "client_state.json" - - -def is_cert_expiring_soon(cert_path: str, threshold_days: int = 30) -> bool: - """Checks if client certificate at cert_path is expiring within threshold_days.""" - if not os.path.exists(cert_path): - return True - try: - from cryptography import x509 - with open(cert_path, "r", encoding="utf-8") as f: - cert = x509.load_pem_x509_certificate(f.read().encode("utf-8")) - expiry = getattr(cert, "not_valid_after_utc", None) - if expiry is None: - expiry = cert.not_valid_after.replace(tzinfo=timezone.utc) - now = datetime.now(timezone.utc) - return expiry <= (now + timedelta(days=threshold_days)) - except Exception: - return True - - -def enroll_client_if_needed( - hub_url: str, - enrollment_secret: str, - cert_dir: str, - client_id: str, - hostname: str, - os_type: str = "windows", - force_renew: bool = False, - threshold_days: int = 30 -): - """Bootstraps client enrollment if certificates are missing or expiring soon.""" - os.makedirs(cert_dir, exist_ok=True) - ca_path = os.path.join(cert_dir, "ca.crt") - cert_path = os.path.join(cert_dir, "client.crt") - key_path = os.path.join(cert_dir, "client.key") - - if not force_renew and os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path): - if not is_cert_expiring_soon(cert_path, threshold_days=threshold_days): - return True - print(f"[*] Client certificate at {cert_path} is expiring within {threshold_days} days. Auto-renewing...") - - action_name = "re-enrolling" if os.path.exists(cert_path) else "enrolling" - print(f"[*] Bootstrapping client {action_name} with LOGAR Hub at {hub_url}...") - enroll_endpoint = f"{hub_url.rstrip('/')}/api/client/enroll" - payload = { - "client_id": client_id, - "hostname": hostname, - "os": os_type, - "enrollment_secret": enrollment_secret - } - req = urllib.request.Request( - enroll_endpoint, - data=json.dumps(payload).encode("utf-8"), - headers={"Content-Type": "application/json"} - ) - with urllib.request.urlopen(req, timeout=10) as resp: - if resp.status != 200: - raise RuntimeError(f"Enrollment failed with status code {resp.status}") - data = json.loads(resp.read().decode("utf-8")) - - with open(ca_path, "w", encoding="utf-8") as f: - f.write(data["ca_cert"]) - with open(cert_path, "w", encoding="utf-8") as f: - f.write(data["client_cert"]) - with open(key_path, "w", encoding="utf-8") as f: - f.write(data["client_key"]) - - try: - os.chmod(key_path, 0o600) - except Exception: - pass - - print(f"[+] Client certificates updated successfully in {os.path.abspath(cert_dir)}") - return True - - -def get_tls_socket(hub_host: str, hub_port: int, cert_dir: str): - """Establishes an mTLS connection with the LOGAR hub using client certificates.""" - ca_path = os.path.join(cert_dir, "ca.crt") - cert_path = os.path.join(cert_dir, "client.crt") - key_path = os.path.join(cert_dir, "client.key") - - if not (os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path)): - raise FileNotFoundError(f"mTLS certificates not found in '{cert_dir}'. Enroll client first.") - - ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca_path) - ctx.load_cert_chain(certfile=cert_path, keyfile=key_path) - ctx.minimum_version = ssl.TLSVersion.TLSv1_3 - ctx.check_hostname = False - - raw_sock = socket.create_connection((hub_host, hub_port), timeout=15) - return ctx.wrap_socket(raw_sock, server_hostname=hub_host) - - -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\\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 - - 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: - 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): - 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): - 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") - save_state(state_path, state) - - -def load_config(config_path: str = CONFIG_FILE_NAME): - actual_config_path = config_path - if not os.path.exists(actual_config_path): - data_dir_cfg = os.path.join(get_default_data_dir("LOGAR"), os.path.basename(config_path)) - if os.path.exists(data_dir_cfg): - actual_config_path = data_dir_cfg - - if not os.path.exists(actual_config_path): - raise FileNotFoundError( - f"Client configuration file not found at: {config_path}\n" - f"Generate one from the server using: python Server.py --create-client-config --client-out {config_path}" - ) - with open(actual_config_path, "r", encoding="utf-8") as f: - config = json.load(f) - - # Resolve cert_dir to a writable directory (defaulting next to the config file if relative) - cfg_dir = os.path.dirname(os.path.abspath(actual_config_path)) - raw_cert_dir = config.get("cert_dir", "certs") - if os.path.isabs(raw_cert_dir) and is_path_writable(raw_cert_dir): - config["cert_dir"] = raw_cert_dir - else: - candidate = os.path.join(cfg_dir, os.path.basename(raw_cert_dir)) - if is_path_writable(candidate) or is_path_writable(cfg_dir): - config["cert_dir"] = candidate - else: - config["cert_dir"] = resolve_writable_path("certs", "LOGAR", "certs") - - # If ca_cert is present in config and ca.crt doesn't exist yet, save it - ca_cert_pem = config.get("ca_cert") - if ca_cert_pem and isinstance(ca_cert_pem, str): - ca_file = os.path.join(config["cert_dir"], "ca.crt") - if not os.path.exists(ca_file): - try: - os.makedirs(config["cert_dir"], exist_ok=True) - with open(ca_file, "w", encoding="utf-8") as f: - f.write(ca_cert_pem) - except Exception: - pass - - return config - - -def get_machine_identifier() -> str: - """ - Returns the hostname of the machine sending the logs, - and appends the network/DNS domain if available. - """ - 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_recent_windows_logs(hours: int = 24, state: Optional[dict] = None) -> list: - """ - Scans the Windows Application Event Log backwards for events within the window. - Edge Filtering: Retains INFO, WARNING, and ERROR. Drops Audit and Debug noise. - State Tracking: Skips events older than lookback window (default 24h) and events - that have already been sent in previous runs. - """ - if win32evtlog is None: - print("[!] pywin32 is not installed or not running on Windows. Returning mock/empty candidate list.") - return [] - - server = "localhost" - log_type = "Application" - flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ - - try: - hand = win32evtlog.OpenEventLog(server, log_type) - except Exception as e: - print(f"[!] Error opening Windows event log: {e}") - 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", [])) - - # Windows Event Log EventTypes: - # 1: EVENTLOG_ERROR_TYPE -> ERROR - # 2: EVENTLOG_WARNING_TYPE -> WARNING - # 4: EVENTLOG_INFORMATION_TYPE -> INFO - # Excludes: 8 (Audit Success), 16 (Audit Failure), and other verbose noise - sev_map = { - 1: "ERROR", - 2: "WARNING", - 4: "INFO" - } - - newest_record_number = 0 - collected_record_ids = [] - - while True: - events = win32evtlog.ReadEventLog(hand, flags, 0) - 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 (default: 24h) - if event.TimeGenerated < cutoff_time: - break - - # 2. Skip already sent events if we've reached records <= last_record_number - # (unless the log was cleared and numbers wrapped, i.e. newest_record_number < last_record_number) - if last_record_number > 0 and newest_record_number >= last_record_number: - if rec_num <= last_record_number: - break - - rec_id = f"{rec_num}:{event.TimeGenerated.isoformat()}" - if rec_id in sent_record_ids: - continue - - # Filter: upload everything from INFO to ERROR only - 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": event.SourceName or "Windows-Event", - "severity": sev_map[event.EventType], - "message": msg[:2048] # Limit message length - }) - 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 - - win32evtlog.CloseEventLog(hand) - - if state is not None: - target_rec = max(newest_record_number, last_record_number) - state["new_last_record_number"] = target_rec - state["new_sent_record_ids"] = (list(sent_record_ids) + collected_record_ids)[-1000:] - state["last_run_timestamp"] = datetime.now(timezone.utc).isoformat() - - return logs - - -def send_encrypted_logs_over_socket(config: dict, logs: list): - """ - Streams logs to the LOGAR hub. - Uses mutual TLS 1.3 (mTLS) with client certificates if available, - or falls back to OpenPGP encrypted envelope over TCP. - """ - server_host = config["server_host"] - server_port = int(config["server_port"]) - cert_dir = config.get("cert_dir", "certs") - enrollment_secret = config.get("enrollment_secret") - machine_id = get_machine_identifier() - - # Attempt automatic enrollment bootstrap if certs are missing and secret is provided - hub_url = None - if enrollment_secret: - hermes_host = config.get("hermes_host", server_host) - if hermes_host in ("0.0.0.0", "::", None): - hermes_host = server_host if server_host not in ("0.0.0.0", "::", None) else "127.0.0.1" - hermes_port = config.get("hermes_port", 8443) - hub_url = f"http://{hermes_host}:{hermes_port}" - try: - enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="windows") - except Exception as e: - print(f"[!] Warning: Enrollment bootstrap failed: {e}") - - ca_path = os.path.join(cert_dir, "ca.crt") - cert_path = os.path.join(cert_dir, "client.crt") - key_path = os.path.join(cert_dir, "client.key") - has_mtls_certs = os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path) - - if not has_mtls_certs and enrollment_secret and hub_url: - try: - enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="windows") - has_mtls_certs = os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path) - except Exception as e: - print(f"[!] Warning: Enrollment before TLS connection failed: {e}") - - if has_mtls_certs: - print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over mTLS (TLS 1.3)...") - sock = None - try: - sock = get_tls_socket(server_host, server_port, cert_dir) - except (ssl.SSLError, ssl.CertificateError, ConnectionResetError) as tls_err: - if enrollment_secret and hub_url: - print(f"[!] TLS handshake error ({tls_err}). Re-enrolling with LOGAR Hub...") - try: - enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="windows", force_renew=True) - sock = get_tls_socket(server_host, server_port, cert_dir) - except Exception as retry_err: - print(f"[!] Re-enrollment or reconnection retry failed: {retry_err}") - raise - else: - raise - - with sock: - payload = { - "server": machine_id, - "timestamp": datetime.now(timezone.utc).isoformat(), - "logs": logs - } - payload_bytes = json.dumps(payload).encode("utf-8") - frame = struct.pack(">I", len(payload_bytes)) + payload_bytes - sock.sendall(frame) - - resp_len_bytes = sock.recv(4) - if not resp_len_bytes: - raise ConnectionError("Server closed mTLS connection without response.") - resp_len = struct.unpack(">I", resp_len_bytes)[0] - resp_bytes = bytearray() - while len(resp_bytes) < resp_len: - chunk = sock.recv(min(4096, resp_len - len(resp_bytes))) - if not chunk: - break - resp_bytes.extend(chunk) - - response = json.loads(resp_bytes.decode("utf-8")) - print(f"[+] Server response: {response}") - return response - - # Fallback to OpenPGP envelope over plain TCP socket - auth_token = config.get("auth_token", "") - pub_key_armored = config.get("server_public_key") - if not pub_key_armored: - raise ValueError("No server public key or mTLS certificates available for connection.") - - expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper() - pub_key, _ = pgpy.PGPKey.from_blob(pub_key_armored) - actual_fp = str(pub_key.fingerprint).replace(" ", "").upper() - if expected_fp and actual_fp != expected_fp: - raise ValueError(f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}.") - - payload = { - "server": machine_id, - "timestamp": datetime.now(timezone.utc).isoformat(), - "logs": logs - } - payload_json = json.dumps(payload) - pgp_msg = pgpy.PGPMessage.new(payload_json) - encrypted_msg = pub_key.encrypt(pgp_msg) - encrypted_armored = str(encrypted_msg) - - envelope = { - "auth_token": auth_token, - "timestamp": datetime.now(timezone.utc).isoformat(), - "encrypted_payload": encrypted_armored - } - envelope_bytes = json.dumps(envelope).encode("utf-8") - - print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over secure TCP socket...") - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(15.0) - sock.connect((server_host, server_port)) - - frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes - sock.sendall(frame) - - resp_len_bytes = sock.recv(4) - if not resp_len_bytes: - raise ConnectionError("Server closed connection without response.") - - resp_len = struct.unpack(">I", resp_len_bytes)[0] - resp_bytes = bytearray() - while len(resp_bytes) < resp_len: - chunk = sock.recv(min(4096, resp_len - len(resp_bytes))) - if not chunk: - break - resp_bytes.extend(chunk) - - response = json.loads(resp_bytes.decode("utf-8")) - print(f"[+] Server response: {response}") - return response - - -def ensure_enrolled(config: dict, machine_id: str) -> bool: - """Proactively ensures client certificate enrollment with the Hub if enrollment_secret is available.""" - cert_dir = config.get("cert_dir", "certs") - enrollment_secret = config.get("enrollment_secret") - if not enrollment_secret: - return False - - server_host = config.get("server_host", "127.0.0.1") - hermes_host = config.get("hermes_host", server_host) - if hermes_host in ("0.0.0.0", "::", None, ""): - hermes_host = server_host if server_host not in ("0.0.0.0", "::", None, "") else "127.0.0.1" - hermes_port = config.get("hermes_port", 8443) - hub_url = f"http://{hermes_host}:{hermes_port}" - try: - return enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type="windows") - except Exception as e: - print(f"[!] Warning: Initial client enrollment bootstrap with Hub at {hub_url} failed: {e}") - return False - - -def main(): - parser = argparse.ArgumentParser(description="LOGAR Windows Edge Log Forwarder with State Tracking") - parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json") - parser.add_argument("--hours", type=int, default=24, help="Lookback window in hours for event logs (default: 24)") - parser.add_argument("--state-file", default=None, help="Path to state tracking file (default: client_state.json next to config)") - parser.add_argument("--no-state", action="store_true", help="Disable state tracking and send all events matching lookback window") - parser.add_argument("--interval", type=int, default=0, help="Continuous polling interval in seconds (0 = run once and exit, >0 = service loop)") - args = parser.parse_args() - - try: - config = load_config(args.config) - except Exception as e: - print(f"[!] Configuration error: {e}") - sys.exit(1) - - state_path = get_state_path(args.config, args.state_file) - state = None if args.no_state else load_state(state_path) - - machine_id = config.get("client_id") or get_machine_identifier() - print(f"[*] Edge Forwarder Node: {machine_id}") - if state and "last_record_number" in state: - print(f"[*] State tracking active: resuming from record #{state['last_record_number']} (state file: {state_path})") - elif not args.no_state: - print(f"[*] State tracking initialized (state file: {state_path})") - - # Proactively ensure dynamic PKI certificate enrollment with Server Hub - ensure_enrolled(config, machine_id) - - def run_forwarder_cycle(): - # Continuously verify/renew PKI enrollment if server is licensed & online - ensure_enrolled(config, machine_id) - - print(f"[*] Scanning Windows Application event log for unsent entries (last {args.hours} hours)...") - candidate_logs = get_recent_windows_logs(hours=args.hours, state=state) - print(f"[*] Found {len(candidate_logs)} unsent candidate entries (INFO to ERROR, entries > {args.hours}h and already-sent skipped).") - - if not candidate_logs: - print("[*] No new unsent events to transmit.") - if state is not None: - commit_state(state, state_path) - return - - try: - resp = send_encrypted_logs_over_socket(config, candidate_logs) - if state is not None and resp and resp.get("status") == "success": - commit_state(state, state_path) - print(f"[+] State successfully committed to {state_path}") - except Exception as e: - print(f"[!] Failed to stream logs to server: {e}") - if not args.interval: - sys.exit(1) - - if args.interval and args.interval > 0: - print(f"[*] Starting continuous forwarder service loop (Interval: {args.interval}s)...") - while True: - try: - run_forwarder_cycle() - except Exception as e: - print(f"[!] Error in forwarder cycle: {e}") - time.sleep(args.interval) - else: - run_forwarder_cycle() - - -if __name__ == "__main__": - main() \ No newline at end of file +""" +LOGAR Windows Edge Log Forwarder +Collects Windows Event Logs (Application/System) and securely streams them +to the central LOGAR Hub via mutual TLS (mTLS 1.3) or OpenPGP envelope. +""" + +import os +import sys +import time +import argparse +import warnings +from typing import Optional + +# Ensure script directory and repo root are in sys.path +SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +if SCRIPT_DIR not in sys.path: + sys.path.insert(0, SCRIPT_DIR) +REPO_ROOT = os.path.abspath(os.path.join(SCRIPT_DIR, "..")) +if REPO_ROOT not in sys.path: + sys.path.insert(0, REPO_ROOT) + +# Suppress cryptography / pgpy deprecation notices +warnings.filterwarnings("ignore") + +import pgpy + +try: + import win32evtlog +except ImportError: + win32evtlog = None + +from log_collector import ( + get_machine_identifier, + is_path_writable, + get_default_data_dir, + resolve_writable_path, + get_state_path, + load_state, + save_state, + commit_state, + WindowsLogCollector, + BaseLogCollector +) +from client_transport import ( + is_cert_expiring_soon, + enroll_client_if_needed, + get_tls_socket, + load_config, + ensure_enrolled as transport_ensure_enrolled, + send_encrypted_logs_over_socket as transport_send_encrypted_logs +) + +CONFIG_FILE_NAME = "client_config.json" +STATE_FILE_NAME = "client_state.json" + + +def get_recent_windows_logs(hours: int = 24, state: Optional[dict] = None) -> list: + """ + Scans the Windows Application Event Log backwards for events within the window. + Edge Filtering: Retains INFO, WARNING, and ERROR. Drops Audit and Debug noise. + State Tracking: Skips events older than lookback window and already sent records. + """ + collector = WindowsLogCollector(state=state, channels=["Application"]) + return collector.collect_logs(hours=hours) + + +def send_encrypted_logs_over_socket(config: dict, logs: list): + """Streams logs to the LOGAR hub over mTLS or encrypted TCP envelope.""" + return transport_send_encrypted_logs(config=config, logs=logs, os_type="windows") + + +def ensure_enrolled(config: dict, machine_id: str) -> bool: + """Proactively ensures client certificate enrollment with the Hub if enrollment_secret is available.""" + return transport_ensure_enrolled(config=config, machine_id=machine_id, os_type="windows") + + +def main(): + parser = argparse.ArgumentParser(description="LOGAR Windows Edge Log Forwarder with State Tracking") + parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json") + parser.add_argument("--hours", type=int, default=24, help="Lookback window in hours for event logs (default: 24)") + parser.add_argument("--state-file", default=None, help="Path to state tracking file (default: client_state.json next to config)") + parser.add_argument("--no-state", action="store_true", help="Disable state tracking and send all events matching lookback window") + parser.add_argument("--interval", type=int, default=0, help="Continuous polling interval in seconds (0 = run once and exit, >0 = service loop)") + args = parser.parse_args() + + try: + config = load_config(args.config) + except Exception as e: + print(f"[!] Configuration error: {e}") + sys.exit(1) + + state_path = get_state_path(args.config, args.state_file) + state = None if args.no_state else load_state(state_path) + + machine_id = config.get("client_id") or get_machine_identifier() + print(f"[*] Edge Forwarder Node: {machine_id}") + if state and "last_record_number" in state: + print(f"[*] State tracking active: resuming from record #{state['last_record_number']} (state file: {state_path})") + elif not args.no_state: + print(f"[*] State tracking initialized (state file: {state_path})") + + # Proactively ensure dynamic PKI certificate enrollment with Server Hub + ensure_enrolled(config, machine_id) + + collector = WindowsLogCollector(state=state, machine_id=machine_id, channels=["Application"]) + + def run_forwarder_cycle(): + # Continuously verify/renew PKI enrollment if server is licensed & online + ensure_enrolled(config, machine_id) + + print(f"[*] Scanning Windows Application event log for unsent entries (last {args.hours} hours)...") + candidate_logs = collector.collect_logs(hours=args.hours) + print(f"[*] Found {len(candidate_logs)} unsent candidate entries (INFO to ERROR, entries > {args.hours}h and already-sent skipped).") + + if not candidate_logs: + print("[*] No new unsent events to transmit.") + if state is not None: + commit_state(state, state_path) + return + + try: + resp = send_encrypted_logs_over_socket(config, candidate_logs) + if state is not None and resp and resp.get("status") == "success": + commit_state(state, state_path) + print(f"[+] State successfully committed to {state_path}") + except Exception as e: + print(f"[!] Failed to stream logs to server: {e}") + if not args.interval: + sys.exit(1) + + if args.interval and args.interval > 0: + print(f"[*] Starting continuous forwarder service loop (Interval: {args.interval}s)...") + while True: + try: + run_forwarder_cycle() + except Exception as e: + print(f"[!] Error in forwarder cycle: {e}") + time.sleep(args.interval) + else: + run_forwarder_cycle() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/client_transport.py b/src/client_transport.py new file mode 100644 index 0000000..ff28363 --- /dev/null +++ b/src/client_transport.py @@ -0,0 +1,322 @@ +""" +LOGAR - Unified Client Transport & PKI Enrollment Module +Encapsulates mutual TLS (mTLS 1.3), OpenPGP fallback envelope framing, +and dynamic certificate enrollment with the LOGAR Server Hub. +""" + +import os +import sys +import json +import socket +import ssl +import struct +import urllib.request +import warnings +from datetime import datetime, timezone, timedelta +from typing import Optional, List, Dict, Any + +# Suppress cryptography / pgpy deprecation notices +warnings.filterwarnings("ignore") + +import pgpy +from log_collector import ( + get_machine_identifier, + is_path_writable, + get_default_data_dir, + resolve_writable_path +) + +CONFIG_FILE_NAME = "client_config.json" + + +def is_cert_expiring_soon(cert_path: str, threshold_days: int = 30) -> bool: + """Checks if client certificate at cert_path is expiring within threshold_days.""" + if not os.path.exists(cert_path): + return True + try: + from cryptography import x509 + with open(cert_path, "r", encoding="utf-8") as f: + cert = x509.load_pem_x509_certificate(f.read().encode("utf-8")) + expiry = getattr(cert, "not_valid_after_utc", None) + if expiry is None: + expiry = cert.not_valid_after.replace(tzinfo=timezone.utc) + now = datetime.now(timezone.utc) + return expiry <= (now + timedelta(days=threshold_days)) + except Exception: + return True + + +def enroll_client_if_needed( + hub_url: str, + enrollment_secret: str, + cert_dir: str, + client_id: str, + hostname: str, + os_type: str = "windows", + force_renew: bool = False, + threshold_days: int = 30 +) -> bool: + """Bootstraps client enrollment if certificates are missing or expiring soon.""" + os.makedirs(cert_dir, exist_ok=True) + ca_path = os.path.join(cert_dir, "ca.crt") + cert_path = os.path.join(cert_dir, "client.crt") + key_path = os.path.join(cert_dir, "client.key") + + if not force_renew and os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path): + if not is_cert_expiring_soon(cert_path, threshold_days=threshold_days): + return True + print(f"[*] Client certificate at {cert_path} is expiring within {threshold_days} days. Auto-renewing...") + + action_name = "re-enrolling" if os.path.exists(cert_path) else "enrolling" + print(f"[*] Bootstrapping client {action_name} with LOGAR Hub at {hub_url}...") + enroll_endpoint = f"{hub_url.rstrip('/')}/api/client/enroll" + payload = { + "client_id": client_id, + "hostname": hostname, + "os": os_type, + "enrollment_secret": enrollment_secret + } + req = urllib.request.Request( + enroll_endpoint, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"} + ) + with urllib.request.urlopen(req, timeout=10) as resp: + if resp.status != 200: + raise RuntimeError(f"Enrollment failed with status code {resp.status}") + data = json.loads(resp.read().decode("utf-8")) + + with open(ca_path, "w", encoding="utf-8") as f: + f.write(data["ca_cert"]) + with open(cert_path, "w", encoding="utf-8") as f: + f.write(data["client_cert"]) + with open(key_path, "w", encoding="utf-8") as f: + f.write(data["client_key"]) + + try: + os.chmod(key_path, 0o600) + except Exception: + pass + + print(f"[+] Client certificates updated successfully in {os.path.abspath(cert_dir)}") + return True + + +def get_tls_socket(hub_host: str, hub_port: int, cert_dir: str): + """Establishes an mTLS connection with the LOGAR hub using client certificates.""" + ca_path = os.path.join(cert_dir, "ca.crt") + cert_path = os.path.join(cert_dir, "client.crt") + key_path = os.path.join(cert_dir, "client.key") + + if not (os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path)): + raise FileNotFoundError(f"mTLS certificates not found in '{cert_dir}'. Enroll client first.") + + ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca_path) + ctx.load_cert_chain(certfile=cert_path, keyfile=key_path) + ctx.minimum_version = ssl.TLSVersion.TLSv1_3 + ctx.check_hostname = False + + raw_sock = socket.create_connection((hub_host, hub_port), timeout=15) + return ctx.wrap_socket(raw_sock, server_hostname=hub_host) + + +def load_config(config_path: str = CONFIG_FILE_NAME) -> dict: + """Loads and normalizes client configuration JSON.""" + actual_config_path = config_path + if not os.path.exists(actual_config_path): + data_dir_cfg = os.path.join(get_default_data_dir("LOGAR"), os.path.basename(config_path)) + if os.path.exists(data_dir_cfg): + actual_config_path = data_dir_cfg + + if not os.path.exists(actual_config_path): + raise FileNotFoundError( + f"Client configuration file not found at: {config_path}\n" + f"Generate one from the server using: python Server.py --create-client-config --client-out {config_path}" + ) + with open(actual_config_path, "r", encoding="utf-8") as f: + config = json.load(f) + + # Resolve cert_dir to a writable directory + cfg_dir = os.path.dirname(os.path.abspath(actual_config_path)) + raw_cert_dir = config.get("cert_dir", "certs") + if os.path.isabs(raw_cert_dir) and is_path_writable(raw_cert_dir): + config["cert_dir"] = raw_cert_dir + else: + candidate = os.path.join(cfg_dir, os.path.basename(raw_cert_dir)) + if is_path_writable(candidate) or is_path_writable(cfg_dir): + config["cert_dir"] = candidate + else: + config["cert_dir"] = resolve_writable_path("certs", "LOGAR", "certs") + + # If ca_cert is present in config and ca.crt doesn't exist yet, save it + ca_cert_pem = config.get("ca_cert") + if ca_cert_pem and isinstance(ca_cert_pem, str): + ca_file = os.path.join(config["cert_dir"], "ca.crt") + if not os.path.exists(ca_file): + try: + os.makedirs(config["cert_dir"], exist_ok=True) + with open(ca_file, "w", encoding="utf-8") as f: + f.write(ca_cert_pem) + except Exception: + pass + + return config + + +def ensure_enrolled(config: dict, machine_id: str, os_type: str = "windows") -> bool: + """Proactively ensures client certificate enrollment with the Hub if enrollment_secret is available.""" + cert_dir = config.get("cert_dir", "certs") + enrollment_secret = config.get("enrollment_secret") + if not enrollment_secret: + return False + + server_host = config.get("server_host", "127.0.0.1") + hermes_host = config.get("hermes_host", server_host) + if hermes_host in ("0.0.0.0", "::", None, ""): + hermes_host = server_host if server_host not in ("0.0.0.0", "::", None, "") else "127.0.0.1" + hermes_port = config.get("hermes_port", 8443) + hub_url = f"http://{hermes_host}:{hermes_port}" + try: + return enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, machine_id, machine_id, os_type=os_type) + except Exception as e: + print(f"[!] Warning: Initial client enrollment bootstrap with Hub at {hub_url} failed: {e}") + return False + + +def send_encrypted_logs_over_socket( + config: dict, + logs: list, + os_type: Optional[str] = None, + machine_id: Optional[str] = None +) -> dict: + """ + Streams logs to the LOGAR hub. + Uses mutual TLS 1.3 (mTLS) with client certificates if available, + or falls back to OpenPGP encrypted envelope over TCP socket. + """ + node_os = os_type or ("windows" if sys.platform == "win32" else "linux") + server_host = config["server_host"] + server_port = int(config["server_port"]) + cert_dir = config.get("cert_dir", "certs") + enrollment_secret = config.get("enrollment_secret") + actual_machine_id = machine_id or config.get("client_id") or get_machine_identifier() + + # Attempt automatic enrollment bootstrap if certs are missing and secret is provided + hub_url = None + if enrollment_secret: + hermes_host = config.get("hermes_host", server_host) + if hermes_host in ("0.0.0.0", "::", None): + hermes_host = server_host if server_host not in ("0.0.0.0", "::", None) else "127.0.0.1" + hermes_port = config.get("hermes_port", 8443) + hub_url = f"http://{hermes_host}:{hermes_port}" + try: + enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, actual_machine_id, actual_machine_id, os_type=node_os) + except Exception as e: + print(f"[!] Warning: Enrollment bootstrap failed: {e}") + + ca_path = os.path.join(cert_dir, "ca.crt") + cert_path = os.path.join(cert_dir, "client.crt") + key_path = os.path.join(cert_dir, "client.key") + has_mtls_certs = os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path) + + if not has_mtls_certs and enrollment_secret and hub_url: + try: + enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, actual_machine_id, actual_machine_id, os_type=node_os) + has_mtls_certs = os.path.exists(ca_path) and os.path.exists(cert_path) and os.path.exists(key_path) + except Exception as e: + print(f"[!] Warning: Enrollment before TLS connection failed: {e}") + + if has_mtls_certs: + print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over mTLS (TLS 1.3)...") + sock = None + try: + sock = get_tls_socket(server_host, server_port, cert_dir) + except (ssl.SSLError, ssl.CertificateError, ConnectionResetError) as tls_err: + if enrollment_secret and hub_url: + print(f"[!] TLS handshake error ({tls_err}). Re-enrolling with LOGAR Hub...") + try: + enroll_client_if_needed(hub_url, enrollment_secret, cert_dir, actual_machine_id, actual_machine_id, os_type=node_os, force_renew=True) + sock = get_tls_socket(server_host, server_port, cert_dir) + except Exception as retry_err: + print(f"[!] Re-enrollment or reconnection retry failed: {retry_err}") + raise + else: + raise + + with sock: + payload = { + "server": actual_machine_id, + "timestamp": datetime.now(timezone.utc).isoformat(), + "logs": logs + } + payload_bytes = json.dumps(payload).encode("utf-8") + frame = struct.pack(">I", len(payload_bytes)) + payload_bytes + sock.sendall(frame) + + resp_len_bytes = sock.recv(4) + if not resp_len_bytes: + raise ConnectionError("Server closed mTLS connection without response.") + resp_len = struct.unpack(">I", resp_len_bytes)[0] + resp_bytes = bytearray() + while len(resp_bytes) < resp_len: + chunk = sock.recv(min(4096, resp_len - len(resp_bytes))) + if not chunk: + break + resp_bytes.extend(chunk) + + response = json.loads(resp_bytes.decode("utf-8")) + print(f"[+] Server response: {response}") + return response + + # Fallback to OpenPGP envelope over plain TCP socket + auth_token = config.get("auth_token", "") + pub_key_armored = config.get("server_public_key") + if not pub_key_armored: + raise ValueError("No server public key or mTLS certificates available for connection.") + + expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper() + pub_key, _ = pgpy.PGPKey.from_blob(pub_key_armored) + actual_fp = str(pub_key.fingerprint).replace(" ", "").upper() + if expected_fp and actual_fp != expected_fp: + raise ValueError(f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}.") + + payload = { + "server": actual_machine_id, + "timestamp": datetime.now(timezone.utc).isoformat(), + "logs": logs + } + payload_json = json.dumps(payload) + pgp_msg = pgpy.PGPMessage.new(payload_json) + encrypted_msg = pub_key.encrypt(pgp_msg) + encrypted_armored = str(encrypted_msg) + + envelope = { + "auth_token": auth_token, + "timestamp": datetime.now(timezone.utc).isoformat(), + "encrypted_payload": encrypted_armored + } + envelope_bytes = json.dumps(envelope).encode("utf-8") + + print(f"[*] Connecting to LOGAR server at {server_host}:{server_port} over secure TCP socket...") + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.settimeout(15.0) + sock.connect((server_host, server_port)) + + frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes + sock.sendall(frame) + + resp_len_bytes = sock.recv(4) + if not resp_len_bytes: + raise ConnectionError("Server closed connection without response.") + + resp_len = struct.unpack(">I", resp_len_bytes)[0] + resp_bytes = bytearray() + while len(resp_bytes) < resp_len: + chunk = sock.recv(min(4096, resp_len - len(resp_bytes))) + if not chunk: + break + resp_bytes.extend(chunk) + + response = json.loads(resp_bytes.decode("utf-8")) + print(f"[+] Server response: {response}") + return response diff --git a/src/log_collector.py b/src/log_collector.py new file mode 100644 index 0000000..1edbd3e --- /dev/null +++ b/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\\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/, /tmp/, ~/.local/share/ + 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) diff --git a/tests/test_log_collector.py b/tests/test_log_collector.py new file mode 100644 index 0000000..36e0c63 --- /dev/null +++ b/tests/test_log_collector.py @@ -0,0 +1,108 @@ +""" +Unit tests for LOGAR log_collector and client_transport modules. +""" + +import os +import sys +import json +import unittest +import warnings +from datetime import datetime, timezone, timedelta + +warnings.filterwarnings("ignore") +warnings.simplefilter("ignore") + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))) + +from log_collector import ( + get_machine_identifier, + is_path_writable, + get_default_data_dir, + resolve_writable_path, + get_state_path, + load_state, + save_state, + commit_state, + BaseLogCollector, + WindowsLogCollector, + LinuxLogCollector, + get_log_collector +) +from client_transport import ( + is_cert_expiring_soon, + load_config +) + + +class TestLogCollectorModule(unittest.TestCase): + def test_machine_identifier(self): + mid = get_machine_identifier() + self.assertIsInstance(mid, str) + self.assertTrue(len(mid) > 0) + self.assertNotEqual(mid, "localhost") + + def test_path_writable_and_data_dir(self): + data_dir = get_default_data_dir("LOGAR_TEST") + self.assertIsInstance(data_dir, str) + self.assertTrue(is_path_writable(data_dir)) + + def test_state_lifecycle_windows_and_linux(self): + state_file = "test_shared_state.json" + try: + state = load_state(state_file) + self.assertEqual(state, {}) + + # Windows staged state + state["new_last_record_number"] = 100 + state["new_sent_record_ids"] = ["100:2026-09-27T00:00:00"] + commit_state(state, state_file) + self.assertNotIn("new_last_record_number", state) + self.assertEqual(state["last_record_number"], 100) + + # Linux staged state + state["new_last_cursor"] = "c_xyz" + state["new_last_timestamp_us"] = 123456789.0 + state["new_sent_cursors"] = ["c_xyz"] + commit_state(state, state_file) + self.assertNotIn("new_last_cursor", state) + self.assertEqual(state["last_cursor"], "c_xyz") + self.assertEqual(state["last_timestamp_us"], 123456789.0) + + # Reload and verify + reloaded = load_state(state_file) + self.assertEqual(reloaded["last_record_number"], 100) + self.assertEqual(reloaded["last_cursor"], "c_xyz") + finally: + if os.path.exists(state_file): + try: + os.remove(state_file) + except Exception: + pass + + def test_windows_log_collector_severity_mapping(self): + state = {} + collector = WindowsLogCollector(state=state, machine_id="test-win-node", channels=["Application"]) + self.assertEqual(collector.machine_id, "test-win-node") + self.assertEqual(collector.channels, ["Application"]) + + def test_linux_log_collector_priority_normalization(self): + state = {} + collector = LinuxLogCollector(state=state, machine_id="test-linux-node") + self.assertEqual(collector.machine_id, "test-linux-node") + + def test_get_log_collector_factory(self): + win_col = get_log_collector(os_type="windows", machine_id="win-1") + self.assertIsInstance(win_col, WindowsLogCollector) + self.assertEqual(win_col.machine_id, "win-1") + + lin_col = get_log_collector(os_type="linux", machine_id="lin-1") + self.assertIsInstance(lin_col, LinuxLogCollector) + self.assertEqual(lin_col.machine_id, "lin-1") + + auto_col = get_log_collector() + self.assertIsInstance(auto_col, BaseLogCollector) + + +if __name__ == "__main__": + unittest.main()