195 lines
8.3 KiB
Python
195 lines
8.3 KiB
Python
import base64
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import patch
|
|
|
|
from app.ndr import NDRAnalyzer, ThreatIntelManager
|
|
from app.store import AlertStore
|
|
|
|
|
|
class DummyRouterOS:
|
|
configured = True
|
|
|
|
def list_dhcp_leases(self):
|
|
return [{"address": "192.168.88.20", "mac": "AA:BB:CC:DD:EE:20", "hostname": "office-pc"}]
|
|
|
|
def list_arp(self):
|
|
return [{"address": "192.168.88.30", "mac": "AA:BB:CC:DD:EE:30"}]
|
|
|
|
def block_ip(self, address, timeout_value, comment):
|
|
raise AssertionError("auto-block is disabled in this test")
|
|
|
|
|
|
def test_threat_intel_materializes_suricata8_datasets_and_matches():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = AlertStore(os.path.join(td, "ids.db"))
|
|
ja3 = "0123456789abcdef0123456789abcdef"
|
|
hassh = "fedcba9876543210fedcba9876543210"
|
|
ja4 = "t13d1516h2_8daaf6152771_02713d6af862"
|
|
store.add_ioc("203.0.113.7", "ip", source="test")
|
|
store.add_ioc("bad.example", "domain", source="test")
|
|
store.add_ioc(ja3, "ja3", source="test")
|
|
store.add_ioc(ja4, "ja4", source="test")
|
|
sha256 = "a" * 64
|
|
store.add_ioc(hassh, "hassh", source="test")
|
|
store.add_ioc(sha256, "sha256", source="test")
|
|
|
|
manager = ThreatIntelManager(store, os.path.join(td, "suricata"))
|
|
counts = manager.sync_suricata_datasets()
|
|
assert counts["ip"] == 1
|
|
assert counts["domain"] == 1
|
|
assert counts["ja3"] == 1
|
|
assert counts["ja4"] == 1
|
|
assert counts["hassh"] == 1
|
|
assert counts["sha256"] == 1
|
|
|
|
state = os.path.join(td, "suricata")
|
|
assert open(os.path.join(state, "ti-ips.lst"), encoding="ascii").read().strip() == "203.0.113.7"
|
|
assert open(os.path.join(state, "ti-sha256.lst"), encoding="ascii").read().strip() == sha256
|
|
for kind, value in (("domains", "bad.example"), ("ja3", ja3), ("ja4", ja4), ("hassh", hassh)):
|
|
encoded = open(os.path.join(state, f"ti-{kind}.lst"), encoding="ascii").read().strip()
|
|
assert base64.b64decode(encoded).decode() == value
|
|
|
|
rules = open(os.path.join(state, "threat-intel.rules"), encoding="utf-8").read()
|
|
assert "sid:1000205" in rules and "ja3.hash" in rules
|
|
assert "sid:1000206" in rules and "alert tls" in rules
|
|
assert "sid:1000207" in rules and "alert quic" in rules
|
|
assert "sid:1000208" in rules and "ssh.hassh" in rules
|
|
assert "sid:1000209" in rules and "ssh.hassh.server" in rules
|
|
assert "sid:1000210" in rules and "filesha256:ti-sha256.lst" in rules
|
|
assert "sid:1000215" in rules and "alert smb" in rules
|
|
assert "type string,load ti-ja3.lst" in rules
|
|
|
|
hits = manager.match({"dest_ip": "203.0.113.7", "dns_query": "sub.bad.example", "tls_ja3": ja3})
|
|
assert {row["indicator_type"] for row in hits} >= {"ip", "domain", "ja3"}
|
|
store.close()
|
|
|
|
|
|
def test_ndr_correlates_multistage_risk_and_status():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = AlertStore(os.path.join(td, "ids.db"))
|
|
first = store.correlate_signal({
|
|
"subject_ip": "192.168.88.10", "timestamp": "2026-08-15T08:00:00+00:00",
|
|
"kind": "behavior", "stage": "recon", "risk": 45, "summary": "scan",
|
|
"dest_ip": "192.168.88.11",
|
|
})
|
|
second = store.correlate_signal({
|
|
"subject_ip": "192.168.88.10", "timestamp": "2026-08-15T08:01:00+00:00",
|
|
"kind": "alert", "stage": "lateral-movement", "risk": 60, "summary": "SMB access",
|
|
"dest_ip": "192.168.88.11",
|
|
})
|
|
assert first == second
|
|
incident = store.ndr_incident(first)
|
|
assert incident["risk_score"] == 70
|
|
assert set(incident["stages"]) == {"recon", "lateral-movement"}
|
|
assert store.set_ndr_incident_status(first, "closed") is True
|
|
assert store.ndr_incident(first)["status"] == "closed"
|
|
assert store.ndr_summary()["open_incidents"] == 0
|
|
store.close()
|
|
|
|
|
|
def test_routeros_inventory_enriches_assets():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = AlertStore(os.path.join(td, "ids.db"))
|
|
ti = ThreatIntelManager(store, os.path.join(td, "suricata"))
|
|
analyzer = NDRAnalyzer(
|
|
store, ti, DummyRouterOS(), "192.168.88.0/24", "", "1h",
|
|
enabled=True, auto_block=False,
|
|
)
|
|
result = analyzer.sync_routeros_inventory()
|
|
assert result == {"arp": 1, "dhcp": 1, "assets": 2}
|
|
assets = {row["ip"]: row for row in store.assets(20)}
|
|
assert assets["192.168.88.20"]["hostname"] == "office-pc"
|
|
assert assets["192.168.88.20"]["mac"] == "AA:BB:CC:DD:EE:20"
|
|
assert assets["192.168.88.30"]["mac"] == "AA:BB:CC:DD:EE:30"
|
|
store.close()
|
|
|
|
|
|
def test_repeated_ip_mac_changes_escalate_to_network_spoofing_and_anomalies_are_cooled_down():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = AlertStore(os.path.join(td, "ids.db"))
|
|
ti = ThreatIntelManager(store, os.path.join(td, "suricata"))
|
|
analyzer = NDRAnalyzer(
|
|
store, ti, DummyRouterOS(), "192.168.88.0/24", "", "1h",
|
|
enabled=True, auto_block=False,
|
|
)
|
|
ip = "192.168.88.44"
|
|
for idx, mac in enumerate((
|
|
"AA:BB:CC:DD:EE:01",
|
|
"AA:BB:CC:DD:EE:02",
|
|
"AA:BB:CC:DD:EE:03",
|
|
"AA:BB:CC:DD:EE:04",
|
|
)):
|
|
analyzer._process({
|
|
"timestamp": f"2026-08-15T08:00:{idx:02d}+00:00",
|
|
"type": "arp", "direction": "outbound", "src_ip": ip,
|
|
"arp_src_ip": ip, "arp_src_mac": mac,
|
|
}, None)
|
|
|
|
incidents = store.recent_ndr_incidents(20)
|
|
incident = next(row for row in incidents if row["subject_ip"] == ip)
|
|
assert "network-spoofing" in incident["stages"]
|
|
assert int(incident["risk_score"]) >= 78
|
|
|
|
anomaly = {
|
|
"timestamp": "2026-08-15T08:10:00+00:00", "type": "anomaly",
|
|
"direction": "outbound", "src_ip": "192.168.88.55",
|
|
"anomaly_event": "APPLAYER_WRONG_DIRECTION_FIRST_DATA",
|
|
}
|
|
analyzer._process(anomaly, None)
|
|
anomaly["timestamp"] = "2026-08-15T08:10:10+00:00"
|
|
analyzer._process(anomaly, None)
|
|
anomaly_incident = next(row for row in store.recent_ndr_incidents(20) if row["subject_ip"] == "192.168.88.55")
|
|
events = store.ndr_incident_events(int(anomaly_incident["id"]), 20)
|
|
assert sum(1 for event in events if event["stage"] == "protocol-anomaly") == 1
|
|
store.close()
|
|
|
|
|
|
def test_baseline_touch_is_cached_for_repeated_values():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = AlertStore(os.path.join(td, "ids.db"))
|
|
ti = ThreatIntelManager(store, os.path.join(td, "suricata"))
|
|
analyzer = NDRAnalyzer(
|
|
store, ti, DummyRouterOS(), "192.168.88.0/24", "", "1h",
|
|
enabled=True, auto_block=False,
|
|
)
|
|
calls = 0
|
|
original = store.baseline_touch
|
|
|
|
def counted(*args, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
return original(*args, **kwargs)
|
|
|
|
store.baseline_touch = counted
|
|
record = {
|
|
"timestamp": "2026-08-15T08:20:00+00:00",
|
|
"type": "flow", "direction": "outbound", "src_ip": "192.168.88.50",
|
|
"dest_ip": "203.0.113.20", "dest_port": 443, "app_proto": "tls",
|
|
}
|
|
for _ in range(10):
|
|
analyzer._process(dict(record), None)
|
|
assert calls == 2 # app + outbound port, only on first sight in this process
|
|
assert analyzer.status()["state_entries"]["baseline_lru"] == 2
|
|
store.close()
|
|
|
|
|
|
def test_beacon_state_is_bounded():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = AlertStore(os.path.join(td, "ids.db"))
|
|
ti = ThreatIntelManager(store, os.path.join(td, "suricata"))
|
|
analyzer = NDRAnalyzer(
|
|
store, ti, DummyRouterOS(), "192.168.88.0/24", "", "1h",
|
|
enabled=True, auto_block=False,
|
|
)
|
|
with patch("app.ndr.NDR_BEACON_MAX_KEYS", 32):
|
|
for idx in range(64):
|
|
analyzer._behavior({
|
|
"timestamp": "2026-08-15T08:30:00+00:00",
|
|
"type": "flow", "direction": "outbound",
|
|
"src_ip": "192.168.88.60", "dest_ip": f"203.0.113.{idx}",
|
|
"dest_port": 443,
|
|
}, None, "192.168.88.60")
|
|
assert len(analyzer._beacon) == 32
|
|
store.close()
|