669 lines
33 KiB
Python
669 lines
33 KiB
Python
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 MemoryWaptAudit(object):
|
|
"""Small fake implementing the WAPT 2.6 audit read/write contract."""
|
|
|
|
def __init__(self):
|
|
self.values = {}
|
|
self.history = []
|
|
|
|
def read_audit_data(self, section, key, default=None, ptype=None, include_expired_data=True):
|
|
return self.values.get((section, key), default)
|
|
|
|
def write_audit_data_if_changed(self, section, key, value, ptype=None,
|
|
value_date=None, expiration_date=None,
|
|
max_count=2, keep_days=None):
|
|
storage_key = (section, key)
|
|
previous = self.values.get(storage_key)
|
|
if previous != value:
|
|
self.values[storage_key] = value
|
|
self.history.append((storage_key, value, max_count, keep_days))
|
|
return previous
|
|
|
|
|
|
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_is_resolved_but_requires_investigation(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)
|
|
item = snapshot["detections"][0]
|
|
self.assertEqual(snapshot["active_threats"], [])
|
|
self.assertEqual(item["remediation_state"], "removed")
|
|
self.assertFalse(item["active"])
|
|
self.assertTrue(item["execution_observed"])
|
|
self.assertTrue(item["requires_investigation"])
|
|
self.assertEqual(item["event_level"], 9) # event severity excludes execution/investigation context
|
|
self.assertEqual(item["investigation_level"], 9)
|
|
self.assertEqual(snapshot["summary"]["max_active_level"], 0)
|
|
self.assertEqual(snapshot["summary"]["max_investigation_level"], 9)
|
|
self.assertEqual(snapshot["summary"]["audit_status"], "WARNING")
|
|
|
|
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"], "currently_executing")
|
|
self.assertTrue(snapshot["active_threats"][0]["execution_observed"])
|
|
|
|
def test_allowed_execution_is_not_active_but_requires_investigation(self):
|
|
event = make_detection(status_id=4)
|
|
event["execution_status_id"] = 2
|
|
snapshot = self.build(make_data([event], [make_threat(severity_id=5, is_active=False)]))
|
|
item = snapshot["detections"][0]
|
|
self.assertFalse(item["active"])
|
|
self.assertFalse(item["execution_observed"])
|
|
self.assertTrue(item["requires_investigation"])
|
|
self.assertEqual(item["execution_status"], "Allowed")
|
|
|
|
def test_blocked_detection_does_not_imply_execution_or_compromise(self):
|
|
event = make_detection(status_id=6)
|
|
event["execution_status_id"] = 1
|
|
snapshot = self.build(make_data([event], [make_threat(severity_id=5, is_active=False)]))
|
|
item = snapshot["detections"][0]
|
|
self.assertEqual(item["remediation_state"], "blocked")
|
|
self.assertFalse(item["active"])
|
|
self.assertFalse(item["execution_observed"])
|
|
self.assertFalse(item["requires_investigation"])
|
|
self.assertEqual(snapshot["summary"]["current_risk_level"], 0)
|
|
self.assertEqual(snapshot["summary"]["audit_status"], "OK")
|
|
|
|
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 + 2 active + 3 failure
|
|
self.assertEqual(snapshot["summary"]["remediation_failed"], True)
|
|
self.assertEqual(snapshot["summary"]["current_risk_level"], 9)
|
|
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
|
|
|
def test_action_success_false_without_error_is_not_assumed_to_be_failed_remediation(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]["remediation_state"], "detected")
|
|
self.assertEqual(snapshot["detections"][0]["event_level"], 6) # 4 + 2 unresolved
|
|
self.assertFalse(snapshot["summary"]["remediation_failed"])
|
|
self.assertNotEqual(snapshot["summary"]["audit_status"], "ERROR")
|
|
|
|
def test_nonzero_error_with_failed_action_forces_error(self):
|
|
data = make_data(
|
|
[make_detection(status_id=1, action_success=False, error_code=2142207991)],
|
|
[make_threat(severity_id=1, is_active=True)],
|
|
)
|
|
snapshot = self.build(data)
|
|
self.assertEqual(snapshot["detections"][0]["remediation_state"], "failed")
|
|
self.assertEqual(snapshot["detections"][0]["event_level"], 9)
|
|
self.assertEqual(snapshot["summary"]["audit_status"], "ERROR")
|
|
|
|
def test_action_success_default_false_does_not_create_failure_correlation(self):
|
|
events = [
|
|
make_detection("default-false-a", status_id=1, action_success=False),
|
|
make_detection("default-false-b", status_id=1, action_success=False),
|
|
]
|
|
snapshot = self.build(make_data(
|
|
events, [make_threat(severity_id=1, is_active=True)],
|
|
))
|
|
self.assertTrue(all(item["remediation_state"] == "detected" for item in snapshot["detections"]))
|
|
rules = {item["rule"] for item in snapshot["correlations"]}
|
|
self.assertNotIn("repeated_remediation_failures", rules)
|
|
|
|
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"] == 6 for item in snapshot["detections"]))
|
|
self.assertEqual(snapshot["summary"]["max_event_level_6h"], 6)
|
|
self.assertEqual(snapshot["summary"]["correlation_level"], 8)
|
|
self.assertEqual(snapshot["summary"]["current_risk_level"], 8)
|
|
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]["count"], 2)
|
|
|
|
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_threat_rollup_isactive_does_not_reopen_a_resolved_detectionid(self):
|
|
events = [
|
|
make_detection("still-detected", status_id=1),
|
|
make_detection("already-removed", status_id=4, detected_hours_ago=2),
|
|
]
|
|
snapshot = self.build(make_data(
|
|
events, [make_threat(severity_id=1, is_active=True)],
|
|
))
|
|
active_ids = [item["detection_id"] for item in snapshot["active_threats"]]
|
|
removed = next(item for item in snapshot["detections"] if item["detection_id"] == "already-removed")
|
|
self.assertEqual(active_ids, ["still-detected"])
|
|
self.assertFalse(removed["active"])
|
|
self.assertEqual(removed["remediation_state"], "removed")
|
|
self.assertTrue(removed["requires_investigation"]) # aggregated IsActive is a conflict signal
|
|
|
|
def test_fallback_event_identity_survives_status_change_and_is_flagged(self):
|
|
first = make_detection(detection_id=None, status_id=1)
|
|
first.pop("detection_id")
|
|
first_snapshot = self.build(make_data([first], [make_threat(severity_id=1, is_active=True)]))
|
|
second = dict(first)
|
|
second["status_id"] = 4
|
|
second["changed_at"] = NOW.isoformat().replace("+00:00", "Z")
|
|
second_snapshot = self.build(make_data([second], [make_threat(severity_id=1, is_active=False)]))
|
|
self.assertEqual(first_snapshot["detections"][0]["event_id"], second_snapshot["detections"][0]["event_id"])
|
|
self.assertFalse(second_snapshot["detections"][0]["identity_complete"])
|
|
self.assertGreater(second_snapshot["summary"]["identity_incomplete_count"], 0)
|
|
|
|
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=False),
|
|
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"], 14) # host protection is not part of event score
|
|
self.assertEqual(snapshot["summary"]["correlation_level"], 16)
|
|
self.assertEqual(snapshot["summary"]["current_risk_level"], 16)
|
|
self.assertFalse(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_current_protection_change_does_not_rewrite_event_level(self):
|
|
event = make_detection(status_id=4)
|
|
threat = make_threat(severity_id=5, is_active=False)
|
|
healthy = self.build(make_data([event], [threat]))
|
|
degraded = self.build(make_data([event], [threat], defender={"realtime_enabled": False}))
|
|
self.assertEqual(healthy["detections"][0]["event_level"], 9)
|
|
self.assertEqual(degraded["detections"][0]["event_level"], 9)
|
|
self.assertEqual(healthy["summary"]["current_risk_level"], 0)
|
|
self.assertGreater(degraded["summary"]["current_risk_level"], 0)
|
|
|
|
def test_old_resolved_event_with_execution_remains_investigation_record(self):
|
|
event = make_detection(status_id=4, detected_hours_ago=24 * 45, changed_hours_ago=24 * 45)
|
|
snapshot = self.build(make_data(
|
|
[event], [make_threat(severity_id=5, is_active=False, did_threat_execute=True)],
|
|
))
|
|
self.assertEqual(len(snapshot["detections"]), 1)
|
|
self.assertFalse(snapshot["detections"][0]["active"])
|
|
self.assertTrue(snapshot["detections"][0]["requires_investigation"])
|
|
self.assertEqual(snapshot["summary"]["max_active_level"], 0)
|
|
self.assertGreater(snapshot["summary"]["max_investigation_level"], 0)
|
|
|
|
def test_distinct_event_and_correlation_levels_have_scoring_reasons(self):
|
|
events = [
|
|
make_detection("corr-a", threat_id="100"),
|
|
make_detection("corr-b", threat_id="200", detected_hours_ago=2),
|
|
]
|
|
snapshot = self.build(make_data(
|
|
events,
|
|
[make_threat("100", severity_id=1, is_active=True), make_threat("200", severity_id=1, is_active=True)],
|
|
))
|
|
self.assertEqual(snapshot["summary"]["max_event_level_6h"], 6)
|
|
self.assertEqual(snapshot["summary"]["correlation_level"], 8)
|
|
self.assertEqual(snapshot["summary"]["max_active_level"], 6)
|
|
self.assertEqual(snapshot["summary"]["current_risk_level"], 8)
|
|
self.assertTrue(snapshot["detections"][0]["event_level_reasons"])
|
|
self.assertTrue(snapshot["correlations"][0]["scoring_reasons"])
|
|
|
|
def test_investigation_state_is_retained_beyond_collection_history(self):
|
|
event = make_detection(status_id=4, detected_hours_ago=24 * 45, changed_hours_ago=24 * 45)
|
|
snapshot = self.build(make_data(
|
|
[event], [make_threat(severity_id=5, is_active=False, did_threat_execute=True)],
|
|
))
|
|
self.assertEqual(len(snapshot["detections"]), 1)
|
|
self.assertFalse(snapshot["detections"][0]["active"])
|
|
self.assertTrue(snapshot["detections"][0]["requires_investigation"])
|
|
self.assertEqual(snapshot["summary"]["max_active_level"], 0)
|
|
self.assertGreater(snapshot["summary"]["max_investigation_level"], 0)
|
|
|
|
def test_serialized_audit_values_shrink_for_small_medium_and_large_sets(self):
|
|
for size in (1, 100, 1000):
|
|
with self.subTest(detections=size):
|
|
events = [
|
|
make_detection("size-%d" % index, threat_id=str(index + 100))
|
|
for index in range(size)
|
|
]
|
|
threats = [
|
|
make_threat(str(index + 100), severity_id=1, is_active=True)
|
|
for index in range(size)
|
|
]
|
|
snapshot = self.build(make_data(events, threats))
|
|
legacy = {
|
|
"snapshot": snapshot,
|
|
"summary": snapshot["summary"],
|
|
"detections": snapshot["detections"],
|
|
"active_threats": snapshot["active_threats"],
|
|
"correlations": snapshot["correlations"],
|
|
}
|
|
compact = defender_setup.build_audit_records(snapshot)
|
|
old_size = len(json.dumps(legacy, separators=(",", ":")).encode("utf-8"))
|
|
new_size = len(json.dumps(compact, separators=(",", ":")).encode("utf-8"))
|
|
self.assertLess(new_size, old_size)
|
|
|
|
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")
|
|
self.assertEqual(
|
|
defender_setup._wapt_status(9, [], True, False, True, 0, 0, remediation_failed=True),
|
|
"ERROR",
|
|
)
|
|
self.assertEqual(
|
|
defender_setup._wapt_status(
|
|
9, [], True, False, True, 0, 0, remediation_failed=True,
|
|
rules={"failed_remediation_forces_error": False},
|
|
),
|
|
"WARNING",
|
|
)
|
|
|
|
|
|
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_summary(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]
|
|
summary_call = next(args for args in stored if args[1] == "summary")
|
|
self.assertFalse(summary_call[2]["collection_complete"])
|
|
self.assertEqual(summary_call[2]["summary"]["audit_status"], "ERROR")
|
|
|
|
def test_wapt_audit_api_uses_official_signature_and_json_value(self):
|
|
wapt = Mock()
|
|
value = {"schema_version": 3, "summary": {}}
|
|
with patch.object(defender_setup, "WAPT", wapt, create=True):
|
|
defender_setup.save_audit("summary", value)
|
|
wapt.write_audit_data_if_changed.assert_called_once_with(
|
|
"Microsoft Defender", "summary", value,
|
|
max_count=defender_setup.AUDIT_SUMMARY_HISTORY_MAX_COUNT, keep_days=90,
|
|
)
|
|
|
|
def test_audit_hook_returns_wapt_status_and_writes_compact_v3_records(self):
|
|
wapt = Mock()
|
|
with patch.object(defender_setup, "WAPT", wapt, create=True), patch.object(
|
|
defender_setup, "collect_defender", return_value=make_data(
|
|
[make_detection()], [make_threat(severity_id=1, is_active=True)],
|
|
)
|
|
):
|
|
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[:2], ["summary", "correlations"])
|
|
self.assertTrue(written_keys[2].startswith("event."))
|
|
summary_call = wapt.write_audit_data_if_changed.call_args_list[0]
|
|
self.assertEqual(summary_call.args[2]["schema_version"], 3)
|
|
correlations_call = wapt.write_audit_data_if_changed.call_args_list[1]
|
|
self.assertEqual(correlations_call.args[2]["schema_version"], 3)
|
|
wapt.read_audit_data.assert_called_once_with(
|
|
"Microsoft Defender", written_keys[2], default=None, include_expired_data=False,
|
|
)
|
|
|
|
def test_event_envelope_is_provider_neutral_and_tracks_resolution(self):
|
|
initial = defender_setup.build_audit_snapshot(make_data(
|
|
[make_detection("stable-detection", status_id=1)],
|
|
[make_threat(severity_id=5, is_active=True)],
|
|
), now=NOW)
|
|
first = defender_setup.build_audit_records(initial)
|
|
event_key = next(key for key in first if key.startswith("event."))
|
|
event = first[event_key]
|
|
self.assertEqual(event["schema_version"], 1)
|
|
self.assertEqual(event["source"], "comi-audit-defender")
|
|
self.assertEqual(event["event_type"], "threat.detected")
|
|
self.assertNotIn("threat_name", event)
|
|
|
|
resolved_input = make_detection("stable-detection", status_id=4, changed_hours_ago=0)
|
|
resolved = defender_setup.build_audit_snapshot(make_data(
|
|
[resolved_input], [make_threat(severity_id=5, is_active=False)],
|
|
), now=NOW)
|
|
previous = {event["event_id"]: event}
|
|
second = defender_setup.build_audit_records(resolved, previous_events=previous)
|
|
resolved_event = second[event_key]
|
|
self.assertEqual(resolved_event["event_id"], event["event_id"])
|
|
self.assertEqual(resolved_event["event_type"], "threat.resolved")
|
|
self.assertEqual(resolved_event["remediation_state"], "removed")
|
|
|
|
def test_repeated_audit_writes_only_changed_values_and_historizes_event_transition(self):
|
|
fake_wapt = MemoryWaptAudit()
|
|
detection = make_detection("audit-transition", status_id=1)
|
|
first_snapshot = defender_setup.build_audit_snapshot(make_data(
|
|
[detection], [make_threat(severity_id=1, is_active=True)],
|
|
), now=NOW)
|
|
with patch.object(defender_setup, "WAPT", fake_wapt, create=True):
|
|
defender_setup._save_snapshot(first_snapshot)
|
|
first_write_count = len(fake_wapt.history)
|
|
defender_setup._save_snapshot(first_snapshot)
|
|
self.assertEqual(len(fake_wapt.history), first_write_count)
|
|
|
|
resolved = make_detection("audit-transition", status_id=4, changed_hours_ago=0)
|
|
second_snapshot = defender_setup.build_audit_snapshot(make_data(
|
|
[resolved], [make_threat(severity_id=1, is_active=False)],
|
|
), now=NOW)
|
|
defender_setup._save_snapshot(second_snapshot)
|
|
|
|
event_key = next(key for section, key in fake_wapt.values if key.startswith("event."))
|
|
stored_event = fake_wapt.values[("Microsoft Defender", event_key)]
|
|
self.assertEqual(stored_event["event_type"], "threat.resolved")
|
|
self.assertEqual(stored_event["remediation_state"], "removed")
|
|
self.assertNotIn(("Microsoft Defender", "snapshot"), fake_wapt.values)
|
|
self.assertNotIn(("Microsoft Defender", "detections"), fake_wapt.values)
|
|
self.assertEqual(fake_wapt.values[("Microsoft Defender", "summary")]["schema_version"], 3)
|
|
|
|
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["version"], "2.0-2")
|
|
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()
|