From ae373370aa19231979f4e8e65f555dd51df98c72 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Fri, 19 Feb 2021 12:07:09 +0100 Subject: [PATCH 1/7] Adding general MQTT utility for stabilizer --- stabilizer.py | 147 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 147 insertions(+) create mode 100644 stabilizer.py diff --git a/stabilizer.py b/stabilizer.py new file mode 100644 index 0000000..6b83bee --- /dev/null +++ b/stabilizer.py @@ -0,0 +1,147 @@ +#!/usr/bin/python +""" +Author: Vertigo Designs, Ryan Summers + +Description: Provides an API for controlling Booster NGFW over MQTT. +""" +import argparse +import asyncio +import json + +from gmqtt import Client as MqttClient + + +def parse_value(value): + """ Parse a command-line value into the most appropriate associated python datatype. """ + if value.isnumeric(): + return int(value) + try: + return float(value) + except ValueError: + return value + + +class MiniconfApi: + """ An asynchronous API for controlling Miniconf devices using the MQTT control interface. """ + + @classmethod + async def create(cls, identifier, broker): + """ Create a connection to MQTT for communication with booster. """ + client = MqttClient(client_id='') + await client.connect(broker) + return cls(client, identifier) + + + def __init__(self, client, identifier): + """ Consructor. + + Args: + client: A connected MQTT5 client. + identifier: The ID of the device to control. + """ + self.response_topic = f'{identifier}/feedback' + self.client = client + self.identifier = identifier + self.command_complete = asyncio.Event() + self.client.on_message = self._handle_response + self.response = None + + self.client.subscribe(self.response_topic) + + + def _handle_response(self, client, topic, payload, *_args, **_kwargs): + """ Callback function for when messages are received over MQTT. + + Args: + client: The MQTT client. + topic: The topic that the message was received on. + payload: The payload of the message. + """ + if topic != self.response_topic: + raise Exception(f'Unknown topic: {topic}') + + # Indicate a response was received. + self.response = payload.decode('ascii') + self.command_complete.set() + + + async def _command(self, topic, message): + """ Send a command to a booster control topic. + + Args: + topic: The topic to send the message to. + message: The message to send to the provided topic. + + Returns: + The received response to the command. + """ + self.command_complete.clear() + self.client.publish(topic, payload=message, qos=0, retain=False, + response_topic=self.response_topic) + await self.command_complete.wait() + + response = self.response + self.response = None + + return response + + + async def set_setting(self, setting, message): + """ Change the provided setting with the provided data. """ + return await self._command(f'{self.identifier}/settings/{setting}', message) + + + async def commit(self): + """ Commit staged settings to become active. """ + return await self._command(f'{self.identifier}/commit', 'commit') + + +async def configure_settings(args): + """ Configure an RF channel. """ + + # Establish a communication interface with Booster. + interface = await MiniconfApi.create(args.stabilizer, args.broker) + + request = None + + # In the exceptional case that this is a terminal value, there is no key available and only a + # single value. + if len(args.values) == 1 and '=' not in args.values[0]: + request = parse_value(args.values[0]) + else: + # Convert all of the values into a key-value list. + request = dict() + for pair in args.values: + key, value = pair.split('=') + request[str(key)] = parse_value(value) + response = json.dumps(request) + + # TODO: Should we escape quotes in the dumped request? + response = await interface.set_setting(args.setting, request) + print(f'+ {response}') + + if args.commit: + response = await interface.commit() + print(f'+ {response}') + + +def main(): + """ Main program entry point. """ + parser = argparse.ArgumentParser( + formatter_class=argparse.RawTextHelpFormatter, + description='Modify booster RF channel configuration') + parser.add_argument('--stabilizer', type=str, default='stabilizer', + help='The identifier of the stabilizer to configure') + parser.add_argument('--setting', required=True, type=str, help='The setting path to configure') + parser.add_argument('--broker', default='10.34.16.1', type=str, help='The MQTT broker address') + parser.add_argument('values', nargs='+', type=str, + help='The value of settings. key=value list or a single value is accepted.') + parser.add_argument('--commit', action='store_true', + help='Specified true to commit after updating settings.') + + loop = asyncio.get_event_loop() + loop.run_until_complete(configure_settings(parser.parse_args())) + + +if __name__ == '__main__': + main() From 86a3f840e98c0e52e53147fa3fa9f38ab0b94c07 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Fri, 19 Feb 2021 12:09:35 +0100 Subject: [PATCH 2/7] Fixing comments --- stabilizer.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/stabilizer.py b/stabilizer.py index 6b83bee..5ecee84 100644 --- a/stabilizer.py +++ b/stabilizer.py @@ -2,7 +2,7 @@ """ Author: Vertigo Designs, Ryan Summers -Description: Provides an API for controlling Booster NGFW over MQTT. +Description: Provides an API for controlling Stabilizer over Miniconf (MQTT). """ import argparse import asyncio @@ -26,7 +26,7 @@ class MiniconfApi: @classmethod async def create(cls, identifier, broker): - """ Create a connection to MQTT for communication with booster. """ + """ Create a connection to MQTT for communication with the device. """ client = MqttClient(client_id='') await client.connect(broker) return cls(client, identifier) @@ -66,7 +66,7 @@ class MiniconfApi: async def _command(self, topic, message): - """ Send a command to a booster control topic. + """ Send a command to a topic. Args: topic: The topic to send the message to. @@ -99,7 +99,7 @@ class MiniconfApi: async def configure_settings(args): """ Configure an RF channel. """ - # Establish a communication interface with Booster. + # Establish a communication interface with stabilizer. interface = await MiniconfApi.create(args.stabilizer, args.broker) request = None @@ -116,7 +116,6 @@ async def configure_settings(args): request[str(key)] = parse_value(value) response = json.dumps(request) - # TODO: Should we escape quotes in the dumped request? response = await interface.set_setting(args.setting, request) print(f'+ {response}') @@ -127,9 +126,7 @@ async def configure_settings(args): def main(): """ Main program entry point. """ - parser = argparse.ArgumentParser( - formatter_class=argparse.RawTextHelpFormatter, - description='Modify booster RF channel configuration') + parser = argparse.ArgumentParser(description='Stabilizer settings modification utility') parser.add_argument('--stabilizer', type=str, default='stabilizer', help='The identifier of the stabilizer to configure') parser.add_argument('--setting', required=True, type=str, help='The setting path to configure') From f6207d86fba57d15dd0e76d0dfb8ceff72313b90 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Fri, 19 Feb 2021 12:15:02 +0100 Subject: [PATCH 3/7] Adding support for array parsing --- stabilizer.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/stabilizer.py b/stabilizer.py index 5ecee84..a78452f 100644 --- a/stabilizer.py +++ b/stabilizer.py @@ -13,6 +13,13 @@ from gmqtt import Client as MqttClient def parse_value(value): """ Parse a command-line value into the most appropriate associated python datatype. """ + # If the value is an array, construct it as such and recurse for individual elements. + if value.startswith('[') and value.endswith(']'): + all_values = [] + for data in value[1:][:-1].split(','): + all_values.append(parse_value(data)) + return all_values + if value.isnumeric(): return int(value) try: From 971c53d082f522d3575c8158fe3b856bf0922580 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Fri, 19 Feb 2021 12:29:55 +0100 Subject: [PATCH 4/7] Fixing JSON dumps --- stabilizer.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/stabilizer.py b/stabilizer.py index a78452f..59c3948 100644 --- a/stabilizer.py +++ b/stabilizer.py @@ -121,9 +121,8 @@ async def configure_settings(args): for pair in args.values: key, value = pair.split('=') request[str(key)] = parse_value(value) - response = json.dumps(request) - response = await interface.set_setting(args.setting, request) + response = await interface.set_setting(args.setting, json.dumps(request)) print(f'+ {response}') if args.commit: From da4d6a749089ec6fc162e08ff12229c7fe7d7b0f Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Fri, 19 Feb 2021 13:04:10 +0100 Subject: [PATCH 5/7] Adding logging, verbosity, updating json loading --- stabilizer.py | 42 ++++++++++++++++++++---------------------- 1 file changed, 20 insertions(+), 22 deletions(-) diff --git a/stabilizer.py b/stabilizer.py index 59c3948..904e2fa 100644 --- a/stabilizer.py +++ b/stabilizer.py @@ -7,27 +7,11 @@ Description: Provides an API for controlling Stabilizer over Miniconf (MQTT). import argparse import asyncio import json +import logging from gmqtt import Client as MqttClient -def parse_value(value): - """ Parse a command-line value into the most appropriate associated python datatype. """ - # If the value is an array, construct it as such and recurse for individual elements. - if value.startswith('[') and value.endswith(']'): - all_values = [] - for data in value[1:][:-1].split(','): - all_values.append(parse_value(data)) - return all_values - - if value.isnumeric(): - return int(value) - try: - return float(value) - except ValueError: - return value - - class MiniconfApi: """ An asynchronous API for controlling Miniconf devices using the MQTT control interface. """ @@ -52,6 +36,7 @@ class MiniconfApi: self.command_complete = asyncio.Event() self.client.on_message = self._handle_response self.response = None + self.logger = logging.getLogger('stabilizer.miniconf') self.client.subscribe(self.response_topic) @@ -83,6 +68,7 @@ class MiniconfApi: The received response to the command. """ self.command_complete.clear() + self.logger.debug('Sending "%s" to "%s"', message, topic) self.client.publish(topic, payload=message, qos=0, retain=False, response_topic=self.response_topic) await self.command_complete.wait() @@ -105,6 +91,7 @@ class MiniconfApi: async def configure_settings(args): """ Configure an RF channel. """ + logger = logging.getLogger('stabilizer') # Establish a communication interface with stabilizer. interface = await MiniconfApi.create(args.stabilizer, args.broker) @@ -114,20 +101,21 @@ async def configure_settings(args): # In the exceptional case that this is a terminal value, there is no key available and only a # single value. if len(args.values) == 1 and '=' not in args.values[0]: - request = parse_value(args.values[0]) + request = json.loads(args.values[0]) else: # Convert all of the values into a key-value list. request = dict() for pair in args.values: key, value = pair.split('=') - request[str(key)] = parse_value(value) + request[str(key)] = json.loads(value) + logger.debug('Parsed request: %s', request) response = await interface.set_setting(args.setting, json.dumps(request)) - print(f'+ {response}') + logger.info(response) if args.commit: response = await interface.commit() - print(f'+ {response}') + logger.info(response) def main(): @@ -141,9 +129,19 @@ def main(): help='The value of settings. key=value list or a single value is accepted.') parser.add_argument('--commit', action='store_true', help='Specified true to commit after updating settings.') + parser.add_argument('-v', '--verbose', action='store_true', help='Enable verbose logging') + + args = parser.parse_args() + + logger = logging.getLogger('stabilizer') + logger.setLevel(logging.INFO) + + if args.verbose: + logger.setLevel(logging.DEBUG) + logging.basicConfig(level=logging.DEBUG) loop = asyncio.get_event_loop() - loop.run_until_complete(configure_settings(parser.parse_args())) + loop.run_until_complete(configure_settings(args)) if __name__ == '__main__': From c943e8d136610134cd43338fe471d9b504c614e6 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Mon, 1 Mar 2021 13:32:53 +0100 Subject: [PATCH 6/7] Updating logging, removing commit support, adding support for multiple in-flight settings --- stabilizer.py | 70 ++++++++++++++++++++++----------------------------- 1 file changed, 30 insertions(+), 40 deletions(-) diff --git a/stabilizer.py b/stabilizer.py index 904e2fa..02444e0 100644 --- a/stabilizer.py +++ b/stabilizer.py @@ -30,65 +30,57 @@ class MiniconfApi: client: A connected MQTT5 client. identifier: The ID of the device to control. """ - self.response_topic = f'{identifier}/feedback' self.client = client self.identifier = identifier - self.command_complete = asyncio.Event() self.client.on_message = self._handle_response - self.response = None + self.inflight_settings = dict() self.logger = logging.getLogger('stabilizer.miniconf') - self.client.subscribe(self.response_topic) + self.client.subscribe(f'{identifier}/feedback/#') - def _handle_response(self, client, topic, payload, *_args, **_kwargs): + def _handle_response(self, _client, topic, payload, *_args, **_kwargs): """ Callback function for when messages are received over MQTT. Args: - client: The MQTT client. + _client: The MQTT client. topic: The topic that the message was received on. payload: The payload of the message. """ - if topic != self.response_topic: - raise Exception(f'Unknown topic: {topic}') + if topic not in self.inflight_settings: + self.logger.warning('Unknown response topic: %s', topic) + return - # Indicate a response was received. - self.response = payload.decode('ascii') - self.command_complete.set() + # Indicate a response was received for the provided topic. + self.inflight_settings[topic].set_result(payload.decode('ascii')) - async def _command(self, topic, message): - """ Send a command to a topic. + async def set_setting(self, setting, value): + """ Change the provided setting with the provided data. Args: - topic: The topic to send the message to. - message: The message to send to the provided topic. + setting: The setting to update + value: The value to configure the setting to (in JSON-serialized format). Returns: The received response to the command. """ - self.command_complete.clear() - self.logger.debug('Sending "%s" to "%s"', message, topic) - self.client.publish(topic, payload=message, qos=0, retain=False, - response_topic=self.response_topic) - await self.command_complete.wait() + setting_topic = f'{self.identifier}/settings/{setting}' + response_topic = f'{self.identifier}/feedback/{setting}' + assert response_topic not in self.inflight_settings, \ + 'Only one in-flight setting is supported' - response = self.response - self.response = None + self.logger.debug('Sending %s to "%s"', value, setting_topic) + self.inflight_settings[response_topic] = asyncio.get_running_loop().create_future() + self.client.publish(setting_topic, payload=value, qos=0, retain=False, + response_topic=response_topic) + + response = await self.inflight_settings[response_topic] + del self.inflight_settings[response_topic] return response - async def set_setting(self, setting, message): - """ Change the provided setting with the provided data. """ - return await self._command(f'{self.identifier}/settings/{setting}', message) - - - async def commit(self): - """ Commit staged settings to become active. """ - return await self._command(f'{self.identifier}/commit', 'commit') - - async def configure_settings(args): """ Configure an RF channel. """ logger = logging.getLogger('stabilizer') @@ -101,7 +93,10 @@ async def configure_settings(args): # In the exceptional case that this is a terminal value, there is no key available and only a # single value. if len(args.values) == 1 and '=' not in args.values[0]: - request = json.loads(args.values[0]) + if args.values[0][0].isalpha(): + request = args.values[0] + else: + request = json.loads(args.values[0]) else: # Convert all of the values into a key-value list. request = dict() @@ -113,10 +108,6 @@ async def configure_settings(args): response = await interface.set_setting(args.setting, json.dumps(request)) logger.info(response) - if args.commit: - response = await interface.commit() - logger.info(response) - def main(): """ Main program entry point. """ @@ -127,8 +118,6 @@ def main(): parser.add_argument('--broker', default='10.34.16.1', type=str, help='The MQTT broker address') parser.add_argument('values', nargs='+', type=str, help='The value of settings. key=value list or a single value is accepted.') - parser.add_argument('--commit', action='store_true', - help='Specified true to commit after updating settings.') parser.add_argument('-v', '--verbose', action='store_true', help='Enable verbose logging') args = parser.parse_args() @@ -138,7 +127,8 @@ def main(): if args.verbose: logger.setLevel(logging.DEBUG) - logging.basicConfig(level=logging.DEBUG) + + logging.basicConfig() loop = asyncio.get_event_loop() loop.run_until_complete(configure_settings(args)) From 9a5de507d12b75d354e98edc06b80be365ccad90 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Mon, 1 Mar 2021 13:47:44 +0100 Subject: [PATCH 7/7] Fixing log format, adding commit support --- stabilizer.py | 24 +++++++++++++++--------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/stabilizer.py b/stabilizer.py index 02444e0..973a7d1 100644 --- a/stabilizer.py +++ b/stabilizer.py @@ -55,20 +55,20 @@ class MiniconfApi: self.inflight_settings[topic].set_result(payload.decode('ascii')) - async def set_setting(self, setting, value): - """ Change the provided setting with the provided data. + async def command(self, path, value): + """ Write the provided data to the specified path. Args: - setting: The setting to update - value: The value to configure the setting to (in JSON-serialized format). + setting: The path to write the message to. + value: The value to write to the path. Returns: The received response to the command. """ - setting_topic = f'{self.identifier}/settings/{setting}' - response_topic = f'{self.identifier}/feedback/{setting}' + setting_topic = f'{self.identifier}/{path}' + response_topic = f'{self.identifier}/feedback/{path}' assert response_topic not in self.inflight_settings, \ - 'Only one in-flight setting is supported' + 'Only one in-flight message per topic is supported' self.logger.debug('Sending %s to "%s"', value, setting_topic) self.inflight_settings[response_topic] = asyncio.get_running_loop().create_future() @@ -105,9 +105,13 @@ async def configure_settings(args): request[str(key)] = json.loads(value) logger.debug('Parsed request: %s', request) - response = await interface.set_setting(args.setting, json.dumps(request)) + response = await interface.command(f'settings/{args.setting}', json.dumps(request)) logger.info(response) + if args.commit: + response = await interface.command('commit', 'commit') + logger.info(response) + def main(): """ Main program entry point. """ @@ -118,6 +122,8 @@ def main(): parser.add_argument('--broker', default='10.34.16.1', type=str, help='The MQTT broker address') parser.add_argument('values', nargs='+', type=str, help='The value of settings. key=value list or a single value is accepted.') + parser.add_argument('--commit', action='store_true', + help='Specified true to commit after updating settings.') parser.add_argument('-v', '--verbose', action='store_true', help='Enable verbose logging') args = parser.parse_args() @@ -128,7 +134,7 @@ def main(): if args.verbose: logger.setLevel(logging.DEBUG) - logging.basicConfig() + logging.basicConfig(format='%(asctime)s [%(levelname)s] %(message)s') loop = asyncio.get_event_loop() loop.run_until_complete(configure_settings(args))