feat(brain): support admin api bearer token

implementations/python/.env.example

@@ -13,3 +13,7 @@ A2C_YOLO_MODE=false
 
 # Optional: set a fixed device ID (if not set, a UUID is auto-generated and persisted)
 # A2C_DEVICE_ID=my-workstation
+
+# Optional: require Authorization: Bearer <token> on the Brain admin API.
+# Recommended whenever A2C_ADMIN_ENABLED=true outside a local-only lab.
+# A2C_ADMIN_TOKEN=change_me

implementations/python/AGENTS.md

@@ -147,6 +147,11 @@ uv run pytest --asyncio-mode=auto
 | `A2C_STORAGE_DIR` | `~/.a2c` | Both | JSON persistence root |
 | `A2C_YOLO_MODE` | `false` | Body | Skip consent dialogs, auto-approve all tool calls |
 | `A2C_DEVICE_ID` | _(auto UUID)_ | Body | Override the persistent device identifier |
+| `A2C_MODE` | `interactive` | Body | Body operating mode: `interactive` or `daemon` |
+| `A2C_ADMIN_ENABLED` | `false` | Brain | Enable the HTTP admin API |
+| `A2C_ADMIN_HOST` | `127.0.0.1` | Brain | Admin API listen host |
+| `A2C_ADMIN_PORT` | `50052` | Brain | Admin API port |
+| `A2C_ADMIN_TOKEN` | — | Brain / Operator | Optional bearer token for the admin API |
 
 `.env` is loaded at startup by both `brain/server.py` and `body/cli.py` / `body/tui.py` via `python-dotenv`. Never commit `.env`.
 

implementations/python/README.md

@@ -217,6 +217,7 @@ A2C_YOLO_MODE=true
 | `A2C_ADMIN_ENABLED` | `false` | Enable the HTTP admin API on the Brain |
 | `A2C_ADMIN_HOST` | `127.0.0.1` | Admin API listen host (set `0.0.0.0` inside Docker) |
 | `A2C_ADMIN_PORT` | `50052` | Admin API port |
+| `A2C_ADMIN_TOKEN` | — | Optional bearer token required by the admin API and Operator TUI |
 
 ---
 
@@ -300,6 +301,8 @@ The Brain's HTTP admin API must be enabled to push tasks to daemon devices:
 # .env
 A2C_ADMIN_ENABLED=true
 A2C_ADMIN_PORT=50052
+# Recommended outside local-only labs:
+A2C_ADMIN_TOKEN=change_me
 ```
 
 Or via Docker (already configured in `docker-compose.yml`):
@@ -328,11 +331,13 @@ environment:
 # Push a task to any live device
 curl -X POST http://localhost:50052/tasks \
   -H 'Content-Type: application/json' \
+  -H "Authorization: Bearer $A2C_ADMIN_TOKEN" \
   -d '{"device_id": "<device_id>", "task": "List files in the home directory"}'
 
 # Push a task in a named conversation thread
 curl -X POST http://localhost:50052/tasks \
   -H 'Content-Type: application/json' \
+  -H "Authorization: Bearer $A2C_ADMIN_TOKEN" \
   -d '{"device_id": "<device_id>", "task": "What is in /tmp?", "conversation_id": "ops-1"}'
 ```
 
@@ -405,6 +410,8 @@ PYTHONPATH=src uv run a2c-operator
 
 # Remote Brain
 PYTHONPATH=src uv run a2c-operator --api http://my-brain.example.com:50052
+# With admin token:
+A2C_ADMIN_TOKEN=change_me PYTHONPATH=src uv run a2c-operator
 ```
 
 ### Features

implementations/python/src/brain/admin.py

@@ -57,6 +57,7 @@
     from a2c.storage import A2CStorage
 
 log = logging.getLogger("a2c.brain.admin")
+ADMIN_TOKEN_KEY = web.AppKey("admin_token", str)
 
 
 def _json(data: object, status: int = 200) -> web.Response:
@@ -67,8 +68,27 @@ def _json(data: object, status: int = 200) -> web.Response:
     )
 
 
-def build_app(registry: "DeviceRegistry", storage: "A2CStorage") -> web.Application:
-    app = web.Application()
+@web.middleware
+async def require_admin_token(
+    request: web.Request,
+    handler,
+) -> web.StreamResponse:
+    token = request.app.get(ADMIN_TOKEN_KEY)
+    if token:
+        expected = f"Bearer {token}"
+        if request.headers.get("Authorization") != expected:
+            return _json({"error": "unauthorized"}, status=401)
+    return await handler(request)
+
+
+def build_app(
+    registry: "DeviceRegistry",
+    storage: "A2CStorage",
+    admin_token: str | None = None,
+) -> web.Application:
+    app = web.Application(middlewares=[require_admin_token])
+    if admin_token:
+        app[ADMIN_TOKEN_KEY] = admin_token
 
     # -- GET /devices ---------------------------------------------------------
 
@@ -214,8 +234,9 @@ async def start_admin_server(
     storage: "A2CStorage",
     host: str = "127.0.0.1",
     port: int = 50052,
+    admin_token: str | None = None,
 ) -> web.AppRunner:
-    app = build_app(registry, storage)
+    app = build_app(registry, storage, admin_token=admin_token)
     runner = web.AppRunner(app)
     await runner.setup()
     site = web.TCPSite(runner, host, port)

implementations/python/src/brain/operator_tui.py

@@ -36,10 +36,17 @@
 class AdminClient:
     """Thin async wrapper around the Brain admin HTTP API."""
 
-    def __init__(self, base_url: str) -> None:
+    def __init__(self, base_url: str, token: str | None = None) -> None:
         self.base_url = base_url.rstrip("/")
+        self.token = token
         self._session: aiohttp.ClientSession | None = None
 
+    @property
+    def _headers(self) -> dict[str, str]:
+        if not self.token:
+            return {}
+        return {"Authorization": f"Bearer {self.token}"}
+
     async def start(self) -> None:
         self._session = aiohttp.ClientSession()
 
@@ -50,14 +57,18 @@ async def close(self) -> None:
     async def _get(self, path: str, **params: Any) -> Any:
         assert self._session
         url = f"{self.base_url}{path}"
-        async with self._session.get(url, params={k: v for k, v in params.items() if v is not None}) as resp:
+        async with self._session.get(
+            url,
+            params={k: v for k, v in params.items() if v is not None},
+            headers=self._headers,
+        ) as resp:
             resp.raise_for_status()
             return await resp.json()
 
     async def _post(self, path: str, body: dict) -> Any:
         assert self._session
         url = f"{self.base_url}{path}"
-        async with self._session.post(url, json=body) as resp:
+        async with self._session.post(url, json=body, headers=self._headers) as resp:
             resp.raise_for_status()
             return await resp.json()
 
@@ -310,9 +321,9 @@ class OperatorApp(App):
     }
     """
 
-    def __init__(self, api_url: str) -> None:
+    def __init__(self, api_url: str, token: str | None = None) -> None:
         super().__init__()
-        self.api = AdminClient(api_url)
+        self.api = AdminClient(api_url, token=token)
         self._api_url = api_url
         self._selected_device: dict | None = None
         self._all_devices: list[dict] = []
@@ -574,9 +585,12 @@ def main() -> None:
 
     import sys
     api_url = None
+    token = os.environ.get("A2C_ADMIN_TOKEN")
     for i, arg in enumerate(sys.argv[1:]):
         if arg == "--api" and i + 1 < len(sys.argv) - 1:
             api_url = sys.argv[i + 2]
+        elif arg == "--token" and i + 1 < len(sys.argv) - 1:
+            token = sys.argv[i + 2]
         elif arg.startswith("http"):
             api_url = arg
 
@@ -588,7 +602,7 @@ def main() -> None:
         format="%(asctime)s %(levelname)-8s %(name)s  %(message)s",
     )
 
-    app = OperatorApp(api_url=api_url)
+    app = OperatorApp(api_url=api_url, token=token)
     app.run()
 
 

implementations/python/src/brain/server.py

@@ -576,12 +576,14 @@ async def serve() -> None:
         from brain.admin import start_admin_server
         admin_host = os.environ.get("A2C_ADMIN_HOST", "127.0.0.1")
         admin_port = int(os.environ.get("A2C_ADMIN_PORT", "50052"))
+        admin_token = os.environ.get("A2C_ADMIN_TOKEN") or None
         try:
             admin_runner = await start_admin_server(
                 registry=servicer.registry,
                 storage=storage,
                 host=admin_host,
                 port=admin_port,
+                admin_token=admin_token,
             )
         except Exception as exc:
             log.warning("Admin API could not start: %s", exc)

implementations/python/tests/test_admin_auth.py

@@ -0,0 +1,56 @@
+import tempfile
+
+import pytest
+from aiohttp.test_utils import TestClient, TestServer
+
+from a2c.storage import A2CStorage
+from brain.admin import build_app
+from brain.server import DeviceRegistry
+
+
+async def make_client(admin_token: str | None = None) -> TestClient:
+    app = build_app(
+        registry=DeviceRegistry(),
+        storage=A2CStorage(tempfile.mkdtemp()),
+        admin_token=admin_token,
+    )
+    client = TestClient(TestServer(app))
+    await client.start_server()
+    return client
+
+
+@pytest.mark.asyncio
+async def test_admin_api_allows_requests_without_configured_token() -> None:
+    client = await make_client()
+    try:
+        response = await client.get("/devices")
+
+        assert response.status == 200
+    finally:
+        await client.close()
+
+
+@pytest.mark.asyncio
+async def test_admin_api_rejects_missing_bearer_token_when_configured() -> None:
+    client = await make_client(admin_token="secret")
+    try:
+        response = await client.get("/devices")
+
+        assert response.status == 401
+        assert await response.json() == {"error": "unauthorized"}
+    finally:
+        await client.close()
+
+
+@pytest.mark.asyncio
+async def test_admin_api_accepts_configured_bearer_token() -> None:
+    client = await make_client(admin_token="secret")
+    try:
+        response = await client.get(
+            "/devices",
+            headers={"Authorization": "Bearer secret"},
+        )
+
+        assert response.status == 200
+    finally:
+        await client.close()