From 30706cc62ea5773df347f6a391a4ec4b33297570 Mon Sep 17 00:00:00 2001 From: Brett Kinny Date: Fri, 5 Jun 2026 20:58:38 +1000 Subject: [PATCH] feat(bridge-auth): send X-Admin-Token on all /xiaozhi/admin/* calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Admin-auth epic, part 4 of 4 (bridge caller — the scattered one). Pairs with the permissive /xiaozhi/admin/* middleware (part 1). Adds a `_xiaozhi_admin_headers()` helper (reads DOTTY_ADMIN_TOKEN into a module-level `_ADMIN_TOKEN`) to both bridge.py and bridge/dashboard.py, threaded through every admin call site: - bridge.py: _dispatch_abort / _dispatch_set_state / _dispatch_set_toggle and the dashboard _dashboard_abort_device / _dashboard_inject_to_device helpers (5 POSTs). - dashboard.py: _xiaozhi_device_count + _xiaozhi_list_songs (urllib GETs, now via urllib.request.Request with headers) and play_song's play-asset POST (3 sites). Calls to dotty-behaviour (vision photo proxy, perception getters) are deliberately NOT given the header. Header only sent when the token is set — no-op until the flip. Tests: new tests/test_admin_headers.py (6 cases) — both helpers set/unset, and an integration through _dispatch_abort capturing the headers kwarg. Full tests/ 71 passed; ruff clean. Completes the foundation: #149 (server) + #150 (behaviour) + #151 (pi-ext) + this. Co-Authored-By: Claude Opus 4.8 (1M context) --- bridge.py | 28 +++++++-- bridge/dashboard.py | 18 +++++- tests/test_admin_headers.py | 117 ++++++++++++++++++++++++++++++++++++ 3 files changed, 155 insertions(+), 8 deletions(-) create mode 100644 tests/test_admin_headers.py diff --git a/bridge.py b/bridge.py index e90efe9..974dbd0 100644 --- a/bridge.py +++ b/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: diff --git a/bridge/dashboard.py b/bridge/dashboard.py index e02b4c8..6ff9754 100644 --- a/bridge/dashboard.py +++ b/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: diff --git a/tests/test_admin_headers.py b/tests/test_admin_headers.py new file mode 100644 index 0000000..e7d5e27 --- /dev/null +++ b/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()