From 1feb72bf1721a29fda7e05e97bab8293465b080e Mon Sep 17 00:00:00 2001 From: Johnny Date: Fri, 4 Sep 2026 11:40:11 +0200 Subject: [PATCH] =?UTF-8?q?Version=20initiale=20:=20diagnostic=20r=C3=A9se?= =?UTF-8?q?au=20AD=20+=20jointure=20de=20domaine?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Application PySide6 (GUI + CLI) fonctionnant sous Linux et Windows : - Diagnostic réseau complet AD (DNS, SRV, ports, NTP, outils requis) - Jointure de domaine via realmd (Linux) ou Add-Computer (Windows) - Installation automatique des prérequis Linux (apt) - CLI colorée (diagnose/join/install-prereqs), sans dépendance PySide6 - Scripts de packaging PyInstaller pour Linux et Windows Co-Authored-By: Claude Sonnet 5 --- .gitignore | 7 + domainjoin/__init__.py | 1 + domainjoin/backend/__init__.py | 19 ++ domainjoin/backend/base.py | 47 +++ domainjoin/backend/linux_realmd.py | 134 +++++++++ domainjoin/backend/windows_addcomputer.py | 118 ++++++++ domainjoin/cli.py | 223 ++++++++++++++ domainjoin/diagnostics.py | 247 ++++++++++++++++ domainjoin/gui.py | 337 ++++++++++++++++++++++ domainjoin/privileges.py | 22 ++ domainjoin/worker.py | 72 +++++ main.py | 48 +++ packaging/build_linux.sh | 25 ++ packaging/build_windows.bat | 26 ++ requirements.txt | 3 + 15 files changed, 1329 insertions(+) create mode 100644 .gitignore create mode 100644 domainjoin/__init__.py create mode 100644 domainjoin/backend/__init__.py create mode 100644 domainjoin/backend/base.py create mode 100644 domainjoin/backend/linux_realmd.py create mode 100644 domainjoin/backend/windows_addcomputer.py create mode 100644 domainjoin/cli.py create mode 100644 domainjoin/diagnostics.py create mode 100644 domainjoin/gui.py create mode 100644 domainjoin/privileges.py create mode 100644 domainjoin/worker.py create mode 100644 main.py create mode 100755 packaging/build_linux.sh create mode 100644 packaging/build_windows.bat create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e3fa3c5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.pyc +build/ +dist/ +*.spec +.build-venv-linux/ +.build-venv-windows/ diff --git a/domainjoin/__init__.py b/domainjoin/__init__.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/domainjoin/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/domainjoin/backend/__init__.py b/domainjoin/backend/__init__.py new file mode 100644 index 0000000..ef6201a --- /dev/null +++ b/domainjoin/backend/__init__.py @@ -0,0 +1,19 @@ +"""Sélection du backend de jointure de domaine selon la plateforme.""" + +import platform + +from .base import DomainJoinBackend, JoinResult + + +def get_backend() -> DomainJoinBackend: + if platform.system() == "Windows": + from .windows_addcomputer import WindowsAddComputerBackend + + return WindowsAddComputerBackend() + + from .linux_realmd import LinuxRealmdBackend + + return LinuxRealmdBackend() + + +__all__ = ["get_backend", "DomainJoinBackend", "JoinResult"] diff --git a/domainjoin/backend/base.py b/domainjoin/backend/base.py new file mode 100644 index 0000000..8b814c5 --- /dev/null +++ b/domainjoin/backend/base.py @@ -0,0 +1,47 @@ +"""Interface commune aux backends de jointure de domaine (Linux/Windows).""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Callable, Optional + + +@dataclass +class JoinResult: + success: bool + message: str + needs_reboot: bool = False + + +OutputCallback = Callable[[str], None] + + +class DomainJoinBackend(ABC): + """Backend de jointure à un domaine Active Directory.""" + + @abstractmethod + def is_available(self) -> tuple[bool, str]: + """Indique si les outils nécessaires sont présents. Retourne (disponible, message).""" + + @abstractmethod + def join( + self, + domain: str, + username: str, + password: str, + ou: Optional[str] = None, + computer_name: Optional[str] = None, + on_output: Optional[OutputCallback] = None, + ) -> JoinResult: + """Rejoint le domaine. Le mot de passe n'est jamais passé en argument + de ligne de commande : il est transmis via l'entrée standard du + sous-processus pour ne pas apparaître dans la liste des processus.""" + + def can_install_prerequisites(self) -> bool: + """True si ce backend sait installer automatiquement les outils manquants.""" + return False + + def install_prerequisites(self, on_output: Optional[OutputCallback] = None) -> JoinResult: + """Installe les outils manquants. À surcharger si can_install_prerequisites() est vrai.""" + return JoinResult(success=False, message="Installation automatique non prise en charge pour ce backend.") diff --git a/domainjoin/backend/linux_realmd.py b/domainjoin/backend/linux_realmd.py new file mode 100644 index 0000000..f8104d0 --- /dev/null +++ b/domainjoin/backend/linux_realmd.py @@ -0,0 +1,134 @@ +"""Backend de jointure de domaine pour Linux, via realmd (`realm join`).""" + +from __future__ import annotations + +import os +import shutil +import subprocess +from typing import Optional + +from .base import DomainJoinBackend, JoinResult, OutputCallback + +REQUIRED_PACKAGES = ["realmd", "sssd", "adcli", "samba-common-bin", "packagekit"] +INSTALL_HINT = "sudo apt install " + " ".join(REQUIRED_PACKAGES) + + +class LinuxRealmdBackend(DomainJoinBackend): + def is_available(self) -> tuple[bool, str]: + if shutil.which("realm") is None: + return False, f"La commande 'realm' est introuvable. Installez-la avec : {INSTALL_HINT}" + return True, "realmd disponible." + + def can_install_prerequisites(self) -> bool: + return shutil.which("apt-get") is not None + + def install_prerequisites(self, on_output: Optional[OutputCallback] = None) -> JoinResult: + def emit(line: str) -> None: + if on_output: + on_output(line) + + if shutil.which("apt-get") is None: + return JoinResult(success=False, message="apt-get est introuvable sur ce système.") + + env = os.environ.copy() + env["DEBIAN_FRONTEND"] = "noninteractive" + + for cmd in (["apt-get", "update"], ["apt-get", "install", "-y", *REQUIRED_PACKAGES]): + emit(f"$ {' '.join(cmd)}") + try: + process = subprocess.Popen( + cmd, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + env=env, + ) + except OSError as exc: + return JoinResult(success=False, message=f"Impossible de lancer apt-get : {exc}") + + if process.stdout: + for line in process.stdout: + emit(line.rstrip("\n")) + returncode = process.wait() + if returncode != 0: + return JoinResult( + success=False, + message=f"Échec de « {' '.join(cmd)} » (code retour {returncode}). " + "Vérifiez que l'application tourne avec les droits root.", + ) + + available, _ = self.is_available() + if available: + return JoinResult(success=True, message="Les prérequis ont été installés avec succès.") + return JoinResult( + success=False, + message="Installation terminée mais 'realm' reste introuvable. Vérifiez le journal ci-dessus.", + ) + + def join( + self, + domain: str, + username: str, + password: str, + ou: Optional[str] = None, + computer_name: Optional[str] = None, + on_output: Optional[OutputCallback] = None, + ) -> JoinResult: + available, reason = self.is_available() + if not available: + return JoinResult(success=False, message=reason) + + def emit(line: str) -> None: + if on_output: + on_output(line) + + if computer_name: + emit( + "Note : realm join utilise le nom d'hôte système du poste comme nom de " + "machine dans l'annuaire. Changez le hostname (hostnamectl) avant la " + f"jointure si vous voulez que la machine apparaisse comme « {computer_name} »." + ) + + cmd = ["realm", "-v", "join", "--user", username] + if ou: + cmd += ["--computer-ou", ou] + cmd.append(domain) + + emit(f"$ realm -v join --user {username}" + (f" --computer-ou {ou}" if ou else "") + f" {domain}") + + try: + process = subprocess.Popen( + cmd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + ) + except OSError as exc: + return JoinResult(success=False, message=f"Impossible de lancer realm : {exc}") + + try: + process.stdin.write(password + "\n") + process.stdin.close() + except BrokenPipeError: + pass + + if process.stdout: + for line in process.stdout: + emit(line.rstrip("\n")) + + returncode = process.wait() + + if returncode == 0: + return JoinResult( + success=True, + message=f"Le poste a rejoint le domaine {domain} avec succès.", + needs_reboot=True, + ) + return JoinResult( + success=False, + message=f"Échec de la jointure au domaine (code retour {returncode}). Voir le journal ci-dessus.", + ) diff --git a/domainjoin/backend/windows_addcomputer.py b/domainjoin/backend/windows_addcomputer.py new file mode 100644 index 0000000..d36fc16 --- /dev/null +++ b/domainjoin/backend/windows_addcomputer.py @@ -0,0 +1,118 @@ +"""Backend de jointure de domaine pour Windows, via la cmdlet PowerShell Add-Computer.""" + +from __future__ import annotations + +import os +import shutil +import subprocess +from typing import Optional + +from .base import DomainJoinBackend, JoinResult, OutputCallback + +_PASSWORD_ENV_VAR = "DOMAINJOIN_PWD" +_SUCCESS_MARKER = "JOIN_SUCCESS" +_FAILURE_PREFIX = "JOIN_FAILURE:" + + +def _ps_quote(value: str) -> str: + """Échappe une valeur pour l'insérer dans une chaîne PowerShell entre quotes simples.""" + return value.replace("'", "''") + + +class WindowsAddComputerBackend(DomainJoinBackend): + def is_available(self) -> tuple[bool, str]: + if shutil.which("powershell") is None and shutil.which("powershell.exe") is None: + return False, "PowerShell est introuvable sur ce poste." + return True, "PowerShell disponible." + + def _build_script(self, domain: str, username: str, ou: Optional[str], computer_name: Optional[str]) -> str: + lines = [ + "$ErrorActionPreference = 'Stop'", + f"$domain = '{_ps_quote(domain)}'", + f"$username = '{_ps_quote(username)}'", + f"$password = $env:{_PASSWORD_ENV_VAR}", + "if ([string]::IsNullOrEmpty($password)) { Write-Output 'JOIN_FAILURE: mot de passe manquant.'; exit 2 }", + "$securePwd = ConvertTo-SecureString -String $password -AsPlainText -Force", + "$cred = New-Object System.Management.Automation.PSCredential($username, $securePwd)", + "$params = @{ DomainName = $domain; Credential = $cred; Force = $true; Restart = $false }", + ] + if ou: + lines.append(f"$params['OUPath'] = '{_ps_quote(ou)}'") + if computer_name: + lines.append(f"$params['NewName'] = '{_ps_quote(computer_name)}'") + lines += [ + "try {", + " Add-Computer @params -ErrorAction Stop", + f" Write-Output '{_SUCCESS_MARKER}'", + "} catch {", + f" Write-Output ('{_FAILURE_PREFIX} ' + $_.Exception.Message)", + " exit 1", + "}", + ] + return "\n".join(lines) + + def join( + self, + domain: str, + username: str, + password: str, + ou: Optional[str] = None, + computer_name: Optional[str] = None, + on_output: Optional[OutputCallback] = None, + ) -> JoinResult: + available, reason = self.is_available() + if not available: + return JoinResult(success=False, message=reason) + + def emit(line: str) -> None: + if on_output: + on_output(line) + + script = self._build_script(domain, username, ou, computer_name) + emit(f"Exécution de Add-Computer pour le domaine {domain}...") + + env = os.environ.copy() + env[_PASSWORD_ENV_VAR] = password + + try: + process = subprocess.Popen( + ["powershell", "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-Command", "-"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + env=env, + creationflags=subprocess.CREATE_NO_WINDOW, + ) + except OSError as exc: + return JoinResult(success=False, message=f"Impossible de lancer PowerShell : {exc}") + finally: + env[_PASSWORD_ENV_VAR] = "" + + try: + process.stdin.write(script) + process.stdin.close() + except BrokenPipeError: + pass + + output_lines: list[str] = [] + if process.stdout: + for line in process.stdout: + line = line.rstrip("\n") + output_lines.append(line) + emit(line) + + returncode = process.wait() + full_output = "\n".join(output_lines) + + if returncode == 0 and _SUCCESS_MARKER in full_output: + return JoinResult( + success=True, + message=f"Le poste a rejoint le domaine {domain} avec succès. Un redémarrage est requis.", + needs_reboot=True, + ) + + failure_detail = next((l for l in output_lines if l.strip().startswith(_FAILURE_PREFIX)), None) + message = failure_detail or f"Échec de la jointure au domaine (code retour {returncode})." + return JoinResult(success=False, message=message) diff --git a/domainjoin/cli.py b/domainjoin/cli.py new file mode 100644 index 0000000..b658fba --- /dev/null +++ b/domainjoin/cli.py @@ -0,0 +1,223 @@ +"""Interface en ligne de commande de JoinDomainGUI. + +N'importe volontairement pas PySide6 : la CLI doit pouvoir tourner sur un +poste sans environnement graphique (déploiement scripté, SSH, etc.). +""" + +from __future__ import annotations + +import argparse +import getpass +import os +import sys +from typing import Sequence + +from . import __version__, diagnostics, privileges +from .backend import get_backend + +DEFAULT_DOMAIN = "h3adm.lan" + +_STATUS_PREFIX = { + diagnostics.Status.OK: "[OK] ", + diagnostics.Status.WARN: "[!!] ", + diagnostics.Status.FAIL: "[XX] ", +} + + +class _Color: + RESET = "\x1b[0m" + BOLD = "\x1b[1m" + DIM = "\x1b[2m" + RED = "\x1b[31m" + GREEN = "\x1b[32m" + YELLOW = "\x1b[33m" + CYAN = "\x1b[36m" + + +_STATUS_COLOR = { + diagnostics.Status.OK: _Color.GREEN, + diagnostics.Status.WARN: _Color.YELLOW, + diagnostics.Status.FAIL: _Color.RED, +} + + +def _enable_windows_ansi() -> None: + """Active l'interprétation des séquences ANSI dans les consoles Windows classiques.""" + if sys.platform != "win32": + return + try: + import ctypes + + kernel32 = ctypes.windll.kernel32 + enable_virtual_terminal_processing = 0x0004 + for std_handle in (-11, -12): # STD_OUTPUT_HANDLE, STD_ERROR_HANDLE + handle = kernel32.GetStdHandle(std_handle) + mode = ctypes.c_uint32() + if kernel32.GetConsoleMode(handle, ctypes.byref(mode)): + kernel32.SetConsoleMode(handle, mode.value | enable_virtual_terminal_processing) + except Exception: + pass + + +def _color_enabled(stream) -> bool: + if os.environ.get("NO_COLOR") is not None: + return False + if os.environ.get("FORCE_COLOR") is not None: + return True + try: + return stream.isatty() + except Exception: + return False + + +def _colorize(text: str, color: str, stream) -> str: + if not _color_enabled(stream): + return text + return f"{color}{text}{_Color.RESET}" + + +def _print_check(result: diagnostics.CheckResult) -> None: + line = f"{_STATUS_PREFIX[result.status]}{result.name} — {result.detail}" + print(_colorize(line, _STATUS_COLOR[result.status], sys.stdout)) + + +def _confirm(prompt: str, assume_yes: bool) -> bool: + if assume_yes: + return True + reply = input(f"{prompt} [o/N] : ").strip().lower() + return reply in ("o", "oui", "y", "yes") + + +def _resolve_password(args: argparse.Namespace) -> str: + if args.password_stdin: + return sys.stdin.readline().rstrip("\n") + if args.password: + print( + "Attention : --password expose le mot de passe dans l'historique du shell " + "et la liste des processus. Préférez --password-stdin.", + file=sys.stderr, + ) + return args.password + return getpass.getpass("Mot de passe : ") + + +def _print_error(message: str) -> None: + print(_colorize(f"Erreur : {message}", _Color.RED, sys.stderr), file=sys.stderr) + + +def _print_result(result) -> None: + color = _Color.GREEN if result.success else _Color.RED + print(_colorize(result.message, color, sys.stdout)) + + +def _require_elevation() -> int | None: + if privileges.is_elevated(): + return None + _print_error(privileges.elevation_hint()) + return 2 + + +def _cmd_diagnose(args: argparse.Namespace) -> int: + results = list(diagnostics.run_diagnostics(args.domain)) + for result in results: + _print_check(result) + + oks = sum(1 for r in results if r.status == diagnostics.Status.OK) + warns = sum(1 for r in results if r.status == diagnostics.Status.WARN) + fails = sum(1 for r in results if r.status == diagnostics.Status.FAIL) + summary = ( + f"\nRésumé : {_colorize(f'{oks} OK', _Color.GREEN, sys.stdout)}, " + f"{_colorize(f'{warns} avertissement(s)', _Color.YELLOW, sys.stdout)}, " + f"{_colorize(f'{fails} échec(s)', _Color.RED, sys.stdout)}." + ) + print(summary) + return 1 if fails else 0 + + +def _cmd_join(args: argparse.Namespace) -> int: + if (code := _require_elevation()) is not None: + return code + + backend = get_backend() + available, reason = backend.is_available() + if not available: + _print_error(reason) + return 2 + + password = _resolve_password(args) + + if not _confirm( + f"Rejoindre le domaine « {args.domain} » avec l'utilisateur « {args.username} » ?", args.yes + ): + print(_colorize("Annulé.", _Color.YELLOW, sys.stdout)) + return 1 + + result = backend.join( + args.domain, + args.username, + password, + ou=args.ou, + computer_name=args.computer_name, + on_output=print, + ) + _print_result(result) + if result.success and result.needs_reboot: + print(_colorize("Un redémarrage est nécessaire pour finaliser la jointure.", _Color.YELLOW, sys.stdout)) + return 0 if result.success else 1 + + +def _cmd_install_prereqs(args: argparse.Namespace) -> int: + if (code := _require_elevation()) is not None: + return code + + backend = get_backend() + if not backend.can_install_prerequisites(): + _print_error("Ce backend ne prend pas en charge l'installation automatique des prérequis.") + return 2 + + if not _confirm("Installer les prérequis système (realmd, sssd, adcli, samba-common-bin, packagekit) ?", args.yes): + print(_colorize("Annulé.", _Color.YELLOW, sys.stdout)) + return 1 + + result = backend.install_prerequisites(on_output=print) + _print_result(result) + return 0 if result.success else 1 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="JoinDomainGUI", + description="Diagnostic réseau et jointure de domaine Active Directory (Linux/Windows).", + ) + parser.add_argument("--version", action="version", version=f"JoinDomainGUI {__version__}") + + sub = parser.add_subparsers(dest="command", required=True) + + p_diag = sub.add_parser("diagnose", help="Lance le diagnostic réseau AD.") + p_diag.add_argument("--domain", default=DEFAULT_DOMAIN, help=f"Domaine à diagnostiquer (défaut : {DEFAULT_DOMAIN}).") + p_diag.set_defaults(func=_cmd_diagnose) + + p_join = sub.add_parser("join", help="Rejoint le domaine (nécessite les droits root/administrateur).") + p_join.add_argument("--domain", default=DEFAULT_DOMAIN, help=f"Domaine à rejoindre (défaut : {DEFAULT_DOMAIN}).") + p_join.add_argument("--username", required=True, help="Compte autorisé à joindre le domaine.") + p_join.add_argument("--password", help="Mot de passe (déconseillé : visible dans l'historique/ps). Préférez --password-stdin.") + p_join.add_argument("--password-stdin", action="store_true", help="Lit le mot de passe depuis l'entrée standard.") + p_join.add_argument("--ou", default=None, help="Unité d'organisation cible (optionnel).") + p_join.add_argument("--computer-name", default=None, help="Nom de machine (optionnel).") + p_join.add_argument("-y", "--yes", action="store_true", help="Ne pas demander de confirmation.") + p_join.set_defaults(func=_cmd_join) + + p_install = sub.add_parser( + "install-prereqs", help="Installe les outils système requis pour la jointure (Linux/apt uniquement)." + ) + p_install.add_argument("-y", "--yes", action="store_true", help="Ne pas demander de confirmation.") + p_install.set_defaults(func=_cmd_install_prereqs) + + return parser + + +def run_cli(argv: Sequence[str]) -> int: + _enable_windows_ansi() + parser = build_parser() + args = parser.parse_args(argv) + return args.func(args) diff --git a/domainjoin/diagnostics.py b/domainjoin/diagnostics.py new file mode 100644 index 0000000..8ff68a0 --- /dev/null +++ b/domainjoin/diagnostics.py @@ -0,0 +1,247 @@ +"""Moteur de diagnostic réseau pour une jointure à un domaine Active Directory. + +Chaque vérification est indépendante de la plateforme et retourne un CheckResult. +`run_diagnostics()` est un générateur : il yield chaque résultat au fur et à +mesure, pour permettre à l'appelant (thread worker) de mettre à jour l'UI en +continu plutôt que d'attendre la fin de tous les checks. +""" + +from __future__ import annotations + +import platform +import shutil +import socket +import subprocess +import time +from dataclasses import dataclass, field +from enum import Enum + +try: + import dns.resolver + import dns.exception + + _HAS_DNSPYTHON = True +except ImportError: + _HAS_DNSPYTHON = False + +try: + import ntplib + + _HAS_NTPLIB = True +except ImportError: + _HAS_NTPLIB = False + + +class Status(Enum): + OK = "OK" + WARN = "AVERTISSEMENT" + FAIL = "ÉCHEC" + + +@dataclass +class CheckResult: + name: str + status: Status + detail: str + + +# Ports requis pour rejoindre/fonctionner dans un domaine AD. +AD_PORTS = { + 53: "DNS", + 88: "Kerberos", + 389: "LDAP", + 445: "SMB", + 464: "Kerberos kpasswd", + 636: "LDAPS", + 3268: "LDAP Global Catalog", +} + +SRV_RECORDS = [ + "_ldap._tcp.{domain}", + "_kerberos._tcp.{domain}", + "_kerberos._tcp.dc._msdcs.{domain}", +] + +REQUIRED_LINUX_TOOLS = ["realm", "adcli", "net"] + + +def _check_hostname() -> CheckResult: + host = socket.gethostname() + fqdn = socket.getfqdn() + if fqdn and fqdn != host and "." in fqdn: + return CheckResult("Nom d'hôte / FQDN", Status.OK, f"hostname={host}, FQDN={fqdn}") + return CheckResult( + "Nom d'hôte / FQDN", + Status.WARN, + f"hostname={host}, FQDN non résolu ({fqdn}) — la résolution DNS locale peut être incomplète.", + ) + + +def _resolve_domain_ips(domain: str) -> list[str]: + try: + infos = socket.getaddrinfo(domain, None) + return sorted({info[4][0] for info in infos}) + except socket.gaierror: + return [] + + +def _check_dns(domain: str) -> tuple[CheckResult, list[str]]: + ips = _resolve_domain_ips(domain) + if ips: + return CheckResult("Résolution DNS du domaine", Status.OK, f"{domain} → {', '.join(ips)}"), ips + return ( + CheckResult( + "Résolution DNS du domaine", + Status.FAIL, + f"Impossible de résoudre {domain}. Vérifiez que le DNS du poste pointe vers un contrôleur de domaine.", + ), + [], + ) + + +def _check_srv_records(domain: str) -> CheckResult: + if not _HAS_DNSPYTHON: + return CheckResult( + "Enregistrements SRV (LDAP/Kerberos)", + Status.WARN, + "Le module dnspython n'est pas installé — vérification ignorée.", + ) + found, missing = [], [] + for template in SRV_RECORDS: + qname = template.format(domain=domain) + try: + dns.resolver.resolve(qname, "SRV", lifetime=3.0) + found.append(qname) + except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.exception.Timeout, dns.resolver.NoNameservers): + missing.append(qname) + if not missing: + return CheckResult("Enregistrements SRV (LDAP/Kerberos)", Status.OK, f"{len(found)}/{len(found)} trouvés") + if found: + return CheckResult( + "Enregistrements SRV (LDAP/Kerberos)", + Status.WARN, + f"Manquants : {', '.join(missing)}", + ) + return CheckResult( + "Enregistrements SRV (LDAP/Kerberos)", + Status.FAIL, + "Aucun enregistrement SRV trouvé — le DNS ne semble pas être celui du domaine AD.", + ) + + +def _ping(host: str, timeout_s: int = 2) -> bool: + is_windows = platform.system() == "Windows" + if is_windows: + cmd = ["ping", "-n", "1", "-w", str(timeout_s * 1000), host] + else: + cmd = ["ping", "-c", "1", "-W", str(timeout_s), host] + try: + result = subprocess.run( + cmd, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=timeout_s + 2, + creationflags=subprocess.CREATE_NO_WINDOW if is_windows else 0, + ) + return result.returncode == 0 + except (subprocess.TimeoutExpired, OSError): + return False + + +def _check_ping(domain: str, ips: list[str]) -> CheckResult: + targets = ips or [domain] + reachable = [ip for ip in targets if _ping(ip)] + if reachable: + return CheckResult("Ping du/des contrôleur(s) de domaine", Status.OK, f"Répond : {', '.join(reachable)}") + return CheckResult( + "Ping du/des contrôleur(s) de domaine", + Status.WARN, + "Aucune réponse au ping (peut être normal si ICMP est filtré par un pare-feu).", + ) + + +def _check_ports(ips: list[str]) -> CheckResult: + if not ips: + return CheckResult("Ports AD (53/88/389/445/464/636/3268)", Status.FAIL, "Pas d'IP de contrôleur à tester.") + ip = ips[0] + open_ports, closed_ports = [], [] + for port, name in AD_PORTS.items(): + try: + with socket.create_connection((ip, port), timeout=2): + open_ports.append(f"{port}/{name}") + except OSError: + closed_ports.append(f"{port}/{name}") + if not closed_ports: + return CheckResult("Ports AD (53/88/389/445/464/636/3268)", Status.OK, f"Tous ouverts sur {ip}") + if open_ports: + return CheckResult( + "Ports AD (53/88/389/445/464/636/3268)", + Status.WARN, + f"Fermés/filtrés sur {ip} : {', '.join(closed_ports)}", + ) + return CheckResult( + "Ports AD (53/88/389/445/464/636/3268)", + Status.FAIL, + f"Aucun port AD accessible sur {ip} — connectivité réseau ou pare-feu à vérifier.", + ) + + +def _check_ntp(domain: str, ips: list[str]) -> CheckResult: + if not _HAS_NTPLIB: + return CheckResult( + "Synchronisation horaire (NTP)", + Status.WARN, + "Le module ntplib n'est pas installé — vérification ignorée.", + ) + targets = ips or [domain] + client = ntplib.NTPClient() + for target in targets: + try: + response = client.request(target, port=123, version=3, timeout=2) + drift = abs(response.offset) + if drift > 300: + return CheckResult( + "Synchronisation horaire (NTP)", + Status.FAIL, + f"Dérive de {drift:.1f}s avec {target} — Kerberos échouera (limite : 300s).", + ) + if drift > 30: + return CheckResult( + "Synchronisation horaire (NTP)", Status.WARN, f"Dérive de {drift:.1f}s avec {target}." + ) + return CheckResult("Synchronisation horaire (NTP)", Status.OK, f"Dérive de {drift:.1f}s avec {target}.") + except Exception: + continue + return CheckResult( + "Synchronisation horaire (NTP)", + Status.WARN, + "Aucun contrôleur n'a répondu au NTP (port 123) — vérification impossible.", + ) + + +def _check_linux_tools() -> CheckResult: + missing = [tool for tool in REQUIRED_LINUX_TOOLS if shutil.which(tool) is None] + if not missing: + return CheckResult("Outils requis (realm/adcli/net)", Status.OK, "Tous les outils sont installés.") + return CheckResult( + "Outils requis (realm/adcli/net)", + Status.FAIL, + "Manquants : " + ", ".join(missing) + ". Installez-les avec : " + "sudo apt install realmd sssd adcli samba-common-bin packagekit", + ) + + +def run_diagnostics(domain: str): + """Générateur qui exécute tous les checks et yield chaque CheckResult au fur et à mesure.""" + yield _check_hostname() + + dns_result, ips = _check_dns(domain) + yield dns_result + + yield _check_srv_records(domain) + yield _check_ping(domain, ips) + yield _check_ports(ips) + yield _check_ntp(domain, ips) + + if platform.system() != "Windows": + yield _check_linux_tools() diff --git a/domainjoin/gui.py b/domainjoin/gui.py new file mode 100644 index 0000000..b3a29a5 --- /dev/null +++ b/domainjoin/gui.py @@ -0,0 +1,337 @@ +"""Interface graphique PySide6 : diagnostic réseau AD + jointure de domaine.""" + +from __future__ import annotations + +import platform +import socket + +from PySide6.QtCore import Qt +from PySide6.QtGui import QColor +from PySide6.QtWidgets import ( + QApplication, + QFormLayout, + QHBoxLayout, + QLabel, + QLineEdit, + QMainWindow, + QMessageBox, + QPlainTextEdit, + QPushButton, + QTabWidget, + QTreeWidget, + QTreeWidgetItem, + QVBoxLayout, + QWidget, +) + +from . import __version__, diagnostics, privileges +from .backend import get_backend +from .worker import DiagnosticsWorker, InstallPrerequisitesWorker, JoinWorker + +DEFAULT_DOMAIN = "h3adm.lan" + +_STATUS_COLORS = { + diagnostics.Status.OK: QColor("#1e7e34"), + diagnostics.Status.WARN: QColor("#b8860b"), + diagnostics.Status.FAIL: QColor("#c0392b"), +} + + +class DiagnosticsTab(QWidget): + def __init__(self, parent=None): + super().__init__(parent) + self._worker: DiagnosticsWorker | None = None + self._results: list[diagnostics.CheckResult] = [] + + layout = QVBoxLayout(self) + + form = QFormLayout() + self.domain_edit = QLineEdit(DEFAULT_DOMAIN) + form.addRow("Domaine :", self.domain_edit) + layout.addLayout(form) + + buttons = QHBoxLayout() + self.run_button = QPushButton("Lancer le diagnostic") + self.run_button.clicked.connect(self._start_diagnostics) + buttons.addWidget(self.run_button) + self.copy_button = QPushButton("Copier le rapport") + self.copy_button.clicked.connect(self._copy_report) + self.copy_button.setEnabled(False) + buttons.addWidget(self.copy_button) + buttons.addStretch() + layout.addLayout(buttons) + + self.tree = QTreeWidget() + self.tree.setHeaderLabels(["Vérification", "Statut", "Détail"]) + self.tree.setColumnWidth(0, 260) + self.tree.setColumnWidth(1, 130) + layout.addWidget(self.tree) + + self.summary_label = QLabel("") + layout.addWidget(self.summary_label) + + def _start_diagnostics(self) -> None: + domain = self.domain_edit.text().strip() + if not domain: + QMessageBox.warning(self, "Domaine manquant", "Veuillez saisir un domaine à diagnostiquer.") + return + + self.tree.clear() + self._results = [] + self.run_button.setEnabled(False) + self.copy_button.setEnabled(False) + self.summary_label.setText("Diagnostic en cours...") + + self._worker = DiagnosticsWorker(domain) + self._worker.check_ready.connect(self._on_check_ready) + self._worker.finished_all.connect(self._on_finished) + self._worker.start() + + def _on_check_ready(self, result: diagnostics.CheckResult) -> None: + self._results.append(result) + item = QTreeWidgetItem([result.name, result.status.value, result.detail]) + color = _STATUS_COLORS.get(result.status) + if color: + item.setForeground(1, color) + self.tree.addTopLevelItem(item) + + def _on_finished(self) -> None: + self.run_button.setEnabled(True) + self.copy_button.setEnabled(True) + ok = sum(1 for r in self._results if r.status == diagnostics.Status.OK) + warn = sum(1 for r in self._results if r.status == diagnostics.Status.WARN) + fail = sum(1 for r in self._results if r.status == diagnostics.Status.FAIL) + self.summary_label.setText(f"Terminé : {ok} OK, {warn} avertissement(s), {fail} échec(s).") + + def _copy_report(self) -> None: + lines = [f"Diagnostic réseau pour le domaine {self.domain_edit.text().strip()}", ""] + for r in self._results: + lines.append(f"[{r.status.value}] {r.name} — {r.detail}") + QApplication.clipboard().setText("\n".join(lines)) + + +class JoinTab(QWidget): + def __init__(self, parent=None): + super().__init__(parent) + self._worker: JoinWorker | None = None + self.backend = get_backend() + + layout = QVBoxLayout(self) + + form = QFormLayout() + self.domain_edit = QLineEdit(DEFAULT_DOMAIN) + self.username_edit = QLineEdit() + self.username_edit.setPlaceholderText("ex : administrateur@h3adm.lan") + self.password_edit = QLineEdit() + self.password_edit.setEchoMode(QLineEdit.EchoMode.Password) + self.ou_edit = QLineEdit() + self.ou_edit.setPlaceholderText("optionnel, ex : OU=Postes,DC=h3adm,DC=lan") + self.computer_name_edit = QLineEdit(socket.gethostname()) + + form.addRow("Domaine :", self.domain_edit) + form.addRow("Utilisateur :", self.username_edit) + form.addRow("Mot de passe :", self.password_edit) + form.addRow("Unité d'organisation :", self.ou_edit) + form.addRow("Nom de la machine :", self.computer_name_edit) + layout.addLayout(form) + + self.prereq_label = QLabel("") + self.prereq_label.setWordWrap(True) + self.prereq_label.setVisible(False) + layout.addWidget(self.prereq_label) + + buttons = QHBoxLayout() + self.join_button = QPushButton("Rejoindre le domaine") + self.join_button.clicked.connect(self._start_join) + buttons.addWidget(self.join_button) + self.install_button = QPushButton("Installer les prérequis") + self.install_button.clicked.connect(self._start_install) + self.install_button.setVisible(False) + buttons.addWidget(self.install_button) + buttons.addStretch() + layout.addLayout(buttons) + + self.log = QPlainTextEdit() + self.log.setReadOnly(True) + layout.addWidget(self.log) + + self._refresh_backend_state() + + def _refresh_backend_state(self) -> None: + """Met à jour l'état des boutons selon les privilèges et la disponibilité du backend.""" + elevated = privileges.is_elevated() + available, reason = self.backend.is_available() + + self.install_button.setVisible(not available and self.backend.can_install_prerequisites()) + self.install_button.setEnabled(elevated) + if not elevated: + self.install_button.setToolTip( + "Droits insuffisants pour installer les prérequis.\n" + privileges.elevation_hint() + ) + else: + self.install_button.setToolTip("") + + self.prereq_label.setVisible(not available) + if not available: + self.prereq_label.setText("⚠ " + reason) + + if not elevated: + self.join_button.setEnabled(False) + self.join_button.setToolTip( + "Droits insuffisants pour rejoindre le domaine.\n" + privileges.elevation_hint() + ) + elif not available: + self.join_button.setEnabled(False) + self.join_button.setToolTip("Prérequis manquants — installez-les d'abord.") + else: + self.join_button.setEnabled(True) + self.join_button.setToolTip("") + + def _start_install(self) -> None: + confirm = QMessageBox.question( + self, + "Confirmer l'installation", + "Installer les paquets système requis pour la jointure de domaine " + "(realmd, sssd, adcli, samba-common-bin, packagekit) ?", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No, + ) + if confirm != QMessageBox.StandardButton.Yes: + return + + self.log.clear() + self.install_button.setEnabled(False) + self.join_button.setEnabled(False) + + self._install_worker = InstallPrerequisitesWorker(self.backend) + self._install_worker.output_line.connect(self.log.appendPlainText) + self._install_worker.finished_install.connect(self._on_install_finished) + self._install_worker.start() + + def _on_install_finished(self, result) -> None: + if result.success: + QMessageBox.information(self, "Installation terminée", result.message) + else: + QMessageBox.critical(self, "Échec de l'installation", result.message) + self._refresh_backend_state() + + def _start_join(self) -> None: + domain = self.domain_edit.text().strip() + username = self.username_edit.text().strip() + password = self.password_edit.text() + ou = self.ou_edit.text().strip() or None + computer_name = self.computer_name_edit.text().strip() or None + + if not domain or not username or not password: + QMessageBox.warning(self, "Champs manquants", "Domaine, utilisateur et mot de passe sont requis.") + return + + available, reason = self.backend.is_available() + if not available: + QMessageBox.critical(self, "Backend indisponible", reason) + return + + confirm = QMessageBox.question( + self, + "Confirmer la jointure", + f"Rejoindre le domaine « {domain} » avec l'utilisateur « {username} » ?\n\n" + "Cette action modifie la configuration système du poste et nécessitera " + "probablement un redémarrage.", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No, + ) + if confirm != QMessageBox.StandardButton.Yes: + return + + self.log.clear() + self.join_button.setEnabled(False) + + self._worker = JoinWorker(self.backend, domain, username, password, ou, computer_name) + self._worker.output_line.connect(self.log.appendPlainText) + self._worker.finished_join.connect(self._on_join_finished) + self._worker.start() + + def _on_join_finished(self, result) -> None: + self._refresh_backend_state() + if result.success: + msg = result.message + if result.needs_reboot: + msg += "\n\nUn redémarrage est nécessaire pour finaliser la jointure." + QMessageBox.information(self, "Jointure réussie", msg) + else: + QMessageBox.critical(self, "Échec de la jointure", result.message) + + +class AboutTab(QWidget): + def __init__(self, parent=None): + super().__init__(parent) + layout = QVBoxLayout(self) + layout.setAlignment(Qt.AlignmentFlag.AlignTop) + + title = QLabel("JoinDomainGUI") + title.setStyleSheet("font-size: 18pt; font-weight: bold;") + layout.addWidget(title) + + version = QLabel(f"Version {__version__}") + layout.addWidget(version) + + description = QLabel( + "Diagnostic réseau et jointure de domaine Active Directory, " + "pour postes Linux (realmd) et Windows (Add-Computer)." + ) + description.setWordWrap(True) + layout.addWidget(description) + + backend = get_backend() + elevated = "élevé (root/administrateur)" if privileges.is_elevated() else "standard (non élevé)" + state = QLabel( + f"Plateforme détectée : {platform.system()} {platform.release()}\n" + f"Backend de jointure : {type(backend).__name__}\n" + f"Droits actuels : {elevated}" + ) + layout.addWidget(state) + + cli_title = QLabel("Utilisation en ligne de commande") + cli_title.setStyleSheet("font-weight: bold; margin-top: 12px;") + layout.addWidget(cli_title) + + cli_hint = QPlainTextEdit( + "python main.py diagnose --domain h3adm.lan\n" + "python main.py join --domain h3adm.lan --username administrateur --password-stdin\n" + "python main.py install-prereqs\n" + "python main.py --help" + ) + cli_hint.setReadOnly(True) + cli_hint.setMaximumHeight(100) + layout.addWidget(cli_hint) + + layout.addStretch() + + +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle("Diagnostic réseau & Jointure de domaine") + self.resize(760, 620) + + central = QWidget() + self.setCentralWidget(central) + layout = QVBoxLayout(central) + + if not privileges.is_elevated(): + banner = QLabel( + "⚠ L'application ne tourne pas avec les droits administrateur/root. " + "Le diagnostic fonctionne, mais la jointure au domaine sera indisponible. " + + privileges.elevation_hint() + ) + banner.setWordWrap(True) + banner.setStyleSheet( + "background-color:#fff3cd; color:#664d03; padding:8px; border:1px solid #ffe69c;" + ) + layout.addWidget(banner) + + tabs = QTabWidget() + tabs.addTab(DiagnosticsTab(), "Diagnostic réseau") + tabs.addTab(JoinTab(), "Rejoindre le domaine") + tabs.addTab(AboutTab(), "À propos") + layout.addWidget(tabs) diff --git a/domainjoin/privileges.py b/domainjoin/privileges.py new file mode 100644 index 0000000..e25499b --- /dev/null +++ b/domainjoin/privileges.py @@ -0,0 +1,22 @@ +"""Détection des privilèges d'exécution (root/Administrateur), indépendante de l'OS.""" + +import ctypes +import os +import platform + + +def is_elevated() -> bool: + """True si le processus courant tourne avec les droits root (Linux) ou Administrateur (Windows).""" + if platform.system() == "Windows": + try: + return bool(ctypes.windll.shell32.IsUserAnAdmin()) + except Exception: + return False + return hasattr(os, "geteuid") and os.geteuid() == 0 + + +def elevation_hint() -> str: + """Message à afficher à l'utilisateur pour élever l'application.""" + if platform.system() == "Windows": + return "Relancez l'application via un clic droit → « Exécuter en tant qu'administrateur »." + return "Relancez l'application avec les droits root, par exemple : sudo python3 main.py" diff --git a/domainjoin/worker.py b/domainjoin/worker.py new file mode 100644 index 0000000..020af3f --- /dev/null +++ b/domainjoin/worker.py @@ -0,0 +1,72 @@ +"""Workers Qt exécutant diagnostics et jointure de domaine en arrière-plan, +pour ne jamais bloquer l'interface graphique.""" + +from __future__ import annotations + +from typing import Optional + +from PySide6.QtCore import QThread, Signal + +from . import diagnostics +from .backend.base import DomainJoinBackend, JoinResult + + +class DiagnosticsWorker(QThread): + check_ready = Signal(object) # diagnostics.CheckResult + finished_all = Signal() + + def __init__(self, domain: str, parent=None): + super().__init__(parent) + self.domain = domain + + def run(self) -> None: + for result in diagnostics.run_diagnostics(self.domain): + self.check_ready.emit(result) + self.finished_all.emit() + + +class JoinWorker(QThread): + output_line = Signal(str) + finished_join = Signal(object) # backend.base.JoinResult + + def __init__( + self, + backend: DomainJoinBackend, + domain: str, + username: str, + password: str, + ou: Optional[str], + computer_name: Optional[str], + parent=None, + ): + super().__init__(parent) + self.backend = backend + self.domain = domain + self.username = username + self.password = password + self.ou = ou + self.computer_name = computer_name + + def run(self) -> None: + result: JoinResult = self.backend.join( + self.domain, + self.username, + self.password, + ou=self.ou, + computer_name=self.computer_name, + on_output=self.output_line.emit, + ) + self.finished_join.emit(result) + + +class InstallPrerequisitesWorker(QThread): + output_line = Signal(str) + finished_install = Signal(object) # backend.base.JoinResult + + def __init__(self, backend: DomainJoinBackend, parent=None): + super().__init__(parent) + self.backend = backend + + def run(self) -> None: + result: JoinResult = self.backend.install_prerequisites(on_output=self.output_line.emit) + self.finished_install.emit(result) diff --git a/main.py b/main.py new file mode 100644 index 0000000..00088cd --- /dev/null +++ b/main.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +"""Point d'entrée : lance la GUI si aucun argument n'est fourni, sinon la CLI. + +Exemples CLI : + python main.py diagnose --domain h3adm.lan + python main.py join --domain h3adm.lan --username administrateur --password-stdin + python main.py install-prereqs + python main.py --help +""" + +import sys + + +def _attach_windows_console() -> None: + """Rattache le binaire packagé (--windowed, sans console) au terminal appelant + afin que la sortie de la CLI reste visible sous Windows.""" + import ctypes + + ATTACH_PARENT_PROCESS = -1 + if ctypes.windll.kernel32.AttachConsole(ATTACH_PARENT_PROCESS): + try: + sys.stdout = open("CONOUT$", "w") + sys.stderr = open("CONOUT$", "w") + except OSError: + pass + + +def main() -> int: + if len(sys.argv) > 1: + if sys.platform == "win32": + _attach_windows_console() + from domainjoin.cli import run_cli + + return run_cli(sys.argv[1:]) + + from PySide6.QtWidgets import QApplication + + from domainjoin.gui import MainWindow + + app = QApplication(sys.argv) + app.setApplicationName("JoinDomainGUI") + window = MainWindow() + window.show() + return app.exec() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packaging/build_linux.sh b/packaging/build_linux.sh new file mode 100755 index 0000000..1c3f18b --- /dev/null +++ b/packaging/build_linux.sh @@ -0,0 +1,25 @@ +#!/usr/bin/env bash +# Fabrique le binaire Linux autonome (ELF) de JoinDomainGUI avec PyInstaller. +# À exécuter SUR LINUX (PyInstaller ne fait pas de cross-compilation). +set -euo pipefail + +cd "$(dirname "$0")/.." + +VENV_DIR=".build-venv-linux" +if [ ! -d "$VENV_DIR" ]; then + python3 -m venv --system-site-packages "$VENV_DIR" +fi + +"$VENV_DIR/bin/pip" install --upgrade pip -q +"$VENV_DIR/bin/pip" install -q -r requirements.txt pyinstaller + +"$VENV_DIR/bin/pyinstaller" \ + --noconfirm \ + --onefile \ + --windowed \ + --name JoinDomainGUI \ + main.py + +echo +echo "Binaire produit : dist/JoinDomainGUI" +echo "Lancement : sudo ./dist/JoinDomainGUI" diff --git a/packaging/build_windows.bat b/packaging/build_windows.bat new file mode 100644 index 0000000..9106f9f --- /dev/null +++ b/packaging/build_windows.bat @@ -0,0 +1,26 @@ +@echo off +REM Fabrique le binaire Windows autonome (.exe) de JoinDomainGUI avec PyInstaller. +REM A executer SUR WINDOWS (PyInstaller ne fait pas de cross-compilation). +setlocal + +cd /d "%~dp0\.." + +set VENV_DIR=.build-venv-windows +if not exist "%VENV_DIR%" ( + python -m venv "%VENV_DIR%" +) + +call "%VENV_DIR%\Scripts\pip.exe" install --upgrade pip -q +call "%VENV_DIR%\Scripts\pip.exe" install -q -r requirements.txt pyinstaller + +"%VENV_DIR%\Scripts\pyinstaller.exe" ^ + --noconfirm ^ + --onefile ^ + --windowed ^ + --name JoinDomainGUI ^ + main.py + +echo. +echo Binaire produit : dist\JoinDomainGUI.exe +echo Lancement : clic droit -^> Executer en tant qu'administrateur +endlocal diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9d1f750 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +PySide6>=6.6 +dnspython>=2.6 +ntplib>=0.4