[FIX] db_restore: ask the master password again instead of dying on a typo
It was asked once. Wrong, and Odoo raises AccessDenied, check_output raises CalledProcessError, nothing catches it, and the migration dies on a traceback. After an hour of version bumps that is a steep price for one letter. Ten attempts now. Only a refused PASSWORD is asked again. Any other failure stops and is shown: asking ten times in front of an unreachable database would hide the real fault behind a prompt, and one would hunt for a password. AccessDenied is matched on the class, never on its message, which is translated. The attempt is probed with --list, which changes nothing. Validating here avoids failing half-way, once the database has already been dropped. --- FR --- Il était demandé une fois. Faux, et Odoo lève AccessDenied, check_output lève CalledProcessError, rien ne l'attrape, la migration meurt sur une trace. Après une heure de paliers, c'est cher payé pour une lettre. Dix essais désormais. Seul un MOT DE PASSE refusé fait reposer la question. Tout autre échec arrête et s'affiche : dix invites devant une base injoignable cacheraient la panne, et l'on chercherait un mot de passe. AccessDenied se reconnaît à la CLASSE, jamais au message, qui est traduit. L'essai est éprouvé sur --list, qui ne modifie rien. Valider là évite d'échouer à mi-parcours, une fois la base déjà supprimée. Assisted-by: Claude Opus 5
This commit is contained in:
parent
6339282668
commit
0ebdc0c710
2 changed files with 242 additions and 3 deletions
|
|
@ -9,6 +9,7 @@ import logging
|
|||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import subprocess
|
||||
from subprocess import check_output
|
||||
|
||||
sys.path.append(
|
||||
|
|
@ -83,6 +84,71 @@ def get_master_password():
|
|||
_logger.error("Password echoed, danger!")
|
||||
|
||||
|
||||
# Assez pour une faute de frappe répétée, pas assez pour qu'une boucle
|
||||
# oubliée tourne toute la nuit devant une invite que personne ne lit.
|
||||
MAX_ESSAIS_MOT_DE_PASSE = 10
|
||||
|
||||
|
||||
def password_refused(sortie):
|
||||
"""Odoo a-t-il refusé le mot de passe maître, ou autre chose ?
|
||||
|
||||
La distinction porte tout. Reposer la question sur n'importe quel
|
||||
échec cacherait la vraie panne derrière dix invites, et l'on
|
||||
chercherait un mot de passe alors que la base est cassée.
|
||||
|
||||
Odoo lève `AccessDenied` — la classe apparaît dans la trace, et son
|
||||
message traduit peut varier. On reconnaît donc la CLASSE.
|
||||
"""
|
||||
return "AccessDenied" in (sortie or "")
|
||||
|
||||
|
||||
def probe_master_password(arg_base):
|
||||
"""(accepté, sortie) — éprouver le mot de passe sur `--list`.
|
||||
|
||||
La commande la plus inoffensive : elle ne touche à rien et rend le
|
||||
même refus qu'une restauration. Valider ici évite d'échouer à
|
||||
mi-parcours, une fois la base déjà supprimée.
|
||||
"""
|
||||
done = subprocess.run(
|
||||
f"{arg_base} --list".split(" "),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
return done.returncode == 0, (done.stdout or "") + (done.stderr or "")
|
||||
|
||||
|
||||
def ask_master_password(arg_base, essais=MAX_ESSAIS_MOT_DE_PASSE):
|
||||
"""Le mot de passe maître, redemandé tant qu'Odoo le refuse.
|
||||
|
||||
None si l'on renonce — invite vide, essais épuisés, ou panne qui
|
||||
n'a rien à voir avec le mot de passe.
|
||||
|
||||
Une faute de frappe arrêtait la migration net, sur une trace
|
||||
`CalledProcessError` que rien n'attrapait. Après une heure de
|
||||
paliers, c'est cher payé pour une lettre.
|
||||
"""
|
||||
for tour in range(1, essais + 1):
|
||||
mot = get_master_password()
|
||||
if not mot:
|
||||
return None
|
||||
candidat = f"{arg_base} --master_password={mot}"
|
||||
accepte, sortie = probe_master_password(candidat)
|
||||
if accepte:
|
||||
return mot
|
||||
if not password_refused(sortie):
|
||||
# Autre chose est cassé : le dire, et ne pas noyer la panne
|
||||
# sous dix invites de mot de passe.
|
||||
_logger.error(sortie.strip()[-1500:])
|
||||
return None
|
||||
restants = essais - tour
|
||||
if restants:
|
||||
_logger.warning(
|
||||
f"Master password refused, {restants} attempt(s) left."
|
||||
)
|
||||
_logger.error("Master password refused too many times.")
|
||||
return None
|
||||
|
||||
|
||||
def get_list_db_cache(arg_base):
|
||||
arg = f"{arg_base} --list"
|
||||
out = check_output(arg.split(" ")).decode()
|
||||
|
|
@ -217,12 +283,11 @@ def main():
|
|||
|
||||
has_admin_password = config_parser.get("options", "admin_passwd")
|
||||
if has_admin_password and has_admin_password != "admin":
|
||||
master_password = get_master_password()
|
||||
master_password = ask_master_password(arg_base)
|
||||
if not master_password:
|
||||
_logger.error("Missing master password, cancel transaction.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
arg_base += f" --master_password={master_password}"
|
||||
arg_base += f" --master_password={master_password}"
|
||||
else:
|
||||
_logger.info("No master password needed... Continue")
|
||||
|
||||
|
|
|
|||
174
test/test_master_password_retry.py
Normal file
174
test/test_master_password_retry.py
Normal file
|
|
@ -0,0 +1,174 @@
|
|||
#!/usr/bin/env python3
|
||||
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
||||
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
||||
|
||||
"""Une faute de frappe ne doit pas coûter une migration.
|
||||
|
||||
Le mot de passe maître était demandé UNE fois. Faux ? Odoo lève
|
||||
`AccessDenied`, `check_output` lève `CalledProcessError`, rien ne
|
||||
l'attrape, et la migration meurt sur une trace. Après une heure de
|
||||
paliers, c'est cher payé pour une lettre.
|
||||
|
||||
La propriété qui porte tout : on ne redemande QUE sur un refus de mot de
|
||||
passe. Reposer la question sur n'importe quel échec cacherait la vraie
|
||||
panne derrière dix invites, et l'on chercherait un mot de passe alors
|
||||
que la base est cassée.
|
||||
"""
|
||||
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
|
||||
sys.path.append(
|
||||
os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
)
|
||||
|
||||
from script.database import db_restore # noqa: E402
|
||||
|
||||
|
||||
class TestRecognisingTheRefusal(unittest.TestCase):
|
||||
def test_an_access_denied_is_a_refusal(self):
|
||||
self.assertTrue(
|
||||
db_restore.password_refused(
|
||||
"Traceback...\nodoo.exceptions.AccessDenied: Access Denied"
|
||||
)
|
||||
)
|
||||
|
||||
def test_the_class_is_what_we_match_not_the_message(self):
|
||||
# Le message est traduit : « Accès refusé » en français. La
|
||||
# CLASSE, elle, ne bouge pas.
|
||||
self.assertTrue(
|
||||
db_restore.password_refused("odoo.exceptions.AccessDenied")
|
||||
)
|
||||
self.assertFalse(db_restore.password_refused("Accès refusé"))
|
||||
|
||||
def test_anything_else_is_NOT_a_refusal(self):
|
||||
# C'est la protection : dix invites de mot de passe devant une
|
||||
# base cassée, et l'on cherche du mauvais côté.
|
||||
for sortie in (
|
||||
"psycopg2.OperationalError: could not connect",
|
||||
"FileNotFoundError: ./odoo_bin.sh",
|
||||
"",
|
||||
None,
|
||||
):
|
||||
self.assertFalse(db_restore.password_refused(sortie), sortie)
|
||||
|
||||
|
||||
class TestTheRetryLoop(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.vrais = (
|
||||
db_restore.get_master_password,
|
||||
db_restore.probe_master_password,
|
||||
)
|
||||
self.demandes = 0
|
||||
self.sondes = []
|
||||
|
||||
def tearDown(self):
|
||||
(
|
||||
db_restore.get_master_password,
|
||||
db_restore.probe_master_password,
|
||||
) = self.vrais
|
||||
|
||||
def branche(self, mots, reponses):
|
||||
suite = iter(mots)
|
||||
rep = iter(reponses)
|
||||
|
||||
def demander():
|
||||
self.demandes += 1
|
||||
return next(suite, "")
|
||||
|
||||
def sonder(arg_base):
|
||||
self.sondes.append(arg_base)
|
||||
return next(rep, (False, "AccessDenied"))
|
||||
|
||||
db_restore.get_master_password = demander
|
||||
db_restore.probe_master_password = sonder
|
||||
|
||||
def lance(self, essais=10):
|
||||
tampon = io.StringIO()
|
||||
with redirect_stdout(tampon), redirect_stderr(tampon):
|
||||
return db_restore.ask_master_password("./odoo_bin.sh db", essais)
|
||||
|
||||
def test_a_good_password_is_returned_at_once(self):
|
||||
self.branche(["bon"], [(True, "db1\ndb2")])
|
||||
self.assertEqual(self.lance(), "bon")
|
||||
self.assertEqual(self.demandes, 1)
|
||||
|
||||
def test_a_typo_is_asked_again(self):
|
||||
self.branche(
|
||||
["faux", "bon"],
|
||||
[(False, "odoo.exceptions.AccessDenied"), (True, "db1")],
|
||||
)
|
||||
self.assertEqual(self.lance(), "bon")
|
||||
self.assertEqual(self.demandes, 2)
|
||||
|
||||
def test_it_stops_after_the_allowed_attempts(self):
|
||||
# Sans borne, une invite non lue tournerait toute la nuit.
|
||||
self.branche(["faux"] * 20, [(False, "AccessDenied")] * 20)
|
||||
self.assertIsNone(self.lance(essais=10))
|
||||
self.assertEqual(self.demandes, 10)
|
||||
|
||||
def test_an_empty_prompt_gives_up_immediately(self):
|
||||
# Ctrl-D ou Entrée : on ne veut pas neuf invites de plus.
|
||||
self.branche([""], [])
|
||||
self.assertIsNone(self.lance())
|
||||
self.assertEqual(self.demandes, 1)
|
||||
self.assertEqual(self.sondes, [])
|
||||
|
||||
def test_an_unrelated_failure_stops_instead_of_asking_again(self):
|
||||
# LA propriété. Une base injoignable n'est pas un mot de passe
|
||||
# faux, et redemander dix fois cacherait la vraie panne.
|
||||
self.branche(["bon"], [(False, "psycopg2.OperationalError: refused")])
|
||||
self.assertIsNone(self.lance())
|
||||
self.assertEqual(self.demandes, 1)
|
||||
|
||||
def test_the_unrelated_failure_is_shown(self):
|
||||
self.branche(["bon"], [(False, "psycopg2.OperationalError: refused")])
|
||||
with self.assertLogs(db_restore._logger, level="ERROR") as journal:
|
||||
db_restore.ask_master_password("./odoo_bin.sh db")
|
||||
self.assertIn("OperationalError", "\n".join(journal.output))
|
||||
|
||||
def test_the_probe_carries_the_password_and_touches_nothing(self):
|
||||
# `--list` ne modifie rien : valider ici évite d'échouer à
|
||||
# mi-parcours, une fois la base déjà supprimée.
|
||||
self.branche(["bon"], [(True, "db1")])
|
||||
self.lance()
|
||||
self.assertEqual(len(self.sondes), 1)
|
||||
self.assertIn("--master_password=bon", self.sondes[0])
|
||||
|
||||
def test_each_attempt_probes_with_ITS_password(self):
|
||||
self.branche(
|
||||
["un", "deux"],
|
||||
[(False, "AccessDenied"), (True, "db1")],
|
||||
)
|
||||
self.lance()
|
||||
self.assertIn("--master_password=un", self.sondes[0])
|
||||
self.assertIn("--master_password=deux", self.sondes[1])
|
||||
|
||||
|
||||
class TestTheWiring(unittest.TestCase):
|
||||
def source(self):
|
||||
with io.open(db_restore.__file__, encoding="utf-8") as handle:
|
||||
return handle.read()
|
||||
|
||||
def test_the_flow_uses_the_retrying_version(self):
|
||||
src = self.source()
|
||||
self.assertIn("master_password = ask_master_password(arg_base)", src)
|
||||
|
||||
def test_the_probe_uses_list_which_changes_nothing(self):
|
||||
src = self.source()
|
||||
debut = src.index("def probe_master_password")
|
||||
fin = src.index("def ask_master_password")
|
||||
bloc = src[debut:fin]
|
||||
self.assertIn("--list", bloc)
|
||||
for danger in ("--drop", "--restore", "--clone"):
|
||||
self.assertNotIn(danger, bloc)
|
||||
|
||||
def test_the_bound_is_ten(self):
|
||||
self.assertEqual(db_restore.MAX_ESSAIS_MOT_DE_PASSE, 10)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in a new issue