LOGAR/tests/test_pipeline.py

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 src/ directory is in sys.path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "src")))

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=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()