"""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) # Paquets retires lors d'une sortie de domaine avec nettoyage. `packagekit` est # volontairement exclu de cette liste : il est installe comme prerequis mais sert # aussi au gestionnaire de logiciels du bureau, et le retirer casserait des # fonctions sans rapport avec le domaine. REMOVABLE_PACKAGES = ["realmd", "sssd", "adcli", "samba-common-bin"] 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 _run_apt(self, commands: list[list[str]], emit: OutputCallback) -> Optional[JoinResult]: """Enchaîne des commandes apt-get. Retourne None si tout a réussi, sinon le JoinResult d'échec décrivant la commande fautive.""" env = os.environ.copy() env["DEBIAN_FRONTEND"] = "noninteractive" for cmd in commands: 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.", ) return 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.") failure = self._run_apt( [["apt-get", "update"], ["apt-get", "install", "-y", *REQUIRED_PACKAGES]], emit ) if failure: return failure 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.", ) def can_leave(self) -> bool: return True def leave( self, domain: str, username: str, password: str, remove_packages: bool = False, 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) # `realm leave` sait travailler sans identifiants (il se sert du compte # machine local). Fournir un compte permet en plus de supprimer l'objet # ordinateur dans l'annuaire ; on ne l'impose donc pas. cmd = ["realm", "-v", "leave"] if username: cmd += ["--user", username] cmd.append(domain) emit(f"$ {' '.join(cmd)}") 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: if username: process.stdin.write(password + "\n") process.stdin.close() except BrokenPipeError: pass if process.stdout: for line in process.stdout: emit(line.rstrip("\n")) if process.wait() != 0: return JoinResult( success=False, message=f"Échec de la sortie du domaine {domain}. Voir le journal ci-dessus.", ) message = f"Le poste a quitté le domaine {domain}." if remove_packages: if shutil.which("apt-get") is None: message += " Les paquets n'ont pas été retirés : apt-get est introuvable." else: emit("Retrait des paquets devenus inutiles...") failure = self._run_apt( [ ["apt-get", "remove", "--purge", "-y", *REMOVABLE_PACKAGES], ["apt-get", "autoremove", "--purge", "-y"], ], emit, ) if failure: # La sortie du domaine, elle, a bien eu lieu : on le dit # plutot que de faire passer l'ensemble pour un echec. return JoinResult( success=True, message=f"{message} En revanche le nettoyage des paquets a échoué : " f"{failure.message}", needs_reboot=True, ) message += " Paquets retirés : " + ", ".join(REMOVABLE_PACKAGES) + "." return JoinResult(success=True, message=message, needs_reboot=True)