#!/usr/bin/env python3 """Import one verified TOTP from a local Yandex Key backup into Keychain.""" import argparse import base64 import binascii from contextlib import ExitStack import getpass import json import os from pathlib import Path import re import shlex import signal import stat import sys import tempfile import unicodedata from urllib.parse import parse_qs, urlsplit MAX_BACKUP_BYTES = 4 * 1024 * 1024 class ImportFailure(ValueError): """Only fixed, secret-free messages may cross the CLI boundary.""" def display_name(value, secrets=()): value = "".join(c for c in value if not unicodedata.category(c).startswith("C")) if "otpauth:" in value.lower() or any(secret and secret in value.upper() for secret in secrets): return "" return value[:120] def read_backup(path): try: # Nonblocking open also lets us reject a mistakenly selected FIFO/device. fd = os.open(path, os.O_RDONLY | os.O_NONBLOCK) with os.fdopen(fd, "rb") as stream: if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode): raise ImportFailure("Выберите обычный файл резервной копии.") data = stream.read(MAX_BACKUP_BYTES + 1) if len(data) > MAX_BACKUP_BYTES: raise ImportFailure("Файл превышает лимит импорта 4 МиБ.") entries = json.loads(data.decode("utf-8-sig")) except (OSError, UnicodeError, ValueError) as error: if isinstance(error, ImportFailure): raise raise ImportFailure("Не удалось прочитать локальный JSON-экспорт Яндекс Ключа.") from None if not isinstance(entries, list) or not 1 <= len(entries) <= 1000: raise ImportFailure("Ожидается резервная копия с 1–1000 записями.") if any(not isinstance(entry, dict) or not isinstance(entry.get("techInfo"), str) for entry in entries): raise ImportFailure("Неверная структура резервной копии Яндекс Ключа.") return entries def normalize_secret(value): if not isinstance(value, str): raise ImportFailure("Запись не содержит корректного секрета 2FA.") value = value.strip().upper().rstrip("=") if not re.fullmatch(r"[A-Z2-7]{16,1024}", value): raise ImportFailure("Некорректный BASE32-секрет в выбранной записи.") try: decoded = base64.b32decode(value + "=" * (-len(value) % 8)) except binascii.Error: raise ImportFailure("Некорректный BASE32-секрет в выбранной записи.") from None if base64.b32encode(decoded).decode().rstrip("=") != value: raise ImportFailure("Некорректный BASE32-секрет в выбранной записи.") return value def selected_secret(entry): try: uri = urlsplit(entry["techInfo"]) query = parse_qs(uri.query, keep_blank_values=True, max_num_fields=100) except ValueError: raise ImportFailure("Некорректные параметры OTP в выбранной записи.") from None if uri.scheme != "otpauth" or uri.netloc != "totp" or uri.fragment: raise ImportFailure("Поддерживается только TOTP; HOTP и собственные коды Яндекса не поддерживаются.") if set(query) - {"secret", "algorithm", "digits", "period", "issuer", "account_id", "created_at", "creation_type", "image", "name", "uid"}: raise ImportFailure("Выбранная запись содержит неподдерживаемые параметры OTP.") for key in ("secret", "algorithm", "digits", "period"): if key in query and len(query[key]) != 1: raise ImportFailure("В записи повторяются параметры OTP.") if (query.get("algorithm", ["SHA1"])[0].upper() != "SHA1" or query.get("digits", ["6"])[0] != "6" or query.get("period", ["30"])[0] != "30" or "counter" in query): raise ImportFailure("Поддерживается только TOTP: SHA1, 6 цифр, период 30 секунд.") secret = normalize_secret(query.get("secret", [""])[0]) if "secret" in entry and normalize_secret(entry["secret"]) != secret: raise ImportFailure("Секрет и параметры выбранной записи не совпадают.") return secret def activate(secret, username, config_file): import keyring # Stage settings before touching Keychain; retain all unrelated local settings. text = config_file.read_text() if config_file.exists() else "" saved = None for line in text.splitlines(): match = re.fullmatch(r'(?:export[ \t]+)?LEMANA_VPN_USERNAME=("(?:\\.|[^"\\])*"|\'[^\']*\'|[^\n]*)[ \t]*(?:#.*)?', line) if match: raw = match[1] if raw.startswith('"'): saved = re.sub(r'\\([$`"\\])', lambda m: m[1], raw[1:-1]) else: values = shlex.split(raw, comments=True) saved = values[0] if len(values) == 1 else None if saved != username: raise ImportFailure("Сохранённый LDAP username изменился или не настроен. Сначала выполните vpn --configure-login.") original_text = text for key, value in (("LEMANA_VPN_MODE", "keychain-auto"), ("LEMANA_VPN_LOCAL_OTP", "1"), ("LEMANA_VPN_AUTO_SOURCE", "keychain"), ("LEMANA_VPN_AUTH_PROFILE", "auto"), ("LEMANA_VPN_CREDENTIAL_SOURCE", "keychain"), ("LEMANA_VPN_USE_BITWARDEN", "0")): assignment = f'{key}="{value}"\n' pattern = r"(?m)^(?:export[ \t]+)?" + key + r"=[^\n]*(?:\n|$)" text = re.sub(pattern, "", text) text = text.rstrip("\n") + "\n" + assignment config_file.parent.mkdir(parents=True, exist_ok=True) fd, staging = tempfile.mkstemp(prefix=".otp-env-", dir=config_file.parent) try: with os.fdopen(fd, "w") as stream: stream.write(text) account = "totp/" + username previous = keyring.get_password("openconnect-lite", account) try: keyring.set_password("openconnect-lite", account, secret) if keyring.get_password("openconnect-lite", account) != secret: raise ImportFailure("Keychain не подтвердил сохранение секрета; локальный код не включён.") if config_file.read_text() != original_text: raise ImportFailure("Настройки изменились во время импорта; повторите импорт.") os.replace(staging, config_file) except BaseException: if not os.path.exists(staging): # replace is the commit point: both stores now contain the new state. raise ImportFailure("Настройка уже сохранена. Проверьте её командой vpn --status.") from None # An interrupted activation must not silently replace working credentials. try: if previous is None: if keyring.get_password("openconnect-lite", account) is not None: keyring.delete_password("openconnect-lite", account) else: keyring.set_password("openconnect-lite", account, previous) if keyring.get_password("openconnect-lite", account) != previous: raise RuntimeError except BaseException: raise ImportFailure("Импорт не завершён, вернуть прежний секрет не удалось. Проверьте запись 2FA в Keychain.") from None raise finally: if os.path.exists(staging): os.unlink(staging) def configure(username, config_file, ask, ask_secret, say): import pyotp say("В Яндекс Ключе: Настройки → Резервная копия данных → Сохранить в файл.") entries = read_backup(Path(ask("Путь к файлу резервной копии: ").strip()).expanduser()) secrets = set() for entry in entries: if isinstance(entry.get("secret"), str) and entry["secret"]: secrets.add(entry["secret"].strip().upper().rstrip("=")) try: for value in parse_qs(urlsplit(entry["techInfo"]).query, max_num_fields=100).get("secret", []): if value: secrets.add(value.strip().upper().rstrip("=")) except ValueError: pass labels = [] for index, entry in enumerate(entries, 1): name = entry.get("name", "") label = display_name(name, secrets) if isinstance(name, str) else "" labels.append(label or f"Запись {index}") say(f"{index}) {labels[-1]}") choice = ask("Номер записи Lemana/VPN (Enter — отмена): ").strip() if not re.fullmatch(r"[0-9]{1,4}", choice) or not 1 <= int(choice) <= len(entries): raise ImportFailure("Запись не выбрана; настройки не изменены.") entry = entries[int(choice) - 1] secret = selected_secret(entry) say(f"Запись: {labels[int(choice) - 1]}; LDAP: {display_name(username)}.") say("Будет включён автоматический вход с ключом на этом Mac: скрытый браузер, подстановка данных и отправка формы без Bitwarden.") code = ask_secret("Введите текущие 6 цифр из этой записи на телефоне для проверки и сохранения: ").strip() if not re.fullmatch(r"[0-9]{6}", code) or not pyotp.TOTP(secret).verify(code, valid_window=1): raise ImportFailure("Код не совпал. Проверьте выбранную запись и время на устройствах; настройки не изменены.") activate(secret, username, config_file) say("Готово: одноразовый код сохранён в Keychain и будет подставляться при следующем подключении.") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--username", required=True) parser.add_argument("--config", required=True, type=Path) args = parser.parse_args() def interrupted(_signum, _frame): raise KeyboardInterrupt signal.signal(signal.SIGTERM, interrupted) try: import toml from xdg.BaseDirectory import load_first_config # Match the actual openconnect-lite read path, including XDG overrides. config_dir = load_first_config("openconnect-lite") if not config_dir or toml.load(Path(config_dir) / "config.toml").get("credentials", {}).get("username") != args.username: raise ImportFailure("LDAP username в профиле SSO отличается. Согласуйте логин через vpn --configure-login или установщик.") with ExitStack() as handles: terminal_input, terminal = sys.stdin, sys.stderr if not sys.stdin.isatty(): terminal_fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY) terminal_input = handles.enter_context(os.fdopen(os.dup(terminal_fd), "r", encoding="utf-8")) terminal = handles.enter_context(os.fdopen(terminal_fd, "w", encoding="utf-8")) def ask(prompt): terminal.write(prompt) terminal.flush() line = terminal_input.readline() if not line: raise EOFError return line.rstrip("\n") configure(args.username, args.config, ask, lambda prompt: getpass.getpass(prompt, stream=terminal), lambda message: print(message, file=terminal, flush=True)) except ImportFailure as error: print(str(error), file=sys.stderr) return 1 except (EOFError, KeyboardInterrupt): print("Импорт отменён.", file=sys.stderr) return 1 except Exception: # Backend/JSON exceptions may contain sensitive data; never print them. print("Не удалось завершить импорт. Запустите команду в Terminal и проверьте доступ к файлу и Keychain.", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())