摘要：问题描述：我正在编写一个脚本来自动执行 Python 中的一些命令行命令。目前，我正在执行如下调用：cmd = "some unix command" retcode = subprocess.call(cmd,shell=True) 但是，我需要在远程机器上运行一些命令。手动地，我将使用...

问题描述：

我正在编写一个脚本来自动执行 Python 中的一些命令行命令。目前，我正在执行如下调用：

cmd = "some unix command"
retcode = subprocess.call(cmd,shell=True)

但是，我需要在远程机器上运行一些命令。手动地，我将使用登录ssh，然后运行命令。如何在 Python 中自动执行此操作？我需要使用（已知）密码登录远程机器，因此我不能只使用cmd = ssh user@remotehost，我想知道是否有我应该使用的模块？

解决方案 1：

我会向你推荐paramiko

看到这个问题

ssh = paramiko.SSHClient()
ssh.connect(server, username=username, password=password)
ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command(cmd_to_execute)

如果您使用 ssh 密钥，请执行以下操作：

k = paramiko.RSAKey.from_private_key_file(keyfilename)
# OR k = paramiko.DSSKey.from_private_key_file(keyfilename)

ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname=host, username=user, pkey=k)

解决方案 2：

保持简单。无需任何库。

import subprocess

# Python 2
subprocess.Popen("ssh {user}@{host} {cmd}".format(user=user, host=host, cmd='ls -l'), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()

# Python 3
subprocess.Popen(f"ssh {user}@{host} {cmd}", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()

解决方案 3：

或者您也可以使用commands.getstatusoutput：

commands.getstatusoutput("ssh machine 1 'your script'")

我广泛使用了它并且效果很好。

在 Python 2.6+ 中，使用subprocess.check_output。

解决方案 4：

我发现 paramiko 有点太低级了，而 Fabric 并不特别适合用作库，因此我组建了自己的库spur，它使用 paramiko 实现了一个稍微好一点的界面：

import spur

shell = spur.SshShell(hostname="localhost", username="bob", password="password1")
result = shell.run(["echo", "-n", "hello"])
print result.output # prints hello

如果需要在 shell 中运行：

shell.run(["sh", "-c", "echo -n hello"])

解决方案 5：

接受的答案对我来说不起作用，下面是我使用的答案：

import paramiko
import os

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# ssh.load_system_host_keys()
ssh.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
ssh.connect("d.d.d.d", username="user", password="pass", port=22222)

ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command("ls -alrt")
exit_code = ssh_stdout.channel.recv_exit_status() # handles async exit error 

for line in ssh_stdout:
    print(line.strip())

total 44
-rw-r--r--.  1 root root  129 Dec 28  2013 .tcshrc
-rw-r--r--.  1 root root  100 Dec 28  2013 .cshrc
-rw-r--r--.  1 root root  176 Dec 28  2013 .bashrc
...

或者，您可以使用sshpass：

import subprocess
cmd = """ sshpass -p "myPas$" ssh user@d.d.d.d -p 12345 'my command; exit' """
print( subprocess.getoutput(cmd) )

参考：

笔记：

只需确保至少通过 ssh ( ssh root@ip) 手动连接远程系统一次并接受公钥，这很多时候是无法使用paramiko或其他自动ssh脚本连接的原因。

解决方案 6：

所有人都已经声明（推荐）使用paramiko，而我只是分享一个 python 代码（可以说是 API），它允许您一次执行多个命令。

在不同的节点上执行命令使用：Commands().run_cmd(host_ip, list_of_commands)

您将看到一个 TODO，如果任何命令执行失败，我将停止执行，但我不知道该怎么做。请分享您的知识

#!/usr/bin/python

import os
import sys
import select
import paramiko
import time


class Commands:
    def __init__(self, retry_time=0):
        self.retry_time = retry_time
        pass

    def run_cmd(self, host_ip, cmd_list):
        i = 0
        while True:
        # print("Trying to connect to %s (%i/%i)" % (self.host, i, self.retry_time))
        try:
            ssh = paramiko.SSHClient()
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh.connect(host_ip)
            break
        except paramiko.AuthenticationException:
            print("Authentication failed when connecting to %s" % host_ip)
            sys.exit(1)
        except:
            print("Could not SSH to %s, waiting for it to start" % host_ip)
            i += 1
            time.sleep(2)

        # If we could not connect within time limit
        if i >= self.retry_time:
            print("Could not connect to %s. Giving up" % host_ip)
            sys.exit(1)
        # After connection is successful
        # Send the command
        for command in cmd_list:
            # print command
            print "> " + command
            # execute commands
            stdin, stdout, stderr = ssh.exec_command(command)
            # TODO() : if an error is thrown, stop further rules and revert back changes
            # Wait for the command to terminate
            while not stdout.channel.exit_status_ready():
                # Only print data if there is data to read in the channel
                if stdout.channel.recv_ready():
                    rl, wl, xl = select.select([ stdout.channel ], [ ], [ ], 0.0)
                    if len(rl) > 0:
                        tmp = stdout.channel.recv(1024)
                        output = tmp.decode()
                        print output

        # Close SSH connection
        ssh.close()
        return

def main(args=None):
    if args is None:
        print "arguments expected"
    else:
        # args = {'<ip_address>', <list_of_commands>}
        mytest = Commands()
        mytest.run_cmd(host_ip=args[0], cmd_list=args[1])
    return


if __name__ == "__main__":
    main(sys.argv[1:])

解决方案 7：

在添加了允许缺少主机密钥策略的附加行后， paramiko终于为我工作了，这确实很重要（第 5 行）：

import paramiko

p = paramiko.SSHClient()
# This script doesn't work for me unless the following line is added!
p.set_missing_host_key_policy(paramiko.AutoAddPolicy())   
p.connect("server", port=22, username="username", password="password")
stdin, stdout, stderr = p.exec_command("your command")
opt = stdout.readlines()
opt = "".join(opt)
print(opt)

确保已安装 paramiko 包。解决方案的原始来源：来源

解决方案 8：

首先：我很惊讶还没有人提及fabric。

第二：针对您描述的这些要求，我实现了一个名为的 Python 模块jk_simpleexec。它的目的是：使运行命令变得简单。

让我为你稍微解释一下。

“本地执行命令”问题

我的 python 模块jk_simpleexec提供了一个名为的函数runCmd(..)，可以在本地或远程执行 shell (!) 命令。这非常简单。以下是本地执行命令的示例：

import jk_simpleexec

cmdResult = jk_simpleexec.runCmd(None, "cd / ; ls -la")

注意：请注意，默认情况下会自动修剪返回的数据，以从 STDOUT 和 STDERR 中删除过多的空行。（当然，可以停用此行为，但就您所考虑的目的而言，该行为正是您想要的。）

“处理结果”问题

您将收到一个包含返回代码、STDOUT 和 STDERR 的对象。因此处理结果非常容易。

这正是您想要做的，因为您执行的命令可能存在并已启动，但可能无法完成预期的操作。在最简单的情况下，如果您对 STDOUT 和 STDERR 不感兴趣，您的代码可能看起来像这样：

cmdResult.raiseExceptionOnError("Something went wrong!", bDumpStatusOnError=True)

出于调试目的，您希望在某个时间将结果输出到 STDOUT，因此您可以这样做：

cmdResult.dump()

如果你想处理 STDOUT，这也很简单。例如：

for line in cmdResult.stdOutLines:
    print(line)

“远程执行命令”问题

现在我们当然可能想在另一个系统上远程执行此命令。为此，我们可以runCmd(..)以完全相同的方式使用相同的函数，但我们需要fabric先指定一个连接对象。可以这样做：

from fabric import Connection

REMOTE_HOST = "myhost"
REMOTE_PORT = 22
REMOTE_LOGIN = "mylogin"
REMOTE_PASSWORD = "mypwd"
c = Connection(host=REMOTE_HOST, user=REMOTE_LOGIN, port=REMOTE_PORT, connect_kwargs={"password": REMOTE_PASSWORD})

cmdResult = jk_simpleexec.runCmd(c, "cd / ; ls -la")

# ... process the result stored in cmdResult ...

c.close()

一切都保持完全相同，但这次我们在另一台主机上运行此命令。这样做的目的是：我希望有一个统一的 API，如果您在某个时候决定从本地主机移动到另一个主机，则无需对软件进行任何修改。

密码输入问题

现在当然还有密码问题。一些用户在上面提到过：我们可能想让执行此 Python 代码的用户输入密码。

为了解决这个问题，我很久以前就创建了一个自己的模块。jk_pwdinput与常规密码输入的不同之处在于，它jk_pwdinput会输出一些星号，而不是什么都不打印。因此，对于您输入的每个密码字符，您都会看到一个星号。这样，您可以更轻松地输入密码。

以下是代码：

import jk_pwdinput

# ... define other 'constants' such as REMOTE_LOGIN, REMOTE_HOST ...

REMOTE_PASSWORD = jk_pwdinput.readpwd("Password for " + REMOTE_LOGIN + "@" + REMOTE_HOST + ": ")

（为了完整性：如果readpwd(..)返回，None则用户使用 Ctrl+C 取消了密码输入。在现实世界中，您可能需要对此采取适当的行动。）

完整示例

以下是完整示例：

import jk_simpleexec
import jk_pwdinput
from fabric import Connection

REMOTE_HOST = "myhost"
REMOTE_PORT = 22
REMOTE_LOGIN = "mylogin"
REMOTE_PASSWORD = jk_pwdinput.readpwd("Password for " + REMOTE_LOGIN + "@" + REMOTE_HOST + ": ")
c = Connection(host=REMOTE_HOST, user=REMOTE_LOGIN, port=REMOTE_PORT, connect_kwargs={"password": REMOTE_PASSWORD})

cmdResult = jk_simpleexec.runCmd(
    c = c,
    command = "cd / ; ls -la"
)
cmdResult.raiseExceptionOnError("Something went wrong!", bDumpStatusOnError=True)

c.close()

最后说明

因此我们有全套内容：

执行命令，
通过相同的 API 远程执行该命令，
通过输入密码以简单而安全的方式创建连接。

上面的代码对我来说很好地解决了这个问题（希望对你来说也是如此）。而且一切都是开源的：Fabric 是BSD-2-Clause，我自己的模块是在Apache-2下提供的。

使用的模块：

面料：http://www.fabfile.org/
jk_pwdinput：https ://github.com/jkpubsrc/python-module-jk-pwdinput
jk_simplexec：https://github.com/jkpubsrc/python-module-jk-simpleexec

编码愉快！;-)

解决方案 9：

您可以使用其中任何一个命令，这也将帮助您提供密码。

cmd = subprocess.run(["sshpass -p 'password' ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null root@domain.com ps | grep minicom"], shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
print(cmd.stdout)
OR
cmd = subprocess.getoutput("sshpass -p 'password' ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null root@domain.com ps | grep minicom")
print(cmd)

解决方案 10：

完美运行...

import paramiko
import time

ssh = paramiko.SSHClient()
#ssh.load_system_host_keys()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('10.106.104.24', port=22, username='admin', password='')

time.sleep(5)
print('connected')
stdin, stdout, stderr = ssh.exec_command(" ")

def execute():
       stdin.write('xcommand SystemUnit Boot Action: Restart
')
       print('success')

execute()

解决方案 11：

要求用户根据其登录的设备输入命令。

以下代码已通过 PEP8online.com 验证。

import paramiko
import xlrd
import time

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
loc = ('/Users/harshgow/Documents/PYTHON_WORK/labcred.xlsx')
wo = xlrd.open_workbook(loc)
sheet = wo.sheet_by_index(0)
Host = sheet.cell_value(0, 1)
Port = int(sheet.cell_value(3, 1))
User = sheet.cell_value(1, 1)
Pass = sheet.cell_value(2, 1)

def details(Host, Port, User, Pass):
    time.sleep(2)
    ssh.connect(Host, Port, User, Pass)
    print('connected to ip ', Host)
    stdin, stdout, stderr = ssh.exec_command("")
    x = input('Enter the command:')
    stdin.write(x)
    stdin.write('
')
    print('success')

details(Host, Port, User, Pass)

解决方案 12：

#Reading the Host,username,password,port from excel file
import paramiko 
import xlrd

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

loc = ('/Users/harshgow/Documents/PYTHON_WORK/labcred.xlsx')
wo = xlrd.open_workbook(loc)
sheet = wo.sheet_by_index(0)
Host = sheet.cell_value(0,1)
Port = int(sheet.cell_value(3,1))
User = sheet.cell_value(1,1)
Pass = sheet.cell_value(2,1)

def details(Host,Port,User,Pass):
    ssh.connect(Host, Port, User, Pass)
    print('connected to ip ',Host)
    stdin, stdout, stderr = ssh.exec_command("")
    stdin.write('xcommand SystemUnit Boot Action: Restart
')
    print('success')

details(Host,Port,User,Pass)

解决方案 13：

最现代的方法可能是使用fabric。该模块允许您设置 SSH 连接，然后运行命令并通过连接对象获取其结果。

这是一个简单的例子：

from fabric import Connection
with Connection("your_hostname") as connection:
    result = connection.run("uname -s", hide=True)
    msg = "Ran {0.command!r} on {0.connection.host}, got stdout:
{0.stdout}"
    print(msg.format(result))

解决方案 14：

我编写了一个简单的类来通过本机 ssh 在远程运行命令，使用以下subprocess模块：

用法

from ssh_utils import SshClient
client = SshClient(user='username', remote='remote_host', key='path/to/key.pem')

# run a list of commands
client.cmd(['mkdir ~/testdir', 'ls -la', 'echo done!'])

# copy files/dirs
client.scp('my_file.txt', '~/testdir')

类源代码

https://gist.github.com/mamaj/a7b378a5c969e3e32a9e4f9bceb0c5eb

import subprocess
from pathlib import Path
from typing import Union

class SshClient():
    """ Perform commands and copy files on ssh using subprocess 
        and native ssh client (OpenSSH).
    """
    
    def __init__(self,
                 user: str,
                 remote: str,
                 key_path: Union[str, Path]) -> None:
        """

        Args:
            user (str): username for the remote
            remote (str): remote host IP/DNS
            key_path (str or pathlib.Path): path to .pem file
        """
        self.user = user
        self.remote = remote
        self.key_path = str(key_path)
        
        
    def cmd(self, 
            cmds: list[str],
            strict_host_key_checking=False) -> None:
        
        """runs commands consecutively, ensuring success of each
            after calling the next command.

        Args:
            cmds (list[str]): list of commands to run.
            strict_host_key_checking (bool, optional): Defaults to True.
        """
        
        strict_host_key_checking = 'yes' if strict_host_key_checking \n                                    else 'no'
        cmd = ' && '.join(cmds)
        subprocess.run(
            [
                'ssh',
                '-i', self.key_path,
                '-o', f'StrictHostKeyChecking={strict_host_key_checking}', 
                '-o', 'UserKnownHostsFile=/dev/null',
                f'{self.user}@{self.remote}', 
                cmd
            ]
        )
        
        
    def scp(self, source: Union[str, Path], destination: Union[str, Path]):
        """Copies `srouce` file to remote `destination` using the 
            native `scp` command.
            
        Args:
            source (Union[str, Path]): Source file path.
            destination (Union[str, Path]): Destination path on remote.
        """
        subprocess.run(
            [
                'scp',
                '-i', self.key_path,
                str(source), 
                f'{self.user}@{self.remote}:{str(destination)}',
            ]
        )

解决方案 15：

sshpass 仅适用于 Linux。sshpass 在 Windows 上不存在

解决方案 16：

下面的例子，如果您希望用户输入主机名、用户名、密码和端口号。

  import paramiko

  ssh = paramiko.SSHClient()

  ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())



  def details():

  Host = input("Enter the Hostname: ")

  Port = input("Enter the Port: ")

  User = input("Enter the Username: ")

  Pass = input("Enter the Password: ")

  ssh.connect(Host, Port, User, Pass, timeout=2)

  print('connected')

  stdin, stdout, stderr = ssh.exec_command("")

  stdin.write('xcommand SystemUnit Boot Action: Restart
')

  print('success')

  details()

使用 Python 通过 ssh 执行命令