我正在编写一个脚本,该脚本使用 Tailscale API 获取设备列表,然后使用 Paramiko 通过 SSH 连接并查询设备上的一些数据。昨天一切正常,但今天所有连接都失败了。我意识到,在幕后,SSH 失败了,因为它由 Tailscale 管理,需要重新身份验证:
ssh root@MyDevice
# Tailscale SSH requires an additional check.
# To authenticate, visit: https://login.tailscale.com/a/XXXXXX
但是我如何使用 Paramiko 检测并解决此问题?我当前的代码非常基本,但我能够添加日志记录:
import requests
from datetime import datetime, timedelta
import paramiko
import logging
url = "https://api.tailscale.com/api/v2/tailnet/-/devices"
# https://tailscale.com/kb/1085/auth-keys#generate-an-auth-key
headers = {"Authorization": "Bearer GENERATED AUTH KEY"}
response = requests.get(url, headers=headers)
# print(response.json())
data = response.json()
devices = data['devices']
current = datetime.utcnow() - timedelta(minutes = 1)
class OnlyTailscaleFilter(logging.Filter):
def filter(self, record):
return not record.getMessage().contains('Tailscale')
logging.basicConfig()
logger = logging.getLogger("paramiko")
logger.setLevel(logging.INFO)
logger.addFilter(OnlyTailscaleFilter())
for device in devices:
if 'lastSeen' in device:
last_seen = datetime.strptime(device['lastSeen'], '%Y-%m-%dT%H:%M:%SZ')
if last_seen > current:
hostname = ''
uptime_seconds = ''
if 'hostname' in device:
hostname = device['hostname']
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(device['addresses'][0], username='root')
except paramiko.AuthenticationException as ex:
print('{}, An authentication exception occurred: {}'.format(hostname, ex))
exit(1)
else:
ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('cat /proc/uptime')
if ssh_stdout.channel.recv_exit_status() == 0:
temp = ssh_stdout.readlines()
if temp:
temp = temp[0].strip('\n').strip('\"')
uptime_seconds = temp.split()[0]
print(hostname, ',', uptime_seconds)
ssh.close()
当我运行它时,我可以在异常之前的日志中看到这一点:
INFO:paramiko.transport:Connected (version 2.0, client Tailscale)
INFO:paramiko.transport:Auth banner: b'# Tailscale SSH requires an additional check.\n# To authenticate, visit: https://login.tailscale.com/a/XXXXX\n'
当然可以使用 Paramiko 检测并处理此 Tailscale 身份验证挑战。诀窍在于识别 Tailscale 的身份验证提示并使用 Tailscale 提供的 URL 进行身份验证。由于此身份验证流在浏览器中发生,因此无法在 Paramiko 中直接处理它。
以下是调整代码以解决此问题的步骤:
- 检测 Tailscale 身份验证提示: 修改的代码以识别 Paramiko 捕获的 SSH 横幅中的 Tailscale 特定消息。
- 提取身份验证 URL: 从 SSH 横幅中解析 Tailscale 提供的 URL。
- 提示用户进行身份验证: 指示用户访问提取的 URL 以完成身份验证过程。
- 重试 SSH 连接: 用户完成身份验证后,重试 SSH 连接。
以下是使用这些修改更新后的代码:
import requests
from datetime import datetime, timedelta
import paramiko
import logging
import re
# ...(现有的代码)...
for device in devices:
if 'lastSeen' in device:
# ...(现有的代码)...
try:
ssh.connect(device['addresses'][0], username='root')
except paramiko.AuthenticationException as ex:
# 检查 Tailscale 身份验证提示
if "Tailscale SSH requires an additional check" in str(ex):
# 从异常中提取身份验证 URL
match = re.search(r"(https://login\.tailscale\.com/a/\w+)", str(ex))
if match:
auth_url = match.group(1)
print(f"需要 Tailscale 身份验证。请访问:{auth_url}")
input("按 Enter 键继续...") # 在用户完成身份验证后等待
try:
ssh.connect(device['addresses'][0], username='root')
except paramiko.AuthenticationException as ex:
print(f"{hostname}, 身份验证失败:{ex}")
exit(1)
else:
print(f"{hostname}, 无法提取 Tailscale 身份验证 URL")
exit(1)
else:
print(f"{hostname}, 发生身份验证异常:{ex}")
exit(1)
else:
# ...(现有的代码)...
此代码执行以下操作:
-
它会在
paramiko.AuthenticationException
中捕获 Tailscale 特定的消息。 - 它使用正则表达式从异常消息中提取身份验证 URL。
- 它提示用户访问提取的 URL 以完成身份验证。
- 在用户按下 Enter 键后,它会尝试重新建立 SSH 连接。
此方法提供了一个交互式流程,供在使用 Paramiko 和 Tailscale 时处理必要的身份验证步骤。请记住,此方法依赖用户交互来完成身份验证。
标签:python,logging,paramiko,tailscale From: 78813596