LOGAR/tests/test_linux_client.py

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"}),
            json.dumps({"PRIORITY": "7", "SYSLOG_IDENTIFIER": "debugd", "MESSAGE": "Verbose debugging log"}),
        ]
        logs = []
        machine_id = Linux_Client.get_machine_identifier()
        for line_str in sample_journal_lines:
            entry = json.loads(line_str)
            priority = int(entry.get("PRIORITY", "6"))
            # Filter: Retain INFO to ERROR (<= 6), drop DEBUG (> 6)
            if priority > 6:
                continue

            if priority <= 3:
                sev = "ERROR"
            elif priority in (4, 5):
                sev = "WARNING"
            else:
                sev = "INFO"

            logs.append({
                "server": machine_id,
                "os_type": "linux",
                "signature": entry.get("SYSLOG_IDENTIFIER", "unknown"),
                "severity": sev,
                "message": entry.get("MESSAGE", "")
            })

        # Priority 7 (DEBUG) must be stripped, while 3 (ERROR), 4 (WARNING), 6 (INFO) are retained
        self.assertEqual(len(logs), 3)
        self.assertEqual(logs[0]["severity"], "ERROR")
        self.assertEqual(logs[1]["severity"], "WARNING")
        self.assertEqual(logs[2]["severity"], "INFO")

    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")

    def test_state_lifecycle(self):
        state_path = "test_linux_state.json"
        try:
            # 1. Load non-existent returns empty dict
            state = Linux_Client.load_state(state_path)
            self.assertEqual(state, {})

            # 2. Stage new cursor and timestamp
            state["new_last_cursor"] = "s=abc;i=123"
            state["new_last_timestamp_us"] = 1700000000000000
            state["new_sent_cursors"] = ["s=abc;i=123"]

            # 3. Commit state moves staged keys to permanent and writes atomically
            Linux_Client.commit_state(state, state_path)
            self.assertNotIn("new_last_cursor", state)
            self.assertEqual(state.get("last_cursor"), "s=abc;i=123")
            self.assertEqual(state.get("last_timestamp_us"), 1700000000000000)
            self.assertEqual(state.get("sent_cursors"), ["s=abc;i=123"])

            # 4. Reload from disk
            reloaded = Linux_Client.load_state(state_path)
            self.assertEqual(reloaded.get("last_cursor"), "s=abc;i=123")
            self.assertEqual(reloaded.get("last_timestamp_us"), 1700000000000000)
        finally:
            if os.path.exists(state_path):
                os.remove(state_path)

    def test_duplicate_suppression_and_lookback_logic(self):
        from datetime import datetime, timezone, timedelta
        now_us = datetime.now(timezone.utc).timestamp() * 1_000_000
        cutoff_epoch_us = (datetime.now(timezone.utc) - timedelta(hours=24)).timestamp() * 1_000_000

        mock_entries = [
            # 1. 26 hours old -> skip (> 24h)
            {"__CURSOR": "c1", "__REALTIME_TIMESTAMP": str(int(now_us - 26 * 3600 * 1_000_000)), "PRIORITY": "3", "MESSAGE": "Old error"},
            # 2. 2 hours old, already sent -> skip
            {"__CURSOR": "c2", "__REALTIME_TIMESTAMP": str(int(now_us - 2 * 3600 * 1_000_000)), "PRIORITY": "4", "MESSAGE": "Already sent warning"},
            # 3. 1 hour old, new entry -> retain
            {"__CURSOR": "c3", "__REALTIME_TIMESTAMP": str(int(now_us - 1 * 3600 * 1_000_000)), "PRIORITY": "6", "MESSAGE": "New info"},
            # 4. 30 mins old, debug -> skip priority
            {"__CURSOR": "c4", "__REALTIME_TIMESTAMP": str(int(now_us - 1800 * 1_000_000)), "PRIORITY": "7", "MESSAGE": "Debug entry"}
        ]

        state = {
            "last_cursor": "c2",
            "last_timestamp_us": int(now_us - 2 * 3600 * 1_000_000),
            "sent_cursors": ["c2"]
        }

        # Simulate the filtering loop from get_recent_linux_logs
        logs = []
        last_cursor = state.get("last_cursor")
        last_timestamp_us = float(state.get("last_timestamp_us", 0))
        sent_cursors = set(state.get("sent_cursors", []))
        newest_cursor = None
        newest_timestamp_us = last_timestamp_us
        collected_cursors = []

        for entry in mock_entries:
            entry_cursor = entry.get("__CURSOR")
            entry_ts_us = float(entry.get("__REALTIME_TIMESTAMP"))

            if entry_ts_us < cutoff_epoch_us:
                continue
            if entry_cursor and (entry_cursor in sent_cursors or entry_cursor == last_cursor):
                continue
            if last_timestamp_us > 0 and entry_ts_us < last_timestamp_us:
                continue

            if entry_cursor:
                newest_cursor = entry_cursor
                collected_cursors.append(entry_cursor)
            if entry_ts_us > newest_timestamp_us:
                newest_timestamp_us = entry_ts_us

            priority = int(entry.get("PRIORITY", "6"))
            if priority > 6:
                continue

            logs.append(entry)

        self.assertEqual(len(logs), 1)
        self.assertEqual(logs[0]["__CURSOR"], "c3")
        self.assertEqual(newest_cursor, "c4")


if __name__ == "__main__":
    unittest.main()