feat(mqtt-auth): per-device topic ACL for phone-app users

POST /mqtt/auth/acl now handles "app_<uid>" users:
- topic must be exactly vesper/{serial}/<a>/<b> with serial in the user's
  device_serials (resolved by the users doc `uid` field),
- publish (acc 2): only control/command,
- subscribe (acc 4) and read/delivery (acc 1): only control/ack,
  status/heartbeat, status/playback,
- wildcard topics (+ / #) are denied,
- clientid must start with "app_<uid>_" so one user cannot reuse another
  user's client id to kick them off,
- blocked users are denied; anything else (incl. other acc values) is 403.

Lookups go through mqtt/app_users.py's 60s TTL cache so per-message
checks don't hit Firestore every time; assign/unassign/block invalidate it.

Also fixes the acc comment: mosquitto passes 1 = read (delivery),
2 = write (publish), 4 = subscribe - not "1 = subscribe, 3 = both".
Device/kiosk ACL is unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

backend/mqtt/auth.py

@@ -161,26 +161,87 @@ def mqtt_auth_user(
     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 = subscribe, 2 = publish, 3 = subscribe+publish
+    acc: int = Form(...),  # 1 = read (delivery), 2 = write (publish), 4 = subscribe
 ):
     """
-    Called by Mosquitto on every SUBSCRIBE and PUBLISH.
+    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)