paquets-publics/comi-audit-gpresult-user
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enrichit l'audit GPResult utilisateur et les groupes de sécurité

Browse Source
This commit is contained in:
slaporte committed 2026-10-30 16:48:03 +01:00
1 parent 010f45b8cf
commit be8cd468f6
1 file changed
+648 -186
Download Patch File Download Diff File Expand all files Collapse all files
setup.py
+648 -186
View File
@@ -1,272 +1,734 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""Audit des GPO de l'utilisateur actif : audit-gpresult.user-gpo.
L'installation ne realise aucune action. La collecte est executee dans
audit(), sous le compte de l'agent WAPT, en ciblant explicitement les
donnees RSoP de l'utilisateur avec l'option /USER de gpresult.exe.
"""
import os
import tempfile
import xml.etree.ElementTree as ET
from setuphelpers import * 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(): def install():
"""Point d'entree obligatoire ; aucune configuration a appliquer.""" """
Ce paquet est uniquement destiné à l'audit.
Aucune modification persistante n'est effectuée sur le poste.
"""
pass 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(): def _get_logged_on_user():
"""Retourne DOMAINE\\utilisateur pour l'unique utilisateur actif.
Les sessions deconnectees sont ignorees. Plusieurs comptes actifs
provoquent une erreur pour eviter de choisir une cible arbitraire.
Sans utilisateur actif, retourne None.
""" """
import win32ts Retourne le nom de l'utilisateur associé à la session interactive active.
server = win32ts.WTS_CURRENT_SERVER_HANDLE Les audits WAPT s'exécutent dans le contexte du service WAPT.
users = [] Il n'est donc pas possible de se fier simplement à l'identité
du processus courant.
for session in win32ts.WTSEnumerateSessions(server, 1, 0): `quser` est utilisé pour inspecter les sessions Windows et récupérer
if session["State"] != win32ts.WTSActive: l'utilisateur de la session interactive active.
continue
session_id = session["SessionId"] Exemples retournés par cette fonction :
formation
sebastien
username = win32ts.WTSQuerySessionInformation( L'identité complète du compte est ensuite résolue par GPResult :
server, session_id, win32ts.WTSUserName WSMANAGE30\\formation
) FORMATION\\sebastien
domain = win32ts.WTSQuerySessionInformation(
server, session_id, win32ts.WTSDomainName
)
username = (username or "").strip() Returns:
domain = (domain or "").strip() str ou None
if not username:
continue
account = (
"%s\\%s" % (domain, username)
if domain else username
)
if account not in users:
users.append(account)
if len(users) > 1:
raise Exception(
"Plusieurs utilisateurs ont une session active : %s"
% ", ".join(users)
)
return users[0] if users else None
def _write_user_audit(data):
"""Enregistre un objet structure, sans conversion en chaine JSON.
La meme cle est utilisee pour les succes, avertissements et erreurs,
afin de ne pas laisser un ancien resultat presente comme actuel.
""" """
WAPT.write_audit_data_if_changed(
"audit-gpresult", try:
"user-gpo",
data, output = run(
keep_days=365, r'quser.exe',
max_count=1 timeout=30,
return_stderr=False
) )
if not output:
def _xml_flag(element, tag):
"""Lit un booleen XML ; None signifie que le champ est absent."""
value = element.findtext(tag)
if value is None:
return None return None
value = value.strip().lower() lines = output.splitlines()
if value in ("true", "1"): # La première ligne correspond aux en-têtes de colonnes.
return True for line in lines[1:]:
if value in ("false", "0"): line = line.strip()
return False
raise Exception( if not line:
"Valeur XML inattendue pour %s : %s" % (tag, value) 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
def _parse_user_gpos(xml_path): # USERNAME correspond à la première colonne.
"""Retourne les listes triees des GPO appliquees et non appliquees. username = parts[0].strip()
denied_gpos regroupe toutes les GPO non appliquees selon les indicateurs if username:
XML : desactivation, invalidite, filtrage ou refus d'acces. 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.
""" """
root = ET.parse(xml_path).getroot()
# Neutraliser les espaces de noms pour lire les balises par leur nom.
for element in root.iter(): for element in root.iter():
if isinstance(element.tag, str):
element.tag = element.tag.rsplit("}", 1)[-1]
user_results = root.find(".//UserResults") if _clean_tag(element.tag) == "UserResults":
return element
if user_results is None: return None
raise Exception(
"Le rapport XML ne contient aucune section UserResults."
) # ---------------------------------------------------------------------------
# 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 = [] applied = []
denied = [] denied = []
# Les references imbriquees ne sont pas recomptees comme des GPO. seen = set()
for gpo in user_results.findall("GPO"):
name = (
gpo.findtext("Name")
or gpo.findtext("Path/Identifier")
or ""
).strip()
if not name: if scope_results is None:
raise Exception( return applied, denied
"GPO sans nom ni identifiant dans le rapport XML."
for element in scope_results.iter():
if _clean_tag(element.tag) != "GPO":
continue
name = _text(
element,
"Name"
) )
enabled = _xml_flag(gpo, "Enabled") if not name:
valid = _xml_flag(gpo, "IsValid") continue
allowed = _xml_flag(gpo, "FilterAllowed")
access_denied = _xml_flag(gpo, "AccessDenied") 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 ( if (
enabled is False enabled
or valid is False and valid
or allowed is False and filter_allowed
or access_denied is True and not access_denied
): ):
denied.append(name)
elif valid is True and allowed is True:
applied.append(name) applied.append(name)
else: else:
# Un etat inconnu ne doit pas devenir un faux succes.
raise Exception( denied.append(name)
"Etat de la GPO indeterminable : %s" % 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
) )
return sorted(applied), sorted(denied) 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(): def audit():
"""Collecte le GPResult utilisateur et retourne OK, WARNING ou ERROR.""" """
user = None Exécute GPResult pour le contexte USER de l'utilisateur interactif actif.
computer = {
"name": os.environ.get("COMPUTERNAME", "")
}
try: Déroulement :
user = _get_logged_on_user() 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.
if not user: L'absence d'utilisateur interactif est considérée comme une situation
data = { normale. Dans ce cas, l'audit WAPT retourne OK avec un statut
"status": "WARNING", fonctionnel NO_USER.
"message": "Aucun utilisateur avec une session active.", """
"computer": computer,
"user": None
}
_write_user_audit(data) xml_file = os.path.join(
print(data["message"]) tempfile.gettempdir(),
return "WARNING" "wapt-gpresult-user.xml"
)
print("Audit GPResult utilisateur : %s" % user)
try:
# Dossier unique, automatiquement supprime a la sortie du bloc,
# y compris si l'export ou l'analyse XML echoue. # WAPT s'exécute en tant que service.
with tempfile.TemporaryDirectory( # Il faut donc identifier explicitement l'utilisateur interactif.
prefix="wapt_gpresult_user_" logged_on_user = _get_logged_on_user()
) as temp_dir:
xml_path = os.path.join(temp_dir, "gpresult.xml") if not logged_on_user:
# Conserver l'appel direct valide sur l'environnement cible. audit_data = {
# /USER cible les donnees de l'utilisateur ; /SCOPE USER
# exclut les GPO machine ; /X exporte les donnees en XML. "status": "NO_USER",
command = (
'gpresult.exe /USER "%s" /SCOPE USER /X "%s" /F' "message":
% (user, xml_path) "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("Commande : %s" % command)
print( print(
"Dossier temporaire present : %s" "No active interactive user session"
% os.path.isdir(temp_dir) )
# 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( result = run(
command, command,
timeout=120, timeout=120,
return_stderr=True return_stderr=False
) )
if result: if not os.path.isfile(xml_file):
print(result)
if not os.path.isfile(xml_path):
raise Exception( raise Exception(
"GPResult n'a pas genere de rapport XML : %s" "GPResult did not generate XML report. "
% xml_path "Output: %s"
% result
) )
applied, denied = _parse_user_gpos(xml_path) tree = ET.parse(
xml_file
)
# Le domaine provient du compte WTS (souvent au format NetBIOS). root = tree.getroot()
domain, separator, username = user.partition("\\")
if not separator: user_results = _find_user_results(
username = user root
domain = "" )
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 = {
data = {
"status": "OK", "status": "OK",
"summary": {
"total_gpo_count": len(applied) + len(denied),
"denied_gpo_count": len(denied),
"applied_gpo_count": len(applied)
},
"computer": computer,
"user": { "user": {
"account": user,
"name": username, # Identité complète du compte telle que résolue par GPResult.
"domain": domain "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,
}, },
"denied_gpos": denied,
"applied_gpos": applied "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,
} }
_write_user_audit(data) WAPT.write_audit_data_if_changed(
AUDIT_SECTION,
AUDIT_NAME,
audit_data,
keep_days=365,
max_count=1
)
print( print(
"Audit disponible : audit-gpresult.user-gpo " "GPResult USER %s: "
"(%s appliquees, %s non appliquees)" "%s applied / %s denied"
% (len(applied), len(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" return "OK"
except Exception as e: except Exception as e:
# Remplacer le resultat precedent par l'erreur de cette collecte.
data = {
"status": "ERROR",
"message": str(e),
"computer": computer,
"user": user
}
_write_user_audit(data) # Conserve également l'erreur dans les données d'audit WAPT
print("Erreur audit GPResult utilisateur : %s" % e) # 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" 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