diff --git a/script/todo/todo.json b/script/todo/todo.json index acf06bc..4bc76d8 100644 --- a/script/todo/todo.json +++ b/script/todo/todo.json @@ -62,6 +62,7 @@ "bash_command": "PATCH=\"/tmp/patch_$(date +%Y%m%d_%H%M%S).patch\"; git -C \"$(pwd)\" diff HEAD > \"$PATCH\" && printf \"\\n✓ Patch created: %s\\n\\n=== Guide to apply the patch ===\\n Check compatibility : git apply --check %s\\n Apply (git) : git apply %s\\n Apply (patch) : patch -p1 < %s\\n Revert (git) : git apply -R %s\\n\" \"$PATCH\" \"$PATCH\" \"$PATCH\" \"$PATCH\" \"$PATCH\"" } ], + "vpn": [], "qemu_from_makefile": [ { "prompt_description_key": "QEMU - Sample dry-run (demo-vm, Ubuntu 24.04)", diff --git a/script/todo/todo.py b/script/todo/todo.py index e579ad1..eeed290 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -37,6 +37,8 @@ from script.todo.qemu_install import QemuInstallMixin from script.todo.qemu_manage import QemuManageMixin from script.todo.qemu_menu import QemuMenuMixin from script.todo.qemu_recover import QemuRecoverMixin +from script.todo.vpn_menu import VpnMenuMixin +from script.todo.kdbx_manager import KdbxManager from script.todo.todo_i18n import get_lang, lang_is_configured, set_lang, t from script.todo.version_manager import get_odoo_version @@ -100,6 +102,7 @@ class TODO( QemuAccessMixin, ProxmoxMenuMixin, LongTestMenuMixin, + VpnMenuMixin, ): def __init__(self): self.dir_path = None @@ -971,6 +974,12 @@ class TODO( "Deploy - Install NTFY notification server" ) }, + {"section": t("VPN & tunnels")}, + { + "prompt_description": t( + "VPN - Tunnels (L2TP/IPsec, WireGuard, OpenVPN...)" + ) + }, ] help_info = self.fill_help_info(choices) @@ -993,6 +1002,8 @@ class TODO( self.prompt_execute_proxmox() elif status == "7": self._deploy_ntfy_server() + elif status == "8": + self.prompt_execute_vpn() else: print(t("Command not found !")) @@ -4942,6 +4953,11 @@ class TODO( "Network performance request per second" ) }, + { + "prompt_description": t( + "VPN - Tunnels (L2TP/IPsec, WireGuard, OpenVPN...)" + ) + }, ] help_info = self.fill_help_info(choices) @@ -4954,6 +4970,8 @@ class TODO( self.generate_network_port_forwarding() elif status == "2": self.generate_network_performance_test() + elif status == "3": + self.prompt_execute_vpn() else: print(t("Command not found !")) diff --git a/script/todo/todo_example.json b/script/todo/todo_example.json index c56739c..243f2a7 100644 --- a/script/todo/todo_example.json +++ b/script/todo/todo_example.json @@ -8,6 +8,46 @@ "kdbx_key": "OpenAI api" } }, + "vpn": [ + { + "name": "client-acme", + "driver": "l2tp_ipsec", + "server": "vpn.acme.example", + "ppp_user": "ACME\\user", + "routes": [ + "10.20.0.0/16" + ], + "default_route": false, + "mtu": 1280, + "use_peer_dns": true, + "dns_search": "acme.example", + "probe": "10.20.0.1" + }, + { + "name": "client-beta", + "driver": "wireguard", + "server": "vpn.beta.example", + "port": 51820, + "wg_address": "10.7.0.2/32", + "wg_peer_key": "SGVsbG9Xb3JsZEV4YW1wbGVLZXkxMjM0NTY3ODkwYWI=", + "routes": [ + "10.7.0.0/24", + "192.168.80.0/24" + ], + "probe": "10.7.0.1" + }, + { + "name": "bastion-gamma", + "driver": "sshuttle", + "server": "user@bastion.gamma.example", + "port": 22, + "ssh_dns": true, + "routes": [ + "10.90.0.0/16" + ], + "probe": "10.90.0.1" + } + ], "instance": [ { "prompt_description": "Test - Instance de base minimale", diff --git a/script/todo/vpn_menu.py b/script/todo/vpn_menu.py new file mode 100644 index 0000000..2457e39 --- /dev/null +++ b/script/todo/vpn_menu.py @@ -0,0 +1,513 @@ +#!/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 : profils, secrets, montée, diagnostic. + +La frontière avec `script/vpn/` est nette : ici on DEMANDE (quel profil, +quelle adresse, quel PSK) et on affiche ; là-bas on décide et on exécute. Ce +fichier ne connaît ni ipsec.conf, ni xl2tpd, ni aucun chemin système. + +Deux chemins d'exécution, pour une raison : + +· les profils et les secrets sont manipulés EN PROCESSUS, par les modules + `script.vpn.profiles` et `script.vpn.vault` — le mot de passe maître du + coffre est déjà en mémoire ici, le redemander à un sous-processus serait + une saisie de plus à chaque geste ; +· le montage, la descente et le diagnostic passent par `script/vpn/vpn.py` + en sous-processus — ils durent, ils parlent, et ils appellent sudo. La + sortie en direct est ce qui rend un « ipsec up » suivable. +""" + +import getpass + +import click + +from script.todo.todo_i18n import t +from script.vpn import profiles +from script.vpn.drivers import DRIVERS, get_driver +from script.vpn.vault import VaultError, VpnVault, secrets_to_env + +# `-u` : sans lui, la sortie du script est mise en tampon par blocs dès +# qu'elle est redirigée, et une montée de tunnel de trente secondes +# n'afficherait rien avant la fin. +VPN_CLI = "./.venv.erplibre/bin/python -u ./script/vpn/vpn.py" + +# Nommé pour que la clé i18n tienne sur une ligne lisible — c'est la même +# chaîne que celle du CLI, et elle est longue parce qu'elle doit dire quoi +# faire, pas seulement que quelque chose ne va pas. +# Ce qu'on dit à qui n'a reçu du site qu'une passerelle et des +# identifiants : le profil est utilisable, et le premier montage dira +# lui-même quel réseau ajouter. +NO_ROUTE_NOTE = ( + "No network routed yet: this tunnel will only reach the remote host." + " Connect once — the address you get tells you which network to add." +) + +# La légende de l'étoile posée sur les technologies non éprouvées. Elle dit +# ce qui manque — la confrontation au terrain — et non que le code serait +# douteux : les tests unitaires, eux, sont là. +UNPROVEN_NOTE = "never mounted against a real server: only unit tests cover it" + +MASTER_PASSWORD_WARNING = ( + "The vault MASTER password is stored in the configuration in clear" + " text. Remove it and type it on demand." +) + + +# Les technologies se choisissent par LETTRE. Le menu qui précède numérote +# ses entrées ; une seconde liste numérotée juste après invite à retaper un +# numéro de menu — et c'est exactement ce qui s'est produit. La lettre dit +# « autre question ». +DRIVER_LETTERS = "abcdefghijklmnopqrstuvwxyz" + + +def match_driver(answer, names): + """Le pilote désigné par `answer`. + + Rend le nom du pilote, "" si rien ne correspond, ou la LISTE des + candidats quand c'est ambigu — le dire vaut mieux qu'en choisir un. + + Trois formes, dans cet ordre : la lettre affichée ; le rang, parce que + quelqu'un tapera un chiffre et qu'il a raison de le faire vu le menu qui + précède ; et un début de libellé, parce que devant « L2TP/IPsec PSK » on + tape « L ». « open » désigne deux pilotes : celui-là est refusé en le + nommant. + """ + answer = (answer or "").strip().lower() + if not answer: + return "" + if len(answer) == 1 and answer in DRIVER_LETTERS: + index = DRIVER_LETTERS.index(answer) + if index < len(names): + return names[index] + if answer.isdigit(): + index = int(answer) - 1 + return names[index] if 0 <= index < len(names) else "" + matches = [ + name + for name in names + if DRIVERS[name].label.lower().startswith(answer) + or name.startswith(answer) + ] + if len(matches) == 1: + return matches[0] + return matches or "" + + +class VpnMenuMixin: + # ------------------------------------------------------------------ + # Menu + # ------------------------------------------------------------------ + def prompt_execute_vpn(self): + print(f"🔐 {t('VPN tunnels: connect, profiles, vault secrets')}") + choices = [ + {"section": t("Connection")}, + {"prompt_description": t("VPN - Connect a profile")}, + {"prompt_description": t("VPN - Disconnect a profile")}, + {"prompt_description": t("VPN - Status and diagnosis")}, + {"section": t("Profiles & secrets")}, + {"prompt_description": t("VPN - Add or edit a profile")}, + {"prompt_description": t("VPN - Store secrets in the vault")}, + { + "prompt_description": t( + "VPN - Show the rendered configuration (dry-run)" + ) + }, + {"prompt_description": t("VPN - Delete a profile")}, + {"section": t("Host")}, + {"prompt_description": t("VPN - Install the client packages")}, + {"prompt_description": t("VPN - What can this machine do?")}, + ] + help_info = self.fill_help_info(choices) + + while True: + status = click.prompt(help_info) + print() + if status == "0": + return False + elif status == "1": + self._vpn_connect() + elif status == "2": + self._vpn_disconnect() + elif status == "3": + self._vpn_diagnose() + elif status == "4": + self._vpn_edit_profile() + elif status == "5": + self._vpn_store_secrets() + elif status == "6": + self._vpn_show_config() + elif status == "7": + self._vpn_delete_profile() + elif status == "8": + self._vpn_install() + elif status == "9": + self._vpn_check() + else: + print(t("Command not found !")) + + # ------------------------------------------------------------------ + # Actions déléguées au CLI + # ------------------------------------------------------------------ + def _vpn_cli(self, arguments, secrets_env=None): + self.execute.exec_command_live( + f"{VPN_CLI} {arguments}", + source_erplibre=False, + new_env=secrets_env or None, + ) + + def _vpn_secrets_env(self, name): + """Secrets du profil, prêts pour l'environnement du sous-processus. + + Le coffre est déjà ouvert ici : le faire rouvrir par `vpn.py` ferait + retaper le mot de passe maître deux fois par connexion, puisqu'un + essai à blanc précède le montage. Rend {} quand le coffre n'est pas + joignable — `vpn.py` demandera alors lui-même, et le dira. + """ + profile = profiles.load(name) + driver_cls = get_driver(profile["driver"]) if profile else None + if driver_cls is None or not driver_cls.secret_fields: + return {} + fields = tuple(key for key, _, _ in driver_cls.secret_fields) + vault = VpnVault(self.config_file, self.kdbx_manager) + if not vault.vault_path(): + return {} + try: + values = vault.read(profiles.secret_title(name), fields=fields) + except VaultError as error: + print(f"! {error}") + return {} + return secrets_to_env({key: values.get(key, "") for key in fields}) + + def _vpn_connect(self): + name = self._vpn_select_profile() + if not name: + return + # Lu UNE fois pour les deux exécutions qui suivent. + secrets_env = self._vpn_secrets_env(name) + # Le plan d'abord, l'exécution ensuite : monter un tunnel réécrit + # /etc/ipsec.conf et la table de routage. Le voir avant coûte une + # touche et évite de découvrir une faute de frappe dans un journal. + self._vpn_cli(f"up --profile {name} --dry-run", secrets_env) + if not self._is_yes(input(f"\n{t('Run this plan? (y/N): ')}")): + return + self._vpn_cli(f"up --profile {name}", secrets_env) + + def _vpn_disconnect(self): + name = self._vpn_select_profile() + if name: + self._vpn_cli(f"down --profile {name}") + + def _vpn_diagnose(self): + name = self._vpn_select_profile() + if name: + self._vpn_cli(f"diagnose --profile {name}") + + def _vpn_show_config(self): + name = self._vpn_select_profile() + if name: + self._vpn_cli( + f"up --profile {name} --dry-run", + self._vpn_secrets_env(name), + ) + + def _vpn_check(self): + self._vpn_cli("check") + + def _vpn_install(self): + driver_cls = self._vpn_pick_driver(None) + if driver_cls is None: + return + print(f"\n{t('The installation requires sudo.')}") + self._vpn_cli(f"install --driver {driver_cls.name}") + + # ------------------------------------------------------------------ + # Profils + # ------------------------------------------------------------------ + def _vpn_select_profile(self): + """Nom du profil choisi, "" si l'utilisateur renonce.""" + all_profiles = [profiles.with_defaults(p) for p in profiles.load_all()] + if not all_profiles: + print(t("No VPN profile yet: create one first.")) + return "" + for index, profile in enumerate(all_profiles, start=1): + target = ( + t("all traffic") + if profile["default_route"] + else ", ".join(profile["routes"]) + ) + print( + f"[{index}] {profile['name']:<20}" + f" {profile['server']:<26} {target}" + ) + answer = input(f"{t('Profile number (0 to go back)')} : ").strip() + if not answer.isdigit() or not 1 <= int(answer) <= len(all_profiles): + if answer not in ("0", ""): + print(t("Unknown choice.")) + return "" + return all_profiles[int(answer) - 1]["name"] + + def _vpn_edit_profile(self): + """Crée ou modifie un profil, quelle que soit la technologie. + + Les questions viennent du PILOTE (`form_fields`) : ce menu ne sait + pas qu'un profil L2TP a un utilisateur PPP ni qu'un profil WireGuard + a une clé de pair. Ajouter une technologie n'ajoute donc pas une + ligne ici. + + Une réponse vide garde la valeur actuelle : modifier une seule route + ne doit pas obliger à ressaisir tout le reste. + """ + name = input(f"{t('Profile name (lowercase, digits, - or _)')} : ") + name = name.strip() + if not name: + return + current = profiles.load(name) or {"name": name} + driver_cls = self._vpn_pick_driver(current.get("driver")) + if driver_cls is None: + return + + # Les défauts DU PILOTE CHOISI, pour que chaque question ait un + # défaut sensé même sur un profil qui change de technologie. + draft = profiles.with_defaults(dict(current, driver=driver_cls.name)) + + draft["server"] = self._vpn_ask( + t(driver_cls.server_label), draft.get("server", "") + ) + # L'identité d'abord, le routage ensuite : c'est l'ordre du document + # que le site remet — passerelle, utilisateur, mot de passe, clé — + # et le routage est une question à part, à laquelle ce document ne + # répond souvent pas. + self._vpn_ask_fields(draft, driver_cls, advanced=False) + draft["routes"] = self._vpn_ask( + t("Networks to reach, comma-separated"), + ", ".join(draft.get("routes", [])), + ) + draft["default_route"] = self._vpn_ask_flag( + t("Send ALL traffic through the tunnel?"), + draft.get("default_route", False), + ) + draft["probe"] = self._vpn_ask( + t("Witness address reachable only through the tunnel (optional)"), + draft.get("probe", ""), + ) + if self._is_yes(input(f"{t('Advanced settings? (y/N)')} : ")): + if driver_cls.uses_mtu: + draft["mtu"] = self._vpn_ask( + t("MTU"), str(draft.get("mtu", 1280)) + ) + self._vpn_ask_fields(draft, driver_cls, advanced=True) + + try: + saved = profiles.save(draft) + except profiles.ProfileError as error: + print(f"\n✗ {t('Profile refused: ')}{error}") + return + print(f"\n✓ {t('Profile saved: ')}{saved['name']}") + if not saved["routes"] and not saved["default_route"]: + print(f" {t(NO_ROUTE_NOTE)}") + if driver_cls.secret_fields: + print(f" {t('Next step: store its secrets in the vault.')}") + else: + print( + f" {t('No secret to store: this one authenticates over SSH.')}" + ) + + def _vpn_ask_fields(self, draft, driver_cls, advanced): + """Déroule les champs déclarés par le pilote.""" + for key, label, kind, is_advanced in driver_cls.form_fields: + if bool(is_advanced) != advanced: + continue + if kind == "flag": + draft[key] = self._vpn_ask_flag( + t(label), draft.get(key, False) + ) + else: + draft[key] = self._vpn_ask( + t(label), str(draft.get(key, "") or "") + ) + + @staticmethod + def _vpn_pick_driver(current): + """La technologie, par lettre, avec un conseil par ligne. + + C'est la seule décision du formulaire où l'utilisateur a besoin + d'aide : le reste se déduit de ce que le site lui a donné. + + Une étoile marque les technologies qu'aucun serveur réel n'a + encore validées, et une légende dit ce qu'elle signifie : la liste + montre autrement cinq choix d'apparence égale. + + `[0] Retour` est là comme dans tous les menus de ce CLI : sans lui, + on est coincé dans le formulaire dès qu'on a tapé un nom de profil. + """ + names = list(DRIVERS) + if len(names) == 1: + return DRIVERS[names[0]] + default = current if current in DRIVERS else names[0] + print(f"\n{t('Which technology?')}") + unproven = False + for letter, name in zip(DRIVER_LETTERS, names): + driver_cls = DRIVERS[name] + mark = " ←" if name == default else "" + # L'étoile occupe une colonne à elle : sans cela, les lignes + # marquées décaleraient leur conseil et la liste se lirait mal. + star = " " if driver_cls.proven else "*" + unproven = unproven or not driver_cls.proven + print( + f"[{letter}] {driver_cls.label:<16}{star}" + f" {t(driver_cls.hint)}{mark}" + ) + if unproven: + print(f" * {t(UNPROVEN_NOTE)}") + print(f"[0] {t('Back')}") + default_letter = DRIVER_LETTERS[names.index(default)] + answer = input( + f"{t('Choice')} [{default_letter} = {DRIVERS[default].label}] : " + ).strip() + if not answer: + return DRIVERS[default] + if answer == "0": + return None + chosen = match_driver(answer, names) + if isinstance(chosen, list): + labels = ", ".join(DRIVERS[name].label for name in chosen) + print(f"{t('Several technologies match: ')}{labels}") + return None + if not chosen: + print(t("Unknown choice.")) + return None + return DRIVERS[chosen] + + def _vpn_delete_profile(self): + name = self._vpn_select_profile() + if not name: + return + if not self._is_yes( + input(f"{t('Delete profile')} « {name} » ? (y/N) : ") + ): + return + if profiles.delete(name): + print(f"✓ {t('Profile deleted.')}") + print(f" {t('Its vault entry is kept: delete it in KeePassXC.')}") + else: + message = t( + "Not deletable here: this profile comes from a shared" + " configuration file." + ) + print(f"✗ {message}") + + # ------------------------------------------------------------------ + # Secrets + # ------------------------------------------------------------------ + def _vpn_store_secrets(self): + name = self._vpn_select_profile() + if not name: + return + profile = profiles.load(name) + driver_cls = get_driver(profile["driver"]) + if driver_cls is None: + print(f"✗ {t('Unknown driver: ')}{profile['driver']}") + return + if not driver_cls.secret_fields: + print( + f"{t('No secret to store: this one authenticates over SSH.')}" + ) + return + vault = VpnVault(self.config_file, self.kdbx_manager) + try: + path = vault.ensure_vault(ask=input) + except VaultError as error: + print(f"✗ {error}") + return + if not path: + print(t("No vault: nothing stored.")) + return + if vault.master_password_is_stored(): + print(f"\n! {t(MASTER_PASSWORD_WARNING)}") + + title = profiles.secret_title(name) + fields = tuple(key for key, _, _ in driver_cls.secret_fields) + # Lu AVANT les invites, pour deux raisons : le mot de passe maître + # est alors demandé avant qu'on tape des secrets, et non après ; et + # chaque invite peut dire s'il y a déjà quelque chose derrière. + # « Une réponse vide garde la valeur en place » est un piège quand + # il n'y a rien en place. + try: + existing = vault.read(title, fields=fields) + except VaultError as error: + print(f"✗ {error}") + return + + print(f"\n{t('Vault entry')} : {title}") + print(f"{t('An empty answer keeps the stored value.')}\n") + values = {} + if driver_cls.user_field: + # Recopié pour que le coffre reste LISIBLE dans KeePassXC ; le + # profil reste la source de vérité de l'identifiant. + values["username"] = profile.get(driver_cls.user_field, "") + for key, label, _required in driver_cls.secret_fields: + state = t("already set") if existing.get(key) else t("empty") + secret = self._vpn_ask_secret(f"{t(label)} [{state}]") + if secret is None: + return + if secret: + values[key] = secret + try: + vault.write(title, values) + except VaultError as error: + print(f"✗ {error}") + return + print(f"\n✓ {t('Secrets stored in the vault.')}") + + # Ce qui reste vide et qui est OBLIGATOIRE : le dire ici, pas au + # premier montage raté. + absents = [ + t(label) + for key, label, required in driver_cls.secret_fields + if required and not (values.get(key) or existing.get(key)) + ] + if absents: + print( + f"✗ {t('Still missing, the tunnel will not come up: ')}" + f"{', '.join(absents)}" + ) + + @staticmethod + def _vpn_ask_secret(label): + """Un secret, saisi deux fois, jamais affiché. + + Deux fois parce qu'une faute de frappe dans un PSK ne se voit pas : + elle ressort en « no matching proposal » côté IKE, trois étages plus + loin, et fait chercher au mauvais endroit pendant une heure. + + Rend "" pour « garder la valeur en place », None pour renoncer. + """ + first = getpass.getpass(f"{label} : ") + if not first: + return "" + if first != getpass.getpass(f"{t('Confirm')} : "): + print(f"✗ {t('The two entries differ, nothing stored.')}") + return None + return first + + # ------------------------------------------------------------------ + @staticmethod + def _vpn_ask_flag(label, current): + """Question oui/non dont le défaut est la valeur ACTUELLE. + + Une réponse vide garde ce qui est en place : rééditer un profil pour + changer une route ne doit pas remettre le mode de routage à zéro. + """ + answer = input(f"{label} [{'O/n' if current else 'o/N'}] : ") + answer = answer.strip().lower() + if not answer: + return bool(current) + return answer in ("y", "yes", "o", "oui") + + @staticmethod + def _vpn_ask(label, default): + """Question à réponse par défaut. Vide = on garde `default`.""" + shown = f" [{default}]" if default else "" + answer = input(f"{label}{shown} : ").strip() + return answer or default diff --git a/test/test_vpn_menu.py b/test/test_vpn_menu.py new file mode 100644 index 0000000..665653c --- /dev/null +++ b/test/test_vpn_menu.py @@ -0,0 +1,601 @@ +#!/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()