Ansible Python API

官方文档

参考

• 官方示例

#!/usr/bin/env python

import json
import shutil
from ansible.module_utils.common.collections import ImmutableDict
from ansible.parsing.dataloader import DataLoader
from ansible.vars.manager import VariableManager
from ansible.inventory.manager import InventoryManager
from ansible.playbook.play import Play
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.plugins.callback import CallbackBase
from ansible import context
import ansible.constants as C

class ResultCallback(CallbackBase):
    """A sample callback plugin used for performing an action as results come in

    If you want to collect all results into a single object for processing at
    the end of the execution, look into utilizing the ``json`` callback plugin
    or writing your own custom callback plugin
    """
    def v2_runner_on_ok(self, result, **kwargs):
        """Print a json representation of the result

        This method could store the result in an instance attribute for retrieval later
        """
        host = result._host
        print(json.dumps({host.name: result._result}, indent=4))

# since the API is constructed for CLI it expects certain options to always be set in the context object
context.CLIARGS = ImmutableDict(connection='local', module_path=['/to/mymodules'], forks=10, become=None,
                                become_method=None, become_user=None, check=False, diff=False)

# initialize needed objects
loader = DataLoader() # Takes care of finding and reading yaml, json and ini files
passwords = dict(vault_pass='secret')

# Instantiate our ResultCallback for handling results as they come in. Ansible expects this to be one of its main display outlets
results_callback = ResultCallback()

# create inventory, use path to host config file as source or hosts in a comma separated string
inventory = InventoryManager(loader=loader, sources='localhost,')

# variable manager takes care of merging all the different sources to give you a unified view of variables available in each context
variable_manager = VariableManager(loader=loader, inventory=inventory)

# create data structure that represents our play, including tasks, this is basically what our YAML loader does internally.
play_source =  dict(
        name = "Ansible Play",
        hosts = 'localhost',
        gather_facts = 'no',
        tasks = [
            dict(action=dict(module='shell', args='ls'), register='shell_out'),
            dict(action=dict(module='debug', args=dict(msg='{{shell_out.stdout}}')))
         ]
    )

# Create play object, playbook objects use .load instead of init or new methods,
# this will also automatically create the task objects from the info provided in play_source
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)

# Run it - instantiate task queue manager, which takes care of forking and setting up all objects to iterate over host list and tasks
tqm = None
try:
    tqm = TaskQueueManager(
              inventory=inventory,
              variable_manager=variable_manager,
              loader=loader,
              passwords=passwords,
              stdout_callback=results_callback,  # Use our custom callback instead of the ``default`` callback plugin, which prints to stdout
          )
    result = tqm.run(play) # most interesting data for a play is actually sent to the callback's methods
finally:
    # we always need to cleanup child procs and the structures we use to communicate with them
    if tqm is not None:
        tqm.cleanup()

    # Remove ansible tmpdir
    shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)

• 执行ad-hoc任务

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
from collections import namedtuple
# 核心类
# 用于读取YAML和JSON格式的文件
from ansible.parsing.dataloader import DataLoader
# 用于存储各类变量信息
from ansible.vars.manager import VariableManager
# 用于导入资产文件
from ansible.inventory.manager import InventoryManager
# 操作单个主机信息
from ansible.inventory.host import Host
# 操作单个主机组信息
from ansible.inventory.group import Group
# 存储执行hosts的角色信息
from ansible.playbook.play import Play
# ansible底层用到的任务队列
from ansible.executor.task_queue_manager import TaskQueueManager
# 核心类执行playbook
from ansible.executor.playbook_executor import PlaybookExecutor


def adhoc():
    """
    ad-hoc 调用
    资产配置信息  这个是通过 InventoryManager和VariableManager 定义
    执行选项 这个是通过namedtuple来定义
    执行对象和模块 通过dict()来定义
    定义play 通过Play来定义
    最后通过 TaskQueueManager 的实例来执行play
    :return:
    """
    # 资产配置信息
    dl = DataLoader()
    im = InventoryManager(loader=dl, sources=["hosts"])
    vm = VariableManager(loader=dl, inventory=im)
    # 执行选项，这个类不是ansible的类，这个的功能就是为了构造参数
    Options = namedtuple("Options", [
        "connection", "remote_user", "ask_sudo_pass", "verbosity", "ack_pass",
        "module_path", "forks", "become", "become_method", "become_user", "check",
        "listhosts", "listtasks", "listtags", "syntax", "sudo_user", "sudo", "diff"
    ])
    """
    这里就是Options的实例，然后你就可以赋值，这个为了给ansible设置执行选项 ansibile 172.16.48.171 -m shell -a 'ls /tmp' -f 5
    这里的选项就是ansible命令中 -f -C -D -m等执行选项
    """
    options = Options(connection='smart', remote_user=None, ack_pass=None, sudo_user=None, forks=5, sudo=None, ask_sudo_pass=False,
                      verbosity=5, module_path=None, become=None, become_method=None, become_user=None, check=False, diff=False,
                      listhosts=None, listtasks=None, listtags=None, syntax=None)
    # play的执行对象和模块，这里设置hosts，其实是因为play把play_source和资产信息关联后，执行的play的时候它会去资产信息中设置的sources的hosts文件中
    # 找你在play_source中设置的hosts是否在资产管理类里面。
    play_source = dict(name="Ansible Play",  # 任务名称
                       hosts="172.16.48.242",  # 目标主机，可以填写具体主机也可以是主机组名称
                       gather_facts="no",  # 是否收集配置信息

                       # tasks是具体执行的任务，列表形式，每个具体任务都是一个字典
                       tasks=[
                           dict(action=dict(module="shell", args="ls /tmp"))
                       ])
    # 定义play
    play = Play().load(play_source, variable_manager=vm, loader=dl)

    passwords = dict()  # 这个可以为空，因为在hosts文件中
    #
    tqm = TaskQueueManager(
        inventory=im,
        variable_manager=vm,
        loader=dl,
        options=options,
        passwords=passwords,
    )
    result = tqm.run(play)
    print(result)


def main():
    adhoc()


if __name__ == "__main__":
    try:
        main()
    finally:
        sys.exit()

• 执行playbook任务

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
from collections import namedtuple
# 核心类
# 用于读取YAML和JSON格式的文件
from ansible.parsing.dataloader import DataLoader
# 用于存储各类变量信息
from ansible.vars.manager import VariableManager
# 用于导入资产文件
from ansible.inventory.manager import InventoryManager
# 操作单个主机信息
from ansible.inventory.host import Host
# 操作单个主机组信息
from ansible.inventory.group import Group
# 存储执行hosts的角色信息
from ansible.playbook.play import Play
# ansible底层用到的任务队列
from ansible.executor.task_queue_manager import TaskQueueManager
# 核心类执行playbook
from ansible.executor.playbook_executor import PlaybookExecutor


def execplaybook():
    """
    调用 playbook
    调用playboo大致和调用ad-hoc相同，只是真正调用的是使用PlaybookExecutor
    :return:
    """
    # 资产配置信息
    dl = DataLoader()
    im = InventoryManager(loader=dl, sources=["hosts"])
    vm = VariableManager(loader=dl, inventory=im)
    # 执行选项，这个类不是ansible的类，这个的功能就是为了构造参数
    Options = namedtuple("Options", [
        "connection", "remote_user", "ask_sudo_pass", "verbosity", "ack_pass",
        "module_path", "forks", "become", "become_method", "become_user", "check",
        "listhosts", "listtasks", "listtags", "syntax", "sudo_user", "sudo", "diff"
    ])
    """
    这里就是Options的实例，然后你就可以赋值，这个为了给ansible设置执行选项 ansibile 172.16.48.171 -m shell -a 'ls /tmp' -f 5
    这里的选项就是ansible命令中 -f -C -D -m等执行选项
    """
    options = Options(connection='smart', remote_user=None, ack_pass=None, sudo_user=None, forks=5, sudo=None,
                      ask_sudo_pass=False,
                      verbosity=5, module_path=None, become=None, become_method=None, become_user=None, check=False,
                      diff=False,
                      listhosts=None, listtasks=None, listtags=None, syntax=None)
    passwords = dict()  # 这个可以为空，因为在hosts文件中
    #
    try:
        # playbooks参数里面放的就是playbook文件路径
        playbook = PlaybookExecutor(playbooks=["f1.yml"], inventory=im, variable_manager=vm, loader=dl, options=options, passwords=passwords)
        playbook.run()
    except Exception as err:
        print(err)


def main():
    execplaybook()

if __name__ == "__main__":
    try:
        main()
    finally:
        sys.exit()

• ansible管理iptables

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""ansible管理iptables"""

import shutil

import ansible.constants as C
from ansible import context
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.inventory.manager import InventoryManager
from ansible.module_utils.common.collections import ImmutableDict
from ansible.parsing.dataloader import DataLoader
from ansible.playbook.play import Play
from ansible.plugins.callback import CallbackBase
from ansible.vars.manager import VariableManager


class ResultCallback(CallbackBase):
    """
    回调类，给ansible执行ad-hoc命令后回调使用
    """

    def __init__(self):
        super(ResultCallback, self).__init__()
        self.result = None

    def v2_runner_on_ok(self, result, **kwargs):
        """
        运行成功时回调
        :param result:
        :param kwargs:
        :return: 返回json对象
        """
        host = result._host
        self.result = {host.name: result._result}

    def v2_runner_on_failed(self, result, ignore_errors=False):
        """
        运行失败时回调
        :param result:
        :param ignore_errors:
        :return: 返回json对象
        """
        host = result._host
        self.result = {host.name: result._result}

    def v2_runner_on_unreachable(self, result):
        """
        主机连接失败时回调
        :param result:
        :return: 返回json对象
        """
        host = result._host
        self.result = {host.name: result._result}


class FireWall(object):
    """
    iptables规则核心类
    """

    def __init__(self, inventory_path, hosts='all', remote_user=None, become_user='root', password=None):
        """

        :param inventory_path: str  inventory文件路径 必传参数
        :param hosts: str 传入的主机，和inventory文件的需要匹配上 默认所有主机
        :param remote_user: str 远程连接用户 非必传参数 默认app用户
        :param become_user: str 远程执行命令用户 非必传参数 默认root用户
        :param password: str 远程主机密码 非必传参数 默认使用ssh公钥对连接
        """
        context.CLIARGS = ImmutableDict(
            connection='smart',
            remote_user=remote_user,
            ack_pass=None,
            sudo_user=None,
            forks=5,
            sudo=True,
            ask_sudo_pass=False,
            verbosity=5,
            module_path=None,
            become=True,
            become_method='sudo',
            become_user=become_user,
            check=False,
            diff=False,
            listhosts=None,
            listtasks=None,
            listtags=None,
            syntax=None
        )
        self._loader = DataLoader()
        self._passwords = dict() if password is None else dict(vault_pass=password)
        self._results_callback = ResultCallback()
        self._inventory = InventoryManager(loader=self._loader, sources=[inventory_path])
        self._variable_manager = VariableManager(loader=self._loader, inventory=self._inventory)
        self._hosts = hosts
        self._redis_key = 'bdc_aomp_firewall_iptables_{}'.format('_'.join(self._hosts.split('.')))

    def _run_shell_task(self, name, cmd):
        """
        执行shell命令私有类
        :param name:  任务名称
        :param cmd:   shell命令
        :return: 返回回调类的结果
        """
        play_source = dict(
            name=name,
            hosts=self._hosts,
            gather_facts='no',
            tasks=[
                dict(action=dict(module='shell', args=cmd), register='shell_out'),
            ]
        )

        play = Play().load(play_source, variable_manager=self._variable_manager, loader=self._loader)

        tqm = None
        try:
            tqm = TaskQueueManager(
                inventory=self._inventory,
                variable_manager=self._variable_manager,
                loader=self._loader,
                passwords=self._passwords,
                stdout_callback=self._results_callback,

            )
            tqm.run(play)
        finally:
            if tqm is not None:
                tqm.cleanup()

            shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)
        return self._results_callback.result

    def _save_rule(self):
        """
        用来保存iptables的规则方法，默认添加和删除的iptables规则保存在内存中，执行完新增和删除规则后需要调用该方法
        :return: tuple (True, None)
        """
        cmd = 'service iptables save'
        res = self._run_shell_task('append rules', cmd)
        if res[self._hosts]['rc'] != 0:
            return False, res[self._hosts]['stderr']
        else:
            return True, None

    def get_iptables(self):
        """
        通过ansible获取远程主机的iptables规则
        :return: tuple (rules, None)
        """
        name = 'get iptables'
        cmd = 'iptables -nL --line-number'
        result = self._run_shell_task(name, cmd)
        if result is None:
            return None, 'get iptables failed!'
        if result[self._hosts].get('unreachable') is True:
            return None, result[self._hosts]['msg']
        if result[self._hosts]['rc'] != 0:
            return None, result[self._hosts]['stderr']
        try:
            forward_index = result[self._hosts]['stdout_lines'].index('Chain FORWARD (policy DROP)')
        except Exception:
            forward_index = result[self._hosts]['stdout_lines'].index('Chain FORWARD (policy ACCEPT)')
        output_index = result[self._hosts]['stdout_lines'].index('Chain OUTPUT (policy ACCEPT)')
        input_list = []
        forward_list = []
        output_list = []
        for i, data in enumerate(result[self._hosts]['stdout_lines']):
            res_dict = {
                'num': None,
                'target': None,
                'prot': None,
                'opt': None,
                'source': None,
                'destination': None,
                'rules': None
            }
            if len(data) > 0 and 'num' not in data and 'Chain' not in data:
                rules = data.split()
                res_dict['num'] = rules[0]
                res_dict['target'] = rules[1]
                res_dict['prot'] = rules[2]
                res_dict['opt'] = rules[3]
                res_dict['source'] = rules[4]
                res_dict['destination'] = rules[5]

                if len(rules) > 6:
                    res_dict['rules'] = ' '.join(rules[6:])
                if i < forward_index:
                    input_list.append(res_dict)
                elif i < output_index:
                    forward_list.append(res_dict)
                else:
                    output_list.append(res_dict)

        res = {"input": input_list, "forward": forward_list, "output": output_list}
        return res, None

    def get_hosts(self):
        """
        获取管理的主机列表
        :return: dict
        """
        return {'hosts': [i.address for i in self._inventory.get_hosts()]}

    def append_rule(self, target, source=None, dpt=None, prot='all', chain='INPUT'):
        """
        向iptables规则后面新增规则
        :param target: str 放行还是屏蔽 指[ACCEPT, DROP]
        :param chain: str 出入站或者转发的目标规则，大些，特指 [INPUT, FORWARD, OUTPUT]三个值
        :param source: str 源ip地址
        :param dpt: str 目的地端口
        :param prot: str 协议
        :return: tuple (True, None)
        """
        if not any([source, dpt]):
            return False, 'source 或者 dpt参数缺失'

        if dpt is not None and prot not in ['tcp', 'udp']:
            return False, 'dpt 参数不能出现在非tcp或者udp协议内'

        if source is None and prot in ['tcp', 'udp']:
            cmd = 'iptables -A {chain} -p {prot} --dport {dpt} -j {target}'.format(
                chain=chain, prot=prot, dpt=dpt, target=target
            )
        elif all([chain, prot, source, dpt]):
            cmd = 'iptables -A {chain} -p {prot} -s {source} --dport {dpt} -j {target}'.format(
                chain=chain, prot=prot, source=source, dpt=dpt, target=target
            )
        else:
            cmd = 'iptables -A {chain} -p {prot} -s {source} -j {target}'.format(
                chain=chain, prot=prot, source=source, target=target
            )
        res = self._run_shell_task('append rules', cmd)
        if res[self._hosts]['rc'] != 0:
            return False, res[self._hosts]['stderr']
        else:
            res, exception = self._save_rule()
            if exception is not None:
                return False, exception

            return True, None

    def delete_rule(self, num, chain='INPUT'):
        """
        删除iptables规则
        :param chain: str 出入站或者转发的目标规则，大些，特指 [INPUT, FORWARD, OUTPUT]三个值
        :param num: str 对应的股则序号id
        :return: tuple (True, None)
        """
        cmd = 'iptables -D {} {}'.format(chain, num)
        res = self._run_shell_task('append rules', cmd)
        if res[self._hosts]['rc'] != 0:
            return False, res[self._hosts]['stderr']
        else:
            res, exception = self._save_rule()
            if exception is not None:
                return False, exception

            return True, None

    def check_rules(self, num, chain='INPUT'):
        """
        检查iptables的规则序号是否正确
        :param chain: str 出入站或者转发的目标规则，大些，特指 [INPUT, FORWARD, OUTPUT]三个值
        :param num: str 出入站或者转发的目标规则，大些，特指 [INPUT, FORWARD, OUTPUT]三个值
        :return: tuple (True, None)
        """
        res, exception = self.get_iptables()
        if exception is not None:
            return False, exception
        data = res[chain.lower()]
        nums = [i['num'] for i in data]
        if num not in nums:
            return False, '{}规则内没有num为{}的规则！'.format(chain, num)
        else:
            return True, None

    def search_rules(self, source=None, dpt=None, prot=None, chain='INPUT'):
        if not any([source, dpt, prot]):
            return None, '缺少查询条件'
        data, err = self.get_iptables()
        if err is not None:
            return None, err

        data = data[chain.lower()]
        res = list()
        for i in data:
            if all([source, dpt, prot]):
                if source in i['source'] and (i['rules'] is not None and dpt in i['rules']) and prot in i['prot']:
                    res.append(i)
            elif all([source, dpt]):
                if source in i['source'] and (i['rules'] is not None and dpt in i['rules']):
                    res.append(i)
            elif all([dpt, prot]):
                if dpt in i['source'] and prot in i['prot']:
                    res.append(i)
            elif source:
                if source in i['source']:
                    res.append(i)
            elif dpt:
                if i['rules'] is not None and dpt in i['rules']:
                    res.append(i)
            elif prot:
                if prot in i['prot']:
                    res.append(i)
        return res, None