Files
LOGAR/tests/test_pipeline.py
T

183 lines
8.1 KiB
Python

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
# Ensure repository root and src/ directory are in sys.path
ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
SRC_DIR = os.path.join(ROOT_DIR, "src")
sys.path.insert(0, ROOT_DIR)
sys.path.insert(0, SRC_DIR)
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 ===")
server_cfg_path = "server_config.json" if os.path.exists("server_config.json") else os.path.join(ROOT_DIR, "server_config.json")
client_cfg_path = "client_config.json" if os.path.exists("client_config.json") else os.path.join(ROOT_DIR, "client_config.json")
assert os.path.exists(server_cfg_path), f"{server_cfg_path} must exist"
assert os.path.exists(client_cfg_path), f"{client_cfg_path} must exist"
with open(client_cfg_path, "r", encoding="utf-8") as f:
client_conf = json.load(f)
with open(server_cfg_path, "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 = "TestServiceDegraded"
candidate_log = [{
"server": "test-edge-node",
"os_type": "linux",
"signature": test_signature,
"severity": "WARNING",
"message": "Resource usage high warning"
}]
print("\n=== [3] Testing Temporal Persistence & 4-Run Rule for Warnings ===")
for run_num in range(1, 5):
resp = send_socket_batch(candidate_log)
assert resp.get("status") == "success", f"Run {run_num} failed: {resp}"
promoted = resp.get("promoted_verified", 0)
print(f"[Run {run_num}/4] Ingested successfully. Promoted to verified: {promoted}")
if run_num < 4:
assert promoted == 0, f"Expected 0 promoted on run {run_num} for warning, got {promoted}"
else:
assert promoted == 1, f"Expected 1 promoted on run 4 for warning, got {promoted}"
# 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: Warning promoted to VERIFIED anomaly on 4th run!")
print("\n=== [3b] Testing Immediate Pass for Errors ===")
error_signature = "TestServiceCrashImmediate"
error_log = [{
"server": "test-edge-node",
"os_type": "linux",
"signature": error_signature,
"severity": "ERROR",
"message": "Fatal process crash occurred"
}]
err_resp = send_socket_batch(error_log)
assert err_resp.get("status") == "success", f"Error run failed: {err_resp}"
print(f"[Run 1/1] Error ingested successfully. Promoted to verified: {err_resp.get('promoted_verified')}")
assert err_resp.get("promoted_verified") == 1, f"Expected error to be promoted to verified immediately on run 1, got {err_resp.get('promoted_verified')}"
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 = ?", (error_signature,))
err_row = cursor.fetchone()
conn.close()
assert err_row is not None, "Error issue not found in SQLite"
err_run_count, err_status = err_row
print(f"[DB Verification] Issue '{error_signature}' -> run_count: {err_run_count}, status: {err_status}")
assert err_run_count == 1, f"Expected run_count == 1, got {err_run_count}"
assert err_status == "VERIFIED", f"Expected status 'VERIFIED', got {err_status}"
print("[OK] Immediate pass verified: Error promoted to VERIFIED anomaly immediately!")
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_warning = False
found_error = 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_warning = True
assert issue["verified"] is True
assert issue["consecutive_runs"] >= 4
if issue["signature"] == error_signature:
found_error = True
assert issue["verified"] is True
assert issue["consecutive_runs"] == 1
assert found_warning, f"Warning issue {test_signature} should be in Hermes report"
assert found_error, f"Error issue {error_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=24)
print(f"[Win_Client] Successfully queried Windows logs: {len(win_logs)} candidate entries.")
print("\n=== [6] Testing Linux Client Script Integration ===")
from Linux_Client import get_recent_linux_logs
linux_logs = get_recent_linux_logs(hours=24)
print(f"[Linux_Client] Successfully queried Linux logs: {len(linux_logs)} candidate entries.")
print("\n==========================================")
print(" ALL VERIFICATION TESTS PASSED SUCCESSFULLY! ")
print("==========================================")
if __name__ == "__main__":
run_tests()