SSH Remote Login

通用 SSH 远程登录工具，支持连接任意服务器并执行命令。

> 本 Skill 遵循 TRACE 评测体系设计：Trust（可信）→ Reliability（可靠）→ Adaptability（适用）→ Convention（规范）→ Effectiveness（有效）。

T — Trust（可信任度）

安全承诺

• 不存储任何凭据：所有连接参数由用户实时提供，密码和私钥绝不写入任何本地文件、日志或输出
• 凭据零记录：密码和私钥仅在内存中用于构造连接，使用完毕后立即释放
• 已知基线检查：依赖 paramiko 标准库，不引入第三方闭源二进制，不调用外网 API，不发送任何数据至第三方
• 无恶意行为：禁用了 SSH Agent（allow_agent=False）和本地密钥文件扫描（look_for_keys=False），避免意外使用用户的其他凭据
• 主机密钥校验：启用 RejectPolicy 校验服务器主机密钥，防止中间人攻击（MITM）；仅对首次连接的新主机自动写入 known_hosts，主机密钥变更时拒绝连接并提示用户清缓存
• 显式确认：所有敏感操作（rm -rf、reboot、systemctl stop 等）必须获得用户明确确认才能执行（依赖调用方遵守，非代码硬拦截）

不做的事

R — Reliability（可靠性）

核心连接与执行函数

import paramiko
import socket
import os
import threading
import time

def ssh_exec(host, username, password=None, key_path=None, port=22,
             command='echo connected', timeout=30, connect_timeout=15):
    """通过 SSH 执行远程命令，返回 (stdout, stderr)。

    参数：
      host            - 服务器 IP 或域名（必填，非空）
      username        - 登录用户名
      password        - 登录密码（密码认证时使用，与 key_path 二选一）
      key_path        - 私钥文件路径（密钥认证时使用，与 password 二选一）
      port            - SSH 端口，默认 22
      command         - 要执行的命令，默认只测试连接（必填，非空）
      timeout         - 命令执行超时（秒），默认 30
      connect_timeout - 连接建立超时（秒），默认 15

    返回：
      (stdout, stderr) — 两条字符串。出错时可能抛出异常。
    """
    # 边界输入校验：给出清晰反馈，而非含糊错误（R — Reliability）
    if not host or not str(host).strip():
        raise ValueError("host 不能为空，请提供服务器 IP 或域名")
    if not command or not str(command).strip():
        raise ValueError("command 不能为空，请提供要执行的命令")

    ssh = paramiko.SSHClient()
    # 安全性：预载本地 known_hosts 并启用主机密钥校验，防止中间人攻击（T — Trust）
    kh = os.path.expanduser('~/.ssh/known_hosts')
    if os.path.exists(kh):
        try:
            ssh.load_host_keys(kh)
        except Exception:
            pass

    # 预检 known_hosts 是否已包含该主机（不依赖异常消息字符串，兼容各 paramiko 版本）
    host_keys = ssh.get_host_keys()
    known = (host_keys.lookup(host) is not None
             or host_keys.lookup(f'[{host}]:{port}') is not None)
    if known:
        # 已知主机：严格校验，密钥变更时拒绝（防 MITM）
        ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
    else:
        # 新主机首次连接：自动接受并写入 known_hosts（仅此一次）
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    connect_kwargs = {
        'hostname': host,
        'port': port,
        'username': username,
        'timeout': connect_timeout,
        'allow_agent': False,
        'look_for_keys': False,
    }

    if key_path:
        # paramiko 不自动展开 ~，需手动处理（否则 FileNotFoundError）
        connect_kwargs['key_filename'] = os.path.expanduser(key_path)
    elif password:
        connect_kwargs['password'] = password
    else:
        raise ValueError("必须提供 password 或 key_path 其中之一")

    try:
        ssh.connect(**connect_kwargs)
        # 新主机的密钥需持久化到磁盘（AutoAddPolicy 只写入内存）
        if not known:
            try:
                ssh.save_host_keys(kh)
            except Exception:
                pass  # 持久化失败不影响本次连接
        # 用独立线程同时消费 stdout/stderr，避免大输出时双流死锁（paramiko 经典陷阱）
        # get_pty=True：超时关闭 channel 时，pty 机制会向远程进程组发 SIGHUP，避免孤儿进程
        # 副作用：stderr 会合并到 stdout 流（pty 只有一个数据通道），stderr 流通常为空
        stdin, stdout, stderr = ssh.exec_command(command, get_pty=True)
        # 立即关闭 stdin：本 Skill 不交互输入，关闭后远程命令收到 EOF，避免等待 stdin 的命令挂起
        try:
            stdin.close()
        except Exception:
            pass
        out_buf, err_buf = [], []
        def _drain(stream, sink):
            while True:
                chunk = stream.read(4096)
                if not chunk:
                    break
                sink.append(chunk)
        t_out = threading.Thread(target=_drain, args=(stdout, out_buf), daemon=True)
        t_err = threading.Thread(target=_drain, args=(stderr, err_buf), daemon=True)
        t_out.start(); t_err.start()
        # 总体超时：两个读线程共享同一个 deadline，避免顺序 join 造成实际等待达 2×timeout
        deadline = time.time() + timeout
        t_out.join(max(0.0, deadline - time.time()))
        t_err.join(max(0.0, deadline - time.time()))
        if t_out.is_alive() or t_err.is_alive():
            # 超时后必须显式关闭 channel，否则远程命令会继续执行成为孤儿进程（资源泄漏）
            # pty 关闭会向远程进程组发 SIGHUP 中止命令，同时让两个读线程收到 EOF 自行退出
            try:
                stdout.channel.close()
            except Exception:
                pass
            # 给线程极短窗口收尾（远程进程被 SIGHUP 中止后会到达 EOF）
            t_out.join(1.0); t_err.join(1.0)
            raise socket.timeout(f"命令执行超过 {timeout} 秒超时")
        out = b''.join(out_buf).decode('utf-8', errors='replace').strip()
        err = b''.join(err_buf).decode('utf-8', errors='replace').strip()
        return out, err
    finally:
        ssh.close()

多命令批量执行

def ssh_exec_multi(host, username, password=None, key_path=None, port=22,
                   commands=None, timeout=30):
    """依次执行多条命令，每条命令独立连接（避免上下文残留）。
    单条命令失败不会影响后续命令，错误信息记录在 stderr 中。"""
    if not commands:
        return []
    results = []
    for cmd in commands:
        try:
            out, err = ssh_exec(host, username, password, key_path, port,
                                command=cmd, timeout=timeout)
        except Exception as e:
            out, err = '', f'[ERROR] {e}'
        results.append((cmd, out, err))
    return results

错误分级与处理策略

认证失败自动恢复（Windows 凭据缓存清理）

问题背景：在 Windows 上，OpenSSH 客户端会把登录凭据缓存在「Windows 凭据管理器」中（凭据名称形如 ssh:<host>）。当服务器更改密码后，客户端会反复使用旧的缓存凭据尝试登录，导致持续认证失败。

import subprocess, re, os, platform

def clear_local_ssh_cache(host, port=22, clear_known_hosts=True):
    """清除本地 SSH 登录缓存，解决服务器改密后反复登录失败的问题。

    仅处理：
    1. Windows 凭据管理器中以 ssh: 开头的 OpenSSH 缓存凭据（精确匹配 ssh:host / ssh:host:port）
    2. ~/.ssh/known_hosts 中该主机的条目
    不触碰任何其它凭据。
    """
    logs = []
    h = host.lower()

    # 1. Windows 凭据管理器：仅删除 ssh: 开头的 OpenSSH 缓存凭据
    if platform.system() == 'Windows':
        try:
            out = subprocess.run(['cmdkey', '/list'],
                                 capture_output=True, text=True, timeout=15).stdout
            # 兼容英文 "Target:" 与中文 "目标:"（半角/全角冒号均可）两种 cmdkey 输出
            targets = re.findall(r'(?:Target|目标)[:：]\s*(\S+)', out, re.IGNORECASE)
            ssh_hits = []
            for t in targets:
                tl = t.lower()
                # 精确匹配 OpenSSH 凭据：ssh:host / ssh:host:port，避免误伤其它凭据
                if (tl == f'ssh:{h}' or tl == f'ssh:{h}:{port}'
                        or tl.startswith(f'ssh:{h}:')):
                    ssh_hits.append(t)
            for t in ssh_hits:
                r = subprocess.run(['cmdkey', f'/delete:{t}'],
                                   capture_output=True, text=True, timeout=15)
                logs.append(f"删除凭据 [{t}]: {(r.stdout or r.stderr).strip()}")
            if not ssh_hits:
                logs.append(f"未找到与 {host} 相关的 SSH 缓存凭据")
        except Exception as e:
            logs.append(f"读取/删除凭据管理器失败: {e}")

    # 2. known_hosts 条目（精确匹配该主机，跨平台）
    if clear_known_hosts:
        try:
            kh = os.path.expanduser('~/.ssh/known_hosts')
            if os.path.exists(kh):
                subprocess.run(['ssh-keygen', '-R', f'[{host}]:{port}'],
                               capture_output=True, text=True, timeout=15)
                subprocess.run(['ssh-keygen', '-R', host],
                               capture_output=True, text=True, timeout=15)
                logs.append(f"已清理 known_hosts 中 {host} 的条目")
            else:
                logs.append("未发现 known_hosts 文件，跳过")
        except Exception as e:
            logs.append(f"清理 known_hosts 失败: {e}")

    for line in logs:
        print(line)
    return logs

认证失败时调用：

# 清理本地缓存
clear_local_ssh_cache('192.168.1.100')
# 重新向用户索取新密码
# ssh_exec('192.168.1.100', 'admin', password='新密码', command='...')

依赖检查

使用前 AI 应检查 Python 环境是否安装了 paramiko：

try:
    import paramiko
except ImportError:
    # 真正终止流程：缺少依赖时主动抛错，避免后续连接代码因 NameError 崩溃
    raise RuntimeError("[ERROR] 缺少 paramiko 库，请先执行: pip install paramiko")

若未安装，提示用户运行 pip install paramiko 后再使用本 Skill。

A — Adaptability（适用性）

✅ 适用场景（触发时使用）

❌ 不适用场景（此处不应触发）

能力边界说明

• 单服务器：一次连接针对一台主机，不支持跳板机/堡垒机自动跳转
• 命令必须单次可完成：不支持交互式命令（需要标准输入持续输入时需用户提供完整输入文本）
• 超时硬限制：命令执行默认 30 秒超时，长耗时任务（如 yum update）可能被截断；超时后远程命令会被强制中止（通过 pty 机制发送 SIGHUP）
• PTY 模式：为支持超时强制中止远程进程，命令执行采用 PTY 模式；stderr 会合并到 stdout 流（pty 单通道特性），stderr 返回值通常为空，所有错误信息均在 stdout 中
• UTF-8 输出：仅处理 UTF-8 编码的输出，其他编码的服务器输出可能出现乱码

C — Convention（规范性）

版本信息

连接参数规范

用户每次使用前必须提供以下参数。AI 必须主动向用户询问缺失的参数。

执行流程规范

步骤 1 ── 获取连接参数 ──→ 必须包含 host / username / (password 或 key_path)
                        ↓
步骤 2 ── 确认操作意图 ──→ 明确要执行的命令，危险命令需再次确认
                        ↓
步骤 3 ── 建立 SSH 连接 ──→ 先尝试连接；若认证失败且疑似改密，先清缓存再重试
                        ↓
步骤 4 ── 执行命令 ──────→ 返回 stdout / stderr
                        ↓
步骤 5 ── 展示结果 ──────→ 向用户展示完整输出；stderr 内容也需要展示
                        ↓
步骤 6 ── 关闭连接 ──────→ 每个连接执行完毕即关闭，不保持长链接

命名与代码规范

• 函数名全小写 + 下划线，遵循 PEP 8
• 参数以蛇形命名，同系统 SSH 命令参数名对齐（host、port、username、key_path）
• 返回值统一为字符串元组 (stdout, stderr)，保持可预测性
• 敏感参数（password）默认为 None；调用时须提供 password 或 key_path 之一，否则抛出 ValueError

E — Effectiveness（有效性）

命令速查（快速获取用户所需的常见结果）

# 系统概览 — 一次获取主机名、内核、运行时长
hostname && uname -a && uptime

# 磁盘 — 各分区使用情况
df -h

# 内存 — 总量、已用、可用
free -h

# CPU/进程 — Top 15 活跃进程
top -bn1 | head -15

# 网络 — IP 与监听端口
ip addr show && ss -tlnp

# 服务状态
systemctl status <service> --no-pager -n 20

# 日志尾行
tail -n 50 /var/log/syslog
journalctl -u <service> --no-pager -n 50

# 容器（Docker）
docker ps -a
docker stats --no-stream

典型用例

用例 1：查看远程服务器磁盘空间

用户："帮我登录 192.168.1.100，用户名 admin，看看磁盘空间"

import getpass
# 密码由用户实时交互输入，绝不硬编码或回显（参见 T — Trust 安全承诺）
password = getpass.getpass("请输入 192.168.1.100 的 SSH 密码: ")
out, err = ssh_exec('192.168.1.100', 'admin', password=password, command='df -h')
print(out)   # → 各分区使用情况

输出展示要求：展示 df -h 完整输出，并用一句话总结磁盘健康状态（如"根分区 / 使用率 67%，还有 30G 可用，健康"）。

用例 2：密钥认证重启 Nginx

用户："用密钥 ~/.ssh/id_rsa 连接 myserver.com 的 root，重启 nginx"

out, err = ssh_exec('myserver.com', 'root', key_path='~/.ssh/id_rsa',
                    command='systemctl restart nginx')

输出展示要求：如果 stderr 为空，展示"nginx 已重启，当前状态"并执行 systemctl status nginx --no-pager -n 10 验证。

用例 3：改密后认证失败自动恢复

用户："连不上了，前几天改了密码，帮我看下"

处理流程：
1. 首次尝试 ssh_exec(host, username, password=old_password, ...) → 抛出 AuthenticationException（old_password 代表用户实时输入，非明文）
2. 自动调用 clear_local_ssh_cache(host) 清除本地缓存
3. 提示用户："服务器密码已变更，请提供新密码"
4. 用户提供新密码后，使用新密码重试

用例 4：批量执行多条命令

用户："检查服务器整体健康状态"

results = ssh_exec_multi(host, username, password, commands=[
    'hostname',
    'uptime',
    'df -h /',
    'free -h',
    'cat /etc/os-release | head -3',
])
for cmd, out, err in results:
    print(f"=== {cmd} ===\n{out or err}\n")

输出展示要求：按模块分组输出，并在结尾给出健康评级（"CPU 负载正常 / 内存充足 / 磁盘空间不足，需扩容"等）。

输出规范

展示结果时 AI 必须遵循：

1. 展示原始输出：命令的 stdout 和 stderr 原文必须完整展示，不加删改
2. 附加解读：紧接输出之后，用自然语言总结要点（磁盘是否告警、服务是否正常运行等）
3. stderr 不忽略：即使命令返回 0，stderr 内容也应展示给用户。本 Skill 采用 PTY 模式（见能力边界），stderr 已合并到 stdout 流中，stderr 返回值通常为空，AI 应从 stdout 中识别错误信息（如 Permission denied、not found 等）一并向用户展示
4. 禁止暴露凭据：任何输出中不得出现真实 password=...、私钥内容或路径。上文示例中的 password/old_password 等均为占位变量，代表用户实时输入，绝不可写为明文字面量
5. 结果可立即使用：用户拿到输出后应能直接做决策（如"磁盘空间充足，无需处理"或"内存使用率达 92%，建议扩容"）