Move source files Server.py, Win_Client.py, and Linux_Client.py into src/ directory
This commit is contained in:
@@ -0,0 +1,343 @@
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import socket
|
||||
import struct
|
||||
import argparse
|
||||
import subprocess
|
||||
import warnings
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Optional, Dict, Any, List
|
||||
|
||||
# Suppress cryptography / pgpy deprecation notices
|
||||
warnings.filterwarnings("ignore")
|
||||
|
||||
import pgpy
|
||||
|
||||
CONFIG_FILE_NAME = "client_config.json"
|
||||
STATE_FILE_NAME = "client_state.json"
|
||||
|
||||
|
||||
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:
|
||||
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 = 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.
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
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.
|
||||
"""
|
||||
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 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 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 journalctl 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")
|
||||
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 = get_machine_identifier()
|
||||
print(f"[*] Edge Forwarder Node: {machine_id}")
|
||||
if state and ("last_cursor" in state or "last_timestamp_us" in state):
|
||||
print(f"[*] State tracking active: resuming after previous cursor/timestamp (state file: {state_path})")
|
||||
elif not args.no_state:
|
||||
print(f"[*] State tracking initialized (state file: {state_path})")
|
||||
|
||||
print(f"[*] Scanning Linux journalctl for unsent entries (last {args.hours} hours)...")
|
||||
candidate_logs = get_recent_linux_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}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+465
@@ -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: retain INFO to ERROR / CRITICAL; strip verbose debug noise
|
||||
if severity in ["DEBUG", "TRACE"]:
|
||||
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()
|
||||
@@ -0,0 +1,310 @@
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import socket
|
||||
import struct
|
||||
import argparse
|
||||
import warnings
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Optional, Dict, Any, List
|
||||
|
||||
# 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 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_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):
|
||||
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 = 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):
|
||||
"""
|
||||
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 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")
|
||||
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 = 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})")
|
||||
|
||||
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}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()␍
|
||||
Reference in New Issue
Block a user