240 lines
12 KiB
Python
240 lines
12 KiB
Python
#!/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())
|