refactor: clear SSH auth flow — passphrase is local only, target password is separate; interactive ssh-copy-id with visible prompt; 90 lines removed, logic simplified

This commit is contained in:
2026-05-22 23:53:15 +04:00
parent bd16e339c0
commit 3ee5104302
2 changed files with 98 additions and 179 deletions

View File

@@ -379,16 +379,15 @@ def do_transfer_offer():
if key_path: if key_path:
state.set_stage("TRANSFER", target_host=host, target_user=user, target_port=port_int, ssh_key=key_path) state.set_stage("TRANSFER", target_host=host, target_user=user, target_port=port_int, ssh_key=key_path)
else: else:
# Нет SSH-ключа — спрашиваем пароль и пробуем sshpass # Нет SSH-ключа — запрашиваем пароль от TARGET сервера
password = prompt_password(f"Пароль для {user}@{host} (Enter для отмены)") password = prompt_password(f"Пароль от target-сервера {user}@{host} (Enter для отмены)")
if not password: if not password:
state.set_error("ssh_key_setup", "", "Пароль не введён", suggestion="Запустите docker-migrate --resume или настройте SSH-ключ") state.set_error("ssh_key_setup", "", "Пароль не введён", suggestion="Запустите docker-migrate --resume или настройте SSH-ключ на target")
raise RuntimeError("Пароль не введён. Установите SSH-ключ или введите пароль.") raise RuntimeError("Пароль от target не введён.")
# Проверяем/устанавливаем sshpass
from transfer.ssh import ensure_sshpass from transfer.ssh import ensure_sshpass
if not ensure_sshpass(): if not ensure_sshpass():
state.set_error("ssh_key_setup", "", "sshpass не найден и не удалось установить", suggestion="Установите sshpass: apt-get install -y sshpass") state.set_error("ssh_key_setup", "", "sshpass не найден", suggestion="apt-get install -y sshpass")
raise RuntimeError("sshpass не найден. Установите: apt-get install -y sshpass") raise RuntimeError("sshpass не установлен.")
state.set_stage("TRANSFER", target_host=host, target_user=user, target_port=port_int, ssh_key=None, ssh_password=password) state.set_stage("TRANSFER", target_host=host, target_user=user, target_port=port_int, ssh_key=None, ssh_password=password)
from transfer.transfer import do_transfer from transfer.transfer import do_transfer

View File

@@ -1,95 +1,56 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
ssh.py — SSH-диагностика: поиск ключей, проверка подключения, генерация временных ключей ssh.py — SSH-подготовка для transfer на target-сервер.
Логика: найти ключ → проверить подключение к target.
Если ключ зашифрован (passphrase) — пропускаем (пользователь должен был разблокировать через ssh-agent).
Если нет ключа или не подходит — меню: пароль target / новый ключ.
""" """
import os import os
import glob import glob
import re
from core.color import info, warn, success, error as cerror, prompt, confirm from core.color import info, warn, success, error as cerror, prompt, confirm
from core.runner import run, exists from core.runner import run, exists
from core import state
_SSH_DIR = os.path.expanduser("~/.ssh") _SSH_DIR = os.path.expanduser("~/.ssh")
def list_private_keys(): def list_private_keys():
"""Ищет приватные SSH-ключи в ~/.ssh/""" """Ищет приватные SSH-ключи в ~/.ssh/ (id_* без .pub)."""
if not os.path.isdir(_SSH_DIR): if not os.path.isdir(_SSH_DIR):
return [] return []
keys = [] keys = []
# Стандартные паттерны приватных ключей (не .pub)
for pattern in ("id_rsa", "id_ed25519", "id_ecdsa", "id_dsa"): for pattern in ("id_rsa", "id_ed25519", "id_ecdsa", "id_dsa"):
for f in glob.glob(os.path.join(_SSH_DIR, pattern)): for f in glob.glob(os.path.join(_SSH_DIR, pattern)):
# Убедимся что это не .pub файл
if not f.endswith(".pub") and os.path.isfile(f): if not f.endswith(".pub") and os.path.isfile(f):
# Проверим что есть соответствующий .pub
pub = f + ".pub" pub = f + ".pub"
has_pub = os.path.isfile(pub)
keys.append({ keys.append({
"private": f, "private": f,
"public": pub if has_pub else None, "public": pub if os.path.isfile(pub) else None,
"type": os.path.basename(f).replace("id_", ""), "type": os.path.basename(f).replace("id_", ""),
"has_pub": has_pub,
}) })
# Ищем и другие файлы без расширения (которые содержат PRIVATE KEY)
for f in glob.glob(os.path.join(_SSH_DIR, "*")):
if os.path.isfile(f) and not f.endswith(".pub") and not f.endswith(".config"):
try:
with open(f, "r", encoding="utf-8", errors="ignore") as fh:
first = fh.read(100)
if "PRIVATE KEY" in first or "OPENSSH" in first:
pub = f + ".pub"
has_pub = os.path.isfile(pub)
if f not in [k["private"] for k in keys]:
keys.append({
"private": f,
"public": pub if has_pub else None,
"type": "unknown",
"has_pub": has_pub,
})
except Exception:
pass
return keys return keys
def check_ssh_connectivity(host, user, port, key_path=None, timeout=10): def check_ssh_keyless(host, user, port, key_path, timeout=10):
"""Проверяет SSH-соединение. Возвращает (ok, stdout, stderr).""" """Проверяет подключение к target с данным ключом. БЕЗ passphrase-промптов."""
if not exists("ssh"): if not exists("ssh"):
return (False, "", "ssh не найден") return (False, "ssh не найден")
identity = f"-i '{key_path}'" if key_path and os.path.isfile(key_path) else "" identity = f"-i '{key_path}'" if key_path and os.path.isfile(key_path) else ""
cmd = f"ssh {identity} -p {port} -o ConnectTimeout={timeout} -o BatchMode=yes -o StrictHostKeyChecking=accept-new {user}@{host} 'echo migrate-ok'" cmd = f"ssh {identity} -p {port} -o ConnectTimeout={timeout} -o BatchMode=yes -o StrictHostKeyChecking=accept-new {user}@{host} 'echo migrate-ok'"
r = run(cmd, check=False, capture=True, timeout=30) r = run(cmd, check=False, capture=True, timeout=30)
if r.returncode == 0 and "migrate-ok" in r.stdout: if r.returncode == 0 and "migrate-ok" in r.stdout:
return (True, r.stdout, r.stderr) return (True, "OK")
return (False, r.stdout, r.stderr) # Разбираем ошибку
err = (r.stderr or "").strip()
if "passphrase" in err.lower():
def test_keys_against_target(host, user, port, keys): return (False, "PASSPHRASE")
"""Перебирает ключи и проверяет какой работает с target.""" if "permission denied" in err.lower() or "too many auth" in err.lower():
if not keys: return (False, "DENIED")
return None return (False, err[:100])
info(f"Проверяем SSH-подключение к {user}@{host}:{port} с найденными ключами ...")
working = None
for k in keys:
label = os.path.basename(k["private"])
ok, out, err = check_ssh_connectivity(host, user, port, key_path=k["private"])
if ok:
success(f" ✓ Ключ '{label}' работает!")
working = k
break
else:
# Проверим, не passphrase ли
if "passphrase" in err.lower() or "password" in err.lower():
warn(f" ! Ключ '{label}' требует passphrase (зашифрован). Можно ввести passphrase позже.")
else:
info(f" ✗ Ключ '{label}' не подходит ({err.strip()[:60]}).")
return working
def generate_temp_keypair(host_label): def generate_temp_keypair(host_label):
"""Генерирует временную пару ed25519 в /tmp/ для миграции.""" """Генерирует временную пару ed25519 в /tmp/."""
temp_dir = "/tmp/docker-migrate-ssh-keys" temp_dir = "/tmp/docker-migrate-ssh-keys"
os.makedirs(temp_dir, exist_ok=True) os.makedirs(temp_dir, exist_ok=True)
priv = os.path.join(temp_dir, f"migrate_{host_label}") priv = os.path.join(temp_dir, f"migrate_{host_label}")
@@ -97,65 +58,52 @@ def generate_temp_keypair(host_label):
info("Генерируем временную SSH-пару ed25519 ...") info("Генерируем временную SSH-пару ed25519 ...")
run(f"ssh-keygen -t ed25519 -C 'docker-migrate-temp' -f '{priv}' -N ''", check=False) run(f"ssh-keygen -t ed25519 -C 'docker-migrate-temp' -f '{priv}' -N ''", check=False)
if os.path.isfile(priv) and os.path.isfile(pub): if os.path.isfile(priv) and os.path.isfile(pub):
success(f"Временная пара создана: {priv}")
# Устанавливаем безопасные права
os.chmod(priv, 0o600) os.chmod(priv, 0o600)
success(f"Временная пара создана: {priv}")
return {"private": priv, "public": pub, "type": "ed25519", "temp": True} return {"private": priv, "public": pub, "type": "ed25519", "temp": True}
return None return None
def ssh_copy_id(host, user, port, pubkey_path): def ssh_copy_id(host, user, port, pubkey_path):
"""Копирует публичный ключ на target через ssh-copy-id. Интерактивно, т.к. запрашивает пароль target.""" """ssh-copy-id на target. Интерактивно (пользователь вводит пароль target)."""
if exists("ssh-copy-id"): if not exists("ssh-copy-id"):
warn("⚠ Сейчас запустится ssh-copy-id — на target запросит ПАРОЛЬ. Введите его ниже ↓") return False
# Запускаем без capture_output, чтобы prompt пароля был виден пользователю warn("⚠ Сейчас запустится ssh-copy-id — введите ПАРОЛЬ от target-сервера ↓")
r = run( r = run(
f"ssh-copy-id -p {port} -o ConnectTimeout=30 -o StrictHostKeyChecking=accept-new -i '{pubkey_path}' {user}@{host}", f"ssh-copy-id -p {port} -o ConnectTimeout=30 -o StrictHostKeyChecking=accept-new -i '{pubkey_path}' {user}@{host}",
check=False, check=False, capture=False, timeout=120
capture=False, )
timeout=120 if r.returncode == 0:
) success("Ключ добавлен на target")
if r.returncode == 0: return True
success("Ключ добавлен на target через ssh-copy-id") warn(f"ssh-copy-id не удался (код: {r.returncode})")
return True
else:
warn(f"ssh-copy-id не удался (код: {r.returncode})")
return False return False
def manual_copy_instructions(host, user, port, pubkey_path): def manual_copy_instructions(host, user, port, pubkey_path):
"""Выводит инструкции по ручному добавлению ключа.""" """Инструкции по ручному добавлению ключа на target."""
print() print()
cerror("SSH-подключение к target не удалось ни с одним ключом.") cerror("SSH-подключение к target не удалось.")
print() info("Варианты:")
info("Варианты решения:")
print(f" {yellow('1')} Добавьте публичный ключ на target вручную:")
try: try:
with open(pubkey_path, "r") as f: with open(pubkey_path, "r") as f:
pubkey_content = f.read().strip() pk = f.read().strip()
except Exception: except Exception:
pubkey_content = "(не удалось прочитать)" pk = "(не удалось прочитать)"
print(f" На {user}@{host} выполните:") print(f" 1) Вручную на target: mkdir -p ~/.ssh; echo '{pk}' >> ~/.ssh/authorized_keys")
print(f" mkdir -p ~/.ssh && chmod 700 ~/.ssh") print(f" 2) Или: ssh-copy-id -p {port} -i {pubkey_path} {user}@{host}")
print(f" echo '{pubkey_content}' >> ~/.ssh/authorized_keys") print(f" 3) Потом: docker-migrate --resume")
print(f" chmod 600 ~/.ssh/authorized_keys")
print()
print(f" {yellow('2')} Или скопируйте через ssh-copy-id (если есть парольный доступ):")
print(f" ssh-copy-id -p {port} -i {pubkey_path} {user}@{host}")
print()
print(f" {yellow('3')} После добавления ключа запустите миграцию заново:")
print(f" docker-migrate --resume")
print() print()
def ensure_sshpass(): def ensure_sshpass():
"""Проверяет наличие sshpass, при необходимости ставит через apt.""" """Устанавливает sshpass если нужно."""
if exists("sshpass"): if exists("sshpass"):
return True return True
info("sshpass не найден. Устанавливаем ...") info("sshpass не найден. Устанавливаем ...")
r1 = run("apt-get update", check=False, timeout=60) r1 = run("apt-get update", check=False, timeout=60)
if r1.returncode != 0: if r1.returncode != 0:
warn(f"apt-get update не удался: {r1.stderr.strip()[:120]}") warn(f"apt-get update: {r1.stderr.strip()[:120]}")
r = run("apt-get install -y sshpass", check=False, timeout=120) r = run("apt-get install -y sshpass", check=False, timeout=120)
if r.returncode != 0: if r.returncode != 0:
warn(f"Не удалось установить sshpass: {r.stderr.strip()[:120]}") warn(f"Не удалось установить sshpass: {r.stderr.strip()[:120]}")
@@ -163,99 +111,71 @@ def ensure_sshpass():
return exists("sshpass") return exists("sshpass")
def try_key_with_passphrase(key_path, passphrase):
"""Проверяет, работает ли ключ с введённым passphrase через ssh-agent или expect."""
# Способ 1: ssh-agent + ssh-add
from core.runner import run
import subprocess
try:
# Запускаем ssh-agent и добавляем ключ
agent_out = subprocess.run(["ssh-agent", "-s"], capture_output=True, text=True)
if agent_out.returncode != 0:
return False
# Парсим переменные окружения из ssh-agent -s
env_lines = agent_out.stdout.strip().split(";")
agent_env = {}
for ln in env_lines:
if "=" in ln and "SSH_" in ln:
k, v = ln.strip().split("=", 1)
agent_env[k] = v
# ssh-add с passphrase через PIPE
add_proc = subprocess.Popen(
["ssh-add", key_path],
env={**os.environ, **agent_env},
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
stdout, stderr = add_proc.communicate(input=passphrase + "\n", timeout=10)
if add_proc.returncode == 0:
return True
# Если не сработало — убиваем ssh-agent
subprocess.run(["ssh-agent", "-k"], env={**os.environ, **agent_env}, capture_output=True)
except Exception:
pass
return False
def pick_or_setup_ssh_key(host, user, port): def pick_or_setup_ssh_key(host, user, port):
""" """
Главная функция SSH-подготовки. Возвращает путь к приватному ключу. Возвращает путь к рабочему приватному ключу для target.
Если None — значит используем пароль (sshpass). Если None — значит нужен пароль target (sshpass).
""" """
# 1. Найти существующие ключи
keys = list_private_keys() keys = list_private_keys()
if keys: if keys:
info(f"Найдено SSH-ключей: {len(keys)}") info(f"Найдено SSH-ключей на этом сервере: {len(keys)}")
for k in keys: for k in keys:
label = os.path.basename(k["private"]) pub_ok = "✓ .pub" if k["public"] else "✗ нет .pub"
status = "(есть .pub)" if k["has_pub"] else "(❗ нет .pub!)" print(f" - {os.path.basename(k['private'])} ({pub_ok})")
print(f" - {label} {status}")
# 2. Проверить ключи БЕЗ passphrase # Проверяем каждый ключ
working_key = test_keys_against_target(host, user, port, keys) usable = None
if working_key: encrypted = 0
return working_key["private"] denied = 0
# 3. Проверить ключи — если зашифрован и не подходит, просто сообщаем и идём дальше
encrypted_count = 0
for k in keys: for k in keys:
label = os.path.basename(k["private"]) label = os.path.basename(k["private"])
ok, _, err = check_ssh_connectivity(host, user, port, key_path=k["private"]) ok, why = check_ssh_keyless(host, user, port, k["private"])
if not ok and ("passphrase" in err.lower() or "password" in err.lower()): if ok:
warn(f" ! Ключ '{label}' зашифрован (passphrase). Пропускаем.") success(f" Ключ '{label}' подходит для {host}")
encrypted_count += 1 usable = k["private"]
if encrypted_count and not working_key: break
info(f"Все найденные ключи зашифрованы или не подходят для {host}.") if why == "PASSPHRASE":
warn(f" ! Ключ '{label}' зашифрован passphrase (разблокируйте через ssh-add, если нужен)")
encrypted += 1
elif why == "DENIED":
info(f" ✗ Ключ '{label}' не принят target (Permission denied)")
denied += 1
else:
info(f" ✗ Ключ '{label}': {why[:60]}")
# 4. Ничего не работает — выбор метода if usable:
return usable
# Выводим диагностику
print() print()
info("Выберите способ подключения к target:") if encrypted and not denied:
print(" 1 Пароль (sshpass) — ввести пароль от target прямо сейчас") info(f"Все ключи зашифрованы passphrase. Если хотите использовать их — разблокируйте:")
print(" 2 Сгенерировать временный ed25519 ключ — скрипт сам сделает ssh-copy-id") info(f" eval $(ssh-agent) && ssh-add")
print(" 0 Отмена — передать архив вручную") info(f" Затем перезапустите: docker-migrate --resume")
elif denied:
info(f"Ключи есть, но target их не принимает (не добавлены в ~/.ssh/authorized_keys).")
# Меню выбора
info("")
info("Как подключиться к target-серверу?")
print(" 1) Пароль — ввести пароль от target (sshpass)")
print(" 2) Новый ключ — сгенерировать ed25519 + ssh-copy-id на target")
print(" 0) Отмена")
choice = prompt("Ваш выбор", default="1").strip() choice = prompt("Ваш выбор", default="1").strip()
if choice == "2": if choice == "2":
# Генерация временного ключа temp = generate_temp_keypair(host.replace(".", "_"))
temp_key = generate_temp_keypair(host.replace(".", "_")) if not temp:
if not temp_key: raise RuntimeError("Не удалось сгенерировать ключ")
cerror("Не удалось сгенерировать ключ.") if ssh_copy_id(host, user, port, temp["public"]):
return None ok, _ = check_ssh_keyless(host, user, port, temp["private"])
if ssh_copy_id(host, user, port, temp_key["public"]):
ok, _, _ = check_ssh_connectivity(host, user, port, key_path=temp_key["private"])
if ok: if ok:
success("Временный ключ настроен и работает!") success("Временный ключ работает!")
return temp_key["private"] return temp["private"]
# ssh-copy-id не сработал — показываем инструкции manual_copy_instructions(host, user, port, temp["public"])
manual_copy_instructions(host, user, port, temp_key["public"]) raise RuntimeError("ssh-copy-id не удался. Настройте вручную и запустите --resume")
raise RuntimeError("ssh-copy-id не удался. Настройте ключ вручную и запустите: docker-migrate --resume")
elif choice == "1": if choice == "1":
# Пароль — caller запросит его позже return None # caller запросит пароль
return None
else: raise RuntimeError("Перенос отменён пользователем.")
raise RuntimeError("Перенос отменён пользователем.")