[eric] credentials: move a rotating token out of 9router's db.json without racing the router for the file

backend/apps/nine_router/credential_store.py

@@ -0,0 +1,167 @@
+"""Safe read/modify/write of 9Router's on-disk provider credentials.
+
+9Router owns ~/.9router/db.json and rewrites the whole file whenever it refreshes a token or a
+user edits a provider, and its HTTP API has no route that can write an OAuth connection's tokens
+(PUT /api/providers/[id] accepts name/priority/isActive/apiKey only, and apiKey only for apikey
+connections). So the only way to move an OAuth credential is to edit the file, which means we have
+to not race the router for it. Every mutation here happens with the router stopped.
+
+The reason any of this exists: providers hand back a NEW refresh token on every refresh and treat
+a replayed one as theft, revoking the whole grant family. So a credential may have exactly one
+holder that can refresh it. Removing `refreshToken` from a connection is what makes a given
+9Router instance structurally unable to rotate, because its refresh dispatcher bails on a falsy
+refreshToken before it ever calls the provider.
+"""
+
+import asyncio
+import json
+import logging
+import os
+import tempfile
+from typing import Any, Dict, List, Optional
+
+import httpx
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.nine_router import process
+
+logger = logging.getLogger(__name__)
+
+P_SHUTDOWN_TIMEOUT_S = 5.0
+P_DOWN_POLL_INTERVAL_S = 0.1
+P_DOWN_WAIT_S = 10.0
+
+
+class ProviderCredential(BaseModel):
+    """The transferable half of a 9Router provider connection."""
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    connection_id: str
+    provider: str
+    access_token: str
+    refresh_token: Optional[str] = None
+    expires_at: Optional[str] = None
+
+
+@typechecked
+def db_path() -> str:
+    return os.path.join(process.nine_router_data_dir(), "db.json")
+
+
+@typechecked
+def p_load_db() -> Optional[Dict[str, Any]]:
+    try:
+        with open(db_path(), encoding="utf-8") as f:
+            db = json.load(f)
+        return db if isinstance(db, dict) else None
+    except (OSError, ValueError):
+        logger.warning("could not read 9router db.json", exc_info=True)
+        return None
+
+
+@typechecked
+def p_write_db(db: Dict[str, Any]) -> bool:
+    """Atomic replace at 0600. A half-written db.json costs the user every provider connection."""
+    path = db_path()
+    directory = os.path.dirname(path)
+    handle, temp_path = tempfile.mkstemp(dir=directory, prefix=".db.json.", suffix=".tmp")
+    try:
+        with os.fdopen(handle, "w", encoding="utf-8") as f:
+            json.dump(db, f, indent=2)
+        os.chmod(temp_path, 0o600)
+        os.replace(temp_path, path)
+        return True
+    except OSError:
+        logger.warning("could not write 9router db.json", exc_info=True)
+        try:
+            os.unlink(temp_path)
+        except OSError:
+            pass
+        return False
+
+
+@typechecked
+def read_credential(connection_id: str) -> Optional[ProviderCredential]:
+    """The tokens for one connection, readable whether or not the router is up."""
+    for c in process.read_persisted_connections():
+        if c.get("id") != connection_id:
+            continue
+        access = c.get("accessToken")
+        if not isinstance(access, str) or not access:
+            return None
+        refresh = c.get("refreshToken")
+        expires = c.get("expiresAt")
+        return ProviderCredential(
+            connection_id=connection_id,
+            provider=str(c.get("provider") or ""),
+            access_token=access,
+            refresh_token=refresh if isinstance(refresh, str) and refresh else None,
+            expires_at=expires if isinstance(expires, str) else None,
+        )
+    return None
+
+
+@typechecked
+def list_oauth_connection_ids() -> List[str]:
+    """Connections that carry a rotating credential; apikey rows have nothing to lease."""
+    return [
+        str(c.get("id"))
+        for c in process.read_persisted_connections()
+        if c.get("authType") == "oauth" and c.get("id")
+    ]
+
+
+@typechecked
+async def p_request_shutdown() -> None:
+    """Ask the router to exit over HTTP. Its own seam so a test can never reach a real router."""
+    try:
+        async with httpx.AsyncClient(timeout=P_SHUTDOWN_TIMEOUT_S, headers=process.cli_auth_headers()) as client:
+            await client.post(f"{process.NINE_ROUTER_API}/shutdown")
+    except (httpx.HTTPError, AttributeError):
+        pass
+
+
+@typechecked
+async def p_stop_router() -> bool:
+    """Down the router however we can reach it. `stop()` alone only kills one we spawned; an
+    adopted port-holder has no handle, so ask it to shut itself down over HTTP first."""
+    await p_request_shutdown()
+    process.stop()
+    waited = 0.0
+    while waited < P_DOWN_WAIT_S:
+        if not process.is_running():
+            return True
+        await asyncio.sleep(P_DOWN_POLL_INTERVAL_S)
+        waited += P_DOWN_POLL_INTERVAL_S
+    return not process.is_running()
+
+
+@typechecked
+async def apply_to_connection(connection_id: str, changes: Dict[str, Any], drop: List[str]) -> bool:
+    """Set `changes` and delete `drop` on one connection, with the router stopped throughout.
+
+    Refuses to run if the router will not go down, because a concurrent refresh would either lose
+    our edit or, far worse, resurrect a refresh token we are in the middle of handing away.
+    """
+    if not await p_stop_router():
+        logger.error("refusing to edit 9router db.json: router would not stop")
+        return False
+    try:
+        db = p_load_db()
+        if db is None:
+            return False
+        connections = db.get("providerConnections")
+        if not isinstance(connections, list):
+            return False
+        target = next((c for c in connections if isinstance(c, dict) and c.get("id") == connection_id), None)
+        if target is None:
+            logger.error("9router connection %s not found", connection_id)
+            return False
+        target.update(changes)
+        for key in drop:
+            target.pop(key, None)
+        return p_write_db(db)
+    finally:
+        await process.ensure_running()

backend/apps/nine_router/process.py

@@ -99,7 +99,7 @@ def is_running() -> bool:
         return False
 
 
-def p_nine_router_data_dir() -> str:
+def nine_router_data_dir() -> str:
     """Where 9Router persists machine-id + auth/cli-secret, the two files we
     hash into the /api/* auth token on 0.4.x. Mirrors 9Router's own default
     (DATA_DIR env, else ~/.9router on unix, %APPDATA%/9router on win) so we read
@@ -123,7 +123,7 @@ def harden_data_dir_permissions() -> None:
     every token refresh, which would drop a chmod on the file itself within the hour."""
     if os.name == "nt":
         return
-    data_dir = p_nine_router_data_dir()
+    data_dir = nine_router_data_dir()
     try:
         if not os.path.isdir(data_dir):
             return
@@ -152,7 +152,7 @@ def cli_auth_token() -> str | None:
     if not is_running():
         return None
     try:
-        data_dir = p_nine_router_data_dir()
+        data_dir = nine_router_data_dir()
         try:
             with open(os.path.join(data_dir, "machine-id"), encoding="utf-8") as f:
                 machine_id = f.read().strip()
@@ -200,7 +200,6 @@ def p_find_9router_dir() -> str | None:
     p_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
 
     if p_is_packaged:
-        import sys
         p_resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
         p_candidate = os.path.join(p_resources, "router")
         if os.path.isdir(p_candidate):
@@ -386,7 +385,7 @@ def read_persisted_connections() -> list[dict]:
     Empty list on any read problem."""
     try:
         import json as p_json
-        with open(os.path.join(p_nine_router_data_dir(), "db.json"), encoding="utf-8") as f:
+        with open(os.path.join(nine_router_data_dir(), "db.json"), encoding="utf-8") as f:
             db = p_json.load(f)
         return [c for c in (db.get("providerConnections") or []) if isinstance(c, dict)]
     except Exception:

backend/tests/test_credential_store.py

@@ -0,0 +1,142 @@
+"""Moving a rotating credential out of 9Router's db.json without racing 9Router for the file.
+
+The stakes: providers issue a NEW refresh token on every refresh and treat a replayed one as
+theft, revoking the whole grant family. So exactly one holder may be able to refresh. Removing
+`refreshToken` from a connection is what makes an instance structurally unable to rotate. If that
+edit were lost to a concurrent router write, or if the router resurrected the token afterwards,
+we would have two rotators and a dead account.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_credential_store.py -v
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+
+import pytest
+
+import backend.apps.nine_router.credential_store as store
+from backend.apps.nine_router import process
+
+P_CONNECTION = {
+    "id": "conn-1",
+    "provider": "claude",
+    "authType": "oauth",
+    "accessToken": "access-value",
+    "refreshToken": "refresh-value",
+    "expiresAt": "2026-08-01T00:00:00.000Z",
+    "isActive": True,
+}
+
+
+@pytest.fixture
+def p_router(tmp_path, monkeypatch):
+    """A stopped router with one oauth connection on disk. Restart is recorded, never real."""
+    data_dir = tmp_path / "9router"
+    data_dir.mkdir()
+    db = {"providerConnections": [dict(P_CONNECTION), {"id": "conn-2", "authType": "apikey"}]}
+    (data_dir / "db.json").write_text(json.dumps(db))
+    monkeypatch.setattr(process, "nine_router_data_dir", lambda: str(data_dir))
+
+    state = {"running": True, "restarts": 0}
+
+    def p_stop() -> None:
+        state["running"] = False
+
+    async def p_ensure() -> None:
+        state["restarts"] += 1
+        state["running"] = True
+
+    async def p_no_http() -> None:
+        state["shutdown_calls"] += 1
+
+    state["shutdown_calls"] = 0
+    monkeypatch.setattr(process, "stop", p_stop)
+    monkeypatch.setattr(process, "is_running", lambda: state["running"])
+    monkeypatch.setattr(process, "ensure_running", p_ensure)
+    # Hard-stubbed: without this the suite would POST /shutdown at whatever real router owns the port.
+    monkeypatch.setattr(store, "p_request_shutdown", p_no_http)
+    return state
+
+
+def p_connection(data_dir_owner) -> dict:
+    db = json.loads(open(store.db_path(), encoding="utf-8").read())
+    return next(c for c in db["providerConnections"] if c["id"] == "conn-1")
+
+
+def test_read_credential_returns_the_tokens(p_router):
+    cred = store.read_credential("conn-1")
+    assert cred is not None
+    assert cred.provider == "claude"
+    assert cred.access_token == "access-value"
+    assert cred.refresh_token == "refresh-value"
+
+
+def test_only_oauth_connections_are_listed(p_router):
+    assert store.list_oauth_connection_ids() == ["conn-1"]
+
+
+@pytest.mark.asyncio
+async def test_dropping_the_refresh_token_removes_the_key(p_router):
+    """Absent, not blank. 9Router's refresh dispatcher bails on a falsy refreshToken, so the key
+    being gone is precisely what makes this instance unable to rotate."""
+    ok = await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    assert ok
+    after = p_connection(p_router)
+    assert "refreshToken" not in after
+    assert after["accessToken"] == "access-value", "must not disturb the rest of the connection"
+
+
+@pytest.mark.asyncio
+async def test_router_is_stopped_then_restarted(p_router):
+    await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    assert p_router["shutdown_calls"] == 1, "an adopted router only goes down over HTTP"
+    assert p_router["restarts"] == 1
+    assert p_router["running"] is True
+
+
+@pytest.mark.asyncio
+async def test_refuses_to_edit_when_the_router_will_not_stop(p_router, monkeypatch):
+    """The load-bearing guard. Editing under a live router risks losing the edit, or worse, the
+    router rewriting the refresh token back after we have already handed it to the cloud."""
+    monkeypatch.setattr(process, "is_running", lambda: True)
+    monkeypatch.setattr(process, "stop", lambda: None)
+    monkeypatch.setattr(store, "P_DOWN_WAIT_S", 0.2)
+
+    ok = await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+
+    assert ok is False
+    assert p_connection(p_router)["refreshToken"] == "refresh-value", "file must be untouched"
+
+
+@pytest.mark.asyncio
+async def test_restoring_a_refresh_token_round_trips(p_router):
+    await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    ok = await store.apply_to_connection("conn-1", changes={"refreshToken": "returned"}, drop=[])
+    assert ok
+    assert p_connection(p_router)["refreshToken"] == "returned"
+
+
+@pytest.mark.asyncio
+async def test_written_db_is_owner_only(p_router):
+    await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    mode = stat.S_IMODE(os.stat(store.db_path()).st_mode)
+    assert mode & 0o077 == 0
+
+
+@pytest.mark.asyncio
+async def test_unknown_connection_changes_nothing(p_router):
+    ok = await store.apply_to_connection("conn-missing", changes={"x": 1}, drop=[])
+    assert ok is False
+    assert p_connection(p_router)["refreshToken"] == "refresh-value"
+
+
+@pytest.mark.asyncio
+async def test_corrupt_db_is_not_overwritten(p_router):
+    open(store.db_path(), "w", encoding="utf-8").write("{not json")
+    ok = await store.apply_to_connection("conn-1", changes={}, drop=["refreshToken"])
+    assert ok is False
+    assert open(store.db_path(), encoding="utf-8").read() == "{not json"