diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..991c3cd --- /dev/null +++ b/.gitignore @@ -0,0 +1,46 @@ +# Do not commit binaries: +*.exe +*.msu +*.msi +*.zip +*.rpm +*.deb +*.yum +*.dmg +*.vsix +*.appx +*.appxbundle +*.msix +*.msixbundle +*.iso +*.cab +*.pkg +*.jar +*.msp +*.part +*.bak + +# Do not commit generated WAPT files: +*.pyc +*.crt +*.sha256 +*.wapt +.env +*.psproj +**/.vscode/ +signature +*.sha1 + +# Do not commit template part: +template.py +.vscode/*.json +code_template.cmd +/template +package_template.py + +# Specific to this package (use ! as prefix to allow commit): +!README.md +!setup.py +!WAPT/control +!tests/ +!tests/** diff --git a/README.md b/README.md new file mode 100644 index 0000000..4289afc --- /dev/null +++ b/README.md @@ -0,0 +1,107 @@ +# comit-audit-defender + +Paquet WAPT Enterprise 2.6 d'audit **en lecture seule** de Microsoft Defender. Il ne déploie pas d'agent supplémentaire, ne modifie aucune préférence Defender, ne supprime aucune quarantaine et n'effectue aucun accès réseau externe. Le hook natif `audit()` retourne exclusivement `OK`, `WARNING` ou `ERROR`. + +Le paquet est planifié toutes les 15 minutes par `WAPT/control`. La collecte PowerShell est locale et bornée à 90 secondes; elle utilise `Get-MpComputerStatus`, `Get-MpThreat` et `Get-MpThreatDetection`. Une erreur PowerShell, un timeout, une réponse vide ou un JSON invalide produit `ERROR` et un état `collection_complete=false`, jamais un poste sain. + +## Classification 0–16 + +La gravité Defender intrinsèque est conservée dans `severity_id` et le risque calculé dans `event_level`. Les identifiants canoniques `MPTHREAT_SEVERITY` sont `0 Unknown`, `1 Low`, `2 Moderate`, `4 High`, `5 Severe`; WMI `SeverityID` (`0..4`) est normalisé (`3 High` vers `4`, `4 Severe` vers `5`) avant classification. Les niveaux de base sont respectivement 5, 4, 6, 9 et 12. + +| Niveau | Catégorie | +|---:|---| +| 0 | Aucun événement | +| 1–3 | Information | +| 4–6 | Faible | +| 7–9 | Modéré | +| 10–12 | Élevé | +| 13–14 | Critique | +| 15–16 | Urgence | + +Règles par défaut, définies dans `DEFAULT_RULES` et surchargeables par les fonctions pures pour les tests : + +| Contexte Defender | Modificateur | +|---|---:| +| Cleaned, Quarantined, Removed | -3 | +| Blocked | -2 | +| Detected / état non résolu | +2 | +| Failed remediation | +3 | +| Allowed | +3 | +| Répétition de la même menace dans la fenêtre | +2 | +| Temps réel désactivé en mode Normal | +2 | +| Antivirus / service désactivé en mode Normal (chaque signal) | +2 | +| Surveillance comportementale ou IOAV désactivée | +1 chacun | +| Signatures anciennes de plus de 3 jours / âge inconnu | +1 | + +Le résultat est borné à `[0,16]`. Ces poids et les corrélations sont des règles métier Comitari, inspirées de Wazuh; ils ne reproduisent pas son moteur. L'échelle WAPT dépend de `current_risk_level`, jamais du maximum historique : 0–6 `OK`, 7–9 `WARNING`, 10–16 `ERROR`. En mode `Normal`, antivirus, service ou temps réel désactivé force `ERROR` indépendamment du score. `Passive`, `SxS Passive Mode` et `EDR Block Mode` ne déclenchent pas à eux seuls de faux incident; les menaces restent classifiées. + +`Quarantined` reçoit une réduction de score mais reste dans `active_threats`, car la quarantaine ne prouve pas l'absence de compromission. `Cleaned`, `Removed` et `Blocked` ne maintiennent pas seuls le risque courant; toutefois `IsActive`, `DidThreatExecute` ou un état d'exécution `Allowed`/`Executing` les maintient comme risque non résolu. Les événements récents restent aussi dans `detections` et l'historique WAPT. `DidThreatExecute` et `CurrentThreatExecutionStatusID` sont des signaux explicites, séparés du nom et de la sévérité de signature; aucun nom (y compris « ransomware ») n'est interprété comme preuve d'exécution ou de comportement. + +## Fenêtres et corrélations + +- Fenêtre glissante par défaut : six heures; une détection y entre si sa première détection **ou** son dernier changement de statut se trouve dans cette fenêtre. +- Les événements résolus sont conservés dans la collecte courante pendant 30 jours. Toute menace non résolue est conservée sans limite d'ancienneté, donc une menace active de plus de six heures ne disparaît pas. +- `max_event_level_6h` porte sur les événements récents, `max_active_level` sur toutes les menaces non résolues et `current_risk_level` sur le maximum du risque actif et de l'état de protection. +- Corrélations sur six heures : répétition d'une même menace, menaces distinctes multiples, menace High/Severe avec temps réel désactivé, échecs de remédiation répétés et menace explicitement autorisée. +- `DetectionID` est la clé de déduplication documentée, combinée à `ThreatID`. En son absence, un composite stable de `ThreatID`, dates, statut et empreinte interne des ressources est utilisé; l'identité incomplète est signalée dans le résumé. Les événements distincts avec des DetectionID différents ne sont pas fusionnés parce qu'ils partagent un ThreatID. +- Il n'y a plus de limite arbitraire à 100 détections. La collecte transmet tous les événements de moins de 30 jours ainsi que les états non résolus plus anciens. Les données étant complètes ou la collecte échouant explicitement, aucun incident critique n'est silencieusement tronqué. + +## Audit WAPT et Reporting + +L'API utilisée est `WAPT.write_audit_data_if_changed(section, key, value, ptype=None, value_date=None, expiration_date=None, max_count=2, keep_days=None)`, documentée dans l'API Setuphelpers WAPT 2.6. Les valeurs sont des dictionnaires/listes JSON natifs, consultables dans Audit Data / Reporting. Le paquet écrit un agrégat `snapshot` version 2 et conserve aussi les clés historiques `summary`, `detections` et `active_threats`, plus `correlations`. La collecte n'ajoute pas d'horodatage courant qui rendrait chaque valeur différente. WAPT conserve jusqu'à 512 changements pendant 90 jours, selon la première limite atteinte. + +Exemple synthétique de structure : + +```json +{ + "schema_version": 2, + "collection_complete": true, + "defender": { + "running_mode": "Normal", + "antivirus_enabled": true, + "realtime_enabled": true, + "behavior_monitor_enabled": true, + "ioav_enabled": true, + "signature_age": 0, + "signature_version": "1.000.0.0", + "signature_last_updated": "2026-10-27T09:00:00Z", + "engine_version": "1.1.0.0", + "service_enabled": true + }, + "summary": { + "max_event_level_6h": 0, + "max_active_level": 0, + "current_risk_level": 0, + "current_risk_category": "Aucun événement", + "audit_status": "OK", + "collection_complete": true, + "detection_count_6h": 0, + "active_count": 0, + "resolved_count_6h": 0, + "identity_incomplete_count": 0, + "unknown_severity_count": 0, + "protection_issues": [], + "correlation_count": 0 + }, + "detections": [], + "active_threats": [], + "correlations": [] +} +``` + +## Données et limites de collecte + +- `Get-MpThreatDetection` fournit le ThreatID, DetectionID, ThreatStatusID, dates de détection/changement/remédiation, code d'erreur, réussite d'action, état d'exécution et ressources. `Get-MpThreat` est joint en Python par `ThreatID` pour ThreatName, sévérité, IsActive et DidThreatExecute. +- Les chemins complets des ressources, noms d'utilisateurs et noms de processus ne sont ni sérialisés ni transmis. Seuls le nombre et les catégories `file`, `process`, `registry`, `service`, `url` ou `other` sont conservés. +- Defender peut ne pas exposer une propriété sur une version donnée. Les valeurs absentes sont traitées comme inconnues; une détection de sévérité inconnue reçoit le niveau de base `Unknown`. Une panne de la cmdlet entière force `ERROR`. +- Les valeurs de `ThreatStatusID` utilisées viennent de `MSFT_MpThreatDetection` : 0 Unknown, 1 Detected, 2 Cleaned, 3 Quarantined, 4 Removed, 5 Allowed, 6 Blocked, 102 QuarantineFailed, 103 RemoveFailed, 104 AllowFailed, 105 Abandoned, 107 BlockedFailed. Les différents enums Microsoft peuvent différer selon l'API; ce paquet cible les objets des cmdlets Defender. +- Defender ne livre pas une chronologie complète des actions humaines ni une preuve d'exécution malveillante. `DidThreatExecute` et `CurrentThreatExecutionStatusID` sont exposés lorsque disponibles, sans déduction comportementale sur le nom. +- La réponse PowerShell contient l'ensemble des événements récents et actifs pertinents; le timeout de 90 s provoque un état `ERROR`, sans limite de 100 qui masquerait une détection. + +## Tests locaux + +Les tests utilisent uniquement des données synthétiques et des mocks Python; aucune commande Defender réelle ni WAPT n'est invoquée : + +```shell +python -m unittest discover -s tests -v +``` diff --git a/WAPT/control b/WAPT/control index 5f24d30..853fce7 100644 --- a/WAPT/control +++ b/WAPT/control @@ -1,43 +1,17 @@ package : comit-audit-defender -version : 0-2 +version : 2.0-1 architecture : all section : base priority : optional -name : -categories : maintainer : Comitari,Pierre Cosson -description : Monitors Microsoft Defender security status, threat detections, and remediation actions through WAPT audits. Classifies threats using a Wazuh-inspired severity scale (0–16), correlates security events over a rolling 6-hour window, and reports active threats, protection issues, and security risks to the WAPT console. -depends : -conflicts : -maturity : -locale : +description : Monitors Microsoft Defender security status, detections, and remediation through WAPT audits. Classifies risk on a Wazuh-inspired 0-16 scale and correlates events over six hours. +description_fr : Audite l’état de Microsoft Defender via WAPT, classe le risque de 0 à 16 et corrèle les événements sur six heures. +depends : +conflicts : +maturity : DEV +locale : fr target_os : windows -min_wapt_version : -sources : -installed_size : -impacted_process : -description_fr : -description_pl : -description_de : -description_es : -description_pt : -description_it : -description_nl : -description_ru : -audit_schedule : -editor : -keywords : -licence : -homepage : -package_uuid : -valid_from : -valid_until : -forced_install_on : -changelog : -min_os_version : -max_os_version : -icon_sha256sum : -signer : -signer_fingerprint: -signature_date : -signed_attributes : +min_wapt_version : 2.6 +audit_schedule : 15m +keywords : defender,audit,security +homepage : https://git.comitari.fr/paquets-publiques/comi-audit-defender diff --git a/setup.py b/setup.py index c4f2ed2..8c8f3b7 100644 --- a/setup.py +++ b/setup.py @@ -1,17 +1,41 @@ - # -*- coding: utf-8 -*- -from setuphelpers import * +"""Read-only Microsoft Defender audit for WAPT Enterprise 2.6.""" + +from setuphelpers import * # noqa: F401,F403 - WAPT injects the audit API here. + import json -import subprocess import os -from datetime import datetime, timedelta +import subprocess +from collections import Counter +from datetime import datetime, timedelta, timezone + AUDIT_CATEGORY = "Microsoft Defender" HISTORY_DAYS = 30 -SIGNATURE_MAX_AGE = 3 -MAX_DETECTIONS = 100 +AUDIT_HISTORY_DAYS = 90 +AUDIT_HISTORY_MAX_COUNT = 512 +POWERSHELL_TIMEOUT_SECONDS = 90 +# MPTHREAT_SEVERITY values used by the Defender threat API. SeverityID on the +# WMI catalog uses 0..4 instead; PowerShell normalizes that representation. +SEVERITY_BASE_LEVELS = { + 0: 5, # Unknown + 1: 4, # Low + 2: 6, # Moderate + 4: 9, # High + 5: 12, # Severe +} +SEVERITY_NAMES = { + 0: "Unknown", + 1: "Low", + 2: "Moderate", + 4: "High", + 5: "Severe", +} + +# Values exposed by the MSFT_MpThreatDetection class used by the Defender +# PowerShell cmdlet. Quarantine remains unresolved; it is never called clean. THREAT_STATUSES = { 0: "Unknown", 1: "Detected", @@ -26,264 +50,856 @@ THREAT_STATUSES = { 105: "Abandoned", 107: "BlockedFailed", } +RESOLVED_STATUS_IDS = {2, 4, 6} +FAILED_STATUS_IDS = {102, 103, 104, 107} -ACTIVE_STATUSES = {1, 5, 105} -FAILED_STATUSES = {102, 103, 104, 107} -RESOLVED_STATUSES = {2, 3, 4, 6} +DEFAULT_RULES = { + "window_hours": 6, + "signature_max_age_days": 3, + "active_modifier": 2, + "repeated_modifier": 2, + "failed_remediation_modifier": 3, + "status_modifiers": { + 2: -3, # cleaned + 3: -3, # quarantined, but still unresolved + 4: -3, # removed + 5: 3, # explicitly allowed + 6: -2, # blocked + 102: 3, + 103: 3, + 104: 3, + 107: 3, + }, + "protection_modifiers": { + "service_disabled": 2, + "antivirus_disabled": 2, + "realtime_disabled": 2, + "behavior_monitor_disabled": 1, + "ioav_disabled": 1, + "signatures_stale": 1, + "signature_age_unknown": 1, + }, + "repeated_threshold": 2, + "remediation_failure_threshold": 2, + "multiple_threats_threshold": 2, + "audit_history_days": AUDIT_HISTORY_DAYS, + "audit_history_max_count": AUDIT_HISTORY_MAX_COUNT, + "expected_active_modes": ("Normal",), + "passive_modes": ("Passive", "SxS Passive Mode", "EDR Block Mode"), + "correlations": { + "same_threat_repeated": True, + "multiple_threats": True, + "severe_with_protection_disabled": True, + "repeated_remediation_failures": True, + "explicitly_allowed": True, + }, +} + + +class DefenderCollectionError(RuntimeError): + """Collection error with a safe, non-sensitive reason code.""" + + def __init__(self, code): + self.code = code + RuntimeError.__init__(self, code) def install(): + """Keep the existing package install hook; no Defender changes are made.""" print("Microsoft Defender audit package installed") -def powershell_json(script): +POWERSHELL_SCRIPT = r''' +$ErrorActionPreference = 'Stop' +$WarningPreference = 'SilentlyContinue' +$ProgressPreference = 'SilentlyContinue' + +function Get-SafeInt($Value) { + if ($null -eq $Value -or "$Value" -eq '') { return $null } + try { return [int]$Value } catch { return $null } +} + +function Format-Utc($Value) { + if ($null -eq $Value -or "$Value" -eq '') { return $null } + try { + if ($Value -is [datetime]) { $date = $Value } + else { + $date = [datetime]::MinValue + if (-not [datetime]::TryParse([string]$Value, [ref]$date)) { + return $null + } + } + return $date.ToUniversalTime().ToString('o') + } catch { return $null } +} + +function Get-ResourceSummary($Values) { + $items = @($Values | Where-Object { $null -ne $_ -and "$($_)" -ne '' }) + $kinds = @( + $items | ForEach-Object { + $text = [string]$_ + $kind = ($text -split ':', 2)[0].ToLowerInvariant() + if ($kind -notin @('file', 'process', 'registry', 'service', 'url')) { + $kind = 'other' + } + $kind + } | Sort-Object -Unique + ) + return [PSCustomObject]@{ + resource_count = [int]$items.Count + resource_kinds = $kinds + } +} + +$now = [datetime]::UtcNow +$historyCutoff = $now.AddDays(-30) +$status = Get-MpComputerStatus -ErrorAction Stop + +$catalog = @( + Get-MpThreat -ErrorAction Stop | ForEach-Object { + $severityId = $null + $severitySource = $null + $severityProperty = $_.PSObject.Properties['ThreatSeverityID'] + if ($null -ne $severityProperty -and $null -ne $severityProperty.Value) { + $severityId = Get-SafeInt $severityProperty.Value + $severitySource = 'ThreatSeverityID' + } else { + $severityProperty = $_.PSObject.Properties['SeverityID'] + if ($null -ne $severityProperty -and $null -ne $severityProperty.Value) { + $rawSeverity = Get-SafeInt $severityProperty.Value + # MSFT_MpThreatCatalog uses 3=High and 4=Severe. Normalize to + # MPTHREAT_SEVERITY's documented 4=High and 5=Severe values. + $catalogSeverityMap = @{ 0 = 0; 1 = 1; 2 = 2; 3 = 4; 4 = 5 } + if ($null -ne $rawSeverity -and $catalogSeverityMap.ContainsKey($rawSeverity)) { + $severityId = $catalogSeverityMap[$rawSeverity] + } + $severitySource = 'SeverityID' + } + } + + $active = $null + $activeProperty = $_.PSObject.Properties['IsActive'] + if ($null -ne $activeProperty -and $null -ne $activeProperty.Value) { + $active = [bool]$activeProperty.Value + } + + [PSCustomObject]@{ + threat_id = if ($null -ne $_.ThreatID) { [string]$_.ThreatID } else { $null } + threat_name = if ($_.ThreatName) { ([string]$_.ThreatName).Substring(0, [Math]::Min(256, ([string]$_.ThreatName).Length)) } else { $null } + threat_severity_id = $severityId + severity_source = $severitySource + is_active = $active + did_threat_execute = if ($null -ne $_.DidThreatExecute) { [bool]$_.DidThreatExecute } else { $null } + } + } +) + +$activeThreatIds = @{} +foreach ($item in $catalog) { + if ($item.threat_id -and $item.is_active -eq $true) { + $activeThreatIds[[string]$item.threat_id] = $true + } +} + +$allDetections = @(Get-MpThreatDetection -ErrorAction Stop) +$selectedDetections = @( + $allDetections | Where-Object { + $initial = $_.InitialDetectionTime + $changed = $_.LastThreatStatusChangeTime + $withinHistory = $false + if ($initial -and $initial.ToUniversalTime() -ge $historyCutoff) { $withinHistory = $true } + if ($changed -and $changed.ToUniversalTime() -ge $historyCutoff) { $withinHistory = $true } + $id = if ($null -ne $_.ThreatID) { [string]$_.ThreatID } else { '' } + $statusId = Get-SafeInt $_.ThreatStatusID + $isKnownResolved = $statusId -in @(2, 4, 6) + $knownActive = $activeThreatIds.ContainsKey($id) + $withinHistory -or $knownActive -or -not $isKnownResolved + } | ForEach-Object { + $resourceSummary = Get-ResourceSummary $_.Resources + [PSCustomObject]@{ + detection_id = if ($_.DetectionID) { [string]$_.DetectionID } else { $null } + threat_id = if ($null -ne $_.ThreatID) { [string]$_.ThreatID } else { $null } + status_id = Get-SafeInt $_.ThreatStatusID + detected_at = Format-Utc $_.InitialDetectionTime + changed_at = Format-Utc $_.LastThreatStatusChangeTime + remediation_at = Format-Utc $_.RemediationTime + error_code = Get-SafeInt $_.ThreatStatusErrorCode + action_success = if ($null -ne $_.ActionSuccess) { [bool]$_.ActionSuccess } else { $null } + execution_status_id = Get-SafeInt $_.CurrentThreatExecutionStatusID + resource_count = $resourceSummary.resource_count + resource_kinds = @($resourceSummary.resource_kinds) + } + } +) + +$result = [PSCustomObject]@{ + collection_complete = $true + history_detection_count = [int]$allDetections.Count + returned_detection_count = [int]$selectedDetections.Count + defender = [PSCustomObject]@{ + running_mode = $status.AMRunningMode + antivirus_enabled = $status.AntivirusEnabled + realtime_enabled = $status.RealTimeProtectionEnabled + behavior_monitor_enabled = $status.BehaviorMonitorEnabled + ioav_enabled = $status.IoavProtectionEnabled + signature_age = Get-SafeInt $status.AntivirusSignatureAge + signature_version = $status.AntivirusSignatureVersion + signature_last_updated = Format-Utc $status.AntivirusSignatureLastUpdated + engine_version = $status.AMEngineVersion + service_enabled = $status.AMServiceEnabled + } + threats = @($catalog) + detections = @($selectedDetections) +} + +ConvertTo-Json -InputObject $result -Depth 8 -Compress +''' + + +def _copy_rules(rules=None): + result = dict(DEFAULT_RULES) + for key in ("status_modifiers", "protection_modifiers", "correlations"): + result[key] = dict(DEFAULT_RULES[key]) + if rules: + for key, value in rules.items(): + if key in ("status_modifiers", "protection_modifiers", "correlations"): + result[key].update(value) + else: + result[key] = value + return result + + +def _as_list(value): + if value is None: + return [] + if isinstance(value, (list, tuple)): + return list(value) + return [value] + + +def _first(mapping, *keys): + if not isinstance(mapping, dict): + return None + for key in keys: + if key in mapping: + return mapping[key] + return None + + +def _to_int(value, default=None): + if value is None or isinstance(value, bool): + return default + try: + return int(value) + except (TypeError, ValueError, OverflowError): + return default + + +def _to_bool(value): + if isinstance(value, bool): + return value + if value in (0, 1): + return bool(value) + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in ("true", "yes", "1"): + return True + if lowered in ("false", "no", "0"): + return False + return None + + +def _threat_id(value): + if value is None: + return None + text = str(value).strip() + return text or None + + +def _parse_datetime(value): + if isinstance(value, datetime): + parsed = value + elif isinstance(value, str) and value.strip(): + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + else: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _iso(value): + parsed = _parse_datetime(value) + return parsed.isoformat().replace("+00:00", "Z") if parsed else None + + +def _catalog_severity(item): + severity_id = _to_int(_first(item, "threat_severity_id", "ThreatSeverityID")) + source = _first(item, "severity_source", "SeveritySource") + if severity_id is None: + severity_id = _to_int(_first(item, "severity_id", "SeverityID")) + source = source or "SeverityID" + # WMI catalog values differ from MPTHREAT_SEVERITY for High and Severe. + if source == "SeverityID": + severity_id = {0: 0, 1: 1, 2: 2, 3: 4, 4: 5}.get(severity_id) + return severity_id if severity_id in SEVERITY_BASE_LEVELS else None + + +def _resource_summary(detection): + kinds = _as_list(_first(detection, "resource_kinds", "ResourcesKinds")) + count = _to_int(_first(detection, "resource_count", "ResourceCount")) + if not kinds: + raw_resources = _as_list(_first(detection, "resources", "Resources")) + kinds = [] + for resource in raw_resources: + prefix = str(resource).split(":", 1)[0].lower() + kinds.append(prefix if prefix in ("file", "process", "registry", "service", "url") else "other") + if count is None: + count = len(raw_resources) + safe_kinds = sorted(set( + kind for kind in (str(value).lower() for value in kinds) + if kind in ("file", "process", "registry", "service", "url", "other") + )) + return safe_kinds, max(0, count or 0) + + +def _detection_identity(detection, index): + threat_id = _threat_id(_first(detection, "threat_id", "ThreatID")) or "unknown" + detection_id = _first(detection, "detection_id", "DetectionID") + if detection_id is not None and str(detection_id).strip(): + detection_id = str(detection_id).strip() + return "%s:%s" % (threat_id, detection_id), detection_id, True + + # DetectionID is the documented unique detection key. If unavailable, use + # all stable fields we have and surface the weaker identity in the summary. + initial = _iso(_first(detection, "detected_at", "initial_detection_time", "InitialDetectionTime")) or "?" + changed = _iso(_first(detection, "changed_at", "last_status_change_at", "LastThreatStatusChangeTime")) or "?" + status_id = _to_int(_first(detection, "status_id", "threat_status_id", "ThreatStatusID")) + resources = _as_list(_first(detection, "resources", "Resources")) + resource_signature = "|".join(sorted(str(value) for value in resources)) + identity = "%s:%s:%s:%s:%s" % (threat_id, initial, changed, status_id, resource_signature) + return identity, None, False + + +def _is_active(status_id, is_active, action_success, did_threat_execute=False, + execution_status_id=None): + if action_success is False: + return True + if did_threat_execute is True or execution_status_id in (2, 3): + return True + if status_id in RESOLVED_STATUS_IDS: + return is_active is True + # In particular, Quarantined and Allowed remain unresolved for scoring. + return True + + +def classify_detection(detection, repeated=False, realtime_expected=True, + protection_context=None, rules=None): + """Return a deterministic 0..16 risk level for one Defender detection.""" + config = _copy_rules(rules) + severity_id = _to_int(_first(detection, "threat_severity_id", "ThreatSeverityID")) + if severity_id is None: + severity_id = _to_int(_first(detection, "severity_id", "SeverityID")) + if _first(detection, "severity_source", "SeveritySource") == "SeverityID": + severity_id = {0: 0, 1: 1, 2: 2, 3: 4, 4: 5}.get(severity_id) + base = config.get("base_levels", SEVERITY_BASE_LEVELS).get(severity_id, SEVERITY_BASE_LEVELS[0]) + status_id = _to_int(_first(detection, "status_id", "threat_status_id", "ThreatStatusID")) + is_active = _to_bool(_first(detection, "is_active", "IsActive")) + action_success = _to_bool(_first(detection, "action_success", "ActionSuccess")) + did_threat_execute = _to_bool(_first(detection, "did_threat_execute", "DidThreatExecute")) + execution_status_id = _to_int(_first(detection, "execution_status_id", "CurrentThreatExecutionStatusID")) + active = _is_active(status_id, is_active, action_success, did_threat_execute, execution_status_id) + + level = base + config["status_modifiers"].get(status_id, 0) + if active: + level += config["active_modifier"] + if repeated: + level += config["repeated_modifier"] + if status_id not in FAILED_STATUS_IDS and action_success is False: + level += config["failed_remediation_modifier"] + if protection_context is None: + protection_context = [] + if realtime_expected and _to_bool(_first(detection, "realtime_enabled")) is False: + protection_context.append("realtime_disabled") + for modifier in protection_context: + level += config["protection_modifiers"].get(modifier, 0) + return max(0, min(16, int(level))) + + +def level_category(level): + level = max(0, min(16, _to_int(level, 0))) + if level == 0: + return "Aucun événement" + if level <= 3: + return "Information" + if level <= 6: + return "Faible" + if level <= 9: + return "Modéré" + if level <= 12: + return "Élevé" + if level <= 14: + return "Critique" + return "Urgence" + + +def _protection_state(defender, rules): + mode = defender.get("running_mode") + mode_text = str(mode or "").strip() + expected_modes = {str(value).lower() for value in rules["expected_active_modes"]} + passive_modes = {str(value).lower() for value in rules["passive_modes"]} + is_expected_active = mode_text.lower() in expected_modes + is_passive = mode_text.lower() in passive_modes + + issues = [] + modifiers = [] + level = 0 + if is_expected_active: + checks = ( + ("service_enabled", "Defender service disabled", "service_disabled"), + ("antivirus_enabled", "Antivirus disabled", "antivirus_disabled"), + ("realtime_enabled", "Realtime protection disabled", "realtime_disabled"), + ("behavior_monitor_enabled", "Behavior monitoring disabled", "behavior_monitor_disabled"), + ("ioav_enabled", "IOAV protection disabled", "ioav_disabled"), + ) + for key, label, modifier in checks: + value = _to_bool(defender.get(key)) + if value is False: + issues.append(label) + modifiers.append(modifier) + level += rules["protection_modifiers"].get(modifier, 0) + elif value is None: + issues.append("Unknown %s" % key) + + age = _to_int(defender.get("signature_age")) + max_age = rules["signature_max_age_days"] + if age is None: + issues.append("Signature age unknown") + modifiers.append("signature_age_unknown") + level += rules["protection_modifiers"].get("signature_age_unknown", 0) + elif age > max_age: + issues.append("Signatures outdated: %s days" % age) + modifiers.append("signatures_stale") + level += rules["protection_modifiers"].get("signatures_stale", 0) + elif not is_passive and not mode_text: + issues.append("Defender running mode unknown") + + return max(0, min(16, level)), issues, is_expected_active, is_passive, modifiers + + +def _deduplicate_detections(detections): + deduped = {} + identity_incomplete = 0 + for index, item in enumerate(_as_list(detections)): + if not isinstance(item, dict): + identity_incomplete += 1 + continue + key, detection_id, has_detection_id = _detection_identity(item, index) + if not has_detection_id: + identity_incomplete += 1 + previous = deduped.get(key) + candidate_changed = _parse_datetime(_first(item, "changed_at", "last_status_change_at", "LastThreatStatusChangeTime")) + previous_changed = _parse_datetime(_first(previous or {}, "changed_at", "last_status_change_at", "LastThreatStatusChangeTime")) + if previous is None or (candidate_changed and (not previous_changed or candidate_changed >= previous_changed)): + deduped[key] = dict(item) + deduped[key]["_detection_key"] = key + deduped[key]["_detection_id"] = detection_id + deduped[key]["_identity_complete"] = has_detection_id + return list(deduped.values()), identity_incomplete + + +def _normalize_detection(item, catalog, now, window, history_cutoff, rules, repeated=False, + realtime_expected=True, realtime_enabled=None, + protection_context=None): + threat_id = _threat_id(_first(item, "threat_id", "ThreatID")) + metadata = catalog.get(threat_id, {}) + severity_id = _catalog_severity(metadata) + status_id = _to_int(_first(item, "status_id", "threat_status_id", "ThreatStatusID")) + is_active_value = _to_bool(_first(metadata, "is_active", "IsActive")) + action_success = _to_bool(_first(item, "action_success", "ActionSuccess")) + did_threat_execute = _to_bool(_first(metadata, "did_threat_execute", "DidThreatExecute")) + execution_status_id = _to_int(_first(item, "execution_status_id", "CurrentThreatExecutionStatusID")) + active = _is_active( + status_id, is_active_value, action_success, did_threat_execute, execution_status_id, + ) + detected = _parse_datetime(_first(item, "detected_at", "initial_detection_time", "InitialDetectionTime")) + changed = _parse_datetime(_first(item, "changed_at", "last_status_change_at", "LastThreatStatusChangeTime")) + remediation = _parse_datetime(_first(item, "remediation_at", "RemediationTime")) + recent = bool((detected and detected >= window) or (changed and changed >= window)) + retained = bool(recent or active or ((detected and detected >= history_cutoff) or (changed and changed >= history_cutoff))) + resource_kinds, resource_count = _resource_summary(item) + enriched = dict(item) + enriched.update({ + "threat_severity_id": severity_id, + "status_id": status_id, + "is_active": active, + "action_success": action_success, + "did_threat_execute": did_threat_execute, + "execution_status_id": execution_status_id, + "realtime_enabled": realtime_enabled, + }) + level = classify_detection( + enriched, repeated=repeated, realtime_expected=realtime_expected, + protection_context=protection_context, rules=rules, + ) + return { + "detection_id": item.get("_detection_id") or _first(item, "detection_id", "DetectionID"), + "threat_id": threat_id, + "threat_name": _first(metadata, "threat_name", "name", "ThreatName") or "Unknown threat", + "did_threat_execute": _to_bool(_first(metadata, "did_threat_execute", "DidThreatExecute")), + "severity_id": severity_id, + "severity": SEVERITY_NAMES.get(severity_id, "Unknown"), + "status_id": status_id, + "status": THREAT_STATUSES.get(status_id, "Unknown"), + "detected_at": _iso(_first(item, "detected_at", "initial_detection_time", "InitialDetectionTime")), + "changed_at": _iso(_first(item, "changed_at", "last_status_change_at", "LastThreatStatusChangeTime")), + "remediation_at": _iso(_first(item, "remediation_at", "RemediationTime")), + "error_code": _to_int(_first(item, "error_code", "threat_status_error_code", "ThreatStatusErrorCode")), + "action_success": action_success, + "execution_status_id": _to_int(_first(item, "execution_status_id", "CurrentThreatExecutionStatusID")), + "resource_kinds": resource_kinds, + "resource_count": resource_count, + "active": active, + "active_reason": ( + "remediation_failed" if action_success is False or status_id in FAILED_STATUS_IDS + else "execution_observed" if did_threat_execute is True + else "execution_allowed_or_running" if execution_status_id in (2, 3) + else "explicitly_allowed" if status_id == 5 + else "quarantined_unresolved" if status_id == 3 + else "defender_active" if is_active_value is True + else "unresolved_status" if status_id not in RESOLVED_STATUS_IDS + else None + ), + "in_window_6h": recent, + "event_level": level, + "event_category": level_category(level), + "identity_complete": bool(item.get("_identity_complete", False)), + "_retained": retained, + } + + +def _append_correlation(result, enabled, key, evidence): + if enabled.get(key) and evidence: + result.append({"rule": key, **evidence}) + + +def _build_correlations(detections, rules, realtime_disabled): + enabled = rules["correlations"] + recent = [item for item in detections if item["in_window_6h"]] + by_threat = {} + failures = Counter() + for item in recent: + if item["threat_id"]: + by_threat.setdefault(item["threat_id"], []).append(item) + if item["status_id"] in FAILED_STATUS_IDS or item["action_success"] is False: + failures[item["threat_id"]] += 1 + + repeated = { + threat_id: items for threat_id, items in by_threat.items() + if len(items) >= rules["repeated_threshold"] + } + correlations = [] + repeated_ids = sorted(repeated) + _append_correlation(correlations, enabled, "same_threat_repeated", { + "count": sum(len(items) for items in repeated.values()), + "threat_ids": repeated_ids, + } if repeated else None) + + unique_recent_threats = sorted(set(item["threat_id"] for item in recent if item["threat_id"])) + _append_correlation(correlations, enabled, "multiple_threats", { + "count": len(unique_recent_threats), + "threat_ids": unique_recent_threats, + } if len(unique_recent_threats) >= rules["multiple_threats_threshold"] else None) + + high_or_active = [item for item in detections if item["in_window_6h"] or item["active"]] + severe_ids = sorted(set( + item["threat_id"] for item in high_or_active + if item["threat_id"] and item["severity_id"] in (4, 5) + )) + _append_correlation(correlations, enabled, "severe_with_protection_disabled", { + "threat_ids": severe_ids, + "realtime_disabled": True, + } if severe_ids and realtime_disabled else None) + + repeated_failure_ids = sorted( + threat_id for threat_id, count in failures.items() + if count >= rules["remediation_failure_threshold"] + ) + _append_correlation(correlations, enabled, "repeated_remediation_failures", { + "threat_ids": repeated_failure_ids, + "count": sum(failures[threat_id] for threat_id in repeated_failure_ids), + } if repeated_failure_ids else None) + + allowed = [item for item in recent if item["status_id"] == 5] + _append_correlation(correlations, enabled, "explicitly_allowed", { + "threat_ids": sorted(set(item["threat_id"] for item in allowed if item["threat_id"])), + "count": len(allowed), + } if allowed else None) + return correlations + + +def _wapt_status(current_level, protection_issues, expected_active, is_passive, collection_complete, identity_incomplete, unknown_severity): + if not collection_complete: + return "ERROR" + if expected_active and any( + issue in protection_issues + for issue in ("Defender service disabled", "Antivirus disabled", "Realtime protection disabled") + ): + return "ERROR" + if current_level >= 10: + return "ERROR" + if current_level >= 7: + return "WARNING" + if protection_issues or identity_incomplete or unknown_severity: + return "WARNING" + if not expected_active and not is_passive: + return "WARNING" + return "OK" + + +def build_audit_snapshot(data, now=None, rules=None): + """Normalize Defender data, classify events, and build the v2 WAPT payload.""" + config = _copy_rules(rules) + now = _parse_datetime(now) or datetime.now(timezone.utc) + window = now - timedelta(hours=config["window_hours"]) + history_cutoff = now - timedelta(days=HISTORY_DAYS) + if not isinstance(data, dict): + data = {} + + defender_raw = data.get("defender") if isinstance(data.get("defender"), dict) else {} + defender = { + "running_mode": defender_raw.get("running_mode"), + "antivirus_enabled": _to_bool(defender_raw.get("antivirus_enabled")), + "realtime_enabled": _to_bool(defender_raw.get("realtime_enabled")), + "behavior_monitor_enabled": _to_bool(defender_raw.get("behavior_monitor_enabled")), + "ioav_enabled": _to_bool(defender_raw.get("ioav_enabled")), + "signature_age": _to_int(defender_raw.get("signature_age")), + "signature_version": defender_raw.get("signature_version"), + "signature_last_updated": _iso(defender_raw.get("signature_last_updated")), + "engine_version": defender_raw.get("engine_version"), + "service_enabled": _to_bool(defender_raw.get("service_enabled")), + } + raw_detections = data.get("detections") + raw_threats = data.get("threats") + supported_collection_types = (list, tuple, dict) + arrays_valid = ( + "detections" in data and "threats" in data + and (raw_detections is None or isinstance(raw_detections, supported_collection_types)) + and (raw_threats is None or isinstance(raw_threats, supported_collection_types)) + ) + collection_complete = data.get("collection_complete") is not False and arrays_valid + if not defender or not defender.get("running_mode"): + collection_complete = False + if defender["antivirus_enabled"] is None or defender["realtime_enabled"] is None: + collection_complete = False + + catalog = {} + for threat in _as_list(raw_threats): + threat_id = _threat_id(_first(threat, "threat_id", "ThreatID")) + if threat_id: + catalog[threat_id] = threat + + deduped, identity_incomplete = _deduplicate_detections(raw_detections) + preliminary = [] + for item in deduped: + threat_id = _threat_id(_first(item, "threat_id", "ThreatID")) + preliminary.append({ + "threat_id": threat_id, + "in_window_6h": bool( + (_parse_datetime(_first(item, "detected_at", "initial_detection_time", "InitialDetectionTime")) or datetime.min.replace(tzinfo=timezone.utc)) >= window + or (_parse_datetime(_first(item, "changed_at", "last_status_change_at", "LastThreatStatusChangeTime")) or datetime.min.replace(tzinfo=timezone.utc)) >= window + ), + }) + recent_counts = Counter(item["threat_id"] for item in preliminary if item["in_window_6h"] and item["threat_id"]) + + protection_level, protection_issues, expected_active, is_passive, protection_context = _protection_state(defender, config) + realtime_disabled = expected_active and defender["realtime_enabled"] is False + detections = [] + for item in deduped: + threat_id = _threat_id(_first(item, "threat_id", "ThreatID")) + repeated = bool(threat_id and recent_counts.get(threat_id, 0) >= config["repeated_threshold"]) + normalized = _normalize_detection( + item, catalog, now, window, history_cutoff, config, + repeated=repeated, realtime_expected=expected_active, + realtime_enabled=defender["realtime_enabled"], + protection_context=protection_context, + ) + if normalized["in_window_6h"] and realtime_disabled: + # classify_detection already applies this configured modifier. + pass + detections.append(normalized) + + detections = [item for item in detections if item["_retained"]] + for item in detections: + item.pop("_retained", None) + + recent = [item for item in detections if item["in_window_6h"]] + active = [item for item in detections if item["active"]] + max_event = max((item["event_level"] for item in recent), default=0) + max_active = max((item["event_level"] for item in active), default=0) + current_risk = max(max_active, protection_level) + if not collection_complete: + current_risk = 16 + unknown_severity_count = sum(1 for item in detections if item["severity_id"] is None) + correlations = _build_correlations(detections, config, realtime_disabled) + status = _wapt_status( + current_risk, protection_issues, expected_active, is_passive, + collection_complete, identity_incomplete, unknown_severity_count, + ) + + # Keep the payload stable between audits; WAPT stores a new historical value + # only when an actual status, detection, or correlation changes. + return { + "schema_version": 2, + "collection_complete": bool(collection_complete), + "history_detection_count": _to_int(data.get("history_detection_count")), + "returned_detection_count": _to_int(data.get("returned_detection_count"), len(_as_list(raw_detections))), + "defender": defender, + "summary": { + "max_event_level_6h": max_event, + "max_active_level": max_active, + "current_risk_level": current_risk, + "current_risk_category": level_category(current_risk), + "audit_status": status, + "collection_complete": bool(collection_complete), + "detection_count_6h": len(recent), + "active_count": len(active), + "resolved_count_6h": sum(1 for item in recent if not item["active"]), + "identity_incomplete_count": identity_incomplete, + "unknown_severity_count": unknown_severity_count, + "protection_issues": protection_issues, + "correlation_count": len(correlations), + }, + "detections": detections, + "active_threats": active, + "correlations": correlations, + } + + +def build_collection_error_snapshot(code): + safe_code = code if code in ("powershell_timeout", "powershell_failed", "powershell_unavailable", "empty_output", "invalid_json", "collection_failed") else "collection_failed" + return { + "schema_version": 2, + "collection_complete": False, + "history_detection_count": None, + "returned_detection_count": 0, + "defender": { + "running_mode": None, + "antivirus_enabled": None, + "realtime_enabled": None, + "behavior_monitor_enabled": None, + "ioav_enabled": None, + "signature_age": None, + "signature_version": None, + "signature_last_updated": None, + "engine_version": None, + "service_enabled": None, + }, + "summary": { + "max_event_level_6h": 0, + "max_active_level": 0, + "current_risk_level": 16, + "current_risk_category": level_category(16), + "audit_status": "ERROR", + "collection_complete": False, + "detection_count_6h": 0, + "active_count": 0, + "resolved_count_6h": 0, + "identity_incomplete_count": 0, + "unknown_severity_count": 0, + "protection_issues": [safe_code], + "correlation_count": 0, + }, + "detections": [], + "active_threats": [], + "correlations": [], + } + + +def powershell_json(script=POWERSHELL_SCRIPT, timeout=POWERSHELL_TIMEOUT_SECONDS): powershell = os.path.join( os.environ.get("SystemRoot", r"C:\Windows"), "System32", "WindowsPowerShell", "v1.0", - "powershell.exe" - ) - - result = subprocess.run( - [ - powershell, - "-NoProfile", - "-NonInteractive", - "-ExecutionPolicy", "Bypass", - "-Command", script, - ], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - timeout=90, + "powershell.exe", ) + try: + result = subprocess.run( + [powershell, "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout, + check=False, + ) + except subprocess.TimeoutExpired: + raise DefenderCollectionError("powershell_timeout") + except OSError: + raise DefenderCollectionError("powershell_unavailable") if result.returncode != 0: - raise RuntimeError( - result.stderr.decode("utf-8", errors="replace") - ) - - output = result.stdout.decode( - "utf-8-sig", errors="replace" - ).strip() - + raise DefenderCollectionError("powershell_failed") + raw_output = result.stdout or b"" + if isinstance(raw_output, bytes): + output = raw_output.decode("utf-8-sig", errors="replace").strip() + else: + output = str(raw_output).lstrip("\ufeff").strip() if not output: - raise RuntimeError("Empty PowerShell response") - - return json.loads(output) + raise DefenderCollectionError("empty_output") + try: + decoded = json.loads(output) + except (TypeError, ValueError): + raise DefenderCollectionError("invalid_json") + if not isinstance(decoded, dict): + raise DefenderCollectionError("invalid_json") + return decoded def collect_defender(): - script = r""" -$ErrorActionPreference = "Stop" -$ProgressPreference = "SilentlyContinue" - -$status = Get-MpComputerStatus -$cutoff = (Get-Date).AddDays(-30) - -$threats = @( - Get-MpThreatDetection | - Where-Object { - $_.InitialDetectionTime -ge $cutoff - } | - Sort-Object InitialDetectionTime -Descending | - Select-Object -First 100 | - ForEach-Object { - [PSCustomObject]@{ - threat_id = $_.ThreatID - status_id = [int]$_.ThreatStatusID - detected_at = if ($_.InitialDetectionTime) { - $_.InitialDetectionTime.ToString("o") - } else { $null } - changed_at = if ($_.LastThreatStatusChangeTime) { - $_.LastThreatStatusChangeTime.ToString("o") - } else { $null } - error_code = $_.ThreatStatusErrorCode - resources = @($_.Resources) - } - } -) - -$threatNames = @{} -Get-MpThreat | ForEach-Object { - $threatNames[[string]$_.ThreatID] = $_.ThreatName -} - -foreach ($threat in $threats) { - $id = [string]$threat.threat_id - $name = $threatNames[$id] - - $threat | Add-Member -NotePropertyName "name" ` - -NotePropertyValue $name -} - -$result = [PSCustomObject]@{ - collected_at = (Get-Date).ToUniversalTime().ToString("o") - computer_name = $env:COMPUTERNAME - running_mode = $status.AMRunningMode - antivirus_enabled = $status.AntivirusEnabled - realtime_enabled = $status.RealTimeProtectionEnabled - behavior_monitor_enabled = $status.BehaviorMonitorEnabled - ioav_enabled = $status.IoavProtectionEnabled - signature_age = $status.AntivirusSignatureAge - signature_version = $status.AntivirusSignatureVersion - engine_version = $status.AMEngineVersion - threats = @($threats) -} - -ConvertTo-Json -InputObject $result -Depth 8 -Compress -""" - return powershell_json(script) + """Read Defender state only; no network or remediation cmdlets are used.""" + return powershell_json() -def normalize_threats(data): - result = [] - - for threat in data.get("threats", []): - status_id = int(threat.get("status_id", 0)) - - if status_id in ACTIVE_STATUSES: - severity = "ERROR" - elif status_id in FAILED_STATUSES: - severity = "ERROR" - elif status_id in RESOLVED_STATUSES: - severity = "RESOLVED" - else: - severity = "WARNING" - - result.append({ - "id": threat.get("threat_id"), - "name": threat.get("name") or "Unknown threat", - "status_id": status_id, - "status": THREAT_STATUSES.get( - status_id, "Unknown" - ), - "severity": severity, - "detected_at": threat.get("detected_at"), - "changed_at": threat.get("changed_at"), - "resources": threat.get("resources") or [], - "error_code": threat.get("error_code"), - }) - - return result - - -def save_audit(key, value): - WAPT.write_audit_data_if_changed( +def save_audit(key, value, rules=None): + config = _copy_rules(rules) + return WAPT.write_audit_data_if_changed( AUDIT_CATEGORY, key, value, - keep_days=90 + max_count=config["audit_history_max_count"], + keep_days=config["audit_history_days"], ) +def _save_snapshot(snapshot): + save_audit("snapshot", snapshot) + # Keep established Reporting keys available while adding the v2 aggregate. + save_audit("summary", snapshot["summary"]) + save_audit("detections", snapshot["detections"]) + save_audit("active_threats", snapshot["active_threats"]) + save_audit("correlations", snapshot["correlations"]) + + def audit(): try: data = collect_defender() - threats = normalize_threats(data) - - active = [ - t for t in threats - if t["severity"] == "ERROR" - ] - uncertain = [ - t for t in threats - if t["severity"] == "WARNING" - ] - resolved = [ - t for t in threats - if t["severity"] == "RESOLVED" - ] - - status = "OK" - problems = [] - - mode = data.get("running_mode") - - if mode != "Normal": - problems.append( - "Defender mode: %s" % mode - ) - status = "WARNING" - - if mode == "Normal": - if not data.get("antivirus_enabled"): - problems.append("Antivirus disabled") - status = "ERROR" - - if not data.get("realtime_enabled"): - problems.append("Realtime protection disabled") - status = "ERROR" - - age = data.get("signature_age") - - if age is None: - problems.append("Unknown signature age") - if status == "OK": - status = "WARNING" - elif int(age) > SIGNATURE_MAX_AGE: - problems.append( - "Signatures outdated: %s days" % age - ) - if status == "OK": - status = "WARNING" - - if active: - problems.append( - "%d unresolved or failed detections" % - len(active) - ) - status = "ERROR" - - if uncertain and status == "OK": - problems.append( - "%d detections require review" % - len(uncertain) - ) - status = "WARNING" - - summary = { - "status": status, - "computer_name": data.get("computer_name"), - "running_mode": mode, - "antivirus_enabled": data.get( - "antivirus_enabled" - ), - "realtime_enabled": data.get( - "realtime_enabled" - ), - "signature_age": data.get("signature_age"), - "signature_version": data.get( - "signature_version" - ), - "engine_version": data.get("engine_version"), - "detection_count": len(threats), - "active_count": len(active), - "review_count": len(uncertain), - "resolved_count": len(resolved), - "problems": problems, - } - - save_audit("summary", summary) - save_audit("detections", threats) - save_audit("active_threats", active) - - print(json.dumps( - summary, - indent=2, - ensure_ascii=False - )) - - return status - + snapshot = build_audit_snapshot(data) except Exception as exc: - error = { - "status": "ERROR", - "error": str(exc) - } + code = getattr(exc, "code", "collection_failed") + snapshot = build_collection_error_snapshot(code) - print("Defender audit error: %s" % exc) - - save_audit("audit_error", error) + try: + _save_snapshot(snapshot) + except Exception: + # Keep the status fail-closed if WAPT audit persistence itself fails. + print("Defender audit data could not be stored by WAPT") return "ERROR" + print(json.dumps(snapshot["summary"], indent=2, ensure_ascii=False, sort_keys=True)) + return snapshot["summary"]["audit_status"] diff --git a/tests/test_setup.py b/tests/test_setup.py new file mode 100644 index 0000000..974c001 --- /dev/null +++ b/tests/test_setup.py @@ -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()