marilib: add MARI_MQTT_INSECURE and survive a failed broker connect

marilib/README.md

@@ -17,6 +17,12 @@ To run with a gateway connected via UART:
 ```
 You can see how it works using `mari-edge --help`.
 
+Broker credentials come from the environment, never the command line:
+`MARI_MQTT_USER` / `MARI_MQTT_PASS`. `MARI_MQTT_INSECURE=1` skips the broker
+certificate check (the link stays encrypted but the broker is no longer
+authenticated), for a bench whose broker certificate has expired or is
+self-signed.
+
 To run with a gateway connected via MQTT:
 ```bash
 # for example, using the Inria Argus MQTT broker

marilib/examples/mari_edge_stats.py

@@ -9,7 +9,7 @@
 from marilib.model import SCHEDULES, EdgeEvent, GatewayInfo, MariNode, TestState
 from marilib.serial_uart import get_default_port
 from marilib.tui_edge import MarilibTUIEdge
-from marilib.cli.edge import mqtt_credentials
+from marilib.communication_adapter import mqtt_options_from_env
 from marilib.communication_adapter import SerialAdapter, MQTTAdapter
 
 
@@ -224,7 +224,7 @@ def main(
         on_event,
         serial_interface=SerialAdapter(port),
         mqtt_interface=(
-            MQTTAdapter.from_url(mqtt_host, is_edge=True, **mqtt_credentials())
+            MQTTAdapter.from_url(mqtt_host, is_edge=True, **mqtt_options_from_env())
             if mqtt_host
             else None
         ),

marilib/marilib/cli/cloud.py

@@ -1,30 +1,14 @@
-import os
 import time
 
 import click
 from marilib.mari_protocol import MARI_BROADCAST_ADDRESS, MARI_NET_ID_DEFAULT, DefaultPayload, Frame
 from marilib.marilib_cloud import MarilibCloud
 from marilib.model import EdgeEvent, GatewayInfo, MariNode
-from marilib.communication_adapter import MQTTAdapter
+from marilib.communication_adapter import MQTTAdapter, mqtt_options_from_env
 from marilib.tui_cloud import MarilibTUICloud
 from marilib.logger import MetricsLogger
 
 
-def mqtt_credentials() -> dict[str, str | None]:
-    """MQTT username and password from the environment, as kwargs.
-
-    Same variable names as the `dotbot` CLI, so one export serves both.
-    Returned as a dict for `**` splatting: positional unpacking collides with
-    `is_edge` at every call site. Keeping credentials out of the URL keeps
-    them out of shell history and out of any command a run sheet reproduces;
-    MQTTAdapter.from_url still accepts the `mqtts://user:pass@host` form.
-    """
-    return {
-        "username": os.environ.get("DOTBOT_MQTT_USER"),
-        "password": os.environ.get("DOTBOT_MQTT_PASS"),
-    }
-
-
 def on_event(event: EdgeEvent, event_data: MariNode | Frame | GatewayInfo):
     """An event handler for the application."""
     pass
@@ -37,7 +21,7 @@ def on_event(event: EdgeEvent, event_data: MariNode | Frame | GatewayInfo):
     type=str,
     default="mqtt://localhost:1883",
     show_default=True,
-    help="MQTT broker to use",
+    help="MQTT broker to use. Credentials come from MARI_MQTT_USER / MARI_MQTT_PASS; MARI_MQTT_INSECURE=1 skips the broker certificate check.",
 )
 @click.option(
     "--network-id",
@@ -74,7 +58,7 @@ def main(mqtt_url: str, network_id: int, send_periodic: float, log_dir: str, max
 
     mari = MarilibCloud(
         on_event,
-        mqtt_interface=MQTTAdapter.from_url(mqtt_url, is_edge=False, **mqtt_credentials()),
+        mqtt_interface=MQTTAdapter.from_url(mqtt_url, is_edge=False, **mqtt_options_from_env()),
         logger=MetricsLogger(
             log_dir_base=log_dir, rotation_interval_minutes=1440, log_interval_seconds=1.0
         ),

marilib/marilib/cli/edge.py

@@ -1,31 +1,15 @@
-import os
 import time
 
 import click
 from marilib.logger import MetricsLogger
 from marilib.mari_protocol import Frame, MARI_BROADCAST_ADDRESS, DefaultPayload
 from marilib.model import EdgeEvent, MariNode
-from marilib.communication_adapter import SerialAdapter, MQTTAdapter
+from marilib.communication_adapter import SerialAdapter, MQTTAdapter, mqtt_options_from_env
 from marilib.serial_uart import get_default_port
 from marilib.tui_edge import MarilibTUIEdge
 from marilib.marilib_edge import MarilibEdge
 
 
-def mqtt_credentials() -> dict[str, str | None]:
-    """MQTT username and password from the environment, as kwargs.
-
-    Same variable names as the `dotbot` CLI, so one export serves both.
-    Returned as a dict for `**` splatting: positional unpacking collides with
-    `is_edge` at every call site. Keeping credentials out of the URL keeps
-    them out of shell history and out of any command a run sheet reproduces;
-    MQTTAdapter.from_url still accepts the `mqtts://user:pass@host` form.
-    """
-    return {
-        "username": os.environ.get("DOTBOT_MQTT_USER"),
-        "password": os.environ.get("DOTBOT_MQTT_PASS"),
-    }
-
-
 def on_event(event: EdgeEvent, event_data: MariNode | Frame):
     """An event handler for the application."""
     pass
@@ -45,7 +29,7 @@ def on_event(event: EdgeEvent, event_data: MariNode | Frame):
     "-m",
     type=str,
     default=None,
-    help="MQTT broker to use (default: None, no cloud)",
+    help="MQTT broker to use (default: None, no cloud). Credentials come from MARI_MQTT_USER / MARI_MQTT_PASS; MARI_MQTT_INSECURE=1 skips the broker certificate check.",
 )
 @click.option(
     "--metrics-probe-interval",
@@ -72,7 +56,9 @@ def main(port: str | None, mqtt_url: str, metrics_probe_interval: float, log_dir
         on_event,
         serial_interface=SerialAdapter(port),
         mqtt_interface=(
-            MQTTAdapter.from_url(mqtt_url, is_edge=True, **mqtt_credentials()) if mqtt_url else None
+            MQTTAdapter.from_url(mqtt_url, is_edge=True, **mqtt_options_from_env())
+            if mqtt_url
+            else None
         ),
         logger=logger,
         tui=MarilibTUIEdge(),

marilib/marilib/communication_adapter.py

@@ -1,4 +1,6 @@
 import base64
+import os
+import ssl
 import time
 from abc import ABC, abstractmethod
 from dataclasses import asdict, dataclass
@@ -136,6 +138,24 @@ def parse_mqtt_url(url: str):
     return parsed.hostname, port, use_tls, parsed.username, parsed.password
 
 
+MQTT_USER_ENV = "MARI_MQTT_USER"
+MQTT_PASS_ENV = "MARI_MQTT_PASS"
+MQTT_INSECURE_ENV = "MARI_MQTT_INSECURE"
+
+
+def mqtt_options_from_env() -> dict:
+    """`MQTTAdapter.from_url` kwargs read from the environment.
+
+    `MARI_MQTT_INSECURE` takes 1/true/yes/on.
+    """
+    insecure = os.environ.get(MQTT_INSECURE_ENV, "").strip().lower()
+    return {
+        "username": os.environ.get(MQTT_USER_ENV),
+        "password": os.environ.get(MQTT_PASS_ENV),
+        "insecure": insecure in {"1", "true", "yes", "on"},
+    }
+
+
 class MQTTAdapter(CommunicationAdapterBase):
     """Class used to interface with MQTT."""
 
@@ -147,7 +167,9 @@ def __init__(
         use_tls: bool = False,
         username: str | None = None,
         password: str | None = None,
+        insecure: bool = False,
     ):
+        """`insecure` skips broker certificate checking on a TLS connection."""
         self.host = host
         self.port = port
         self.is_edge = is_edge
@@ -157,6 +179,7 @@ def __init__(
         self.use_tls = use_tls
         self.username = username
         self.password = password
+        self.insecure = insecure
         # optimize qos for throughput
         # 0 = no delivery guarantee, 1 = at least once, 2 = exactly once
         self.qos = 0
@@ -168,6 +191,7 @@ def from_url(
         is_edge: bool,
         username: str | None = None,
         password: str | None = None,
+        insecure: bool = False,
     ):
         """Build an MQTTAdapter from a `mqtt(s)://[user:pass@]host[:port]` URL.
 
@@ -183,6 +207,7 @@ def from_url(
             use_tls=use_tls,
             username=username if username is not None else url_user,
             password=password if password is not None else url_pass,
+            insecure=insecure,
         )
 
     # ==== public methods ====
@@ -226,14 +251,35 @@ def init(self):
         if self.username is not None:
             self.client.username_pw_set(self.username, self.password)
         if self.use_tls:
-            self.client.tls_set_context(context=None)
+            self.client.tls_set_context(context=self._tls_context())
         self.client.on_log = self._on_log
         self.client.on_connect = self._on_connect_edge if self.is_edge else self._on_connect_cloud
         self.client.on_message = self._on_message_edge if self.is_edge else self._on_message_cloud
-        self.client.connect(self.host, self.port, 60)
-        print(f"[yellow]Connected to MQTT broker on {self.host}:{self.port}[/]")
+        # On the edge this runs on the serial reader thread, so a failed
+        # connect must not raise: paho's loop retries in the background.
+        try:
+            self.client.connect(self.host, self.port, 60)
+            print(f"[yellow]Connected to MQTT broker on {self.host}:{self.port}[/]")
+        except (OSError, ValueError) as exc:
+            print(f"[red]MQTT broker {self.host}:{self.port} unreachable: {exc}[/]")
+            if isinstance(exc, ssl.SSLCertVerificationError):
+                print(f"[red]{MQTT_INSECURE_ENV}=1 skips the certificate check[/]")
+            print("[yellow]Retrying the broker in the background[/]")
         self.client.loop_start()
 
+    def _tls_context(self) -> ssl.SSLContext | None:
+        """None asks paho for its default, verifying context."""
+        if not self.insecure:
+            return None
+        print(
+            "[red]Broker certificate checking is OFF: the connection is encrypted "
+            "but the broker is not authenticated[/]"
+        )
+        context = ssl.create_default_context()
+        context.check_hostname = False
+        context.verify_mode = ssl.CERT_NONE
+        return context
+
     def close(self):
         if self.client is None:
             return
@@ -285,11 +331,23 @@ def _on_log(self, client, userdata, paho_log_level, messages):
         # print(messages)
         pass
 
+    def _refused(self, reason_code) -> bool:
+        """Report a CONNACK the broker refused; paho keeps retrying either way."""
+        if not reason_code.is_failure:
+            return False
+        print(f"[red]MQTT broker {self.host}:{self.port} refused the connection: {reason_code}[/]")
+        print(f"[red]Credentials come from {MQTT_USER_ENV} and {MQTT_PASS_ENV}[/]")
+        return True
+
     def _on_connect_edge(self, client, userdata, flags, reason_code, properties):
+        if self._refused(reason_code):
+            return
         self.client.subscribe(f"/mari/{self.network_id}/to_edge", qos=self.qos)
         print(f"[yellow]Subscribed to /mari/{self.network_id}/to_edge[/]")
 
     def _on_connect_cloud(self, client, userdata, flags, reason_code, properties):
+        if self._refused(reason_code):
+            return
         self.client.subscribe(f"/mari/{self.network_id}/to_cloud", qos=self.qos)
         print(f"[yellow]Subscribed to /mari/{self.network_id}/to_cloud[/]")
 

marilib/marilib/tests/test_communication_adapter.py

@@ -1,6 +1,16 @@
 """Tests for MQTTAdapter URL parsing and credential handling."""
 
-from marilib.communication_adapter import MQTTAdapter, parse_mqtt_url
+import socket
+import ssl
+
+import pytest
+
+from marilib.communication_adapter import (
+    MQTT_INSECURE_ENV,
+    MQTTAdapter,
+    mqtt_options_from_env,
+    parse_mqtt_url,
+)
 
 
 def test_parse_mqtt_url_parts():
@@ -84,3 +94,67 @@ def test_constructor_credentials_default_none():
     a = MQTTAdapter("h", 1883, is_edge=True)
     assert a.username is None
     assert a.password is None
+
+
+@pytest.mark.parametrize("value", ["1", "true", "YES", " on "])
+def test_env_insecure_accepts_truthy_values(monkeypatch, value):
+    monkeypatch.setenv(MQTT_INSECURE_ENV, value)
+    assert mqtt_options_from_env()["insecure"] is True
+
+
+@pytest.mark.parametrize("value", ["", "0", "no", "off", "maybe"])
+def test_env_insecure_rejects_anything_else(monkeypatch, value):
+    monkeypatch.setenv(MQTT_INSECURE_ENV, value)
+    assert mqtt_options_from_env()["insecure"] is False
+
+
+def test_env_options_carry_credentials(monkeypatch):
+    monkeypatch.setenv("MARI_MQTT_USER", "alice")
+    monkeypatch.setenv("MARI_MQTT_PASS", "pw")
+    monkeypatch.delenv(MQTT_INSECURE_ENV, raising=False)
+    a = MQTTAdapter.from_url("mqtts://argus:8883", is_edge=True, **mqtt_options_from_env())
+    assert (a.username, a.password, a.insecure) == ("alice", "pw", False)
+
+
+def test_default_tls_context_is_paho_verifying_default():
+    assert MQTTAdapter("h", 8883, is_edge=True, use_tls=True)._tls_context() is None
+
+
+def test_insecure_tls_context_skips_verification():
+    a = MQTTAdapter.from_url("mqtts://argus:8883", is_edge=True, insecure=True)
+    context = a._tls_context()
+    assert context.verify_mode == ssl.CERT_NONE
+    assert context.check_hostname is False
+
+
+def test_unreachable_broker_does_not_raise():
+    """init() runs on the serial reader thread; raising there kills it."""
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        port = s.getsockname()[1]
+    a = MQTTAdapter("127.0.0.1", port, is_edge=True)
+    try:
+        a.update("1234", lambda data: None)
+        assert a.client is not None
+        assert not a.is_ready()
+    finally:
+        a.close()
+
+
+@pytest.mark.parametrize("is_edge", [True, False])
+def test_a_refused_connack_is_reported_and_not_subscribed(capsys, is_edge):
+    from unittest.mock import MagicMock
+
+    from paho.mqtt.packettypes import PacketTypes
+    from paho.mqtt.reasoncodes import ReasonCode
+
+    a = MQTTAdapter("h", 1883, is_edge=is_edge)
+    a.network_id = "1234"
+    a.client = MagicMock()
+    refused = ReasonCode(PacketTypes.CONNACK, "Bad user name or password")
+    on_connect = a._on_connect_edge if is_edge else a._on_connect_cloud
+    on_connect(a.client, None, None, refused, None)
+    a.client.subscribe.assert_not_called()
+    out = capsys.readouterr().out
+    assert "refused" in out
+    assert "MARI_MQTT_USER" in out