Collection of CheckMK checks (see https://checkmk.com/). All checks and plugins are provided as is. Absolutely no warranty. Send any comments to thl-cmk[at]outlook[dot]com

Skip to content
Snippets Groups Projects
Commit 7bdc35fe authored by thl-cmk's avatar thl-cmk :flag_na:
Browse files

Delete agent_fritzbox_smarthome.py

parent ec6574bb
No related branches found
No related tags found
No related merge requests found
#!/usr/bin/env python3
# -*- encoding: utf-8; py-indent-offset: 4 -*-
#
# modifications by thl-cmk[at]outlook[dot]com
# 2023-12-18: modified to work with cmk 2.2.x
# changed to return the complete XML response back as json
# 2023-12-28: added data/option for testing
import sys
import traceback
import ssl
import json
import time
from urllib.request import urlopen
import argparse
import xml.etree.ElementTree as ET
import hashlib
from re import sub as re_sub
from cmk.utils.password_store import replace_passwords
# based on: https://stackoverflow.com/a/47081240
def parse_xml_to_json(xml):
response = {}
for key in xml.keys():
response[key] = xml.get(key)
for child in list(xml):
for key in child.keys():
response[key] = child.get(key)
if len(list(child)) > 0:
response[child.tag] = parse_xml_to_json(child)
else:
response[child.tag] = child.text or ''
return response
def parse_args():
parser = argparse.ArgumentParser(
description='Check_MK Fritz!Box Smarthome Agent\n'
'This is an additional check_MK Fritz!Box Agent which can gather information\'s over the \n'
'AVM AHA HTTP Interface about SmartHome Devices connected to an Fritz!Box.',
formatter_class=argparse.RawTextHelpFormatter,
)
parser.add_argument(
'host',
help='Host name or IP address of your Fritz!Box',
)
parser.add_argument(
'--debug', action='store_true', default=False,
help='Debug mode: let Python exceptions come through',
)
parser.add_argument(
'--ignore_ssl', action='store_true', default=False,
help='Tha agent will ignores SSL errors',
)
parser.add_argument(
'--no-piggyback', action='store_true', default=False,
help='By default the agent generates the output as piggyback data for each\n'
'Samrthome device. If you want to attach all your Smarthome devices directly\n'
' to your Fritz!Box use this option.',
)
parser.add_argument(
'--password', nargs='?',
help='The password to logon the Fritz!Box',
)
parser.add_argument(
'--username', nargs='?',
help='The username to logon to the Fritz!Box',
)
parser.add_argument(
'--port', nargs='?', type=int, default=443,
help='The TCP port on witch to access the Fritz!Box',
)
parser.add_argument(
'--prefix', nargs='?',
help='The prefix is used to group all the Smarthome devices from one Fritz!Box in CMK.'
)
parser.add_argument(
'--protocol', nargs='?', choices=['http', 'https'], default='https',
help='The protocol used to access the Fritz!Box',
)
parser.add_argument(
'--testing', action='store_true', default=False,
help='Development usage only (might be ignored)'
)
args = parser.parse_args()
return args
def check_fritzbox_smarthome(args):
base_address = '%s://%s:%d' % (args.protocol, args.host, args.port)
ctx = ssl.create_default_context()
if args.ignore_ssl:
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
# CALL /login_sid.lua
# and grab challenge
response = urlopen(base_address + '/login_sid.lua', context=ctx)
if args.password:
xml_login = ET.fromstring(response.read())
challenge = xml_login.find('Challenge').text
blocktime = int(xml_login.find('BlockTime').text)
if blocktime > 0:
sys.stdout.write('<<<fritzbox_smarthome:sep(0)>>>')
sys.stdout.write(json.dumps({'block_time': blocktime}))
exit()
# create challenge_response (hash with md5: '<challenge>-<password>')
# TODO: check if challenge is PBKDF2 (startswith $2)
digest = hashlib.md5()
digest.update(challenge.encode('utf-16le'))
digest.update('-'.encode('utf-16le'))
digest.update(args.password.encode('utf-16le'))
challenge_response = challenge + '-' + digest.hexdigest()
# CALL /login_sid.lua?username=<username>&response=<challenge_response>
# and grab sessionid
if args.username:
response = urlopen(
base_address + '/login_sid.lua?username=%s&response=%s' % (args.username, challenge_response),
context=ctx)
else:
response = urlopen(base_address + '/login_sid.lua?response=%s' % challenge_response, context=ctx)
xml_login_solve = ET.fromstring(response.read())
sessionid = xml_login_solve.find('SID').text
blocktime = int(xml_login_solve.find('BlockTime').text)
if blocktime > 0:
sys.stdout.write('<<<fritzbox_smarthome:sep(0)>>>')
sys.stdout.write(json.dumps({'block_time': blocktime}))
exit()
if args.password and sessionid == '0000000000000000':
raise Exception('Check credentials\n')
# Write section header
response = urlopen(
base_address + '/webservices/homeautoswitch.lua?switchcmd=getdevicelistinfos&sid=%s' % sessionid, context=ctx)
response_read = response.read()
if args.debug:
sys.stdout.write('Raw XML:\n')
sys.stdout.write(str(response_read))
sys.stdout.write('\n')
xml_devicelist = ET.fromstring(response_read)
devices = []
if args.testing:
__switch_01 = {
"identifier": "08761 0116372",
"id": "99",
"functionbitmask": "35712",
"fwversion": "04.26",
"manufacturer": "AVM",
"productname": "FRITZ!DECT 200",
"present": "1",
"txbusy": "0",
"name": "TV-living_room",
"switch": {
"state": "1",
"mode": "manuell",
"lock": "0",
"devicelock": "0"
},
"simpleonoff": {
"state": "1"
},
"powermeter": {
"voltage": "235814",
"power": "4220",
"energy": "145427"
},
"temperature": {
"celsius": "190",
"offset": "0"
}
}
__repeater_01 = {
"identifier": "11657 0057950",
"id": "98",
"functionbitmask": "1024",
"fwversion": "04.16",
"manufacturer": "AVM",
"productname": "FRITZ!DECT Repeater 100",
"present": "0",
"txbusy": "0",
"name": "FRITZ!DECT Rep 100 #1"
}
__repeater_02 = {
"identifier": "11657 0170905",
"id": "97",
"functionbitmask": "1280",
"fwversion": "04.25",
"manufacturer": "AVM",
"productname": "FRITZ!DECT Repeater 100",
"present": "1",
"txbusy": "0",
"name": "FRITZ!DECT Repeater 100 #2",
"temperature": {
"celsius": "245",
"offset": "0"
}
}
__thermostat_01 = {
"identifier": "13979 0878454",
"id": "96",
"functionbitmask": "320",
"fwversion": "05.16",
"manufacturer": "AVM",
"productname": "Comet DECT",
"present": "1",
"name": "Temp02",
"temperature": {
"celsius": "210",
"offset": "-10"
},
"hkr": {
"tist": "42",
"tsoll": "32",
"absenk": "32",
"komfort": "38",
"lock": "1",
"devicelock": "1",
"errorcode": "0",
"batterylow": "0",
"nextchange": {
"endperiod": "1704888000",
"tchange": "32"
}
}
}
energy = int(__switch_01["powermeter"]["energy"])
power = int(__switch_01["powermeter"]["power"])
start_time = 1703883617
energy_up = int(time.time() - start_time) / 3600 * (int(power) / 1000)
__switch_01["powermeter"]["energy"] = str(int(energy + energy_up))
devices.append(__switch_01)
devices.append(__repeater_01)
devices.append(__repeater_02)
devices.append(__thermostat_01)
for xml_device in xml_devicelist.findall('device'):
devices.append(parse_xml_to_json(xml_device))
if args.no_piggyback:
sys.stdout.write('<<<fritzbox_smarthome:sep(0)>>>\n')
# if len(devices) == 1:
# sys.stdout.write(json.dumps(devices[0])) # single device
# else:
sys.stdout.write(json.dumps(devices))
sys.stdout.write('\n')
else:
for json_device in devices:
name = json_device["name"].replace(' ', '_')
name = re_sub(r'[^.\-_a-zA-Z0-9]', '', name)
if args.prefix:
name = f'{args.prefix}-{name}'
sys.stdout.write(f'<<<<{name}>>>>\n')
sys.stdout.write('<<<fritzbox_smarthome:sep(0)>>>\n')
sys.stdout.write(json.dumps(json_device))
sys.stdout.write('\n')
def main():
replace_passwords()
args = parse_args()
try:
check_fritzbox_smarthome(args)
except:
if args.debug:
raise
sys.stderr.write('fritzbox_smarthome\n %s\n' % traceback.format_exc())
sys.exit(2)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment