JoinDomainGUI/domainjoin/backend/windows_addcomputer.py

"""Backend de jointure de domaine pour Windows, via la cmdlet PowerShell Add-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 = "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("'", "''")


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, et la jointure echouait sans
    le moindre message. 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")


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",
                    "-EncodedCommand",
                    _encode_command(script),
                ],
                stdin=subprocess.DEVNULL,
                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] = ""

        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)
        if failure_detail:
            message = failure_detail
        else:
            # Sans ce report de la sortie brute, un echec inattendu se resumait a
            # « code retour 0 », indiagnosticable. Le mot de passe n'apparait
            # jamais dans cette sortie : il ne transite que par l'environnement.
            detail = full_output.strip() or "aucune sortie produite par PowerShell"
            message = (
                f"Échec de la jointure au domaine (code retour {returncode}).\n"
                f"Sortie de PowerShell : {detail}"
            )
        return JoinResult(success=False, message=message)