From 315e45e5f0a11e1d2556ac85264798c1b05e6e6d Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Fri, 4 Sep 2026 03:43:02 +0000 Subject: [PATCH] =?UTF-8?q?[ADD]=20menu=20vpn=20:=20cr=C3=A9er=20un=20prof?= =?UTF-8?q?il,=20d=C3=A9poser=20les=20secrets,=20monter=20le=20tunnel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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 --- script/todo/todo.json | 1 + script/todo/todo.py | 18 + script/todo/todo_example.json | 40 +++ script/todo/vpn_menu.py | 513 +++++++++++++++++++++++++++++ test/test_vpn_menu.py | 601 ++++++++++++++++++++++++++++++++++ 5 files changed, 1173 insertions(+) create mode 100644 script/todo/vpn_menu.py create mode 100644 test/test_vpn_menu.py 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()