erplibre/script/execute/execute.py

#!/usr/bin/env python3
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)

# Annotations différées : la migration charge ce module sous le Python
# d'Odoo 12 — 3.7 — où « dict | None » et « tuple[str, str] » n'existent
# pas encore. Sans ceci, l'annotation est ÉVALUÉE au chargement et la
# migration meurt sur un TypeError avant d'avoir rien fait.
from __future__ import annotations

import codecs
import datetime
import logging
import os
import re
import shutil
import subprocess
import sys
import time

try:
    import humanize
except ModuleNotFoundError as e:
    humanize = None

VENV_ERPLIBRE = ".venv.erplibre"

# Une commande construite ailleurs peut porter un secret en clair : todo.py et
# kdbx_manager.py y mettent « --default_password_auth '<mot de passe KeePass>' ».
# Cette commande est affichée avant et après l'exécution, et journalisée en
# erreur : le secret finissait donc dans le terminal, dans les journaux et dans
# toute sortie CI qui les capture.
#
# Ce filtre reste le dernier rempart, pas le premier : un secret n'a rien à
# faire sur argv, que /proc/<pid>/cmdline expose à tout utilisateur de la
# machine et qu'aucun caviardage n'atteint. db_restore.py est passé à
# MASTER_PWD dans l'environnement pour cette raison.
#
# On caviarde la VALEUR, jamais le nom de l'option : la commande reste lisible et
# reproductible, il ne manque que ce qui ne doit pas être lu.
_SECRET_OPTION = re.compile(
    r"(?P<opt>--?[\w-]*"
    r"(?:password|passwd|pwd|secret|token|api[-_]?key)[\w-]*"
    r"(?:\s+|=))"
    r"(?P<val>'[^']*'|\"[^\"]*\"|\S+)",
    re.IGNORECASE,
)
_SECRET_ENV = re.compile(
    r"(?P<var>\b\w*(?:PASSWORD|PASSWD|SECRET|TOKEN)\w*=)"
    r"(?P<val>'[^']*'|\"[^\"]*\"|\S+)"
)


def redact_secrets(text):
    """Remplace la valeur des options et variables porteuses de secret.

    Appliqué à CHAQUE affichage d'une commande. Filtrer au point d'affichage
    plutôt qu'à la construction est ce qui rend la garantie tenable : il n'y a
    qu'une poignée de sorties ici, alors que les commandes se construisent
    partout dans le dépôt.
    """
    if not text:
        return text
    text = _SECRET_OPTION.sub(lambda m: m.group("opt") + "'***'", text)
    return _SECRET_ENV.sub(lambda m: m.group("var") + "'***'", text)


new_path = os.path.normpath(
    os.path.join(os.path.dirname(__file__), "..", "..")
)
sys.path.append(new_path)


logging.basicConfig(
    format=(
        "%(asctime)s,%(msecs)d %(levelname)-8s [%(filename)s:%(lineno)d]"
        " %(message)s"
    ),
    datefmt="%Y-%m-%d:%H:%M:%S",
    level=logging.INFO,
)
_logger = logging.getLogger(__name__)


class Execute:
    def __init__(self) -> None:
        self.cmd_source_erplibre: str = ""
        self.cmd_source_default: str = ""
        exec_path_gnome_terminal = shutil.which("gnome-terminal")
        if exec_path_gnome_terminal:
            self.cmd_source_erplibre = (
                f"gnome-terminal -- bash -c 'source"
                f" ./{VENV_ERPLIBRE}/bin/activate;%s'"
            )
            self.cmd_source_default = "gnome-terminal -- bash -c '" f"%s'"
        else:
            exec_path_tell = shutil.which("osascript")
            if exec_path_tell:
                self.cmd_source_erplibre = (
                    "osascript -e 'tell application \"Terminal\"'"
                )
                self.cmd_source_erplibre += " -e 'tell application \"System Events\" to keystroke \"t\" using {command down}' -e 'delay 0.1' -e 'do script \""
                self.cmd_source_erplibre += f"cd {os.getcwd()}; source ./{VENV_ERPLIBRE}/bin/activate; %s\" in front window'"
                self.cmd_source_erplibre += " -e 'end tell'"
            else:
                self.cmd_source_erplibre = (
                    f"source ./{VENV_ERPLIBRE}/bin/activate;%s"
                )

    def exec_command_live(
        self,
        command: str,
        source_erplibre: bool = True,
        quiet: bool = False,
        single_source_erplibre: bool = False,
        new_window: bool = False,
        single_source_odoo: bool = False,
        source_odoo: str = "",
        new_env: dict | None = None,
        return_status_and_command: bool = False,
        return_status_and_output: bool = False,
        return_status_and_output_and_command: bool = False,
    ) -> (
        int
        | tuple[int, str]
        | tuple[int, list[str]]
        | tuple[int, str, list[str]]
    ):
        """
        Execute a command and display its output live.

        Args:
            command (str): The command to execute.
        """

        my_env = os.environ.copy()
        if new_env:
            my_env.update(new_env)

        process_start_time = time.time()
        exit_code = None
        if source_erplibre:
            # command = f"source ./{VENV_ERPLIBRE}/bin/activate && " + command
            # cmd = (
            #     f"gnome-terminal --tab -- bash -c 'source"
            #     f" ./{VENV_ERPLIBRE}/bin/activate;{command}'"
            # )
            command = self.cmd_source_erplibre % command
            # os.system(f"./script/terminal/open_terminal.sh {command}")
        elif single_source_erplibre:
            command = f"source ./{VENV_ERPLIBRE}/bin/activate && %s" % command
        elif single_source_odoo:
            if not source_odoo and os.path.exists("./.erplibre-version"):
                with open("./.erplibre-version") as f:
                    source_odoo = f.read()
            if not source_odoo:
                _logger.error(
                    "You cannot execute Odoo command if no version is"
                    f" installed. Command : {redact_secrets(command)}"
                )
                # Return the SAME shape the caller asked for. A bare int here
                # made callers doing « status, cmd = exec_command_live(...) »
                # crash with ValueError instead of seeing the failure.
                if return_status_and_output_and_command:
                    return 1, command, []
                if return_status_and_command:
                    return 1, command
                if return_status_and_output:
                    return 1, []
                return 1
            command = f"source ./.venv.{source_odoo}/bin/activate && {command}"
        if new_window and self.cmd_source_default:
            command = self.cmd_source_default % command

        if not quiet:
            print("🏠 ⬇ Execute command :\n")
            print(redact_secrets(command))
        output_lines = []

        try:
            process = subprocess.Popen(
                command,
                shell=True,
                executable="/bin/bash",
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                # Octets bruts, SANS tampon. « readline » attendait le saut de
                # ligne pour rendre la main : une invite qui n'en porte pas —
                # « Continuer ? [o/N] » — restait donc invisible jusqu'à ce que
                # la réponse soit déjà tapée. La question s'affichait APRÈS la
                # réponse, et l'on répondait à l'aveugle.
                bufsize=0,
                env=my_env,
            )

            sink = getattr(self, "log_sink", None)
            # Le tube porte des octets, et une lecture peut couper un caractère
            # accentué ou un emoji en deux. Le décodeur incrémental garde le
            # morceau incomplet en attente au lieu de rendre un « ? ».
            decoder = codecs.getincrementaldecoder("utf-8")("replace")
            fd = process.stdout.fileno()
            # « pending » est la ligne en cours, pas encore terminée ; « shown »
            # compte ce qui en a déjà été envoyé au terminal, pour ne jamais
            # afficher deux fois le même morceau d'invite quand la ligne finit
            # par se terminer.
            pending = ""
            shown = 0

            def retenir(ligne):
                """Journaliser et retenir une ligne complète, caviardée."""
                nonlocal sink
                # La sortie du sous-processus passe par le MÊME filtre que la
                # commande : un outil qui réaffiche ses propres arguments
                # (« set -x », une trace, odoo_bin.sh) y remettrait le secret
                # que l'affichage de la commande venait d'écarter.
                clean = redact_secrets(ligne)
                if sink:
                    # Chaque ligne passe DÉJÀ ici : c'est le seul endroit où
                    # journaliser sans rien changer à ce que le terminal
                    # montre. Une erreur d'écriture ne doit jamais faire
                    # échouer la commande qu'on est en train de suivre.
                    try:
                        sink.write(clean)
                    except Exception:
                        sink = None
                if (
                    return_status_and_output
                    or return_status_and_output_and_command
                ):
                    # Remove last \n char
                    output_lines.append(
                        clean.removesuffix("\r\n")
                        .removesuffix("\n")
                        .removesuffix("\r")
                    )

            while True:
                chunk = os.read(fd, 65536)
                if not chunk:
                    break
                pending += decoder.decode(chunk)
                while True:
                    coupe = pending.find("\n")
                    if coupe < 0:
                        break
                    ligne = pending[: coupe + 1]
                    pending = pending[coupe + 1 :]
                    if not quiet:
                        print(redact_secrets(ligne[shown:]), end="")
                    shown = 0
                    retenir(ligne)
                if not quiet:
                    if len(pending) > shown:
                        # Le reliquat sans saut de ligne EST l'invite : la
                        # montrer tout de suite, avant que la commande ne se
                        # bloque sur la lecture de la réponse.
                        print(redact_secrets(pending[shown:]), end="")
                        shown = len(pending)
                    # Sans vidage explicite, cette invite resterait dans le
                    # tampon de Python : second endroit où la question se
                    # perdait, la sortie n'étant vidée qu'au saut de ligne.
                    sys.stdout.flush()

            pending += decoder.decode(b"", True)
            if pending:
                if not quiet:
                    print(redact_secrets(pending[shown:]), end="")
                    sys.stdout.flush()
                retenir(pending)

            process.wait()
            exit_code = process.returncode
            if process.returncode != 0 and not quiet:
                print("Command returned error code:" f" {process.returncode}")

        # An exception MUST report a failure. exit_code stays None otherwise,
        # and None is falsy: callers testing « if not status: » would mark the
        # step as done, and « if status and wait_at_error » would skip the error
        # prompt. A crashed command was therefore recorded as a success.
        except FileNotFoundError:
            exit_code = 1
            if not quiet:
                print(f"Error: Command '{redact_secrets(command)}' not found.")
        except Exception as e:
            exit_code = 1
            if not quiet:
                print(f"An error occurred: {redact_secrets(str(e))}")
        process_end_time = time.time()
        duration_sec = process_end_time - process_start_time
        if humanize:
            duration_delta = datetime.timedelta(seconds=duration_sec)
            human_time = humanize.precisedelta(duration_delta)
            if not quiet:
                print(f"🏠 ⬆ Executed ({human_time}) :\n")
        else:
            if not quiet:
                print(f"🏠 ⬆ Executed ({duration_sec:.2f} sec.) :\n")
        if not quiet:
            print(redact_secrets(command))
            print()
        if return_status_and_output_and_command:
            return exit_code, command, output_lines
        if return_status_and_command:
            return exit_code, command
        if return_status_and_output:
            return exit_code, output_lines
        return exit_code