[eric] credentials: hand custody to the cloud strip-first, so two rotators is unrepresentable

backend/apps/nine_router/credential_lease.py

@@ -0,0 +1,209 @@
+"""Handing custody of a rotating provider credential between this device and the cloud.
+
+Cloud runs execute a user's workflow while their laptop is off, using the user's OWN subscription.
+That means our server needs to refresh their token, and providers issue a new refresh token on every
+refresh while treating a replayed one as theft, revoking the entire grant family. So a credential
+gets exactly ONE holder that can rotate it, and the handover has to be ordered so there is never an
+instant where both sides can.
+
+The order is strip-then-upload, never the reverse:
+  - Strip first, then upload: worst case nobody can rotate for a moment, which is harmless because
+    the access token stays valid for hours. We restore on failure.
+  - Upload first, then strip: if the strip fails, BOTH sides hold a rotating token, which is the
+    exact incident this whole design exists to prevent.
+
+The one genuinely ambiguous case is an upload that times out after the server already committed.
+Restoring blindly there would recreate the two-holder state, so we ask the server who owns it
+before deciding.
+"""
+
+import logging
+from datetime import datetime
+from typing import Literal, Optional
+
+import httpx
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.nine_router import credential_store
+from backend.apps.settings.credentials import proxy_auth
+from backend.apps.settings.store import load_settings
+
+logger = logging.getLogger(__name__)
+
+P_TIMEOUT_S = 20.0
+
+LeaseStatus = Literal[
+    "leased",
+    "released",
+    "refreshed",
+    "not_signed_in",
+    "no_such_connection",
+    "not_rotatable",
+    "cloud_rejected",
+    "local_write_failed",
+    "ownership_unknown",
+]
+
+
+class LeaseOutcome(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    status: LeaseStatus
+    detail: str = ""
+
+
+@typechecked
+def p_cloud() -> Optional[tuple[str, str]]:
+    """(bearer, base_url) for the signed-in user, or None when there is nothing to talk to."""
+    token, base = proxy_auth(load_settings())
+    if not token or not base:
+        return None
+    return (token, base)
+
+
+@typechecked
+async def p_lease_is_cloud_owned(connection_id: str) -> Optional[bool]:
+    """True/False if we can read ownership, None if we cannot tell. The None case is load-bearing:
+    guessing here is how you end up with two rotators."""
+    cloud = p_cloud()
+    if cloud is None:
+        return None
+    token, base = cloud
+    try:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_S) as client:
+            r = await client.get(
+                f"{base}/api/credentials/status",
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        if r.status_code != 200:
+            return None
+        for lease in r.json().get("leases") or []:
+            if lease.get("connection_id") == connection_id:
+                return lease.get("owner") == "cloud"
+        return False
+    except (httpx.HTTPError, ValueError):
+        return None
+
+
+@typechecked
+async def lease_to_cloud(connection_id: str) -> LeaseOutcome:
+    """Give the cloud sole custody so it can run this user's workflows while the laptop is off."""
+    cloud = p_cloud()
+    if cloud is None:
+        return LeaseOutcome(status="not_signed_in")
+    token, base = cloud
+
+    cred = credential_store.read_credential(connection_id)
+    if cred is None:
+        return LeaseOutcome(status="no_such_connection")
+    if not cred.refresh_token:
+        # Already stripped, or an api-key row. Either way there is no rotating secret to hand over.
+        return LeaseOutcome(status="not_rotatable")
+
+    refresh_token = cred.refresh_token
+    if not await credential_store.apply_to_connection(connection_id, changes={}, drop=["refreshToken"]):
+        return LeaseOutcome(status="local_write_failed")
+
+    payload = {
+        "connection_id": connection_id,
+        "provider": cred.provider,
+        "access_token": cred.access_token,
+        "refresh_token": refresh_token,
+        "expires_at": expires_ms(cred.expires_at),
+    }
+    try:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_S) as client:
+            r = await client.post(
+                f"{base}/api/credentials/lease",
+                headers={"Authorization": f"Bearer {token}"},
+                json=payload,
+            )
+        if r.status_code == 200:
+            return LeaseOutcome(status="leased")
+        await p_restore(connection_id, refresh_token)
+        return LeaseOutcome(status="cloud_rejected", detail=f"HTTP {r.status_code}")
+    except httpx.HTTPError as exc:
+        # The request may still have committed server-side, so ask before putting the token back.
+        owned = await p_lease_is_cloud_owned(connection_id)
+        if owned is True:
+            return LeaseOutcome(status="leased", detail="upload reported an error but the lease exists")
+        if owned is False:
+            await p_restore(connection_id, refresh_token)
+            return LeaseOutcome(status="cloud_rejected", detail=str(exc))
+        logger.error("lease upload outcome unknown for %s; leaving the token off this device", connection_id)
+        return LeaseOutcome(status="ownership_unknown", detail=str(exc))
+
+
+@typechecked
+async def p_restore(connection_id: str, refresh_token: str) -> None:
+    if not await credential_store.apply_to_connection(
+        connection_id, changes={"refreshToken": refresh_token}, drop=[]
+    ):
+        logger.error("could not restore the local refresh token for %s; the user must reconnect", connection_id)
+
+
+@typechecked
+def expires_ms(expires_at: Optional[str]) -> int:
+    """9Router stores an ISO string; the cloud wants unix ms. Unparseable reads as already expired,
+    which makes the server refresh on first use instead of trusting a bad clock."""
+    if not expires_at:
+        return 0
+    try:
+        return int(datetime.fromisoformat(expires_at.replace("Z", "+00:00")).timestamp() * 1000)
+    except ValueError:
+        return 0
+
+
+@typechecked
+async def release_to_device(connection_id: str) -> LeaseOutcome:
+    """Take custody back. The server hands the live refresh token home and drops its own copy."""
+    cloud = p_cloud()
+    if cloud is None:
+        return LeaseOutcome(status="not_signed_in")
+    token, base = cloud
+    try:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_S) as client:
+            r = await client.post(
+                f"{base}/api/credentials/release",
+                headers={"Authorization": f"Bearer {token}"},
+                json={"connection_id": connection_id},
+            )
+    except httpx.HTTPError as exc:
+        return LeaseOutcome(status="cloud_rejected", detail=str(exc))
+    if r.status_code != 200:
+        # 409 means a refresh is mid-exchange; the caller retries rather than taking a doomed token.
+        return LeaseOutcome(status="cloud_rejected", detail=f"HTTP {r.status_code}")
+    body = r.json()
+    ok = await credential_store.apply_to_connection(
+        connection_id,
+        changes={"accessToken": body["access_token"], "refreshToken": body["refresh_token"]},
+        drop=[],
+    )
+    return LeaseOutcome(status="released" if ok else "local_write_failed")
+
+
+@typechecked
+async def pull_access_token(connection_id: str) -> LeaseOutcome:
+    """Get a usable access token for a cloud-owned credential. This is what keeps LOCAL work going
+    once this device can no longer mint one itself."""
+    cloud = p_cloud()
+    if cloud is None:
+        return LeaseOutcome(status="not_signed_in")
+    token, base = cloud
+    try:
+        async with httpx.AsyncClient(timeout=P_TIMEOUT_S) as client:
+            r = await client.get(
+                f"{base}/api/credentials/access",
+                headers={"Authorization": f"Bearer {token}"},
+                params={"connection_id": connection_id},
+            )
+    except httpx.HTTPError as exc:
+        return LeaseOutcome(status="cloud_rejected", detail=str(exc))
+    if r.status_code != 200:
+        return LeaseOutcome(status="cloud_rejected", detail=f"HTTP {r.status_code}")
+    body = r.json()
+    ok = await credential_store.apply_to_connection(
+        connection_id, changes={"accessToken": body["access_token"]}, drop=[]
+    )
+    return LeaseOutcome(status="refreshed" if ok else "local_write_failed")

backend/apps/nine_router/credential_store.py

@@ -114,7 +114,7 @@ def list_oauth_connection_ids() -> List[str]:
 
 
 @typechecked
-async def p_request_shutdown() -> None:
+async def 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:
@@ -127,7 +127,7 @@ async def p_request_shutdown() -> None:
 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()
+    await request_shutdown()
     process.stop()
     waited = 0.0
     while waited < P_DOWN_WAIT_S:

backend/tests/test_credential_lease.py

@@ -0,0 +1,256 @@
+"""Custody handover of a rotating credential, ordered so two holders is unrepresentable.
+
+Providers issue a new refresh token on every refresh and treat a replayed one as theft, revoking
+the whole grant family. So the invariant under test is not "the happy path works", it is: at no
+point does BOTH this device and the cloud hold a token that can rotate.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_credential_lease.py -v
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List
+
+import httpx
+import pytest
+
+import backend.apps.nine_router.credential_lease as lease
+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-old",
+    "refreshToken": "refresh-live",
+    "expiresAt": "2026-08-01T00:00:00.000Z",
+    "isActive": True,
+}
+
+
+@pytest.fixture
+def p_device(tmp_path, monkeypatch):
+    """A stopped-on-demand router with one oauth connection, and a signed-in cloud identity."""
+    data_dir = tmp_path / "9router"
+    data_dir.mkdir()
+    (data_dir / "db.json").write_text(json.dumps({"providerConnections": [dict(P_CONNECTION)]}))
+    monkeypatch.setattr(process, "nine_router_data_dir", lambda: str(data_dir))
+
+    state = {"running": True}
+    monkeypatch.setattr(process, "stop", lambda: state.__setitem__("running", False))
+    monkeypatch.setattr(process, "is_running", lambda: state["running"])
+
+    async def p_ensure() -> None:
+        state["running"] = True
+
+    async def p_no_http() -> None:
+        return None
+
+    monkeypatch.setattr(process, "ensure_running", p_ensure)
+    monkeypatch.setattr(store, "request_shutdown", p_no_http)
+    monkeypatch.setattr(lease, "p_cloud", lambda: ("bearer-xyz", "https://api.example.test"))
+    return state
+
+
+def p_local() -> Dict[str, Any]:
+    db = json.loads(open(store.db_path(), encoding="utf-8").read())
+    return next(c for c in db["providerConnections"] if c["id"] == "conn-1")
+
+
+class FakeClient:
+    """Records calls and replays scripted responses; never touches the network."""
+
+    def __init__(self, script: List[Any], calls: List[Dict[str, Any]]):
+        self.script = script
+        self.calls = calls
+
+    async def __aenter__(self) -> "FakeClient":
+        return self
+
+    async def __aexit__(self, *exc: Any) -> None:
+        return None
+
+    def p_next(self, method: str, url: str, kwargs: Dict[str, Any]) -> Any:
+        self.calls.append({"method": method, "url": url, **kwargs})
+        outcome = self.script.pop(0)
+        if isinstance(outcome, Exception):
+            raise outcome
+        return outcome
+
+    async def post(self, url: str, **kwargs: Any) -> Any:
+        return self.p_next("POST", url, kwargs)
+
+    async def get(self, url: str, **kwargs: Any) -> Any:
+        return self.p_next("GET", url, kwargs)
+
+
+def p_response(status: int, body: Dict[str, Any] | None = None) -> Any:
+    return httpx.Response(status, json=body if body is not None else {})
+
+
+async def p_already_leased(harness: Dict[str, Any]) -> None:
+    """Put the device in the post-handover state: cloud owns the refresh token, device does not."""
+    harness["script"].append(p_response(200, {}))
+    await lease.lease_to_cloud("conn-1")
+    harness["script"].clear()
+    harness["calls"].clear()
+
+
+@pytest.fixture
+def p_cloud_calls(monkeypatch):
+    calls: List[Dict[str, Any]] = []
+    script: List[Any] = []
+
+    def p_factory(*args: Any, **kwargs: Any) -> FakeClient:
+        return FakeClient(script, calls)
+
+    monkeypatch.setattr(lease.httpx, "AsyncClient", p_factory)
+    return {"calls": calls, "script": script}
+
+
+@pytest.mark.asyncio
+async def test_lease_strips_locally_and_uploads(p_device, p_cloud_calls):
+    p_cloud_calls["script"].append(p_response(200, {"owner": "cloud"}))
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "leased"
+    assert "refreshToken" not in p_local(), "the device must not keep a token it could rotate"
+    assert p_local()["accessToken"] == "access-old", "the access token still has to work locally"
+    sent = p_cloud_calls["calls"][0]["json"]
+    assert sent["refresh_token"] == "refresh-live"
+    assert sent["provider"] == "claude"
+
+
+@pytest.mark.asyncio
+async def test_device_is_stripped_before_the_upload_is_attempted(p_device, p_cloud_calls, monkeypatch):
+    """The ordering IS the safety property. If the upload could run first, a failed strip would
+    leave two live rotators, which is the incident this design exists to prevent."""
+    observed: List[bool] = []
+
+    def p_factory(*args: Any, **kwargs: Any) -> FakeClient:
+        observed.append("refreshToken" in p_local())
+        return FakeClient(p_cloud_calls["script"], p_cloud_calls["calls"])
+
+    p_cloud_calls["script"].append(p_response(200, {}))
+    monkeypatch.setattr(lease.httpx, "AsyncClient", p_factory)
+
+    await lease.lease_to_cloud("conn-1")
+
+    assert observed == [False], "the local token was still present when the upload began"
+
+
+@pytest.mark.asyncio
+async def test_rejected_upload_restores_the_local_token(p_device, p_cloud_calls):
+    p_cloud_calls["script"].append(p_response(500, {}))
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "cloud_rejected"
+    assert p_local()["refreshToken"] == "refresh-live", "custody never moved, so it must come back"
+
+
+@pytest.mark.asyncio
+async def test_ambiguous_upload_asks_who_owns_it_before_restoring(p_device, p_cloud_calls):
+    """A timeout can mean the server committed anyway. Restoring blindly would recreate exactly the
+    two-holder state, so ownership is checked rather than assumed."""
+    p_cloud_calls["script"].append(httpx.ReadTimeout("boom"))
+    p_cloud_calls["script"].append(
+        p_response(200, {"leases": [{"connection_id": "conn-1", "owner": "cloud"}]})
+    )
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "leased"
+    assert "refreshToken" not in p_local(), "the cloud owns it; putting it back makes two rotators"
+
+
+@pytest.mark.asyncio
+async def test_ambiguous_upload_restores_when_the_cloud_does_not_have_it(p_device, p_cloud_calls):
+    p_cloud_calls["script"].append(httpx.ReadTimeout("boom"))
+    p_cloud_calls["script"].append(p_response(200, {"leases": []}))
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "cloud_rejected"
+    assert p_local()["refreshToken"] == "refresh-live"
+
+
+@pytest.mark.asyncio
+async def test_unknown_ownership_leaves_the_token_off_the_device(p_device, p_cloud_calls):
+    """Fail safe: when we cannot learn who owns it, the safe guess is 'not us'. Worst case the user
+    reconnects; the alternative risks revoking their whole grant."""
+    p_cloud_calls["script"].append(httpx.ReadTimeout("boom"))
+    p_cloud_calls["script"].append(httpx.ReadTimeout("also boom"))
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "ownership_unknown"
+    assert "refreshToken" not in p_local()
+
+
+@pytest.mark.asyncio
+async def test_release_brings_the_refresh_token_home(p_device, p_cloud_calls):
+    await p_already_leased(p_cloud_calls)
+    p_cloud_calls["script"].append(
+        p_response(200, {"access_token": "access-new", "refresh_token": "refresh-rotated"})
+    )
+
+    result = await lease.release_to_device("conn-1")
+
+    assert result.status == "released"
+    assert p_local()["refreshToken"] == "refresh-rotated", "must be the CURRENT token, not the old one"
+    assert p_local()["accessToken"] == "access-new"
+
+
+@pytest.mark.asyncio
+async def test_release_conflict_does_not_write_a_doomed_token(p_device, p_cloud_calls):
+    """409 means a refresh is mid-exchange. Taking that token would hand the device one the provider
+    is about to invalidate."""
+    await p_already_leased(p_cloud_calls)
+    p_cloud_calls["script"].append(p_response(409, {}))
+
+    result = await lease.release_to_device("conn-1")
+
+    assert result.status == "cloud_rejected"
+    assert "refreshToken" not in p_local()
+
+
+@pytest.mark.asyncio
+async def test_pull_access_token_updates_only_the_access_token(p_device, p_cloud_calls):
+    await p_already_leased(p_cloud_calls)
+    p_cloud_calls["script"].append(p_response(200, {"access_token": "access-fresh"}))
+
+    result = await lease.pull_access_token("conn-1")
+
+    assert result.status == "refreshed"
+    assert p_local()["accessToken"] == "access-fresh"
+    assert "refreshToken" not in p_local(), "pulling a token must never re-arm local rotation"
+
+
+@pytest.mark.asyncio
+async def test_api_key_connection_is_not_leasable(p_device, p_cloud_calls):
+    """No rotating secret means nothing to hand over, and no reason to touch the row."""
+    await p_already_leased(p_cloud_calls)
+
+    result = await lease.lease_to_cloud("conn-1")
+
+    assert result.status == "not_rotatable"
+    assert p_cloud_calls["calls"] == [], "a row with nothing to rotate must not be uploaded at all"
+
+
+@pytest.mark.asyncio
+async def test_signed_out_device_does_nothing(p_device, monkeypatch):
+    monkeypatch.setattr(lease, "p_cloud", lambda: None)
+    result = await lease.lease_to_cloud("conn-1")
+    assert result.status == "not_signed_in"
+    assert p_local()["refreshToken"] == "refresh-live"
+
+
+def test_expiry_converts_to_unix_ms():
+    assert lease.expires_ms("2026-08-01T00:00:00.000Z") == 1785542400000
+    assert lease.expires_ms("garbage") == 0, "an unreadable clock must read as expired, not valid"
+    assert lease.expires_ms(None) == 0

backend/tests/test_credential_store.py

@@ -58,7 +58,7 @@ def p_router(tmp_path, monkeypatch):
     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)
+    monkeypatch.setattr(store, "request_shutdown", p_no_http)
     return state