Complete IoT Node: from Sensor to Dashboard

1Objectives of the lab

• Understanding the publish-subscribe model and why it replaces request-response in IoT
• Designing a topic hierarchy and mastering the + and # wildcards
• Installing and configuring an MQTT broker on a Raspberry Pi
• Publishing from an ESP32 and consuming the data in a bridge written in Python
• Using the last-will message and retained messages to detect nodes that have gone down
• Choosing the quality-of-service level based on the energy cost
• Practically demonstrating the vulnerabilities of a default configuration and securing it

2System architecture

3Materials needed

• 1 Raspberry Pi 5 or Pi 3 (from Lab 1)
• 2 ESP32-C6 with a sensor (from Lab 2)
• 1 16×2 I²C LCD
• 1 Lab Wi-Fi network you have access to
• 1 Computer for the web dashboard

Wiring diagram

4Why MQTT and not HTTP

In the second lab, the node sent its data with an HTTP request. It worked. Why change anything?

The broker's role

The node does not send data to someone. It publishes it on a topic, and the broker sends it to everyone subscribed to that topic. The node does not know how many consumers it has - maybe zero, maybe ten. The consumers do not know how many nodes exist. This decoupling is the whole idea: you can add a web dashboard, an alarm and a database without reprogramming a single sensor.

5The topic hierarchy

A topic is a string with levels separated by /, like a file path. A subscription can use wildcards. This is where most misunderstandings are born, so it is worth practicing interactively.

How to design a good hierarchy

6Installing the broker

1. Install Mosquitto on the Raspberry Pi
   installationcopy

   sudo apt update
   sudo apt install -y mosquitto mosquitto-clients
   sudo systemctl enable mosquitto

2. Configure it for the lab

   The default configuration only accepts local connections. We open the port for the network - for now with no security measures at all, precisely so we can see in section 10 what that means.

   create the configuration filecopy

   sudo nano /etc/mosquitto/conf.d/lab.conf

   /etc/mosquitto/conf.d/lab.confcopy

   # WARNING: insecure configuration, used deliberately for
   # the first part of the lab. Replaced in section 10.
   listener 1883 0.0.0.0
   allow_anonymous true
   
   # Verbose logging, so we can see what is happening.
   log_type all
   log_dest file /var/log/mosquitto/mosquitto.log

   restart and checkcopy

   sudo systemctl restart mosquitto
   sudo systemctl status mosquitto --no-pager

3. Find the broker's address
   network addresscopy

   hostname -I

   Note down the address - you will write it into the node's program. Ideally, reserve it a fixed address on the lab router, otherwise it changes on every reboot and the nodes are left orphaned.

4. Test the broker with itself

   Open two terminal windows. In the first:

   terminal 1 - listening to everythingcopy

   mosquitto_sub -h localhost -t 'lab/si/#' -v

   In the second:

   terminal 2 - publishingcopy

   mosquitto_pub -h localhost -t 'lab/si/node1/temperature' -m '23.5'
   mosquitto_pub -h localhost -t 'lab/si/node2/humidity'    -m '61'

   The messages must appear immediately in the first window. If they do, the broker works and you can move on to the node. If not, the problem is local and has nothing to do with the ESP32 - fix it here, it is much easier to debug.

7The publishing node

Install the PubSubClient library from the Arduino library manager, then upload the program below. It is the node from the second lab, with HTTP replaced by MQTT.

#include <WiFi.h>
#include <PubSubClient.h>
#include <Wire.h>
#include <Adafruit_BME280.h>
#include <esp_sleep.h>

#define uS_PER_S     1000000ULL
#define PERIOD_S     60
#define NODE_ID      "node1"

const char* SSID     = "lab_network";
const char* PASSWORD = "password";
const char* BROKER   = "192.168.1.10";
const uint16_t PORT  = 1883;

// The topics are built once, from the node's identifier.
const char* T_TEMP    = "lab/si/" NODE_ID "/temperature";
const char* T_HUMID   = "lab/si/" NODE_ID "/humidity";
const char* T_BATTERY = "lab/si/" NODE_ID "/battery";
const char* T_STATUS  = "lab/si/" NODE_ID "/status";

RTC_DATA_ATTR uint32_t cycle = 0;

WiFiClient network;
PubSubClient mqtt(network);
Adafruit_BME280 bme;

bool connect_broker() {
  mqtt.setServer(BROKER, PORT);

  // The last-will message: the broker publishes it AUTOMATICALLY if the
  // node disappears without disconnecting politely. It is the only way to
  // tell "the node is quiet because it has nothing to say" apart from
  // "the node has died".
  return mqtt.connect(NODE_ID,          // unique identifier on the network
                      NULL, NULL,       // username and password (none for now)
                      T_STATUS,         // the last-will message's topic
                      1,                // quality of service
                      true,             // retained: whoever subscribes later also sees it
                      "offline");       // the last-will message's content
}

void setup() {
  Serial.begin(115200);
  delay(150);
  cycle++;

  Wire.begin(6, 7);
  if (!bme.begin(0x76)) {
    Serial.println("the sensor is not responding");
  }
  bme.setSampling(Adafruit_BME280::MODE_FORCED,
                  Adafruit_BME280::SAMPLING_X1, Adafruit_BME280::SAMPLING_X1,
                  Adafruit_BME280::SAMPLING_X1, Adafruit_BME280::FILTER_OFF);
  bme.takeForcedMeasurement();
  float temperature = bme.readTemperature();
  float humidity    = bme.readHumidity();

  WiFi.mode(WIFI_STA);
  WiFi.begin(SSID, PASSWORD);
  uint32_t t0 = millis();
  while (WiFi.status() != WL_CONNECTED && millis() - t0 < 10000) delay(10);

  if (WiFi.status() == WL_CONNECTED && connect_broker()) {
    char value[16];

    // RETAINED message: the broker keeps the last value and sends it
    // immediately to anyone who subscribes. A dashboard opened an hour
    // later sees numbers, not a blank screen until the next measurement.
    snprintf(value, sizeof(value), "%.2f", temperature);
    mqtt.publish(T_TEMP, value, true);

    snprintf(value, sizeof(value), "%.1f", humidity);
    mqtt.publish(T_HUMID, value, true);

    snprintf(value, sizeof(value), "%d", (int) (analogRead(0) / 40.95));
    mqtt.publish(T_BATTERY, value, true);

    // Announce that we are alive; replaces the retained last-will message.
    mqtt.publish(T_STATUS, "online", true);

    mqtt.loop();                      // let the library actually send it
    delay(100);
    mqtt.disconnect();                // polite disconnect: does NOT trigger the last will
    Serial.printf("cycle %lu published in %lu ms\n", cycle, millis());
  } else {
    Serial.println("could not publish");
  }

  WiFi.disconnect(true);
  WiFi.mode(WIFI_OFF);
  Serial.flush();
  esp_sleep_enable_timer_wakeup((uint64_t) PERIOD_S * uS_PER_S);
  esp_deep_sleep_start();
}

void loop() {}

Check it right away, from the Raspberry Pi:

mosquitto_sub -h localhost -t 'lab/si/#' -v

8The bridge: collecting, storing, displaying

The bridge runs on the Raspberry Pi. It subscribes to everything the nodes publish, saves it to a database, updates the LCD and serves a web dashboard.

source ~/si-lab/bin/activate
pip install paho-mqtt
mkdir -p ~/si-lab/lab04 && cd ~/si-lab/lab04

#!/usr/bin/env python3
"""MQTT bridge: collects the readings, saves them and displays them."""

import json
import sqlite3
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

import paho.mqtt.client as mqtt

BROKER = "localhost"
PORT = 1883
TOPIC = "lab/si/#"
DATABASE = "readings.db"
WEB_PORT = 8080

# Last value of every topic, for a quick display.
latest = {}
lock = threading.Lock()


def prepare_database():
    con = sqlite3.connect(DATABASE)
    con.execute("""
        CREATE TABLE IF NOT EXISTS readings (
            timestamp REAL,
            node      TEXT,
            quantity  TEXT,
            value     TEXT
        )
    """)
    con.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON readings(timestamp)")
    con.commit()
    return con


def on_connect(client, userdata, flags, rc, properties=None):
    if rc == 0:
        print("connected to the broker")
        client.subscribe(TOPIC, qos=1)
    else:
        print("connection failed, code", rc)


def on_message(client, userdata, msg):
    text = msg.payload.decode("utf-8", errors="replace")
    levels = msg.topic.split("/")

    # We expect the form lab/si/<node>/<quantity>; ignore the rest.
    if len(levels) != 4:
        return
    _, _, node, quantity = levels

    with lock:
        latest[(node, quantity)] = (text, time.time())

    con = userdata["con"]
    con.execute("INSERT INTO readings VALUES (?,?,?,?)",
                (time.time(), node, quantity, text))
    con.commit()

    tag = " (retained)" if msg.retain else ""
    print("%-34s %s%s" % (msg.topic, text, tag))


def update_lcd(panel):
    """Rotates through the nodes on the LCD, one every 3 seconds."""
    while True:
        with lock:
            nodes = sorted({n for (n, _) in latest})
            snapshot = dict(latest)
        if not nodes:
            panel.write("waiting for nodes...", "")
        for node in nodes:
            t = snapshot.get((node, "temperature"), ("--", 0))[0]
            h = snapshot.get((node, "humidity"), ("--", 0))[0]
            status, moment = snapshot.get((node, "status"), ("?", 0))
            age = int(time.time() - moment) if moment else 999
            panel.write("%-6s %5s C" % (node, t),
                        "%3s%% %s %ds" % (h, status[:3], min(age, 999)))
            time.sleep(3)


PAGE = """<!DOCTYPE html><html lang="en"><head><meta charset="utf-8">
<meta http-equiv="refresh" content="5"><title>Nodes</title>
<style>body{font-family:monospace;background:#123526;color:#efede7;padding:24px}
table{border-collapse:collapse}td,th{border:1px solid #2a513d;padding:6px 12px;text-align:left}
th{color:#a0dfba}.old{color:#ff9d80}</style></head><body>
<h2>Active nodes</h2>%s</body></html>"""


class Server(BaseHTTPRequestHandler):
    def do_GET(self):
        with lock:
            snapshot = dict(latest)

        rows = ["<tr><th>node</th><th>quantity</th><th>value</th><th>age</th></tr>"]
        for (node, quantity), (value, moment) in sorted(snapshot.items()):
            age = time.time() - moment
            css_class = ' class="old"' if age > 180 else ""
            rows.append("<tr%s><td>%s</td><td>%s</td><td>%s</td><td>%d s</td></tr>"
                           % (css_class, node, quantity, value, age))

        body = (PAGE % ("<table>" + "".join(rows) + "</table>")).encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass          # do not flood the console with the web server's log


def main():
    from display import Panel          # reuse the class from Lab 1

    con = prepare_database()
    panel = Panel()

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                         client_id="bridge", userdata={"con": con})
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, PORT, keepalive=60)

    threading.Thread(target=update_lcd, args=(panel,), daemon=True).start()
    threading.Thread(target=HTTPServer(("", WEB_PORT), Server).serve_forever,
                     daemon=True).start()

    print("bridge active - web dashboard on port %d" % WEB_PORT)
    try:
        client.loop_forever()
    except KeyboardInterrupt:
        panel.close()
        con.close()
        print("\nstopped")


if __name__ == "__main__":
    main()

python3 bridge.py
# then open in a browser  http://<pi-address>:8080

sqlite3 readings.db "SELECT node, ROUND(AVG(CAST(value AS REAL)),2), COUNT(*)
  FROM readings WHERE quantity='temperature' AND timestamp > strftime('%s','now')-3600
  GROUP BY node;"

9Quality of service and its cost

MQTT offers three levels of delivery guarantee. The choice is not free: every extra level means more radio frames, so more transmission time, so more battery consumed.

Retained messages

A message published with the "retained" flag stays at the broker. Anyone who subscribes later gets it immediately, without waiting for the next measurement. For a node that wakes up once an hour, the difference is between a working dashboard and a blank screen.

# publish a retained message and exit
mosquitto_pub -h localhost -t 'lab/si/test/value' -m '42' -r

# subscribe ONLY NOW - and we still get the message
mosquitto_sub -h localhost -t 'lab/si/test/value' -v

# deleting a retained message is done by publishing EMPTY content
mosquitto_pub -h localhost -t 'lab/si/test/value' -m '' -r

10What an eavesdropper sees

So far we have run the broker with allow_anonymous true, exactly like most tutorials online. Now we see what that means.

Step 1 - anyone can subscribe to everything

mosquitto_sub -h 192.168.1.10 -t '#' -v

No password, no certificate. You see every measurement from every node. And if anything in the system reacted to commands, it could also receive commands:

mosquitto_pub -h 192.168.1.10 -t 'lab/si/node1/command' -m 'reset'

Step 2 - the traffic is plain text

On the Raspberry Pi, while the nodes are publishing:

sudo tcpdump -i any -A -n port 1883

In the output you will make out, in the clear: the topic names, the measured values, the node identifier - and, had you configured authentication, the username and password, just as readable.

Step 3 - authentication

sudo mosquitto_passwd -c /etc/mosquitto/passwd node1
sudo mosquitto_passwd    /etc/mosquitto/passwd node2
sudo mosquitto_passwd    /etc/mosquitto/passwd bridge

We also add access control: a sensor must be able to publish on its own topics, but has no reason to read the traffic of the entire network.

# The bridge reads everything, but publishes nothing.
user bridge
topic read lab/si/#

# Each node publishes ONLY on its own branch.
user node1
topic write lab/si/node1/#

user node2
topic write lab/si/node2/#

listener 1883 0.0.0.0
allow_anonymous false
password_file /etc/mosquitto/passwd
acl_file /etc/mosquitto/acl

sudo systemctl restart mosquitto

# without credentials - this must now fail
mosquitto_sub -h localhost -t '#' -v

# with credentials - this works
mosquitto_sub -h localhost -u bridge -P bridge_password -t 'lab/si/#' -v

# node 1 tries to publish on node 2's branch - must be rejected
mosquitto_pub -h localhost -u node1 -P node1_password -t 'lab/si/node2/temp' -m '99'

Step 4 - encryption

We generate our own certificate authority and a certificate for the broker. On a closed lab network, this is the correct solution; on a public system a publicly issued certificate would be used instead.

mkdir -p ~/certs && cd ~/certs

# 1. our own certificate authority
openssl req -new -x509 -days 3650 -extensions v3_ca \
  -keyout ca.key -out ca.crt -nodes -subj "/CN=SI Lab CA"

# 2. the key and request for the broker (CN = the broker's address!)
openssl req -new -nodes -keyout broker.key -out broker.csr \
  -subj "/CN=192.168.1.10"

# 3. signing the request with our own authority
openssl x509 -req -in broker.csr -CA ca.crt -CAkey ca.key \
  -CAcreateserial -out broker.crt -days 3650

sudo cp ca.crt broker.crt broker.key /etc/mosquitto/certs/
sudo chown mosquitto:mosquitto /etc/mosquitto/certs/broker.key
sudo chmod 600 /etc/mosquitto/certs/broker.key

# The insecure port stays closed for good.
listener 8883 0.0.0.0
allow_anonymous false
password_file /etc/mosquitto/passwd
acl_file /etc/mosquitto/acl

cafile   /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/broker.crt
keyfile  /etc/mosquitto/certs/broker.key
tls_version tlsv1.2

sudo systemctl restart mosquitto

# listen to the traffic again, now on the secure port
sudo tcpdump -i any -A -n port 8883

#include <WiFiClientSecure.h>

// Our own authority's certificate, copied from ca.crt.
// This is how the node verifies it is talking to OUR broker, not an impostor.
const char* CA_CERT = R"EOF(
-----BEGIN CERTIFICATE-----
... the contents of the ca.crt file ...
-----END CERTIFICATE-----
)EOF";

WiFiClientSecure network;
PubSubClient mqtt(network);

void setup() {
  // ... Wi-Fi connection, as before ...

  network.setCACert(CA_CERT);         // do NOT use setInsecure() in a real system
  mqtt.setServer(BROKER, 8883);

  mqtt.connect(NODE_ID, "node1", "node1_password",
               T_STATUS, 1, true, "offline");
  // ... the rest of the program is unchanged ...
}

11Assignments

1. Install the broker and verify it with mosquitto_pub and mosquitto_sub. Include a screenshot with the two terminals in your report.
2. Using the widget from section 5, fill in a table with eight published topics and, for each, which of the eight subscriptions receive it. Explain the two cases that surprised you.
3. Program two nodes with different identifiers and verify that the bridge displays both of them, alternating, on the LCD.
4. Demonstrate the last-will message working: cut the power to a connected node and show the moment the broker publishes the offline status on its own.
5. Compare quality levels 0 and 1: measure the duration of the node's active phase in both cases and calculate the autonomy difference with the widget from the second lab.
6. Capture the traffic on port 1883 and identify the topic and value of a measurement in it. Then switch to the secure port and repeat the capture.
7. Configure the access-control list and demonstrate that node 1 cannot publish on node 2's branch.

12Deeper-dive challenge

13Self-check questions

14Resources

• Chapter 11 - Internet of Things and smart device networks
• Eclipse Mosquitto documentation
• MQTT 5.0 specification (OASIS)
• Paho MQTT library for Python