Mathieu Benoit
eb28952d6c
Le filtre ne connaissait que trois noms de variable — mot de passe, secret, jeton — donc `OPENAI_API_KEY=` partait en clair dans le terminal, dans les journaux et dans toute sortie CI qui les capture. Un jeton d'en-tête échappait aux deux règles par construction : il ne porte ni nom d'option ni nom de variable, il suit le mot « Bearer ». Le filtre couvre maintenant `API_KEY` côté variables et `Authorization: Bearer|Basic` côté en-têtes, sans casse, la valeur allant jusqu'au premier blanc. Il protège au même titre la restauration de base, qui l'appelle sur six sorties. Reste le dernier rempart et non le premier : argv est lisible par tout compte de la machine, où aucun caviardage n'atteint. --- EN --- The filter knew only three variable names — password, secret, token — so `OPENAI_API_KEY=` went out in the clear to the terminal, to the logs and to any CI output capturing them. A header token escaped both rules by construction: it carries neither an option name nor a variable name, it follows the word "Bearer". The filter now covers `API_KEY` on the variable side and `Authorization: Bearer|Basic` on the header side, case-insensitively, the value running to the first blank. It protects database restore just as much, which calls it on six outputs. It stays the last line of defence, not the first: argv is readable by every account on the machine, where no redaction reaches. Assisted-by: Claude Opus 5
316 lines
13 KiB
Python
316 lines
13 KiB
Python
#!/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|API_?KEY)\w*=)"
|
|
r"(?P<val>'[^']*'|\"[^\"]*\"|\S+)"
|
|
)
|
|
# Un jeton porté par un en-tête n'a ni nom d'option ni nom de variable : il
|
|
# suit le mot « Bearer », et les deux règles ci-dessus passent à côté. Le
|
|
# schéma est nommé par la RFC 7235 et se compare sans casse, la valeur allant
|
|
# jusqu'à la fin de la ligne — un jeton ne porte pas d'espace.
|
|
_SECRET_HEADER = re.compile(
|
|
r"(?P<schema>\bAuthorization:\s*(?:Bearer|Basic)\s+)(?P<val>\S+)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def redact_secrets(text):
|
|
"""Remplace la valeur des options, variables et en-têtes 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)
|
|
text = _SECRET_ENV.sub(lambda m: m.group("var") + "'***'", text)
|
|
return _SECRET_HEADER.sub(lambda m: m.group("schema") + "'***'", 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
|