# -*- 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 * def install(): """Point d'entree obligatoire ; aucune configuration a appliquer.""" pass 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 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. """ WAPT.write_audit_data_if_changed( "audit-gpresult", "user-gpo", data, keep_days=365, max_count=1 ) 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 value = value.strip().lower() 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." ) 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() if not name: raise Exception( "GPO sans nom ni identifiant dans le rapport XML." ) enabled = _xml_flag(gpo, "Enabled") valid = _xml_flag(gpo, "IsValid") allowed = _xml_flag(gpo, "FilterAllowed") access_denied = _xml_flag(gpo, "AccessDenied") if ( enabled is False or valid is False or allowed is False or access_denied is True ): 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) def audit(): """Collecte le GPResult utilisateur et retourne OK, WARNING ou ERROR.""" user = None computer = { "name": os.environ.get("COMPUTERNAME", "") } try: user = _get_logged_on_user() if not user: data = { "status": "WARNING", "message": "Aucun utilisateur avec une session active.", "computer": computer, "user": None } _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) ) print("Commande : %s" % command) print( "Dossier temporaire present : %s" % os.path.isdir(temp_dir) ) 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) print( "Audit disponible : audit-gpresult.user-gpo " "(%s appliquees, %s non appliquees)" % (len(applied), len(denied)) ) 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) return "ERROR"