diff --git a/platypush/backend/mqtt.py b/platypush/backend/mqtt.py index 7527d42b..af57ba48 100644 --- a/platypush/backend/mqtt.py +++ b/platypush/backend/mqtt.py @@ -1,10 +1,12 @@ import json +import os import threading import paho.mqtt.client as mqtt import paho.mqtt.publish as publisher from platypush.backend import Backend +from platypush.context import get_plugin from platypush.message import Message from platypush.message.request import Request @@ -19,7 +21,10 @@ class MqttBackend(Backend): * **paho-mqtt** (``pip install paho-mqtt``) """ - def __init__(self, host, port=1883, topic='platypush_bus_mq', *args, **kwargs): + def __init__(self, host, port=1883, topic='platypush_bus_mq', tls_cafile=None, + tls_certfile=None, tls_keyfile=None, + tls_version=None, tls_ciphers=None, username=None, + password=None, *args, **kwargs): """ :param host: MQTT broker host :type host: str @@ -29,6 +34,27 @@ class MqttBackend(Backend): :param topic: Topic to read messages from (default: ``platypush_bus_mq/``) :type topic: str + + :param tls_cafile: If TLS/SSL is enabled on the MQTT server and the certificate requires a certificate authority to authenticate it, `ssl_cafile` will point to the provided ca.crt file (default: None) + :type tls_cafile: str + + :param tls_certfile: If TLS/SSL is enabled on the MQTT server and a client certificate it required, specify it here (default: None) + :type tls_certfile: str + + :param tls_keyfile: If TLS/SSL is enabled on the MQTT server and a client certificate key it required, specify it here (default: None) + :type tls_keyfile: str + + :param tls_version: If TLS/SSL is enabled on the MQTT server and it requires a certain TLS version, specify it here (default: None) + :type tls_version: str + + :param tls_ciphers: If TLS/SSL is enabled on the MQTT server and an explicit list of supported ciphers is required, specify it here (default: None) + :type tls_ciphers: str + + :param username: Specify it if the MQTT server requires authentication (default: None) + :type username: str + + :param password: Specify it if the MQTT server requires authentication (default: None) + :type password: str """ super().__init__(*args, **kwargs) @@ -36,15 +62,31 @@ class MqttBackend(Backend): self.host = host self.port = port self.topic = '{}/{}'.format(topic, self.device_id) + self.username = username + self.password = password + + self.tls_cafile = os.path.abspath(os.path.expanduser(tls_cafile)) \ + if tls_cafile else None + + self.tls_certfile = os.path.abspath(os.path.expanduser(tls_certfile)) \ + if tls_certfile else None + + self.tls_keyfile = os.path.abspath(os.path.expanduser(tls_keyfile)) \ + if tls_keyfile else None + + self.tls_version = tls_version + self.tls_ciphers = tls_ciphers def send_message(self, msg): - try: - msg = Message.build(json.loads(msg)) - except: - pass - - publisher.single(self.topic, str(msg), hostname=self.host, port=self.port) + client = get_plugin('mqtt') + client.send_message(topic=self.topic, msg=msg, host=self.host, + port=self.port, username=self.username, + password=self.password, tls_cafile=self.tls_cafile, + tls_certfile=self.tls_certfile, + tls_keyfile=self.tls_keyfile, + tls_version=self.tls_version, + tls_ciphers=self.tls_ciphers) def run(self): def on_connect(client, userdata, flags, rc): @@ -58,8 +100,14 @@ class MqttBackend(Backend): self.logger.info('Processing response on the MQTT topic {}: {}'. format(response_topic, response)) - publisher.single(response_topic, str(response), - hostname=self.host, port=self.port) + client = get_plugin('mqtt') + client.send_message(topic=self.topic, msg=msg, host=self.host, + port=self.port, username=self.username, + password=self.password, tls_cafile=self.tls_cafile, + tls_certfile=self.tls_certfile, + tls_keyfile=self.tls_keyfile, + tls_version=self.tls_version, + tls_ciphers=self.tls_ciphers) msg = msg.payload.decode('utf-8') try: msg = Message.build(json.loads(msg)) @@ -75,6 +123,15 @@ class MqttBackend(Backend): client = mqtt.Client() client.on_connect = on_connect client.on_message = on_message + + if self.username and self.password: + client.username_pw_set(self.username, self.password) + + if self.tls_cafile: + client.tls_set(ca_certs=self.tls_cafile, certfile=self.tls_certfile, + keyfile=self.tls_keyfile, tls_version=self.tls_version, + ciphers=self.tls_ciphers) + client.connect(self.host, self.port, 60) self.logger.info('Initialized MQTT backend on host {}:{}, topic {}'. format(self.host, self.port, self.topic)) diff --git a/platypush/plugins/mqtt.py b/platypush/plugins/mqtt.py index 9a3a6ade..75f8b217 100644 --- a/platypush/plugins/mqtt.py +++ b/platypush/plugins/mqtt.py @@ -15,7 +15,10 @@ class MqttPlugin(Plugin): super().__init__(*args, **kwargs) @action - def send_message(self, topic, msg, host, port=1883, *args, **kwargs): + def send_message(self, topic, msg, host, port=1883, tls_cafile=None, + tls_certfile=None, tls_keyfile=None, + tls_version=None, tls_ciphers=None, username=None, + password=None, *args, **kwargs): """ Sends a message to a topic/channel. @@ -29,15 +32,58 @@ class MqttPlugin(Plugin): :param port: MQTT broker port (default: 1883) :type port: int + + :param tls_cafile: If TLS/SSL is enabled on the MQTT server and the certificate requires a certificate authority to authenticate it, `ssl_cafile` will point to the provided ca.crt file (default: None) + :type tls_cafile: str + + :param tls_certfile: If TLS/SSL is enabled on the MQTT server and a client certificate it required, specify it here (default: None) + :type tls_certfile: str + + :param tls_keyfile: If TLS/SSL is enabled on the MQTT server and a client certificate key it required, specify it here (default: None) + :type tls_keyfile: str + + :param tls_version: If TLS/SSL is enabled on the MQTT server and it requires a certain TLS version, specify it here (default: None) + :type tls_version: str + + :param tls_ciphers: If TLS/SSL is enabled on the MQTT server and an explicit list of supported ciphers is required, specify it here (default: None) + :type tls_ciphers: str + + :param username: Specify it if the MQTT server requires authentication (default: None) + :type username: str + + :param password: Specify it if the MQTT server requires authentication (default: None) + :type password: str """ + publisher_args = { + 'hostname': host, + 'port': port, + } + + if username and password: + publisher_args['auth'] = { + 'username': username, + 'password': password, + } + + if tls_cafile: + publisher_args['tls'] = { 'ca_certs': tls_cafile } + if tls_certfile: + publishers_args['tls']['certfile'] = tls_certfile + if tls_keyfile: + publishers_args['tls']['keyfile'] = tls_keyfile + if tls_version: + publishers_args['tls']['tls_version'] = tls_version + if tls_ciphers: + publishers_args['tls']['ciphers'] = tls_ciphers + try: msg = json.dumps(msg) except: pass try: msg = Message.build(json.loads(msg)) except: pass - publisher.single(topic, str(msg), hostname=host, port=port) + publisher.single(topic, str(msg), **publisher_args) # vim:sw=4:ts=4:et: