commit c5d364b289b1041a7e334573df3f4508df396492 Author: max Date: Thu Sep 3 21:05:28 2026 +0200 Initial commit: LOGAR edge-thin log analysis system with OpenPGP encryption, authenticated TCP sockets, 4-run persistence rule, Hermes reporting, and modular shippables diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d48306f --- /dev/null +++ b/.gitignore @@ -0,0 +1,30 @@ +# Virtual Environment +venv/ +env/ +ENV/ + +# Python cache & build +__pycache__/ +*.py[cod] +*$py.class +build/ +dist/ +*.spec + +# Database files & state +*.db +*.sqlite +*.sqlite3 + +# Live configuration with generated private keys & tokens (samples are tracked) +server_config.json +client_config.json +test_*.json +test_*.db + +# IDE & OS files +.vscode/ +.idea/ +Thumbs.db +Desktop.ini +.DS_Store diff --git a/Linux_Client.py b/Linux_Client.py new file mode 100644 index 0000000..6ac8d31 --- /dev/null +++ b/Linux_Client.py @@ -0,0 +1,206 @@ +import os +import sys +import json +import socket +import struct +import argparse +import subprocess +import warnings +from datetime import datetime, timezone + +# Suppress cryptography / pgpy deprecation notices +warnings.filterwarnings("ignore") + +import pgpy + +CONFIG_FILE_NAME = "client_config.json" + + +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: + return json.load(f) + + +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 = 6): + """ + Collects warnings and errors from systemd journalctl over the lookback window. + Edge Thinness: Drops INFO and DEBUG entries at the source. + """ + cmd = ["journalctl", "--since", f"{hours} hours ago", "-p", "warning", "--output=json"] + 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() + + for line in result.stdout.splitlines(): + line_str = line.strip() + if not line_str: + continue + try: + entry = json.loads(line_str) + priority = str(entry.get("PRIORITY", "4")) + # Priority 0: Emerg, 1: Alert, 2: Crit, 3: Err, 4: Warning. + # Strip anything above 4 (5: Notice, 6: Info, 7: Debug) + if int(priority) > 4: + continue + + sev = "WARNING" if priority == "4" else "ERROR" + 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 + + return logs + + +def send_encrypted_logs_over_socket(config: dict, logs: list): + """ + Encrypts the payload using the server's OpenPGP public key and streams + over an authenticated TCP socket. Zero local state is maintained on the client. + """ + server_host = config["server_host"] + server_port = int(config["server_port"]) + auth_token = config["auth_token"] + pub_key_armored = config["server_public_key"] + expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper() + + # Load and verify server public key + 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}." + ) + + machine_id = get_machine_identifier() + + # Prepare zero-state candidate batch + payload = { + "server": machine_id, + "timestamp": datetime.now(timezone.utc).isoformat(), + "logs": logs + } + payload_json = json.dumps(payload) + + # Encrypt payload with server's encryption-only key + pgp_msg = pgpy.PGPMessage.new(payload_json) + encrypted_msg = pub_key.encrypt(pgp_msg) + encrypted_armored = str(encrypted_msg) + + # Envelope with socket authentication header + envelope = { + "auth_token": auth_token, + "timestamp": datetime.now(timezone.utc).isoformat(), + "encrypted_payload": encrypted_armored + } + envelope_bytes = json.dumps(envelope).encode("utf-8") + + # Connect over TCP socket and transmit with 4-byte length prefix framing + 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)) + + # Send frame: length (4 bytes big-endian) + envelope + frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes + sock.sendall(frame) + + # Receive response length + 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 main(): + parser = argparse.ArgumentParser(description="LOGAR Linux Edge Log Forwarder (Zero State)") + parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json") + parser.add_argument("--hours", type=int, default=6, help="Lookback window in hours for journalctl logs") + args = parser.parse_args() + + try: + config = load_config(args.config) + except Exception as e: + print(f"[!] Configuration error: {e}") + sys.exit(1) + + machine_id = get_machine_identifier() + print(f"[*] Edge Forwarder Node: {machine_id}") + print(f"[*] Scanning Linux journalctl for candidate anomalies (last {args.hours} hours)...") + candidate_logs = get_recent_linux_logs(hours=args.hours) + print(f"[*] Found {len(candidate_logs)} candidate anomalies (noise stripped at source).") + + try: + send_encrypted_logs_over_socket(config, candidate_logs) + except Exception as e: + print(f"[!] Failed to stream logs to server: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..5197227 --- /dev/null +++ b/README.md @@ -0,0 +1,273 @@ +# LOGAR: Edge-Thin Log Analysis & Temporal Verification System + +**LOGAR** is an enterprise log aggregation, verification, and anomaly detection architecture designed for heterogeneous server fleets (Windows & Linux). It combines lightweight zero-state edge forwarders with a centralized cloud hub that applies OpenPGP encryption, authenticated TCP streaming, temporal persistence tracking across 12-hour evaluation windows, and an automated 4-run rule to filter out transient infrastructure blips before reporting verified anomalies to **Hermes**. + +--- + +## Table of Contents +1. [Core Philosophy](#core-philosophy) +2. [Architecture & Data Flow](#architecture--data-flow) +3. [Security & Cryptographic Model](#security--cryptographic-model) +4. [Cloud-Side Temporal Persistence & 4-Run Rule](#cloud-side-temporal-persistence--4-run-rule) +5. [Agentic Hermes Integration](#agentic-hermes-integration) +6. [Dynamic Machine & Domain Identification](#dynamic-machine--domain-identification) +7. [Repository & Shippables Structure](#repository--shippables-structure) +8. [Getting Started & Installation](#getting-started--installation) +9. [Running Tests](#running-tests) + +--- + +## Core Philosophy + +### 1. Edge Thinness & Zero State +Site agents running on Windows and Linux act strictly as lightweight forwarders: +- **No Local Database**: Clients maintain zero state and no local SQLite or cache files. +- **Source-Level Noise Stripping**: Conversational, informational, and debugging log noise (`INFO`, `DEBUG`, audit entries) is dropped directly at the source. +- **End-to-End Encryption**: Logs are encrypted using the server's OpenPGP public key before leaving the edge node. +- **Secure TCP Sockets**: Ingestion occurs over low-overhead authenticated TCP sockets rather than bulky HTTP/HTTPS endpoints. + +### 2. Cloud-Side Temporal Persistence +The central Python/TCP hub handles the heavy lifting: +- State tracking is managed centrally in SQLite (`logar_state.db`). +- Candidate issues are evaluated over a **12-hour temporal evaluation window**. +- An issue must persist across **at least 4 consecutive runs / cycles** to be confirmed as a genuine system anomaly. Transient blips and sporadic spikes are filtered out automatically. + +### 3. Agentic Integration with Hermes +Instead of human engineers manually diving through noisy logs, **Hermes** ingests pre-filtered, 4-run validated anomalies directly from the cloud hub (`GET /api/hermes/report`), treating them as verified system artifacts to trigger precise team notifications. + +--- + +## Architecture & Data Flow + +```mermaid +graph TB + subgraph Edge Nodes [Zero-State Edge Forwarders] + W[Win_Client.py / Win_Client.exe
Windows Event Log Application] + L[Linux_Client.py / Linux_Client.bin
systemd journalctl -p warning] + end + + subgraph Security Layer [Security & Framing] + E[OpenPGP Payload Encryption
Server Public Key & Fingerprint] + S[Length-Prefixed Framing
4-byte Big-Endian + Auth Envelope] + end + + subgraph Cloud Hub [LOGAR Central Server Hub] + TCP[Authenticated TCP Listener
Port 9443] + DEC[OpenPGP Decryption
Server Private Key] + DB[(SQLite Persistence
active_issues & ingest_runs)] + RULE{12h Window &
4-Run Rule} + end + + subgraph Agentic Reporting [Downstream Integration] + API[FastAPI / Uvicorn Reporting
Port 8443] + HERMES[Hermes Agent
GET /api/hermes/report] + end + + W --> E + L --> E + E --> S + S -->|TCP Stream| TCP + TCP --> DEC + DEC --> RULE + RULE --> DB + DB --> API + API --> HERMES +``` + +--- + +## Security & Cryptographic Model + +### Pure-Python OpenPGP (RFC 4880) +- **Zero OS Binary Dependency**: Utilizes `pgpy` and `cryptography` in pure Python. **No native GnuPG or `gpg` binary installation is required** on the server, Windows nodes, or Linux nodes. +- **First-Run Automatic Key Generation**: On the first launch, if `server_config.json` is missing, `Server.py` automatically generates: + - An OpenPGP RSA 2048 keypair with encryption-only usage flags. + - An armored private key (`private_key`) and public key (`public_key`). + - A SHA-256 public encryption fingerprint (`server_fingerprint`). + - A cryptographically random authentication secret token (`auth_token`). +- **Client Configuration Exporter**: + ```bash + python Server.py --create-client-config --server-host 127.0.0.1 --server-port 9443 --client-out client_config.json + ``` + Produces an anonymous client config containing only the server socket coordinates, authentication token, and the encryption-only public key & fingerprint. +- **Socket Protocol Framing**: + - `[4 bytes big-endian unsigned int]` : Total envelope length. + - `[JSON Envelope]` : + ```json + { + "auth_token": "", + "timestamp": "2026-09-03T...", + "encrypted_payload": "-----BEGIN PGP MESSAGE-----\n..." + } + ``` + - Unauthorized clients or invalid authentication tokens are rejected immediately. + +--- + +## Cloud-Side Temporal Persistence & 4-Run Rule + +Incoming candidate logs are tracked in SQLite table `active_issues`: +- **Issue Fingerprint**: Formatted as `{site_name}:{server}:{signature}`. +- **12-Hour Evaluation Window**: + - When an issue is observed, the hub compares `(now - last_seen)`. + - If more than 12 hours have passed since the issue was last recorded, the previous window is expired and the cycle resets to `run_count = 1` with status `TRANSIENT`. +- **4-Run Rule**: + - For each distinct run batch, `run_count` increments. + - Issues with `run_count < 4` are marked as `TRANSIENT` and ignored by downstream reporting. + - When `run_count >= 4` within the active 12-hour window, the status transitions to `VERIFIED`. + +--- + +## Agentic Hermes Integration + +The server hub serves a REST reporting API (default port `8443`): + +### `GET /api/hermes/report` +Returns exclusively **verified anomalies** that have satisfied the 4-run rule within the active 12-hour evaluation window: + +```json +[ + { + "fingerprint": "corp.internal:web-app-01.corp.internal:NginxWorkerCrash", + "site": "corp.internal", + "server": "web-app-01.corp.internal", + "signature": "NginxWorkerCrash", + "severity": "ERROR", + "message": "Worker process 4120 terminated with signal 11", + "os_type": "linux", + "first_seen": "2026-09-03T09:00:00+00:00", + "last_seen": "2026-09-03T21:00:00+00:00", + "consecutive_runs": 4, + "evaluation_window": "12h", + "verified": true, + "status": "VERIFIED" + } +] +``` + +### `GET /api/hermes/all` +Diagnostic endpoint listing all active issues (both `TRANSIENT` candidate blips and `VERIFIED` anomalies). + +### `GET /health` +Returns hub health, encryption fingerprint, and listener ports. + +--- + +## Dynamic Machine & Domain Identification + +Client configurations intentionally contain **no machine name or site name**. Both forwarders dynamically identify their host and domain at runtime via `get_machine_identifier()`: +1. **Fully Qualified Domain Name (FQDN)**: Checked via `socket.getfqdn()`. +2. **OS-Specific Domain Discovery**: + - **Windows**: Checks Active Directory environment variable `USERDNSDOMAIN` / `USERDOMAIN`. + - **Linux**: Parses `/etc/resolv.conf` `domain` and `search` directives. +3. **Reverse DNS Lookup**: Resolves canonical hostname via `socket.gethostbyaddr`. +4. **Fallback**: Local hostname `socket.gethostname()`. + +The server automatically infers site attribution from domain qualifiers (e.g. `node01.corp.internal` $\rightarrow$ site `corp.internal`). + +--- + +## Repository & Shippables Structure + +``` +LOGAR/ +├── .gitignore # Ignore venv, caches, DBs, and private keys +├── requirements.txt # Unified dependencies +├── README.md # Comprehensive documentation +├── Server.py # Central TCP server and Hermes API +├── Win_Client.py # Windows edge forwarder +├── Linux_Client.py # Linux edge forwarder +├── test_pipeline.py # End-to-end integration test +└── out/ # Standalone shippable distributions + ├── server/ + │ ├── Server.exe # Standalone Windows executable + │ ├── Server.py # Python source + │ ├── server_config.sample.json + │ ├── requirements.txt + │ ├── README.md + │ └── test/ + │ └── test_server.py # Server unit tests + ├── win_client/ + │ ├── Win_Client.exe # Standalone Windows executable + │ ├── Win_Client.py # Python source + │ ├── client_config.sample.json + │ ├── requirements.txt + │ ├── README.md + │ └── test/ + │ └── test_win_client.py # Windows client unit tests + └── linux_client/ + ├── Linux_Client.bin # Standalone executable binary (zipapp) + ├── build_bin.sh # PyInstaller ELF compiler script + ├── Linux_Client.py # Python source + ├── client_config.sample.json + ├── requirements.txt + ├── README.md + └── test/ + └── test_linux_client.py# Linux client unit tests +``` + +--- + +## Getting Started & Installation + +### 1. Central Server Hub + +1. **Install dependencies**: + ```bash + pip install -r requirements.txt + ``` +2. **Start the server** (generates `server_config.json` and keypair on first run): + ```bash + python Server.py + # Or run the standalone executable: + ./out/server/Server.exe + ``` +3. **Export a client configuration**: + ```bash + python Server.py --create-client-config --server-host --server-port 9443 --client-out client_config.json + ``` + +### 2. Windows Client Deployment + +1. Copy `Win_Client.exe` (or `Win_Client.py`) and `client_config.json` to the target machine. +2. Run manually or schedule via Task Scheduler (every 3 hours): + ```powershell + Win_Client.exe --hours 6 + ``` + +### 3. Linux Client Deployment + +1. Copy `Linux_Client.bin` (or `Linux_Client.py`) and `client_config.json` to `/opt/logar/`. +2. Ensure executable permissions: + ```bash + chmod +x /opt/logar/Linux_Client.bin + ``` +3. Run via cron or systemd timer: + ```bash + 0 */3 * * * /opt/logar/Linux_Client.bin --hours 6 + ``` + +--- + +## Running Tests + +### 1. Component-Specific Unit Tests +Each component in `out/` includes its own isolated test suite: + +```bash +# Server tests (config generation, SQLite persistence, 4-run rule) +python out/server/test/test_server.py + +# Windows client tests (config anonymity, machine ID, OpenPGP encryption) +python out/win_client/test/test_win_client.py + +# Linux client tests (config anonymity, journalctl priority filter, OpenPGP) +python out/linux_client/test/test_linux_client.py +``` + +### 2. End-to-End Pipeline Integration Test +Start the server in one shell and run the pipeline test: +```bash +python test_pipeline.py +``` +This tests invalid token rejection, encrypted socket streaming, database persistence, status promotion upon the 4th run, and the Hermes API output. diff --git a/Server.py b/Server.py new file mode 100644 index 0000000..ffa6683 --- /dev/null +++ b/Server.py @@ -0,0 +1,465 @@ +import os +import sys +import json +import uuid +import struct +import socket +import sqlite3 +import argparse +import asyncio +import secrets +import warnings +from datetime import datetime, timezone, timedelta +from typing import Dict, Any, List, Optional + +# Suppress cryptography / pgpy deprecation notices for a clean terminal output +warnings.filterwarnings("ignore") + +import pgpy +from pgpy.constants import ( + PubKeyAlgorithm, + KeyFlags, + HashAlgorithm, + SymmetricKeyAlgorithm, + CompressionAlgorithm +) +from fastapi import FastAPI, HTTPException +import uvicorn + +CONFIG_FILE_NAME = "server_config.json" +DEFAULT_DB_FILE = "logar_state.db" +EVALUATION_WINDOW_HOURS = 12 +RUN_THRESHOLD = 4 + +app = FastAPI(title="LOGAR Cloud Ingestion & Hermes Hub", version="2.0.0") + +# Global context holding server state +SERVER_STATE: Dict[str, Any] = {} + + +def generate_server_keypair(server_name: str): + """Generates an OpenPGP RSA 2048 key with encryption capability.""" + key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048) + uid = pgpy.PGPUID.new(server_name) + key.add_uid( + uid, + usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage}, + hashes=[HashAlgorithm.SHA256], + ciphers=[SymmetricKeyAlgorithm.AES256], + compression=[CompressionAlgorithm.Uncompressed] + ) + private_key_armored = str(key) + public_key_armored = str(key.pubkey) + fingerprint = str(key.pubkey.fingerprint) + return private_key_armored, public_key_armored, fingerprint + + +def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]: + """Loads existing server_config.json or creates a new one on first run.""" + if os.path.exists(config_path): + print(f"[*] Loading server configuration from: {os.path.abspath(config_path)}") + with open(config_path, "r", encoding="utf-8") as f: + config = json.load(f) + return config + + print(f"[!] Config '{config_path}' not found. Initializing first-run configuration...") + server_name = "LOGAR-Cloud-Hub" + private_key, public_key, fingerprint = generate_server_keypair(server_name) + auth_token = secrets.token_hex(24) + + config = { + "server_name": server_name, + "tcp_host": "0.0.0.0", + "tcp_port": 9443, + "hermes_host": "0.0.0.0", + "hermes_port": 8443, + "auth_token": auth_token, + "db_path": DEFAULT_DB_FILE, + "evaluation_window_hours": EVALUATION_WINDOW_HOURS, + "min_persistence_runs": RUN_THRESHOLD, + "server_fingerprint": fingerprint, + "public_key": public_key, + "private_key": private_key + } + + with open(config_path, "w", encoding="utf-8") as f: + json.dump(config, f, indent=2) + + print(f"[+] Successfully generated new server config and OpenPGP keypair.") + print(f"[+] Server Encryption Fingerprint: {fingerprint}") + print(f"[+] Saved to: {os.path.abspath(config_path)}") + return config + + +def create_client_config( + server_host: str, + server_port: int, + output_path: str, + config_path: str = CONFIG_FILE_NAME +) -> Dict[str, Any]: + """Creates a client configuration file containing the server address, auth token, and encryption-only key/fingerprint.""" + server_conf = load_or_init_config(config_path) + + client_conf = { + "server_host": server_host, + "server_port": server_port, + "server_fingerprint": server_conf["server_fingerprint"], + "server_public_key": server_conf["public_key"], + "auth_token": server_conf["auth_token"] + } + + out_dir = os.path.dirname(os.path.abspath(output_path)) + if out_dir and not os.path.exists(out_dir): + os.makedirs(out_dir, exist_ok=True) + + with open(output_path, "w", encoding="utf-8") as f: + json.dump(client_conf, f, indent=2) + + print(f"[+] Client configuration successfully written to: {os.path.abspath(output_path)}") + print(f" - Server Target: {server_host}:{server_port}") + print(f" - Encryption Fingerprint: {server_conf['server_fingerprint']}") + return client_conf + + +def init_db(db_path: str): + """Initializes the SQLite schema for multi-run temporal tracking.""" + conn = sqlite3.connect(db_path) + conn.execute(""" + CREATE TABLE IF NOT EXISTS active_issues ( + fingerprint TEXT PRIMARY KEY, + site_name TEXT, + server TEXT, + signature TEXT, + severity TEXT, + message TEXT, + os_type TEXT, + first_seen TEXT, + last_seen TEXT, + run_count INTEGER, + status TEXT, + last_run_id TEXT + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS ingest_runs ( + run_id TEXT PRIMARY KEY, + site_name TEXT, + server TEXT, + timestamp TEXT, + log_count INTEGER + ) + """) + conn.commit() + conn.close() + + +def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: int, min_runs: int) -> Dict[str, Any]: + """ + Evaluates candidate issues against the 12-hour evaluation window and 4-run rule. + Zero-state clients send raw candidate entries; this engine handles temporal state. + """ + client_server = payload.get("server", "unknown-host") + site_name = payload.get("site_name") or (client_server.split(".", 1)[1] if "." in client_server else "default") + logs = payload.get("logs", []) + run_id = str(uuid.uuid4()) + now = datetime.now(timezone.utc) + now_iso = now.isoformat() + + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + + # Record the batch run + cursor.execute( + "INSERT INTO ingest_runs (run_id, site_name, server, timestamp, log_count) VALUES (?, ?, ?, ?, ?)", + (run_id, site_name, client_server, now_iso, len(logs)) + ) + + processed_count = 0 + promoted_to_verified = 0 + + for log in logs: + severity = str(log.get("severity", "WARNING")).upper() + # Edge forwarder filter safeguard (strip informational noise) + if severity in ["INFO", "DEBUG"]: + continue + + signature = log.get("signature", "unknown") + server = log.get("server", client_server) + message = log.get("message", "") + os_type = log.get("os_type", "unknown") + fp = f"{site_name}:{server}:{signature}" + + cursor.execute( + "SELECT run_count, first_seen, last_seen, status, last_run_id FROM active_issues WHERE fingerprint = ?", + (fp,) + ) + row = cursor.fetchone() + + if row: + run_count, first_seen_str, last_seen_str, current_status, last_run_id = row + try: + last_seen_dt = datetime.fromisoformat(last_seen_str) + except Exception: + last_seen_dt = now + + # 12-hour evaluation window expiry check + if (now - last_seen_dt) > timedelta(hours=window_hours): + # Window elapsed: reset to new cycle + new_runs = 1 + new_first_seen = now_iso + new_status = "TRANSIENT" + else: + # Same run guard: only increment count once per distinct run batch + if last_run_id != run_id: + new_runs = run_count + 1 + else: + new_runs = run_count + new_first_seen = first_seen_str + # 4-run rule enforcement + new_status = "VERIFIED" if new_runs >= min_runs else "TRANSIENT" + + if new_status == "VERIFIED" and current_status != "VERIFIED": + promoted_to_verified += 1 + + cursor.execute(""" + UPDATE active_issues + SET run_count = ?, last_seen = ?, first_seen = ?, status = ?, last_run_id = ?, message = ?, severity = ? + WHERE fingerprint = ? + """, (new_runs, now_iso, new_first_seen, new_status, run_id, message, severity, fp)) + else: + initial_status = "VERIFIED" if 1 >= min_runs else "TRANSIENT" + cursor.execute(""" + INSERT INTO active_issues + (fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status, last_run_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, (fp, site_name, server, signature, severity, message, os_type, now_iso, now_iso, 1, initial_status, run_id)) + + processed_count += 1 + + conn.commit() + conn.close() + + return { + "status": "success", + "run_id": run_id, + "processed": processed_count, + "promoted_verified": promoted_to_verified + } + + +async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + """ + Authenticated TCP socket handler. + Protocol: + - 4-byte big-endian prefix: payload length + - Payload: JSON with auth_token and encrypted_payload (OpenPGP ASCII armored) + - Response: 4-byte length + JSON confirmation + """ + addr = writer.get_extra_info("peername") + try: + # Read 4-byte length prefix + length_bytes = await reader.readexactly(4) + length = struct.unpack(">I", length_bytes)[0] + if length <= 0 or length > 10 * 1024 * 1024: # 10MB limit + raise ValueError(f"Invalid frame size: {length}") + + payload_bytes = await reader.readexactly(length) + envelope = json.loads(payload_bytes.decode("utf-8")) + + # Authenticate socket client + expected_token = SERVER_STATE["config"]["auth_token"] + provided_token = envelope.get("auth_token") + if not secrets.compare_digest(str(provided_token), str(expected_token)): + err_msg = json.dumps({"status": "error", "message": "Authentication failed"}).encode("utf-8") + writer.write(struct.pack(">I", len(err_msg)) + err_msg) + await writer.drain() + writer.close() + await writer.wait_closed() + return + + # Decrypt payload using server's OpenPGP private key + encrypted_armored = envelope.get("encrypted_payload", "") + pgp_msg = pgpy.PGPMessage.from_blob(encrypted_armored) + priv_key = SERVER_STATE["private_key_obj"] + decrypted_obj = priv_key.decrypt(pgp_msg) + decrypted_json_str = decrypted_obj.message + log_payload = json.loads(decrypted_json_str) + + # Ingest and apply 12h window / 4-run rule + res = process_ingested_logs( + log_payload, + db_path=SERVER_STATE["config"]["db_path"], + window_hours=SERVER_STATE["config"]["evaluation_window_hours"], + min_runs=SERVER_STATE["config"]["min_persistence_runs"] + ) + + resp_bytes = json.dumps(res).encode("utf-8") + writer.write(struct.pack(">I", len(resp_bytes)) + resp_bytes) + await writer.drain() + + except Exception as e: + err = json.dumps({"status": "error", "message": str(e)}).encode("utf-8") + try: + writer.write(struct.pack(">I", len(err)) + err) + await writer.drain() + except Exception: + pass + finally: + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + + +@app.get("/api/hermes/report") +def get_hermes_report(): + """ + Agentic Integration endpoint: Consumed by Hermes to fetch anomalies that have persisted + across the 12-hour evaluation window and satisfied the 4-run rule. + """ + db_path = SERVER_STATE["config"]["db_path"] + window_hours = SERVER_STATE["config"]["evaluation_window_hours"] + min_runs = SERVER_STATE["config"]["min_persistence_runs"] + now = datetime.now(timezone.utc) + + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + cursor.execute(""" + SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status + FROM active_issues + WHERE status = 'VERIFIED' AND run_count >= ? + """, (min_runs,)) + rows = cursor.fetchall() + conn.close() + + report = [] + for r in rows: + last_seen_dt = datetime.fromisoformat(r[8]) + # Only return anomalies active within the evaluation window + if (now - last_seen_dt) <= timedelta(hours=window_hours): + report.append({ + "fingerprint": r[0], + "site": r[1], + "server": r[2], + "signature": r[3], + "severity": r[4], + "message": r[5], + "os_type": r[6], + "first_seen": r[7], + "last_seen": r[8], + "consecutive_runs": r[9], + "evaluation_window": f"{window_hours}h", + "verified": True, + "status": r[10] + }) + + return report + + +@app.get("/api/hermes/all") +def get_all_issues(): + """Diagnostic endpoint to inspect both transient candidate blips and verified anomalies.""" + db_path = SERVER_STATE["config"]["db_path"] + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + cursor.execute(""" + SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status + FROM active_issues + """) + rows = cursor.fetchall() + conn.close() + + return [ + { + "fingerprint": r[0], + "site": r[1], + "server": r[2], + "signature": r[3], + "severity": r[4], + "message": r[5], + "os_type": r[6], + "first_seen": r[7], + "last_seen": r[8], + "run_count": r[9], + "status": r[10] + } + for r in rows + ] + + +@app.get("/health") +def health_check(): + return { + "status": "healthy", + "server_name": SERVER_STATE["config"]["server_name"], + "fingerprint": SERVER_STATE["config"]["server_fingerprint"], + "tcp_port": SERVER_STATE["config"]["tcp_port"], + "hermes_port": SERVER_STATE["config"]["hermes_port"] + } + + +async def run_server(): + """Runs the TCP socket listener and the Hermes REST API concurrently.""" + config = SERVER_STATE["config"] + tcp_host = config["tcp_host"] + tcp_port = int(config["tcp_port"]) + hermes_host = config["hermes_host"] + hermes_port = int(config["hermes_port"]) + + # Start TCP Socket Server + tcp_server = await asyncio.start_server(handle_socket_client, tcp_host, tcp_port) + print(f"[*] LOGAR TCP Socket Server listening on {tcp_host}:{tcp_port}") + + # Start FastAPI / Uvicorn server for Hermes + uv_config = uvicorn.Config(app, host=hermes_host, port=hermes_port, log_level="warning") + uv_server = uvicorn.Server(uv_config) + print(f"[*] Hermes Reporting API available at http://{hermes_host}:{hermes_port}/api/hermes/report") + + await asyncio.gather( + tcp_server.serve_forever(), + uv_server.serve() + ) + + +def main(): + parser = argparse.ArgumentParser(description="LOGAR Cloud Hub & TCP Socket Ingestion Server") + parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to server_config.json") + parser.add_argument("--create-client-config", action="store_true", help="Generate a client config with encryption-only fingerprint and server address") + parser.add_argument("--client-out", default="client_config.json", help="Output file path for generated client config") + parser.add_argument("--server-host", default="127.0.0.1", help="Server address to embed in client config") + parser.add_argument("--server-port", type=int, default=None, help="TCP port to embed in client config") + args = parser.parse_args() + + config = load_or_init_config(args.config) + init_db(config["db_path"]) + + # Load OpenPGP private key into memory + priv_key_obj, _ = pgpy.PGPKey.from_blob(config["private_key"]) + SERVER_STATE["config"] = config + SERVER_STATE["private_key_obj"] = priv_key_obj + + if args.create_client_config: + port = args.server_port or config["tcp_port"] + create_client_config( + server_host=args.server_host, + server_port=port, + output_path=args.client_out, + config_path=args.config + ) + sys.exit(0) + + print("=" * 60) + print(f" LOGAR Server Hub: {config['server_name']}") + print(f" Encryption Fingerprint: {config['server_fingerprint']}") + print(f" Evaluation Window: {config['evaluation_window_hours']} hours | Rule: {config['min_persistence_runs']}+ consecutive runs") + print("=" * 60) + + try: + asyncio.run(run_server()) + except KeyboardInterrupt: + print("\n[!] Server shutting down.") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/Win_Client.py b/Win_Client.py new file mode 100644 index 0000000..722a9d2 --- /dev/null +++ b/Win_Client.py @@ -0,0 +1,209 @@ +import os +import sys +import json +import socket +import struct +import argparse +import warnings +from datetime import datetime, timezone, timedelta + +# Suppress cryptography / pgpy deprecation notices +warnings.filterwarnings("ignore") + +import pgpy + +try: + import win32evtlog +except ImportError: + win32evtlog = None + +CONFIG_FILE_NAME = "client_config.json" + + +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: + return json.load(f) + + +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 = 6): + """ + Scans the Windows Application Event Log backwards for events within the window. + Edge Thinness & Noise Stripping: INFO and DEBUG events are dropped at the source. + """ + 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() + + sev_map = { + 1: "CRITICAL", + 2: "ERROR", + 3: "WARNING" + } + + while True: + events = win32evtlog.ReadEventLog(hand, flags, 0) + if not events: + break + + for event in events: + if event.TimeGenerated < cutoff_time: + break + + # Drop conversational or informational noise (INFO=4, etc.) at source + # Only retain Critical (1), Error (2), and Warning (3) + 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 + }) + + if events[-1].TimeGenerated < cutoff_time: + break + + win32evtlog.CloseEventLog(hand) + return logs + + +def send_encrypted_logs_over_socket(config: dict, logs: list): + """ + Encrypts the payload using the server's OpenPGP public key and streams + over an authenticated TCP socket. Zero local state is maintained on the client. + """ + server_host = config["server_host"] + server_port = int(config["server_port"]) + auth_token = config["auth_token"] + pub_key_armored = config["server_public_key"] + expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper() + + # Load and verify server public key + 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}." + ) + + machine_id = get_machine_identifier() + + # Prepare zero-state candidate batch + payload = { + "server": machine_id, + "timestamp": datetime.now(timezone.utc).isoformat(), + "logs": logs + } + payload_json = json.dumps(payload) + + # Encrypt payload with server's encryption-only key + pgp_msg = pgpy.PGPMessage.new(payload_json) + encrypted_msg = pub_key.encrypt(pgp_msg) + encrypted_armored = str(encrypted_msg) + + # Envelope with socket authentication header + envelope = { + "auth_token": auth_token, + "timestamp": datetime.now(timezone.utc).isoformat(), + "encrypted_payload": encrypted_armored + } + envelope_bytes = json.dumps(envelope).encode("utf-8") + + # Connect over TCP socket and transmit with 4-byte length prefix framing + 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)) + + # Send frame: length (4 bytes big-endian) + envelope + frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes + sock.sendall(frame) + + # Receive response length + 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 main(): + parser = argparse.ArgumentParser(description="LOGAR Windows Edge Log Forwarder (Zero State)") + parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json") + parser.add_argument("--hours", type=int, default=6, help="Lookback window in hours for event logs") + args = parser.parse_args() + + try: + config = load_config(args.config) + except Exception as e: + print(f"[!] Configuration error: {e}") + sys.exit(1) + + print(f"[*] Scanning Windows Application event log for candidate anomalies (last {args.hours} hours)...") + candidate_logs = get_recent_windows_logs(hours=args.hours) + print(f"[*] Found {len(candidate_logs)} candidate anomalies (noise stripped at source).") + + try: + send_encrypted_logs_over_socket(config, candidate_logs) + except Exception as e: + print(f"[!] Failed to stream logs to server: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/out/linux_client/Linux_Client.bin b/out/linux_client/Linux_Client.bin new file mode 100644 index 0000000..279f601 Binary files /dev/null and b/out/linux_client/Linux_Client.bin differ diff --git a/out/linux_client/Linux_Client.py b/out/linux_client/Linux_Client.py new file mode 100644 index 0000000..4e8cfec --- /dev/null +++ b/out/linux_client/Linux_Client.py @@ -0,0 +1,194 @@ +import os +import sys +import json +import socket +import struct +import argparse +import subprocess +import warnings +from datetime import datetime, timezone + +# Suppress cryptography / pgpy deprecation notices +warnings.filterwarnings("ignore") + +import pgpy + +CONFIG_FILE_NAME = "client_config.json" + + +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: + return json.load(f) + + +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() + + 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 + + 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 = 6): + """ + Collects warnings and errors from systemd journalctl over the lookback window. + Edge Thinness: Drops INFO and DEBUG entries at the source. + """ + cmd = ["journalctl", "--since", f"{hours} hours ago", "-p", "warning", "--output=json"] + 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() + + for line in result.stdout.splitlines(): + line_str = line.strip() + if not line_str: + continue + try: + entry = json.loads(line_str) + priority = str(entry.get("PRIORITY", "4")) + if int(priority) > 4: + continue + + sev = "WARNING" if priority == "4" else "ERROR" + 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 + + return logs + + +def send_encrypted_logs_over_socket(config: dict, logs: list): + """ + Encrypts the payload using the server's OpenPGP public key and streams + over an authenticated TCP socket. Zero local state is maintained on the client. + """ + server_host = config["server_host"] + server_port = int(config["server_port"]) + auth_token = config["auth_token"] + pub_key_armored = config["server_public_key"] + 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}." + ) + + machine_id = get_machine_identifier() + + 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 main(): + parser = argparse.ArgumentParser(description="LOGAR Linux Edge Log Forwarder (Zero State)") + parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json") + parser.add_argument("--hours", type=int, default=6, help="Lookback window in hours for journalctl logs") + args = parser.parse_args() + + try: + config = load_config(args.config) + except Exception as e: + print(f"[!] Configuration error: {e}") + sys.exit(1) + + machine_id = get_machine_identifier() + print(f"[*] Edge Forwarder Node: {machine_id}") + print(f"[*] Scanning Linux journalctl for candidate anomalies (last {args.hours} hours)...") + candidate_logs = get_recent_linux_logs(hours=args.hours) + print(f"[*] Found {len(candidate_logs)} candidate anomalies (noise stripped at source).") + + try: + send_encrypted_logs_over_socket(config, candidate_logs) + except Exception as e: + print(f"[!] Failed to stream logs to server: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/out/linux_client/README.md b/out/linux_client/README.md new file mode 100644 index 0000000..d536983 --- /dev/null +++ b/out/linux_client/README.md @@ -0,0 +1,62 @@ +# LOGAR Linux Edge Forwarder + +Lightweight edge log forwarder for Linux servers running systemd. + +## Features +- **Zero Local State**: No local SQLite database or state tracking on the edge server. +- **Edge Noise Stripping**: Strips conversational/informational noise (`INFO`, `DEBUG`) directly at the source via `journalctl -p warning`. +- **End-to-End OpenPGP Encryption**: Encrypts logs using the server's public key; decrypted exclusively on the cloud hub. +- **Authenticated TCP Socket**: Direct, low-overhead TCP streaming with token authentication. +- **No GPG Binary Required**: Pure-Python implementation (`pgpy` + `cryptography`). + +## Installation +```bash +python3 -m pip install -r requirements.txt +``` + +## Configuration +Place `client_config.json` generated by the server (`Server.py --create-client-config`) in the same directory as `Linux_Client.py`. + +## Running the Forwarder +```bash +python3 Linux_Client.py --hours 6 +``` + +## Cron / Systemd Timer Deployment +### Option A: Cron Job (Every 3 hours) +```bash +0 */3 * * * cd /opt/logar && /usr/bin/python3 Linux_Client.py --hours 6 >> /var/log/logar_client.log 2>&1 +``` + +### Option B: Systemd Service & Timer +1. Create `/etc/systemd/system/logar-forwarder.service`: +```ini +[Unit] +Description=LOGAR Edge Forwarder +After=network.target + +[Service] +Type=oneshot +WorkingDirectory=/opt/logar +ExecStart=/usr/bin/python3 /opt/logar/Linux_Client.py --hours 6 +``` + +2. Create `/etc/systemd/system/logar-forwarder.timer`: +```ini +[Unit] +Description=Run LOGAR Edge Forwarder every 3 hours + +[Timer] +OnBootSec=5min +OnUnitActiveSec=3h +Persistent=true + +[Install] +WantedBy=timers.target +``` + +3. Enable and start: +```bash +sudo systemctl daemon-reload +sudo systemctl enable --now logar-forwarder.timer +``` diff --git a/out/linux_client/build_bin.sh b/out/linux_client/build_bin.sh new file mode 100644 index 0000000..1b63e54 --- /dev/null +++ b/out/linux_client/build_bin.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +# Build script to compile Linux_Client into a standalone native ELF binary on Linux +set -e + +echo "[*] Installing build requirements..." +pip3 install pyinstaller pgpy cryptography standard-imghdr + +echo "[*] Compiling Linux_Client native binary..." +pyinstaller --onefile --clean --name Linux_Client.bin Linux_Client.py + +echo "[+] Compilation successful: dist/Linux_Client.bin" diff --git a/out/linux_client/client_config.sample.json b/out/linux_client/client_config.sample.json new file mode 100644 index 0000000..45dc8ff --- /dev/null +++ b/out/linux_client/client_config.sample.json @@ -0,0 +1,8 @@ +{ + "server_host": "192.168.1.100", + "server_port": 9443, + "server_fingerprint": "PASTE_SERVER_FINGERPRINT_HERE", + "server_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n...\n-----END PGP PUBLIC KEY BLOCK-----\n", + "auth_token": "PASTE_AUTH_TOKEN_HERE", + "site_name": "Frankfurt-DC" +} diff --git a/out/linux_client/requirements.txt b/out/linux_client/requirements.txt new file mode 100644 index 0000000..e7f7058 --- /dev/null +++ b/out/linux_client/requirements.txt @@ -0,0 +1,3 @@ +pgpy>=0.6.0 +standard-imghdr>=3.13.0; python_version >= "3.13" +cryptography>=42.0.0 diff --git a/out/linux_client/test/test_linux_client.py b/out/linux_client/test/test_linux_client.py new file mode 100644 index 0000000..90cc1ae --- /dev/null +++ b/out/linux_client/test/test_linux_client.py @@ -0,0 +1,107 @@ +import os +import sys +import json +import struct +import unittest +import warnings + +warnings.filterwarnings("ignore") + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +import Linux_Client +import pgpy +from pgpy.constants import PubKeyAlgorithm, KeyFlags, HashAlgorithm, SymmetricKeyAlgorithm, CompressionAlgorithm + + +class TestLinuxClientComponent(unittest.TestCase): + def setUp(self): + self.dummy_config = "test_linux_client_config.json" + # Generate dummy PGP key for testing + key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048) + uid = pgpy.PGPUID.new("TestHub") + key.add_uid( + uid, + usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage}, + hashes=[HashAlgorithm.SHA256], + ciphers=[SymmetricKeyAlgorithm.AES256], + compression=[CompressionAlgorithm.Uncompressed] + ) + self.server_priv = key + self.server_pub = key.pubkey + self.fingerprint = str(key.pubkey.fingerprint) + + with open(self.dummy_config, "w", encoding="utf-8") as f: + json.dump({ + "server_host": "127.0.0.1", + "server_port": 9443, + "server_fingerprint": self.fingerprint, + "server_public_key": str(self.server_pub), + "auth_token": "secret-test-token" + }, f) + + def tearDown(self): + if os.path.exists(self.dummy_config): + try: + os.remove(self.dummy_config) + except Exception: + pass + + def test_client_config_anonymity(self): + config = Linux_Client.load_config(self.dummy_config) + self.assertNotIn("server_name", config) + self.assertNotIn("name", config) + self.assertNotIn("site_name", config) + self.assertEqual(config["server_fingerprint"], self.fingerprint) + + def test_get_machine_identifier(self): + machine_id = Linux_Client.get_machine_identifier() + self.assertIsInstance(machine_id, str) + self.assertGreater(len(machine_id), 0) + self.assertNotEqual(machine_id, "localhost") + + def test_journalctl_parsing_and_priority_filter(self): + sample_journal_lines = [ + json.dumps({"PRIORITY": "3", "SYSLOG_IDENTIFIER": "sshd", "MESSAGE": "Failed password for root"}), + json.dumps({"PRIORITY": "4", "SYSLOG_IDENTIFIER": "systemd", "MESSAGE": "Unit entered failed state"}), + json.dumps({"PRIORITY": "6", "SYSLOG_IDENTIFIER": "cron", "MESSAGE": "Informational session opened"}), + ] + logs = [] + machine_id = Linux_Client.get_machine_identifier() + for line_str in sample_journal_lines: + entry = json.loads(line_str) + priority = str(entry.get("PRIORITY", "4")) + if int(priority) > 4: + continue + sev = "WARNING" if priority == "4" else "ERROR" + logs.append({ + "server": machine_id, + "os_type": "linux", + "signature": entry.get("SYSLOG_IDENTIFIER", "unknown"), + "severity": sev, + "message": entry.get("MESSAGE", "") + }) + + # Priority 6 must be stripped (INFO noise) + self.assertEqual(len(logs), 2) + self.assertEqual(logs[0]["severity"], "ERROR") + self.assertEqual(logs[1]["severity"], "WARNING") + + def test_encryption_and_decryption(self): + config = Linux_Client.load_config(self.dummy_config) + pub_key, _ = pgpy.PGPKey.from_blob(config["server_public_key"]) + payload = { + "server": Linux_Client.get_machine_identifier(), + "logs": [{"signature": "kernel", "severity": "ERROR", "message": "Kernel panic - not syncing"}] + } + msg = pgpy.PGPMessage.new(json.dumps(payload)) + enc = pub_key.encrypt(msg) + self.assertTrue(str(enc).startswith("-----BEGIN PGP MESSAGE-----")) + + dec = self.server_priv.decrypt(enc) + restored = json.loads(dec.message) + self.assertEqual(restored["logs"][0]["signature"], "kernel") + + +if __name__ == "__main__": + unittest.main() diff --git a/out/server/README.md b/out/server/README.md new file mode 100644 index 0000000..4384bd0 --- /dev/null +++ b/out/server/README.md @@ -0,0 +1,36 @@ +# LOGAR Server Hub + +Central Python/TCP ingestion server for the LOGAR Log Analysis System. + +## Features +- **Zero External GPG Requirement**: Uses pure-Python OpenPGP (`pgpy` + `cryptography`), no native GnuPG binary needed. +- **First-Run Key & Config Auto-generation**: Generates OpenPGP keypairs, auth tokens, and `server_config.json` automatically on first launch. +- **Client Config Exporter**: Generates `client_config.json` bundles containing the server's encryption-only fingerprint and address. +- **Cloud-Side Temporal Persistence**: SQLite database tracking candidate anomalies over 12-hour evaluation windows. +- **4-Run Persistence Rule**: Filters out transient infrastructure blips, promoting issues to `VERIFIED` anomalies only after persisting across $\ge 4$ runs. +- **Agentic Hermes Endpoint**: REST API (`GET /api/hermes/report`) providing verified system artifacts for Hermes agent alerts. + +## Installation +```bash +pip install -r requirements.txt +``` + +## Running the Server +```bash +# Starts both the TCP socket listener (port 9443) and the Hermes API (port 8443) +python Server.py +``` + +## Generating Client Configurations +To deploy edge forwarders, generate a client config file: +```bash +python Server.py --create-client-config --server-host --server-port 9443 --site-name "Frankfurt-DC" --client-out client_config.json +``` +Copy the generated `client_config.json` into the deployment directory of `Win_Client.py` or `Linux_Client.py`. + +## Hermes Agent Integration +Hermes queries the verified anomalies via: +``` +GET http://:8443/api/hermes/report +``` +Only issues meeting the 4-run persistence rule within the active 12-hour evaluation window are returned. diff --git a/out/server/Server.exe b/out/server/Server.exe new file mode 100644 index 0000000..96f3586 Binary files /dev/null and b/out/server/Server.exe differ diff --git a/out/server/Server.py b/out/server/Server.py new file mode 100644 index 0000000..e3071e8 --- /dev/null +++ b/out/server/Server.py @@ -0,0 +1,437 @@ +# Copy of Server.py without site_name in client_config +import os +import sys +import json +import uuid +import struct +import socket +import sqlite3 +import argparse +import asyncio +import secrets +import warnings +from datetime import datetime, timezone, timedelta +from typing import Dict, Any, List, Optional + +# Suppress cryptography / pgpy deprecation notices for a clean terminal output +warnings.filterwarnings("ignore") + +import pgpy +from pgpy.constants import ( + PubKeyAlgorithm, + KeyFlags, + HashAlgorithm, + SymmetricKeyAlgorithm, + CompressionAlgorithm +) +from fastapi import FastAPI, HTTPException +import uvicorn + +CONFIG_FILE_NAME = "server_config.json" +DEFAULT_DB_FILE = "logar_state.db" +EVALUATION_WINDOW_HOURS = 12 +RUN_THRESHOLD = 4 + +app = FastAPI(title="LOGAR Cloud Ingestion & Hermes Hub", version="2.0.0") + +SERVER_STATE: Dict[str, Any] = {} + + +def generate_server_keypair(server_name: str): + """Generates an OpenPGP RSA 2048 key with encryption capability.""" + key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048) + uid = pgpy.PGPUID.new(server_name) + key.add_uid( + uid, + usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage}, + hashes=[HashAlgorithm.SHA256], + ciphers=[SymmetricKeyAlgorithm.AES256], + compression=[CompressionAlgorithm.Uncompressed] + ) + private_key_armored = str(key) + public_key_armored = str(key.pubkey) + fingerprint = str(key.pubkey.fingerprint) + return private_key_armored, public_key_armored, fingerprint + + +def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]: + """Loads existing server_config.json or creates a new one on first run.""" + if os.path.exists(config_path): + print(f"[*] Loading server configuration from: {os.path.abspath(config_path)}") + with open(config_path, "r", encoding="utf-8") as f: + config = json.load(f) + return config + + print(f"[!] Config '{config_path}' not found. Initializing first-run configuration...") + server_name = "LOGAR-Cloud-Hub" + private_key, public_key, fingerprint = generate_server_keypair(server_name) + auth_token = secrets.token_hex(24) + + config = { + "server_name": server_name, + "tcp_host": "0.0.0.0", + "tcp_port": 9443, + "hermes_host": "0.0.0.0", + "hermes_port": 8443, + "auth_token": auth_token, + "db_path": DEFAULT_DB_FILE, + "evaluation_window_hours": EVALUATION_WINDOW_HOURS, + "min_persistence_runs": RUN_THRESHOLD, + "server_fingerprint": fingerprint, + "public_key": public_key, + "private_key": private_key + } + + with open(config_path, "w", encoding="utf-8") as f: + json.dump(config, f, indent=2) + + print(f"[+] Successfully generated new server config and OpenPGP keypair.") + print(f"[+] Server Encryption Fingerprint: {fingerprint}") + print(f"[+] Saved to: {os.path.abspath(config_path)}") + return config + + +def create_client_config( + server_host: str, + server_port: int, + output_path: str, + config_path: str = CONFIG_FILE_NAME +) -> Dict[str, Any]: + """Creates a client configuration file containing the server address, auth token, and encryption-only key/fingerprint.""" + server_conf = load_or_init_config(config_path) + + client_conf = { + "server_host": server_host, + "server_port": server_port, + "server_fingerprint": server_conf["server_fingerprint"], + "server_public_key": server_conf["public_key"], + "auth_token": server_conf["auth_token"] + } + + out_dir = os.path.dirname(os.path.abspath(output_path)) + if out_dir and not os.path.exists(out_dir): + os.makedirs(out_dir, exist_ok=True) + + with open(output_path, "w", encoding="utf-8") as f: + json.dump(client_conf, f, indent=2) + + print(f"[+] Client configuration successfully written to: {os.path.abspath(output_path)}") + print(f" - Server Target: {server_host}:{server_port}") + print(f" - Encryption Fingerprint: {server_conf['server_fingerprint']}") + return client_conf + + +def init_db(db_path: str): + """Initializes the SQLite schema for multi-run temporal tracking.""" + conn = sqlite3.connect(db_path) + conn.execute(""" + CREATE TABLE IF NOT EXISTS active_issues ( + fingerprint TEXT PRIMARY KEY, + site_name TEXT, + server TEXT, + signature TEXT, + severity TEXT, + message TEXT, + os_type TEXT, + first_seen TEXT, + last_seen TEXT, + run_count INTEGER, + status TEXT, + last_run_id TEXT + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS ingest_runs ( + run_id TEXT PRIMARY KEY, + site_name TEXT, + server TEXT, + timestamp TEXT, + log_count INTEGER + ) + """) + conn.commit() + conn.close() + + +def process_ingested_logs(payload: Dict[str, Any], db_path: str, window_hours: int, min_runs: int) -> Dict[str, Any]: + """ + Evaluates candidate issues against the 12-hour evaluation window and 4-run rule. + Zero-state clients send raw candidate entries; this engine handles temporal state. + """ + client_server = payload.get("server", "unknown-host") + site_name = payload.get("site_name") or (client_server.split(".", 1)[1] if "." in client_server else "default") + logs = payload.get("logs", []) + run_id = str(uuid.uuid4()) + now = datetime.now(timezone.utc) + now_iso = now.isoformat() + + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + + cursor.execute( + "INSERT INTO ingest_runs (run_id, site_name, server, timestamp, log_count) VALUES (?, ?, ?, ?, ?)", + (run_id, site_name, client_server, now_iso, len(logs)) + ) + + processed_count = 0 + promoted_to_verified = 0 + + for log in logs: + severity = str(log.get("severity", "WARNING")).upper() + if severity in ["INFO", "DEBUG"]: + continue + + signature = log.get("signature", "unknown") + server = log.get("server", client_server) + message = log.get("message", "") + os_type = log.get("os_type", "unknown") + fp = f"{site_name}:{server}:{signature}" + + cursor.execute( + "SELECT run_count, first_seen, last_seen, status, last_run_id FROM active_issues WHERE fingerprint = ?", + (fp,) + ) + row = cursor.fetchone() + + if row: + run_count, first_seen_str, last_seen_str, current_status, last_run_id = row + try: + last_seen_dt = datetime.fromisoformat(last_seen_str) + except Exception: + last_seen_dt = now + + if (now - last_seen_dt) > timedelta(hours=window_hours): + new_runs = 1 + new_first_seen = now_iso + new_status = "TRANSIENT" + else: + if last_run_id != run_id: + new_runs = run_count + 1 + else: + new_runs = run_count + new_first_seen = first_seen_str + new_status = "VERIFIED" if new_runs >= min_runs else "TRANSIENT" + + if new_status == "VERIFIED" and current_status != "VERIFIED": + promoted_to_verified += 1 + + cursor.execute(""" + UPDATE active_issues + SET run_count = ?, last_seen = ?, first_seen = ?, status = ?, last_run_id = ?, message = ?, severity = ? + WHERE fingerprint = ? + """, (new_runs, now_iso, new_first_seen, new_status, run_id, message, severity, fp)) + else: + initial_status = "VERIFIED" if 1 >= min_runs else "TRANSIENT" + cursor.execute(""" + INSERT INTO active_issues + (fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status, last_run_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, (fp, site_name, server, signature, severity, message, os_type, now_iso, now_iso, 1, initial_status, run_id)) + + processed_count += 1 + + conn.commit() + conn.close() + + return { + "status": "success", + "run_id": run_id, + "processed": processed_count, + "promoted_verified": promoted_to_verified + } + + +async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + try: + length_bytes = await reader.readexactly(4) + length = struct.unpack(">I", length_bytes)[0] + if length <= 0 or length > 10 * 1024 * 1024: + raise ValueError(f"Invalid frame size: {length}") + + payload_bytes = await reader.readexactly(length) + envelope = json.loads(payload_bytes.decode("utf-8")) + + expected_token = SERVER_STATE["config"]["auth_token"] + provided_token = envelope.get("auth_token") + if not secrets.compare_digest(str(provided_token), str(expected_token)): + err_msg = json.dumps({"status": "error", "message": "Authentication failed"}).encode("utf-8") + writer.write(struct.pack(">I", len(err_msg)) + err_msg) + await writer.drain() + writer.close() + await writer.wait_closed() + return + + encrypted_armored = envelope.get("encrypted_payload", "") + pgp_msg = pgpy.PGPMessage.from_blob(encrypted_armored) + priv_key = SERVER_STATE["private_key_obj"] + decrypted_obj = priv_key.decrypt(pgp_msg) + decrypted_json_str = decrypted_obj.message + log_payload = json.loads(decrypted_json_str) + + res = process_ingested_logs( + log_payload, + db_path=SERVER_STATE["config"]["db_path"], + window_hours=SERVER_STATE["config"]["evaluation_window_hours"], + min_runs=SERVER_STATE["config"]["min_persistence_runs"] + ) + + resp_bytes = json.dumps(res).encode("utf-8") + writer.write(struct.pack(">I", len(resp_bytes)) + resp_bytes) + await writer.drain() + + except Exception as e: + err = json.dumps({"status": "error", "message": str(e)}).encode("utf-8") + try: + writer.write(struct.pack(">I", len(err)) + err) + await writer.drain() + except Exception: + pass + finally: + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + + +@app.get("/api/hermes/report") +def get_hermes_report(): + db_path = SERVER_STATE["config"]["db_path"] + window_hours = SERVER_STATE["config"]["evaluation_window_hours"] + min_runs = SERVER_STATE["config"]["min_persistence_runs"] + now = datetime.now(timezone.utc) + + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + cursor.execute(""" + SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status + FROM active_issues + WHERE status = 'VERIFIED' AND run_count >= ? + """, (min_runs,)) + rows = cursor.fetchall() + conn.close() + + report = [] + for r in rows: + last_seen_dt = datetime.fromisoformat(r[8]) + if (now - last_seen_dt) <= timedelta(hours=window_hours): + report.append({ + "fingerprint": r[0], + "site": r[1], + "server": r[2], + "signature": r[3], + "severity": r[4], + "message": r[5], + "os_type": r[6], + "first_seen": r[7], + "last_seen": r[8], + "consecutive_runs": r[9], + "evaluation_window": f"{window_hours}h", + "verified": True, + "status": r[10] + }) + + return report + + +@app.get("/api/hermes/all") +def get_all_issues(): + db_path = SERVER_STATE["config"]["db_path"] + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + cursor.execute(""" + SELECT fingerprint, site_name, server, signature, severity, message, os_type, first_seen, last_seen, run_count, status + FROM active_issues + """) + rows = cursor.fetchall() + conn.close() + + return [ + { + "fingerprint": r[0], + "site": r[1], + "server": r[2], + "signature": r[3], + "severity": r[4], + "message": r[5], + "os_type": r[6], + "first_seen": r[7], + "last_seen": r[8], + "run_count": r[9], + "status": r[10] + } + for r in rows + ] + + +@app.get("/health") +def health_check(): + return { + "status": "healthy", + "server_name": SERVER_STATE["config"]["server_name"], + "fingerprint": SERVER_STATE["config"]["server_fingerprint"], + "tcp_port": SERVER_STATE["config"]["tcp_port"], + "hermes_port": SERVER_STATE["config"]["hermes_port"] + } + + +async def run_server(): + config = SERVER_STATE["config"] + tcp_host = config["tcp_host"] + tcp_port = int(config["tcp_port"]) + hermes_host = config["hermes_host"] + hermes_port = int(config["hermes_port"]) + + tcp_server = await asyncio.start_server(handle_socket_client, tcp_host, tcp_port) + print(f"[*] LOGAR TCP Socket Server listening on {tcp_host}:{tcp_port}") + + uv_config = uvicorn.Config(app, host=hermes_host, port=hermes_port, log_level="warning") + uv_server = uvicorn.Server(uv_config) + print(f"[*] Hermes Reporting API available at http://{hermes_host}:{hermes_port}/api/hermes/report") + + await asyncio.gather( + tcp_server.serve_forever(), + uv_server.serve() + ) + + +def main(): + parser = argparse.ArgumentParser(description="LOGAR Cloud Hub & TCP Socket Ingestion Server") + parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to server_config.json") + parser.add_argument("--create-client-config", action="store_true", help="Generate a client config with encryption-only fingerprint and server address") + parser.add_argument("--client-out", default="client_config.json", help="Output file path for generated client config") + parser.add_argument("--server-host", default="127.0.0.1", help="Server address to embed in client config") + parser.add_argument("--server-port", type=int, default=None, help="TCP port to embed in client config") + args = parser.parse_args() + + config = load_or_init_config(args.config) + init_db(config["db_path"]) + + priv_key_obj, _ = pgpy.PGPKey.from_blob(config["private_key"]) + SERVER_STATE["config"] = config + SERVER_STATE["private_key_obj"] = priv_key_obj + + if args.create_client_config: + port = args.server_port or config["tcp_port"] + create_client_config( + server_host=args.server_host, + server_port=port, + output_path=args.client_out, + config_path=args.config + ) + sys.exit(0) + + print("=" * 60) + print(f" LOGAR Server Hub: {config['server_name']}") + print(f" Encryption Fingerprint: {config['server_fingerprint']}") + print(f" Evaluation Window: {config['evaluation_window_hours']} hours | Rule: {config['min_persistence_runs']}+ consecutive runs") + print("=" * 60) + + try: + asyncio.run(run_server()) + except KeyboardInterrupt: + print("\n[!] Server shutting down.") + + +if __name__ == "__main__": + main() diff --git a/out/server/requirements.txt b/out/server/requirements.txt new file mode 100644 index 0000000..e9c6032 --- /dev/null +++ b/out/server/requirements.txt @@ -0,0 +1,6 @@ +pgpy>=0.6.0 +standard-imghdr>=3.13.0; python_version >= "3.13" +cryptography>=42.0.0 +fastapi>=0.110.0 +uvicorn>=0.28.0 +pydantic>=2.6.0 diff --git a/out/server/server_config.sample.json b/out/server/server_config.sample.json new file mode 100644 index 0000000..153d792 --- /dev/null +++ b/out/server/server_config.sample.json @@ -0,0 +1,14 @@ +{ + "server_name": "LOGAR-Cloud-Hub", + "tcp_host": "0.0.0.0", + "tcp_port": 9443, + "hermes_host": "0.0.0.0", + "hermes_port": 8443, + "auth_token": "replace_with_secure_random_hex_token", + "db_path": "logar_state.db", + "evaluation_window_hours": 12, + "min_persistence_runs": 4, + "server_fingerprint": "AUTO_GENERATED_ON_FIRST_RUN", + "public_key": "AUTO_GENERATED_ON_FIRST_RUN", + "private_key": "AUTO_GENERATED_ON_FIRST_RUN" +} diff --git a/out/server/test/test_server.py b/out/server/test/test_server.py new file mode 100644 index 0000000..d1e0dde --- /dev/null +++ b/out/server/test/test_server.py @@ -0,0 +1,119 @@ +import os +import sys +import json +import socket +import struct +import sqlite3 +import unittest +import urllib.request +import warnings +from datetime import datetime, timezone, timedelta + +warnings.filterwarnings("ignore") + +# Ensure parent directory is in path to import Server +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +import Server +import pgpy + + +class TestServerComponent(unittest.TestCase): + def setUp(self): + self.test_db = "test_server_state.db" + self.test_config = "test_server_config.json" + if os.path.exists(self.test_db): + os.remove(self.test_db) + if os.path.exists(self.test_config): + os.remove(self.test_config) + + def tearDown(self): + if os.path.exists(self.test_db): + try: + os.remove(self.test_db) + except Exception: + pass + if os.path.exists(self.test_config): + try: + os.remove(self.test_config) + except Exception: + pass + + def test_first_run_config_and_keypair_generation(self): + config = Server.load_or_init_config(self.test_config) + self.assertTrue(os.path.exists(self.test_config)) + self.assertIn("server_fingerprint", config) + self.assertIn("public_key", config) + self.assertIn("private_key", config) + self.assertIn("auth_token", config) + self.assertNotIn("site_name", config) + + # Verify keypair + priv_key, _ = pgpy.PGPKey.from_blob(config["private_key"]) + pub_key, _ = pgpy.PGPKey.from_blob(config["public_key"]) + self.assertEqual(str(pub_key.fingerprint), config["server_fingerprint"]) + + def test_create_client_config(self): + Server.load_or_init_config(self.test_config) + client_out = "test_client_out.json" + try: + client_conf = Server.create_client_config( + server_host="10.0.0.1", + server_port=9443, + output_path=client_out, + config_path=self.test_config + ) + self.assertTrue(os.path.exists(client_out)) + self.assertEqual(client_conf["server_host"], "10.0.0.1") + self.assertEqual(client_conf["server_port"], 9443) + # Verify no machine name or site_name is included + self.assertNotIn("server_name", client_conf) + self.assertNotIn("name", client_conf) + self.assertNotIn("site_name", client_conf) + finally: + if os.path.exists(client_out): + os.remove(client_out) + + def test_4_run_rule_and_12h_window(self): + Server.init_db(self.test_db) + log_entry = { + "server": "app-worker-01.corp.local", + "signature": "PostgresConnTimeout", + "severity": "ERROR", + "message": "Connection to database pool timed out after 30s", + "os_type": "linux" + } + payload = { + "server": "app-worker-01.corp.local", + "logs": [log_entry] + } + + # Runs 1 to 3: should remain TRANSIENT + for run_idx in range(1, 4): + res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4) + self.assertEqual(res["status"], "success") + self.assertEqual(res["promoted_verified"], 0) + + conn = sqlite3.connect(self.test_db) + c = conn.cursor() + c.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", ("PostgresConnTimeout",)) + row = c.fetchone() + conn.close() + self.assertEqual(row[0], 3) + self.assertEqual(row[1], "TRANSIENT") + + # Run 4: promotes to VERIFIED! + res4 = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4) + self.assertEqual(res4["promoted_verified"], 1) + + conn = sqlite3.connect(self.test_db) + c = conn.cursor() + c.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", ("PostgresConnTimeout",)) + row = c.fetchone() + conn.close() + self.assertEqual(row[0], 4) + self.assertEqual(row[1], "VERIFIED") + + +if __name__ == "__main__": + unittest.main() diff --git a/out/win_client/README.md b/out/win_client/README.md new file mode 100644 index 0000000..8b155c5 --- /dev/null +++ b/out/win_client/README.md @@ -0,0 +1,31 @@ +# LOGAR Windows Edge Forwarder + +Lightweight edge log forwarder for Windows servers. + +## Features +- **Zero Local State**: No local database or state tracking. Forwarder simply scans recent logs and streams candidates. +- **Edge Noise Stripping**: Strips conversational/informational noise (INFO, DEBUG, Audit) at the source. +- **End-to-End OpenPGP Encryption**: Encrypts logs using the server's public key so that only the server can decrypt them. +- **Authenticated TCP Socket**: Connects directly via raw TCP framing with token verification. +- **No GPG Binary Required**: Pure-Python cryptography (`pgpy` + `cryptography`). + +## Installation +```powershell +python -m pip install -r requirements.txt +``` + +## Configuration +Place the `client_config.json` generated by the server (`Server.py --create-client-config`) in the same directory as `Win_Client.py`. + +## Running the Forwarder +```powershell +python Win_Client.py --hours 6 +``` + +## Scheduled Task Deployment +To run periodically via Windows Task Scheduler (e.g., every 3 hours): +```powershell +$Action = New-ScheduledTaskAction -Execute "python.exe" -Argument "C:\LOGAR\Win_Client.py --hours 6" -WorkingDirectory "C:\LOGAR" +$Trigger = New-ScheduledTaskTrigger -Once -At (Get-Date) -RepetitionInterval (New-TimeSpan -Hours 3) +Register-ScheduledTask -TaskName "LOGAR_Windows_Forwarder" -Action $Action -Trigger $Trigger -Description "LOGAR Edge Forwarder" +``` diff --git a/out/win_client/Win_Client.exe b/out/win_client/Win_Client.exe new file mode 100644 index 0000000..2c1f974 Binary files /dev/null and b/out/win_client/Win_Client.exe differ diff --git a/out/win_client/Win_Client.py b/out/win_client/Win_Client.py new file mode 100644 index 0000000..90d1a87 --- /dev/null +++ b/out/win_client/Win_Client.py @@ -0,0 +1,211 @@ +import os +import sys +import json +import socket +import struct +import argparse +import warnings +from datetime import datetime, timezone, timedelta + +# Suppress cryptography / pgpy deprecation notices +warnings.filterwarnings("ignore") + +import pgpy + +try: + import win32evtlog +except ImportError: + win32evtlog = None + +CONFIG_FILE_NAME = "client_config.json" + + +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: + return json.load(f) + + +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 = 6): + """ + Scans the Windows Application Event Log backwards for events within the window. + Edge Thinness & Noise Stripping: INFO and DEBUG events are dropped at the source. + """ + 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() + + sev_map = { + 1: "CRITICAL", + 2: "ERROR", + 3: "WARNING" + } + + while True: + events = win32evtlog.ReadEventLog(hand, flags, 0) + if not events: + break + + for event in events: + if event.TimeGenerated < cutoff_time: + break + + # Drop conversational or informational noise (INFO=4, etc.) at source + # Only retain Critical (1), Error (2), and Warning (3) + 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 + }) + + if events[-1].TimeGenerated < cutoff_time: + break + + win32evtlog.CloseEventLog(hand) + return logs + + +def send_encrypted_logs_over_socket(config: dict, logs: list): + """ + Encrypts the payload using the server's OpenPGP public key and streams + over an authenticated TCP socket. Zero local state is maintained on the client. + """ + server_host = config["server_host"] + server_port = int(config["server_port"]) + auth_token = config["auth_token"] + pub_key_armored = config["server_public_key"] + expected_fp = config.get("server_fingerprint", "").replace(" ", "").upper() + + # Load and verify server public key + 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}." + ) + + machine_id = get_machine_identifier() + + # Prepare zero-state candidate batch + payload = { + "server": machine_id, + "timestamp": datetime.now(timezone.utc).isoformat(), + "logs": logs + } + payload_json = json.dumps(payload) + + # Encrypt payload with server's encryption-only key + pgp_msg = pgpy.PGPMessage.new(payload_json) + encrypted_msg = pub_key.encrypt(pgp_msg) + encrypted_armored = str(encrypted_msg) + + # Envelope with socket authentication header + envelope = { + "auth_token": auth_token, + "timestamp": datetime.now(timezone.utc).isoformat(), + "encrypted_payload": encrypted_armored + } + envelope_bytes = json.dumps(envelope).encode("utf-8") + + # Connect over TCP socket and transmit with 4-byte length prefix framing + 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)) + + # Send frame: length (4 bytes big-endian) + envelope + frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes + sock.sendall(frame) + + # Receive response length + 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 main(): + parser = argparse.ArgumentParser(description="LOGAR Windows Edge Log Forwarder (Zero State)") + parser.add_argument("--config", default=CONFIG_FILE_NAME, help="Path to client_config.json") + parser.add_argument("--hours", type=int, default=6, help="Lookback window in hours for event logs") + args = parser.parse_args() + + try: + config = load_config(args.config) + except Exception as e: + print(f"[!] Configuration error: {e}") + sys.exit(1) + + machine_id = get_machine_identifier() + print(f"[*] Edge Forwarder Node: {machine_id}") + print(f"[*] Scanning Windows Application event log for candidate anomalies (last {args.hours} hours)...") + candidate_logs = get_recent_windows_logs(hours=args.hours) + print(f"[*] Found {len(candidate_logs)} candidate anomalies (noise stripped at source).") + + try: + send_encrypted_logs_over_socket(config, candidate_logs) + except Exception as e: + print(f"[!] Failed to stream logs to server: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/out/win_client/client_config.sample.json b/out/win_client/client_config.sample.json new file mode 100644 index 0000000..45dc8ff --- /dev/null +++ b/out/win_client/client_config.sample.json @@ -0,0 +1,8 @@ +{ + "server_host": "192.168.1.100", + "server_port": 9443, + "server_fingerprint": "PASTE_SERVER_FINGERPRINT_HERE", + "server_public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n...\n-----END PGP PUBLIC KEY BLOCK-----\n", + "auth_token": "PASTE_AUTH_TOKEN_HERE", + "site_name": "Frankfurt-DC" +} diff --git a/out/win_client/requirements.txt b/out/win_client/requirements.txt new file mode 100644 index 0000000..bfbe56a --- /dev/null +++ b/out/win_client/requirements.txt @@ -0,0 +1,4 @@ +pgpy>=0.6.0 +standard-imghdr>=3.13.0; python_version >= "3.13" +cryptography>=42.0.0 +pywin32>=306 diff --git a/out/win_client/test/test_win_client.py b/out/win_client/test/test_win_client.py new file mode 100644 index 0000000..a999570 --- /dev/null +++ b/out/win_client/test/test_win_client.py @@ -0,0 +1,95 @@ +import os +import sys +import json +import struct +import unittest +import warnings + +warnings.filterwarnings("ignore") + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +import Win_Client +import pgpy +from pgpy.constants import PubKeyAlgorithm, KeyFlags, HashAlgorithm, SymmetricKeyAlgorithm, CompressionAlgorithm + + +class TestWinClientComponent(unittest.TestCase): + def setUp(self): + self.dummy_config = "test_win_client_config.json" + # Generate dummy PGP key for testing + key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 2048) + uid = pgpy.PGPUID.new("TestHub") + key.add_uid( + uid, + usage={KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage}, + hashes=[HashAlgorithm.SHA256], + ciphers=[SymmetricKeyAlgorithm.AES256], + compression=[CompressionAlgorithm.Uncompressed] + ) + self.server_priv = key + self.server_pub = key.pubkey + self.fingerprint = str(key.pubkey.fingerprint) + + with open(self.dummy_config, "w", encoding="utf-8") as f: + json.dump({ + "server_host": "127.0.0.1", + "server_port": 9443, + "server_fingerprint": self.fingerprint, + "server_public_key": str(self.server_pub), + "auth_token": "secret-test-token" + }, f) + + def tearDown(self): + if os.path.exists(self.dummy_config): + try: + os.remove(self.dummy_config) + except Exception: + pass + + def test_client_config_anonymity(self): + config = Win_Client.load_config(self.dummy_config) + self.assertNotIn("server_name", config) + self.assertNotIn("name", config) + self.assertNotIn("site_name", config) + self.assertEqual(config["server_fingerprint"], self.fingerprint) + + def test_get_machine_identifier(self): + machine_id = Win_Client.get_machine_identifier() + self.assertIsInstance(machine_id, str) + self.assertGreater(len(machine_id), 0) + self.assertNotEqual(machine_id, "localhost") + + def test_encryption_and_envelope_creation(self): + config = Win_Client.load_config(self.dummy_config) + logs = [{ + "server": Win_Client.get_machine_identifier(), + "signature": "TestWinSignature", + "severity": "WARNING", + "message": "Disk space threshold warning" + }] + + pub_key, _ = pgpy.PGPKey.from_blob(config["server_public_key"]) + payload = { + "server": Win_Client.get_machine_identifier(), + "logs": logs + } + msg = pgpy.PGPMessage.new(json.dumps(payload)) + enc = pub_key.encrypt(msg) + self.assertTrue(str(enc).startswith("-----BEGIN PGP MESSAGE-----")) + + # Decrypt with private key to verify end-to-end payload integrity + dec = self.server_priv.decrypt(enc) + restored = json.loads(dec.message) + self.assertEqual(restored["logs"][0]["signature"], "TestWinSignature") + + def test_framing_protocol(self): + envelope_data = json.dumps({"test": "data"}).encode("utf-8") + frame = struct.pack(">I", len(envelope_data)) + envelope_data + self.assertEqual(len(frame), 4 + len(envelope_data)) + length = struct.unpack(">I", frame[:4])[0] + self.assertEqual(length, len(envelope_data)) + + +if __name__ == "__main__": + unittest.main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2f53992 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +pgpy>=0.6.0 +standard-imghdr>=3.13.0; python_version >= "3.13" +cryptography>=42.0.0 +fastapi>=0.110.0 +uvicorn>=0.28.0 +pydantic>=2.6.0 +pywin32>=306; sys_platform == "win32" diff --git a/test_pipeline.py b/test_pipeline.py new file mode 100644 index 0000000..8014988 --- /dev/null +++ b/test_pipeline.py @@ -0,0 +1,131 @@ +import os +import sys +import json +import time +import socket +import struct +import sqlite3 +import urllib.request +import warnings +from datetime import datetime, timezone, timedelta + +warnings.filterwarnings("ignore") +import pgpy + +# Test server endpoints +TCP_HOST = "127.0.0.1" +TCP_PORT = 9443 +HERMES_HOST = "127.0.0.1" +HERMES_PORT = 8443 + +def run_tests(): + print("=== [1] Verifying server_config.json & client_config.json ===") + assert os.path.exists("server_config.json"), "server_config.json must exist" + assert os.path.exists("client_config.json"), "client_config.json must exist" + + with open("client_config.json", "r", encoding="utf-8") as f: + client_conf = json.load(f) + + with open("server_config.json", "r", encoding="utf-8") as f: + server_conf = json.load(f) + + assert "server_name" not in client_conf, "client_config.json must NOT contain server_name" + assert "name" not in client_conf, "client_config.json must NOT contain name" + assert "site_name" not in client_conf, "client_config.json must NOT contain site_name" + assert client_conf["server_fingerprint"] == server_conf["server_fingerprint"], "Fingerprints must match" + print(f"[OK] Verified client_config.json contains no machine/server/site name.") + print(f"[OK] Fingerprint verified: {client_conf['server_fingerprint']}") + + # Load public key + pub_key, _ = pgpy.PGPKey.from_blob(client_conf["server_public_key"]) + + def send_socket_batch(logs, auth_token=client_conf["auth_token"]): + payload = { + "server": "test-edge-node.corp.internal", + "timestamp": datetime.now(timezone.utc).isoformat(), + "logs": logs + } + pgp_msg = pgpy.PGPMessage.new(json.dumps(payload)) + enc = pub_key.encrypt(pgp_msg) + + envelope = { + "auth_token": auth_token, + "timestamp": datetime.now(timezone.utc).isoformat(), + "encrypted_payload": str(enc) + } + envelope_bytes = json.dumps(envelope).encode("utf-8") + + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(5.0) + s.connect((TCP_HOST, TCP_PORT)) + frame = struct.pack(">I", len(envelope_bytes)) + envelope_bytes + s.sendall(frame) + + resp_len_bytes = s.recv(4) + resp_len = struct.unpack(">I", resp_len_bytes)[0] + resp_bytes = s.recv(resp_len) + return json.loads(resp_bytes.decode("utf-8")) + + print("\n=== [2] Testing Socket Authentication Failure ===") + bad_resp = send_socket_batch([], auth_token="invalid-token-12345") + assert bad_resp.get("status") == "error", f"Expected error, got: {bad_resp}" + print(f"[OK] Bad auth rejected correctly: {bad_resp['message']}") + + test_signature = "TestServiceCrash" + candidate_log = [{ + "server": "test-edge-node", + "os_type": "linux", + "signature": test_signature, + "severity": "ERROR", + "message": "Out of memory killer triggered" + }] + + print("\n=== [3] Testing Temporal Persistence & 4-Run Rule ===") + for run_num in range(1, 5): + resp = send_socket_batch(candidate_log) + assert resp.get("status") == "success", f"Run {run_num} failed: {resp}" + print(f"[Run {run_num}/4] Ingested successfully. Promoted to verified: {resp.get('promoted_verified')}") + + # Inspect SQLite database directly + conn = sqlite3.connect(server_conf.get("db_path", "logar_state.db")) + cursor = conn.cursor() + cursor.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", (test_signature,)) + row = cursor.fetchone() + conn.close() + + assert row is not None, "Issue not found in SQLite" + run_count, status = row + print(f"[DB Verification] Issue '{test_signature}' -> run_count: {run_count}, status: {status}") + assert run_count >= 4, f"Expected run_count >= 4, got {run_count}" + assert status == "VERIFIED", f"Expected status 'VERIFIED', got {status}" + print("[OK] 4-Run Rule verified: Transient issue promoted to VERIFIED anomaly!") + + print("\n=== [4] Testing Hermes Reporting Endpoint (/api/hermes/report) ===") + req = urllib.request.Request(f"http://{HERMES_HOST}:{HERMES_PORT}/api/hermes/report") + with urllib.request.urlopen(req, timeout=5) as response: + assert response.status == 200, f"Expected 200, got {response.status}" + hermes_data = json.loads(response.read().decode("utf-8")) + + print(f"[Hermes API] Returned {len(hermes_data)} verified anomalies:") + found_issue = False + for issue in hermes_data: + print(f" - Fingerprint: {issue['fingerprint']} | Consecutive Runs: {issue['consecutive_runs']} | Status: {issue['status']}") + if issue["signature"] == test_signature: + found_issue = True + assert issue["verified"] is True + assert issue["consecutive_runs"] >= 4 + + assert found_issue, f"Test issue {test_signature} should be in Hermes report" + print("[OK] Hermes reporting validated!") + + print("\n=== [5] Testing Windows Client Script Integration ===") + from Win_Client import get_recent_windows_logs + win_logs = get_recent_windows_logs(hours=6) + print(f"[Win_Client] Successfully queried Windows logs: {len(win_logs)} candidate entries.") + + print("\n==========================================") + print(" ALL VERIFICATION TESTS PASSED SUCCESSFULLY! ") + print("==========================================") + +if __name__ == "__main__": + run_tests()