我会把你介绍给 paramiko

this question

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)

commands.getstatusoutput :

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

我用得很广泛,效果很好。

subprocess.check_output .

保持简单。不需要库。

import subprocess

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

我发现paramiko有点太低级了,织物不太适合用作图书馆,所以我把自己的图书馆叫做 spur

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)

#!/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:])

谢谢您!

帕拉米科

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)

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

我用过 paramiko 一堆(好的)和 pxssh

公认的答案对我不适用,以下是我所用的:

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 22222 'my command; exit' """
print( subprocess.getoutput(cmd) )

1. https://github.com/onyxfish/relay/issues/11
2. https://stackoverflow.com/a/61016663/797495

笔记:

1. 只需确保通过ssh手动连接到远程系统至少一次( ssh root@ip )以及 接受公钥 ,这是很多次无法使用连接的原因 paramiko ssh 脚本。

完美地工作。。。

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

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

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)

看一看 spurplus ,我们开发的包装器 spur 它提供了类型注释和一些小技巧(重新连接SFTP、md5) 等 .): https://pypi.org/project/spurplus/

要求用户根据登录的设备输入命令。
以下代码已通过PEP8验证在线.

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)

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

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

第二:对于您描述的那些需求,我已经实现了一个自己的python模块,名为 jk_simpleexec . 它的目的是:使运行命令变得简单。

让我给你解释一下。

我的python模块 jk\ U simpleexec公司 提供一个名为 runCmd(..) 可以执行shell(!)本地或远程命令。这很简单。以下是本地执行命令的示例:

import jk_simpleexec

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

注意:请注意,默认情况下,返回的数据会自动修剪,以从STDOUT和STDERR中删除过多的空行。(当然,可以停用此行为,但出于特定目的,您必须清楚地知道,此行为正是您所需要的。)

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

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

出于调试目的,您希望在某个时候将结果输出到STDOUT,因此可以执行以下操作:

cmdResult.dump()

如果您想处理STDOUT,它也很简单。例子:

for line in cmdResult.stdOutLines:
    print(line)

“远程执行命令”问题

当然,我们可能希望在另一个系统上远程执行这个命令。为此,我们可以使用相同的函数 以完全相同的方式,但我们需要指定一个

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,如果您在某个时间决定从本地主机移动到另一个主机,则不需要对软件进行任何修改。

密码输入问题

jk_pwdinput . 与常规密码输入不同的是 jk\U输入 将输出一些星星而不是什么都不打印。因此,对于您键入的每个密码字符,您将看到一个星。这样你就更容易输入密码了。

代码如下:

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远程执行命令,
• 使用密码输入以简单安全的方式创建连接。

上面的代码很好地解决了这个问题(希望你也一样)。一切都是开源的:结构是 BSD-2-条款 ,下面提供了我自己的模块 阿帕奇-2

使用的模块:

• : http://www.fabfile.org/
• : https://github.com/jkpubsrc/python-module-jk-pwdinput
• jk\ U simplexec公司 https://github.com/jkpubsrc/python-module-jk-simpleexec

快乐的编码!;-)

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