使用Python脚本对Zabbix关键模版进行审计

前言

当一个企业的zabbix监控系统运行了很久之后，必然会产生某些机器的某几个模版很重要的情况，另外由于经常维护这些模版需要频繁的重新绑定，要是解绑和再次绑定之间有一定时间差的话那就很容易造成漏绑定，一旦发生漏绑定且此时业务系统刚好出现了问题没有监控到那么后果就很严重了，本篇博客提供的脚本就是为了解决这个问题而存在的。

本次用到的脚本如下所示：

 ~/Downloads/tmp  tree                 
.
├── zabbix_template_audit_every_hour.py
├── zbx_prod.conf
└── zhtmplfinder.py

~/Downloads/tmp  tree

├── zabbix_template_audit_every_hour.py

├── zbx_prod.conf

└── zhtmplfinder.py

代码分两块，一个是zabbix_template_audit_every_hour.py，一个是github上的开源脚本zhtmplfinder.py，仓库名叫q1x/zabbix-gnomes，感谢，该仓库年久失修，为了防止失联本人fork了一下，也在下面把源码直接贴了出来，请把两块代码放到同一级目录。

整个项目依赖python2，主要是这个七八年前的上古开源脚本zhtmplfinder.py也依赖python2。

依赖安装

python2 -m pip install pyzabbix==1.0.0 requests==2.26.0

1	python2 -m pip install pyzabbix==1.0.0 requests==2.26.0

使用

上述脚本的最后一个叫zhtmplfinder.py依赖一个下面格式叫zbx_prod.conf的配置文件：

[Zabbix API]
     username=johndoe
     password=verysecretpassword
     api=https://zabbix.mycompany.com/path/to/zabbix/frontend/
     no_verify=true

[Zabbix API]

username=johndoe

password=verysecretpassword

api=https://zabbix.mycompany.com/path/to/zabbix/frontend/

no_verify=true

主代码zabbix_template_audit_every_hour.py的运行方式如下，建议自行配置定时任务，每个小时进行一次审计:

python2 zabbix_template_audit_every_hour.py

1	python2 zabbix_template_audit_every_hour.py

代码

主程序zabbix_template_audit_every_hour.py

#!/usr/bin/env python
# encoding=utf-8

import json
import subprocess
import requests


host_with_templates_list = {
    '192.168.0.1': ['customize-discover-xxx-check1', 'customize-discover-xxx-check2'],
    '192.168.0.2': ['customize-discover-xxx-check1', 'customize-discover-xxx-check2'],
}


def send_msg_to_wework(chat_id, content):
    diag = {"chatid": chat_id,
            "msgtype": "markdown",
            "markdown": {
                "content": content}}

    headers = {"Content-Type": "application/json"}

    diag = json.dumps(diag)
    requests.post('http://xxxx.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxx',
                  data=diag, headers=headers)


def main():
    template_not_in_result_list = []
    try:
        for host in host_with_templates_list:
            templates_list= host_with_templates_list[host]
            cmd = "python zhtmplfinder.py -c zbx_prod.conf {0}".format(host)
            result = subprocess.check_output(cmd, shell=True).strip().split('\n')
            for template in templates_list:
                if template not in result:
                    template_not_in_result_list.append(host + ':' + template)
        if template_not_in_result_list:
            send_msg_to_wework('xxxxx',
            "# <font color='warning'>** 以下Zabbix模板审计异常,请及时进行绑定! <@xxxx> <@xxxx>**</font>\n{0}".format('\n'.join(template_not_in_result_list)))
    except subprocess.CalledProcessError:
        send_msg_to_wework('xxxxx', 
        "# <font color='warning'>**Zabbix模板审计脚本192.168.0.1:/usr/local/zabbix/scripts/zabbix_template_audit/zabbix_template_audit_every_hour.py发生错误,请检查! <@xxx> <@xxx>**</font>")


if __name__ == '__main__':
    main()

#!/usr/bin/env python

# encoding=utf-8

import json

import subprocess

import requests

host_with_templates_list = {

'192.168.0.1': ['customize-discover-xxx-check1', 'customize-discover-xxx-check2'],

'192.168.0.2': ['customize-discover-xxx-check1', 'customize-discover-xxx-check2'],

}

def send_msg_to_wework(chat_id, content):

diag = {"chatid": chat_id,

"msgtype": "markdown",

"markdown": {

"content": content}}

headers = {"Content-Type": "application/json"}

diag = json.dumps(diag)

requests.post('http://xxxx.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxx',

data=diag, headers=headers)

def main():

template_not_in_result_list = []

try:

for host in host_with_templates_list:

templates_list= host_with_templates_list[host]

cmd = "python zhtmplfinder.py -c zbx_prod.conf {0}".format(host)

result = subprocess.check_output(cmd, shell=True).strip().split('\n')

for template in templates_list:

if template not in result:

template_not_in_result_list.append(host + ':' + template)

if template_not_in_result_list:

send_msg_to_wework('xxxxx',

"# <font color='warning'>** 以下Zabbix模板审计异常,请及时进行绑定! <@xxxx> <@xxxx>**</font>\n{0}".format('\n'.join(template_not_in_result_list)))

except subprocess.CalledProcessError:

send_msg_to_wework('xxxxx',

"# <font color='warning'>**Zabbix模板审计脚本192.168.0.1:/usr/local/zabbix/scripts/zabbix_template_audit/zabbix_template_audit_every_hour.py发生错误,请检查! <@xxx> <@xxx>**</font>")

if __name__ == '__main__':

main()

依赖脚本zhtmplfinder.py：

#!/usr/bin/env python
#
# import needed modules.
# pyzabbix is needed, see https://github.com/lukecyca/pyzabbix
#
import argparse
import ConfigParser
import os
import os.path
import sys
import distutils.util
from pyzabbix import ZabbixAPI

# define config helper function
def ConfigSectionMap(section):
    dict1 = {}
    options = Config.options(section)
    for option in options:
 	try:
		dict1[option] = Config.get(section, option)
		if dict1[option] == -1:
			DebugPrint("skip: %s" % option)
	except:
		print("exception on %s!" % option)
		dict1[option] = None
    return dict1


# set default vars
defconf = os.getenv("HOME") + "/.zbx.conf"
username = ""
password = ""
api = ""
noverify = ""

# Define commandline arguments
parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,description='Tries to get the linked templates for the specified Zabbix host.', epilog="""
This program can use .ini style configuration files to retrieve the needed API connection information.
To use this type of storage, create a conf file (the default is $HOME/.zbx.conf) that contains at least the [Zabbix API] section and any of the other parameters:
       
 [Zabbix API]
 username=johndoe
 password=verysecretpassword
 api=https://zabbix.mycompany.com/path/to/zabbix/frontend/
 no_verify=true

""")
parser.add_argument('hostname', help='Hostname to find the linked templates for')
parser.add_argument('-u', '--username', help='User for the Zabbix api')
parser.add_argument('-p', '--password', help='Password for the Zabbix api user')
parser.add_argument('-a', '--api', help='Zabbix API URL')
parser.add_argument('--no-verify', help='Disables certificate validation when using a secure connection',action='store_true') 
parser.add_argument('-c','--config', help='Config file location (defaults to $HOME/.zbx.conf)')
parser.add_argument('-n', '--numeric', help='Return numeric templateids instead of template names',action='store_true')
parser.add_argument('-e', '--extended', help='Return both templateid and template name separated with a ":"',action='store_true')
args = parser.parse_args()

# load config module
Config = ConfigParser.ConfigParser()
Config

# if configuration argument is set, test the config file
if args.config:
 if os.path.isfile(args.config) and os.access(args.config, os.R_OK):
  Config.read(args.config)

# if not set, try default config file
else:
 if os.path.isfile(defconf) and os.access(defconf, os.R_OK):
  Config.read(defconf)

# try to load available settings from config file
try:
 username=ConfigSectionMap("Zabbix API")['username']
 password=ConfigSectionMap("Zabbix API")['password']
 api=ConfigSectionMap("Zabbix API")['api']
 noverify=bool(distutils.util.strtobool(ConfigSectionMap("Zabbix API")["no_verify"]))
except:
 pass

# override settings if they are provided as arguments
if args.username:
 username = args.username

if args.password:
 password = args.password

if args.api:
 api = args.api

if args.no_verify:
 noverify = args.no_verify

# test for needed params
if not username:
 sys.exit("Error: API User not set")

if not password:
 sys.exit("Error: API Password not set")
 
if not api:
 sys.exit("Error: API URL is not set")

# Setup Zabbix API connection
zapi = ZabbixAPI(api)

if noverify is True:
 zapi.session.verify = False

# Login to the Zabbix API
zapi.login(username, password)

##################################
# Start actual API logic
##################################

# set the hostname we are looking for
host_name = args.hostname

# Find specified host from API
hosts = zapi.host.get(output="extend", filter={"host": host_name})

if hosts:
    # Find linked templates
    templates = zapi.template.get(output="extend", hostids=hosts[0]["hostid"])
    if templates:
      if args.extended:
         # print ids and names
	 for template in templates:
	   print(format(template["templateid"])+":"+format(template["host"]))
      else:
        if args.numeric:
           # print template ids
	   for template in templates:
	     print(format(template["templateid"]))
        else:
           # print template names
	   for template in templates:
	     print(format(template["host"]))
    else:
       sys.exit("Error: No templates linked to "+ host_name)
else:
    sys.exit("Error: Could not find host "+ host_name)

# And we're done...

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

#!/usr/bin/env python

# import needed modules.

# pyzabbix is needed, see https://github.com/lukecyca/pyzabbix

import argparse

import ConfigParser

import os

import os.path

import sys

import distutils.util

from pyzabbix import ZabbixAPI

# define config helper function

def ConfigSectionMap(section):

dict1 = {}

options = Config.options(section)

for option in options:

try:

dict1[option] = Config.get(section, option)

if dict1[option] == -1:

DebugPrint("skip: %s" % option)

except:

print("exception on %s!" % option)

dict1[option] = None

return dict1

# set default vars

defconf = os.getenv("HOME") + "/.zbx.conf"

username = ""

password = ""

api = ""

noverify = ""

# Define commandline arguments

parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,description='Tries to get the linked templates for the specified Zabbix host.', epilog="""

This program can use .ini style configuration files to retrieve the needed API connection information.

To use this type of storage, create a conf file (the default is $HOME/.zbx.conf) that contains at least the [Zabbix API] section and any of the other parameters:

[Zabbix API]

username=johndoe

password=verysecretpassword

api=https://zabbix.mycompany.com/path/to/zabbix/frontend/

no_verify=true

""")

parser.add_argument('hostname', help='Hostname to find the linked templates for')

parser.add_argument('-u', '--username', help='User for the Zabbix api')

parser.add_argument('-p', '--password', help='Password for the Zabbix api user')

parser.add_argument('-a', '--api', help='Zabbix API URL')

parser.add_argument('--no-verify', help='Disables certificate validation when using a secure connection',action='store_true')

parser.add_argument('-c','--config', help='Config file location (defaults to $HOME/.zbx.conf)')

parser.add_argument('-n', '--numeric', help='Return numeric templateids instead of template names',action='store_true')

parser.add_argument('-e', '--extended', help='Return both templateid and template name separated with a ":"',action='store_true')

args = parser.parse_args()

# load config module

Config = ConfigParser.ConfigParser()

Config

# if configuration argument is set, test the config file

if args.config:

if os.path.isfile(args.config) and os.access(args.config, os.R_OK):

Config.read(args.config)

# if not set, try default config file

else:

if os.path.isfile(defconf) and os.access(defconf, os.R_OK):

Config.read(defconf)

# try to load available settings from config file

try:

username=ConfigSectionMap("Zabbix API")['username']

password=ConfigSectionMap("Zabbix API")['password']

api=ConfigSectionMap("Zabbix API")['api']

noverify=bool(distutils.util.strtobool(ConfigSectionMap("Zabbix API")["no_verify"]))

except:

pass

# override settings if they are provided as arguments

if args.username:

username = args.username

if args.password:

password = args.password

if args.api:

api = args.api

if args.no_verify:

noverify = args.no_verify

# test for needed params

if not username:

sys.exit("Error: API User not set")

if not password:

sys.exit("Error: API Password not set")

if not api:

sys.exit("Error: API URL is not set")

# Setup Zabbix API connection

zapi = ZabbixAPI(api)

if noverify is True:

zapi.session.verify = False

# Login to the Zabbix API

zapi.login(username, password)

##################################

# Start actual API logic

##################################

# set the hostname we are looking for

host_name = args.hostname

# Find specified host from API

hosts = zapi.host.get(output="extend", filter={"host": host_name})

if hosts:

# Find linked templates

templates = zapi.template.get(output="extend", hostids=hosts[0]["hostid"])

if templates:

if args.extended:

# print ids and names

for template in templates:

print(format(template["templateid"])+":"+format(template["host"]))

else:

if args.numeric:

# print template ids

for template in templates:

print(format(template["templateid"]))

else:

# print template names

for template in templates:

print(format(template["host"]))

else:

sys.exit("Error: No templates linked to "+ host_name)

else:

sys.exit("Error: Could not find host "+ host_name)

# And we're done...

请自行替换主程序中的机器ip和zabbix模版名，以及企业微信api地址和chat id。

本作品采用知识共享署名 4.0 国际许可协议进行许可