feat(mqtt-auth): put legacy "vesper" password behind MQTT_ALLOW_LEGACY_PASSWORD
The shared legacy password is still needed for boards on pre-HMAC firmware, but it was accepted for any username. Now: - controlled by MQTT_ALLOW_LEGACY_PASSWORD (config.py, default true; documented in .env.example) so it can be switched off without a deploy, - only accepted for device-shaped usernames (uppercase alphanumeric segments joined by "-", optional "-kiosk"), never for app_ users or any other shape, - every successful legacy login is logged at WARNING with the username, rate-limited to once per username per hour, so the boards still depending on it are visible before the flag is turned off. HMAC auth is unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -17,6 +17,10 @@ MOSQUITTO_PASSWORD_FILE=/etc/mosquitto/passwd
|
||||
MQTT_CLIENT_ID=bellsystems-admin-panel
|
||||
# HMAC secret used to derive per-device MQTT passwords (must match firmware)
|
||||
MQTT_SECRET=change-me-in-production
|
||||
# Accept the legacy shared "vesper" MQTT password for device usernames
|
||||
# (old firmware). Each use is logged once/hour per board; set to false
|
||||
# once no boards show up in those logs.
|
||||
MQTT_ALLOW_LEGACY_PASSWORD=true
|
||||
|
||||
# App
|
||||
BACKEND_CORS_ORIGINS=["http://localhost:5173"]
|
||||
|
||||
@@ -19,6 +19,9 @@ class Settings(BaseSettings):
|
||||
mqtt_admin_username: str = "admin"
|
||||
mqtt_admin_password: str = ""
|
||||
mqtt_secret: str = "change-me-in-production"
|
||||
# Accept the old shared "vesper" password for device-shaped usernames
|
||||
# (boards on pre-HMAC firmware). Turn off once the logs show no users.
|
||||
mqtt_allow_legacy_password: bool = True
|
||||
mosquitto_password_file: str = "/etc/mosquitto/passwd"
|
||||
mqtt_client_id: str = "bellsystems-admin-panel"
|
||||
|
||||
|
||||
+47
-8
@@ -9,8 +9,10 @@ Password strategy: HMAC-SHA256(MQTT_SECRET, username)[:32]
|
||||
- Deterministic: no storage needed, re-derive on every auth check
|
||||
- Rotating MQTT_SECRET invalidates all passwords at once if needed
|
||||
|
||||
Transition support: during rollout, the legacy password "vesper" is also
|
||||
accepted so that devices still on old firmware stay connected.
|
||||
Transition support: while MQTT_ALLOW_LEGACY_PASSWORD is on (default), the
|
||||
legacy password "vesper" is also accepted for device-shaped usernames so
|
||||
boards still on old firmware stay connected. Each such login is logged
|
||||
(once per username per hour) to show which boards still depend on it.
|
||||
|
||||
User types handled:
|
||||
- Device users (e.g. "PV25L22BP01R01", "PV-26A18-BC02R-X7KQA"):
|
||||
@@ -35,6 +37,9 @@ stalling the event loop.
|
||||
import hmac
|
||||
import hashlib
|
||||
import logging
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
|
||||
from fastapi import APIRouter, Form, Response
|
||||
from firebase_admin import auth as firebase_auth
|
||||
@@ -50,6 +55,14 @@ LEGACY_PASSWORD = "vesper"
|
||||
|
||||
APP_USER_PREFIX = "app_"
|
||||
|
||||
# Uppercase alphanumeric segments joined by "-", optional "-kiosk" suffix.
|
||||
# Covers PV25L22BP01R01, PV-26A18-BC02R-X7KQA, BSVSPR-26C13X-STD01R-X7KQA.
|
||||
DEVICE_USERNAME_RE = re.compile(r"[A-Z0-9]{2,}(?:-[A-Z0-9]+)*(?:-kiosk)?")
|
||||
|
||||
LEGACY_LOG_INTERVAL_SECONDS = 3600
|
||||
_legacy_log_last: dict[str, float] = {}
|
||||
_legacy_log_lock = threading.Lock()
|
||||
|
||||
# Users authenticated via passwd file (go-auth file backend).
|
||||
# If they somehow reach the HTTP ACL endpoint, grant full access.
|
||||
SUPERUSERS = {"admin", "bonamin", "NodeRED"}
|
||||
@@ -64,18 +77,44 @@ def _derive_password(username: str) -> str:
|
||||
).hexdigest()[:32]
|
||||
|
||||
|
||||
def _is_device_username(username: str) -> bool:
|
||||
"""Board serial (optionally with -kiosk), e.g. "PV25L22BP01R01",
|
||||
"BSVSPR-26C13X-STD01R-X7KQA", "PV25L22BP01R01-kiosk"."""
|
||||
return bool(DEVICE_USERNAME_RE.fullmatch(username))
|
||||
|
||||
|
||||
def _log_legacy_auth(username: str) -> None:
|
||||
"""Log a legacy-password login, at most once per username per interval."""
|
||||
now = time.monotonic()
|
||||
with _legacy_log_lock:
|
||||
last = _legacy_log_last.get(username)
|
||||
if last is not None and now - last < LEGACY_LOG_INTERVAL_SECONDS:
|
||||
return
|
||||
_legacy_log_last[username] = now
|
||||
logger.warning("MQTT legacy password accepted for %s — board still on pre-HMAC firmware", username)
|
||||
|
||||
|
||||
def _is_valid_password(username: str, password: str) -> bool:
|
||||
"""
|
||||
Accept the password if it matches either:
|
||||
- The HMAC-derived password (new firmware)
|
||||
- The legacy hardcoded "vesper" password (old firmware, transition period)
|
||||
|
||||
Remove the legacy check in Stage 7 once all devices are on new firmware.
|
||||
- The legacy hardcoded "vesper" password (old firmware, transition period),
|
||||
only when MQTT_ALLOW_LEGACY_PASSWORD is on and the username is
|
||||
device-shaped (never app_ users or unknown shapes).
|
||||
"""
|
||||
expected = _derive_password(username)
|
||||
hmac_ok = hmac.compare_digest(expected, password)
|
||||
legacy_ok = hmac.compare_digest(LEGACY_PASSWORD, password)
|
||||
return hmac_ok or legacy_ok
|
||||
if hmac.compare_digest(expected, password):
|
||||
return True
|
||||
|
||||
if (
|
||||
settings.mqtt_allow_legacy_password
|
||||
and _is_device_username(username)
|
||||
and hmac.compare_digest(LEGACY_PASSWORD, password)
|
||||
):
|
||||
_log_legacy_auth(username)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def _base_sn(username: str) -> str:
|
||||
|
||||
Reference in New Issue
Block a user