You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用Python检测远程Unix服务器的特定命令提示符

用Python检测远程Unix服务器的特定命令提示符

实现思路

借助paramiko库建立SSH交互式连接,登录后读取远程服务器返回的输出内容,检查是否包含目标提示符INPUT NAME:,以此判定服务器状态。

步骤与代码实现

  1. 先安装依赖库:
pip install paramiko
  1. 编写检测脚本:
import paramiko

def check_server_prompt(server, username, password, target_prompt="INPUT NAME:", timeout=10):
    status = {
        "server": server,
        "status": "good",
        "message": ""
    }
    ssh_client = paramiko.SSHClient()
    # 自动添加未知主机密钥(生产环境建议手动管理密钥,避免安全风险)
    ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        # 建立SSH连接
        ssh_client.connect(
            hostname=server,
            username=username,
            password=password,
            timeout=timeout,
            banner_timeout=timeout
        )
        # 创建交互式shell会话(适配专有shell的输出场景)
        channel = ssh_client.invoke_shell()
        channel.settimeout(timeout)
        
        # 循环读取输出,直到检测到目标提示符或超时
        output = b""
        while True:
            if channel.recv_ready():
                output += channel.recv(1024)
                if target_prompt.encode() in output:
                    break
            else:
                status["status"] = "bad"
                status["message"] = "未检测到目标提示符"
                break
    except paramiko.AuthenticationException:
        status["status"] = "bad"
        status["message"] = "认证失败"
    except paramiko.SSHException as e:
        status["status"] = "bad"
        status["message"] = f"SSH连接错误: {str(e)}"
    except TimeoutError:
        status["status"] = "bad"
        status["message"] = "连接超时"
    except Exception as e:
        status["status"] = "bad"
        status["message"] = f"未知错误: {str(e)}"
    finally:
        # 确保连接关闭
        if ssh_client.get_transport() and ssh_client.get_transport().is_active():
            ssh_client.close()
    return status

# 服务器列表示例
server_list = [
    "server1.example.com",
    "server2.example.com",
    "server3.example.com"
]
test_username = "你的测试账号"
test_password = "你的测试密码"

# 批量检测服务器
for server in server_list:
    result = check_server_prompt(server, test_username, test_password)
    print(f"服务器 {result['server']}: {result['status']} - {result['message']}")

关键说明

  • invoke_shell():创建交互式会话,适配专有shell登录后直接输出提示符的场景,而非执行单条命令的模式。
  • 缓冲区读取:循环读取通道输出,避免因输出分段导致遗漏目标提示符。
  • 异常覆盖:处理了认证失败、连接超时、SSH协议错误等常见问题,保证每个服务器的检测流程完整。
  • 生产优化:不要使用AutoAddPolicy(),建议手动维护已知主机密钥文件;优先采用密钥认证替代密码认证,提升安全性。

内容的提问来源于stack exchange,提问作者Chris Boone

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.26 12:01:10