Mathieu Benoit
315e45e5f0
`vpn.py` savait tout faire en ligne de commande, mais il fallait écrire le profil JSON à la main et déposer les secrets dans KeePassXC soi-même. Le menu pose les questions que le pilote choisi déclare, et lui seul : ajouter une technologie n'ajoute pas une ligne ici. Le coffre est manipulé EN PROCESSUS — le mot de passe maître est déjà en mémoire, le redemander à un sous-processus serait une saisie de plus à chaque geste. Montage, démontage et diagnostic passent au contraire par `vpn.py` en sous-processus : ils durent, ils parlent, et la sortie en direct est ce qui rend une montée de tunnel suivable. Vérifié : 37 tests sur le menu, aucune saisie réelle. --- EN --- `vpn.py` could already do everything from the command line, but the JSON profile had to be written by hand and the secrets filed into KeePassXC by hand. The menu asks the questions the chosen driver declares, and only those: adding a technology adds no line here. The vault is handled IN PROCESS — the master password is already in memory, and asking a subprocess for it again would be one more entry at every step. Raising, tearing down and diagnosing go through `vpn.py` as a subprocess instead: they last, they talk, and live output is what makes a tunnel coming up followable. Checked: 37 tests on the menu, no real input. Assisted-by: Claude Opus 5
601 lines
22 KiB
Python
601 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
# © 2026 TechnoLibre (http://www.technolibre.ca)
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|
"""Le menu VPN : le chemin que l'utilisateur emprunte vraiment.
|
|
|
|
Les pilotes sont testés ailleurs. Ici on vérifie que le FORMULAIRE reste
|
|
agnostique : il déroule les questions déclarées par le pilote choisi, sans
|
|
rien savoir de L2TP ni de WireGuard. C'est ce qui fait qu'ajouter une
|
|
technologie n'ajoute pas une ligne au menu — et c'est donc ce qui doit
|
|
casser bruyamment si quelqu'un y remet un cas particulier.
|
|
|
|
Aucune saisie réelle : `input` est remplacé par une liste de réponses, dans
|
|
l'ordre où les questions sont posées.
|
|
"""
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from unittest.mock import patch
|
|
|
|
sys.path.append(
|
|
os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
|
|
)
|
|
sys.argv = ["todo.py"]
|
|
|
|
from script.todo.todo import TODO # noqa: E402
|
|
from script.todo.todo_i18n import t # noqa: E402
|
|
from script.todo.vpn_menu import ( # noqa: E402
|
|
DRIVER_LETTERS,
|
|
UNPROVEN_NOTE,
|
|
match_driver,
|
|
)
|
|
from script.vpn import profiles # noqa: E402
|
|
from script.vpn.drivers import DRIVERS # noqa: E402
|
|
|
|
WG_PUBLIC = base64.b64encode(bytes(range(32, 64))).decode()
|
|
|
|
|
|
class MenuBase(unittest.TestCase):
|
|
"""Les trois fichiers de configuration dans un temporaire : un test qui
|
|
écrirait dans le fichier privé détruirait les profils de qui le lance."""
|
|
|
|
def setUp(self):
|
|
self.todo = TODO()
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
base = os.path.join(self.tmp.name, "todo.json")
|
|
with open(base, "w") as fh:
|
|
json.dump({"vpn": [], "kdbx": {"path": "", "password": ""}}, fh)
|
|
self.patches = [
|
|
patch("script.config.config_file.CONFIG_FILE", base),
|
|
patch(
|
|
"script.config.config_file.CONFIG_OVERRIDE_FILE",
|
|
os.path.join(self.tmp.name, "override.json"),
|
|
),
|
|
patch(
|
|
"script.config.config_file.CONFIG_OVERRIDE_PRIVATE_FILE",
|
|
os.path.join(self.tmp.name, "private.json"),
|
|
),
|
|
]
|
|
for item in self.patches:
|
|
item.start()
|
|
|
|
def tearDown(self):
|
|
for item in self.patches:
|
|
item.stop()
|
|
self.tmp.cleanup()
|
|
|
|
def answering(self, *answers):
|
|
"""Remplace `input` par une suite de réponses. Une question de plus
|
|
que prévu lève StopIteration — et c'est bien : cela veut dire que le
|
|
formulaire a changé sans que le test le sache."""
|
|
return patch("builtins.input", side_effect=list(answers))
|
|
|
|
|
|
class MatchDriver(unittest.TestCase):
|
|
"""La correspondance est pure : elle se juge sans menu ni saisie.
|
|
|
|
Elle existe parce qu'une liste numérotée juste après un menu numéroté
|
|
fait taper un numéro de menu — et que devant « [L2TP/IPsec PSK] », on
|
|
tape « L ». Les deux doivent marcher.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.names = list(DRIVERS)
|
|
|
|
def test_the_displayed_letter(self):
|
|
for index, name in enumerate(self.names):
|
|
with self.subTest(name=name):
|
|
letter = DRIVER_LETTERS[index]
|
|
self.assertEqual(match_driver(letter, self.names), name)
|
|
self.assertEqual(
|
|
match_driver(letter.upper(), self.names), name
|
|
)
|
|
|
|
def test_the_rank_still_works(self):
|
|
"""Quelqu'un tapera un chiffre, et il a raison de le faire vu le
|
|
menu qui précède."""
|
|
for index, name in enumerate(self.names, start=1):
|
|
with self.subTest(rang=index):
|
|
self.assertEqual(match_driver(str(index), self.names), name)
|
|
|
|
def test_the_start_of_the_label(self):
|
|
"""« L » devant « L2TP/IPsec PSK » : le geste qui a motivé tout
|
|
ceci."""
|
|
for answer, expected in (
|
|
("L", "l2tp_ipsec"),
|
|
("l2tp", "l2tp_ipsec"),
|
|
("w", "wireguard"),
|
|
("WireG", "wireguard"),
|
|
("openv", "openvpn"),
|
|
("openc", "openconnect"),
|
|
("ssh", "sshuttle"),
|
|
):
|
|
with self.subTest(answer=answer):
|
|
self.assertEqual(match_driver(answer, self.names), expected)
|
|
|
|
def test_an_ambiguous_prefix_names_the_candidates(self):
|
|
"""« open » désigne deux pilotes : le dire vaut mieux qu'en choisir
|
|
un au hasard."""
|
|
result = match_driver("open", self.names)
|
|
self.assertIsInstance(result, list)
|
|
self.assertEqual(sorted(result), ["openconnect", "openvpn"])
|
|
|
|
def test_nothing_matches_nothing(self):
|
|
for answer in ("x", "9", "0", "", " ", "carrier-pigeon"):
|
|
with self.subTest(answer=answer):
|
|
self.assertEqual(match_driver(answer, self.names), "")
|
|
|
|
|
|
class DriverPicker(MenuBase):
|
|
def test_an_empty_answer_keeps_the_current_driver(self):
|
|
with self.answering(""):
|
|
with redirect_stdout(io.StringIO()):
|
|
chosen = self.todo._vpn_pick_driver("wireguard")
|
|
self.assertEqual(chosen.name, "wireguard")
|
|
|
|
def test_a_letter_picks_from_the_list(self):
|
|
names = list(DRIVERS)
|
|
with self.answering("c"):
|
|
with redirect_stdout(io.StringIO()):
|
|
chosen = self.todo._vpn_pick_driver(None)
|
|
self.assertEqual(chosen.name, names[2])
|
|
|
|
def test_a_number_still_picks_from_the_list(self):
|
|
names = list(DRIVERS)
|
|
with self.answering("3"):
|
|
with redirect_stdout(io.StringIO()):
|
|
chosen = self.todo._vpn_pick_driver(None)
|
|
self.assertEqual(chosen.name, names[2])
|
|
|
|
def test_the_start_of_a_label_picks_too(self):
|
|
with self.answering("L"):
|
|
with redirect_stdout(io.StringIO()):
|
|
chosen = self.todo._vpn_pick_driver(None)
|
|
self.assertEqual(chosen.name, "l2tp_ipsec")
|
|
|
|
def test_the_list_is_lettered_and_offers_a_way_back(self):
|
|
buffer = io.StringIO()
|
|
with self.answering(""):
|
|
with redirect_stdout(buffer):
|
|
self.todo._vpn_pick_driver(None)
|
|
printed = buffer.getvalue()
|
|
for letter in DRIVER_LETTERS[: len(DRIVERS)]:
|
|
self.assertIn(f"[{letter}]", printed)
|
|
self.assertNotIn("[1]", printed)
|
|
self.assertIn("[0]", printed)
|
|
|
|
def test_zero_goes_back_without_scolding(self):
|
|
"""Sans sortie explicite, on est coincé dans le formulaire dès
|
|
qu'on a tapé un nom de profil."""
|
|
buffer = io.StringIO()
|
|
with self.answering("0"):
|
|
with redirect_stdout(buffer):
|
|
self.assertIsNone(self.todo._vpn_pick_driver(None))
|
|
self.assertNotIn("✗", buffer.getvalue())
|
|
self.assertNotIn("inconnu", buffer.getvalue().lower())
|
|
|
|
def test_an_ambiguous_answer_says_which_ones(self):
|
|
buffer = io.StringIO()
|
|
with self.answering("open"):
|
|
with redirect_stdout(buffer):
|
|
self.assertIsNone(self.todo._vpn_pick_driver(None))
|
|
printed = buffer.getvalue()
|
|
self.assertIn("OpenVPN", printed)
|
|
self.assertIn("OpenConnect", printed)
|
|
|
|
def test_an_out_of_range_answer_gives_up(self):
|
|
with self.answering("99"):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.assertIsNone(self.todo._vpn_pick_driver(None))
|
|
|
|
def test_every_driver_shows_its_hint(self):
|
|
"""C'est la seule décision où l'utilisateur a besoin d'un conseil."""
|
|
buffer = io.StringIO()
|
|
with self.answering(""):
|
|
with redirect_stdout(buffer):
|
|
self.todo._vpn_pick_driver(None)
|
|
printed = buffer.getvalue()
|
|
for cls in DRIVERS.values():
|
|
self.assertIn(cls.label, printed)
|
|
|
|
def test_only_the_unproven_technologies_wear_a_star(self):
|
|
"""Sans marque, la liste montre des choix d'apparence égale, et
|
|
rien ne dit lequel a déjà abouti contre un vrai serveur."""
|
|
buffer = io.StringIO()
|
|
with self.answering(""):
|
|
with redirect_stdout(buffer):
|
|
self.todo._vpn_pick_driver(None)
|
|
starred = {
|
|
line.split("]")[1].strip().split(" ")[0]
|
|
for line in buffer.getvalue().splitlines()
|
|
if line.startswith("[") and "*" in line
|
|
}
|
|
expected = {
|
|
cls.label.split(" ")[0]
|
|
for cls in DRIVERS.values()
|
|
if not cls.proven
|
|
}
|
|
self.assertEqual(starred, expected)
|
|
|
|
def test_the_star_is_explained(self):
|
|
"""Une marque sans légende inquiète sans informer."""
|
|
buffer = io.StringIO()
|
|
with self.answering(""):
|
|
with redirect_stdout(buffer):
|
|
self.todo._vpn_pick_driver(None)
|
|
self.assertIn(t(UNPROVEN_NOTE), buffer.getvalue())
|
|
|
|
|
|
class TheFormIsDriverAgnostic(MenuBase):
|
|
def test_it_builds_a_wireguard_profile_from_typed_answers(self):
|
|
names = list(DRIVERS)
|
|
answers = [
|
|
"acme-wg", # nom du profil
|
|
DRIVER_LETTERS[names.index("wireguard")], # technologie
|
|
"vpn.acme.example", # serveur
|
|
"10.7.0.2/32", # wg_address
|
|
WG_PUBLIC, # wg_peer_key
|
|
"10.7.0.0/24", # réseaux
|
|
"", # tout le trafic ? défaut non
|
|
"", # témoin
|
|
"n", # réglages avancés ?
|
|
]
|
|
with self.answering(*answers):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.todo._vpn_edit_profile()
|
|
saved = profiles.load("acme-wg")
|
|
self.assertIsNotNone(saved, "profil non enregistré")
|
|
self.assertEqual(saved["driver"], "wireguard")
|
|
self.assertEqual(saved["wg_address"], "10.7.0.2/32")
|
|
self.assertEqual(saved["wg_peer_key"], WG_PUBLIC)
|
|
self.assertEqual(saved["routes"], ["10.7.0.0/24"])
|
|
self.assertFalse(saved["default_route"])
|
|
# Le défaut du pilote, jamais demandé, doit être là quand même.
|
|
self.assertEqual(saved["port"], 51820)
|
|
|
|
def test_it_builds_an_sshuttle_profile_with_no_secret_question(self):
|
|
names = list(DRIVERS)
|
|
answers = [
|
|
"acme-ssh",
|
|
DRIVER_LETTERS[names.index("sshuttle")],
|
|
"erplibre@bastion.acme.example",
|
|
"10.40.0.0/16",
|
|
"",
|
|
"10.40.0.1", # témoin
|
|
"n", # pas de réglages avancés
|
|
]
|
|
with self.answering(*answers):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.todo._vpn_edit_profile()
|
|
saved = profiles.load("acme-ssh")
|
|
self.assertIsNotNone(saved)
|
|
self.assertEqual(saved["server"], "erplibre@bastion.acme.example")
|
|
self.assertEqual(saved["probe"], "10.40.0.1")
|
|
|
|
def test_the_mtu_is_not_asked_when_the_driver_ignores_it(self):
|
|
"""sshuttle ne prend pas le MTU du profil : le demander serait une
|
|
question sans effet. Si le formulaire le demandait, la liste de
|
|
réponses serait épuisée et le test lèverait StopIteration."""
|
|
names = list(DRIVERS)
|
|
answers = [
|
|
"acme-ssh2",
|
|
str(names.index("sshuttle") + 1),
|
|
"bastion.acme.example",
|
|
"10.41.0.0/16",
|
|
"",
|
|
"",
|
|
"o", # réglages avancés OUI
|
|
"2222", # port SSH
|
|
"", # DNS dans le tunnel : défaut
|
|
]
|
|
with self.answering(*answers):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.todo._vpn_edit_profile()
|
|
saved = profiles.load("acme-ssh2")
|
|
self.assertIsNotNone(saved)
|
|
self.assertEqual(saved["port"], 2222)
|
|
|
|
def test_editing_keeps_what_is_not_retyped(self):
|
|
profiles.save(
|
|
{
|
|
"name": "acme-wg",
|
|
"driver": "wireguard",
|
|
"server": "vpn.acme.example",
|
|
"wg_address": "10.7.0.2/32",
|
|
"wg_peer_key": WG_PUBLIC,
|
|
"routes": ["10.7.0.0/24"],
|
|
"wg_keepalive": 17,
|
|
}
|
|
)
|
|
names = list(DRIVERS)
|
|
answers = [
|
|
"acme-wg",
|
|
str(names.index("wireguard") + 1),
|
|
"", # serveur inchangé
|
|
"", # wg_address inchangée
|
|
"", # wg_peer_key inchangée
|
|
"10.7.0.0/24, 10.9.0.0/16", # routes élargies
|
|
"",
|
|
"",
|
|
"n",
|
|
]
|
|
with self.answering(*answers):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.todo._vpn_edit_profile()
|
|
saved = profiles.load("acme-wg")
|
|
self.assertEqual(saved["server"], "vpn.acme.example")
|
|
self.assertEqual(saved["wg_peer_key"], WG_PUBLIC)
|
|
self.assertEqual(saved["routes"], ["10.7.0.0/24", "10.9.0.0/16"])
|
|
# Un réglage avancé jamais réaffiché ne doit pas être perdu.
|
|
self.assertEqual(saved["wg_keepalive"], 17)
|
|
|
|
def test_a_refused_profile_saves_nothing(self):
|
|
names = list(DRIVERS)
|
|
answers = [
|
|
"acme-bad",
|
|
str(names.index("wireguard") + 1),
|
|
"vpn.acme.example",
|
|
"10.7.0.2/32",
|
|
"pas-une-cle", # clé de pair invalide
|
|
"10.7.0.0/24",
|
|
"",
|
|
"",
|
|
"n",
|
|
]
|
|
buffer = io.StringIO()
|
|
with self.answering(*answers):
|
|
with redirect_stdout(buffer):
|
|
self.todo._vpn_edit_profile()
|
|
self.assertIsNone(profiles.load("acme-bad"))
|
|
self.assertIn("✗", buffer.getvalue())
|
|
|
|
|
|
class OnlyWhatTheSiteGaveYou(MenuBase):
|
|
"""Le cas réel : le site remet une passerelle, un utilisateur, un mot de
|
|
passe et une clé. Rien sur les réseaux derrière.
|
|
|
|
Ce profil DOIT s'enregistrer. Il ne joint que l'hôte distant, le menu le
|
|
dit, et le premier montage proposera le réseau que l'adresse révèle —
|
|
refuser l'enregistrement laissait sans issue.
|
|
"""
|
|
|
|
def test_a_profile_without_routes_is_accepted_and_flagged(self):
|
|
names = list(DRIVERS)
|
|
answers = [
|
|
"novipro",
|
|
DRIVER_LETTERS[names.index("l2tp_ipsec")],
|
|
"vpn.novipro.example", # la passerelle
|
|
"user", # l'utilisateur PPP
|
|
"", # réseaux : le site n'en a pas donné
|
|
"", # tout le trafic ? non
|
|
"", # témoin
|
|
"n",
|
|
]
|
|
buffer = io.StringIO()
|
|
with self.answering(*answers):
|
|
with redirect_stdout(buffer):
|
|
self.todo._vpn_edit_profile()
|
|
saved = profiles.load("novipro")
|
|
self.assertIsNotNone(saved, "profil refusé alors qu'il est utilisable")
|
|
self.assertEqual(saved["routes"], [])
|
|
self.assertFalse(saved["default_route"])
|
|
printed = buffer.getvalue()
|
|
self.assertIn("✓", printed)
|
|
self.assertIn("hôte distant", printed)
|
|
|
|
def test_the_first_mount_suggests_the_network(self):
|
|
"""Sans route déclarée, le montage propose le /24 de l'adresse
|
|
obtenue — en disant que c'est une hypothèse."""
|
|
from script.vpn.drivers.l2tp_ipsec import L2tpIpsecDriver
|
|
from script.vpn.runner import Runner
|
|
|
|
profile = profiles.validate(
|
|
{
|
|
"name": "novipro",
|
|
"driver": "l2tp_ipsec",
|
|
"server": "127.0.0.1",
|
|
"ppp_user": "user",
|
|
}
|
|
)
|
|
driver = L2tpIpsecDriver(profile, {"psk": "x", "password": "y"})
|
|
runner = Runner(dry_run=True)
|
|
buffer = io.StringIO()
|
|
with patch(
|
|
"script.vpn.drivers.base.interface_addresses",
|
|
return_value=["192.168.50.20", "192.168.50.1"],
|
|
):
|
|
with redirect_stdout(buffer):
|
|
driver.suggest_routes(runner, "ppp0")
|
|
printed = buffer.getvalue()
|
|
self.assertIn("192.168.50.0/24", printed)
|
|
self.assertIn("hypothèse", printed)
|
|
|
|
def test_nothing_is_suggested_when_routes_are_declared(self):
|
|
"""La suggestion ne s'invite pas quand la question est réglée."""
|
|
from script.vpn.drivers.l2tp_ipsec import L2tpIpsecDriver
|
|
from script.vpn.runner import Runner
|
|
|
|
profile = profiles.validate(
|
|
{
|
|
"name": "novipro",
|
|
"driver": "l2tp_ipsec",
|
|
"server": "127.0.0.1",
|
|
"ppp_user": "user",
|
|
"routes": ["10.20.0.0/16"],
|
|
}
|
|
)
|
|
driver = L2tpIpsecDriver(profile, {"psk": "x", "password": "y"})
|
|
buffer = io.StringIO()
|
|
with redirect_stdout(buffer):
|
|
driver.suggest_routes(Runner(dry_run=True), "ppp0")
|
|
self.assertEqual(buffer.getvalue(), "")
|
|
|
|
def test_wireguard_still_requires_its_allowed_ips(self):
|
|
"""L'exigence reste DURE là où elle l'est vraiment : sans
|
|
AllowedIPs, wg-quick refuse la configuration entière."""
|
|
from script.vpn.valid import ProfileError
|
|
|
|
with self.assertRaises(ProfileError):
|
|
profiles.validate(
|
|
{
|
|
"name": "beta",
|
|
"driver": "wireguard",
|
|
"server": "127.0.0.1",
|
|
"wg_address": "10.7.0.2/32",
|
|
"wg_peer_key": WG_PUBLIC,
|
|
}
|
|
)
|
|
|
|
|
|
class SecretsOnlyWhenThereAreSome(MenuBase):
|
|
def test_a_driver_without_secrets_does_not_open_the_vault(self):
|
|
"""sshuttle s'authentifie par SSH. Demander le mot de passe maître
|
|
du coffre pour lui serait une saisie pour rien."""
|
|
profiles.save(
|
|
{
|
|
"name": "acme-ssh",
|
|
"driver": "sshuttle",
|
|
"server": "bastion.acme.example",
|
|
"routes": ["10.40.0.0/16"],
|
|
}
|
|
)
|
|
buffer = io.StringIO()
|
|
with patch.object(
|
|
self.todo, "_vpn_select_profile", return_value="acme-ssh"
|
|
):
|
|
with patch.object(self.todo.kdbx_manager, "get_kdbx") as opened:
|
|
with redirect_stdout(buffer):
|
|
self.todo._vpn_store_secrets()
|
|
opened.assert_not_called()
|
|
self.assertIn("SSH", buffer.getvalue())
|
|
|
|
def test_creating_the_vault_does_not_reask_the_master_password(self):
|
|
"""Six saisies masquées, pas sept.
|
|
|
|
Deux pour créer le coffre, quatre pour les deux secrets confirmés.
|
|
`create_database` rend la base DÉJÀ ouverte : sans l'adopter, le mot
|
|
de passe maître était redemandé dans la seconde suivant les deux
|
|
saisies de la création.
|
|
"""
|
|
profiles.save(
|
|
{
|
|
"name": "novipro",
|
|
"driver": "l2tp_ipsec",
|
|
"server": "vpn.novipro.example",
|
|
"ppp_user": "user",
|
|
}
|
|
)
|
|
coffre = os.path.join(self.tmp.name, "secrets.kdbx")
|
|
typed = []
|
|
|
|
def masked(prompt=""):
|
|
typed.append(prompt)
|
|
if len(typed) <= 2:
|
|
return "maitre"
|
|
return "secret"
|
|
|
|
with patch.object(
|
|
self.todo, "_vpn_select_profile", return_value="novipro"
|
|
):
|
|
with self.answering(coffre, "o"):
|
|
with patch("getpass.getpass", masked):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.todo._vpn_store_secrets()
|
|
self.assertEqual(len(typed), 6, typed)
|
|
self.assertTrue(os.path.exists(coffre))
|
|
|
|
def test_an_empty_answer_on_an_empty_field_is_reported(self):
|
|
"""« Une réponse vide garde la valeur en place » est un piège quand
|
|
il n'y a RIEN en place : le secret restait vide en silence, et le
|
|
premier montage échouait sur « Secrets manquants ». L'invite dit
|
|
maintenant l'état, et le bilan nomme ce qui manque encore.
|
|
"""
|
|
profiles.save(
|
|
{
|
|
"name": "novipro",
|
|
"driver": "l2tp_ipsec",
|
|
"server": "vpn.novipro.example",
|
|
"ppp_user": "user",
|
|
}
|
|
)
|
|
coffre = os.path.join(self.tmp.name, "secrets.kdbx")
|
|
typed = []
|
|
|
|
def masked(prompt=""):
|
|
typed.append(prompt)
|
|
if len(typed) <= 2:
|
|
return "maitre"
|
|
# La PSK et sa confirmation, puis Entrée sur le mot de passe.
|
|
return "LaClePSK" if len(typed) <= 4 else ""
|
|
|
|
buffer = io.StringIO()
|
|
with patch.object(
|
|
self.todo, "_vpn_select_profile", return_value="novipro"
|
|
):
|
|
with self.answering(coffre, "o"):
|
|
with patch("getpass.getpass", masked):
|
|
with redirect_stdout(buffer):
|
|
self.todo._vpn_store_secrets()
|
|
printed = buffer.getvalue()
|
|
self.assertIn("Mot de passe PPP", printed)
|
|
self.assertIn("Toujours manquant", printed)
|
|
# Chaque invite annonce ce qu'il y a derrière.
|
|
self.assertTrue(
|
|
[p for p in typed if "[vide]" in p],
|
|
typed,
|
|
)
|
|
|
|
def test_a_field_already_set_says_so(self):
|
|
profiles.save(
|
|
{
|
|
"name": "novipro",
|
|
"driver": "l2tp_ipsec",
|
|
"server": "vpn.novipro.example",
|
|
"ppp_user": "user",
|
|
}
|
|
)
|
|
coffre = os.path.join(self.tmp.name, "secrets.kdbx")
|
|
typed = []
|
|
|
|
def masked(prompt=""):
|
|
typed.append(prompt)
|
|
if len(typed) <= 2:
|
|
return "maitre"
|
|
return "valeur"
|
|
|
|
with patch.object(
|
|
self.todo, "_vpn_select_profile", return_value="novipro"
|
|
):
|
|
with self.answering(coffre, "o"):
|
|
with patch("getpass.getpass", masked):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.todo._vpn_store_secrets()
|
|
# Deuxième passage : tout est en place, et les invites le disent.
|
|
typed.clear()
|
|
with patch(
|
|
"getpass.getpass",
|
|
lambda prompt="": (typed.append(prompt) or ""),
|
|
):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.todo._vpn_store_secrets()
|
|
self.assertTrue([p for p in typed if "[déjà en place]" in p], typed)
|
|
self.assertFalse([p for p in typed if "[vide]" in p], typed)
|
|
|
|
def test_a_mismatched_confirmation_stores_nothing(self):
|
|
with patch("getpass.getpass", side_effect=["un", "deux"]):
|
|
with redirect_stdout(io.StringIO()):
|
|
self.assertIsNone(self.todo._vpn_ask_secret("PSK"))
|
|
|
|
def test_an_empty_answer_keeps_the_stored_value(self):
|
|
with patch("getpass.getpass", return_value=""):
|
|
self.assertEqual(self.todo._vpn_ask_secret("PSK"), "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|