poc2_worked

tests/test_autonomous_ids.py

@@ -0,0 +1,102 @@
+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