feat(bridge-auth): send X-Admin-Token on all /xiaozhi/admin/* calls

bridge.py

@@ -166,6 +166,14 @@ def _write_smart_mode(enabled: bool) -> None:
 
 _XIAOZHI_HOST = os.environ.get("XIAOZHI_HOST", "")
 _XIAOZHI_HTTP_PORT = int(os.environ.get("XIAOZHI_OTA_PORT", "8003"))
+_ADMIN_TOKEN = os.environ.get("DOTTY_ADMIN_TOKEN", "").strip()
+
+
+def _xiaozhi_admin_headers() -> dict:
+    """X-Admin-Token header for /xiaozhi/admin/* requests when DOTTY_ADMIN_TOKEN
+    is set (matches the xiaozhi-server middleware); empty dict otherwise, so the
+    bridge is a no-op until the coordinated enforcement flip."""
+    return {"X-Admin-Token": _ADMIN_TOKEN} if _ADMIN_TOKEN else {}
 
 
 async def _dispatch_abort(device_id: str) -> None:
@@ -177,7 +185,9 @@ async def _dispatch_abort(device_id: str) -> None:
 
     def _post() -> None:
         try:
-            r = requests.post(url, json=payload, timeout=3)
+            r = requests.post(
+                url, json=payload, headers=_xiaozhi_admin_headers(), timeout=3
+            )
             if r.status_code >= 400:
                 log.warning("abort %s: %s", r.status_code, r.text[:200])
         except Exception as exc:
@@ -198,7 +208,9 @@ async def _dispatch_set_state(device_id: str, state: str) -> bool:
 
     def _post() -> bool:
         try:
-            r = requests.post(url, json=payload, timeout=3)
+            r = requests.post(
+                url, json=payload, headers=_xiaozhi_admin_headers(), timeout=3
+            )
             if r.status_code >= 400:
                 log.warning("set_state %s: %s", r.status_code, r.text[:200])
                 return False
@@ -221,7 +233,9 @@ async def _dispatch_set_toggle(device_id: str, name: str, enabled: bool) -> bool
 
     def _post() -> bool:
         try:
-            r = requests.post(url, json=payload, timeout=3)
+            r = requests.post(
+                url, json=payload, headers=_xiaozhi_admin_headers(), timeout=3
+            )
             if r.status_code >= 400:
                 log.warning("set_toggle %s: %s", r.status_code, r.text[:200])
                 return False
@@ -653,7 +667,9 @@ async def _dashboard_abort_device(*, device_id: str = "") -> dict:
 
         def _post() -> dict:
             try:
-                r = requests.post(url, json=payload, timeout=3)
+                r = requests.post(
+                    url, json=payload, headers=_xiaozhi_admin_headers(), timeout=3
+                )
                 if r.status_code == 200:
                     return {"ok": True, **r.json()}
                 if r.status_code == 503 and "no device connected" in r.text:
@@ -677,7 +693,9 @@ async def _dashboard_inject_to_device(*, text: str, device_id: str = "") -> dict
 
         def _post() -> dict:
             try:
-                r = requests.post(url, json=payload, timeout=3)
+                r = requests.post(
+                    url, json=payload, headers=_xiaozhi_admin_headers(), timeout=3
+                )
                 if r.status_code == 200:
                     return {"ok": True, **r.json()}
                 if r.status_code == 503 and "no device connected" in r.text:

bridge/dashboard.py

@@ -290,6 +290,13 @@ def _verify_dashboard_auth(
 XIAOZHI_HOST = os.environ.get("XIAOZHI_HOST", "")
 XIAOZHI_OTA_PORT = int(os.environ.get("XIAOZHI_OTA_PORT", "8003"))
 XIAOZHI_WS_PORT = int(os.environ.get("XIAOZHI_WS_PORT", "8000"))
+_ADMIN_TOKEN = os.environ.get("DOTTY_ADMIN_TOKEN", "").strip()
+
+
+def _xiaozhi_admin_headers() -> dict[str, str]:
+    """X-Admin-Token header for /xiaozhi/admin/* requests when DOTTY_ADMIN_TOKEN
+    is set (matches the xiaozhi-server middleware); empty otherwise."""
+    return {"X-Admin-Token": _ADMIN_TOKEN} if _ADMIN_TOKEN else {}
 LOG_DIR = Path(os.environ.get("CONVO_LOG_DIR", "/var/lib/dotty-bridge/logs"))
 VOICE_CHANNELS = ("dotty", "stackchan")
 
@@ -440,7 +447,8 @@ async def _xiaozhi_device_count() -> int | None:
     import urllib.request
     def _fetch() -> int | None:
         try:
-            with urllib.request.urlopen(url, timeout=2) as r:
+            req = urllib.request.Request(url, headers=_xiaozhi_admin_headers())
+            with urllib.request.urlopen(req, timeout=2) as r:
                 if r.status != 200:
                     return None
                 data = json.loads(r.read())
@@ -598,7 +606,8 @@ async def _xiaozhi_list_songs() -> tuple[list[str], str | None]:
     import urllib.request
     def _fetch() -> tuple[list[str], str | None]:
         try:
-            with urllib.request.urlopen(url, timeout=2) as r:
+            req = urllib.request.Request(url, headers=_xiaozhi_admin_headers())
+            with urllib.request.urlopen(req, timeout=2) as r:
                 data = json.loads(r.read())
                 files = data.get("files") or []
                 return [f for f in files if isinstance(f, str)], None
@@ -638,7 +647,10 @@ async def play_song(request: Request, filename: str = Form(...)) -> Any:
     url = f"http://{XIAOZHI_HOST}:{XIAOZHI_OTA_PORT}/xiaozhi/admin/play-asset"
     def _post() -> dict:
         try:
-            r = requests.post(url, json={"asset": asset_path}, timeout=3)
+            r = requests.post(
+                url, json={"asset": asset_path},
+                headers=_xiaozhi_admin_headers(), timeout=3,
+            )
             if r.status_code == 200:
                 return {"ok": True, "sent": base, "response": f"playing {base}"}
             if r.status_code == 503 and "no device connected" in r.text:

tests/test_admin_headers.py

@@ -0,0 +1,117 @@
+"""Tests for the bridge's X-Admin-Token wiring on /xiaozhi/admin/* calls.
+
+Admin-auth epic (bridge caller). bridge.py and bridge.dashboard each expose a
+`_xiaozhi_admin_headers()` helper that returns the X-Admin-Token header when
+DOTTY_ADMIN_TOKEN is set (read into a module-level `_ADMIN_TOKEN`) and an empty
+dict otherwise. The helper is threaded through every admin call site.
+"""
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import os
+import sys
+import tempfile
+import unittest
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+_state_dir = Path(tempfile.mkdtemp(prefix="dotty-adminhdr-state-"))
+os.environ.setdefault("DOTTY_KID_MODE_STATE", str(_state_dir / "kid-mode"))
+os.environ.setdefault("DOTTY_SMART_MODE_STATE", str(_state_dir / "smart-mode"))
+
+_repo_root = Path(__file__).resolve().parents[1]
+_spec = importlib.util.spec_from_file_location("bridge_app", _repo_root / "bridge.py")
+assert _spec is not None and _spec.loader is not None
+bridge_app = importlib.util.module_from_spec(_spec)
+sys.modules["bridge_app"] = bridge_app
+_spec.loader.exec_module(bridge_app)
+
+
+@asynccontextmanager
+async def _noop_lifespan(_app):
+    yield
+
+
+bridge_app.app.router.lifespan_context = _noop_lifespan
+
+import bridge.dashboard as dash  # noqa: E402
+
+
+@dataclass
+class _Resp:
+    status_code: int = 200
+    text: str = ""
+
+    def json(self) -> dict:
+        return {}
+
+
+class _PostRecorder:
+    def __init__(self) -> None:
+        self.headers: dict[str, str] | None = None
+
+    def __call__(self, url: str, *, json: Any = None, headers: Any = None,
+                 timeout: float = 0) -> _Resp:
+        self.headers = headers
+        return _Resp()
+
+
+class BridgeAdminHeaderTests(unittest.TestCase):
+
+    def setUp(self):
+        self._tok = bridge_app._ADMIN_TOKEN
+        self._host = bridge_app._XIAOZHI_HOST
+        self._post = bridge_app.requests.post
+
+    def tearDown(self):
+        bridge_app._ADMIN_TOKEN = self._tok
+        bridge_app._XIAOZHI_HOST = self._host
+        bridge_app.requests.post = self._post
+
+    def test_helper_returns_header_when_set(self):
+        bridge_app._ADMIN_TOKEN = "tok"
+        self.assertEqual(bridge_app._xiaozhi_admin_headers(), {"X-Admin-Token": "tok"})
+
+    def test_helper_empty_when_unset(self):
+        bridge_app._ADMIN_TOKEN = ""
+        self.assertEqual(bridge_app._xiaozhi_admin_headers(), {})
+
+    def test_dispatch_abort_sends_header_when_set(self):
+        bridge_app._ADMIN_TOKEN = "tok"
+        bridge_app._XIAOZHI_HOST = "127.0.0.1"
+        rec = _PostRecorder()
+        bridge_app.requests.post = rec
+        asyncio.run(bridge_app._dispatch_abort("dev-1"))
+        self.assertEqual(rec.headers, {"X-Admin-Token": "tok"})
+
+    def test_dispatch_abort_no_header_when_unset(self):
+        bridge_app._ADMIN_TOKEN = ""
+        bridge_app._XIAOZHI_HOST = "127.0.0.1"
+        rec = _PostRecorder()
+        bridge_app.requests.post = rec
+        asyncio.run(bridge_app._dispatch_abort("dev-1"))
+        self.assertEqual(rec.headers, {})
+
+
+class DashboardAdminHeaderTests(unittest.TestCase):
+
+    def setUp(self):
+        self._tok = dash._ADMIN_TOKEN
+
+    def tearDown(self):
+        dash._ADMIN_TOKEN = self._tok
+
+    def test_helper_returns_header_when_set(self):
+        dash._ADMIN_TOKEN = "tok"
+        self.assertEqual(dash._xiaozhi_admin_headers(), {"X-Admin-Token": "tok"})
+
+    def test_helper_empty_when_unset(self):
+        dash._ADMIN_TOKEN = ""
+        self.assertEqual(dash._xiaozhi_admin_headers(), {})
+
+
+if __name__ == "__main__":
+    unittest.main()