108 lines
4.1 KiB
Python
108 lines
4.1 KiB
Python
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()
|