bellsystems-cp/backend/mqtt/auth.py

"""
MQTT authentication and ACL endpoints for mosquitto-go-auth HTTP backend.

Mosquitto calls these on every CONNECT, SUBSCRIBE, and PUBLISH.
- POST /mqtt/auth/user  → validate device credentials
- POST /mqtt/auth/acl   → enforce per-device topic isolation

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: 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"):
    Authenticated via HMAC. ACL restricted to their own vesper/{sn}/... topics.
- Kiosk users (e.g. "PV25L22BP01R01-kiosk"):
    Same HMAC auth derived from the full kiosk username.
    ACL: allowed to access topics of their base device (suffix stripped).
- App users (e.g. "app_<firebase_uid>"):
    Remote phone app. Password = a Firebase ID token, verified with
    firebase_admin (revocation checked). No per-user MQTT accounts exist.
    Never accepted via HMAC or the legacy password.
- admin, bonamin, NodeRED, and other non-device users:
    These connect via the passwd file backend (go-auth file backend).
    They never reach this HTTP backend — go-auth resolves them first.
    The ACL endpoint below handles them defensively anyway (superuser list).

The handlers are plain `def` on purpose: they make blocking Firestore /
Firebase Auth calls, which FastAPI then runs in its threadpool instead of
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

from config import settings
from mqtt import app_users

logger = logging.getLogger("mqtt.auth")

router = APIRouter(prefix="/mqtt/auth", tags=["mqtt-auth"])

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"}


def _derive_password(username: str) -> str:
    """Derive the expected MQTT password for a given username."""
    return hmac.new(
        settings.mqtt_secret.encode(),
        username.encode(),
        hashlib.sha256,
    ).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),
      only when MQTT_ALLOW_LEGACY_PASSWORD is on and the username is
      device-shaped (never app_ users or unknown shapes).
    """
    expected = _derive_password(username)
    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:
    """
    Strip the -kiosk suffix if present, returning the base serial number.
    e.g. "PV25L22BP01R01-kiosk" -> "PV25L22BP01R01"
         "PV25L22BP01R01"       -> "PV25L22BP01R01"
    """
    if username.endswith("-kiosk"):
        return username[: -len("-kiosk")]
    return username


def _deny_app(username: str, reason: str) -> Response:
    # Never log the password — for app users it is a bearer token.
    logger.warning("MQTT app auth denied for %s: %s", username, reason)
    return Response(status_code=403)


def _auth_app_user(username: str, token: str) -> Response:
    """Authenticate "app_<firebase_uid>" with a Firebase ID token as password."""
    uid = username[len(APP_USER_PREFIX):]
    if not uid:
        return _deny_app(username, "empty uid")
    if not token:
        return _deny_app(username, "empty token")

    try:
        decoded = firebase_auth.verify_id_token(token, check_revoked=True)
    except firebase_auth.RevokedIdTokenError:
        return _deny_app(username, "token revoked")
    except firebase_auth.ExpiredIdTokenError:
        return _deny_app(username, "token expired")
    except firebase_auth.UserDisabledError:
        return _deny_app(username, "firebase account disabled")
    except firebase_auth.InvalidIdTokenError as e:
        return _deny_app(username, f"invalid token ({type(e).__name__})")
    except Exception as e:
        return _deny_app(username, f"token verification failed ({type(e).__name__})")

    if decoded.get("uid") != uid:
        return _deny_app(username, "token uid does not match username")

    try:
        # Fresh read on CONNECT (also refreshes the ACL cache).
        user = app_users.get_app_user(uid, use_cache=False)
    except Exception as e:
        return _deny_app(username, f"user lookup failed ({type(e).__name__})")

    if user is None:
        return _deny_app(username, "no user doc with this uid")
    if user.blocked:
        return _deny_app(username, "user is blocked")

    return Response(status_code=200)


@router.post("/user")
def mqtt_auth_user(
    username: str = Form(...),
    password: str = Form(...),
    clientid: str = Form(default=""),
):
    """
    Called by Mosquitto on every CONNECT.
    Returns 200 to allow, 403 to deny.

    Username = device SN (new format: "PV-26A18-BC02R-X7KQA", old format: "PV25L22BP01R01")
               or kiosk variant: "PV25L22BP01R01-kiosk"
               or app user: "app_<firebase_uid>"
    Password = HMAC-derived (new firmware) or "vesper" (legacy firmware)
               or, for app users, a Firebase ID token

    Note: admin, bonamin and NodeRED authenticate via the go-auth passwd file backend
    and never reach this endpoint.
    """
    if username.startswith(APP_USER_PREFIX):
        return _auth_app_user(username, password)

    if _is_valid_password(username, password):
        return Response(status_code=200)

    return Response(status_code=403)


# Mosquitto access values as passed through by go-auth (`acc`).
ACC_READ = 1       # message delivery to the client
ACC_WRITE = 2      # publish
ACC_SUBSCRIBE = 4  # subscribe

# App-user topic allowlist, relative to vesper/{serial}/
APP_WRITE_TOPICS = frozenset({"control/command"})
APP_READ_TOPICS = frozenset({"control/ack", "status/heartbeat", "status/playback"})


def _app_acl_allowed(username: str, clientid: str, topic: str, acc: int) -> tuple[bool, str]:
    """ACL decision for "app_<uid>". Returns (allowed, reason)."""
    uid = username[len(APP_USER_PREFIX):]
    if not uid:
        return False, "empty uid"

    # Client id must be owned by this user, so one user can't take over
    # (and disconnect) another user's session by reusing its client id.
    if not clientid.startswith(f"{APP_USER_PREFIX}{uid}_"):
        return False, f"clientid {clientid!r} not prefixed with app_<uid>_"

    if "+" in topic or "#" in topic:
        return False, "wildcards not allowed"

    parts = topic.split("/")
    if len(parts) != 4 or parts[0] != "vesper" or not parts[1]:
        return False, "topic not vesper/{serial}/<a>/<b>"
    serial, leaf = parts[1], f"{parts[2]}/{parts[3]}"

    if acc == ACC_WRITE:
        if leaf not in APP_WRITE_TOPICS:
            return False, "publish not allowed on this topic"
    elif acc in (ACC_READ, ACC_SUBSCRIBE):
        if leaf not in APP_READ_TOPICS:
            return False, "read/subscribe not allowed on this topic"
    else:
        return False, f"unsupported acc={acc}"

    try:
        user = app_users.get_app_user(uid)
    except Exception as e:
        return False, f"user lookup failed ({type(e).__name__})"
    if user is None:
        return False, "no user doc with this uid"
    if user.blocked:
        return False, "user is blocked"
    if serial not in user.device_serials:
        return False, "serial not assigned to user"

    return True, ""


@router.post("/acl")
def mqtt_auth_acl(
    username: str = Form(...),
    topic: str = Form(...),
    clientid: str = Form(default=""),
    acc: int = Form(...),  # 1 = read (delivery), 2 = write (publish), 4 = subscribe
):
    """
    Called by Mosquitto on every SUBSCRIBE, PUBLISH and message delivery.
    Returns 200 to allow, 403 to deny.

    Topic pattern: vesper/{sn}/...
    - Device users: may only access their own SN segment
    - Kiosk users: stripped of -kiosk suffix, then same rule applies
    - App users (app_<uid>): only their assigned serials, only the
      command/ack/heartbeat/playback topics — see _app_acl_allowed
    - Superusers (bonamin, NodeRED): full access
    """
    # Superusers get full access (shouldn't reach here but handled defensively)
    if username in SUPERUSERS:
        return Response(status_code=200)

    if username.startswith(APP_USER_PREFIX):
        allowed, reason = _app_acl_allowed(username, clientid, topic, acc)
        if allowed:
            return Response(status_code=200)
        logger.info("MQTT app ACL denied for %s (acc=%s, topic=%s): %s", username, acc, topic, reason)
        return Response(status_code=403)

    # Derive the base SN (handles -kiosk suffix)
    base = _base_sn(username)

    # Topic must be vesper/{base_sn}/...
    parts = topic.split("/")
    if len(parts) >= 2 and parts[0] == "vesper" and parts[1] == base:
        return Response(status_code=200)

    return Response(status_code=403)