如何使用Paramiko向Linux运行中shell命令输入密码等字符串

原有代码问题 #

• 仅发送命令字符串，未读取shell返回内容，无法识别密码输入类交互提示
• 命令内MONIKERNAME、WALLET_NAME为硬编码占位符，未替换为传入的业务参数
• 无命令执行完成判断逻辑，命令顺序执行时容易出现输入串扰、执行截断问题

实现思路 #

• 基于invoke_shell创建的交互式会话，发送命令后循环读取缓冲区返回内容
• 对读取到的内容做关键字匹配，识别到对应交互提示时自动发送预设参数
  • 匹配Enter keyring passphrase：发送钱包密码nodepass
  • 匹配Re-enter keyring passphrase：二次发送钱包密码完成确认
  • 匹配sudo密码提示：发送服务器登录密码
• 识别到shell命令提示符（$ /# ）时判定当前命令执行完成，再发送下一条命令
• 增加超时保护，避免异常场景下程序死循环

可直接运行的修正代码 #

import paramiko
import time

def send_command(
        ip,
        username,
        password,
        monikername,
        walletname,
        nodepass,
        commands,
        short_pause = 1,
        long_pause = 5,
        cmd_timeout = 300
):
    cl = paramiko.SSHClient()
    cl.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    cl.connect(
        hostname=ip,
        username=username,
        password=password,
        look_for_keys=False,
        allow_agent=False,
        timeout=10
    )
    full_output = ""
    with cl.invoke_shell(width=200, height=1000) as ssh:
        # 清空登录初始欢迎内容
        time.sleep(short_pause)
        ssh.recv(65535).decode("utf-8", errors="ignore")
        
        for raw_cmd in commands:
            start_time = time.time()
            # 替换命令中的占位参数
            cmd = raw_cmd.replace("MONIKERNAME", monikername).replace("WALLET_NAME", walletname)
            ssh.send(f"{cmd}\n")
            time.sleep(short_pause)
            
            cmd_output = ""
            while True:
                time.sleep(0.5)
                # 读取当前缓冲区内容
                try:
                    chunk = ssh.recv(65535).decode("utf-8", errors="ignore")
                except Exception:
                    chunk = ""
                cmd_output += chunk
                full_output += chunk

                # 处理首次输入钱包密码提示
                if "Enter keyring passphrase" in chunk:
                    ssh.send(f"{nodepass}\n")
                    time.sleep(short_pause)
                    continue
                # 处理二次确认钱包密码提示
                if "Re-enter keyring passphrase" in chunk:
                    ssh.send(f"{nodepass}\n")
                    time.sleep(short_pause)
                    continue
                # 处理sudo密码提示
                if f"[sudo] password for {username}" in chunk:
                    ssh.send(f"{password}\n")
                    time.sleep(short_pause)
                    continue
                # 匹配shell提示符，判定当前命令执行结束
                if chunk.endswith("$ ") or chunk.endswith("# "):
                    break
                # 单命令超时退出
                if time.time() - start_time > cmd_timeout:
                    break
            time.sleep(long_pause)
    cl.close()
    return full_output

if __name__ == "__main__":
    # 替换为实际的服务器配置列表
    data = []
    commands = [
        "sudo apt update && sudo apt upgrade -y",
        'ver="1.18.1" && \
        wget "https://golang.org/dl/go$ver.linux-amd64.tar.gz" && \
        sudo rm -rf /usr/local/go && \
        sudo tar -C /usr/local -xzf "go$ver.linux-amd64.tar.gz" && \
        rm "go$ver.linux-amd64.tar.gz" && \
        echo "export PATH=$PATH:/usr/local/go/bin:$HOME/go/bin" >> $HOME/.bash_profile && \
        source $HOME/.bash_profile && \
        go version',
        "git clone https://github.com/sei-protocol/sei-chain.git && cd sei-chain",
        "git checkout 1.0.6beta",
        "make install",
        "seid version --long | head",
        "seid init MONIKERNAME --chain-id atlantic-1",
        "seid keys add WALLET_NAME",
        "seid add-genesis-account WALLET_NAME 100000000usei",
        'seid gentx WALLET_NAME 100000000usei \
        --chain-id atlantic-1 \
        --commission-rate=0.1 \
        --commission-max-rate=0.2 \
        --commission-max-change-rate=0.1 \
        --pubkey $(seid tendermint show-validator) \
        --moniker "MONIKERNAME"',
        "cat ~/.sei/config/gentx/gentx-*"
    ]
    for key in data:
        ip = key['ip']
        username = key['username']
        password = key['pass']
        monikername = key['monikername']
        walletname = key['walletname']
        nodepass = key['nodepass']
        result = send_command(ip, username, password, monikername, walletname, nodepass, commands)
        print(f"===== 服务器{ip}执行输出 =====")
        print(result)

调优提示 #

• 如果你的服务器shell提示符做过自定义，结尾不是$ 或# ，可以自行调整提示符匹配规则，比如增加主机名、用户名特征匹配
• 网络延迟高的场景可以适当调大short_pause、long_pause参数值，避免读取输出不全
• 其他需要交互的命令场景，只要拿到交互提示的固定关键字，都可以按同样的匹配逻辑加自动输入规则

内容的提问来源于stack exchange，提问作者TokyoDead