734 lines
17 KiB
Python
734 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from setuphelpers import *
|
|
|
|
import os
|
|
import re
|
|
import tempfile
|
|
import xml.etree.ElementTree as ET
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Identifiants de l'audit WAPT
|
|
# ---------------------------------------------------------------------------
|
|
|
|
AUDIT_SECTION = "audit-gpresult"
|
|
AUDIT_NAME = "user-gpo"
|
|
|
|
|
|
def install():
|
|
"""
|
|
Ce paquet est uniquement destiné à l'audit.
|
|
|
|
Aucune modification persistante n'est effectuée sur le poste.
|
|
"""
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fonctions utilitaires XML
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _clean_tag(tag):
|
|
"""
|
|
Supprime le namespace XML d'un tag ElementTree.
|
|
"""
|
|
return tag.split("}")[-1]
|
|
|
|
|
|
def _text(element, tag, default=""):
|
|
"""
|
|
Retourne récursivement la première valeur XML correspondant au tag
|
|
demandé, en ignorant les namespaces.
|
|
"""
|
|
|
|
if element is None:
|
|
return default
|
|
|
|
for child in element.iter():
|
|
|
|
if _clean_tag(child.tag).lower() == tag.lower():
|
|
|
|
if child.text:
|
|
return child.text.strip()
|
|
|
|
return default
|
|
|
|
|
|
def _to_bool(value):
|
|
"""
|
|
Convertit une valeur booléenne textuelle issue de GPResult
|
|
en booléen Python.
|
|
"""
|
|
return str(value).strip().lower() in (
|
|
"true",
|
|
"1",
|
|
"yes",
|
|
"oui"
|
|
)
|
|
|
|
|
|
def _split_account(account):
|
|
"""
|
|
Sépare un compte Windows entre son préfixe et son nom de compte.
|
|
|
|
Exemples :
|
|
FORMATION\\sebastien
|
|
-> ("FORMATION", "sebastien")
|
|
|
|
WSMANAGE30\\formation
|
|
-> ("WSMANAGE30", "formation")
|
|
"""
|
|
|
|
if not account:
|
|
return "", ""
|
|
|
|
if "\\" in account:
|
|
return account.split("\\", 1)
|
|
|
|
return "", account
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Détection de l'utilisateur interactif
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _get_logged_on_user():
|
|
"""
|
|
Retourne le nom de l'utilisateur associé à la session interactive active.
|
|
|
|
Les audits WAPT s'exécutent dans le contexte du service WAPT.
|
|
Il n'est donc pas possible de se fier simplement à l'identité
|
|
du processus courant.
|
|
|
|
`quser` est utilisé pour inspecter les sessions Windows et récupérer
|
|
l'utilisateur de la session interactive active.
|
|
|
|
Exemples retournés par cette fonction :
|
|
formation
|
|
sebastien
|
|
|
|
L'identité complète du compte est ensuite résolue par GPResult :
|
|
WSMANAGE30\\formation
|
|
FORMATION\\sebastien
|
|
|
|
Returns:
|
|
str ou None
|
|
"""
|
|
|
|
try:
|
|
|
|
output = run(
|
|
r'quser.exe',
|
|
timeout=30,
|
|
return_stderr=False
|
|
)
|
|
|
|
if not output:
|
|
return None
|
|
|
|
lines = output.splitlines()
|
|
|
|
# La première ligne correspond aux en-têtes de colonnes.
|
|
for line in lines[1:]:
|
|
|
|
line = line.strip()
|
|
|
|
if not line:
|
|
continue
|
|
|
|
# quser peut préfixer la session courante avec ">".
|
|
line = line.lstrip(">").strip()
|
|
|
|
# Prend en charge les sorties Windows en anglais et en français.
|
|
if (
|
|
" Active " not in " %s " % line
|
|
and
|
|
" Actif " not in " %s " % line
|
|
):
|
|
continue
|
|
|
|
parts = re.split(
|
|
r"\s+",
|
|
line
|
|
)
|
|
|
|
if not parts:
|
|
continue
|
|
|
|
# USERNAME correspond à la première colonne.
|
|
username = parts[0].strip()
|
|
|
|
if username:
|
|
return username
|
|
|
|
return None
|
|
|
|
except Exception as e:
|
|
|
|
print(
|
|
"Unable to detect interactive user: %s"
|
|
% e
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Recherche des données GPResult dans le XML
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _find_user_results(root):
|
|
"""
|
|
Recherche la section UserResults dans le document XML GPResult.
|
|
"""
|
|
|
|
for element in root.iter():
|
|
|
|
if _clean_tag(element.tag) == "UserResults":
|
|
return element
|
|
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Analyse des GPO
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _get_gpos(scope_results):
|
|
"""
|
|
Extrait les noms des GPO appliquées et refusées depuis le RSoP USER.
|
|
|
|
Les structures GPO internes sans nom sont ignorées
|
|
et les doublons sont supprimés.
|
|
"""
|
|
|
|
applied = []
|
|
denied = []
|
|
|
|
seen = set()
|
|
|
|
if scope_results is None:
|
|
return applied, denied
|
|
|
|
for element in scope_results.iter():
|
|
|
|
if _clean_tag(element.tag) != "GPO":
|
|
continue
|
|
|
|
name = _text(
|
|
element,
|
|
"Name"
|
|
)
|
|
|
|
if not name:
|
|
continue
|
|
|
|
enabled = _to_bool(
|
|
_text(element, "Enabled")
|
|
)
|
|
|
|
valid = _to_bool(
|
|
_text(element, "IsValid")
|
|
)
|
|
|
|
filter_allowed = _to_bool(
|
|
_text(element, "FilterAllowed")
|
|
)
|
|
|
|
access_denied = _to_bool(
|
|
_text(element, "AccessDenied")
|
|
)
|
|
|
|
version_directory = _text(
|
|
element,
|
|
"VersionDirectory"
|
|
)
|
|
|
|
version_sysvol = _text(
|
|
element,
|
|
"VersionSysvol"
|
|
)
|
|
|
|
identity = (
|
|
name,
|
|
version_directory,
|
|
version_sysvol
|
|
)
|
|
|
|
if identity in seen:
|
|
continue
|
|
|
|
seen.add(identity)
|
|
|
|
if (
|
|
enabled
|
|
and valid
|
|
and filter_allowed
|
|
and not access_denied
|
|
):
|
|
|
|
applied.append(name)
|
|
|
|
else:
|
|
|
|
denied.append(name)
|
|
|
|
return applied, denied
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Analyse des groupes de sécurité
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _get_security_groups(scope_results, principal_name):
|
|
"""
|
|
Extrait et classe les groupes de sécurité utilisés dans le RSoP USER.
|
|
|
|
Catégories :
|
|
|
|
domain:
|
|
Groupes dont le préfixe correspond au préfixe de domaine
|
|
du compte utilisateur.
|
|
|
|
Exemple :
|
|
FORMATION\\Domain Users
|
|
|
|
local:
|
|
Groupes BUILTIN ou groupes appartenant au poste local.
|
|
|
|
Exemples :
|
|
BUILTIN\\Users
|
|
WSMANAGE30\\Administrators
|
|
|
|
system:
|
|
Identités NT AUTHORITY.
|
|
|
|
Exemples :
|
|
NT AUTHORITY\\Authenticated Users
|
|
NT AUTHORITY\\INTERACTIVE
|
|
|
|
other:
|
|
Identités de sécurité qui ne peuvent pas être classées
|
|
de manière fiable.
|
|
|
|
Cette distinction évite notamment qu'un compte local tel que :
|
|
|
|
WSMANAGE30\\formation
|
|
|
|
soit considéré à tort comme un compte Active Directory.
|
|
"""
|
|
|
|
groups = {
|
|
"domain": [],
|
|
"local": [],
|
|
"system": [],
|
|
"other": [],
|
|
}
|
|
|
|
seen = set()
|
|
|
|
if scope_results is None:
|
|
return groups
|
|
|
|
principal_domain, _ = _split_account(
|
|
principal_name
|
|
)
|
|
|
|
principal_domain = principal_domain.upper()
|
|
|
|
computer_name = os.environ.get(
|
|
"COMPUTERNAME",
|
|
""
|
|
).upper()
|
|
|
|
# Le préfixe du compte n'est considéré comme un domaine AD
|
|
# que s'il est différent du nom du poste local.
|
|
ad_domain = ""
|
|
|
|
if (
|
|
principal_domain
|
|
and principal_domain != computer_name
|
|
):
|
|
ad_domain = principal_domain
|
|
|
|
for element in scope_results.iter():
|
|
|
|
if _clean_tag(element.tag) != "SecurityGroup":
|
|
continue
|
|
|
|
name = _text(
|
|
element,
|
|
"Name"
|
|
)
|
|
|
|
if not name:
|
|
continue
|
|
|
|
name = name.strip()
|
|
|
|
key = name.lower()
|
|
|
|
if key in seen:
|
|
continue
|
|
|
|
seen.add(key)
|
|
|
|
upper_name = name.upper()
|
|
|
|
if upper_name.startswith(
|
|
"NT AUTHORITY\\"
|
|
):
|
|
|
|
groups["system"].append(name)
|
|
|
|
elif upper_name.startswith(
|
|
"BUILTIN\\"
|
|
):
|
|
|
|
groups["local"].append(name)
|
|
|
|
elif (
|
|
computer_name
|
|
and upper_name.startswith(
|
|
computer_name + "\\"
|
|
)
|
|
):
|
|
|
|
groups["local"].append(name)
|
|
|
|
elif (
|
|
ad_domain
|
|
and upper_name.startswith(
|
|
ad_domain + "\\"
|
|
)
|
|
):
|
|
|
|
groups["domain"].append(name)
|
|
|
|
else:
|
|
|
|
groups["other"].append(name)
|
|
|
|
# Garantit un ordre stable dans les résultats.
|
|
for category in groups:
|
|
|
|
groups[category] = sorted(
|
|
groups[category],
|
|
key=lambda x: x.lower()
|
|
)
|
|
|
|
return groups
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Audit WAPT
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def audit():
|
|
"""
|
|
Exécute GPResult pour le contexte USER de l'utilisateur interactif actif.
|
|
|
|
Déroulement :
|
|
1. Détecte la session utilisateur Windows active.
|
|
2. Exécute GPResult pour le RSoP USER de ce compte.
|
|
3. Analyse les GPO appliquées et refusées.
|
|
4. Extrait et classe les groupes de sécurité.
|
|
5. Publie le résultat structuré dans les données d'audit WAPT.
|
|
|
|
L'absence d'utilisateur interactif est considérée comme une situation
|
|
normale. Dans ce cas, l'audit WAPT retourne OK avec un statut
|
|
fonctionnel NO_USER.
|
|
"""
|
|
|
|
xml_file = os.path.join(
|
|
tempfile.gettempdir(),
|
|
"wapt-gpresult-user.xml"
|
|
)
|
|
|
|
try:
|
|
|
|
# WAPT s'exécute en tant que service.
|
|
# Il faut donc identifier explicitement l'utilisateur interactif.
|
|
logged_on_user = _get_logged_on_user()
|
|
|
|
if not logged_on_user:
|
|
|
|
audit_data = {
|
|
|
|
"status": "NO_USER",
|
|
|
|
"message":
|
|
"No active interactive user session",
|
|
|
|
"user": None,
|
|
|
|
"applied_gpos": [],
|
|
|
|
"denied_gpos": [],
|
|
|
|
"domain_groups": [],
|
|
|
|
"security_groups": {
|
|
"domain": [],
|
|
"local": [],
|
|
"system": [],
|
|
"other": [],
|
|
}
|
|
}
|
|
|
|
WAPT.write_audit_data_if_changed(
|
|
AUDIT_SECTION,
|
|
AUDIT_NAME,
|
|
audit_data,
|
|
keep_days=365,
|
|
max_count=1
|
|
)
|
|
|
|
print(
|
|
"No active interactive user session"
|
|
)
|
|
|
|
# L'absence d'utilisateur connecté n'est pas considérée
|
|
# comme un échec de l'audit.
|
|
return "OK"
|
|
|
|
print(
|
|
"Active interactive user: %s"
|
|
% logged_on_user
|
|
)
|
|
|
|
# Supprime un éventuel rapport restant d'une exécution précédente.
|
|
if os.path.isfile(xml_file):
|
|
os.remove(xml_file)
|
|
|
|
gpresult_path = makepath(
|
|
system32(),
|
|
"gpresult.exe"
|
|
)
|
|
|
|
# Le nom utilisateur retourné par quser suffit à GPResult
|
|
# pour résoudre le compte local ou de domaine réellement associé
|
|
# au RSoP.
|
|
command = (
|
|
'"%s" '
|
|
'/USER "%s" '
|
|
'/SCOPE USER '
|
|
'/X "%s" '
|
|
'/F'
|
|
% (
|
|
gpresult_path,
|
|
logged_on_user,
|
|
xml_file
|
|
)
|
|
)
|
|
|
|
print(
|
|
"Running GPResult USER for %s..."
|
|
% logged_on_user
|
|
)
|
|
|
|
result = run(
|
|
command,
|
|
timeout=120,
|
|
return_stderr=False
|
|
)
|
|
|
|
if not os.path.isfile(xml_file):
|
|
|
|
raise Exception(
|
|
"GPResult did not generate XML report. "
|
|
"Output: %s"
|
|
% result
|
|
)
|
|
|
|
tree = ET.parse(
|
|
xml_file
|
|
)
|
|
|
|
root = tree.getroot()
|
|
|
|
user_results = _find_user_results(
|
|
root
|
|
)
|
|
|
|
if user_results is None:
|
|
|
|
raise Exception(
|
|
"UserResults section not found"
|
|
)
|
|
|
|
applied_gpos, denied_gpos = _get_gpos(
|
|
user_results
|
|
)
|
|
|
|
# On laisse GPResult fournir l'identité de compte faisant foi.
|
|
#
|
|
# Exemples :
|
|
# FORMATION\\sebastien
|
|
# WSMANAGE30\\formation
|
|
rsop_user = _text(
|
|
user_results,
|
|
"Name"
|
|
)
|
|
|
|
if not rsop_user:
|
|
rsop_user = logged_on_user
|
|
|
|
account_domain, username = _split_account(
|
|
rsop_user
|
|
)
|
|
|
|
security_groups = _get_security_groups(
|
|
user_results,
|
|
rsop_user
|
|
)
|
|
|
|
computer_name = os.environ.get(
|
|
"COMPUTERNAME",
|
|
""
|
|
)
|
|
|
|
# Détermine si GPResult a résolu le compte
|
|
# comme compte local ou compte de domaine.
|
|
if (
|
|
not account_domain
|
|
or account_domain.upper()
|
|
== computer_name.upper()
|
|
):
|
|
account_type = "local"
|
|
|
|
else:
|
|
account_type = "domain"
|
|
|
|
audit_data = {
|
|
|
|
"status": "OK",
|
|
|
|
"user": {
|
|
|
|
# Identité complète du compte telle que résolue par GPResult.
|
|
"account": rsop_user,
|
|
|
|
"username": username,
|
|
|
|
# Préfixe au format NetBIOS :
|
|
# nom de domaine pour un compte AD,
|
|
# nom du poste pour un compte local.
|
|
"account_domain": account_domain,
|
|
|
|
"account_type": account_type,
|
|
},
|
|
|
|
"summary": {
|
|
|
|
"applied_gpo_count": len(
|
|
applied_gpos
|
|
),
|
|
|
|
"denied_gpo_count": len(
|
|
denied_gpos
|
|
),
|
|
|
|
"total_gpo_count": (
|
|
len(applied_gpos)
|
|
+ len(denied_gpos)
|
|
),
|
|
|
|
"domain_group_count": len(
|
|
security_groups["domain"]
|
|
),
|
|
|
|
"local_group_count": len(
|
|
security_groups["local"]
|
|
),
|
|
|
|
"system_group_count": len(
|
|
security_groups["system"]
|
|
),
|
|
|
|
"other_group_count": len(
|
|
security_groups["other"]
|
|
),
|
|
},
|
|
|
|
# Des listes simples sont volontairement conservées
|
|
# afin de faciliter l'exploitation en SQL / reporting.
|
|
"applied_gpos": applied_gpos,
|
|
|
|
"denied_gpos": denied_gpos,
|
|
|
|
# Liste pratique contenant uniquement les groupes AD,
|
|
# généralement les plus intéressants dans un audit GPO.
|
|
"domain_groups":
|
|
security_groups["domain"],
|
|
|
|
# Le contexte complet est conservé pour faciliter le diagnostic.
|
|
"security_groups":
|
|
security_groups,
|
|
}
|
|
|
|
WAPT.write_audit_data_if_changed(
|
|
AUDIT_SECTION,
|
|
AUDIT_NAME,
|
|
audit_data,
|
|
keep_days=365,
|
|
max_count=1
|
|
)
|
|
|
|
print(
|
|
"GPResult USER %s: "
|
|
"%s applied / %s denied"
|
|
% (
|
|
rsop_user,
|
|
len(applied_gpos),
|
|
len(denied_gpos)
|
|
)
|
|
)
|
|
|
|
print(
|
|
"Account type: %s"
|
|
% account_type
|
|
)
|
|
|
|
print(
|
|
"AD groups: %s"
|
|
% len(
|
|
security_groups["domain"]
|
|
)
|
|
)
|
|
|
|
return "OK"
|
|
|
|
except Exception as e:
|
|
|
|
# Conserve également l'erreur dans les données d'audit WAPT
|
|
# afin qu'elle reste visible depuis la console.
|
|
WAPT.write_audit_data_if_changed(
|
|
AUDIT_SECTION,
|
|
AUDIT_NAME,
|
|
{
|
|
"status": "ERROR",
|
|
"message": str(e)
|
|
},
|
|
keep_days=365,
|
|
max_count=1
|
|
)
|
|
|
|
print(
|
|
"GPResult USER audit error: %s"
|
|
% str(e)
|
|
)
|
|
|
|
return "ERROR"
|
|
|
|
finally:
|
|
|
|
# Le XML GPResult est uniquement utilisé comme fichier temporaire.
|
|
try:
|
|
|
|
if os.path.isfile(xml_file):
|
|
os.remove(xml_file)
|
|
|
|
except Exception:
|
|
pass |