From 51972f7e2f20d1242abcf74f7bbee694961dbc76 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 28 Jul 2026 02:00:33 -0400 Subject: [PATCH 1/4] feat(auth): add loopback pairing and credential rotation --- docs/DEVICE_AUTHENTICATION.md | 63 +++ engine/auth/__init__.py | 12 +- engine/auth/browser_pkce.py | 400 ++++++++++--------- engine/auth/lifetime.py | 135 +++++++ engine/auth/pairing.py | 214 ++++++++-- engine/auth/paste.py | 41 +- engine/auth/provider.py | 16 +- engine/auth/rotation.py | 385 ++++++++++++++++++ engine/auth/store.py | 132 ++++++ engine/cli.py | 45 +++ src/lib/types.ts | 2 +- tests/test_engine/conftest.py | 10 +- tests/test_engine/test_auth_browser_pkce.py | 421 ++++++++++++++++---- tests/test_engine/test_auth_dispatch.py | 31 ++ tests/test_engine/test_auth_pairing.py | 205 ++++++++-- tests/test_engine/test_auth_paste.py | 79 +++- tests/test_engine/test_auth_rotation.py | 365 +++++++++++++++++ tests/test_engine/test_cli.py | 39 ++ 18 files changed, 2238 insertions(+), 357 deletions(-) create mode 100644 docs/DEVICE_AUTHENTICATION.md create mode 100644 engine/auth/lifetime.py create mode 100644 engine/auth/rotation.py create mode 100644 tests/test_engine/test_auth_rotation.py diff --git a/docs/DEVICE_AUTHENTICATION.md b/docs/DEVICE_AUTHENTICATION.md new file mode 100644 index 0000000..ff6ff54 --- /dev/null +++ b/docs/DEVICE_AUTHENTICATION.md @@ -0,0 +1,63 @@ +# Desktop Cloud authentication + +OpenAdapt Desktop uses one Cloud bearer format: `oai_ingest_…`. It stores the +bearer in the operating system credential store. The token-paste path remains +available for headless systems and recovery. + +## Browser sign-in + +`openadapt-desktop login --provider browser_pkce` starts an RFC 8252 loopback +flow: + +1. Desktop binds an ephemeral `127.0.0.1` port and serves only `/callback`. +2. Desktop creates an S256 PKCE verifier and challenge plus a random `state`. +3. Desktop opens the OpenAdapt Cloud login page in the system browser. +4. After sign-in, Cloud shows the user a Connect confirmation. +5. Cloud redirects to the exact loopback URL with a five-minute, one-use + `oap_…` pairing code and the exact `state`. +6. Desktop checks the state and code prefix. It claims the pairing with the + PKCE verifier through `/api/local-bridge/pairings/claim`. +7. Desktop stages the returned `oai_ingest_…` bearer, validates it, promotes it + atomically, and confirms the connection. A failure invokes the existing + acknowledged rollback path. + +The code is not a Supabase authorization code. Desktop does not call a +Supabase token endpoint, store a Supabase session, or add a loopback URL to a +Supabase allowlist. `oapp_…` and `oaps_…` credentials belong only to the local +attended-decision portal and cannot authenticate a Cloud request. + +Browser sign-in is unavailable on a headless machine, without a display on +Linux, or without an unlocked secure credential store. Desktop checks the +store before it claims the one-use code. Token paste remains available: + +```console +openadapt-desktop login --provider paste +``` + +## Expiry and renewal + +New Cloud credentials expire within 90 days. Cloud enforces the deadline on +every authenticated request. Desktop treats its local deadline as a warning, +not as proof that a credential remains valid. + +Cloud reports the warning contract through `/api/needs-attention/count`. +Desktop checks that the JSON and the `x-openadapt-credential-*` headers agree. +Cloud sets `expiring_soon` from the exact server timestamp. The +`expires_in_days` value is a floor-rounded display value. Thus, day 14 can +report either state. Desktop uses the server `expiring_soon` value and does not +recompute it from the local clock. A legacy credential can report that it has +no expiry; the operator can rotate it to enter the 90-day policy. + +```console +openadapt-desktop credential +openadapt-desktop rotate +``` + +Rotation creates the replacement first. The old bearer stays valid for at +most seven days. Desktop durably stages the one-time response before it changes +the active keychain entries. After a process interruption, Desktop finishes +that exact stage and does not send a second rotation request. + +If the response is lost before Desktop can stage it, the old bearer remains +usable during the overlap. Cloud cannot replay the raw replacement, so the +operator must sign in again. Do not retry rotation with the same old bearer. diff --git a/engine/auth/__init__.py b/engine/auth/__init__.py index 9cf688f..e1e2703 100644 --- a/engine/auth/__init__.py +++ b/engine/auth/__init__.py @@ -72,14 +72,16 @@ def login(host: str = DEFAULT_HOST, prefer: str | None = None) -> Credential: if prefer: providers = [p for p in providers if p.name == prefer] or providers - last_error: Exception | None = None for provider in providers: if not provider.is_available(): continue try: return provider.login() - except Exception as exc: # try the next provider (e.g. browser -> paste) - last_error = exc - if last_error: - raise RuntimeError(f"Login failed: {last_error}") from last_error + except Exception as exc: + # An available provider can fail after it receives or stages a + # one-use credential. Starting another provider in the same call + # could overwrite its retained recovery state. Token paste remains + # the fallback when browser login is unavailable, and the operator + # can select it explicitly after a completed browser failure. + raise RuntimeError(f"Login failed: {exc}") from exc raise RuntimeError("No auth provider is available on this machine.") diff --git a/engine/auth/browser_pkce.py b/engine/auth/browser_pkce.py index 0bcb079..70eaf23 100644 --- a/engine/auth/browser_pkce.py +++ b/engine/auth/browser_pkce.py @@ -1,52 +1,37 @@ -"""BrowserPkceProvider -- v1 "click Login" for interactive desktop users. - -Flow (spec section 3a): - 1. Generate a PKCE verifier/challenge and bind an ephemeral loopback - listener on ``127.0.0.1:`` at path ``/callback``. - 2. Open the hosted login page in the SYSTEM browser (so Google / magic-link - "just work" with zero forked auth), passing the loopback redirect URI + - PKCE challenge. - 3. The browser redirects back to the loopback with ``?code=…``; the listener - captures it and serves a minimal "you can close this tab" page. - 4. Exchange the code for a Supabase session via PKCE (using the code - verifier), then MINT an ingest token via the hosted API. - 5. Store a single active credential whose bearer ``token`` is the minted - ingest token (so the headless push / count path always has a bearer), - carrying the Supabase ``refresh_token`` so the session can be renewed. - -``is_available()`` is False on a headless server (no browser / no loopback), so -the UI falls back to :class:`~engine.auth.paste.PasteTokenProvider`. - -NOT WIRED END TO END -- do not enable this provider by setting the Supabase -environment variables alone. Steps 2-4 above each depend on a hosted contract -that ``app.openadapt.ai`` does not implement today: - -* ``GET /login`` reads only ``checkout_session_id`` and ``next`` (clamped to a - same-origin ``/dashboard`` destination). It ignores ``redirect_to``, - ``code_challenge``, ``code_challenge_method`` and ``state``, so the browser - never navigates to the loopback listener and step 3 cannot happen. -* The Supabase ``uri_allow_list`` is rewritten on every production deploy and - contains no loopback entry, so Supabase would reject a ``127.0.0.1`` redirect - even if the page forwarded one. -* ``POST /api/ingest-tokens`` authenticates the browser session cookie and - requires a ``name`` field. It rejects a ``Authorization: Bearer `` with 401, so step 4 cannot mint. - -Because ``_exchange_code`` is only reached AFTER the loopback wait, leaving the -provider enabled makes ``login`` open a browser tab and then block for the full -``timeout`` before failing over to token paste. ``is_available()`` therefore -requires the Supabase configuration up front: the provider reports unavailable -rather than costing the operator a silent multi-minute stall. Restoring it is a -hosted-contract change, not a client change -- see the three bullets above. - -The working, shipped path to a credential is the one-time pairing flow -(:mod:`engine.auth.pairing`), which the hosted control plane does implement. - -Reconciliation note: the shared store (:mod:`engine.auth.store`) keys ONE active -credential per host and ``auth_header()`` returns exactly one bearer. We -therefore fold the Supabase session and the minted ingest token into a single -stored ``Credential`` (``kind="ingest_token"``, ``token`` = ingest token, -``refresh_token`` = Supabase refresh) rather than two competing entries. +"""BrowserPkceProvider -- "click Login" for interactive desktop users. + +An RFC 8252 native-app login. The desktop cannot keep a client secret, so it +uses the SYSTEM browser (where Google and magic-link sign-in already work) plus +an ephemeral loopback listener, and binds the exchange with S256 PKCE. + +Flow: + 1. Generate a PKCE verifier/challenge and bind a listener on + ``127.0.0.1:`` that serves ONLY ``/callback``. + 2. Open ``{host}/login`` in the system browser, passing the loopback + redirect URI, the challenge, and an opaque ``state``. + 3. The user signs in normally. OpenAdapt's own authenticated + ``/auth/loopback`` page then redirects to the listener with + ``?code=oap_...&state=...``. + 4. The listener validates ``state`` and redeems the code through the + EXISTING pairing path (:func:`engine.auth.pairing.claim_pairing`), + sending the verifier. Cloud consumes the one-use code, checks the PKCE + binding, and mints a user/org-bound ingest credential. + 5. That path stages, independently validates, atomically promotes, and + confirms the credential in the OS keychain, rolling back on any failure. + +Why the redirect comes from our own page rather than from Supabase: Supabase's +``uri_allow_list`` is rewritten by ``scripts/setup-supabase.mjs`` on every +deploy, and RFC 8252 section 7.3 requires the authorization server to accept an +arbitrary loopback port -- a promise that allow-list glob grammar cannot make. +Routing the redirect through an authenticated OpenAdapt page keeps Supabase +configuration untouched and reuses the hardened pairing redemption whole. It +also means browser login introduces NO new credential format: the ``code`` is +an ``oap_`` pairing secret, and the result is the same ``oai_ingest_...`` +credential every other path produces. + +``is_available()`` is False on a headless box (no browser, no user at the +keyboard) and False when no OS keychain can hold the result, so the UI falls +back to :class:`~engine.auth.paste.PasteTokenProvider`. """ from __future__ import annotations @@ -60,22 +45,26 @@ import urllib.parse from http.server import BaseHTTPRequestHandler, HTTPServer -import httpx from loguru import logger +from engine.auth.pairing import ( + PAIRING_SECRET_RE, + PairingError, + _validate_destination, + claim_pairing, +) from engine.auth.provider import Credential -from engine.auth.store import DEFAULT_HOST, store_credential +from engine.auth.store import DEFAULT_HOST, load_credential, secure_store_available # Hosted login page opened in the system browser. LOGIN_PATH = "/login" -# Path the loopback listener serves for the OAuth redirect. +# The ONLY path the loopback listener serves. CALLBACK_PATH = "/callback" -# Hosted endpoint that mints an ingest token from an authenticated session. -MINT_PATH = "/api/ingest-tokens" -# Supabase project config (coordinator-provisioned). Absent in CI/tests. -SUPABASE_URL_ENV = "OPENADAPT_SUPABASE_URL" -SUPABASE_ANON_KEY_ENV = "OPENADAPT_SUPABASE_ANON_KEY" +# How long to wait for the browser round trip. The user can complete a magic +# link and then explicitly choose Connect before Cloud mints the five-minute +# pairing code. No server secret exists during this wait. +DEFAULT_TIMEOUT_S = 300.0 _CALLBACK_HTML = ( b"" @@ -87,6 +76,26 @@ b"" ) +_ERROR_HTML = ( + b"" + b"OpenAdapt login failed" + b"" + b"

Sign-in did not complete.

" + b"

Return to OpenAdapt for the reason.

" + b"" +) + + +def _write_callback_body(stream, body: bytes, event: threading.Event) -> None: + """Signal callback completion even when the browser closes before reading.""" + try: + stream.write(body) + finally: + # The callback parameters were already retained. A browser disconnect + # during this courtesy page must not make login wait for the full + # timeout or discard a valid one-use code. + event.set() + def generate_pkce_pair() -> tuple[str, str]: """Return ``(code_verifier, code_challenge)`` for the S256 PKCE method.""" @@ -97,11 +106,17 @@ def generate_pkce_pair() -> tuple[str, str]: class _LoopbackReceiver: - """Ephemeral 127.0.0.1 listener that captures the OAuth ``code``.""" + """Ephemeral 127.0.0.1 listener that captures one authorization code. + + Binds ``127.0.0.1`` explicitly -- never ``0.0.0.0`` -- so nothing outside + this machine can reach it, and takes an ephemeral port because RFC 8252 + section 7.3 requires the client to choose one at request time. + """ def __init__(self) -> None: self.code: str | None = None self.state: str | None = None + self.error_code: str | None = None self.error: str | None = None self._event = threading.Event() parent = self @@ -112,21 +127,58 @@ def log_message(self, *args) -> None: # noqa: D401 - silence stdlib logging def do_GET(self) -> None: # noqa: N802 - stdlib naming parsed = urllib.parse.urlparse(self.path) - if parsed.path.rstrip("/") not in ("", CALLBACK_PATH.rstrip("/")): + # Exactly one path. A scan of the ephemeral port finds nothing + # else, and no other route can deliver a code. + if parsed.path != CALLBACK_PATH: self.send_response(404) self.end_headers() return - params = urllib.parse.parse_qs(parsed.query) - parent.code = (params.get("code") or [None])[0] - parent.state = (params.get("state") or [None])[0] - parent.error = (params.get("error") or [None])[0] + params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True) + + def single(key: str) -> str | None: + values = params.get(key) or [] + return values[0] if len(values) == 1 else None + + # First delivery wins. A later request cannot overwrite a + # captured code with a different one. + if not parent._event.is_set(): + parent.state = single("state") + code = single("code") + error_code = single("error") + description = single("error_description") + repeated = any(len(values) != 1 for values in params.values()) + success_shape = set(params) == {"code", "state"} + error_shape = set(params) in ( + {"error", "state"}, + {"error", "error_description", "state"}, + ) + safe_description = description is None or ( + len(description) <= 256 + and not any(ord(char) < 32 or ord(char) == 127 for char in description) + ) + if repeated or not safe_description: + parent.error = "The login callback was malformed." + elif success_shape and code is not None: + parent.code = code + elif error_shape and error_code is not None: + parent.error_code = error_code + parent.error = description or error_code + else: + parent.error = "The login callback was malformed." + failed = parent.error is not None or parent.code is None self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Cache-Control", "no-store") + self.send_header("Referrer-Policy", "no-referrer") self.end_headers() - self.wfile.write(_CALLBACK_HTML) - parent._event.set() + _write_callback_body( + self.wfile, + _ERROR_HTML if failed else _CALLBACK_HTML, + parent._event, + ) self._server = HTTPServer(("127.0.0.1", 0), Handler) + self._thread: threading.Thread | None = None @property def port(self) -> int: @@ -136,20 +188,34 @@ def port(self) -> int: def redirect_uri(self) -> str: return f"http://127.0.0.1:{self.port}{CALLBACK_PATH}" + def __enter__(self) -> "_LoopbackReceiver": + return self + + def __exit__(self, *exc_info) -> None: + self.close() + def serve_until_code(self, timeout: float) -> None: - thread = threading.Thread(target=self._server.serve_forever, daemon=True) - thread.start() - try: - self._event.wait(timeout=timeout) - finally: - self._server.shutdown() - self._server.server_close() + """Serve until a callback arrives or ``timeout`` elapses.""" + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + self._event.wait(timeout=timeout) def close(self) -> None: + """Tear the listener down. Safe to call twice, and on any exit path.""" + # HTTPServer.shutdown() deadlocks when serve_forever() never started, + # for example when the browser launcher itself raises. + if self._thread is not None: + try: + self._server.shutdown() + except Exception: # pragma: no cover - defensive stdlib boundary + pass try: self._server.server_close() - except Exception: + except Exception: # pragma: no cover - already closed pass + if self._thread is not None: + self._thread.join(timeout=5.0) + self._thread = None class BrowserPkceProvider: @@ -158,8 +224,6 @@ class BrowserPkceProvider: Args: host: Hosted base URL. open_browser: Callable that opens a URL in the system browser. - supabase_url: Supabase project URL (defaults to the ``OPENADAPT_SUPABASE_URL`` env). - supabase_anon_key: Supabase anon key (defaults to the env var). timeout: Seconds to wait for the browser redirect. """ @@ -169,36 +233,42 @@ def __init__( self, host: str = DEFAULT_HOST, open_browser=None, - supabase_url: str | None = None, - supabase_anon_key: str | None = None, - timeout: float = 180.0, + timeout: float = DEFAULT_TIMEOUT_S, ) -> None: self.host = host.rstrip("/") + self._uses_system_browser = open_browser is None self._open_browser = open_browser or self._default_open_browser - self._supabase_url = (supabase_url or os.environ.get(SUPABASE_URL_ENV, "")).rstrip("/") - self._supabase_anon_key = supabase_anon_key or os.environ.get(SUPABASE_ANON_KEY_ENV, "") self._timeout = timeout @staticmethod def _default_open_browser(url: str) -> None: import webbrowser - webbrowser.open(url) + if not webbrowser.open(url): + raise RuntimeError("OpenAdapt could not open the system browser.") def is_available(self) -> bool: - """False on a headless box, and False while the flow is unconfigured. - - The Supabase check is deliberately first. ``_exchange_code`` cannot run - without a project URL and anon key, but it is only reached after the - loopback wait -- so an unconfigured provider would open a browser tab - and stall for the whole ``timeout`` before failing over to token paste. - Reporting unavailable up front keeps ``login`` responsive and keeps the - provider chain honest about what it can actually complete. + """False on a headless box, and False without a usable OS keychain. + + The keychain check is not cosmetic: this flow spends a one-use server + secret, so it must refuse BEFORE claiming when the resulting credential + could not be stored securely. """ - if not (self._supabase_url and self._supabase_anon_key): - return False if os.environ.get("OPENADAPT_HEADLESS", "").strip(): return False + try: + _validate_destination(self.host, self._destination_kind()) + except PairingError: + return False + if not secure_store_available(): + return False + if self._uses_system_browser: + import webbrowser + + try: + webbrowser.get() + except webbrowser.Error: + return False if sys.platform.startswith("linux"): # No X11 / Wayland display -> no system browser to drive. return bool(os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")) @@ -206,56 +276,78 @@ def is_available(self) -> bool: return True def login(self) -> Credential: - """Run the browser-PKCE flow and store the resulting credential. + """Run the browser login and return the stored credential. Returns: - The stored ``Credential`` (bearer = minted ingest token). + The stored ``Credential`` (bearer = the minted ingest token). Raises: - RuntimeError: If the flow cannot complete (headless, denied, timeout, - or a code/token/mint failure). + RuntimeError: If the flow cannot complete (headless, denied, + timeout, state mismatch, or a claim failure). """ if not self.is_available(): - raise RuntimeError( - "Browser login is unavailable on this machine; use token paste." - ) + raise RuntimeError("Browser login is unavailable on this machine; use token paste.") verifier, challenge = generate_pkce_pair() state = secrets.token_urlsafe(24) - receiver = _LoopbackReceiver() - - auth_url = self._build_login_url(receiver.redirect_uri, challenge, state) - logger.info("Opening system browser for hosted login") - self._open_browser(auth_url) - - receiver.serve_until_code(self._timeout) - if receiver.error: - raise RuntimeError(f"Login was denied: {receiver.error}") - if not receiver.code: - raise RuntimeError("Timed out waiting for the browser login to complete.") - if receiver.state and receiver.state != state: + # `with` rather than a caller-side try/finally: the listener is closed + # on success, on failure, and on an exception raised inside the block. + with _LoopbackReceiver() as receiver: + auth_url = self._build_login_url(receiver.redirect_uri, challenge, state) + logger.info("Opening the system browser to sign in to {host}", host=self.host) + self._open_browser(auth_url) + receiver.serve_until_code(self._timeout) + code = receiver.code + received_state = receiver.state + error_code = receiver.error_code + error = receiver.error + + # Every delivered callback, including a user refusal, must bind the + # exact request. A missing state never degrades into an accepted error. + if (code is not None or error is not None) and not secrets.compare_digest( + received_state or "", state + ): raise RuntimeError("Login state mismatch (possible CSRF); aborting.") + if error: + if error_code not in {"access_denied", "server_error"}: + raise RuntimeError("Login failed because the callback was malformed.") + raise RuntimeError(f"Login was denied: {error}") + if not code: + raise RuntimeError( + "Timed out waiting for the browser login to complete. " + "Finish sign-in and choose Connect, then try again. " + "You can also paste an ingest token." + ) + # This is our one-use pairing grant, not a Supabase authorization code + # and not either runner-local portal credential. Refuse the confusion + # before any network claim can spend or disclose it. + if not PAIRING_SECRET_RE.fullmatch(code): + raise RuntimeError("Login returned an invalid OpenAdapt pairing code.") + + try: + claim_pairing( + self.host, + code, + code_verifier=verifier, + destination_kind=self._destination_kind(), + ) + except PairingError as exc: + raise RuntimeError(f"Login failed: {exc}") from exc - session = self._exchange_code(receiver.code, verifier, receiver.redirect_uri) - access_token = session["access_token"] - refresh_token = session.get("refresh_token") - expires_at = session.get("expires_at") - - ingest_token, org_id = self._mint_ingest_token(access_token) - - cred: Credential = { - "kind": "ingest_token", - "token": ingest_token, - "refresh_token": refresh_token, - "org_id": org_id, - "host": self.host, - "expires_at": expires_at, - } - store_credential(cred) - logger.info("Browser login complete; ingest token stored for {host}", host=self.host) + cred = load_credential(self.host) + if cred is None: + raise RuntimeError( + "Login completed but the credential could not be read back from the keychain." + ) + logger.info("Browser login complete; credential stored for {host}", host=self.host) return cred + def _destination_kind(self) -> str | None: + """Classify the host for the pairing destination policy.""" + hostname = urllib.parse.urlparse(self.host).hostname + return "local" if hostname in {"localhost", "127.0.0.1", "::1"} else None + def _build_login_url(self, redirect_uri: str, challenge: str, state: str) -> str: """Build the hosted login URL carrying the loopback redirect + PKCE.""" query = urllib.parse.urlencode( @@ -267,57 +359,3 @@ def _build_login_url(self, redirect_uri: str, challenge: str, state: str) -> str } ) return f"{self.host}{LOGIN_PATH}?{query}" - - def _exchange_code(self, code: str, verifier: str, redirect_uri: str) -> dict: - """Exchange the auth code for a Supabase session via PKCE. - - Raises: - RuntimeError: If Supabase is not configured or the exchange fails. - """ - if not (self._supabase_url and self._supabase_anon_key): - raise RuntimeError( - "Supabase is not configured " - f"(set {SUPABASE_URL_ENV} / {SUPABASE_ANON_KEY_ENV}); " - "browser login is unavailable." - ) - url = f"{self._supabase_url}/auth/v1/token" - try: - resp = httpx.post( - url, - params={"grant_type": "pkce"}, - headers={"apikey": self._supabase_anon_key}, - json={"auth_code": code, "code_verifier": verifier, "redirect_to": redirect_uri}, - timeout=30.0, - ) - except httpx.HTTPError as exc: - raise RuntimeError(f"Token exchange failed: {exc}") from exc - if resp.status_code >= 400: - raise RuntimeError(f"Token exchange rejected ({resp.status_code}).") - return resp.json() - - def _mint_ingest_token(self, access_token: str) -> tuple[str, str | None]: - """Mint an ingest token from an authenticated Supabase session. - - Returns: - ``(ingest_token, org_id)``. - - Raises: - RuntimeError: If minting fails. - """ - url = f"{self.host}{MINT_PATH}" - try: - resp = httpx.post( - url, - headers={"Authorization": f"Bearer {access_token}"}, - json={"label": "desktop"}, - timeout=30.0, - ) - except httpx.HTTPError as exc: - raise RuntimeError(f"Could not mint an ingest token: {exc}") from exc - if resp.status_code >= 400: - raise RuntimeError(f"Ingest-token mint rejected ({resp.status_code}).") - body = resp.json() - token = body.get("token") or body.get("ingest_token") - if not token: - raise RuntimeError("Mint response did not include an ingest token.") - return token, body.get("org_id") diff --git a/engine/auth/lifetime.py b/engine/auth/lifetime.py new file mode 100644 index 0000000..300b879 --- /dev/null +++ b/engine/auth/lifetime.py @@ -0,0 +1,135 @@ +"""Strict Cloud credential lifetime and prefix contracts. + +Cloud exposes one credential to a local client: an ``oai_ingest_`` bearer. +Pairing codes and runner-local portal credentials must never cross that role +boundary. This module also validates the additive lifetime block returned by +Cloud, including its response headers, so the Desktop never invents a safe +deadline from a partial or contradictory response. +""" + +from __future__ import annotations + +import re +from datetime import datetime +from typing import Any, Mapping + +INGEST_TOKEN_RE = re.compile(r"^oai_ingest_[A-Za-z0-9_-]{43}$") +WARNING_DAYS = 14 + +_LIFETIME_FIELDS = frozenset( + { + "expires_at", + "expires_in_days", + "expiring_soon", + "legacy_non_expiring", + "warning_days", + } +) + + +class CredentialContractError(ValueError): + """Cloud returned a malformed or contradictory credential contract.""" + + +def valid_ingest_token(value: Any) -> bool: + """Return true only for the exact Cloud ingest-bearer format.""" + return isinstance(value, str) and INGEST_TOKEN_RE.fullmatch(value) is not None + + +def _header(headers: Mapping[str, str] | Any, name: str) -> str | None: + if headers is None: + return None + try: + value = headers.get(name) + if value is None: + value = headers.get(name.lower()) + except (AttributeError, TypeError): + return None + return str(value) if value is not None else None + + +def _timestamp(value: str) -> float: + try: + parsed = datetime.fromisoformat(value[:-1] + "+00:00" if value.endswith("Z") else value) + except ValueError as exc: + raise CredentialContractError("credential expiry is not a valid timestamp") from exc + if parsed.tzinfo is None: + raise CredentialContractError("credential expiry does not include a timezone") + return parsed.timestamp() + + +def parse_credential_lifetime( + body: Any, + *, + headers: Mapping[str, str] | Any | None = None, + require_headers: bool = False, + require_fresh: bool = False, + require_no_store: bool = False, +) -> dict[str, Any]: + """Validate and normalize Cloud's ``credential`` block. + + ``require_fresh`` applies to a newly minted credential. Such a response + must report 89 or 90 remaining whole days, no expiry warning, and no legacy + exemption. ``require_headers`` applies to the needs-attention endpoint, + where the warning and remaining-day headers are part of the contract. + """ + if not isinstance(body, dict): + raise CredentialContractError("credential response is not an object") + block = body.get("credential") + if not isinstance(block, dict) or set(block) != _LIFETIME_FIELDS: + raise CredentialContractError("credential lifetime block is incomplete") + + expires_at = block["expires_at"] + days = block["expires_in_days"] + expiring = block["expiring_soon"] + legacy = block["legacy_non_expiring"] + warning_days = block["warning_days"] + + if type(expiring) is not bool or type(legacy) is not bool: + raise CredentialContractError("credential lifetime flags are invalid") + if type(warning_days) is not int or warning_days != WARNING_DAYS: + raise CredentialContractError("credential warning window is invalid") + + expires_at_timestamp: float | None + if legacy: + if expires_at is not None or days is not None or expiring: + raise CredentialContractError("legacy credential lifetime is contradictory") + expires_at_timestamp = None + else: + if not isinstance(expires_at, str) or not expires_at: + raise CredentialContractError("credential expiry is missing") + if type(days) is not int or days < 0: + raise CredentialContractError("credential remaining days are invalid") + # Cloud computes the renewal trigger from the exact server-side + # timestamp. ``expires_in_days`` is floor-rounded display data. Thus + # day 14 can report either state: false above the exact 14-day + # boundary, then true at or below it. Other whole-day values are + # unambiguous. Do not recompute this control from the client clock. + if (days < WARNING_DAYS and not expiring) or (days > WARNING_DAYS and expiring): + raise CredentialContractError("credential warning state is contradictory") + expires_at_timestamp = _timestamp(expires_at) + + if require_fresh and ( + legacy or days not in {89, 90} or expiring or expires_at_timestamp is None + ): + raise CredentialContractError("new credential does not have a fresh 90-day lifetime") + + if require_no_store and (_header(headers, "cache-control") or "").strip().lower() != "no-store": + raise CredentialContractError("credential response is not marked no-store") + + if require_headers: + if _header(headers, "x-openadapt-credential-warning-days") != str(WARNING_DAYS): + raise CredentialContractError("credential warning header is missing") + expires_header = _header(headers, "x-openadapt-credential-expires-in-days") + expected = None if days is None else str(days) + if expires_header != expected: + raise CredentialContractError("credential expiry header disagrees with its body") + + return { + "expires_at": expires_at, + "expires_at_timestamp": expires_at_timestamp, + "expires_in_days": days, + "expiring_soon": expiring, + "legacy_non_expiring": legacy, + "warning_days": warning_days, + } diff --git a/engine/auth/pairing.py b/engine/auth/pairing.py index d58fc5c..e5ece3d 100644 --- a/engine/auth/pairing.py +++ b/engine/auth/pairing.py @@ -4,6 +4,12 @@ Python engine even though the Tauri shell validates it first, so neither IPC nor an operating-system protocol invocation can become a general command, browser-navigation, or arbitrary-network surface. + +The same claim -> validate -> promote -> confirm -> rollback path also redeems +the authorization code delivered to the RFC 8252 loopback listener in +:mod:`engine.auth.browser_pkce`. That code IS an ``oap_`` pairing secret, so +browser login adds a way to OBTAIN one rather than a second protocol: see +:func:`claim_pairing`. """ from __future__ import annotations @@ -16,6 +22,11 @@ import httpx +from engine.auth.lifetime import ( + CredentialContractError, + parse_credential_lifetime, + valid_ingest_token, +) from engine.auth.provider import Credential from engine.auth.store import ( DEFAULT_HOST, @@ -30,7 +41,8 @@ ) PAIRING_SECRET_RE = re.compile(r"^oap_[A-Za-z0-9_-]{43}$") -INGEST_TOKEN_RE = re.compile(r"^oai_ingest_[A-Za-z0-9_-]{32,}$") +# RFC 7636 code verifier: 43-128 unreserved characters. +PKCE_VERIFIER_RE = re.compile(r"^[A-Za-z0-9\-._~]{43,128}$") ALLOWED_FIELDS = frozenset({"pairing", "host", "destination_kind"}) ALLOWED_DESTINATIONS = frozenset({"openadapt-managed", "local"}) API_TIMEOUT_S = 8.0 @@ -72,9 +84,7 @@ def _origin(raw: str) -> str: def _validate_destination(host: str, destination_kind: str | None) -> str: origin = _origin(host) managed_origin = _origin(DEFAULT_HOST) - kind = destination_kind or ( - "openadapt-managed" if origin == managed_origin else None - ) + kind = destination_kind or ("openadapt-managed" if origin == managed_origin else None) if kind not in ALLOWED_DESTINATIONS: raise PairingError("Connect link has an unsupported destination") if kind == "openadapt-managed": @@ -125,6 +135,19 @@ def parse_connect_uri(uri: str) -> dict[str, str]: return result +def credential_expires_at(body: Any) -> float | None: + """Read the credential deadline Cloud reports, or None when it reports none. + + Cloud is the authority on when a credential dies; this value is stored so + the local surfaces can WARN early. It is never used to decide that a + credential is still good -- every request is checked server-side. + """ + try: + return cast(float | None, parse_credential_lifetime(body)["expires_at_timestamp"]) + except CredentialContractError: + return None + + def _safe_device_name() -> str: value = re.sub(r"[\x00-\x1f\x7f]", "", socket.gethostname()).strip()[:80] return value or "this computer" @@ -158,9 +181,7 @@ def _stage_identity(stage: dict) -> tuple[str, str, str, str, Credential]: state = stage.get("state") credential = stage.get("credential") try: - canonical_pairing_id = ( - str(UUID(pairing_id)) if isinstance(pairing_id, str) else None - ) + canonical_pairing_id = str(UUID(pairing_id)) if isinstance(pairing_id, str) else None except ValueError: canonical_pairing_id = None if ( @@ -169,36 +190,30 @@ def _stage_identity(stage: dict) -> tuple[str, str, str, str, Credential]: or canonical_pairing_id != pairing_id or not isinstance(device_name, str) or not 1 <= len(device_name) <= 80 - or state not in { + or state + not in { "claimed", "canonical_written", "confirm_ambiguous", "abort_acknowledged", } or not isinstance(credential, dict) - or set(credential) - != {"kind", "token", "refresh_token", "org_id", "host", "expires_at"} + or set(credential) != {"kind", "token", "refresh_token", "org_id", "host", "expires_at"} or credential.get("kind") != "ingest_token" or credential.get("refresh_token") is not None or credential.get("org_id") is not None - or credential.get("expires_at") is not None - or not isinstance(credential.get("token"), str) - or not INGEST_TOKEN_RE.fullmatch(credential["token"]) + # Cloud now returns a deadline with the credential. Accept a number or + # None, but nothing else: a corrupt value must not reach the keychain. + or not isinstance(credential.get("expires_at"), (int, float, type(None))) + or isinstance(credential.get("expires_at"), bool) + or not valid_ingest_token(credential.get("token")) or not isinstance(credential.get("host"), str) ): - raise PairingError( - "Desktop found an invalid pairing recovery record in the keychain" - ) + raise PairingError("Desktop found an invalid pairing recovery record in the keychain") host = credential["host"] - destination = ( - "openadapt-managed" - if _origin(host) == _origin(DEFAULT_HOST) - else "local" - ) + destination = "openadapt-managed" if _origin(host) == _origin(DEFAULT_HOST) else "local" if _validate_destination(host, destination) != host: - raise PairingError( - "Desktop found an invalid pairing recovery record in the keychain" - ) + raise PairingError("Desktop found an invalid pairing recovery record in the keychain") return pairing_id, host, credential["token"], state, cast(Credential, credential) @@ -271,7 +286,7 @@ def _fail_staged_pairing(stage: dict, message: str) -> NoReturn: def _finish_staged_pairing(stage: dict) -> dict[str, Any]: - pairing_id, host, token, state, _ = _stage_identity(stage) + pairing_id, host, token, state, credential = _stage_identity(stage) device_name = cast(str, stage["device_name"]) if state == "abort_acknowledged": if restore_pairing_stage(pairing_id) and clear_pairing_stage(pairing_id): @@ -299,6 +314,24 @@ def _finish_staged_pairing(stage: dict) -> dict[str, Any]: stage, f"Cloud could not verify the new credential ({validation.status_code}).", ) + try: + validation_lifetime = parse_credential_lifetime( + validation.json(), + headers=validation.headers, + require_headers=True, + require_fresh=True, + require_no_store=True, + ) + except (AttributeError, TypeError, ValueError, CredentialContractError): + _fail_staged_pairing( + stage, + "Cloud returned an incomplete credential lifetime contract.", + ) + if validation_lifetime["expires_at_timestamp"] != credential["expires_at"]: + _fail_staged_pairing( + stage, + "Cloud returned a different credential deadline during validation.", + ) if not commit_pairing_stage(pairing_id): _fail_staged_pairing( @@ -342,7 +375,48 @@ def recover_pending_pairing() -> dict[str, Any] | None: def connect_uri(uri: str) -> dict[str, Any]: """Claim, stage, verify, commit, and confirm one Desktop pairing URI.""" request = parse_connect_uri(uri) - host = request["host"] + return claim_pairing( + request["host"], + request["pairing"], + destination_kind=request.get("destination_kind"), + ) + + +def claim_pairing( + host: str, + pairing_secret: str, + code_verifier: str | None = None, + destination_kind: str | None = None, +) -> dict[str, Any]: + """Claim, stage, verify, commit, and confirm one pairing secret. + + This is the single redemption path. A deep link supplies the secret through + ``openadapt://connect``; the RFC 8252 loopback listener in + :mod:`engine.auth.browser_pkce` receives the same kind of secret as an + authorization ``code`` and supplies the PKCE ``code_verifier`` that Cloud + bound to it. Everything after the claim -- durable staging, independent + validation, atomic keychain promotion, idempotent confirmation, and + acknowledged rollback -- is identical either way. + + Args: + host: The Cloud origin, already validated by the caller. + pairing_secret: A one-use ``oap_`` secret. + code_verifier: The PKCE verifier for a loopback grant, or None for a + deep-link pairing that carries no challenge. + + Returns: + The paired result dictionary. + + Raises: + PairingError: On any failure, with the prior connection preserved. + """ + if not PAIRING_SECRET_RE.fullmatch(str(pairing_secret)): + raise PairingError("Pairing code is malformed") + if code_verifier is not None and not PKCE_VERIFIER_RE.fullmatch(code_verifier): + raise PairingError("Login proof is malformed") + # Re-validated here, not only at the caller: every entry point to this + # redemption path must prove its destination before a secret is spent. + host = _validate_destination(host, destination_kind) if not secure_store_available(): raise PairingError( "Secure pairing needs an unlocked operating-system keychain. " @@ -354,15 +428,19 @@ def connect_uri(uri: str) -> dict[str, Any]: return recovered previous = snapshot_pairing_canonical(host) if previous is None: - raise PairingError( - "Desktop could not safely snapshot the current keychain connection" - ) + raise PairingError("Desktop could not safely snapshot the current keychain connection") device_name = _safe_device_name() + payload: dict[str, Any] = { + "pairing_secret": pairing_secret, + "device_name": device_name, + } + if code_verifier is not None: + payload["code_verifier"] = code_verifier try: claim = httpx.post( f"{host}/api/local-bridge/pairings/claim", - json={"pairing_secret": request["pairing"], "device_name": device_name}, + json=payload, timeout=API_TIMEOUT_S, follow_redirects=False, ) @@ -372,14 +450,22 @@ def connect_uri(uri: str) -> dict[str, Any]: raise PairingError("Pairing code expired, was cancelled, or was already used") if claim.status_code >= 400: raise PairingError(f"Pairing failed ({claim.status_code})") + if claim.status_code != 201: + raise PairingError("Pairing response used an unexpected success status") try: body = claim.json() - token = str(body["ingest_token"]) - pairing_id = str(UUID(str(body["pairing_id"]))) - except (KeyError, TypeError, ValueError) as exc: - raise PairingError("Pairing response did not contain a valid credential") from exc - if not INGEST_TOKEN_RE.fullmatch(token): - raise PairingError("Pairing response contained a malformed credential") + except (AttributeError, TypeError, ValueError) as exc: + raise PairingError("Pairing response was not valid JSON") from exc + try: + token, pairing_id, expiry = _validated_claim(body, claim.headers) + except PairingError: + rollback = _claim_abort_coordinates(body) + if rollback and _abort_claim(host, rollback[0], rollback[1]): + raise PairingError( + "Cloud rolled back a pairing response that did not match the " + "Desktop credential contract. Your previous connection is unchanged." + ) from None + raise credential: Credential = { "kind": "ingest_token", @@ -387,7 +473,7 @@ def connect_uri(uri: str) -> dict[str, Any]: "refresh_token": None, "org_id": None, "host": host, - "expires_at": None, + "expires_at": expiry, } if not stage_pairing_credential( pairing_id, @@ -412,3 +498,57 @@ def connect_uri(uri: str) -> dict[str, Any]: "Desktop could not recover the newly staged credential from the keychain" ) return _finish_staged_pairing(stage) + + +def _canonical_uuid(value: Any) -> str: + if not isinstance(value, str): + raise PairingError("Pairing response did not contain a valid credential") + try: + canonical = str(UUID(value)) + except ValueError as exc: + raise PairingError("Pairing response did not contain a valid credential") from exc + if canonical != value: + raise PairingError("Pairing response did not contain a valid credential") + return canonical + + +def _validated_claim(body: Any, headers: Any) -> tuple[str, str, float]: + """Validate the complete Cloud 201 response before staging its bearer.""" + expected = { + "paired", + "pairing_id", + "ingest_token_id", + "ingest_token", + "credential", + } + if not isinstance(body, dict) or set(body) != expected or body.get("paired") is not True: + raise PairingError("Pairing response did not match the credential contract") + pairing_id = _canonical_uuid(body.get("pairing_id")) + _canonical_uuid(body.get("ingest_token_id")) + token = body.get("ingest_token") + if not valid_ingest_token(token): + raise PairingError("Pairing response contained a malformed credential") + try: + lifetime = parse_credential_lifetime( + body, + headers=headers, + require_fresh=True, + require_no_store=True, + ) + except CredentialContractError as exc: + raise PairingError("Pairing response contained an invalid credential lifetime") from exc + expiry = lifetime["expires_at_timestamp"] + if not isinstance(expiry, float): + raise PairingError("Pairing response contained an invalid credential lifetime") + return token, pairing_id, expiry + + +def _claim_abort_coordinates(body: Any) -> tuple[str, str] | None: + """Salvage only the exact values needed to revoke a malformed 201.""" + if not isinstance(body, dict) or not valid_ingest_token(body.get("ingest_token")): + return None + try: + pairing_id = _canonical_uuid(body.get("pairing_id")) + except PairingError: + return None + return pairing_id, cast(str, body["ingest_token"]) diff --git a/engine/auth/paste.py b/engine/auth/paste.py index 1d0cac0..4d06a98 100644 --- a/engine/auth/paste.py +++ b/engine/auth/paste.py @@ -21,6 +21,11 @@ import httpx from loguru import logger +from engine.auth.lifetime import ( + CredentialContractError, + parse_credential_lifetime, + valid_ingest_token, +) from engine.auth.provider import Credential from engine.auth.store import ( DEFAULT_HOST, @@ -92,7 +97,13 @@ def login(self, token: str | None = None) -> Credential: if not token: raise TokenValidationError("No ingest token provided.") - org_id = self._validate(token) + if not valid_ingest_token(token): + raise TokenValidationError( + "Expected an OpenAdapt Cloud ingest token that starts with " + "oai_ingest_. A pairing code or portal credential cannot be used here." + ) + + org_id, expires_at = self._validate(token) cred: Credential = { "kind": "ingest_token", @@ -100,7 +111,7 @@ def login(self, token: str | None = None) -> Credential: "refresh_token": None, "org_id": org_id, "host": self.host, - "expires_at": None, + "expires_at": expires_at, } store_credential(cred) logger.info("Stored ingest token for {host}", host=self.host) @@ -114,8 +125,8 @@ def _prompt_for_token(self) -> str: except (EOFError, KeyboardInterrupt): return "" - def _validate(self, token: str) -> str | None: - """Validate a token via the count endpoint; return org_id if exposed. + def _validate(self, token: str) -> tuple[str | None, float | None]: + """Validate a token and retain its Cloud-reported deadline when present. Raises: TokenValidationError: on any non-2xx / network failure. @@ -130,11 +141,25 @@ def _validate(self, token: str) -> str | None: if resp.status_code == 401: raise TokenValidationError("Ingest token was rejected (401).") if resp.status_code >= 400: - raise TokenValidationError( - f"Token validation failed ({resp.status_code})." - ) + raise TokenValidationError(f"Token validation failed ({resp.status_code}).") try: body = resp.json() except ValueError: body = {} - return body.get("org_id") + expires_at: float | None = None + if isinstance(body, dict) and "credential" in body: + try: + lifetime = parse_credential_lifetime( + body, + headers=resp.headers, + require_headers=True, + require_no_store=True, + ) + expires_at = lifetime["expires_at_timestamp"] + except (AttributeError, CredentialContractError): + # A working bearer remains a valid fallback when an older Cloud + # does not yet supply the additive warning contract. Never + # convert a malformed warning into a claim that expiry is safe. + expires_at = None + org_id = body.get("org_id") if isinstance(body, dict) else None + return org_id if isinstance(org_id, str) else None, expires_at diff --git a/engine/auth/provider.py b/engine/auth/provider.py index 5d15cbe..f782db8 100644 --- a/engine/auth/provider.py +++ b/engine/auth/provider.py @@ -28,22 +28,20 @@ class Credential(TypedDict): """A stored-ready hosted credential. Both providers write this exact shape via - :func:`engine.auth.store.store_credential`. The ``kind`` discriminates - between a bare ingest token (the machine/push path) and a full Supabase - session (the interactive browser path, which additionally mints an ingest - token so the headless push/count path always has a bearer token). + :func:`engine.auth.store.store_credential`. Both browser pairing and token + paste produce the same Cloud ingest credential. A Supabase authorization + code or session is never a Desktop bearer. Attributes: - kind: ``"ingest_token"`` for an ``oai_ingest_…`` token, or - ``"supabase_session"`` for a Supabase access token. - token: The ingest token (``oai_ingest_…``) OR the Supabase access token. - refresh_token: Supabase refresh token (Supabase sessions only), else None. + kind: Always ``"ingest_token"``. + token: The ingest token (``oai_ingest_…``). + refresh_token: Reserved compatibility field. Always None. org_id: The organization the token resolves to, if known. host: The hosted base URL, e.g. ``https://app.openadapt.ai``. expires_at: POSIX timestamp when the credential expires, or None. """ - kind: Literal["ingest_token", "supabase_session"] + kind: Literal["ingest_token"] token: str refresh_token: str | None org_id: str | None diff --git a/engine/auth/rotation.py b/engine/auth/rotation.py new file mode 100644 index 0000000..41b25c0 --- /dev/null +++ b/engine/auth/rotation.py @@ -0,0 +1,385 @@ +"""Credential lifetime: report the deadline, and renew it without an outage. + +A Cloud ingest credential lives 90 days. Two things follow, and this module +owns both: + +* **Warn early.** ``GET /api/needs-attention/count`` -- the one authenticated + call this machine already makes on a timer -- returns how long the credential + it just used has left. :func:`credential_status` reads it, and + :func:`expiry_warning` turns it into a line the operator sees. +* **Renew without downtime.** ``POST /api/ingest-tokens/rotate`` mints the + replacement FIRST and only then shortens the outgoing credential to a bounded + overlap. So :func:`rotate_credential` can store the replacement knowing the + credential currently in the keychain keeps working meanwhile: if the write + fails, or the process dies between the two, the machine is still connected. + +Nothing here decides that a credential is still valid. Expiry is enforced +server-side on every request; this is a warning and a renewal, not a check. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any +from uuid import UUID + +import httpx +from loguru import logger + +from engine.auth.lifetime import ( + CredentialContractError, + parse_credential_lifetime, + valid_ingest_token, +) +from engine.auth.provider import Credential +from engine.auth.store import ( + DEFAULT_HOST, + INGEST_TOKEN_ENV, + clear_rotation_stage, + commit_rotation_stage, + load_credential, + load_rotation_stage, + secure_store_available, + snapshot_pairing_canonical, + stage_rotation_credential, +) + +COUNT_PATH = "/api/needs-attention/count" +ROTATE_PATH = "/api/ingest-tokens/rotate" +API_TIMEOUT_S = 10.0 + +# Mirrors the server's own threshold. The server sends `warning_days` with +# every answer, so this is only the fallback when it does not. +DEFAULT_WARNING_DAYS = 14 + + +class RotationError(RuntimeError): + """A safe, user-facing renewal failure that never contains a secret.""" + + +def credential_status(host: str = DEFAULT_HOST, token: str | None = None) -> dict[str, Any] | None: + """Ask Cloud how long the credential for ``host`` has left. + + Args: + host: Hosted base URL. + token: Bearer to ask about. Defaults to the stored credential. + + Returns: + The server's ``credential`` block, or None when there is no credential, + Cloud cannot be reached, or the answer is not readable. None means + "unknown", never "fine". + """ + if token is not None and not valid_ingest_token(token): + return None + bearer = token if token is not None else _stored_token(host) + if not bearer: + return None + try: + response = httpx.get( + f"{host.rstrip('/')}{COUNT_PATH}", + headers={"Authorization": f"Bearer {bearer}"}, + timeout=API_TIMEOUT_S, + follow_redirects=False, + ) + except httpx.HTTPError: + return None + if not 200 <= response.status_code < 300: + return None + try: + lifetime = parse_credential_lifetime( + response.json(), + headers=response.headers, + require_headers=True, + require_no_store=True, + ) + except (AttributeError, TypeError, ValueError, CredentialContractError): + return None + return lifetime + + +def expiry_warning(status: dict[str, Any] | None) -> str | None: + """Render the operator-visible warning, or None when there is nothing to say. + + Returns None for an unknown status too: inventing "your credential is fine" + from a failed request would be worse than saying nothing. + """ + if not status: + return None + if status.get("legacy_non_expiring") is True: + return ( + "This connection was created before connections expired, so it has " + "no renewal date. Run `openadapt-desktop rotate` to move it to a " + "90-day credential." + ) + if status.get("expiring_soon") is not True: + return None + days = status.get("expires_in_days") + if not isinstance(days, int): + return "This connection expires soon. Run `openadapt-desktop rotate` to renew it." + if days <= 0: + return "This connection expires today. Run `openadapt-desktop rotate` to renew it." + unit = "day" if days == 1 else "days" + return ( + f"This connection expires in {days} {unit}. Run `openadapt-desktop rotate` " + "to renew it; the current one keeps working while the new one takes over." + ) + + +def rotate_credential(host: str = DEFAULT_HOST) -> Credential: + """Renew the stored credential for ``host`` without interrupting this machine. + + Args: + host: Hosted base URL. + + Returns: + The newly stored ``Credential``. + + Raises: + RotationError: If there is nothing to renew, Cloud refuses, or the + replacement could not be stored. In every failure case the existing + credential is left in place and keeps working. + """ + host = host.rstrip("/") + recovered = recover_pending_rotation(host) + if recovered is not None: + return recovered + if not secure_store_available(): + raise RotationError( + "Credential renewal needs an unlocked operating-system keychain. " + "Unlock it before OpenAdapt asks Cloud for a one-time replacement." + ) + current = load_credential(host) + if current is None or not valid_ingest_token(current.get("token")): + raise RotationError( + "There is no stored connection for this workspace. " + "Run `openadapt-desktop login` to connect." + ) + previous = snapshot_pairing_canonical(host) + if previous is None: + raise RotationError( + "OpenAdapt could not safely snapshot the current keychain connection. " + "Cloud did not receive a renewal request." + ) + + try: + response = httpx.post( + f"{host}{ROTATE_PATH}", + json={}, + headers={"Authorization": f"Bearer {current['token']}"}, + timeout=API_TIMEOUT_S, + follow_redirects=False, + ) + except httpx.HTTPError as exc: + raise RotationError( + "OpenAdapt did not receive the renewal response. The old credential " + "remains valid for at most seven days if Cloud rotated it. Sign in " + "again; do not retry the same renewal." + ) from exc + + if response.status_code == 401: + raise RotationError( + "This connection is no longer valid. Run `openadapt-desktop login` to connect again." + ) + if response.status_code == 409: + raise RotationError( + "This connection was already renewed once, and the replacement " + "cannot be issued twice. Run `openadapt-desktop login` to connect again." + ) + if response.status_code != 201: + raise RotationError( + f"Renewal did not complete with the required response ({response.status_code}). " + "Keep the current credential and sign in again if Cloud already rotated it." + ) + + try: + body = response.json() + except (AttributeError, TypeError, ValueError) as exc: + raise RotationError( + "Cloud renewed the connection but returned an unreadable one-time response. " + "The old credential remains valid for at most seven days. Sign in again." + ) from exc + try: + token, previous_id, expires_at = _validated_rotation(body, response.headers) + except RotationError as exc: + raise RotationError( + "Cloud renewed the connection, but its one-time response did not " + "match the Desktop credential contract. The old credential remains " + "valid for at most seven days. Sign in again; do not retry the same renewal." + ) from exc + + replacement: Credential = { + "kind": "ingest_token", + "token": token, + "refresh_token": None, + "org_id": current.get("org_id"), + "host": host, + "expires_at": expires_at, + } + # Retain the one-time response before changing any canonical entry. A crash + # after this point resumes from the exact retained replacement and does not + # issue an unsafe second rotation request. + if not stage_rotation_credential(previous_id, replacement, previous): + raise RotationError( + "Cloud renewed the connection, but OpenAdapt could not retain its " + "one-time response. The old credential remains valid for at most " + "seven days. Sign in again; do not retry the same renewal." + ) + if not commit_rotation_stage(previous_id): + raise RotationError( + "OpenAdapt retained the renewed credential but could not finish its " + "keychain update. Recovery state remains available. Restart OpenAdapt " + "before you sign in again." + ) + if not clear_rotation_stage(previous_id): + raise RotationError( + "The renewed credential is active, but OpenAdapt could not clear its " + "recovery record. Restart OpenAdapt to finish recovery." + ) + logger.info("Renewed the hosted credential for {host}", host=host) + return replacement + + +def recover_pending_rotation(host: str | None = None) -> Credential | None: + """Promote an exact retained one-time response without another Cloud call.""" + try: + stage = load_rotation_stage() + except RuntimeError as exc: + raise RotationError( + "OpenAdapt could not read the retained credential renewal state." + ) from exc + if stage is None: + return None + previous_id, credential = _validated_rotation_stage(stage) + if host is not None and credential["host"] != host.rstrip("/"): + raise RotationError("A credential renewal for another Cloud host needs recovery first.") + if not commit_rotation_stage(previous_id): + raise RotationError( + "OpenAdapt could not safely finish the retained credential renewal. " + "Recovery state remains available." + ) + if not clear_rotation_stage(previous_id): + raise RotationError("The renewed credential is active, but its recovery record remains.") + logger.info("Recovered the renewed hosted credential for {host}", host=credential["host"]) + return credential + + +def _stored_token(host: str) -> str | None: + """Resolve the bearer this machine would actually send.""" + import os + + env_token = os.environ.get(INGEST_TOKEN_ENV, "").strip() + if valid_ingest_token(env_token): + return env_token + cred = load_credential(host.rstrip("/")) + return cred.get("token") if cred and valid_ingest_token(cred.get("token")) else None + + +def _validated_rotation(body: Any, headers: Any) -> tuple[str, str, float]: + """Validate the exact one-time response and its seven-day overlap.""" + expected = {"token", "record", "previous_id", "previous_expires_at", "credential"} + if not isinstance(body, dict) or set(body) != expected: + raise RotationError( + "Cloud renewed the connection but returned an incomplete one-time response. " + "The old credential remains valid for at most seven days. Sign in again." + ) + token = body.get("token") + previous_id = body.get("previous_id") + record = body.get("record") + if not valid_ingest_token(token): + raise RotationError("Renewal response contained a malformed credential") + if not _is_canonical_uuid(previous_id): + raise RotationError("Renewal response contained an invalid previous credential id") + credential_block = body.get("credential") + if ( + not isinstance(record, dict) + or set(record) + != { + "id", + "org_id", + "name", + "token_prefix", + "created_at", + "last_used_at", + "expires_at", + "revoked_at", + "rotated_to_id", + } + or not _is_canonical_uuid(record.get("id")) + or not isinstance(record.get("org_id"), str) + or not 1 <= len(record["org_id"]) <= 128 + or not isinstance(record.get("name"), str) + or record.get("token_prefix") != token[:20] + or record.get("last_used_at") is not None + or not isinstance(credential_block, dict) + or record.get("expires_at") != credential_block.get("expires_at") + or record.get("revoked_at") is not None + or record.get("rotated_to_id") is not None + ): + raise RotationError("Renewal response contained a contradictory credential record") + try: + _parse_timestamp(record.get("created_at")) + lifetime = parse_credential_lifetime( + body, + headers=headers, + require_fresh=True, + require_no_store=True, + ) + previous_expiry = _parse_timestamp(body.get("previous_expires_at")) + except CredentialContractError as exc: + raise RotationError("Renewal response contained an invalid credential lifetime") from exc + now = datetime.now(timezone.utc) + if previous_expiry > now + timedelta(days=7, minutes=5): + raise RotationError("Renewal response exceeded the seven-day overlap contract") + expiry = lifetime["expires_at_timestamp"] + if not isinstance(expiry, float): + raise RotationError("Renewal response contained an invalid credential lifetime") + return token, previous_id, expiry + + +def _parse_timestamp(value: Any) -> datetime: + if not isinstance(value, str) or not value: + raise CredentialContractError("credential overlap expiry is missing") + try: + parsed = datetime.fromisoformat(value[:-1] + "+00:00" if value.endswith("Z") else value) + except ValueError as exc: + raise CredentialContractError("credential overlap expiry is invalid") from exc + if parsed.tzinfo is None: + raise CredentialContractError("credential overlap expiry does not include a timezone") + return parsed.astimezone(timezone.utc) + + +def _is_canonical_uuid(value: Any) -> bool: + if not isinstance(value, str): + return False + try: + return str(UUID(value)) == value + except ValueError: + return False + + +def _validated_rotation_stage(stage: Any) -> tuple[str, Credential]: + if not isinstance(stage, dict) or set(stage) != { + "version", + "previous_id", + "credential", + "previous", + "state", + }: + raise RotationError("The retained credential renewal state is malformed") + previous_id = stage.get("previous_id") + credential = stage.get("credential") + if ( + stage.get("version") != 1 + or stage.get("state") not in {"received", "canonical_written"} + or not _is_canonical_uuid(previous_id) + or not isinstance(credential, dict) + or set(credential) != {"kind", "token", "refresh_token", "org_id", "host", "expires_at"} + or credential.get("kind") != "ingest_token" + or credential.get("refresh_token") is not None + or not valid_ingest_token(credential.get("token")) + or not isinstance(credential.get("host"), str) + or not isinstance(credential.get("expires_at"), (int, float)) + or isinstance(credential.get("expires_at"), bool) + ): + raise RotationError("The retained credential renewal state is malformed") + return previous_id, credential # type: ignore[return-value] diff --git a/engine/auth/store.py b/engine/auth/store.py index ae87ca6..5036dfe 100644 --- a/engine/auth/store.py +++ b/engine/auth/store.py @@ -49,6 +49,8 @@ _ACTIVE_HOST_ACCOUNT = "__active_host__" _PAIRING_STAGE_ACCOUNT = "__pairing_stage__" _PAIRING_STAGE_VERSION = 1 +_ROTATION_STAGE_ACCOUNT = "__rotation_stage__" +_ROTATION_STAGE_VERSION = 1 # Suffix for the companion account that holds the full Credential JSON. The bare # ``host`` account holds the RAW bearer token (what the tray reads). @@ -425,6 +427,136 @@ def clear_pairing_stage(pairing_id: str) -> bool: return _apply_exact(_keyring(), _PAIRING_STAGE_ACCOUNT, None) +def stage_rotation_credential( + previous_id: str, + credential: Credential, + previous: dict[str, str | None], +) -> bool: + """Durably retain a one-time rotation response before canonical writes.""" + kr = _keyring() + readable, current = _strict_get(kr, _ROTATION_STAGE_ACCOUNT) + if not readable or current is not None or previous.get("host") != credential["host"]: + return False + payload = json.dumps( + { + "version": _ROTATION_STAGE_VERSION, + "previous_id": previous_id, + "credential": credential, + "previous": previous, + "state": "received", + }, + sort_keys=True, + separators=(",", ":"), + ) + return _apply_exact(kr, _ROTATION_STAGE_ACCOUNT, payload) + + +def load_rotation_stage() -> dict | None: + """Load the one-time rotation response retained for crash recovery.""" + readable, raw = _strict_get(_keyring(), _ROTATION_STAGE_ACCOUNT) + if not readable: + raise RuntimeError("rotation recovery keychain entry is unreadable") + if raw is None: + return None + try: + value = json.loads(raw) + except (json.JSONDecodeError, TypeError) as exc: + raise RuntimeError("rotation recovery keychain entry is invalid") from exc + if not isinstance(value, dict): + raise RuntimeError("rotation recovery keychain entry is invalid") + return value + + +def _rotation_stage_values( + stage: dict, +) -> tuple[str, tuple[str | None, ...], tuple[str, ...]] | None: + credential = stage.get("credential") + previous = stage.get("previous") + if ( + stage.get("version") != _ROTATION_STAGE_VERSION + or stage.get("state") not in {"received", "canonical_written"} + or not isinstance(stage.get("previous_id"), str) + or not isinstance(credential, dict) + or not isinstance(previous, dict) + or not isinstance(credential.get("host"), str) + or previous.get("host") != credential["host"] + or not isinstance(credential.get("token"), str) + ): + return None + if any( + previous.get(key) is not None and not isinstance(previous.get(key), str) + for key in ("token", "credential", "active_host") + ): + return None + host = credential["host"] + prior = ( + previous.get("token"), + previous.get("credential"), + previous.get("active_host"), + ) + replacement = (credential["token"], json.dumps(credential), host) + return host, prior, replacement + + +def _mark_rotation_stage(previous_id: str, state: str) -> bool: + if state not in {"received", "canonical_written"}: + return False + try: + stage = load_rotation_stage() + except RuntimeError: + return False + if stage is None or stage.get("previous_id") != previous_id: + return False + stage["state"] = state + payload = json.dumps(stage, sort_keys=True, separators=(",", ":")) + return _apply_exact(_keyring(), _ROTATION_STAGE_ACCOUNT, payload) + + +def commit_rotation_stage(previous_id: str) -> bool: + """Finish or resume the canonical write for one retained replacement.""" + try: + stage = load_rotation_stage() + except RuntimeError: + return False + if stage is None or stage.get("previous_id") != previous_id: + return False + values = _rotation_stage_values(stage) + if values is None: + return False + host, previous, replacement = values + kr = _keyring() + accounts = (host, host + _CRED_SUFFIX, _ACTIVE_HOST_ACCOUNT) + observed: list[str | None] = [] + for account in accounts: + readable, value = _strict_get(kr, account) + if not readable: + return False + observed.append(value) + if tuple(observed) == replacement: + return _mark_rotation_stage(previous_id, "canonical_written") + if any(value not in {prior, new} for value, prior, new in zip(observed, previous, replacement)): + # Another login won after the rotation request. Preserve both the + # newer canonical value and this one-time response for operator review. + return False + if not all(_apply_exact(kr, account, value) for account, value in zip(accounts, replacement)): + # The stage remains, so the next process can finish any partial write. + return False + return _mark_rotation_stage(previous_id, "canonical_written") + + +def clear_rotation_stage(previous_id: str) -> bool: + """Delete only the exact rotation recovery record after promotion.""" + try: + stage = load_rotation_stage() + except RuntimeError: + return False + if stage is None: + return True + if stage.get("previous_id") != previous_id: + return False + return _apply_exact(_keyring(), _ROTATION_STAGE_ACCOUNT, None) + + def load_credential(host: str) -> Credential | None: """Load a stored credential for ``host``, or None if absent/unreadable. diff --git a/engine/cli.py b/engine/cli.py index b1f3412..8028f6e 100644 --- a/engine/cli.py +++ b/engine/cli.py @@ -262,6 +262,43 @@ def cmd_login(args: argparse.Namespace, engine: types.SimpleNamespace) -> None: engine.audit.log("hosted_login", host=cred["host"], kind=cred["kind"]) +def cmd_credential(args: argparse.Namespace, engine: types.SimpleNamespace) -> None: + """Show the server-authoritative credential lifetime without its bearer.""" + from engine.auth.rotation import credential_status, expiry_warning + + host = getattr(args, "host", None) or engine.config.hosted_host + status = credential_status(host) + if status is None: + print("Credential status is unavailable. Sign in again if Cloud rejects requests.") + return + warning = expiry_warning(status) + if warning: + print(warning) + return + days = status.get("expires_in_days") + if isinstance(days, int): + print(f"This connection expires in {days} days. OpenAdapt warns at 14 days.") + else: + print("This connection has no server-reported expiry date.") + + +def cmd_rotate(args: argparse.Namespace, engine: types.SimpleNamespace) -> None: + """Renew the stored Cloud credential through the seven-day overlap.""" + from engine.auth.rotation import RotationError, rotate_credential + + host = getattr(args, "host", None) or engine.config.hosted_host + try: + credential = rotate_credential(host) + except RotationError as exc: + print(f"Credential renewal failed: {exc}") + sys.exit(1) + print( + "Credential renewed. The old credential remains valid for at most " + "seven days while this computer uses the replacement." + ) + engine.audit.log("hosted_credential_rotated", host=credential["host"]) + + def cmd_push(args: argparse.Namespace, engine: types.SimpleNamespace) -> None: """Zip a recording/bundle directory and push it to /api/ingest.""" from engine import hosted @@ -711,6 +748,8 @@ def cmd_doctor(args: argparse.Namespace, engine: types.SimpleNamespace) -> None: "dismiss": cmd_dismiss, "upload": cmd_upload, "login": cmd_login, + "credential": cmd_credential, + "rotate": cmd_rotate, "push": cmd_push, "compile": cmd_compile, "replay": cmd_replay, @@ -775,6 +814,12 @@ def main(argv: list[str] | None = None) -> None: p.add_argument("--provider", default=None, choices=["paste", "browser_pkce"], help="Force an auth provider") + # credential lifetime / rotation + p = subparsers.add_parser("credential", help="Show Cloud credential lifetime") + p.add_argument("--host", default=None, help="Hosted base URL") + p = subparsers.add_parser("rotate", help="Renew the stored Cloud credential") + p.add_argument("--host", default=None, help="Hosted base URL") + # push p = subparsers.add_parser("push", help="Push a recording/bundle to /api/ingest") p.add_argument("path", nargs="?", default=None, help="Recording/bundle dir (default: latest)") diff --git a/src/lib/types.ts b/src/lib/types.ts index 3a5cf67..bc48179 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -70,7 +70,7 @@ export type StepState = export interface AuthStatus { authenticated: boolean; - kind?: "ingest_token" | "supabase_session"; + kind?: "ingest_token"; host?: string; org_id?: string | null; } diff --git a/tests/test_engine/conftest.py b/tests/test_engine/conftest.py index 4b50819..94d97b3 100644 --- a/tests/test_engine/conftest.py +++ b/tests/test_engine/conftest.py @@ -64,11 +64,17 @@ def fake_keyring(monkeypatch) -> FakeKeyring: class FakeResponse: """Minimal httpx.Response stand-in for monkeypatched requests.""" - def __init__(self, status_code: int = 200, json_body: dict | None = None, - text: str = "") -> None: + def __init__( + self, + status_code: int = 200, + json_body: dict | None = None, + text: str = "", + headers: dict[str, str] | None = None, + ) -> None: self.status_code = status_code self._json = json_body if json_body is not None else {} self.text = text + self.headers = headers or {} def json(self) -> dict: return self._json diff --git a/tests/test_engine/test_auth_browser_pkce.py b/tests/test_engine/test_auth_browser_pkce.py index 4247505..9cec28d 100644 --- a/tests/test_engine/test_auth_browser_pkce.py +++ b/tests/test_engine/test_auth_browser_pkce.py @@ -1,4 +1,4 @@ -"""Tests for BrowserPkceProvider + the loopback receiver.""" +"""End-to-end contract tests for the RFC 8252 Desktop browser login.""" from __future__ import annotations @@ -7,141 +7,402 @@ import threading import time import urllib.parse +from pathlib import Path +from uuid import uuid4 import httpx import pytest -from engine.auth import store +from engine.auth import pairing, store from engine.auth.browser_pkce import ( BrowserPkceProvider, _LoopbackReceiver, + _write_callback_body, generate_pkce_pair, ) +HOST = "https://app.openadapt.ai" +SECRET = "oap_" + "A" * 43 +TOKEN = "oai_ingest_" + "B" * 43 +EXPIRES_AT = "2026-10-26T12:00:00+00:00" +EXPIRES_TS = 1793016000.0 +REAL_HTTPX_GET = httpx.get + + +class _Response: + def __init__( + self, + status_code: int, + body: dict, + headers: dict[str, str] | None = None, + ) -> None: + self.status_code = status_code + self._body = body + self.headers = headers or {} + + def json(self) -> dict: + return self._body + + +def _lifetime() -> dict: + return { + "expires_at": EXPIRES_AT, + "expires_in_days": 90, + "expiring_soon": False, + "legacy_non_expiring": False, + "warning_days": 14, + } + + +def _claim_body(pairing_id: str) -> dict: + return { + "paired": True, + "pairing_id": pairing_id, + "ingest_token_id": str(uuid4()), + "ingest_token": TOKEN, + "credential": _lifetime(), + } + + +def _validation() -> _Response: + return _Response( + 200, + {"count": 0, "credential": _lifetime()}, + { + "cache-control": "no-store", + "x-openadapt-credential-warning-days": "14", + "x-openadapt-credential-expires-in-days": "90", + }, + ) + + +def _desktop_available(monkeypatch) -> None: + monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False) + monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin") + monkeypatch.setattr("engine.auth.browser_pkce.secure_store_available", lambda: True) + monkeypatch.setattr(pairing, "secure_store_available", lambda: True) + + +def _deliver_from_login_url( + url: str, + *, + code: str | None = SECRET, + state: str | None = None, + error: str | None = None, + include_state: bool = True, +) -> dict[str, list[str]]: + query = urllib.parse.parse_qs( + urllib.parse.urlparse(url).query, + keep_blank_values=True, + ) + redirect = query["redirect_to"][0] + callback_state = query["state"][0] if state is None else state + + def _deliver() -> None: + time.sleep(0.02) + params: dict[str, str] = {} + if code is not None: + params["code"] = code + if include_state and callback_state is not None: + params["state"] = callback_state + if error is not None: + params["error"] = error + params["error_description"] = "The user refused the connection." + REAL_HTTPX_GET(redirect, params=params, timeout=5) + + threading.Thread(target=_deliver, daemon=True).start() + return query + class TestPkce: - def test_pair_is_s256(self) -> None: + def test_pair_is_s256_and_within_rfc_7636_bounds(self) -> None: verifier, challenge = generate_pkce_pair() - expected = base64.urlsafe_b64encode( - hashlib.sha256(verifier.encode()).digest() - ).rstrip(b"=").decode() + expected = ( + base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()) + .rstrip(b"=") + .decode() + ) assert challenge == expected - assert "=" not in challenge # base64url, unpadded + assert 43 <= len(verifier) <= 128 + assert len(challenge) == 43 + assert "=" not in verifier + challenge def test_pairs_are_unique(self) -> None: assert generate_pkce_pair()[0] != generate_pkce_pair()[0] class TestLoopbackReceiver: - def test_captures_code(self) -> None: - receiver = _LoopbackReceiver() - redirect = receiver.redirect_uri - assert redirect.startswith("http://127.0.0.1:") - assert redirect.endswith("/callback") + def test_captures_only_the_exact_callback(self) -> None: + with _LoopbackReceiver() as receiver: + redirect = receiver.redirect_uri + assert redirect.startswith("http://127.0.0.1:") + assert redirect.endswith("/callback") + + def _deliver() -> None: + time.sleep(0.02) + wrong = redirect.replace("/callback", "/") + assert httpx.get(wrong, timeout=5).status_code == 404 + response = httpx.get( + redirect, + params={"code": SECRET, "state": "state_1234567890"}, + timeout=5, + ) + assert response.headers["cache-control"] == "no-store" + assert response.headers["referrer-policy"] == "no-referrer" - def _deliver(): - time.sleep(0.1) - httpx.get(redirect, params={"code": "abc", "state": "s1"}, timeout=5) + threading.Thread(target=_deliver, daemon=True).start() + receiver.serve_until_code(timeout=5) + assert receiver.code == SECRET + assert receiver.state == "state_1234567890" + + def test_close_before_server_start_does_not_call_blocking_shutdown(self) -> None: + receiver = _LoopbackReceiver() + receiver.close() + receiver.close() - threading.Thread(target=_deliver, daemon=True).start() - receiver.serve_until_code(timeout=5) - assert receiver.code == "abc" - assert receiver.state == "s1" + def test_browser_disconnect_after_callback_still_signals_completion(self) -> None: + event = threading.Event() + class _DisconnectedStream: + def write(self, body: bytes) -> None: + raise BrokenPipeError("browser closed") -def _configured(**kwargs) -> BrowserPkceProvider: - """A provider whose Supabase project is configured, as in a wired deployment.""" - kwargs.setdefault("supabase_url", "https://project.supabase.co") - kwargs.setdefault("supabase_anon_key", "anon_key") - return BrowserPkceProvider(**kwargs) + with pytest.raises(BrokenPipeError): + _write_callback_body(_DisconnectedStream(), b"done", event) + assert event.is_set() class TestIsAvailable: def test_headless_env_false(self, monkeypatch) -> None: monkeypatch.setenv("OPENADAPT_HEADLESS", "1") - assert _configured().is_available() is False + monkeypatch.setattr("engine.auth.browser_pkce.secure_store_available", lambda: True) + assert BrowserPkceProvider().is_available() is False def test_linux_without_display_false(self, monkeypatch) -> None: monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False) monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "linux") + monkeypatch.setattr("engine.auth.browser_pkce.secure_store_available", lambda: True) monkeypatch.delenv("DISPLAY", raising=False) monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) - assert _configured().is_available() is False - - def test_macos_true_when_configured(self, monkeypatch) -> None: - monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False) - monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin") - assert _configured().is_available() is True - - def test_unconfigured_is_unavailable_even_on_a_desktop(self, monkeypatch) -> None: - """An unconfigured provider must not lead the chain and stall the login. + assert BrowserPkceProvider().is_available() is False - Without a Supabase project the exchange cannot succeed, but it is only - attempted after the loopback wait -- so reporting available here would - open a browser tab and block for the full timeout before falling back - to token paste. Regression guard for that stall. - """ - monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False) + def test_complete_cloud_contract_needs_no_supabase_client_configuration( + self, monkeypatch + ) -> None: + _desktop_available(monkeypatch) monkeypatch.delenv("OPENADAPT_SUPABASE_URL", raising=False) monkeypatch.delenv("OPENADAPT_SUPABASE_ANON_KEY", raising=False) - monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin") - assert BrowserPkceProvider().is_available() is False + assert BrowserPkceProvider(open_browser=lambda _: None).is_available() is True + source = Path("engine/auth/browser_pkce.py").read_text() + assert "/auth/v1/token" not in source + assert "OPENADAPT_SUPABASE_" not in source - def test_partial_configuration_is_unavailable(self, monkeypatch) -> None: - monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False) - monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin") - url_only = BrowserPkceProvider( - supabase_url="https://project.supabase.co", supabase_anon_key="" + def test_missing_system_browser_is_unavailable(self, monkeypatch) -> None: + import webbrowser + + _desktop_available(monkeypatch) + monkeypatch.setattr( + webbrowser, + "get", + lambda: (_ for _ in ()).throw(webbrowser.Error("no browser")), ) - key_only = BrowserPkceProvider(supabase_url="", supabase_anon_key="anon_key") - assert url_only.is_available() is False - assert key_only.is_available() is False + assert BrowserPkceProvider().is_available() is False - def test_unconfigured_provider_never_opens_a_browser(self, monkeypatch) -> None: - """The stall is user-visible as a stray tab; assert none is opened.""" + def test_locked_keychain_refuses_before_opening_browser(self, monkeypatch) -> None: monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False) - monkeypatch.delenv("OPENADAPT_SUPABASE_URL", raising=False) - monkeypatch.delenv("OPENADAPT_SUPABASE_ANON_KEY", raising=False) monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin") + monkeypatch.setattr("engine.auth.browser_pkce.secure_store_available", lambda: False) opened: list[str] = [] - provider = BrowserPkceProvider(open_browser=opened.append) + with pytest.raises(RuntimeError, match="unavailable"): + BrowserPkceProvider(open_browser=opened.append).login() + assert opened == [] + + def test_untrusted_cloud_origin_refuses_before_opening_browser(self, monkeypatch) -> None: + _desktop_available(monkeypatch) + opened: list[str] = [] + provider = BrowserPkceProvider( + host="https://app.openadapt.ai.evil.example", + open_browser=opened.append, + ) with pytest.raises(RuntimeError, match="unavailable"): provider.login() assert opened == [] class TestLogin: - def test_full_flow(self, fake_keyring, monkeypatch) -> None: - monkeypatch.delenv("OPENADAPT_HEADLESS", raising=False) - monkeypatch.setattr("engine.auth.browser_pkce.sys.platform", "darwin") + def test_full_flow_uses_one_oap_claim_and_the_shared_durable_pipeline( + self, fake_keyring, monkeypatch + ) -> None: + _desktop_available(monkeypatch) + pairing_id = str(uuid4()) + events: list[str] = [] + sent_challenge: list[str] = [] + real_snapshot = pairing.snapshot_pairing_canonical + real_stage = pairing.stage_pairing_credential + real_commit = pairing.commit_pairing_stage + real_clear = pairing.clear_pairing_stage def _open_browser(url: str) -> None: - # Extract the loopback redirect and deliver a code asynchronously. - qs = urllib.parse.parse_qs(urllib.parse.urlparse(url).query) - redirect = qs["redirect_to"][0] - state = qs["state"][0] + query = _deliver_from_login_url(url) + assert set(query) == { + "redirect_to", + "code_challenge", + "code_challenge_method", + "state", + } + assert all(len(values) == 1 for values in query.values()) + assert query["code_challenge_method"] == ["S256"] + assert len(query["code_challenge"][0]) == 43 + sent_challenge.append(query["code_challenge"][0]) + assert 16 <= len(query["state"][0]) <= 128 - def _deliver(): - time.sleep(0.1) - httpx.get(redirect, params={"code": "auth_code_1", "state": state}, timeout=5) + def _post(url, *, json, headers=None, **kwargs): + if url.endswith("/claim"): + events.append("claim") + assert headers is None + assert set(json) == {"pairing_secret", "device_name", "code_verifier"} + assert json["pairing_secret"] == SECRET + assert 43 <= len(json["code_verifier"]) <= 128 + challenge = ( + base64.urlsafe_b64encode( + hashlib.sha256(json["code_verifier"].encode()).digest() + ) + .rstrip(b"=") + .decode() + ) + # It is the verifier for the challenge sent in the login URL. + assert sent_challenge == [challenge] + return _Response( + 201, + _claim_body(pairing_id), + {"cache-control": "no-store", "referrer-policy": "no-referrer"}, + ) + events.append("confirm") + assert url.endswith("/confirm") + assert json == {"pairing_id": pairing_id} + assert headers == {"Authorization": f"Bearer {TOKEN}"} + return _Response(200, {"connected": True}) - threading.Thread(target=_deliver, daemon=True).start() + def _snapshot(host): + events.append("snapshot") + return real_snapshot(host) + + def _stage(*args): + events.append("stage") + return real_stage(*args) + + def _validate(*args, **kwargs): + events.append("validate") + return _validation() + + def _commit(value): + events.append("commit") + return real_commit(value) + + def _clear(value): + events.append("clear") + return real_clear(value) + + monkeypatch.setattr(pairing, "_safe_device_name", lambda: "test-device") + monkeypatch.setattr(pairing, "snapshot_pairing_canonical", _snapshot) + monkeypatch.setattr(pairing, "stage_pairing_credential", _stage) + monkeypatch.setattr(pairing, "commit_pairing_stage", _commit) + monkeypatch.setattr(pairing, "clear_pairing_stage", _clear) + monkeypatch.setattr(pairing.httpx, "post", _post) + monkeypatch.setattr(pairing.httpx, "get", _validate) + + credential = BrowserPkceProvider(host=HOST, open_browser=_open_browser).login() - provider = _configured(host="https://app.openadapt.ai", open_browser=_open_browser) - provider._exchange_code = lambda code, verifier, redirect_uri: { # type: ignore[assignment] - "access_token": "supabase_access", - "refresh_token": "supabase_refresh", - "expires_at": 1234.0, + assert credential == { + "kind": "ingest_token", + "token": TOKEN, + "refresh_token": None, + "org_id": None, + "host": HOST, + "expires_at": EXPIRES_TS, } - provider._mint_ingest_token = lambda access_token: ("oai_ingest_minted", "org_9") # type: ignore[assignment] - - cred = provider.login() - assert cred["kind"] == "ingest_token" - assert cred["token"] == "oai_ingest_minted" - assert cred["refresh_token"] == "supabase_refresh" - assert cred["org_id"] == "org_9" - # The bearer path resolves the minted ingest token. - assert store.auth_header() == {"Authorization": "Bearer oai_ingest_minted"} + assert store.auth_header() == {"Authorization": f"Bearer {TOKEN}"} + assert events == [ + "snapshot", + "claim", + "stage", + "validate", + "commit", + "confirm", + "clear", + ] + + @pytest.mark.parametrize( + "wrong_code", + [ + "supabase_authorization_code", + "oai_ingest_" + "A" * 43, + "oapp_" + "A" * 43, + "oaps_" + "A" * 43, + ], + ) + def test_rejects_supabase_and_other_credential_prefixes_before_claim( + self, wrong_code, monkeypatch + ) -> None: + _desktop_available(monkeypatch) + claimed = False + + def _post(*args, **kwargs): + nonlocal claimed + claimed = True + raise AssertionError("a wrong credential role must not reach claim") + + monkeypatch.setattr(pairing.httpx, "post", _post) + provider = BrowserPkceProvider( + open_browser=lambda url: _deliver_from_login_url(url, code=wrong_code) + ) + with pytest.raises(RuntimeError, match="invalid OpenAdapt pairing code"): + provider.login() + assert claimed is False + + @pytest.mark.parametrize("callback_state", ["", "wrong_state_123456"]) + def test_requires_present_exact_state(self, callback_state, monkeypatch) -> None: + _desktop_available(monkeypatch) + provider = BrowserPkceProvider( + open_browser=lambda url: _deliver_from_login_url( + url, + state=callback_state, + ) + ) + with pytest.raises(RuntimeError, match="state mismatch"): + provider.login() + + def test_access_denied_stops_without_a_claim(self, monkeypatch) -> None: + _desktop_available(monkeypatch) + provider = BrowserPkceProvider( + open_browser=lambda url: _deliver_from_login_url( + url, + code=None, + error="access_denied", + ) + ) + with pytest.raises(RuntimeError, match="user refused"): + provider.login() + + def test_access_denied_without_state_is_rejected_as_a_mismatch(self, monkeypatch) -> None: + _desktop_available(monkeypatch) + provider = BrowserPkceProvider( + open_browser=lambda url: _deliver_from_login_url( + url, + code=None, + error="access_denied", + include_state=False, + ) + ) + with pytest.raises(RuntimeError, match="state mismatch"): + provider.login() + + def test_pairing_410_is_a_safe_login_failure(self, monkeypatch) -> None: + _desktop_available(monkeypatch) + monkeypatch.setattr(pairing.httpx, "post", lambda *a, **k: _Response(410, {})) + provider = BrowserPkceProvider(open_browser=lambda url: _deliver_from_login_url(url)) + with pytest.raises(RuntimeError, match="expired.*already used"): + provider.login() def test_headless_login_raises(self, monkeypatch) -> None: monkeypatch.setenv("OPENADAPT_HEADLESS", "1") diff --git a/tests/test_engine/test_auth_dispatch.py b/tests/test_engine/test_auth_dispatch.py index 9a7907e..393c5f8 100644 --- a/tests/test_engine/test_auth_dispatch.py +++ b/tests/test_engine/test_auth_dispatch.py @@ -69,6 +69,37 @@ def login(self): auth.login(prefer="paste") assert called == ["paste"] + def test_available_browser_failure_does_not_overwrite_recovery_with_paste( + self, monkeypatch + ) -> None: + called = [] + + class _Browser: + name = "browser_pkce" + + def is_available(self): + return True + + def login(self): + called.append("browser") + raise RuntimeError("retained one-use credential needs recovery") + + class _Paste: + name = "paste" + + def is_available(self): + return True + + def login(self): + called.append("paste") + return {} + + monkeypatch.setattr(auth, "available_providers", lambda host="": [_Browser(), _Paste()]) + + with pytest.raises(RuntimeError, match="needs recovery"): + auth.login() + assert called == ["browser"] + def test_no_provider_raises(self, monkeypatch) -> None: class _Unavailable: name = "browser_pkce" diff --git a/tests/test_engine/test_auth_pairing.py b/tests/test_engine/test_auth_pairing.py index 96af644..3bdea5a 100644 --- a/tests/test_engine/test_auth_pairing.py +++ b/tests/test_engine/test_auth_pairing.py @@ -9,23 +9,67 @@ from engine.auth import pairing, store SECRET = "oap_" + "A" * 43 -TOKEN = "oai_ingest_" + "B" * 32 -OLD_TOKEN = "oai_ingest_" + "C" * 32 -OTHER_TOKEN = "oai_ingest_" + "D" * 32 +TOKEN = "oai_ingest_" + "B" * 43 +OLD_TOKEN = "oai_ingest_" + "C" * 43 +OTHER_TOKEN = "oai_ingest_" + "D" * 43 HOST = "https://app.openadapt.ai" OTHER_HOST = "https://previous.example" VALID_URI = f"openadapt://connect?pairing={SECRET}&host=https%3A%2F%2Fapp.openadapt.ai" +EXPIRES_AT = "2026-10-26T12:00:00+00:00" +EXPIRES_TS = 1793016000.0 class _Response: - def __init__(self, status_code: int, body: dict) -> None: + def __init__( + self, + status_code: int, + body: dict, + headers: dict[str, str] | None = None, + ) -> None: self.status_code = status_code self._body = body + self.headers = headers or {} def json(self) -> dict: return self._body +def _credential_lifetime() -> dict: + return { + "expires_at": EXPIRES_AT, + "expires_in_days": 90, + "expiring_soon": False, + "legacy_non_expiring": False, + "warning_days": 14, + } + + +def _claim_response(pairing_id: str) -> _Response: + return _Response( + 201, + { + "paired": True, + "pairing_id": pairing_id, + "ingest_token_id": str(uuid4()), + "ingest_token": TOKEN, + "credential": _credential_lifetime(), + }, + {"cache-control": "no-store", "referrer-policy": "no-referrer"}, + ) + + +def _validation_response(status: int = 200) -> _Response: + return _Response( + status, + {"count": 0, "credential": _credential_lifetime()}, + { + "cache-control": "no-store", + "x-openadapt-credential-warning-days": "14", + "x-openadapt-credential-expires-in-days": "90", + }, + ) + + def test_parser_accepts_only_the_fixed_connect_action() -> None: assert pairing.parse_connect_uri(VALID_URI) == { "pairing": SECRET, @@ -123,7 +167,7 @@ def _post(url, *, json, headers=None, **kwargs): events.append("claim") assert json == {"pairing_secret": SECRET, "device_name": "test-device"} assert headers is None - return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id}) + return _claim_response(pairing_id) assert not url.endswith("/abort") events.append("confirm") assert url.endswith("/confirm") @@ -143,7 +187,7 @@ def _get(url, *, headers, **kwargs): events.append("validate") assert url == f"{HOST}/api/needs-attention/count" assert headers == {"Authorization": f"Bearer {TOKEN}"} - return _Response(200, {"count": 0}) + return _validation_response() def _commit(value): events.append("commit") @@ -167,7 +211,7 @@ def _clear(value): assert result["authenticated"] is True assert result["host"] == HOST assert "token" not in result - assert store.load_credential(HOST) == _credential(HOST, TOKEN) + assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS) assert store.load_pairing_stage() is None assert events == [ "snapshot", @@ -180,14 +224,14 @@ def _clear(value): ] -def _credential(host: str, token: str) -> dict: +def _credential(host: str, token: str, expires_at: float | None = None) -> dict: return { "kind": "ingest_token", "token": token, "refresh_token": None, "org_id": None, "host": host, - "expires_at": None, + "expires_at": expires_at, } @@ -202,7 +246,7 @@ def test_rejected_verification_aborts_new_claim_without_touching_prior_state( def _post(url, *, json, headers=None, **kwargs): posts.append((url, json, headers or {})) if url.endswith("/claim"): - return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id}) + return _claim_response(pairing_id) assert url.endswith("/abort") return _Response(200, {"revoked": True}) @@ -239,7 +283,7 @@ def test_confirm_ambiguity_preserves_new_canonical_and_recovers_idempotently( def _post(url, *, json, headers=None, **kwargs): endpoints.append(url.rsplit("/", 1)[-1]) if url.endswith("/claim"): - return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id}) + return _claim_response(pairing_id) if url.endswith("/confirm"): if confirm_ambiguous: return _Response(503, {}) @@ -248,11 +292,11 @@ def _post(url, *, json, headers=None, **kwargs): monkeypatch.setattr(pairing, "secure_store_available", lambda: True) monkeypatch.setattr(pairing.httpx, "post", _post) - monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {})) + monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response()) with pytest.raises(pairing.PairingError, match="confirmation.*uncertain"): pairing.connect_uri(VALID_URI) - assert store.load_credential(HOST) == _credential(HOST, TOKEN) + assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS) assert store.active_host() == HOST assert store.load_pairing_stage()["state"] == "confirm_ambiguous" assert endpoints == ["claim", "confirm", "confirm"] @@ -260,7 +304,7 @@ def _post(url, *, json, headers=None, **kwargs): confirm_ambiguous = False recovered = pairing.recover_pending_pairing() assert recovered is not None and recovered["authenticated"] is True - assert store.load_credential(HOST) == _credential(HOST, TOKEN) + assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS) assert store.load_pairing_stage() is None assert endpoints[-1] == "confirm" @@ -274,7 +318,7 @@ def test_crash_after_canonical_write_recovers_without_a_second_claim( assert previous is not None assert store.stage_pairing_credential( pairing_id, - _credential(HOST, TOKEN), + _credential(HOST, TOKEN, EXPIRES_TS), previous, "crash-device", ) @@ -293,13 +337,13 @@ def _post(url, **kwargs): return _Response(200, {"connected": True}) monkeypatch.setattr(pairing.httpx, "post", _post) - monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {})) + monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response()) recovered = pairing.recover_pending_pairing() assert recovered is not None assert recovered["device_name"] == "crash-device" assert posts == ["confirm"] - assert store.load_credential(HOST) == _credential(HOST, TOKEN) + assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS) assert store.load_pairing_stage() is None @@ -315,14 +359,14 @@ def _post(url, **kwargs): endpoint = url.rsplit("/", 1)[-1] endpoints.append(endpoint) if endpoint == "claim": - return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id}) + return _claim_response(pairing_id) if endpoint == "confirm": return _Response(409, {}) return _Response(200, {"revoked": True}) monkeypatch.setattr(pairing, "secure_store_available", lambda: True) monkeypatch.setattr(pairing.httpx, "post", _post) - monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {})) + monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response()) with pytest.raises(pairing.PairingError, match="previous Desktop connection is unchanged"): pairing.connect_uri(VALID_URI) @@ -341,16 +385,16 @@ def test_abort_409_never_blindly_restores_over_possibly_confirmed_token( def _post(url, **kwargs): if url.endswith("/claim"): - return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id}) + return _claim_response(pairing_id) return _Response(409, {}) monkeypatch.setattr(pairing, "secure_store_available", lambda: True) monkeypatch.setattr(pairing.httpx, "post", _post) - monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {})) + monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response()) with pytest.raises(pairing.PairingError, match="preserved.*safe recovery"): pairing.connect_uri(VALID_URI) - assert store.load_credential(HOST) == _credential(HOST, TOKEN) + assert store.load_credential(HOST) == _credential(HOST, TOKEN, EXPIRES_TS) assert store.active_host() == HOST assert store.load_pairing_stage() is not None @@ -375,13 +419,13 @@ def _fail_new_companion(service, account, value): def _post(url, *, json, headers=None, **kwargs): endpoints.append(url.rsplit("/", 1)[-1]) if url.endswith("/claim"): - return _Response(200, {"ingest_token": TOKEN, "pairing_id": pairing_id}) + return _claim_response(pairing_id) assert url.endswith("/abort") return _Response(200, {"revoked": True}) monkeypatch.setattr(pairing, "secure_store_available", lambda: True) monkeypatch.setattr(pairing.httpx, "post", _post) - monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _Response(200, {})) + monkeypatch.setattr(pairing.httpx, "get", lambda *args, **kwargs: _validation_response()) with pytest.raises(pairing.PairingError, match="atomically write"): pairing.connect_uri(VALID_URI) @@ -390,3 +434,116 @@ def _post(url, *, json, headers=None, **kwargs): assert store.active_host() == OTHER_HOST assert store.load_pairing_stage() is None assert endpoints == ["claim", "abort"] + + +def test_pkce_claim_sends_the_exact_verifier_and_maps_410(monkeypatch) -> None: + verifier = "V" * 43 + requests: list[dict] = [] + + def _post(url, *, json, headers=None, **kwargs): + requests.append(json) + assert url == f"{HOST}/api/local-bridge/pairings/claim" + assert headers is None + return _Response(410, {}) + + monkeypatch.setattr(pairing, "secure_store_available", lambda: True) + monkeypatch.setattr(pairing, "_safe_device_name", lambda: "test-device") + monkeypatch.setattr(pairing.httpx, "post", _post) + + with pytest.raises(pairing.PairingError, match="expired.*already used"): + pairing.claim_pairing(HOST, SECRET, code_verifier=verifier) + assert requests == [ + { + "pairing_secret": SECRET, + "device_name": "test-device", + "code_verifier": verifier, + } + ] + + +@pytest.mark.parametrize( + "wrong_role", + [ + "supabase_authorization_code", + "oai_ingest_" + "A" * 43, + "oapp_" + "A" * 43, + "oaps_" + "A" * 43, + ], +) +def test_claim_rejects_every_non_pairing_credential_before_network(wrong_role, monkeypatch) -> None: + monkeypatch.setattr(pairing, "secure_store_available", lambda: True) + monkeypatch.setattr( + pairing.httpx, + "post", + lambda *a, **k: pytest.fail("wrong credential role reached the claim route"), + ) + with pytest.raises(pairing.PairingError, match="malformed"): + pairing.claim_pairing(HOST, wrong_role, code_verifier="V" * 43) + + +def test_malformed_201_is_aborted_when_the_rollback_identity_is_available( + monkeypatch, +) -> None: + pairing_id = str(uuid4()) + endpoints: list[str] = [] + malformed = _claim_response(pairing_id)._body + malformed.pop("ingest_token_id") + + def _post(url, **kwargs): + endpoints.append(url.rsplit("/", 1)[-1]) + if url.endswith("/claim"): + return _Response(201, malformed) + assert kwargs["headers"] == {"Authorization": f"Bearer {TOKEN}"} + return _Response(200, {"revoked": True}) + + monkeypatch.setattr(pairing, "secure_store_available", lambda: True) + monkeypatch.setattr(pairing.httpx, "post", _post) + with pytest.raises(pairing.PairingError, match="rolled back.*previous connection"): + pairing.claim_pairing(HOST, SECRET) + assert endpoints == ["claim", "abort"] + assert store.load_credential(HOST) is None + assert store.load_pairing_stage() is None + + +def test_claim_without_no_store_is_aborted(monkeypatch) -> None: + pairing_id = str(uuid4()) + endpoints: list[str] = [] + claim = _claim_response(pairing_id) + claim.headers = {} + + def _post(url, **kwargs): + endpoints.append(url.rsplit("/", 1)[-1]) + if url.endswith("/claim"): + return claim + return _Response(200, {"revoked": True}) + + monkeypatch.setattr(pairing, "secure_store_available", lambda: True) + monkeypatch.setattr(pairing.httpx, "post", _post) + with pytest.raises(pairing.PairingError, match="rolled back"): + pairing.claim_pairing(HOST, SECRET) + assert endpoints == ["claim", "abort"] + assert store.load_credential(HOST) is None + + +def test_validation_requires_matching_14_day_headers_and_rolls_back(monkeypatch) -> None: + pairing_id = str(uuid4()) + endpoints: list[str] = [] + + def _post(url, **kwargs): + endpoints.append(url.rsplit("/", 1)[-1]) + if url.endswith("/claim"): + return _claim_response(pairing_id) + if url.endswith("/abort"): + return _Response(200, {"revoked": True}) + raise AssertionError("a malformed validation must not confirm") + + invalid_headers = _validation_response() + invalid_headers.headers["x-openadapt-credential-warning-days"] = "30" + monkeypatch.setattr(pairing, "secure_store_available", lambda: True) + monkeypatch.setattr(pairing.httpx, "post", _post) + monkeypatch.setattr(pairing.httpx, "get", lambda *a, **k: invalid_headers) + + with pytest.raises(pairing.PairingError, match="lifetime contract"): + pairing.claim_pairing(HOST, SECRET) + assert endpoints == ["claim", "abort"] + assert store.load_credential(HOST) is None diff --git a/tests/test_engine/test_auth_paste.py b/tests/test_engine/test_auth_paste.py index 55c9bfe..3072cd0 100644 --- a/tests/test_engine/test_auth_paste.py +++ b/tests/test_engine/test_auth_paste.py @@ -10,6 +10,33 @@ from .conftest import FakeResponse +TOKEN = "oai_ingest_" + "A" * 43 +ENV_TOKEN = "oai_ingest_" + "B" * 43 +PASTED_TOKEN = "oai_ingest_" + "C" * 43 +EXPIRES_AT = "2026-10-26T12:00:00+00:00" +EXPIRES_TS = 1793016000.0 + + +def _lifetime_response(org_id: str = "org_42") -> FakeResponse: + return FakeResponse( + 200, + { + "org_id": org_id, + "credential": { + "expires_at": EXPIRES_AT, + "expires_in_days": 90, + "expiring_soon": False, + "legacy_non_expiring": False, + "warning_days": 14, + }, + }, + headers={ + "cache-control": "no-store", + "x-openadapt-credential-warning-days": "14", + "x-openadapt-credential-expires-in-days": "90", + }, + ) + class TestPasteTokenProvider: def test_is_available_always_true(self) -> None: @@ -25,18 +52,19 @@ def test_settings_url(self) -> None: def test_login_with_explicit_token(self, fake_keyring, monkeypatch) -> None: monkeypatch.setattr( "engine.auth.paste.httpx.get", - lambda *a, **k: FakeResponse(200, {"org_id": "org_42"}), + lambda *a, **k: _lifetime_response(), ) provider = PasteTokenProvider(host="https://app.openadapt.ai") - cred = provider.login(token="oai_ingest_xyz") + cred = provider.login(token=TOKEN) assert cred["kind"] == "ingest_token" - assert cred["token"] == "oai_ingest_xyz" + assert cred["token"] == TOKEN assert cred["org_id"] == "org_42" + assert cred["expires_at"] == EXPIRES_TS # Persisted + resolvable via auth_header. - assert store.auth_header() == {"Authorization": "Bearer oai_ingest_xyz"} + assert store.auth_header() == {"Authorization": f"Bearer {TOKEN}"} def test_login_reads_env_when_headless(self, fake_keyring, monkeypatch) -> None: - monkeypatch.setenv("OPENADAPT_INGEST_TOKEN", "oai_ingest_env") + monkeypatch.setenv("OPENADAPT_INGEST_TOKEN", ENV_TOKEN) monkeypatch.setattr( "engine.auth.paste.httpx.get", lambda *a, **k: FakeResponse(200, {"org_id": "org_env"}), @@ -46,15 +74,27 @@ def _no_prompt(_): raise AssertionError("should not prompt when env is set") cred = PasteTokenProvider(prompt=_no_prompt).login() - assert cred["token"] == "oai_ingest_env" + assert cred["token"] == ENV_TOKEN def test_login_prompts_interactively(self, fake_keyring, monkeypatch) -> None: monkeypatch.setattr( "engine.auth.paste.httpx.get", lambda *a, **k: FakeResponse(200, {}), ) - cred = PasteTokenProvider(prompt=lambda _: " oai_ingest_pasted ").login() - assert cred["token"] == "oai_ingest_pasted" + cred = PasteTokenProvider(prompt=lambda _: f" {PASTED_TOKEN} ").login() + assert cred["token"] == PASTED_TOKEN + + def test_working_token_fallback_keeps_unknown_expiry_from_an_older_cloud( + self, fake_keyring, monkeypatch + ) -> None: + response = _lifetime_response() + response.headers.pop("cache-control") + monkeypatch.setattr("engine.auth.paste.httpx.get", lambda *a, **k: response) + + cred = PasteTokenProvider().login(token=TOKEN) + + assert cred["token"] == TOKEN + assert cred["expires_at"] is None def test_login_rejects_bad_token(self, fake_keyring, monkeypatch) -> None: monkeypatch.setattr( @@ -62,7 +102,7 @@ def test_login_rejects_bad_token(self, fake_keyring, monkeypatch) -> None: lambda *a, **k: FakeResponse(401), ) with pytest.raises(TokenValidationError, match="rejected"): - PasteTokenProvider().login(token="oai_ingest_bad") + PasteTokenProvider().login(token=TOKEN) def test_login_network_error(self, fake_keyring, monkeypatch) -> None: def _raise(*a, **k): @@ -70,7 +110,26 @@ def _raise(*a, **k): monkeypatch.setattr("engine.auth.paste.httpx.get", _raise) with pytest.raises(TokenValidationError, match="Could not reach"): - PasteTokenProvider().login(token="oai_ingest_x") + PasteTokenProvider().login(token=TOKEN) + + @pytest.mark.parametrize( + "wrong_role", + [ + "supabase_authorization_code", + "oap_" + "A" * 43, + "oapp_" + "A" * 43, + "oaps_" + "A" * 43, + ], + ) + def test_rejects_every_non_ingest_credential_before_network( + self, wrong_role, monkeypatch + ) -> None: + monkeypatch.setattr( + "engine.auth.paste.httpx.get", + lambda *a, **k: pytest.fail("wrong credential role reached Cloud validation"), + ) + with pytest.raises(TokenValidationError, match="pairing code or portal"): + PasteTokenProvider().login(token=wrong_role) def test_login_no_token_raises(self, fake_keyring) -> None: with pytest.raises(TokenValidationError, match="No ingest token"): diff --git a/tests/test_engine/test_auth_rotation.py b/tests/test_engine/test_auth_rotation.py new file mode 100644 index 0000000..9ae312e --- /dev/null +++ b/tests/test_engine/test_auth_rotation.py @@ -0,0 +1,365 @@ +"""Credential lifetime warnings and no-outage rotation contracts.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from uuid import uuid4 + +import httpx +import pytest + +from engine.auth import rotation, store + +HOST = "https://app.openadapt.ai" +OLD_TOKEN = "oai_ingest_" + "A" * 43 +NEW_TOKEN = "oai_ingest_" + "B" * 43 +PREVIOUS_ID = str(uuid4()) +REPLACEMENT_ID = str(uuid4()) + + +class _Response: + def __init__( + self, + status_code: int, + body: dict, + headers: dict[str, str] | None = None, + ) -> None: + self.status_code = status_code + self._body = body + self.headers = headers or {} + + def json(self) -> dict: + return self._body + + +def _iso(days: int) -> str: + return (datetime.now(timezone.utc) + timedelta(days=days)).isoformat() + + +def _lifetime(days: int = 90) -> dict: + return { + "expires_at": _iso(days), + "expires_in_days": days, + "expiring_soon": days <= 14, + "legacy_non_expiring": False, + "warning_days": 14, + } + + +def _rotation_response() -> _Response: + lifetime = _lifetime(90) + return _Response( + 201, + { + "token": NEW_TOKEN, + "record": { + "id": REPLACEMENT_ID, + "org_id": "org_1", + "name": "Desktop", + "token_prefix": NEW_TOKEN[:20], + "created_at": _iso(0), + "last_used_at": None, + "expires_at": lifetime["expires_at"], + "revoked_at": None, + "rotated_to_id": None, + }, + "previous_id": PREVIOUS_ID, + "previous_expires_at": _iso(7), + "credential": lifetime, + }, + {"cache-control": "no-store", "referrer-policy": "no-referrer"}, + ) + + +def _credential(token: str, expires_at: float | None = None) -> dict: + return { + "kind": "ingest_token", + "token": token, + "refresh_token": None, + "org_id": "org_1", + "host": HOST, + "expires_at": expires_at, + } + + +def test_14_day_warning_requires_matching_header_and_body(monkeypatch) -> None: + status_body = {"count": 0, "credential": _lifetime(14)} + response = _Response( + 200, + status_body, + { + "cache-control": "no-store", + "x-openadapt-credential-warning-days": "14", + "x-openadapt-credential-expires-in-days": "14", + }, + ) + monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response) + + status = rotation.credential_status(HOST, OLD_TOKEN) + + assert status is not None + assert status["expires_in_days"] == 14 + assert status["expiring_soon"] is True + assert "expires in 14 days" in rotation.expiry_warning(status) + + +def test_status_rejects_a_wrong_credential_role_before_network(monkeypatch) -> None: + monkeypatch.setattr( + rotation.httpx, + "get", + lambda *a, **k: pytest.fail("wrong credential role reached Cloud status"), + ) + assert rotation.credential_status(HOST, "oap_" + "A" * 43) is None + + +def test_day_14_accepts_server_authoritative_false_before_exact_boundary(monkeypatch) -> None: + lifetime = _lifetime(14) + lifetime["expiring_soon"] = False + response = _Response( + 200, + {"count": 0, "credential": lifetime}, + { + "cache-control": "no-store", + "x-openadapt-credential-warning-days": "14", + "x-openadapt-credential-expires-in-days": "14", + }, + ) + monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response) + + status = rotation.credential_status(HOST, OLD_TOKEN) + + assert status is not None + assert status["expires_in_days"] == 14 + assert status["expiring_soon"] is False + assert rotation.expiry_warning(status) is None + + +@pytest.mark.parametrize( + ("days", "expiring"), + [(13, False), (15, True)], +) +def test_warning_rejects_impossible_whole_day_state(days, expiring, monkeypatch) -> None: + lifetime = _lifetime(days) + lifetime["expiring_soon"] = expiring + response = _Response( + 200, + {"count": 0, "credential": lifetime}, + { + "cache-control": "no-store", + "x-openadapt-credential-warning-days": "14", + "x-openadapt-credential-expires-in-days": str(days), + }, + ) + monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response) + assert rotation.credential_status(HOST, OLD_TOKEN) is None + + +def test_warning_rejects_a_header_body_disagreement(monkeypatch) -> None: + response = _Response( + 200, + {"count": 0, "credential": _lifetime(14)}, + { + "x-openadapt-credential-warning-days": "14", + "x-openadapt-credential-expires-in-days": "13", + }, + ) + monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response) + assert rotation.credential_status(HOST, OLD_TOKEN) is None + + +def test_legacy_non_expiring_status_has_no_expiry_header(monkeypatch) -> None: + response = _Response( + 200, + { + "count": 0, + "credential": { + "expires_at": None, + "expires_in_days": None, + "expiring_soon": False, + "legacy_non_expiring": True, + "warning_days": 14, + }, + }, + { + "cache-control": "no-store", + "x-openadapt-credential-warning-days": "14", + }, + ) + monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: response) + status = rotation.credential_status(HOST, OLD_TOKEN) + assert status is not None + assert "no renewal date" in rotation.expiry_warning(status) + + +def test_rotation_stages_then_atomically_promotes_the_one_time_replacement( + monkeypatch, +) -> None: + store.store_credential(_credential(OLD_TOKEN)) + requests: list[tuple[str, dict, dict]] = [] + + def _post(url, *, json, headers, **kwargs): + requests.append((url, json, headers)) + return _rotation_response() + + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr(rotation.httpx, "post", _post) + + replacement = rotation.rotate_credential(HOST) + + assert replacement["token"] == NEW_TOKEN + assert isinstance(replacement["expires_at"], float) + assert store.load_credential(HOST) == replacement + assert store.load_rotation_stage() is None + assert requests == [ + ( + f"{HOST}/api/ingest-tokens/rotate", + {}, + {"Authorization": f"Bearer {OLD_TOKEN}"}, + ) + ] + + +@pytest.mark.parametrize("field", ["previous_id", "record_id"]) +def test_rotation_response_requires_production_uuid_ids(field) -> None: + response = _rotation_response() + body = response.json() + if field == "previous_id": + body["previous_id"] = "ingest_previous_1" + else: + body["record"]["id"] = "ingest_replacement_1" + + with pytest.raises(rotation.RotationError, match="credential id|credential record"): + rotation._validated_rotation(body, response.headers) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("token_prefix", "oai_ingest_wrong"), + ("last_used_at", "2026-07-28T00:00:00+00:00"), + ("revoked_at", "2026-07-28T00:00:00+00:00"), + ("rotated_to_id", str(uuid4())), + ], +) +def test_rotation_response_rejects_a_noncurrent_record(field, value) -> None: + response = _rotation_response() + response.json()["record"][field] = value + + with pytest.raises(rotation.RotationError, match="credential record"): + rotation._validated_rotation(response.json(), response.headers) + + +def test_rotation_response_rejects_extra_record_fields() -> None: + response = _rotation_response() + response.json()["record"]["token_hash"] = "must-not-cross-the-wire" + + with pytest.raises(rotation.RotationError, match="credential record"): + rotation._validated_rotation(response.json(), response.headers) + + +def test_rotation_refuses_before_network_when_keychain_is_unavailable(monkeypatch) -> None: + store.store_credential(_credential(OLD_TOKEN)) + monkeypatch.setattr(rotation, "secure_store_available", lambda: False) + monkeypatch.setattr( + rotation.httpx, + "post", + lambda *a, **k: pytest.fail("rotation must not consume a response"), + ) + with pytest.raises(rotation.RotationError, match="keychain"): + rotation.rotate_credential(HOST) + + +def test_lost_rotation_response_preserves_old_token_and_requires_reconnect( + monkeypatch, +) -> None: + store.store_credential(_credential(OLD_TOKEN)) + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr( + rotation.httpx, + "post", + lambda *a, **k: (_ for _ in ()).throw(httpx.ReadError("lost")), + ) + + with pytest.raises(rotation.RotationError, match="did not receive.*do not retry"): + rotation.rotate_credential(HOST) + assert store.load_credential(HOST)["token"] == OLD_TOKEN + assert store.load_rotation_stage() is None + + +def test_stage_failure_preserves_old_token_and_does_not_recommend_replay(monkeypatch) -> None: + store.store_credential(_credential(OLD_TOKEN)) + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr(rotation.httpx, "post", lambda *a, **k: _rotation_response()) + monkeypatch.setattr(rotation, "stage_rotation_credential", lambda *a, **k: False) + + with pytest.raises(rotation.RotationError, match="Sign in again; do not retry"): + rotation.rotate_credential(HOST) + assert store.load_credential(HOST)["token"] == OLD_TOKEN + + +def test_rotation_rejects_a_one_time_response_without_no_store(monkeypatch) -> None: + store.store_credential(_credential(OLD_TOKEN)) + response = _rotation_response() + response.headers = {} + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr(rotation.httpx, "post", lambda *a, **k: response) + + with pytest.raises(rotation.RotationError, match="Sign in again; do not retry"): + rotation.rotate_credential(HOST) + assert store.load_credential(HOST)["token"] == OLD_TOKEN + assert store.load_rotation_stage() is None + + +def test_retained_rotation_recovers_without_a_second_cloud_request(monkeypatch) -> None: + store.store_credential(_credential(OLD_TOKEN)) + response = _rotation_response().json() + token, previous_id, expiry = rotation._validated_rotation( + response, + {"cache-control": "no-store"}, + ) + replacement = _credential(token, expiry) + previous = store.snapshot_pairing_canonical(HOST) + assert previous is not None + assert store.stage_rotation_credential(previous_id, replacement, previous) + monkeypatch.setattr( + rotation.httpx, + "post", + lambda *a, **k: pytest.fail("recovery must not issue another rotation"), + ) + + recovered = rotation.rotate_credential(HOST) + + assert recovered == replacement + assert store.load_credential(HOST) == replacement + assert store.load_rotation_stage() is None + + +def test_second_rotation_409_requires_a_new_login(monkeypatch) -> None: + store.store_credential(_credential(OLD_TOKEN)) + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr(rotation.httpx, "post", lambda *a, **k: _Response(409, {})) + with pytest.raises(rotation.RotationError, match="already renewed.*login"): + rotation.rotate_credential(HOST) + + +@pytest.mark.parametrize( + "wrong_role", + [ + "supabase_authorization_code", + "oap_" + "A" * 43, + "oapp_" + "A" * 43, + "oaps_" + "A" * 43, + ], +) +def test_rotation_rejects_every_non_ingest_credential_before_network( + wrong_role, monkeypatch +) -> None: + store.store_credential(_credential(wrong_role)) + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr( + rotation.httpx, + "post", + lambda *a, **k: pytest.fail("wrong credential role reached Cloud rotation"), + ) + with pytest.raises(rotation.RotationError, match="no stored connection"): + rotation.rotate_credential(HOST) diff --git a/tests/test_engine/test_cli.py b/tests/test_engine/test_cli.py index 8d4a545..5d86dae 100644 --- a/tests/test_engine/test_cli.py +++ b/tests/test_engine/test_cli.py @@ -131,6 +131,45 @@ def test_login_success(self, cli_config: EngineConfig, capsys) -> None: assert "Logged in" in captured.out assert "org_5" in captured.out + def test_credential_command_surfaces_the_14_day_warning( + self, cli_config: EngineConfig, capsys + ) -> None: + status = { + "expires_at": "2026-08-10T00:00:00+00:00", + "expires_at_timestamp": 1786320000.0, + "expires_in_days": 14, + "expiring_soon": True, + "legacy_non_expiring": False, + "warning_days": 14, + } + with ( + patch("engine.cli.EngineConfig", return_value=cli_config), + patch("engine.auth.rotation.credential_status", return_value=status), + ): + main(["credential"]) + assert "expires in 14 days" in capsys.readouterr().out + + def test_rotate_command_reports_the_overlap_without_printing_the_token( + self, cli_config: EngineConfig, capsys + ) -> None: + token = "oai_ingest_" + "S" * 43 + credential = { + "kind": "ingest_token", + "token": token, + "refresh_token": None, + "org_id": "org_5", + "host": "https://app.openadapt.ai", + "expires_at": 1793016000.0, + } + with ( + patch("engine.cli.EngineConfig", return_value=cli_config), + patch("engine.auth.rotation.rotate_credential", return_value=credential), + ): + main(["rotate"]) + output = capsys.readouterr().out + assert "old credential remains valid for at most seven days" in output + assert token not in output + def test_push_success(self, cli_config: EngineConfig, capsys) -> None: """push should print the returned workflow id + dashboard URL.""" result = {"success": True, "workflow_id": "wf_2", From 1a12e8ec512c6468cdce81c0fa61cff8cf46be6a Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 28 Jul 2026 02:09:17 -0400 Subject: [PATCH 2/4] fix(auth): verify staged rotation replacements --- docs/DEVICE_AUTHENTICATION.md | 7 ++- engine/auth/browser_pkce.py | 21 +++++-- engine/auth/rotation.py | 58 ++++++++++++++++++++ tests/test_engine/test_auth_browser_pkce.py | 34 ++++++++++++ tests/test_engine/test_auth_rotation.py | 61 ++++++++++++++++++++- 5 files changed, 173 insertions(+), 8 deletions(-) diff --git a/docs/DEVICE_AUTHENTICATION.md b/docs/DEVICE_AUTHENTICATION.md index ff6ff54..80dca67 100644 --- a/docs/DEVICE_AUTHENTICATION.md +++ b/docs/DEVICE_AUTHENTICATION.md @@ -55,8 +55,11 @@ openadapt-desktop rotate Rotation creates the replacement first. The old bearer stays valid for at most seven days. Desktop durably stages the one-time response before it changes -the active keychain entries. After a process interruption, Desktop finishes -that exact stage and does not send a second rotation request. +the active keychain entries. It then authenticates one independent +`/api/needs-attention/count` request with the replacement and checks the exact +lifetime contract before promotion. After a process interruption, Desktop +validates and finishes that exact stage. It does not send a second rotation +request. If the response is lost before Desktop can stage it, the old bearer remains usable during the overlap. Cloud cannot replay the raw replacement, so the diff --git a/engine/auth/browser_pkce.py b/engine/auth/browser_pkce.py index 70eaf23..a0f7640 100644 --- a/engine/auth/browser_pkce.py +++ b/engine/auth/browser_pkce.py @@ -235,7 +235,16 @@ def __init__( open_browser=None, timeout: float = DEFAULT_TIMEOUT_S, ) -> None: - self.host = host.rstrip("/") + raw_host = host.rstrip("/") + kind = self._kind_for_host(raw_host) + try: + self.host = _validate_destination(raw_host, kind) + self._host_is_valid = True + except PairingError: + # Retain the safe, secret-free input only so `is_available()` can + # refuse it. No login URL or claim uses an invalid destination. + self.host = raw_host + self._host_is_valid = False self._uses_system_browser = open_browser is None self._open_browser = open_browser or self._default_open_browser self._timeout = timeout @@ -256,9 +265,7 @@ def is_available(self) -> bool: """ if os.environ.get("OPENADAPT_HEADLESS", "").strip(): return False - try: - _validate_destination(self.host, self._destination_kind()) - except PairingError: + if not self._host_is_valid: return False if not secure_store_available(): return False @@ -345,7 +352,11 @@ def login(self) -> Credential: def _destination_kind(self) -> str | None: """Classify the host for the pairing destination policy.""" - hostname = urllib.parse.urlparse(self.host).hostname + return self._kind_for_host(self.host) + + @staticmethod + def _kind_for_host(host: str) -> str | None: + hostname = urllib.parse.urlparse(host).hostname return "local" if hostname in {"localhost", "127.0.0.1", "::1"} else None def _build_login_url(self, redirect_uri: str, challenge: str, state: str) -> str: diff --git a/engine/auth/rotation.py b/engine/auth/rotation.py index 41b25c0..1926700 100644 --- a/engine/auth/rotation.py +++ b/engine/auth/rotation.py @@ -224,6 +224,7 @@ def rotate_credential(host: str = DEFAULT_HOST) -> Credential: "one-time response. The old credential remains valid for at most " "seven days. Sign in again; do not retry the same renewal." ) + _validate_staged_replacement(previous_id, replacement) if not commit_rotation_stage(previous_id): raise RotationError( "OpenAdapt retained the renewed credential but could not finish its " @@ -252,6 +253,7 @@ def recover_pending_rotation(host: str | None = None) -> Credential | None: previous_id, credential = _validated_rotation_stage(stage) if host is not None and credential["host"] != host.rstrip("/"): raise RotationError("A credential renewal for another Cloud host needs recovery first.") + _validate_staged_replacement(previous_id, credential) if not commit_rotation_stage(previous_id): raise RotationError( "OpenAdapt could not safely finish the retained credential renewal. " @@ -263,6 +265,62 @@ def recover_pending_rotation(host: str | None = None) -> Credential | None: return credential +def _validate_staged_replacement(previous_id: str, credential: Credential) -> None: + """Prove the retained replacement bearer before canonical promotion.""" + try: + response = httpx.get( + f"{credential['host']}{COUNT_PATH}", + headers={"Authorization": f"Bearer {credential['token']}"}, + timeout=API_TIMEOUT_S, + follow_redirects=False, + ) + except httpx.HTTPError as exc: + raise RotationError( + "OpenAdapt retained the renewed credential but could not verify it. " + "The old credential remains active, and recovery will retry the " + "retained replacement without another renewal request." + ) from exc + if not 200 <= response.status_code < 300: + raise RotationError( + "OpenAdapt retained the renewed credential, but Cloud did not accept " + f"it during verification ({response.status_code}). The old credential " + "remains active. Sign in again if verification does not recover." + ) + try: + lifetime = parse_credential_lifetime( + response.json(), + headers=response.headers, + require_headers=True, + require_fresh=True, + require_no_store=True, + ) + except (AttributeError, TypeError, ValueError) as exc: + raise RotationError( + "OpenAdapt retained the renewed credential, but Cloud returned an " + "incomplete verification contract. The old credential remains active." + ) from exc + if lifetime["expires_at_timestamp"] != credential["expires_at"]: + raise RotationError( + "OpenAdapt retained the renewed credential, but Cloud verified a " + "different expiry. The old credential remains active." + ) + # The recovery stage identity must remain unchanged across the network + # check. A concurrent operation cannot swap in another retained bearer and + # then have this successful response authorize its promotion. + try: + current_stage = load_rotation_stage() + except RuntimeError as exc: + raise RotationError("The retained credential changed during verification.") from exc + if current_stage is None: + raise RotationError("The retained credential changed during verification.") + try: + current_previous_id, current_credential = _validated_rotation_stage(current_stage) + except RotationError as exc: + raise RotationError("The retained credential changed during verification.") from exc + if current_previous_id != previous_id or current_credential != credential: + raise RotationError("The retained credential changed during verification.") + + def _stored_token(host: str) -> str | None: """Resolve the bearer this machine would actually send.""" import os diff --git a/tests/test_engine/test_auth_browser_pkce.py b/tests/test_engine/test_auth_browser_pkce.py index 9cec28d..93d754f 100644 --- a/tests/test_engine/test_auth_browser_pkce.py +++ b/tests/test_engine/test_auth_browser_pkce.py @@ -332,6 +332,40 @@ def _clear(value): "clear", ] + def test_local_alias_uses_one_canonical_host_for_login_claim_and_readback( + self, fake_keyring, monkeypatch + ) -> None: + _desktop_available(monkeypatch) + pairing_id = str(uuid4()) + login_urls: list[str] = [] + + def _open_browser(url: str) -> None: + login_urls.append(url) + _deliver_from_login_url(url) + + def _post(url, *, json, headers=None, **kwargs): + if url.endswith("/claim"): + assert url == "http://localhost/api/local-bridge/pairings/claim" + return _Response( + 201, + _claim_body(pairing_id), + {"cache-control": "no-store", "referrer-policy": "no-referrer"}, + ) + assert url == "http://localhost/api/local-bridge/pairings/confirm" + return _Response(200, {"connected": True}) + + monkeypatch.setattr(pairing.httpx, "post", _post) + monkeypatch.setattr(pairing.httpx, "get", lambda *a, **k: _validation()) + + credential = BrowserPkceProvider( + host="http://LOCALHOST:80/", + open_browser=_open_browser, + ).login() + + assert login_urls[0].startswith("http://localhost/login?") + assert credential["host"] == "http://localhost" + assert store.load_credential("http://localhost") == credential + @pytest.mark.parametrize( "wrong_code", [ diff --git a/tests/test_engine/test_auth_rotation.py b/tests/test_engine/test_auth_rotation.py index 9ae312e..8597f66 100644 --- a/tests/test_engine/test_auth_rotation.py +++ b/tests/test_engine/test_auth_rotation.py @@ -71,6 +71,18 @@ def _rotation_response() -> _Response: ) +def _replacement_validation_response(credential: dict) -> _Response: + return _Response( + 200, + {"count": 0, "credential": dict(credential)}, + { + "cache-control": "no-store", + "x-openadapt-credential-warning-days": "14", + "x-openadapt-credential-expires-in-days": str(credential["expires_in_days"]), + }, + ) + + def _credential(token: str, expires_at: float | None = None) -> dict: return { "kind": "ingest_token", @@ -196,13 +208,19 @@ def test_rotation_stages_then_atomically_promotes_the_one_time_replacement( ) -> None: store.store_credential(_credential(OLD_TOKEN)) requests: list[tuple[str, dict, dict]] = [] + rotated = _rotation_response() def _post(url, *, json, headers, **kwargs): requests.append((url, json, headers)) - return _rotation_response() + return rotated monkeypatch.setattr(rotation, "secure_store_available", lambda: True) monkeypatch.setattr(rotation.httpx, "post", _post) + monkeypatch.setattr( + rotation.httpx, + "get", + lambda *a, **k: _replacement_validation_response(rotated.json()["credential"]), + ) replacement = rotation.rotate_credential(HOST) @@ -326,6 +344,11 @@ def test_retained_rotation_recovers_without_a_second_cloud_request(monkeypatch) "post", lambda *a, **k: pytest.fail("recovery must not issue another rotation"), ) + monkeypatch.setattr( + rotation.httpx, + "get", + lambda *a, **k: _replacement_validation_response(response["credential"]), + ) recovered = rotation.rotate_credential(HOST) @@ -334,6 +357,42 @@ def test_retained_rotation_recovers_without_a_second_cloud_request(monkeypatch) assert store.load_rotation_stage() is None +def test_bad_replacement_stays_staged_and_recovery_validates_without_reissuing( + monkeypatch, +) -> None: + store.store_credential(_credential(OLD_TOKEN)) + rotated = _rotation_response() + post_calls = 0 + + def _post(*args, **kwargs): + nonlocal post_calls + post_calls += 1 + return rotated + + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr(rotation.httpx, "post", _post) + monkeypatch.setattr(rotation.httpx, "get", lambda *a, **k: _Response(401, {})) + + with pytest.raises(rotation.RotationError, match="did not accept.*old credential"): + rotation.rotate_credential(HOST) + + assert post_calls == 1 + assert store.load_credential(HOST)["token"] == OLD_TOKEN + assert store.load_rotation_stage() is not None + + monkeypatch.setattr( + rotation.httpx, + "get", + lambda *a, **k: _replacement_validation_response(rotated.json()["credential"]), + ) + recovered = rotation.rotate_credential(HOST) + + assert post_calls == 1 + assert recovered["token"] == NEW_TOKEN + assert store.load_credential(HOST) == recovered + assert store.load_rotation_stage() is None + + def test_second_rotation_409_requires_a_new_login(monkeypatch) -> None: store.store_credential(_credential(OLD_TOKEN)) monkeypatch.setattr(rotation, "secure_store_available", lambda: True) From 2683fe79a2f95a7bf8106acaf3e5dac0af9c0119 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 28 Jul 2026 02:15:17 -0400 Subject: [PATCH 3/4] fix(auth): supersede stale rotation recovery --- docs/DEVICE_AUTHENTICATION.md | 4 +- engine/auth/rotation.py | 13 +++- engine/auth/store.py | 63 ++++++++++++++++++ tests/test_engine/test_auth_rotation.py | 87 ++++++++++++++++++++++--- 4 files changed, 154 insertions(+), 13 deletions(-) diff --git a/docs/DEVICE_AUTHENTICATION.md b/docs/DEVICE_AUTHENTICATION.md index 80dca67..032fc19 100644 --- a/docs/DEVICE_AUTHENTICATION.md +++ b/docs/DEVICE_AUTHENTICATION.md @@ -59,7 +59,9 @@ the active keychain entries. It then authenticates one independent `/api/needs-attention/count` request with the replacement and checks the exact lifetime contract before promotion. After a process interruption, Desktop validates and finishes that exact stage. It does not send a second rotation -request. +request. Recovery requires the same stored expiry, but the remaining-day count +can decrease while the machine is offline. A later successful login safely +supersedes an older rejected rotation stage. If the response is lost before Desktop can stage it, the old bearer remains usable during the overlap. Cloud cannot replay the raw replacement, so the diff --git a/engine/auth/rotation.py b/engine/auth/rotation.py index 1926700..cfb81be 100644 --- a/engine/auth/rotation.py +++ b/engine/auth/rotation.py @@ -36,6 +36,7 @@ DEFAULT_HOST, INGEST_TOKEN_ENV, clear_rotation_stage, + clear_superseded_rotation_stage, commit_rotation_stage, load_credential, load_rotation_stage, @@ -253,6 +254,12 @@ def recover_pending_rotation(host: str | None = None) -> Credential | None: previous_id, credential = _validated_rotation_stage(stage) if host is not None and credential["host"] != host.rstrip("/"): raise RotationError("A credential renewal for another Cloud host needs recovery first.") + if clear_superseded_rotation_stage(previous_id): + logger.info( + "Cleared a stale credential renewal after a later login for {host}", + host=credential["host"], + ) + return None _validate_staged_replacement(previous_id, credential) if not commit_rotation_stage(previous_id): raise RotationError( @@ -291,7 +298,6 @@ def _validate_staged_replacement(previous_id: str, credential: Credential) -> No response.json(), headers=response.headers, require_headers=True, - require_fresh=True, require_no_store=True, ) except (AttributeError, TypeError, ValueError) as exc: @@ -299,7 +305,10 @@ def _validate_staged_replacement(previous_id: str, credential: Credential) -> No "OpenAdapt retained the renewed credential, but Cloud returned an " "incomplete verification contract. The old credential remains active." ) from exc - if lifetime["expires_at_timestamp"] != credential["expires_at"]: + if ( + lifetime["legacy_non_expiring"] + or lifetime["expires_at_timestamp"] != credential["expires_at"] + ): raise RotationError( "OpenAdapt retained the renewed credential, but Cloud verified a " "different expiry. The old credential remains active." diff --git a/engine/auth/store.py b/engine/auth/store.py index 5036dfe..cb4e116 100644 --- a/engine/auth/store.py +++ b/engine/auth/store.py @@ -43,6 +43,7 @@ from loguru import logger +from engine.auth.lifetime import valid_ingest_token from engine.auth.provider import Credential SERVICE_NAME = "ai.openadapt.desktop" @@ -557,6 +558,68 @@ def clear_rotation_stage(previous_id: str) -> bool: return _apply_exact(_keyring(), _ROTATION_STAGE_ACCOUNT, None) +def clear_superseded_rotation_stage(previous_id: str) -> bool: + """Clear an exact stale stage only after a coherent later login won. + + A rejected replacement remains staged so transient propagation can recover. + If a later login writes a third, internally consistent credential for the + same host, that credential supersedes both the old snapshot and the staged + replacement. Retaining the dead stage would otherwise block every future + rotation of the new connection. + """ + try: + stage = load_rotation_stage() + except RuntimeError: + return False + if stage is None or stage.get("previous_id") != previous_id: + return False + values = _rotation_stage_values(stage) + if values is None: + return False + host, previous, replacement = values + kr = _keyring() + accounts = (host, host + _CRED_SUFFIX, _ACTIVE_HOST_ACCOUNT) + observed: list[str | None] = [] + for account in accounts: + readable, value = _strict_get(kr, account) + if not readable: + return False + observed.append(value) + current = tuple(observed) + if current in {previous, replacement}: + return False + if all(value in {prior, new} for value, prior, new in zip(current, previous, replacement)): + # A partial old/replacement write needs recovery, not deletion. + return False + raw_token, raw_credential, active = current + if not valid_ingest_token(raw_token) or not isinstance(raw_credential, str) or active != host: + return False + try: + credential = json.loads(raw_credential) + except (json.JSONDecodeError, TypeError): + return False + if ( + not isinstance(credential, dict) + or set(credential) + != {"kind", "token", "refresh_token", "org_id", "host", "expires_at"} + or credential.get("kind") != "ingest_token" + or credential.get("refresh_token") is not None + or credential.get("token") != raw_token + or credential.get("host") != host + ): + return False + + # Re-read the stage after inspecting canonical state. Delete only the same + # stage identity; a concurrent rotation cannot be mistaken for this one. + try: + current_stage = load_rotation_stage() + except RuntimeError: + return False + if current_stage != stage: + return False + return clear_rotation_stage(previous_id) + + def load_credential(host: str) -> Credential | None: """Load a stored credential for ``host``, or None if absent/unreadable. diff --git a/tests/test_engine/test_auth_rotation.py b/tests/test_engine/test_auth_rotation.py index 8597f66..917c18e 100644 --- a/tests/test_engine/test_auth_rotation.py +++ b/tests/test_engine/test_auth_rotation.py @@ -13,6 +13,8 @@ HOST = "https://app.openadapt.ai" OLD_TOKEN = "oai_ingest_" + "A" * 43 NEW_TOKEN = "oai_ingest_" + "B" * 43 +THIRD_TOKEN = "oai_ingest_" + "C" * 43 +FOURTH_TOKEN = "oai_ingest_" + "D" * 43 PREVIOUS_ID = str(uuid4()) REPLACEMENT_ID = str(uuid4()) @@ -46,24 +48,29 @@ def _lifetime(days: int = 90) -> dict: } -def _rotation_response() -> _Response: +def _rotation_response( + *, + token: str = NEW_TOKEN, + previous_id: str = PREVIOUS_ID, + replacement_id: str = REPLACEMENT_ID, +) -> _Response: lifetime = _lifetime(90) return _Response( 201, { - "token": NEW_TOKEN, + "token": token, "record": { - "id": REPLACEMENT_ID, + "id": replacement_id, "org_id": "org_1", "name": "Desktop", - "token_prefix": NEW_TOKEN[:20], + "token_prefix": token[:20], "created_at": _iso(0), "last_used_at": None, "expires_at": lifetime["expires_at"], "revoked_at": None, "rotated_to_id": None, }, - "previous_id": PREVIOUS_ID, + "previous_id": previous_id, "previous_expires_at": _iso(7), "credential": lifetime, }, @@ -71,14 +78,22 @@ def _rotation_response() -> _Response: ) -def _replacement_validation_response(credential: dict) -> _Response: +def _replacement_validation_response( + credential: dict, + *, + days: int | None = None, +) -> _Response: + current = dict(credential) + if days is not None: + current["expires_in_days"] = days + current["expiring_soon"] = days < 14 return _Response( 200, - {"count": 0, "credential": dict(credential)}, + {"count": 0, "credential": current}, { "cache-control": "no-store", "x-openadapt-credential-warning-days": "14", - "x-openadapt-credential-expires-in-days": str(credential["expires_in_days"]), + "x-openadapt-credential-expires-in-days": str(current["expires_in_days"]), }, ) @@ -328,7 +343,9 @@ def test_rotation_rejects_a_one_time_response_without_no_store(monkeypatch) -> N assert store.load_rotation_stage() is None -def test_retained_rotation_recovers_without_a_second_cloud_request(monkeypatch) -> None: +def test_delayed_rotation_recovery_accepts_decreased_days_without_a_second_request( + monkeypatch, +) -> None: store.store_credential(_credential(OLD_TOKEN)) response = _rotation_response().json() token, previous_id, expiry = rotation._validated_rotation( @@ -347,7 +364,7 @@ def test_retained_rotation_recovers_without_a_second_cloud_request(monkeypatch) monkeypatch.setattr( rotation.httpx, "get", - lambda *a, **k: _replacement_validation_response(response["credential"]), + lambda *a, **k: _replacement_validation_response(response["credential"], days=80), ) recovered = rotation.rotate_credential(HOST) @@ -393,6 +410,56 @@ def _post(*args, **kwargs): assert store.load_rotation_stage() is None +def test_later_login_supersedes_rejected_stage_before_future_rotation(monkeypatch) -> None: + store.store_credential(_credential(OLD_TOKEN)) + rejected = _rotation_response() + replacement = _rotation_response( + token=FOURTH_TOKEN, + previous_id=str(uuid4()), + replacement_id=str(uuid4()), + ) + posted_bearers: list[str] = [] + validated_bearers: list[str] = [] + + def _post(url, *, headers, **kwargs): + posted_bearers.append(headers["Authorization"]) + return rejected if len(posted_bearers) == 1 else replacement + + def _get(url, *, headers, **kwargs): + bearer = headers["Authorization"] + validated_bearers.append(bearer) + if bearer == f"Bearer {NEW_TOKEN}": + return _Response(401, {}) + assert bearer == f"Bearer {FOURTH_TOKEN}" + return _replacement_validation_response(replacement.json()["credential"]) + + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr(rotation.httpx, "post", _post) + monkeypatch.setattr(rotation.httpx, "get", _get) + + with pytest.raises(rotation.RotationError, match="did not accept"): + rotation.rotate_credential(HOST) + assert store.load_rotation_stage() is not None + assert store.load_credential(HOST)["token"] == OLD_TOKEN + + # A later browser/deep-link login writes a coherent third credential. + store.store_credential(_credential(THIRD_TOKEN)) + + rotated = rotation.rotate_credential(HOST) + + assert posted_bearers == [ + f"Bearer {OLD_TOKEN}", + f"Bearer {THIRD_TOKEN}", + ] + assert validated_bearers == [ + f"Bearer {NEW_TOKEN}", + f"Bearer {FOURTH_TOKEN}", + ] + assert rotated["token"] == FOURTH_TOKEN + assert store.load_credential(HOST) == rotated + assert store.load_rotation_stage() is None + + def test_second_rotation_409_requires_a_new_login(monkeypatch) -> None: store.store_credential(_credential(OLD_TOKEN)) monkeypatch.setattr(rotation, "secure_store_available", lambda: True) From 716c1989e932ddff8dfe8eabe4375b5e6d603cd3 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 28 Jul 2026 02:17:17 -0400 Subject: [PATCH 4/4] fix(auth): keep immediate rotation verification fresh --- engine/auth/rotation.py | 19 +++++++++++++++---- tests/test_engine/test_auth_rotation.py | 23 +++++++++++++++++++++++ 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/engine/auth/rotation.py b/engine/auth/rotation.py index cfb81be..a1c52c9 100644 --- a/engine/auth/rotation.py +++ b/engine/auth/rotation.py @@ -225,7 +225,7 @@ def rotate_credential(host: str = DEFAULT_HOST) -> Credential: "one-time response. The old credential remains valid for at most " "seven days. Sign in again; do not retry the same renewal." ) - _validate_staged_replacement(previous_id, replacement) + _validate_staged_replacement(previous_id, replacement, require_fresh=True) if not commit_rotation_stage(previous_id): raise RotationError( "OpenAdapt retained the renewed credential but could not finish its " @@ -260,7 +260,7 @@ def recover_pending_rotation(host: str | None = None) -> Credential | None: host=credential["host"], ) return None - _validate_staged_replacement(previous_id, credential) + _validate_staged_replacement(previous_id, credential, require_fresh=False) if not commit_rotation_stage(previous_id): raise RotationError( "OpenAdapt could not safely finish the retained credential renewal. " @@ -272,8 +272,18 @@ def recover_pending_rotation(host: str | None = None) -> Credential | None: return credential -def _validate_staged_replacement(previous_id: str, credential: Credential) -> None: - """Prove the retained replacement bearer before canonical promotion.""" +def _validate_staged_replacement( + previous_id: str, + credential: Credential, + *, + require_fresh: bool, +) -> None: + """Prove the retained replacement bearer before canonical promotion. + + The immediate check must still show a new 89/90-day credential. Recovery + can occur later, so it binds the exact stored expiry while permitting the + display day count to decrease. + """ try: response = httpx.get( f"{credential['host']}{COUNT_PATH}", @@ -298,6 +308,7 @@ def _validate_staged_replacement(previous_id: str, credential: Credential) -> No response.json(), headers=response.headers, require_headers=True, + require_fresh=require_fresh, require_no_store=True, ) except (AttributeError, TypeError, ValueError) as exc: diff --git a/tests/test_engine/test_auth_rotation.py b/tests/test_engine/test_auth_rotation.py index 917c18e..0495506 100644 --- a/tests/test_engine/test_auth_rotation.py +++ b/tests/test_engine/test_auth_rotation.py @@ -252,6 +252,29 @@ def _post(url, *, json, headers, **kwargs): ] +def test_immediate_replacement_verification_still_requires_a_fresh_lifetime( + monkeypatch, +) -> None: + store.store_credential(_credential(OLD_TOKEN)) + rotated = _rotation_response() + monkeypatch.setattr(rotation, "secure_store_available", lambda: True) + monkeypatch.setattr(rotation.httpx, "post", lambda *a, **k: rotated) + monkeypatch.setattr( + rotation.httpx, + "get", + lambda *a, **k: _replacement_validation_response( + rotated.json()["credential"], + days=1, + ), + ) + + with pytest.raises(rotation.RotationError, match="incomplete verification contract"): + rotation.rotate_credential(HOST) + + assert store.load_credential(HOST)["token"] == OLD_TOKEN + assert store.load_rotation_stage() is not None + + @pytest.mark.parametrize("field", ["previous_id", "record_id"]) def test_rotation_response_requires_production_uuid_ids(field) -> None: response = _rotation_response()