小开

最佳答案

我会把你介绍给帕米科

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)

小开

或者你可以使用 Getstatusoutput:

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

我广泛使用它，效果很好。

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

小开

我已经使用了帕米科一串(不错)和嘘(也不错)。我推荐任何一种。它们的工作原理略有不同，但在使用上有相对较大的重叠。

小开

我发现 parmiko 有点太低级了，Fabric 不是特别适合作为库使用，所以我把我自己的名为马刺的库放在一起，它使用 parmiko 实现了一个稍微好一点的接口:

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"])

小开

所有这些都已经说明(推荐)使用 帕米科，我只是共享一个 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:])

小开

看一下 spurplus，这是我们围绕 spur开发的一个包装器，它提供了类型注释和一些小技巧(重新连接 SFTP，md5 等等)校对: https://pypi.org/project/spurplus/

小开

在添加了额外的一行之后，parmiko 终于为我工作了，这是非常重要的一行(第3行) :

import paramiko


p = paramiko.SSHClient()
p.set_missing_host_key_policy(paramiko.AutoAddPolicy())   # This script doesn't work for me unless this line is added!
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)

确保安装了 parmiko 包。解决方案的原始来源: http://www.youtube.com/watch

小开

效果很好。

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\n')
print('success')


execute()

小开

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

  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\n')


print('success')


details()

小开

#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\n')
print('success')


details(Host,Port,User,Pass)

小开

要求用户根据他们正在登录的设备输入命令。
下面的代码由 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('\n')
print('success')


details(Host, Port, User, Pass)

小开

保持简单，不需要库。

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()

小开

这个被接受的答案对我不起作用，下面是我用的方法:

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
...

或者，你可以使用小意思:

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

参考文献:

备注:

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

小开

您可以使用这些命令中的任何一个，这也将帮助您输入密码。

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)

小开

第一: 我很惊讶没有人提到 fabric。

第二: 对于您描述的那些需求，我已经实现了一个名为 jk_simpleexec的 Python 模块。它的目的是: 使运行命令变得容易。

让我给你解释一下。

“在本地执行命令”的问题

我的 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-条款，我自己的模块是在 阿帕奇2号下提供的。

使用的模块:

面料 http://www.fabfile.org/

Jk _ pwdinput : < a href = “ https://github.com/jkpubsrc/python-module-jk-pwdinput”rel = “ nofollow norefrer”> https://github.com/jkpubsrc/python-module-jk-pwdinput

Jk _ simplexec : < a href = “ https://github.com/jkpubsrc/python-module-jk-simpleexec”rel = “ nofollow norefrer”> https://github.com/jkpubsrc/python-module-jk-simpleexec

编码愉快! ; -)

小开

最现代的方法可能是使用布料。这个模块允许您设置 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:\n{0.stdout}" print(msg.format(result))

小开

我编写了一个简单的类，使用 subprocess模块在本机 ssh 上远程运行命令:

用法

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 \ 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)}', ] )

使用 Python 在 ssh 上执行命令