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