Version initiale : diagnostic réseau AD + jointure de domaine
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 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1 @@
|
||||
__version__ = "0.1.0"
|
||||
@@ -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"]
|
||||
@@ -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.")
|
||||
@@ -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.",
|
||||
)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
@@ -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)
|
||||
@@ -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"
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user