jakezp/mqttwarn repository overview

⁠mqttwarn

To warn, alert, or notify.

This program subscribes to any number of MQTT topics (which may include wildcards) and publishes received payloads to one or more notification services, including support for notifying more than one distinct service for the same message.

Notifications are transmitted to the appropriate service via plugins. We provide plugins for the list of services below⁠, and you can easily add your own.

I've written an introductory post, explaining what mqttwarn can be used for⁠. For example, you may wish to notify via e-mail and to Pushover of an alarm published as text to the MQTT topic home/monitoring/+.

• Getting started⁠

  • Requirements⁠
  • Installation⁠
  • Configuration⁠

• Supported Notification Services⁠

  • Configuration of service plugins⁠
  • Creating Custom Service Plugins⁠

• Outbound messages⁠

  • Message forwarding⁠
  • Transforming inbound JSON⁠
  • Custom functions⁠
  • Templates⁠

• Periodic tasks⁠

• Running with Docker⁠

  • Run the Image⁠
  • Build the image⁠

• Examples⁠

  • Low battery notifications⁠
  • Producing JSON⁠

• Notes⁠

• Press⁠

⁠Getting started

⁠Requirements

You'll need at least the following components:

• Python 2.x (tested with 2.6 and 2.7)
• An MQTT broker (e.g. Mosquitto⁠)
• The Paho Python module: pip install paho-mqtt

⁠Installation

1. Clone this repository into a fresh directory.
2. Copy mqttwarn.ini.sample to mqttwarn.ini and edit to your taste
3. Install the prerequisite Python modules for the services you want to use
4. Launch mqttwarn.py

I recommend you use Supervisor⁠ for running this.

Alternatively, a systemd-based installation using a Python virtualenv might be handy, see systemd unit configuration file for mqttwarn⁠ for step-by-step instructions about doing this.

⁠Configuration

I recommend you start off with the following simple configuration which will log messages received on the MQTT topic test/+ to a file. Create the following configuration file:

[defaults]
hostname  = 'localhost'
port      = 1883

; name the service providers you will be using.
launch	 = file, log

[config:file]
append_newline = True
targets = {
    'mylog'     : ['/tmp/mqtt.log']
    }

[config:log]
targets = {
    'info'   : [ 'info' ]
  }

[test/+]
targets = file:mylog, log:info

Note: the closing brace } of the targets dict must be indented; this is an artifact of ConfigParser.

Launch mqttwarn.py and keep an eye on its log file (mqttwarn.log by default). Publish two messages to the subscribed topic, using

mosquitto_pub -t test/1 -m "Hello"
mosquitto_pub -t test/name -m '{ "name" : "Jane" }'

and our output file /tmp/mqtt.log should contain the payload of both messages:

Hello
{ "name" : "Jane" }

Both payloads where copied verbatim to the target.

Stop mqttwarn, and add the following line to the [test/+] section:

format  = -->{name}<--

What we are configuring mqttwarn to do here, is to try and decode the incoming JSON payload and format the output in such a way as that the JSON name element is copied to the output (surrounded with a bit of sugar to illustrate the fact that we can output whatever text we want).

If you repeat the publish of the second message, you should see the following in your output file /tmp/mqtt.log:

-->Jane<--

⁠The [defaults] section

Most of the options in the configuration file have sensible defaults, and/or ought to be self-explanatory:

[defaults]
hostname     = 'localhost'         ; default
port         = 1883
username     = None
password     = None
clientid     = 'mqttwarn'
lwt          = 'clients/mqttwarn'  ; lwt payload is '0' or '1'
skipretained = True
cleansession = False

; logging
logformat = '%(asctime)-15s %(levelname)-5s [%(module)s] %(message)s'
logfile   = 'mqttwarn.log'

; one of: CRITICAL, DEBUG, ERROR, INFO, WARN
loglevel     = DEBUG

; path to file containing self-defined functions for formatmap, alldata, and datamap
functions = 'myfuncs.py'

; name the service providers you will be using.
launch   = file, log, osxnotify, mysql, smtp

; the directory to which we should cd after startup (default: ".")
; the cd is performed before loading service plugins, so it should
; contain a `services/' directory with the required service plugins.
directory = /tmp/


; optional: TLS parameters. (Don't forget to set the port number for
; TLS (probably 8883).
; You will need to set at least `ca_certs' if you want TLS support
; ca_certs: path to the Certificate Authority certificate file (concatenated
;           PEM file)
; tls_version: currently either 'tlsv1' or 'sslv3'
; tls_insecure: True or False (False is default): Do or do not verify
;               broker's certificate CN
; certfile: path to PEM encode client certificate file
; keyfile: path to PEM encode client private key file
ca_certs = '/path/to/ca-certs.pem'
certfile = '/path/to/client.crt'
keyfile = '/path/to/client.key'
tls_version = 'tlsv1'
tls_insecure = False

⁠functions

The functions option specifies the path to a Python file containing functions you use in formatting or filtering data (see below). The .py extension to the path name you configure here must be specified.

⁠launch

In the launch option you specify which services (of those available in the services/ directory of mqttwarn or using the module option, see the following paragraphs) you want to be able to use in target definitions.

⁠The [config:xxx] sections

Sections called [config:xxx] configure settings for a service xxx. Each of these sections has a mandatory option called targets, which is a dictionary of target names, each pointing to an array of "addresses". Address formats depend on the particular service.

A service section may have an option called module, which refers to the name of the actual service module to use. A service called filetruncate - and referenced as such in the launch option - may have module = file, in which case the service works like a regular file service, with its own distinct set of service options. It is thus possible to have several different service configurations for the same underlying service, with different configurations, e.g. one for files that should have notifications appended, and one for files that should get truncated before writes.

As an example for module consider this INI file in which we want two services of type log. We actually launch an xxxlog (which doesn't physically exist), but due to the module=log setting in its configuration it is instantiated:

[defaults]
hostname  = 'localhost'  ; default
port      = 1883

launch	 = log, xxxlog

[config:log]
targets = {
    'debug'  : [ 'debug' ],
  }

[config:xxxlog]
# Note how the xxxlog is instantiated from log and both must be launched
module = log
targets = {
    'debug'  : [ 'debug' ],
  }


[topic/1]
targets = log:debug, xxxlog:debug

⁠The [failover] section

There is a special section (optional) for defining a target (or targets) for internal error conditions. Currently there is only one error handled by this logic, broker disconnection.

This allows you to setup a target for receiving errors generated within mqttwarn. The message is handled like any other with an error code passed as the topic and the error details as the message. You can use formatting and transformations as well as filters, just like any other topic.

Below is an example which will log any failover events to an error log, and display them on all XBMC targets:

[failover]
targets  = log:error, xbmc
title    = mqttwarn

⁠The [__topic__] sections

All sections not called [defaults] or [config:xxx] are treated as MQTT topics to subscribe to. mqttwarn handles each message received on this subscription by handing it off to one or more service targets.

Section names must be unique and must specify the name of the topic to be processed. If the section block does not have a topic option, then the section name will be used.

Consider the following example:

[icinga/+/+]
targets = log:info, file:f01, mysql:nagios

[my/special]
targets = mysql:m1, log:info

[my/other/special]
topic = another/topic
targets = log:debug

MQTT messages received at icinga/+/+ will be directed to the three specified targets, whereas messages received at my/special will be stored in a mysql target and will be logged at level "INFO". Messages received at another/topic (not at my/other/special) will be logged at level "DEBUG".

When a message is received at a topic with more than one matching section it will be directed to the targets in all matching sections. For consistency, it's a good practice to explicitly provide topic options to all such sections.

Targets can be also defined as a dictionary containing the pairs of topic and targets. In that case message matching the section can be dispatched in more flexible ways to selected targets. Consider the following example:

[#]
targets = {
    '/#': 'file:0',
    '/test/#': 'file:1',
    '/test/out/#': 'file:2',
    '/test/out/+': 'file:3',
    '/test/out/+/+': 'file:4',
    '/test/out/+/state': 'file:5',
    '/test/out/FL_power_consumption/state': [ 'file:6', 'file:7' ],
    '/test/out/BR_ambient_power_sensor/state': 'file:8',
  }

With the message dispatching configuration the message is dispatched to the targets matching the most specific topic. If the message is received at /test/out/FL_power_consumption/state it will be directed to file:6 and file:7 targets only. Message received at /test/out/AR_lamp/state will be directed to file:5, but received at /test/out/AR_lamp/command will go to file:4. The dispatcher mechanism is always trying to find the most specific match. It allows to define the wide topic with default targets while some more specific topic can be handled differently. It gives additional flexibility in a message routing.

Each of these sections has a number of optional (O) or mandatory (M) options:

⁠Supported Notification Services

mqttwarn supports a number of services (listed alphabetically below):

• alexa-notify-me⁠
• amqp⁠
• apns⁠
• asterisk⁠
• autoremote⁠
• carbon⁠
• celery⁠
• dbus⁠
• dnsupdate⁠
• emoncms⁠
• execute⁠
• facebook messenger⁠
• file⁠
• freeswitch⁠
• gss⁠
• gss2⁠
• hangbot⁠
• hipchat⁠
• http⁠
• icinga2⁠
• ifttt⁠
• influxdb⁠
• instapush⁠
• ionic⁠
• iothub⁠
• irccat⁠
• linuxnotify⁠
• log⁠
• mastodon (see tootpaste⁠)
• mattermost⁠
• mqtt⁠
• mqttpub⁠
• mysql⁠
• mysql_dynamic⁠
• mysql_remap⁠
• mythtv⁠
• nntp⁠
• nsca⁠
• osxnotify⁠
• osxsay⁠
• pastebinpub⁠
• pipe⁠
• postgres⁠
• prowl⁠
• pushalot⁠
• pushbullet⁠
• pushover⁠
• pushsafer⁠
• redispub⁠
• rrdtool⁠
• serial⁠
• slack⁠
• sqlite⁠
• sqlite_json2cols⁠
• sqlite_timestamp⁠
• smtp⁠
• ssh⁠
• syslog⁠
• telegram⁠
• thingspeak⁠
• tootpaste⁠
• twilio⁠
• twitter⁠
• websocket⁠
• xbmc⁠
• xmpp⁠
• xively⁠
• zabbix⁠

⁠Configuration of service plugins

Service plugins are configured in the main mqttwarn.ini file. Each service has a mandatory section named [config:xxx], where xxx is the name of the service. This section may have some settings which are required for a particular service, and all services have an rarely used option called module (see The config:xxx sections⁠) and one mandatory option called targets. This defines individual "service points" for a particular service, e.g. different paths for the file service, distinct database tables for mysql, etc.

We term the array for each target an "address list" for the particular service. These may be path names (in the case of the file service), topic names (for outgoing mqtt publishes), hostname/port number combinations for xbmc, etc.

⁠alexa-notify-me

The alexa-notify-me service implements a gateway to make Alexa notifications using the Notify-Me voice app. http://www.thomptronics.com/notify-me⁠

[config:alexa-notify-me]
targets = {
	'account1' : [ 'Access Code' ]
  }

[alexa/notify]
targets = alexa-notify-me:account1

The access code is emailed to the user upon setup of Notify-Me

⁠amqp

The amqp service basically implements an MQTT to AMQP gateway which is a little bit overkill as, say, RabbitMQ already has a pretty versatile MQTT plugin. The that as it may, the configuration is as follows:

[config:amqp]
uri     =  'amqp://user:password@localhost:5672/'
    'test01'     : [ 'name_of_exchange',    'routing_key' ],
    }

The exchange specified in the target configuration must exist prior to using this target.

Requires: Puka⁠ (pip install puka)

⁠apns

The apns service interacts with the Apple Push Notification Service (APNS) and is a bit special (and one of mqttwarn's more complex services) in as much as it requires an X.509 certificate and a key which are typically available to developers only.

The following discussion assumes one of these payloads published via MQTT:

{"alert": "Vehicle moved" }

{"alert": "Vehicle moved", "custom" : { "tid": "C2" }}

In both cases, the message which will be displayed in the notification of the iOS device is "Vehicle moved". The second example depends on the app which receives the notification. This custom data is per/app. This example app uses the custom data to show a button:

This is the configuration we'll discuss.

[defaults]
hostname  = 'localhost'
port      = 1883
functions = 'myfuncs'

launch	 = apns

[config:apns]
targets = {
                 # path to cert in PEM format   # key in PEM format
    'prod'     : ['/path/to/prod.crt',          '/path/to/prod.key'],
    }

[test/token/+]
targets = apns
alldata = apnsdata()
format  = {alert}

Certificate and Key files are in PEM format, and the key file must not be password-protected. (The PKCS#12 file you get as a developer can be extracted thusly:

openssl pkcs12 -in apns-CTRL.p12 -nocerts -nodes | openssl rsa > prod.key
openssl pkcs12 -in apns-CTRL.p12 -clcerts -nokeys  > xxxx

then copy/paste from xxxx the sandbox or production certificate into prod.crt.)

The myfuncs function apnsdata() extracts the last part of the topic into apns_token, the hex token for the target device, which is required within the apns service.

def apnsdata(topic, data, srv=None):
    return dict(apns_token = topic.split('/')[-1])

A publish to topic test/token/380757b117f15a46dff2bd0be1d57929c34124dacb28d346dedb14d3464325e5 would thus emit the APNS notification to the specified device.

Requires PyAPNs⁠

⁠autoremote

The autoremote service forwards messages from desired topics to autoremote clients.

[config:autoremote]
targets = {
	'conv2' : [ 'ApiKey', 'Password', 'Target', 'Group', 'TTL' ]
  }

[autoremote/user]
targets = autoremote:conv2

Any messages published to autoremote/user would be sent the autoremote client designated to the ApiKey provided. The "sender" variable of autoremote is equal to the topic address.

https://joaoapps.com/autoremote/⁠

⁠carbon

The carbon service sends a metric to a Carbon-enabled server over TCP.

[config:carbon]
targets = {
        'c1' : [ '172.16.153.110', 2003 ],
  }

[c/#]
targets = carbon:c1

In this configuration, all messages published to c/# would be forwarded to the Carbon server at the specified IP address and TCP port number.

The topic name is translated into a Carbon metric name by replacing slashes by periods. A timestamp is appended to the message automatically.

For example, publishing this:

mosquitto_pub -t c/temp/arduino -m 12

would result in the value 12 being used as the value for the Carbon metric c.temp.arduino. The published payload may contain up to three white-space-separated parts.

1. The carbon metric name, dot-separated (e.g. room.temperature) If this is omitted, the MQTT topic name will be used as described above.
2. The integer value for the metric
3. An integer timestamp (UNIX epoch time) which defaults to "now".

In other words, the following payloads are valid:

15					just the value (metric name will be MQTT topic)
room.living 15				metric name and value
room.living 15 1405014635		metric name, value, and timestamp

⁠celery

The celery service sends messages to celery which celery workers can consume.

[config:celery]
broker_url = 'redis://localhost:6379/5'
app_name = 'celery'
celery_serializer = 'json'
targets = {
   'hello': [
      {
        'task': 'myapp.hello',
        'message_format': 'json'
        }
      ],
  }

[hello/#]
targets = celery:hello

Broker URL can be any broker supported by celery. Celery serializer is usually json or pickle. Json is recommended for security. Targets are selected by task name. Message_format can be either "json" or "text". If it is json, the message will be sent as a json payload rather than a string. In this configuration, all messages that match hello/ will be sent to the celery task "myapp.hello". The first argument of the celery task will be the message from mqtt.

⁠dbus

The dbus service send a message over the dbus to the user's desktop (only tested with Gnome3).

[config:dbus]
targets = {
    'warn' : [ 'Warning' ],
    'note' : [ 'Note' ]
    }

Requires:

• Python dbus⁠ bindings

⁠dnsupdate

The dnsupdate service updates an authoritative DNS server via RFC 2136 DNS Updates. Consider the following configuration:

[config:dnsupdate]
dns_nameserver = '127.0.0.2'
dns_keyname= 'mqttwarn-auth'
dns_keyblob= 'kQNwTJ ... evi2DqP5UA=='
targets = {
   #target             DNS-Zone      DNS domain              TTL,  type
   'temp'         :  [ 'foo.aa.',     'temperature.foo.aa.', 300, 'TXT'   ],
   'addr'         :  [ 'foo.aa.',     'www.foo.aa.',         60,  'A'   ],
  }

[test/temp]
targets = log:info, dnsupdate:temp
format = Current temperature: {payload}C

[test/a]
targets = log:info, dnsupdate:addr
format = {payload}

dns_nameserver is the address of the authoritative server the update should be sent to via a TCP update. dns_keyname and dns_keyblob are the TSIG key names and base64-representation of the key respectively. These can be created with either of:

ldns-keygen  -a hmac-sha256 -b 256 keyname
dnssec-keygen -n HOST -a HMAC-SHA256 -b 256 keyname

where keyname is the name then added to dns_keyname (in this example: mqttwarn-auth).

Supposing a BIND DNS server configured to allow updates, you would then configure it as follows:

key "mqttwarn-auth" {
  algorithm hmac-sha256;
  secret "kQNwTJ ... evi2DqP5UA==";
};

...
zone "foo.aa" in {
   type master;
   file "keytest/foo.aa";
   update-policy {
      grant mqttwarn-auth. zonesub ANY;
   };
};

For the test/temp topic, a pub and the resulting DNS query:

$ mosquitto_pub -t test/temp -m 42'
$ dig @127.0.0.2 +noall +answer temperature.foo.aa txt
temperature.foo.aa. 300 IN  TXT "Current temperature: 42C"

The test/a topic expects an address:

$ mosquitto_pub -t test/a -m 172.16.153.44
$ dig @127.0.0.2 +short www.foo.aa
172.16.153.44

Ensure you watch both mqttwarn's logfile as well as the log of your authoritative name server which will show you what's going on:

client 127.0.0.2#52786/key mqttwarn-auth: view internal: updating zone 'foo.aa/IN': adding an RR at 'www.foo.aa' A 172.16.153.44

Requires:

• dnspython⁠

⁠emoncms

The emoncms service sends a numerical payload to an EmonCMS⁠ instance. EmonCMS⁠ is a powerful open-source web-app for processing, logging and visualising energy, temperature and other environmental data.

The web-app can run locally or you can upload your readings to their server for viewing and monitoring via your own login (note this is likely to become a paid service in the medium term). See http://emoncms.org⁠ for details on installing and configuring your own instance.

By specifying the node id and input name in the mqttwarn target (see the ini example below) you can split different feeds into different nodes, and give each one a human readable name to identify them in EmonCMS.

[config:emoncms]
url     = <url of emoncms server e.g. http://localhost/emoncms or http://emoncms.org/emoncms>
apikey  = <apikey generated by the emoncms server>
timeout = 5
targets = {
    'usage'  : [ 1, 'usage' ],  # [ <nodeid>, <name> ]
    'solar'  : [ 1, 'solar' ]
    }

⁠execute

The execute target launches the specified program and its arguments. It is similar to pipe but it doesn't open a pipe to the program. Example use cases are f.e. IoT buttons which publish a message when they are pushed and the execute an external program. It is also a light version of mqtt-launcher⁠.

[config:execute]
targets = {
             # argv0 .....
   'touch' : [ 'touch', '/tmp/executed' ]
   }

To pass the published data (text) to the command, use [TEXT] which then gets replaced. This can also be configured with the text_replace parameter.

Note, that for each message targeted to the execute service, a new process is spawned (fork/exec), so it is quite "expensive".

⁠fbchat

Notification of one Facebook⁠ account requires an account. For now, this is only done for messaging from one account to another.

Upon configuring this service's targets, make sure the three (3) elements of the list are in the order specified!

[config:fbchat]
targets = {
  'janejol'   :  [ 'vvvvvvvvvvvvvvvvvvvvvv',                              #