ubuntu 在远程计算机上使用Python语言执行命令

bqf10yzr  于 2022-10-27  发布在  Python
关注(0)|答案(4)|浏览(138)

我在Ubuntu上用python语言编写了一个程序,在RaspberryPI上执行命令ls -l,连接到网络。
有人能指导我如何做到这一点吗?

7vux5j2d

7vux5j2d1#

当然,有几种方法可以做到这一点!
假设您在raspberry.lan主机上有一个Raspberry PI,您的用户名是irfan

子流程

它是运行命令的默认Python库。
您可以让它运行ssh,并在远程服务器上执行您需要的任何操作。
Scrat has it covered in his answer。如果您不想使用任何第三方库,则绝对应该这样做。
您还可以使用pexpect自动输入密码/密码短语。

pariko

paramiko是一个第三方库,它添加了SSH协议支持,因此它可以像SSH客户端一样工作。
连接到服务器、执行并获取ls -l命令结果的示例代码如下所示:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('raspberry.lan', username='irfan', password='my_strong_password')

stdin, stdout, stderr = client.exec_command('ls -l')

for line in stdout:
    print line.strip('\n')

client.close()

面料

您也可以使用fabric来实现它。
Fabric是一种部署工具,可在远程服务器上执行各种命令。
它经常被用来在远程服务器上运行东西,所以你可以很容易地安装你的最新版本的Web应用程序,重新启动Web服务器等等,只需一个命令。实际上,您可以在多个服务器上运行相同的命令,这太棒了!
尽管它是作为部署和远程管理工具开发的,但您仍然可以使用它来执行基本命令。


# fabfile.py

from fabric.api import *

def list_files():
    with cd('/'):  # change the directory to '/'
        result = run('ls -l')  # run a 'ls -l' command
        # you can do something with the result here,
        # though it will still be displayed in fabric itself.

这就像在远程服务器上输入cd /ls -l,因此您将获得根文件夹中的目录列表。
然后在外壳中运行:

fab list_files

它将提示输入服务器地址:

No hosts found. Please specify (single) host string for connection: irfan@raspberry.lan

快速提示:您还可以在fab命令中分配用户名和主机:

fab list_files -U irfan -H raspberry.lan

或者,您可以将一个主机放入Fabfile中的env.hosts变量中。以下是如何做到这一点。
然后,系统将提示您输入SSH密码:

[irfan@raspberry.lan] run: ls -l
[irfan@raspberry.lan] Login password for 'irfan':

然后,该命令将成功运行。

[irfan@raspberry.lan] out: total 84
[irfan@raspberry.lan] out: drwxr-xr-x   2 root root  4096 Feb  9 05:54 bin
[irfan@raspberry.lan] out: drwxr-xr-x   3 root root  4096 Dec 19 08:19 boot
...
yyyllmsg

yyyllmsg2#

来自here的简单示例:

import subprocess
import sys

HOST="www.example.org"

# Ports are handled in ~/.ssh/config since we use OpenSSH

COMMAND="uname -a"

ssh = subprocess.Popen(["ssh", "%s" % HOST, COMMAND],
                       shell=False,
                       stdout=subprocess.PIPE,
                       stderr=subprocess.PIPE)
result = ssh.stdout.readlines()
if result == []:
    error = ssh.stderr.readlines()
    print >>sys.stderr, "ERROR: %s" % error
else:
    print result

它所做的正是您想要的:通过ssh连接,执行命令,返回输出。不需要第三方库。

olhwl3o2

olhwl3o23#

您可以将以下方法与Linux/Unix的内置ssh命令一起使用。

import os
   os.system('ssh username@ip  bash < local_script.sh >> /local/path/output.txt 2>&1')
   os.system('ssh username@ip  python < local_program.py >> /local/path/output.txt 2>&1')
hfyxw5xn

hfyxw5xn4#

通过调用外壳,可以使用Paramiko模块来运行多个命令。在这里,我创建了类来调用ssh外壳
类外壳处理程序:

def __init__(self, host, user, psw):
    logger.debug("Initialising instance of ShellHandler host:{0}".format(host))
    try:
        self.ssh = paramiko.SSHClient()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.ssh.connect(host, username=user, password=psw, port=22)
        self.channel = self.ssh.invoke_shell()
    except:
        logger.error("Error Creating ssh connection to {0}".format(host))
        logger.error("Exiting ShellHandler")
        return
    self.psw=psw
    self.stdin = self.channel.makefile('wb')
    self.stdout = self.channel.makefile('r')
    self.host=host
    time.sleep(2)

    while not self.channel.recv_ready():
        time.sleep(2)
    self.initialprompt=""
    while self.channel.recv_ready():

        rl, wl, xl = select.select([ self.stdout.channel ], [ ], [ ], 0.0)
        if len(rl) > 0:
            tmp = self.stdout.channel.recv(24)
            self.initialprompt=self.initialprompt+str(tmp.decode())

def __del__(self):
    self.ssh.close()
    logger.info("closed connection to {0}".format(self.host))

def execute(self, cmd):
    cmd = cmd.strip('\n')
    self.stdin.write(cmd + '\n')
    #self.stdin.write(self.psw +'\n')
    self.stdin.flush()
    time.sleep(1)
    while not self.stdout.channel.recv_ready():
        time.sleep(2)
        logger.debug("Waiting for recv_ready")

    output=""
    while self.channel.recv_ready():
        rl, wl, xl = select.select([ self.stdout.channel ], [ ], [ ], 0.0)
        if len(rl) > 0:
            tmp = self.stdout.channel.recv(24)
            output=output+str(tmp.decode())
    return output

如果每次创建不同的外壳对您来说并不重要,那么您可以使用下面的方法。

def run_cmd(self,cmd):
    try:
        cmd=cmd+'\n'
        #self.ssh.settimeout(60)
        stdin,stdout,stderr=self.ssh.exec_command(cmd)
        while not stdout.channel.eof_received:
           time.sleep(3)
           logger.debug("Waiting for eof_received")
        out=""
        while stdout.channel.recv_ready():
            err=stderr.read()
            if err:
                print("Error: ",my_hostname, str(err))
                return False 

            out=out+stdout.read()
        if out:
               return out 

    except:
        error=sys.exc_info()
        logger.error(error)
        return False

相关问题