JoinDomainGUI/domainjoin/backend/windows_addcomputer.py

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