diff --git a/.env.example b/.env.example index fbce5ec..1115f4b 100644 --- a/.env.example +++ b/.env.example @@ -40,6 +40,8 @@ MQTT_BROKER_PORT=1883 MQTT_USERNAME= MQTT_PASSWORD= MQTT_TLS_ENABLED=false +# MQTT_CLIENT_ID must be a non-empty ASCII slug; use a distinct value per deployment. +MQTT_CLIENT_ID=home-automation # Optional: Home Assistant MQTT Discovery. # Requires MQTT_ENABLED=true and a running MQTT broker. diff --git a/app/api/routes/api/config.py b/app/api/routes/api/config.py index ffe1120..aeea72f 100644 --- a/app/api/routes/api/config.py +++ b/app/api/routes/api/config.py @@ -9,7 +9,7 @@ from sqlalchemy.orm import Session from app.api.routes.api.deps import require_csrf, require_session from app.config import Settings, get_settings from app.dependencies import get_app_settings, get_db -from app.integrations.mqtt import MQTT_SETTINGS_KEYS, mqtt_manager +from app.integrations.mqtt import MQTT_SETTINGS_KEYS, mqtt_manager, mqtt_test_client_id from app.schemas.config import ( ConfigField, ConfigResponse, @@ -233,7 +233,7 @@ def _run_mqtt_test(settings: Settings) -> None: client = mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, - client_id="home-automation-test", + client_id=mqtt_test_client_id(settings.mqtt_client_id), ) def _on_connect( diff --git a/app/config.py b/app/config.py index ece2eac..aa80b79 100644 --- a/app/config.py +++ b/app/config.py @@ -1,7 +1,8 @@ from functools import lru_cache from pathlib import Path +import re -from pydantic import computed_field +from pydantic import computed_field, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -53,6 +54,7 @@ class Settings(BaseSettings): mqtt_username: str = "" mqtt_password: str = "" mqtt_tls_enabled: bool = False + mqtt_client_id: str = "home-automation" # Home Assistant MQTT Discovery (T08 wires into CONFIG_FIELDS/UI; T11 does publishing). ha_discovery_enabled: bool = False @@ -81,6 +83,17 @@ class Settings(BaseSettings): extra="ignore", ) + @field_validator("mqtt_client_id", mode="before") + @classmethod + def validate_mqtt_client_id(cls, value: object) -> str: + """Normalize a broker-safe base client identity used by every MQTT client.""" + if not isinstance(value, str): + raise ValueError("MQTT client ID must be a string") + normalized = value.strip() + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}", normalized): + raise ValueError("MQTT client ID must be a non-empty ASCII slug") + return normalized + @computed_field @property def is_development(self) -> bool: diff --git a/app/integrations/mqtt.py b/app/integrations/mqtt.py index 6db1316..0fb575d 100644 --- a/app/integrations/mqtt.py +++ b/app/integrations/mqtt.py @@ -48,6 +48,7 @@ MQTT_SETTINGS_KEYS = { "mqtt_username", "mqtt_password", "mqtt_tls_enabled", + "mqtt_client_id", } @@ -65,6 +66,16 @@ def _is_configured(settings: Settings) -> bool: return bool(settings.mqtt_enabled and settings.mqtt_broker_host) +def mqtt_source_client_id(base_client_id: str, source_id: int) -> str: + """Return a stable, deployment-scoped identity for one DSMR source.""" + return f"{base_client_id}-dsmr-source-{source_id}" + + +def mqtt_test_client_id(base_client_id: str) -> str: + """Return a transient test identity that cannot evict a long-lived client.""" + return f"{base_client_id}-test" + + class MqttManager: """Long-lived MQTT client wrapper. @@ -236,6 +247,7 @@ class MqttManager: password: str, tls_enabled: bool, subscriptions: dict[str, Callable[[bytes], None]], + base_client_id: str = "home-automation", ) -> bool: """Replace one source-owned client and its handlers. @@ -253,7 +265,7 @@ class MqttManager: captured_subscriptions = dict(subscriptions) client = mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, - client_id=f"home-automation-dsmr-{source_id}", + client_id=mqtt_source_client_id(base_client_id, source_id), ) def _on_connect( @@ -360,7 +372,7 @@ class MqttManager: """Build a fresh paho Client, configure it, and call loop_start + connect.""" client = mqtt.Client( callback_api_version=mqtt.CallbackAPIVersion.VERSION2, - client_id="home-automation", + client_id=settings.mqtt_client_id, ) # Callbacks — VERSION2 on_connect signature: diff --git a/app/main.py b/app/main.py index 801d882..bd16fe3 100644 --- a/app/main.py +++ b/app/main.py @@ -305,7 +305,7 @@ async def lifespan(_: FastAPI): # DSMR sources carry their own runtime configuration and are reconciled # after the MQTT manager is connected. - apply_dsmr_subscription() + apply_dsmr_subscription(_startup_runtime_settings) # Mark it before reconcile: a partial reconcile can already own a fd or # a non-daemon thread and must receive the same orderly shutdown. serial_started = True diff --git a/app/services/config_page.py b/app/services/config_page.py index 9fd434a..e000641 100644 --- a/app/services/config_page.py +++ b/app/services/config_page.py @@ -109,6 +109,7 @@ CONFIG_FIELDS: tuple[ConfigField, ...] = ( ConfigField("MQTT", "MQTT_USERNAME", "mqtt_username", "MQTT Username"), ConfigField("MQTT", "MQTT_PASSWORD", "mqtt_password", "MQTT Password", secret=True), ConfigField("MQTT", "MQTT_TLS_ENABLED", "mqtt_tls_enabled", "MQTT TLS Enabled", input_type="checkbox"), + ConfigField("MQTT", "MQTT_CLIENT_ID", "mqtt_client_id", "MQTT Client ID"), ConfigField( "Home Assistant Discovery", "HA_DISCOVERY_ENABLED", @@ -222,7 +223,12 @@ def save_config_updates(session: Session, form_data: dict[str, str], bootstrap_s else: merged_values[field.env_name] = submitted_value - _validate_config_values(merged_values, bootstrap_settings) + validated_settings = _validate_config_values(merged_values, bootstrap_settings) + # Persist the canonical client identity as well as using it at runtime. A + # whitespace-padded value must not survive in app_config and unexpectedly + # reappear in another consumer of the stored settings. + if "MQTT_CLIENT_ID" in merged_values: + merged_values["MQTT_CLIENT_ID"] = validated_settings.mqtt_client_id _persist_config_values(session, merged_values) get_settings.cache_clear() reset_db_caches() @@ -237,7 +243,9 @@ def save_config_value( ) -> None: current_values = _read_config_values(session) current_values[env_name] = value - _validate_config_values(current_values, bootstrap_settings) + validated_settings = _validate_config_values(current_values, bootstrap_settings) + if env_name == "MQTT_CLIENT_ID": + current_values[env_name] = validated_settings.mqtt_client_id _persist_config_values(session, current_values) get_settings.cache_clear() reset_db_caches() @@ -256,14 +264,14 @@ def _read_config_values(session: Session) -> dict[str, str]: return {row.key: row.value for row in rows} -def _validate_config_values(config_values: dict[str, str], bootstrap_settings: Settings) -> None: +def _validate_config_values(config_values: dict[str, str], bootstrap_settings: Settings) -> Settings: payload = _settings_payload(bootstrap_settings) for field in CONFIG_FIELDS: if field.env_name in config_values: payload[field.setting_attr] = config_values[field.env_name] try: - Settings(_env_file=None, **payload) + return Settings(_env_file=None, **payload) except Exception as exc: raise ConfigSaveError("invalid config submission") from exc @@ -334,6 +342,7 @@ def _settings_payload(settings: Settings) -> dict[str, Any]: "mqtt_username": settings.mqtt_username, "mqtt_password": settings.mqtt_password, "mqtt_tls_enabled": settings.mqtt_tls_enabled, + "mqtt_client_id": settings.mqtt_client_id, "ha_discovery_enabled": settings.ha_discovery_enabled, "ha_discovery_prefix": settings.ha_discovery_prefix, "ha_state_topic_prefix": settings.ha_state_topic_prefix, diff --git a/app/services/dsmr_ingest.py b/app/services/dsmr_ingest.py index fd2489c..6d71b4d 100644 --- a/app/services/dsmr_ingest.py +++ b/app/services/dsmr_ingest.py @@ -39,6 +39,7 @@ class DsmrSourceSnapshot: _subscriptions: dict[int, DsmrSourceSnapshot] = {} +_subscription_client_ids: dict[int, str] = {} _subscription_lock = threading.RLock() # A configuration value is not an ownership identity: disable and re-enable # can produce an equal snapshot. Each installed handler therefore captures a @@ -160,13 +161,16 @@ def _enabled_snapshots() -> list[DsmrSourceSnapshot]: session.close() -def apply_dsmr_subscription(_settings: "Settings | None" = None) -> None: +def apply_dsmr_subscription(settings: "Settings | None" = None) -> None: """Reconcile enabled DSMR source subscriptions from the database. - The optional, ignored settings parameter preserves the old config-route - call shape while ensuring flat DSMR settings are no longer read. + ``settings`` provides the DB-merged app-wide MQTT identity. Individual + DSMR broker settings continue to come solely from MeterSource records. """ from app.integrations.mqtt import mqtt_manager + from app.config import get_settings + + base_client_id = (settings or get_settings()).mqtt_client_id try: desired = {snapshot.source_id: snapshot for snapshot in _enabled_snapshots()} @@ -191,9 +195,11 @@ def apply_dsmr_subscription(_settings: "Settings | None" = None) -> None: source_id for source_id, current in _subscriptions.items() if desired.get(source_id) != current + or _subscription_client_ids.get(source_id) != base_client_id ] for source_id in stale_source_ids: _subscriptions.pop(source_id, None) + _subscription_client_ids.pop(source_id, None) _subscription_tokens.pop(source_id, None) set_current_tariff(source_id, None) @@ -203,7 +209,12 @@ def apply_dsmr_subscription(_settings: "Settings | None" = None) -> None: for source_id, snapshot in desired.items(): with _subscription_lock: current = _subscriptions.get(source_id) - if current == snapshot and mqtt_manager.source_is_active(source_id): + current_client_id = _subscription_client_ids.get(source_id) + if ( + current == snapshot + and current_client_id == base_client_id + and mqtt_manager.source_is_active(source_id) + ): continue if current is not None: # The client went inactive outside reconcile. Invalidate its @@ -211,6 +222,7 @@ def apply_dsmr_subscription(_settings: "Settings | None" = None) -> None: with _subscription_lock: if _subscriptions.get(source_id) == current: _subscriptions.pop(source_id, None) + _subscription_client_ids.pop(source_id, None) _subscription_tokens.pop(source_id, None) mqtt_manager.remove_source(source_id) token = object() @@ -227,6 +239,7 @@ def apply_dsmr_subscription(_settings: "Settings | None" = None) -> None: ) with _subscription_lock: _subscriptions[source_id] = snapshot + _subscription_client_ids[source_id] = base_client_id _subscription_tokens[source_id] = token applied = mqtt_manager.replace_source( source_id, @@ -236,11 +249,13 @@ def apply_dsmr_subscription(_settings: "Settings | None" = None) -> None: password=snapshot.password, tls_enabled=snapshot.tls_enabled, subscriptions=handlers, + base_client_id=base_client_id, ) if not applied: with _subscription_lock: if _subscription_tokens.get(source_id) is token: _subscriptions.pop(source_id, None) + _subscription_client_ids.pop(source_id, None) _subscription_tokens.pop(source_id, None) diff --git a/tests/test_api_config.py b/tests/test_api_config.py index 0ae25ee..fa08453 100644 --- a/tests/test_api_config.py +++ b/tests/test_api_config.py @@ -454,6 +454,7 @@ def test_get_config_includes_mqtt_section(client: TestClient) -> None: assert "MQTT_USERNAME" in env_names assert "MQTT_PASSWORD" in env_names assert "MQTT_TLS_ENABLED" in env_names + assert "MQTT_CLIENT_ID" in env_names def test_get_config_includes_ha_discovery_section(client: TestClient) -> None: @@ -578,6 +579,18 @@ def test_put_config_invalid_mqtt_port_returns_422_and_does_not_write( assert rows.get("MQTT_BROKER_PORT") != "not-a-number" +def test_put_config_invalid_mqtt_client_id_returns_422(client: TestClient) -> None: + _login(client) + + response = client.put( + "/api/config", + json={"updates": _full_config_payload({"MQTT_CLIENT_ID": "invalid id"})}, + headers={"X-CSRF-Token": "token"}, + ) + + assert response.status_code == 422 + + # --------------------------------------------------------------------------- # M5-polish2 Area B: bool fields have input_type="checkbox" # --------------------------------------------------------------------------- @@ -722,6 +735,32 @@ def test_put_config_mqtt_reconnect_uses_db_merged_settings( ) +def test_put_config_mqtt_client_id_reconnects_and_trims_value( + client: TestClient, test_database_urls +) -> None: + _login(client) + mock_mgr = MagicMock() + + with patch("app.api.routes.api.config.mqtt_manager", mock_mgr): + response = client.put( + "/api/config", + json={"updates": _full_config_payload({"MQTT_CLIENT_ID": " home-automation-dev "})}, + headers={"X-CSRF-Token": "token"}, + ) + + assert response.status_code == 200 + reconnect_settings = mock_mgr.reconnect.call_args.args[0] + assert reconnect_settings.mqtt_client_id == "home-automation-dev" + conn = sqlite3.connect(test_database_urls["app_path"]) + try: + stored_value = conn.execute( + "SELECT value FROM app_config WHERE key = 'MQTT_CLIENT_ID'" + ).fetchone() + finally: + conn.close() + assert stored_value == ("home-automation-dev",) + + # --------------------------------------------------------------------------- # M6-T02: DSMR + Tibber CONFIG_FIELDS # --------------------------------------------------------------------------- diff --git a/tests/test_app.py b/tests/test_app.py index 2d4a002..256a219 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -112,6 +112,7 @@ def test_app_start_seeds_missing_config_from_env_without_overwriting_existing_va assert rows["APP_NAME"] == "Database Owned Name" assert rows["HOME_ASSISTANT_BASE_URL"] == "http://bootstrap-ha.local:8123" assert rows["AUTH_SESSION_COOKIE_NAME"] == "home_automation_session" + assert rows["MQTT_CLIENT_ID"] == "home-automation" get_settings.cache_clear() reset_db_caches() diff --git a/tests/test_dsmr_subscription_apply.py b/tests/test_dsmr_subscription_apply.py index b3defe6..9625a96 100644 --- a/tests/test_dsmr_subscription_apply.py +++ b/tests/test_dsmr_subscription_apply.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import replace +from types import SimpleNamespace import pytest @@ -44,6 +45,7 @@ def fake_mqtt(monkeypatch): fake = _FakeMqtt() monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake) monkeypatch.setattr(dsmr_ingest, "_subscriptions", {}) + monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {}) monkeypatch.setattr(dsmr_ingest, "_tariffs", {}) return fake @@ -81,6 +83,18 @@ def test_same_snapshot_has_no_subscription_churn(fake_mqtt, monkeypatch): assert fake_mqtt.remove_calls == [] +def test_changed_base_client_id_replaces_all_enabled_sources(fake_mqtt, monkeypatch): + snapshot = _source(1, "topic", broker_host="broker.test") + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot]) + + dsmr_ingest.apply_dsmr_subscription(SimpleNamespace(mqtt_client_id="home-automation")) + dsmr_ingest.apply_dsmr_subscription(SimpleNamespace(mqtt_client_id="home-automation-dev")) + + assert fake_mqtt.remove_calls == [1] + assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 1] + assert fake_mqtt.replace_calls[-1][1]["base_client_id"] == "home-automation-dev" + + def test_same_topic_on_different_brokers_is_allowed(fake_mqtt, monkeypatch): monkeypatch.setattr( dsmr_ingest, diff --git a/tests/test_mqtt_client.py b/tests/test_mqtt_client.py index 88b4a9d..f1d18c0 100644 --- a/tests/test_mqtt_client.py +++ b/tests/test_mqtt_client.py @@ -9,7 +9,7 @@ from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient -from app.integrations.mqtt import MqttManager +from app.integrations.mqtt import MqttManager, mqtt_source_client_id, mqtt_test_client_id # --------------------------------------------------------------------------- @@ -24,6 +24,7 @@ def _make_settings( mqtt_username: str = "", mqtt_password: str = "", mqtt_tls_enabled: bool = False, + mqtt_client_id: str = "home-automation", ha_discovery_prefix: str = "homeassistant", ): """Return a simple namespace that acts like a Settings object for MqttManager tests.""" @@ -34,6 +35,7 @@ def _make_settings( s.mqtt_username = mqtt_username s.mqtt_password = mqtt_password s.mqtt_tls_enabled = mqtt_tls_enabled + s.mqtt_client_id = mqtt_client_id s.ha_discovery_prefix = ha_discovery_prefix return s @@ -120,6 +122,34 @@ def test_connect_creates_paho_client_with_version2() -> None: ) +def test_connect_uses_configured_client_id() -> None: + manager = MqttManager() + settings = _make_settings(mqtt_client_id="home-automation-dev") + mock_client = MagicMock() + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client) as mock_cls: + manager.connect(settings) + + assert mock_cls.call_args.kwargs["client_id"] == "home-automation-dev" + + +def test_deployment_client_id_variants_are_distinct() -> None: + production_ids = { + "home-automation", + mqtt_source_client_id("home-automation", 7), + mqtt_test_client_id("home-automation"), + } + development_ids = { + "home-automation-dev", + mqtt_source_client_id("home-automation-dev", 7), + mqtt_test_client_id("home-automation-dev"), + } + + assert len(production_ids) == 3 + assert len(development_ids) == 3 + assert production_ids.isdisjoint(development_ids) + + def test_connect_sets_credentials_when_username_provided() -> None: manager = MqttManager() settings = _make_settings(mqtt_username="user", mqtt_password="s3cr3t") @@ -431,3 +461,17 @@ def test_run_mqtt_test_raises_connection_error_on_timeout() -> None: with patch("threading.Event", side_effect=_make_event): with pytest.raises(_MqttConnectionError, match="timed out"): _run_mqtt_test(settings) + + +def test_run_mqtt_test_uses_deployment_scoped_client_id() -> None: + from app.api.routes.api.config import _run_mqtt_test, _MqttConnectionError + + settings = _make_settings(mqtt_client_id="home-automation-dev") + mock_client = MagicMock() + mock_client.connect.side_effect = OSError("Connection refused") + + with patch("paho.mqtt.client.Client", return_value=mock_client) as mock_cls: + with pytest.raises(_MqttConnectionError): + _run_mqtt_test(settings) + + assert mock_cls.call_args.kwargs["client_id"] == "home-automation-dev-test" diff --git a/tests/test_mqtt_subscribe.py b/tests/test_mqtt_subscribe.py index b9e4afa..bf89091 100644 --- a/tests/test_mqtt_subscribe.py +++ b/tests/test_mqtt_subscribe.py @@ -251,6 +251,7 @@ def test_replace_source_uses_isolated_client_and_source_credentials() -> None: password="one-secret", tls_enabled=True, subscriptions={"one/topic": lambda payload: received.append(("one", payload))}, + base_client_id="home-automation-dev", ) manager.replace_source( 2, @@ -284,6 +285,25 @@ def test_replace_source_uses_isolated_client_and_source_credentials() -> None: assert received == [("two", b"two"), ("changed", b"changed")] +def test_source_client_id_is_scoped_by_deployment_and_source() -> None: + manager = MqttManager() + client = MagicMock() + + with patch("app.integrations.mqtt.mqtt.Client", return_value=client) as mock_cls: + assert manager.replace_source( + 42, + host="broker.test", + port=1883, + username="", + password="", + tls_enabled=False, + subscriptions={"topic": lambda _payload: None}, + base_client_id="home-automation-dev", + ) + + assert mock_cls.call_args.kwargs["client_id"] == "home-automation-dev-dsmr-source-42" + + def test_replaced_source_client_callbacks_cannot_reach_new_generation() -> None: """A retained old paho client cannot subscribe, mutate state, or dispatch new handlers.""" manager = MqttManager()