JoinDomainGUI/domainjoin/backend/windows_addcomputer.py

"""Backend de jointure/sortie de domaine pour Windows, via les cmdlets PowerShell
`Add-Computer` et `Remove-Computer`."""

from __future__ import annotations

import base64
import os
import shutil
import subprocess
from typing import Optional

from .base import DomainJoinBackend, JoinResult, OutputCallback

_PASSWORD_ENV_VAR = "DOMAINJOIN_PWD"
_SUCCESS_MARKER = "DOMAINJOIN_OK"
_FAILURE_PREFIX = "DOMAINJOIN_ERROR:"

# En-tête du flux d'objets sérialisés que PowerShell émet sur sa sortie d'erreur
# lorsqu'elle est redirigée. Ce n'est pas un message pour l'utilisateur.
_CLIXML_HEADER = "#< CLIXML"


def _ps_quote(value: str) -> str:
    """Échappe une valeur pour l'insérer dans une chaîne PowerShell entre quotes simples."""
    return value.replace("'", "''")


def _encode_command(script: str) -> str:
    """Encode un script pour `powershell -EncodedCommand` (base64 d'UTF-16LE).

    Le script est ainsi transmis comme un bloc unique et atomique. L'alternative
    `-Command -` fait analyser l'entrée standard ligne par ligne : la dernière
    instruction d'un bloc `try`/`catch` restait en attente dans le parseur et
    n'était jamais exécutée à la fermeture de stdin — PowerShell rendait alors
    la main avec le code 0 sans avoir rien fait. Le mot de passe reste hors du
    script (variable d'environnement) : la ligne de commande ne l'expose pas.
    """
    return base64.b64encode(script.encode("utf-16-le")).decode("ascii")


def _is_noise(line: str) -> bool:
    """Écarte la sérialisation CLIXML que PowerShell mêle à sa sortie."""
    stripped = line.strip()
    return stripped.startswith(_CLIXML_HEADER) or stripped.startswith("<Objs ")


_PREAMBLE = [
    "$ErrorActionPreference = 'Stop'",
    # Sans cela, les barres de progression (« Préparation des modules… ») sont
    # serialisees en CLIXML sur la sortie d'erreur et polluent le journal.
    "$ProgressPreference = 'SilentlyContinue'",
    # La console Windows sort en codepage OEM (850/437) : sans cette bascule en
    # UTF-8, les accents des messages d'erreur arrivent illisibles cote Python.
    "[Console]::OutputEncoding = [System.Text.Encoding]::UTF8",
]


def _credential_lines(username: str) -> list[str]:
    return [
        f"$username = '{_ps_quote(username)}'",
        f"$password = $env:{_PASSWORD_ENV_VAR}",
        "if ([string]::IsNullOrEmpty($password)) { "
        f"Write-Output '{_FAILURE_PREFIX} mot de passe manquant.'; exit 2 }}",
        "$securePwd = ConvertTo-SecureString -String $password -AsPlainText -Force",
        "$cred = New-Object System.Management.Automation.PSCredential($username, $securePwd)",
    ]


def _guarded(body: list[str]) -> list[str]:
    return [
        "try {",
        *[f"    {line}" for line in body],
        f"    Write-Output '{_SUCCESS_MARKER}'",
        "} catch {",
        f"    Write-Output ('{_FAILURE_PREFIX} ' + $_.Exception.Message)",
        "    exit 1",
        "}",
    ]


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

    # ------------------------------------------------------------------ scripts

    def _join_script(self, domain: str, username: str, ou: Optional[str], computer_name: Optional[str]) -> str:
        lines = [
            *_PREAMBLE,
            f"$domain = '{_ps_quote(domain)}'",
            *_credential_lines(username),
            "$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 += _guarded(["Add-Computer @params -ErrorAction Stop"])
        return "\n".join(lines)

    def _leave_script(self, username: str, workgroup: str) -> str:
        lines = [
            *_PREAMBLE,
            *_credential_lines(username),
            f"$workgroup = '{_ps_quote(workgroup)}'",
        ]
        lines += _guarded(
            [
                "Remove-Computer -UnjoinDomainCredential $cred -WorkgroupName $workgroup "
                "-Force -Restart:$false -ErrorAction Stop | Out-Null"
            ]
        )
        return "\n".join(lines)

    # ------------------------------------------------------------------ execution

    def _run_powershell(
        self, script: str, password: str, emit: OutputCallback
    ) -> tuple[Optional[int], list[str]]:
        """Exécute un script PowerShell et retourne (code retour, lignes utiles).

        Un code retour None signale que PowerShell n'a pas pu être lancé ; le
        motif est alors la première ligne retournée.
        """
        env = os.environ.copy()
        env[_PASSWORD_ENV_VAR] = password

        try:
            process = subprocess.Popen(
                [
                    "powershell",
                    "-NoProfile",
                    "-NonInteractive",
                    "-ExecutionPolicy",
                    "Bypass",
                    "-EncodedCommand",
                    _encode_command(script),
                ],
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                encoding="utf-8",
                errors="replace",
                bufsize=1,
                env=env,
                creationflags=subprocess.CREATE_NO_WINDOW,
            )
        except OSError as exc:
            return None, [f"Impossible de lancer PowerShell : {exc}"]
        finally:
            # Le processus fils a déjà reçu sa copie de l'environnement : on
            # efface le mot de passe de la nôtre sans l'en priver.
            env[_PASSWORD_ENV_VAR] = ""

        lines: list[str] = []
        if process.stdout:
            for raw in process.stdout:
                line = raw.rstrip("\n")
                if _is_noise(line):
                    continue
                lines.append(line)
                emit(line)
        return process.wait(), lines

    def _interpret(
        self,
        returncode: Optional[int],
        lines: list[str],
        success_message: str,
        needs_reboot: bool,
    ) -> JoinResult:
        if returncode is None:
            return JoinResult(success=False, message=lines[0])

        if returncode == 0 and any(line.strip() == _SUCCESS_MARKER for line in lines):
            return JoinResult(success=True, message=success_message, needs_reboot=needs_reboot)

        detail = next((l for l in lines if l.strip().startswith(_FAILURE_PREFIX)), None)
        if detail:
            # Le marqueur est un détail d'implémentation : l'utilisateur ne voit
            # que le message d'erreur de PowerShell.
            return JoinResult(success=False, message=detail.strip()[len(_FAILURE_PREFIX) :].strip())

        # Sans ce report de la sortie brute, un echec inattendu se resumerait a
        # « code retour 0 », indiagnosticable. Le mot de passe n'y apparait
        # jamais : il ne transite que par l'environnement du sous-processus.
        output = "\n".join(l for l in lines if l.strip() and l.strip() != _SUCCESS_MARKER)
        return JoinResult(
            success=False,
            message=f"Échec (code retour {returncode}).\nSortie de PowerShell : "
            f"{output.strip() or 'aucune sortie produite par PowerShell'}",
        )

    # ------------------------------------------------------------------ API

    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)

        emit: OutputCallback = on_output or (lambda _line: None)
        emit(f"Exécution de Add-Computer pour le domaine {domain}...")

        returncode, lines = self._run_powershell(
            self._join_script(domain, username, ou, computer_name), password, emit
        )
        return self._interpret(
            returncode,
            lines,
            f"Le poste a rejoint le domaine {domain} avec succès. Un redémarrage est requis.",
            needs_reboot=True,
        )

    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)

        emit: OutputCallback = on_output or (lambda _line: None)
        if remove_packages:
            emit(
                "Note : la désinstallation de paquets ne concerne que Linux. "
                "Aucun composant Windows n'est retiré."
            )
        emit(f"Exécution de Remove-Computer pour quitter le domaine {domain}...")

        returncode, lines = self._run_powershell(
            self._leave_script(username, "WORKGROUP"), password, emit
        )
        return self._interpret(
            returncode,
            lines,
            f"Le poste a quitté le domaine {domain} et rejoint le groupe de travail "
            "WORKGROUP. Un redémarrage est requis.",
            needs_reboot=True,
        )