feat(defender): classify audit threat risk
This commit is contained in:
PierreCsn committed
1 parent
2d3d316741
commit
d402d5f541
5 files changed
+1428
-270
No files matched your search
@@ -0,0 +1,415 @@
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
import types
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT))
|
||||
sys.modules.setdefault("setuphelpers", types.ModuleType("setuphelpers"))
|
||||
SPEC = importlib.util.spec_from_file_location("defender_setup", ROOT / "setup.py")
|
||||
defender_setup = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(defender_setup)
|
||||
|
||||
|
||||
NOW = datetime(2026, 10, 27, 12, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def defender_state(**overrides):
|
||||
state = {
|
||||
"running_mode": "Normal",
|
||||
"antivirus_enabled": True,
|
||||
"realtime_enabled": True,
|
||||
"behavior_monitor_enabled": True,
|
||||
"ioav_enabled": True,
|
||||
"signature_age": 0,
|
||||
"signature_version": "synthetic-signature-version",
|
||||
"signature_last_updated": "2026-10-27T11:00:00Z",
|
||||
"engine_version": "synthetic-engine-version",
|
||||
"service_enabled": True,
|
||||
}
|
||||
state.update(overrides)
|
||||
return state
|
||||
|
||||
|
||||
def make_threat(threat_id="100", severity_id=1, name="Synthetic threat", is_active=False,
|
||||
source="ThreatSeverityID", did_threat_execute=False):
|
||||
return {
|
||||
"threat_id": str(threat_id),
|
||||
"threat_name": name,
|
||||
"threat_severity_id": severity_id,
|
||||
"severity_source": source,
|
||||
"is_active": is_active,
|
||||
"did_threat_execute": did_threat_execute,
|
||||
}
|
||||
|
||||
|
||||
def make_detection(
|
||||
detection_id="detection-1",
|
||||
threat_id="100",
|
||||
status_id=1,
|
||||
detected_hours_ago=1,
|
||||
changed_hours_ago=None,
|
||||
error_code=0,
|
||||
action_success=True,
|
||||
resources=None,
|
||||
):
|
||||
detected = NOW - timedelta(hours=detected_hours_ago)
|
||||
changed = NOW - timedelta(hours=changed_hours_ago) if changed_hours_ago is not None else detected
|
||||
return {
|
||||
"detection_id": detection_id,
|
||||
"threat_id": str(threat_id),
|
||||
"status_id": status_id,
|
||||
"detected_at": detected.isoformat().replace("+00:00", "Z"),
|
||||
"changed_at": changed.isoformat().replace("+00:00", "Z"),
|
||||
"remediation_at": None,
|
||||
"error_code": error_code,
|
||||
"action_success": action_success,
|
||||
"execution_status_id": 0,
|
||||
"resources": resources or [],
|
||||
}
|
||||
|
||||
|
||||
def make_data(detections=None, threats=None, defender=None, collection_complete=True, **extra):
|
||||
detection_rows = detections if isinstance(detections, list) else ([] if detections is None else [detections])
|
||||
threat_rows = threats if isinstance(threats, list) else ([] if threats is None else [threats])
|
||||
return {
|
||||
"collection_complete": collection_complete,
|
||||
"history_detection_count": len(detection_rows),
|
||||
"returned_detection_count": len(detection_rows),
|
||||
"defender": defender_state(**(defender or {})),
|
||||
"threats": threat_rows,
|
||||
"detections": detection_rows,
|
||||
**extra,
|
||||
}
|
||||
|
||||
|
||||
class TestDefenderClassification(unittest.TestCase):
|
||||
def build(self, data, rules=None):
|
||||
return defender_setup.build_audit_snapshot(data, now=NOW, rules=rules)
|
||||
|
||||
def test_no_events_with_operational_defender_is_ok(self):
|
||||
snapshot = self.build(make_data())
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "OK")
|
||||
self.assertEqual(snapshot["summary"]["max_event_level_6h"], 0)
|
||||
self.assertEqual(snapshot["summary"]["max_active_level"], 0)
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 0)
|
||||
self.assertEqual(snapshot["detections"], [])
|
||||
|
||||
def test_low_quarantined_threat_has_residual_risk_and_remains_active(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=3)],
|
||||
[make_threat(severity_id=1, is_active=False)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
item = snapshot["active_threats"][0]
|
||||
self.assertEqual(item["event_level"], 3) # 4 - 3 + 2 active
|
||||
self.assertEqual(item["status"], "Quarantined")
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "OK")
|
||||
|
||||
def test_severe_removed_is_historical_event_not_current_error(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=4)],
|
||||
[make_threat(severity_id=5, is_active=False)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertEqual(snapshot["summary"]["max_event_level_6h"], 9)
|
||||
self.assertEqual(snapshot["summary"]["max_active_level"], 0)
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 0)
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "OK")
|
||||
self.assertEqual(snapshot["summary"]["resolved_count_6h"], 1)
|
||||
|
||||
def test_removed_threat_with_execution_observed_remains_unresolved_risk(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=4)],
|
||||
[make_threat(severity_id=5, is_active=False, did_threat_execute=True)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertEqual(snapshot["active_threats"][0]["active_reason"], "execution_observed")
|
||||
self.assertEqual(snapshot["active_threats"][0]["event_level"], 11) # 12 - 3 removed + 2 observed
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
||||
|
||||
def test_current_executing_state_keeps_removed_detection_active(self):
|
||||
event = make_detection(status_id=4)
|
||||
event["execution_status_id"] = 3
|
||||
snapshot = self.build(make_data([event], [make_threat(severity_id=2, is_active=False)]))
|
||||
self.assertEqual(snapshot["active_threats"][0]["active_reason"], "execution_allowed_or_running")
|
||||
|
||||
def test_severe_detected_threat_is_error(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=1)],
|
||||
[make_threat(severity_id=5, is_active=True)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertEqual(snapshot["summary"]["max_active_level"], 14)
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 14)
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
||||
|
||||
def test_failed_remediation_adds_failure_and_active_modifiers(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=103, action_success=False)],
|
||||
[make_threat(severity_id=1, is_active=False)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertEqual(snapshot["active_threats"][0]["event_level"], 9) # 4 + 3 + 2
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "WARNING")
|
||||
|
||||
def test_action_success_false_adds_failure_modifier_even_before_failed_status_rollup(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=1, action_success=False)],
|
||||
[make_threat(severity_id=1, is_active=True)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertEqual(snapshot["detections"][0]["event_level"], 9) # 4 + 2 active + 3 failure
|
||||
|
||||
def test_explicitly_allowed_threat_is_scored_and_correlated(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=5)],
|
||||
[make_threat(severity_id=2, is_active=False)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertEqual(snapshot["active_threats"][0]["event_level"], 11) # 6 + 3 + 2
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
||||
self.assertIn("explicitly_allowed", [item["rule"] for item in snapshot["correlations"]])
|
||||
|
||||
def test_repeated_detections_raise_score_and_correlation(self):
|
||||
events = [make_detection("detection-a"), make_detection("detection-b", detected_hours_ago=2)]
|
||||
snapshot = self.build(make_data(events, [make_threat(severity_id=1, is_active=True)]))
|
||||
self.assertEqual(len(snapshot["detections"]), 2)
|
||||
self.assertTrue(all(item["event_level"] == 8 for item in snapshot["detections"]))
|
||||
self.assertIn("same_threat_repeated", [item["rule"] for item in snapshot["correlations"]])
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "WARNING")
|
||||
|
||||
def test_repeated_remediation_failures_correlate_once_per_detection(self):
|
||||
events = [
|
||||
make_detection("failure-a", status_id=103, detected_hours_ago=1, action_success=False),
|
||||
make_detection("failure-b", status_id=103, detected_hours_ago=2, action_success=False),
|
||||
]
|
||||
snapshot = self.build(make_data(events, [make_threat(severity_id=1, is_active=False)]))
|
||||
failures = [item for item in snapshot["correlations"] if item["rule"] == "repeated_remediation_failures"]
|
||||
self.assertEqual(len(failures), 1)
|
||||
self.assertEqual(failures[0]["count"], 2)
|
||||
self.assertEqual(failures[0]["threat_ids"], ["100"])
|
||||
|
||||
def test_old_active_threat_remains_visible_outside_six_hour_window(self):
|
||||
event = make_detection(detected_hours_ago=10, changed_hours_ago=10)
|
||||
snapshot = self.build(make_data([event], [make_threat(severity_id=5, is_active=True)]))
|
||||
self.assertEqual(snapshot["summary"]["detection_count_6h"], 0)
|
||||
self.assertEqual(snapshot["summary"]["max_event_level_6h"], 0)
|
||||
self.assertEqual(snapshot["summary"]["active_count"], 1)
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 14)
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
||||
|
||||
def test_realtime_disabled_in_normal_mode_forces_error(self):
|
||||
snapshot = self.build(make_data(defender={"realtime_enabled": False}))
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 2)
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
||||
self.assertIn("Realtime protection disabled", snapshot["summary"]["protection_issues"])
|
||||
|
||||
def test_passive_mode_does_not_create_protection_false_positive(self):
|
||||
snapshot = self.build(make_data(defender={"running_mode": "Passive", "realtime_enabled": False}))
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 0)
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "OK")
|
||||
|
||||
def test_collection_error_is_fail_closed(self):
|
||||
snapshot = defender_setup.build_collection_error_snapshot("powershell_timeout")
|
||||
self.assertFalse(snapshot["collection_complete"])
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 16)
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
||||
|
||||
def test_missing_history_arrays_is_incomplete_and_error(self):
|
||||
snapshot = self.build({"defender": defender_state()})
|
||||
self.assertFalse(snapshot["collection_complete"])
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 16)
|
||||
|
||||
def test_singleton_powershell_objects_are_normalized_as_one_row(self):
|
||||
data = make_data(
|
||||
detections=make_detection(status_id=1),
|
||||
threats=make_threat(severity_id=1, is_active=True),
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertTrue(snapshot["collection_complete"])
|
||||
self.assertEqual(len(snapshot["detections"]), 1)
|
||||
self.assertEqual(snapshot["detections"][0]["threat_id"], "100")
|
||||
|
||||
def test_null_empty_collections_are_treated_as_empty_not_failed(self):
|
||||
data = make_data()
|
||||
data["detections"] = None
|
||||
data["threats"] = None
|
||||
snapshot = self.build(data)
|
||||
self.assertTrue(snapshot["collection_complete"])
|
||||
self.assertEqual(snapshot["detections"], [])
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "OK")
|
||||
|
||||
def test_same_threatid_with_distinct_detectionids_remains_two_incidents(self):
|
||||
detections = [
|
||||
make_detection("distinct-a", detected_hours_ago=1),
|
||||
make_detection("distinct-b", detected_hours_ago=2),
|
||||
make_detection("distinct-a", detected_hours_ago=1), # duplicate record of same detection
|
||||
]
|
||||
snapshot = self.build(make_data(detections, [make_threat(severity_id=1, is_active=True)]))
|
||||
self.assertEqual(len(snapshot["detections"]), 2)
|
||||
self.assertEqual(snapshot["summary"]["detection_count_6h"], 2)
|
||||
self.assertIn("same_threat_repeated", [item["rule"] for item in snapshot["correlations"]])
|
||||
|
||||
def test_old_detection_with_recent_status_change_enters_window(self):
|
||||
event = make_detection(status_id=6, detected_hours_ago=20, changed_hours_ago=1)
|
||||
snapshot = self.build(make_data([event], [make_threat(severity_id=5, is_active=False)]))
|
||||
self.assertEqual(snapshot["summary"]["detection_count_6h"], 1)
|
||||
self.assertEqual(snapshot["summary"]["max_event_level_6h"], 10)
|
||||
self.assertEqual(snapshot["summary"]["current_risk_level"], 0)
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "OK")
|
||||
|
||||
def test_legacy_catalog_severityid_is_normalized_to_documented_enum(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=1)],
|
||||
[make_threat(severity_id=3, source="SeverityID", is_active=True)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertEqual(snapshot["detections"][0]["severity_id"], 4)
|
||||
self.assertEqual(snapshot["detections"][0]["severity"], "High")
|
||||
self.assertEqual(snapshot["summary"]["max_active_level"], 11)
|
||||
|
||||
def test_severe_status_mapping_uses_threatseverityid_five(self):
|
||||
self.assertEqual(
|
||||
defender_setup.classify_detection({"threat_severity_id": 5, "status_id": 2}),
|
||||
9,
|
||||
)
|
||||
self.assertEqual(
|
||||
defender_setup.classify_detection({"threat_severity_id": 4, "status_id": 2}),
|
||||
6,
|
||||
)
|
||||
|
||||
def test_two_distinct_threats_correlate_and_protection_disabled_is_combined(self):
|
||||
events = [
|
||||
make_detection("threat-a", threat_id="100", status_id=1),
|
||||
make_detection("threat-b", threat_id="200", status_id=1, detected_hours_ago=2),
|
||||
]
|
||||
threats = [
|
||||
make_threat("100", severity_id=5, is_active=True, did_threat_execute=True),
|
||||
make_threat("200", severity_id=1, is_active=True),
|
||||
]
|
||||
snapshot = self.build(make_data(events, threats, defender={"realtime_enabled": False}))
|
||||
rules = [item["rule"] for item in snapshot["correlations"]]
|
||||
self.assertIn("multiple_threats", rules)
|
||||
self.assertIn("severe_with_protection_disabled", rules)
|
||||
severe = next(item for item in snapshot["detections"] if item["threat_id"] == "100")
|
||||
self.assertEqual(severe["event_level"], 16) # 12 + 2 active + 2 realtime disabled
|
||||
self.assertTrue(severe["did_threat_execute"])
|
||||
|
||||
def test_resources_are_reduced_to_types_and_counts(self):
|
||||
event = make_detection(resources=[r"file:_C:\Users\synthetic\private.txt", "registry:synthetic"])
|
||||
snapshot = self.build(make_data([event], [make_threat(is_active=True)]))
|
||||
serialized = json.dumps(snapshot)
|
||||
self.assertEqual(snapshot["detections"][0]["resource_count"], 2)
|
||||
self.assertEqual(snapshot["detections"][0]["resource_kinds"], ["file", "registry"])
|
||||
self.assertNotIn("synthetic\\private.txt", serialized)
|
||||
self.assertNotIn("Users", serialized)
|
||||
|
||||
def test_wapt_status_uses_current_risk_not_historical_event_maximum(self):
|
||||
data = make_data(
|
||||
[make_detection(status_id=4)],
|
||||
[make_threat(severity_id=5, is_active=False)],
|
||||
)
|
||||
snapshot = self.build(data)
|
||||
self.assertEqual(snapshot["summary"]["max_event_level_6h"], 9)
|
||||
self.assertEqual(snapshot["summary"]["audit_status"], "OK")
|
||||
|
||||
def test_classifier_modifiers_are_configurable_and_clamped(self):
|
||||
rules = {
|
||||
"active_modifier": 20,
|
||||
"status_modifiers": {1: 20},
|
||||
}
|
||||
self.assertEqual(
|
||||
defender_setup.classify_detection({"threat_severity_id": 5, "status_id": 1}, rules=rules),
|
||||
16,
|
||||
)
|
||||
|
||||
def test_wapt_thresholds_are_ok_warning_error(self):
|
||||
self.assertEqual(defender_setup._wapt_status(6, [], True, False, True, 0, 0), "OK")
|
||||
self.assertEqual(defender_setup._wapt_status(7, [], True, False, True, 0, 0), "WARNING")
|
||||
self.assertEqual(defender_setup._wapt_status(10, [], True, False, True, 0, 0), "ERROR")
|
||||
|
||||
|
||||
class TestDefenderCollectionAndWaptAdapter(unittest.TestCase):
|
||||
def test_powershell_timeout_is_a_safe_collection_error(self):
|
||||
with patch.object(defender_setup.subprocess, "run", side_effect=defender_setup.subprocess.TimeoutExpired("powershell", 90)):
|
||||
with self.assertRaises(defender_setup.DefenderCollectionError) as raised:
|
||||
defender_setup.powershell_json("synthetic script", timeout=90)
|
||||
self.assertEqual(raised.exception.code, "powershell_timeout")
|
||||
|
||||
def test_powershell_json_accepts_bom_and_never_uses_executionpolicy_bypass(self):
|
||||
payload = {"collection_complete": True, "defender": {}, "threats": [], "detections": []}
|
||||
completed = types.SimpleNamespace(returncode=0, stdout=("\ufeff" + json.dumps(payload)).encode("utf-8"), stderr=b"")
|
||||
with patch.object(defender_setup.subprocess, "run", return_value=completed) as run:
|
||||
result = defender_setup.powershell_json("synthetic script")
|
||||
args = run.call_args.args[0]
|
||||
self.assertEqual(result, payload)
|
||||
self.assertNotIn("-ExecutionPolicy", args)
|
||||
self.assertNotIn("Bypass", args)
|
||||
self.assertEqual(run.call_args.kwargs["timeout"], defender_setup.POWERSHELL_TIMEOUT_SECONDS)
|
||||
|
||||
def test_audit_timeout_returns_error_and_stores_failed_snapshot(self):
|
||||
wapt = Mock()
|
||||
with patch.object(defender_setup, "WAPT", wapt, create=True), patch.object(
|
||||
defender_setup, "collect_defender", side_effect=defender_setup.DefenderCollectionError("powershell_timeout")
|
||||
):
|
||||
result = defender_setup.audit()
|
||||
self.assertEqual(result, "ERROR")
|
||||
stored = [call.args for call in wapt.write_audit_data_if_changed.call_args_list]
|
||||
snapshot_call = next(args for args in stored if args[1] == "snapshot")
|
||||
self.assertFalse(snapshot_call[2]["collection_complete"])
|
||||
self.assertEqual(snapshot_call[2]["summary"]["audit_status"], "ERROR")
|
||||
|
||||
def test_wapt_audit_api_uses_official_signature_and_json_value(self):
|
||||
wapt = Mock()
|
||||
value = {"schema_version": 2, "detections": [], "correlations": []}
|
||||
with patch.object(defender_setup, "WAPT", wapt, create=True):
|
||||
defender_setup.save_audit("snapshot", value)
|
||||
wapt.write_audit_data_if_changed.assert_called_once_with(
|
||||
"Microsoft Defender", "snapshot", value,
|
||||
max_count=512, keep_days=90,
|
||||
)
|
||||
|
||||
def test_audit_hook_returns_wapt_status_and_writes_v2_snapshot(self):
|
||||
wapt = Mock()
|
||||
with patch.object(defender_setup, "WAPT", wapt, create=True), patch.object(
|
||||
defender_setup, "collect_defender", return_value=make_data()
|
||||
):
|
||||
result = defender_setup.audit()
|
||||
self.assertEqual(result, "OK")
|
||||
written_keys = [call.args[1] for call in wapt.write_audit_data_if_changed.call_args_list]
|
||||
self.assertEqual(written_keys, ["snapshot", "summary", "detections", "active_threats", "correlations"])
|
||||
snapshot_call = wapt.write_audit_data_if_changed.call_args_list[0]
|
||||
self.assertEqual(snapshot_call.args[2]["schema_version"], 2)
|
||||
|
||||
def test_control_declares_windows_wapt_26_and_15_minute_audit(self):
|
||||
fields = {}
|
||||
for line in (ROOT / "WAPT" / "control").read_text(encoding="utf-8").splitlines():
|
||||
if line.strip() and not line.lstrip().startswith("#"):
|
||||
key, value = line.split(":", 1)
|
||||
fields[key.strip()] = value.strip()
|
||||
self.assertEqual(fields["package"], "comit-audit-defender")
|
||||
self.assertEqual(fields["min_wapt_version"], "2.6")
|
||||
self.assertEqual(fields["audit_schedule"], "15m")
|
||||
self.assertEqual(fields["target_os"], "windows")
|
||||
|
||||
def test_powershell_collection_is_local_read_only_and_does_not_emit_raw_resources(self):
|
||||
script = defender_setup.POWERSHELL_SCRIPT
|
||||
self.assertIn("Get-MpComputerStatus", script)
|
||||
self.assertIn("Get-MpThreatDetection", script)
|
||||
self.assertIn("Get-MpThreat -ErrorAction Stop", script)
|
||||
self.assertNotIn("Invoke-WebRequest", script)
|
||||
self.assertNotIn("Remove-MpThreat", script)
|
||||
self.assertNotIn("Set-MpPreference", script)
|
||||
self.assertNotIn("resources = @($_.Resources)", script)
|
||||
self.assertNotIn("Select-Object -First 100", script)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user