250 lines
10 KiB
Python
250 lines
10 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__), "..")))
|
|
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")
|
|
|
|
def test_mtls_client_certificate_handling(self):
|
|
import shutil
|
|
test_dir = "test_linux_mtls_certs"
|
|
os.makedirs(test_dir, exist_ok=True)
|
|
try:
|
|
from src import server_enrollment as se
|
|
ca_cert, ca_key, ca_pem, _ = se.generate_ca_if_needed(test_dir)
|
|
client_cert_pem, client_key_pem = se.issue_client_cert("linux-client-test", ca_cert, ca_key)
|
|
|
|
with open(os.path.join(test_dir, "ca.crt"), "w") as f:
|
|
f.write(ca_pem)
|
|
with open(os.path.join(test_dir, "client.crt"), "w") as f:
|
|
f.write(client_cert_pem)
|
|
with open(os.path.join(test_dir, "client.key"), "w") as f:
|
|
f.write(client_key_pem)
|
|
|
|
# Test missing certs exception
|
|
empty_dir = "test_empty_linux_certs"
|
|
os.makedirs(empty_dir, exist_ok=True)
|
|
with self.assertRaises(FileNotFoundError):
|
|
Linux_Client.get_tls_socket("127.0.0.1", 9443, empty_dir)
|
|
shutil.rmtree(empty_dir, ignore_errors=True)
|
|
finally:
|
|
shutil.rmtree(test_dir, ignore_errors=True)
|
|
|
|
def test_client_certificate_validity_and_proactive_check(self):
|
|
import shutil
|
|
test_dir = "test_linux_client_validity"
|
|
os.makedirs(test_dir, exist_ok=True)
|
|
try:
|
|
from src import server_enrollment as se
|
|
ca_cert, ca_key, _, _ = se.generate_ca_if_needed(test_dir)
|
|
client_cert_pem, client_key_pem = se.issue_client_cert("linux-validity-test", ca_cert, ca_key, days_valid=365)
|
|
cert_path = os.path.join(test_dir, "client.crt")
|
|
with open(cert_path, "w", encoding="utf-8") as f:
|
|
f.write(client_cert_pem)
|
|
|
|
# Newly issued cert (365 days) is not expiring soon at 30 days
|
|
self.assertFalse(Linux_Client.is_cert_expiring_soon(cert_path, threshold_days=30))
|
|
# Large threshold (500 days) reports expiring soon
|
|
self.assertTrue(Linux_Client.is_cert_expiring_soon(cert_path, threshold_days=500))
|
|
# Non-existent file reports expiring / missing
|
|
self.assertTrue(Linux_Client.is_cert_expiring_soon(os.path.join(test_dir, "missing.crt")))
|
|
finally:
|
|
shutil.rmtree(test_dir, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|