diff --git a/marilib/README.md b/marilib/README.md index 4e5e5d8..cbfb932 100644 --- a/marilib/README.md +++ b/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 diff --git a/marilib/examples/mari_edge_stats.py b/marilib/examples/mari_edge_stats.py index 90f3217..03dd8d9 100644 --- a/marilib/examples/mari_edge_stats.py +++ b/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 ), diff --git a/marilib/marilib/cli/cloud.py b/marilib/marilib/cli/cloud.py index 2b86b9e..2ff28a2 100644 --- a/marilib/marilib/cli/cloud.py +++ b/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 ), diff --git a/marilib/marilib/cli/edge.py b/marilib/marilib/cli/edge.py index d4316ee..9a1d31f 100644 --- a/marilib/marilib/cli/edge.py +++ b/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(), diff --git a/marilib/marilib/communication_adapter.py b/marilib/marilib/communication_adapter.py index dd8bc47..335c677 100644 --- a/marilib/marilib/communication_adapter.py +++ b/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[/]") diff --git a/marilib/marilib/tests/test_communication_adapter.py b/marilib/marilib/tests/test_communication_adapter.py index 35a5e79..2173421 100644 --- a/marilib/marilib/tests/test_communication_adapter.py +++ b/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