通过python脚本实现ansible远程命令

Posted 2023-05-15

参考技术A import paramiko

import getpass

import sys

import threading

import os

def rcmd(host, port=22, user='root', passwd=None, cmd=None):

    ssh = paramiko.SSHClient()

    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    ssh.connect(host, port=port, username=user, password=passwd)

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

    out = stdout.read()

    err = stderr.read()

    if out:

        print(f'[host] \033[32;1mOUT\033[0m:\nout.decode()')

    if err:

        print(f'[host] \033[31;1mERROR\033[0m:\nerr.decode()')

if __name__ == '__main__':

    # rcmd('127.0.0.1', passwd='redhat', cmd='id root; id zhangsan')

    if len(sys.argv) < 3:  # 命令加参数长度不能小于3

        print(f'Usage: sys.argv[0] ipfile commands')

        exit(1)

    if not os.path.isfile(sys.argv[1]):  # ip地址文件必须存在

        print(f'No such file: sys.argv[1]')

        exit(2)

    ipfile = sys.argv[1]

    cmd = ' '.join(sys.argv[2:])  # 拼接出命令

    passwd = getpass.getpass()

    with open(ipfile) as fobj:

        for line in fobj:

            ip = line.strip()  # 去除行尾的的\n

            t = threading.Thread(target=rcmd, args=(ip, 22, 'root', passwd, cmd))

            t.start()

python脚本实现本地或远程执行命令

功能：
1、执行本地shell命令，执行完成后获取结果
2、执行本地shell命令，执行中实时获取输出结果
3、执行远程shell命令，执行完成后获取结果
4、执行远程shell命令，执行中实时获取输出结果

实际操作：
1、安装paramiko

apt-get install  python3-pip libevent-dev libffi-dev libssl-dev -y
pip3 install paramiko -i https://pypi.mirrors.ustc.edu.cn/simple/  --trusted-host https://pypi.mirrors.ustc.edu.cn

2、创建脚本

[email protected]:~# mkdir /scripts/python -p
[email protected]:~# touch /scripts/python/shell.py
[email protected]:~# cat /scripts/python/shell.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import subprocess
import paramiko
import re
import sys
class Cmd(object):
    def onetime_shell(self,cmd):
        cmd = subprocess.Popen(cmd,shell=True,stdout=subprocess.PIPE)
        cmd = cmd.communicate()
        cmd = cmd[0].decode().rstrip()
        return cmd
    def realtime_shell(self,cmd):
        cmd = subprocess.call(cmd, shell=True)
        return cmd
class Remote_cmd(object):
    def __init__(self,PrivateKey,IP,Port,User):
        self.private_key = paramiko.RSAKey.from_private_key_file(PrivateKey)
        self.ssh = paramiko.SSHClient()
        self.set_missing_host_key_policy = self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.connect = self.ssh.connect(hostname=IP, port=Port, username=User,pkey=self.private_key)
    def onetime_shell(self,cmd,notice=False):
        stdin, stdout, stderr = self.ssh.exec_command(cmd)
        result = stdout.read().decode(‘utf-8‘).rstrip()
        if notice:
           self.ssh.close()
        return result

    def realtime_shell(self,cmd,notice=False):
        try:
           stdin, stdout, stderr = self.ssh.exec_command(cmd)
           for line in stdout:
               print(line.strip("
"))
           for error in stderr:
               print(error.strip("
"))
           if notice:
              self.ssh.close()
        except Exception as e:
           print("execute command %s error, error message is %s" % (cmd, e))
           return ""

例子：
1、本地执行shell命令，执行完成后获取结果：
mkdir /tmp/shell #创建目录/tmp/shell
echo shell >> /tmp/shell/shell.log # 输出shell 写入/tmp/shell/shell.log
2、本地执行shell命令，实时获取输出结果
apt-get update #更新
3、远程执行shell命令，执行完成后获取结果
mkdir /tmp/remote_shell #创建目录/tmp/remote_shell
echo remote_shell >> /tmp/remote_shell/remote_shell.log #输出remote_shell写入/tmp/remote_shell/remote_shell.log
4、远程执行shell命令，实时获取输出结果
apt-get update #更新

[email protected]:~# cat exec_shell.py 
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
sys.path.append(‘/scripts/python/‘)
from shell import Cmd,Remote_cmd
class ExecShell (object):
   def __init__(self):
       self.cmd = Cmd()
       self.remote_nfs_server = Remote_cmd(‘/root/.ssh/id_rsa‘,‘nfs-server‘,‘22‘,‘root‘)
   def local_onetime_shell(self):
       print("执行本地shell命令，执行完成后获取结果")
       self.cmd.onetime_shell(‘mkdir /tmp/shell‘)
       self.cmd.onetime_shell(‘echo shell >> /tmp/shell/shell.log‘)
       re = self.cmd.onetime_shell(‘cat  /tmp/shell/shell.log‘)
       print(re)
   def local_realtime_shell(self):
       print("执行本地shell命令，执行中实时获取输出结果")
       self.cmd.realtime_shell(‘apt-get update‘)
   def remote_onetime_shell(self):
       print("执行远程shell命令，执行完成后获取结果")
       self.remote_nfs_server.onetime_shell(‘mkdir /tmp/remote_shell‘)
       self.remote_nfs_server.onetime_shell(‘echo remote_shell >> /tmp/remote_shell/remote_shell.log‘)
       re = self.remote_nfs_server.onetime_shell(‘cat /tmp/remote_shell/remote_shell.log‘)
       print(re)
   def remote_realtime_shell(self):
       print("执行远程shell命令，执行中实时获取输出结果")
       self.cmd.realtime_shell(‘apt-get update‘)
execshell = ExecShell()
execshell.local_onetime_shell()
execshell.local_realtime_shell()
execshell.remote_onetime_shell()
execshell.remote_realtime_shell()

# 执行脚本结果
[email protected]:~# ./exec_shell.py 
执行本地shell命令，执行完成后获取结果
shell
执行本地shell命令，执行中实时获取输出结果
Hit:1 http://mirrors.aliyun.com/docker-ce/linux/ubuntu xenial InRelease
Get:2 http://repo.percona.com/apt jessie InRelease [15.9 kB]                                                                                                                            
Hit:3 http://us.archive.ubuntu.com/ubuntu xenial InRelease                                                                                                   
Get:4 http://security.ubuntu.com/ubuntu xenial-security InRelease [107 kB]                                
Hit:5 http://us.archive.ubuntu.com/ubuntu xenial-updates InRelease                                                   
Ign:2 http://repo.percona.com/apt jessie InRelease                                  
Hit:6 http://us.archive.ubuntu.com/ubuntu xenial-backports InRelease                
Fetched 123 kB in 1s (68.6 kB/s)                             
Reading package lists... Done
W: GPG error: http://repo.percona.com/apt jessie InRelease: The following signatures couldn‘t be verified because the public key is not available: NO_PUBKEY 9334A25F8507EFA5
W: The repository ‘http://repo.percona.com/apt jessie InRelease‘ is not signed.
N: Data from such a repository can‘t be authenticated and is therefore potentially dangerous to use.
N: See apt-secure(8) manpage for repository creation and user configuration details.
执行远程shell命令，执行完成后获取结果
remote_shell
执行远程shell命令，执行中实时获取输出结果
Hit:1 http://mirrors.aliyun.com/docker-ce/linux/ubuntu xenial InRelease
Get:2 http://repo.percona.com/apt jessie InRelease [15.9 kB]                                                                                                                            
Hit:3 http://us.archive.ubuntu.com/ubuntu xenial InRelease                                                                                                   
Get:4 http://security.ubuntu.com/ubuntu xenial-security InRelease [107 kB]                                
Hit:5 http://us.archive.ubuntu.com/ubuntu xenial-updates InRelease                                                   
Ign:2 http://repo.percona.com/apt jessie InRelease                                  
Hit:6 http://us.archive.ubuntu.com/ubuntu xenial-backports InRelease                
Fetched 123 kB in 1s (63.9 kB/s)                             
Reading package lists... Done
W: GPG error: http://repo.percona.com/apt jessie InRelease: The following signatures couldn‘t be verified because the public key is not available: NO_PUBKEY 9334A25F8507EFA5
W: The repository ‘http://repo.percona.com/apt jessie InRelease‘ is not signed.
N: Data from such a repository can‘t be authenticated and is therefore potentially dangerous to use.
N: See apt-secure(8) manpage for repository creation and user configuration details.