diff --git a/setup.py b/setup.py index be20796..51f443a 100644 --- a/setup.py +++ b/setup.py @@ -1,272 +1,734 @@ # -*- 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 * +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(): - """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 -def _get_logged_on_user(): - """Retourne DOMAINE\\utilisateur pour l'unique utilisateur actif. +# --------------------------------------------------------------------------- +# Fonctions utilitaires XML +# --------------------------------------------------------------------------- - Les sessions deconnectees sont ignorees. Plusieurs comptes actifs - provoquent une erreur pour eviter de choisir une cible arbitraire. - Sans utilisateur actif, retourne None. +def _clean_tag(tag): """ - import win32ts - - server = win32ts.WTS_CURRENT_SERVER_HANDLE - users = [] - - for session in win32ts.WTSEnumerateSessions(server, 1, 0): - if session["State"] != win32ts.WTSActive: - continue - - session_id = session["SessionId"] - - username = win32ts.WTSQuerySessionInformation( - server, session_id, win32ts.WTSUserName - ) - domain = win32ts.WTSQuerySessionInformation( - server, session_id, win32ts.WTSDomainName - ) - - username = (username or "").strip() - domain = (domain or "").strip() - - 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. + Supprime le namespace XML d'un tag ElementTree. """ - WAPT.write_audit_data_if_changed( - "audit-gpresult", - "user-gpo", - data, - keep_days=365, - max_count=1 + 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 _xml_flag(element, tag): - """Lit un booleen XML ; None signifie que le champ est absent.""" - value = element.findtext(tag) +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 - if value is None: return None - value = value.strip().lower() + except Exception as e: - if value in ("true", "1"): - return True - - if value in ("false", "0"): - return False - - raise Exception( - "Valeur XML inattendue pour %s : %s" % (tag, value) - ) - - -def _parse_user_gpos(xml_path): - """Retourne les listes triees des GPO appliquees et non appliquees. - - denied_gpos regroupe toutes les GPO non appliquees selon les indicateurs - XML : desactivation, invalidite, filtrage ou refus d'acces. - """ - root = ET.parse(xml_path).getroot() - - # Neutraliser les espaces de noms pour lire les balises par leur nom. - for element in root.iter(): - if isinstance(element.tag, str): - element.tag = element.tag.rsplit("}", 1)[-1] - - user_results = root.find(".//UserResults") - - if user_results is None: - raise Exception( - "Le rapport XML ne contient aucune section UserResults." + 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 = [] - # Les references imbriquees ne sont pas recomptees comme des GPO. - for gpo in user_results.findall("GPO"): - name = ( - gpo.findtext("Name") - or gpo.findtext("Path/Identifier") - or "" - ).strip() + 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: - raise Exception( - "GPO sans nom ni identifiant dans le rapport XML." - ) + continue - enabled = _xml_flag(gpo, "Enabled") - valid = _xml_flag(gpo, "IsValid") - 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 ( - enabled is False - or valid is False - or allowed is False - or access_denied is True + enabled + and valid + and filter_allowed + and not access_denied ): - denied.append(name) - elif valid is True and allowed is True: applied.append(name) else: - # Un etat inconnu ne doit pas devenir un faux succes. - raise Exception( - "Etat de la GPO indeterminable : %s" % name - ) - return sorted(applied), sorted(denied) + denied.append(name) + + return applied, denied -def audit(): - """Collecte le GPResult utilisateur et retourne OK, WARNING ou ERROR.""" - user = None - computer = { - "name": os.environ.get("COMPUTERNAME", "") +# --------------------------------------------------------------------------- +# 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": [], } - try: - user = _get_logged_on_user() + seen = set() - if not user: - data = { - "status": "WARNING", - "message": "Aucun utilisateur avec une session active.", - "computer": computer, - "user": None + 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": [], + } } - _write_user_audit(data) - print(data["message"]) - return "WARNING" - - print("Audit GPResult utilisateur : %s" % user) - - # Dossier unique, automatiquement supprime a la sortie du bloc, - # y compris si l'export ou l'analyse XML echoue. - with tempfile.TemporaryDirectory( - prefix="wapt_gpresult_user_" - ) as temp_dir: - xml_path = os.path.join(temp_dir, "gpresult.xml") - - # Conserver l'appel direct valide sur l'environnement cible. - # /USER cible les donnees de l'utilisateur ; /SCOPE USER - # exclut les GPO machine ; /X exporte les donnees en XML. - command = ( - 'gpresult.exe /USER "%s" /SCOPE USER /X "%s" /F' - % (user, xml_path) + WAPT.write_audit_data_if_changed( + AUDIT_SECTION, + AUDIT_NAME, + audit_data, + keep_days=365, + max_count=1 ) - print("Commande : %s" % command) print( - "Dossier temporaire present : %s" - % os.path.isdir(temp_dir) + "No active interactive user session" ) - result = run( - command, - timeout=120, - return_stderr=True - ) - - if result: - print(result) - - if not os.path.isfile(xml_path): - raise Exception( - "GPResult n'a pas genere de rapport XML : %s" - % xml_path - ) - - applied, denied = _parse_user_gpos(xml_path) - - # Le domaine provient du compte WTS (souvent au format NetBIOS). - domain, separator, username = user.partition("\\") - - if not separator: - username = user - domain = "" - - data = { - "status": "OK", - "summary": { - "total_gpo_count": len(applied) + len(denied), - "denied_gpo_count": len(denied), - "applied_gpo_count": len(applied) - }, - "computer": computer, - "user": { - "account": user, - "name": username, - "domain": domain - }, - "denied_gpos": denied, - "applied_gpos": applied - } - - _write_user_audit(data) + # L'absence d'utilisateur connecté n'est pas considérée + # comme un échec de l'audit. + return "OK" print( - "Audit disponible : audit-gpresult.user-gpo " - "(%s appliquees, %s non appliquees)" - % (len(applied), len(denied)) + "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: - # Remplacer le resultat precedent par l'erreur de cette collecte. - data = { - "status": "ERROR", - "message": str(e), - "computer": computer, - "user": user - } - _write_user_audit(data) - print("Erreur audit GPResult utilisateur : %s" % 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 \ No newline at end of file