Initial commit: LOGAR edge-thin log analysis system with OpenPGP encryption, authenticated TCP sockets, 4-run persistence rule, Hermes reporting, and modular shippables

2026-09-03 21:05:28 +02:00
commit c5d364b289
26 changed files with 2667 additions and 0 deletions
@@ -0,0 +1,107 @@
+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"}),
+        ]
+        logs = []
+        machine_id = Linux_Client.get_machine_identifier()
+        for line_str in sample_journal_lines:
+            entry = json.loads(line_str)
+            priority = str(entry.get("PRIORITY", "4"))
+            if int(priority) > 4:
+                continue
+            sev = "WARNING" if priority == "4" else "ERROR"
+            logs.append({
+                "server": machine_id,
+                "os_type": "linux",
+                "signature": entry.get("SYSLOG_IDENTIFIER", "unknown"),
+                "severity": sev,
+                "message": entry.get("MESSAGE", "")
+            })
+
+        # Priority 6 must be stripped (INFO noise)
+        self.assertEqual(len(logs), 2)
+        self.assertEqual(logs[0]["severity"], "ERROR")
+        self.assertEqual(logs[1]["severity"], "WARNING")
+
+    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")
+
+
+if __name__ == "__main__":
+    unittest.main()