Commit 608137ba authored by Olav Morken's avatar Olav Morken

Initial commit

#!/usr/bin/env python3
import argparse
import getpass
import json
import os
import requests
import sys
base_dir = os.path.dirname(os.path.realpath(__file__))
auth = None
def get_auth():
with open(os.path.join(base_dir, '.zabbixauth'), 'r') as fh:
return None
def invoke(method, **kwargs):
message = {
"jsonrpc": "2.0",
"method": method,
"params": kwargs,
"id": 1,
"auth": get_auth(),
url = ''
headers = {
'Content-Type': 'application/json-rpc',
r =, headers=headers, json=message)
response = r.json()
if not 'result' in response:
raise Exception('Invalid response from Zabbix: {response}'.format(response=repr(response)))
return response['result']
def login():
username = input('Username: ')
password = getpass.getpass()
result = invoke('user.login', user=username, password=password)
with open(os.path.join(base_dir, '.zabbixauth'), 'w') as fh:
fh.write(result + "\n")
def get_items(args):
if not args.include_inherited:
item_filter = {'inherited': False}
item_filter = {}
for template in invoke('template.get'):
for item in invoke('item.get', templateids=template['templateid'], **item_filter):
if args.only_external and item['type'] != '10':
# Not an external check
if args.only_enabled and item['status'] == '1':
print('{itemid} {key}'.format(itemid=item['itemid'], key=item['key_']))
for host in invoke('host.get'):
for item in invoke('item.get', hostids=host['hostid'], **item_filter):
if args.only_external and item['type'] != '10':
# Not an external check
if args.only_enabled and item['status'] == '1':
print('{hostid}&itemid={itemid} {key}'.format(hostid=item['hostid'], itemid=item['itemid'], key=item['key_']))
def dump_vkilo():
ilo_template = 10874
template = invoke('template.get', templateids=ilo_template)
ilohosts = invoke('host.get', templateids=ilo_template, selectMacros='extend')
for host in ilohosts:
vkilo = [ macro['value'] for macro in host['macros'] if macro['macro'] == '{$VKILO}' ]
vkilo = vkilo[0] if vkilo else None
print('{hostid}'.format(hostid=host['hostid']), host['host'], vkilo)
def get_item(args):
items = invoke('item.get', itemids=args.itemid)
if not items:
print('No item with ID {itemid}'.format(itemid=args.itemid), file=sys.stderr)
item = items[0]
json.dump(item, sys.stdout, sort_keys=True, indent=4)
print() # Add trailing newline
def get_web_tests(args):
if not args.include_inherited:
web_test_filter = {'inherited': False}
web_test_filter = {}
for httptest in invoke('httptest.get', selectSteps='extend', **web_test_filter):
url = '{httptestid}&hostid={hostid}&templateid={templateid}'.format(httptestid=httptest['httptestid'], hostid=httptest['hostid'], templateid=httptest['templateid'])
print(url, httptest['name'])
for index, step in enumerate(httptest['steps']):
print('#{index} {url}'.format(index=index, url=step['url']))
def get_unreachable():
for host in invoke('host.get'):
if int(host['available']) != 2:
continue # Host is up
url = '{hostid}'.format(hostid=host['hostid'])
print(url, host['name'], host['error'])
def get_hosts():
for host in invoke('host.get'):
url = '{hostid}'.format(hostid=host['hostid'])
print(url, host['name'])
def parse_args():
parser = argparse.ArgumentParser('Zabbix cmd line client')
subparsers = parser.add_subparsers(help='Command', dest='command')
subparsers.required = True
parser_login = subparsers.add_parser('login', help='login')
parser_get_items = subparsers.add_parser('get-items', help='Get script items')
parser_get_items.add_argument('--include-inherited', action='store_true', help='Include inherited items in output')
parser_get_items.add_argument('--only-enabled', action='store_true', help='Show only enabled items')
parser_get_items.add_argument('--only-external', action='store_true', help='Show only external check items')
parser_dump_vkilo = subparsers.add_parser('dump-vkilo', help='Get verktøykasse ILO config')
parser_get_item = subparsers.add_parser('get-item', help='Get item by ID')
parser_get_item.add_argument('itemid', type=int, help='Item ID')
parser_get_web_tests = subparsers.add_parser('get-web-tests', help='Get Web tests')
parser_get_web_tests.add_argument('--include-inherited', action='store_true', help='Include inherited items in output')
parser_get_unreachable = subparsers.add_parser('get-unreachable', help='Get unreachable hosts')
parser_get_hosts = subparsers.add_parser('get-hosts', help='Get hosts')
return parser.parse_args()
def main():
args = parse_args()
if args.command == 'login':
return login()
elif args.command == 'get-items':
return get_items(args)
elif args.command == 'dump-vkilo':
return dump_vkilo()
elif args.command == 'get-item':
return get_item(args)
elif args.command == 'get-web-tests':
return get_web_tests(args)
elif args.command == 'get-unreachable':
return get_unreachable()
elif args.command == 'get-hosts':
return get_hosts()
raise NotImplementedError('Unknown command: {command}'.format(command=args.command))
if __name__ == '__main__':
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment