JoinDomainGUI/domainjoin/backend/linux_realmd.py

"""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.",
        )