103 lines
4.4 KiB
Python
103 lines
4.4 KiB
Python
import os
|
|
import tarfile
|
|
import tempfile
|
|
from datetime import datetime, timezone
|
|
|
|
from app.adaptive import score_rule
|
|
from app.backup import BackupManager
|
|
from app.mitre import classify, merge
|
|
from app.store import AlertStore
|
|
|
|
|
|
def test_mitre_network_evidence_mapping_is_conservative_and_specific():
|
|
rdp = classify("lateral-movement", "RDP access", {"dest_port": 3389})
|
|
assert rdp[0]["tactic_id"] == "TA0008"
|
|
assert rdp[0]["technique_id"] == "T1021.001"
|
|
dns = classify("command-and-control", "DNS beacon", {"dns_query": "x.example"})
|
|
assert dns[0]["technique_id"] == "T1071.004"
|
|
assert classify("unknown-stage", "opaque event", {}) == []
|
|
assert len(merge(rdp, rdp + dns)) == 2
|
|
|
|
|
|
def test_adaptive_rule_scoring_never_disables_and_limits_only_high_noise():
|
|
noisy = score_rule({
|
|
"signature_id": 9001, "hits": 1800, "rows": 200, "unique_src": 2,
|
|
"unique_dst": 2, "incidents": 0, "blocked": 0, "severity": 3,
|
|
})
|
|
assert noisy["recommendation"] == "limit"
|
|
assert noisy["proposed_threshold"]["type"] == "limit"
|
|
assert noisy["proposed_threshold"]["track"] == "by_src"
|
|
valuable = score_rule({
|
|
"signature_id": 9002, "hits": 500, "rows": 100, "unique_src": 30,
|
|
"unique_dst": 30, "incidents": 40, "blocked": 3, "severity": 1,
|
|
})
|
|
assert valuable["recommendation"] == "keep"
|
|
assert valuable["proposed_threshold"] is None
|
|
|
|
|
|
def test_backup_contains_persistent_state_but_excludes_runtime_streams():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
db = os.path.join(td, "ids.db")
|
|
store = AlertStore(db)
|
|
store.audit("admin", "test.action", target="unit")
|
|
os.makedirs(os.path.join(td, "suricata"), exist_ok=True)
|
|
with open(os.path.join(td, "suricata", "custom.rules"), "w", encoding="utf-8") as f:
|
|
f.write('alert ip any any -> any any (msg:"test"; sid:9900001;)\n')
|
|
os.makedirs(os.path.join(td, "lib", "suricata", "update", "sources"), exist_ok=True)
|
|
with open(os.path.join(td, "lib", "suricata", "update", "sources", "oisf.yaml"), "w", encoding="utf-8") as f:
|
|
f.write("enabled: true\n")
|
|
os.makedirs(os.path.join(td, "redis"), exist_ok=True)
|
|
with open(os.path.join(td, "redis", "appendonly.aof"), "w", encoding="utf-8") as f:
|
|
f.write("runtime")
|
|
manager = BackupManager(db, td, keep=3)
|
|
item = manager.create("unit")
|
|
assert item["id"].startswith("mikrosuricata-")
|
|
with tarfile.open(os.path.join(td, "backups", item["id"]), "r:gz") as tar:
|
|
names = set(tar.getnames())
|
|
assert "ids.db" in names
|
|
assert "suricata/custom.rules" in names
|
|
assert any(name.startswith("lib/suricata/update/sources") for name in names)
|
|
assert not any(name.startswith("redis/") for name in names)
|
|
store.close()
|
|
|
|
|
|
def test_store_persists_mitre_audit_and_rule_intelligence():
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = AlertStore(os.path.join(td, "ids.db"))
|
|
incident_id = store.correlate_signal({
|
|
"subject_ip": "192.168.1.10",
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"kind": "behavior", "stage": "lateral-movement", "risk": 60,
|
|
"summary": "RDP access", "dest_ip": "192.168.1.11",
|
|
"mitre": classify("lateral-movement", "RDP access", {"dest_port": 3389}),
|
|
})
|
|
incident = store.ndr_incident(incident_id)
|
|
assert incident["mitre"][0]["technique_id"] == "T1021.001"
|
|
store.audit("admin", "rules.threshold", target="1234", details={"count": 5})
|
|
event = store.audit_events(1)[0]
|
|
assert event["username"] == "admin"
|
|
assert event["details"]["count"] == 5
|
|
store.close()
|
|
|
|
|
|
def test_evewatcher_constructor_call_has_no_unknown_keywords():
|
|
import ast
|
|
import inspect
|
|
from pathlib import Path
|
|
|
|
from app.eve import EVEWatcher
|
|
|
|
root = Path(__file__).resolve().parents[1]
|
|
tree = ast.parse((root / "app" / "main.py").read_text())
|
|
calls = [
|
|
node for node in ast.walk(tree)
|
|
if isinstance(node, ast.Call)
|
|
and isinstance(node.func, ast.Name)
|
|
and node.func.id == "EVEWatcher"
|
|
]
|
|
assert len(calls) == 1
|
|
allowed = set(inspect.signature(EVEWatcher.__init__).parameters) - {"self"}
|
|
passed = {kw.arg for kw in calls[0].keywords if kw.arg is not None}
|
|
assert passed <= allowed
|
|
assert "backup_manager" not in passed
|