From 345000e8dc0c966b88106126e1c7fb9c2b1c8ceb Mon Sep 17 00:00:00 2001 From: ciregenz Date: Fri, 31 Jul 2026 13:12:11 -0700 Subject: [PATCH] [eric] credentials: hand custody to the cloud strip-first, so two rotators is unrepresentable --- backend/apps/nine_router/credential_lease.py | 209 +++++++++++++++ backend/apps/nine_router/credential_store.py | 4 +- backend/tests/test_credential_lease.py | 256 +++++++++++++++++++ backend/tests/test_credential_store.py | 2 +- 4 files changed, 468 insertions(+), 3 deletions(-) create mode 100644 backend/apps/nine_router/credential_lease.py create mode 100644 backend/tests/test_credential_lease.py diff --git a/backend/apps/nine_router/credential_lease.py b/backend/apps/nine_router/credential_lease.py new file mode 100644 index 00000000..425fb761 --- /dev/null +++ b/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") diff --git a/backend/apps/nine_router/credential_store.py b/backend/apps/nine_router/credential_store.py index ca64257d..9825a53f 100644 --- a/backend/apps/nine_router/credential_store.py +++ b/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: diff --git a/backend/tests/test_credential_lease.py b/backend/tests/test_credential_lease.py new file mode 100644 index 00000000..e10a6a4f --- /dev/null +++ b/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 diff --git a/backend/tests/test_credential_store.py b/backend/tests/test_credential_store.py index a89a58c6..df305bdf 100644 --- a/backend/tests/test_credential_store.py +++ b/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