diff --git a/lib/python3/cmk/special_agents/agent_fritzbox_smarthome.py b/lib/python3/cmk/special_agents/agent_fritzbox_smarthome.py deleted file mode 100644 index b6696503e4188220a6e1d5ce6a2e53ca2ee23581..0000000000000000000000000000000000000000 --- a/lib/python3/cmk/special_agents/agent_fritzbox_smarthome.py +++ /dev/null @@ -1,284 +0,0 @@ -#!/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)