From cc67af75c43276e45f5162c1ce47ecf9e7de2e2e Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:07:48 -0700 Subject: [PATCH] refactor(honcho): compact oauth + oauth_flow; drop _atomic_write_config for utils.atomic_json_write --- plugins/memory/honcho/oauth.py | 321 ++++++++-------------------- plugins/memory/honcho/oauth_flow.py | 320 ++++++++------------------- 2 files changed, 185 insertions(+), 456 deletions(-) diff --git a/plugins/memory/honcho/oauth.py b/plugins/memory/honcho/oauth.py index 8adbd46585..44cdfd6e00 100644 --- a/plugins/memory/honcho/oauth.py +++ b/plugins/memory/honcho/oauth.py @@ -1,16 +1,10 @@ """OAuth credential storage and refresh for the Honcho memory provider. -An access token authenticates like a scoped API key, so it is stored as the -host's ``apiKey``; the refresh token is exchanged before expiry to keep it live. - -Refresh tokens rotate with single-use reuse detection: replaying a stale token -revokes the grant, so every refresh persists the rotated token atomically and is -serialized (in-process lock + cross-process file lock). A failed exchange never -raises into the agent: transient failures retry once immediately (the replay -grace window is short); a permanent error such as invalid_grant marks the grant -dead so nothing keeps hitting the endpoint and callers surface a re-login -prompt. A server-side 401 on a locally-valid token is recovered via -``force_refresh_token``. +The access token is stored as the host's ``apiKey``; the refresh token is exchanged before expiry. +Refresh tokens rotate with single-use reuse detection, so every refresh persists atomically and is +serialized (in-process lock + cross-process file lock). A failed exchange never raises into the agent: +transient failures retry once immediately; a permanent error (invalid_grant) marks the grant dead so +callers surface a re-login prompt. A server-side 401 on a locally-valid token uses ``force_refresh_token``. """ from __future__ import annotations @@ -35,71 +29,55 @@ REFRESH_TOKEN_PREFIX = "hch-rt-" _REFRESH_SKEW_SECONDS = 120 # refresh this early so an in-flight request never races expiry _REFRESH_TIMEOUT_SECONDS = 15.0 # short: sits on the path to a memory call _REFRESH_RETRY_DELAY_SECONDS = 2.0 # replayed refresh tokens are honored only briefly after rotation -# Cap for one exchange cycle (attempt + pause + retry): it runs under the global -# refresh locks, so it must not hold them for two full HTTP timeouts. +# One exchange cycle (attempt + pause + retry) runs under the global refresh locks: cap it below 2 timeouts. _REFRESH_TOTAL_BUDGET_SECONDS = 20.0 -# After a transient failure, fail open without re-exchanging for this long so N -# waiting threads don't serialize N exchange cycles against a failing endpoint. +# After a transient failure, fail open without re-exchanging for this long so N waiting threads +# don't serialize N exchange cycles against a failing endpoint. _REFRESH_FAILURE_COOLDOWN_SECONDS = 30.0 # OAuth error codes a retry can never fix — the grant itself is dead. _PERMANENT_OAUTH_ERRORS = frozenset({"invalid_grant", "invalid_client", "unauthorized_client"}) - # Derived from the canonical prefixes so a prefix change can't silently break redaction. -_TOKEN_VALUE_RE = re.compile( - rf"({re.escape(ACCESS_TOKEN_PREFIX)}|{re.escape(REFRESH_TOKEN_PREFIX)})[A-Za-z0-9._~+/=-]+" -) - +_TOKEN_VALUE_RE = re.compile(rf"({re.escape(ACCESS_TOKEN_PREFIX)}|{re.escape(REFRESH_TOKEN_PREFIX)})[A-Za-z0-9._~+/=-]+") def redact_tokens(text: str) -> str: """Replace any embedded token values with their prefix plus a placeholder.""" return _TOKEN_VALUE_RE.sub(lambda m: f"{m.group(1)}[redacted]", text) - _redact_tokens = redact_tokens # backward-compat alias for older importers - class OAuthRefreshError(Exception): """Token endpoint rejected the refresh. ``permanent`` means re-login is required.""" def __init__(self, message: str, *, error: str = "", permanent: bool = False): super().__init__(message) - self.error = error - self.permanent = permanent + self.error, self.permanent = error, permanent - -# Serializes refresh across threads; state is re-checked under it so racing -# callers don't replay a rotated refresh token. +# Serializes refresh across threads; state is re-checked under it so racers don't replay a rotated token. _refresh_lock = threading.Lock() - def _os_lock(fh, lock: bool) -> None: if os.name == "nt": import msvcrt - fh.seek(0) msvcrt.locking(fh.fileno(), msvcrt.LK_LOCK if lock else msvcrt.LK_UNLCK, 1) else: import fcntl - fcntl.flock(fh.fileno(), fcntl.LOCK_EX if lock else fcntl.LOCK_UN) - @contextmanager def _config_refresh_lock(path: Path): - """Machine-wide advisory lock (``.lock``) around read-refresh-persist: - a sibling process sharing this honcho.json must not replay the single-use - refresh token. Best-effort — without flock it degrades to in-process only.""" + """Machine-wide advisory lock (``.lock``) around read-refresh-persist: a sibling process sharing + this honcho.json must not replay the single-use refresh token. Best-effort — degrades to in-process only.""" fh = None try: - lock_path = Path(f"{path}.lock") - lock_path.parent.mkdir(parents=True, exist_ok=True) - fh = open(lock_path, "a+b") + path.parent.mkdir(parents=True, exist_ok=True) + fh = open(f"{path}.lock", "a+b") _os_lock(fh, True) except Exception: logger.debug("Honcho OAuth cross-process lock unavailable; in-process only", exc_info=True) if fh is not None: fh.close() - fh = None + fh = None try: yield finally: @@ -109,46 +87,39 @@ def _config_refresh_lock(path: Path): fh.close() -# Per-grant state keyed by (config path, host); single-key dict ops are atomic -# under the GIL, so no separate lock. -# (expires_at, access): lets the hot path skip the honcho.json read while the -# token is well clear of expiry. A stale entry can't break auth (the token stays -# valid until its own expiry); it only defers noticing out-of-band rotation. +# Per-grant state keyed by (config path, host); single-key dict ops are atomic under the GIL. +# (expires_at, access): lets the hot path skip the honcho.json read while the token is well clear of +# expiry. A stale entry can't break auth; it only defers noticing out-of-band rotation. _expiry_cache: dict[tuple[str, str], tuple[float, str]] = {} -# sha256 of the permanently rejected refresh token; a re-login rotates the -# token, so the digest comparison self-clears. +# sha256 of the permanently rejected refresh token; a re-login rotates the token, so it self-clears. _dead_grants: dict[tuple[str, str], str] = {} # monotonic time of the last transient exchange failure (drives the fail-open cooldown). _refresh_failure_at: dict[tuple[str, str], float] = {} -# (config mtime_ns, verdict): reauth_required only changes when the file is -# rewritten, so an unchanged mtime skips the parse. +# (config mtime_ns, verdict): reauth_required only changes when the file is rewritten. _reauth_check_cache: dict[tuple[str, str], tuple[int, bool]] = {} - def _in_failure_cooldown(key: tuple[str, str]) -> bool: failed_at = _refresh_failure_at.get(key) return failed_at is not None and (time.monotonic() - failed_at) < _REFRESH_FAILURE_COOLDOWN_SECONDS - -def _refresh_token_digest(cred: OAuthCredential) -> str: - return hashlib.sha256(cred.refresh_token.encode("utf-8")).hexdigest() - - def _grant_is_dead(key: tuple[str, str], cred: OAuthCredential) -> bool: - return _dead_grants.get(key) == _refresh_token_digest(cred) - + return _dead_grants.get(key) == hashlib.sha256(cred.refresh_token.encode("utf-8")).hexdigest() def _mark_grant_dead(key: tuple[str, str], cred: OAuthCredential) -> None: - _dead_grants[key] = _refresh_token_digest(cred) + _dead_grants[key] = hashlib.sha256(cred.refresh_token.encode("utf-8")).hexdigest() _reauth_check_cache.pop(key, None) # verdict changed without a config rewrite +def _read_config(path: Path) -> dict[str, Any]: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} def _load_cred(path: Path, host: str, raw: dict[str, Any] | None = None) -> OAuthCredential | None: """Credential from ``host``'s block in ``raw`` (or the file at ``path``).""" source = raw if raw is not None else _read_config(path) return OAuthCredential.from_host_block((source.get("hosts") or {}).get(host) or {}) - def reauth_required(path: Path, host: str) -> bool: """True when ``host``'s OAuth grant is dead and only a new login fixes it.""" key = (str(path), host) @@ -166,12 +137,10 @@ def reauth_required(path: Path, host: str) -> bool: _reauth_check_cache[key] = (mtime, result) return result - def any_dead_grants() -> bool: """Cheap predicate letting hot-path callers skip config resolution when healthy.""" return bool(_dead_grants) - def _num(value, cast): """``cast(value)``, or ``cast(0)`` when the stored value is malformed.""" try: @@ -179,17 +148,14 @@ def _num(value, cast): except (TypeError, ValueError): return cast(0) - def is_oauth_access_token(value: str | None) -> bool: """True when ``value`` is an OAuth access token (vs a static API key).""" return bool(value) and value.startswith(ACCESS_TOKEN_PREFIX) - @dataclass class OAuthCredential: - """An OAuth grant as stored in a honcho.json host block: ``access_token`` is - the host's ``apiKey``, the rest lives in its ``oauth`` sub-block. - ``expires_at`` is absolute epoch seconds.""" + """An OAuth grant as stored in a honcho.json host block: ``access_token`` is the host's + ``apiKey``, the rest lives in its ``oauth`` sub-block. ``expires_at`` is absolute epoch seconds.""" access_token: str refresh_token: str @@ -198,64 +164,37 @@ class OAuthCredential: token_endpoint: str scope: str = "write" token_type: str = "Bearer" - # Transient consent peer name — set only on a fresh grant, never persisted. - consent_peer_name: str | None = None + consent_peer_name: str | None = None # transient: set only on a fresh grant, never persisted @classmethod def from_host_block(cls, block: dict[str, Any]) -> "OAuthCredential | None": """Build a credential from a honcho.json host block, or None if incomplete.""" - oauth = block.get("oauth") - access = block.get("apiKey") + oauth, access = block.get("oauth"), block.get("apiKey") if not isinstance(oauth, dict) or not is_oauth_access_token(access): return None refresh, endpoint, client_id = oauth.get("refreshToken"), oauth.get("tokenEndpoint"), oauth.get("clientId") if not (refresh and endpoint and client_id): return None - return cls( - access_token=access, - refresh_token=str(refresh), - expires_at=_num(oauth.get("expiresAt", 0), float), - client_id=str(client_id), - token_endpoint=str(endpoint), - scope=str(oauth.get("scope", "write")), - token_type=str(oauth.get("tokenType", "Bearer")), - ) + return cls(access, str(refresh), _num(oauth.get("expiresAt", 0), float), str(client_id), str(endpoint), + scope=str(oauth.get("scope", "write")), token_type=str(oauth.get("tokenType", "Bearer"))) @classmethod def from_token_response( - cls, - body: dict[str, Any], - *, - now: float, - client_id: str, - token_endpoint: str, - scope: str = "write", - token_type: str = "Bearer", - what: str = "grant", + cls, body: dict[str, Any], *, now: float, client_id: str, token_endpoint: str, + scope: str = "write", token_type: str = "Bearer", what: str = "grant", ) -> "OAuthCredential": """Build a credential from an OAuth token response; ``expires_in`` is relative to ``now``.""" access, refresh = body.get("access_token"), body.get("refresh_token") if not is_oauth_access_token(access) or not refresh: raise ValueError(f"{what} missing access_token/refresh_token") - return cls( - access_token=access, - refresh_token=str(refresh), - expires_at=now + _num(body.get("expires_in", 0), int), - client_id=client_id, - token_endpoint=token_endpoint, - scope=str(body.get("scope", scope)), - token_type=str(body.get("token_type", token_type)), - ) + return cls(access, str(refresh), now + _num(body.get("expires_in", 0), int), client_id, token_endpoint, + scope=str(body.get("scope", scope)), token_type=str(body.get("token_type", token_type))) def oauth_block(self) -> dict[str, Any]: """The ``oauth`` sub-block to persist (the access token lives in apiKey).""" return { - "refreshToken": self.refresh_token, - "expiresAt": int(self.expires_at), - "clientId": self.client_id, - "tokenEndpoint": self.token_endpoint, - "scope": self.scope, - "tokenType": self.token_type, + "refreshToken": self.refresh_token, "expiresAt": int(self.expires_at), "clientId": self.client_id, + "tokenEndpoint": self.token_endpoint, "scope": self.scope, "tokenType": self.token_type, } def is_expired(self, *, now: float, skew: float = _REFRESH_SKEW_SECONDS) -> bool: @@ -263,68 +202,51 @@ class OAuthCredential: return now >= (self.expires_at - skew) -# HTTP indirection: tests monkeypatch these module attributes, and the refresh -# path looks them up at call time. +# HTTP indirection: tests monkeypatch these module attributes; callers look them up at call time. def _http_json(method: str, url: str, *, data=None, timeout: float, strict: bool = True) -> tuple[int, Any]: - """Return ``(status, parsed JSON body)``. ``strict`` raises on non-2xx / - non-JSON; otherwise a 4xx passes through (RFC 8628 polling reads the OAuth - error off a 400) and a non-JSON / non-object body parses to ``{}``.""" + """Return ``(status, parsed JSON body)``. ``strict`` raises on non-2xx / non-JSON; otherwise a 4xx + passes through (RFC 8628 polling reads the OAuth error off a 400) and a non-object body parses to ``{}``.""" import httpx resp = httpx.request(method, url, data=data, timeout=timeout) if strict: resp.raise_for_status() return resp.status_code, resp.json() - try: + body = {} + with suppress(ValueError): body = resp.json() - except ValueError: - body = {} return resp.status_code, body if isinstance(body, dict) else {} - def _http_post_form_status(url: str, data: dict[str, str], timeout: float) -> tuple[int, dict[str, Any]]: """POST form-encoded ``data``; return ``(status, body)`` without raising on 4xx.""" return _http_json("POST", url, data=data, timeout=timeout, strict=False) - def _exchange_refresh_token( cred: OAuthCredential, *, now: float, timeout: float = _REFRESH_TIMEOUT_SECONDS ) -> OAuthCredential: - """Run the refresh_token grant and return the rotated credential. Raises - ``OAuthRefreshError`` (with the endpoint's error body) on an error response, - transport errors as-is; callers fail open.""" - status, body = _http_post_form_status( - cred.token_endpoint, - {"grant_type": "refresh_token", "client_id": cred.client_id, "refresh_token": cred.refresh_token}, - timeout, - ) + """Run the refresh_token grant and return the rotated credential. Raises ``OAuthRefreshError`` (with + the endpoint's error body) on an error response, transport errors as-is; callers fail open.""" + form = {"grant_type": "refresh_token", "client_id": cred.client_id, "refresh_token": cred.refresh_token} + status, body = _http_post_form_status(cred.token_endpoint, form, timeout) if status >= 400: - error = str(body.get("error") or "") - description = str(body.get("error_description") or "") + error, description = str(body.get("error") or ""), str(body.get("error_description") or "") detail = " — ".join(p for p in (error, description) if p) or "no error body" - raise OAuthRefreshError( - _redact_tokens(f"token endpoint returned HTTP {status}: {detail}"), - error=error, - permanent=error in _PERMANENT_OAUTH_ERRORS, - ) + message = _redact_tokens(f"token endpoint returned HTTP {status}: {detail}") + raise OAuthRefreshError(message, error=error, permanent=error in _PERMANENT_OAUTH_ERRORS) return OAuthCredential.from_token_response( body, now=now, client_id=cred.client_id, token_endpoint=cred.token_endpoint, scope=cred.scope, token_type=cred.token_type, what="refresh response", ) - def _exchange_with_retry(cred: OAuthCredential, *, now: float) -> OAuthCredential: - """Exchange the refresh token, retrying once on transient failure. The retry - cannot wait (replay grace window is short) and the cycle is capped by - ``_REFRESH_TOTAL_BUDGET_SECONDS`` because it runs under the refresh locks.""" + """Exchange the refresh token, retrying once on transient failure. The retry cannot wait (replay grace + window is short) and the cycle is capped by ``_REFRESH_TOTAL_BUDGET_SECONDS`` (runs under the locks).""" deadline = time.monotonic() + _REFRESH_TOTAL_BUDGET_SECONDS try: return _exchange_refresh_token(cred, now=now) - except OAuthRefreshError as exc: - if exc.permanent: - raise - first: Exception = exc except Exception as exc: + if isinstance(exc, OAuthRefreshError) and exc.permanent: + raise first = exc remaining = deadline - time.monotonic() - _REFRESH_RETRY_DELAY_SECONDS if remaining <= 0: @@ -333,119 +255,71 @@ def _exchange_with_retry(cred: OAuthCredential, *, now: float) -> OAuthCredentia time.sleep(_REFRESH_RETRY_DELAY_SECONDS) return _exchange_refresh_token(cred, now=now, timeout=min(remaining, _REFRESH_TIMEOUT_SECONDS)) - def _rotate_and_persist( - path: Path, - host: str, - key: tuple[str, str], - cred: OAuthCredential, - *, - now: float, - op_label: str = "refresh", + path: Path, host: str, key: tuple[str, str], cred: OAuthCredential, *, now: float, op_label: str = "refresh" ) -> OAuthCredential | None: - """Exchange ``cred`` and persist the rotation; ``None`` on failure (logged). - A permanent OAuth error marks the grant dead so later calls skip the endpoint - until a new login rotates the refresh token.""" + """Exchange ``cred`` and persist the rotation; ``None`` on failure (logged). A permanent OAuth error + marks the grant dead so later calls skip the endpoint until a new login rotates the refresh token.""" try: rotated = _exchange_with_retry(cred, now=now) - except OAuthRefreshError as exc: - if exc.permanent: - _mark_grant_dead(key, cred) - logger.error( - "Honcho OAuth grant for host %s is no longer valid (%s); " - "run 'hermes honcho setup' to re-authenticate", host, exc, - ) - return None - _refresh_failure_at[key] = time.monotonic() - logger.warning("Honcho OAuth %s failed for host %s: %s", op_label, host, exc) - return None except Exception as exc: + if isinstance(exc, OAuthRefreshError) and exc.permanent: + _mark_grant_dead(key, cred) + logger.error("Honcho OAuth grant for host %s is no longer valid (%s); " + "run 'hermes honcho setup' to re-authenticate", host, exc) + return None _refresh_failure_at[key] = time.monotonic() logger.warning("Honcho OAuth %s failed for host %s: %s", op_label, host, _redact_tokens(str(exc))) return None _persist_credential(path, host, rotated) return rotated - -def _read_config(path: Path) -> dict[str, Any]: - try: - return json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return {} - - -def _atomic_write_config(path: Path, raw: dict[str, Any]) -> None: - """Write ``raw`` to ``path`` atomically with 0600 on the new file.""" - path.parent.mkdir(parents=True, exist_ok=True) - tmp = path.with_name(f".{path.name}.tmp") - fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - try: - with os.fdopen(fd, "w", encoding="utf-8") as fh: - fh.write(json.dumps(raw, indent=2) + "\n") - except Exception: - tmp.unlink(missing_ok=True) - raise - os.replace(tmp, path) - - def _deep_merge(base: dict[str, Any], overlay: dict[str, Any]) -> dict[str, Any]: - """Recursively merge ``overlay`` into ``base`` (overlay wins on scalars/lists).""" + """Recursively merge ``overlay`` into ``base`` in place (overlay wins on scalars/lists).""" for key, value in overlay.items(): if isinstance(value, dict) and isinstance(base.get(key), dict): - _deep_merge(base[key], value) - else: - base[key] = value + value = _deep_merge(base[key], value) + base[key] = value return base - def _persist_credential(path: Path, host: str, cred: OAuthCredential, raw: dict[str, Any] | None = None) -> None: """Write ``cred`` into ``host``'s block (apiKey + oauth) of ``raw`` (default: the file's current content), leaving the rest intact; marks the grant live.""" + from utils import atomic_json_write + raw = _read_config(path) if raw is None else raw block = raw.setdefault("hosts", {}).setdefault(host, {}) - block["apiKey"] = cred.access_token - block["oauth"] = cred.oauth_block() - _atomic_write_config(path, raw) + block["apiKey"], block["oauth"] = cred.access_token, cred.oauth_block() + atomic_json_write(path, raw, mode=0o600) key = (str(path), host) _expiry_cache[key] = (cred.expires_at, cred.access_token) _dead_grants.pop(key, None) _refresh_failure_at.pop(key, None) - def ensure_fresh_token( - path: Path, - host: str, - raw: dict[str, Any] | None = None, - *, - now: float | None = None, + path: Path, host: str, raw: dict[str, Any] | None = None, *, now: float | None = None ) -> tuple[str | None, bool]: - """Return ``(access_token, refreshed)`` for ``host``, refreshing if near expiry. - ``(None, False)`` when the host has no OAuth credential (plain API key). - Refresh failures are swallowed: the current (possibly stale) token comes back - with ``refreshed=False``; a permanently rejected grant is marked dead and the - session's 401 recovery escalates it to the user.""" + """Return ``(access_token, refreshed)`` for ``host``, refreshing if near expiry; ``(None, False)`` when + the host has no OAuth credential. Refresh failures are swallowed: the current (possibly stale) token comes + back with ``refreshed=False``; a permanently rejected grant is marked dead and 401 recovery escalates it.""" now = time.time() if now is None else now key = (str(path), host) - - # Hot path: trust the cached expiry while well clear of the skew window (no - # disk read). Bypassed when an explicit ``raw`` is supplied. + # Hot path: trust the cached expiry while well clear of the skew window (no disk read); bypassed + # when an explicit ``raw`` is supplied. if raw is None: cached = _expiry_cache.get(key) if cached is not None and now < cached[0] - _REFRESH_SKEW_SECONDS: return cached[1], False - cred = _load_cred(path, host, raw) if cred is None: _expiry_cache.pop(key, None) return None, False - _expiry_cache[key] = (cred.expires_at, cred.access_token) if not cred.is_expired(now=now) or _in_failure_cooldown(key): return cred.access_token, False with _refresh_lock, _config_refresh_lock(path): - # Re-read under both locks: another thread or process may have just - # rotated the token — adopt theirs instead of replaying the old one. + # Re-read under both locks: another thread/process may have just rotated — adopt theirs. current = _load_cred(path, host) or cred if not current.is_expired(now=now): return current.access_token, current.access_token != cred.access_token @@ -453,11 +327,9 @@ def ensure_fresh_token( if _grant_is_dead(key, current) or _in_failure_cooldown(key): return current.access_token, False rotated = _rotate_and_persist(path, host, key, current, now=now) - if rotated is None: - return current.access_token, False - logger.info("Honcho OAuth token refreshed for host %s", host) - return rotated.access_token, True - + if rotated is not None: + logger.info("Honcho OAuth token refreshed for host %s", host) + return (rotated.access_token, True) if rotated is not None else (current.access_token, False) def force_refresh_token(path: Path, host: str) -> str | None: """Rotate ``host``'s token now, ignoring local expiry (recovers a 401 on a @@ -478,26 +350,17 @@ def force_refresh_token(path: Path, host: str) -> str | None: _expiry_cache[key] = (cred.expires_at, cred.access_token) return cred.access_token rotated = _rotate_and_persist(path, host, key, cred, now=now, op_label="forced refresh") - if rotated is None: - return None - logger.info("Honcho OAuth token force-refreshed for host %s after an auth failure", host) - return rotated.access_token - + if rotated is not None: + logger.info("Honcho OAuth token force-refreshed for host %s after an auth failure", host) + return rotated.access_token if rotated is not None else None def install_grant( - path: Path, - host: str, - grant: dict[str, Any], - *, - client_id: str, - token_endpoint: str, - apply_config: bool = True, - now: float | None = None, + path: Path, host: str, grant: dict[str, Any], *, + client_id: str, token_endpoint: str, apply_config: bool = True, now: float | None = None, ) -> OAuthCredential: - """Apply a fresh OAuth grant (an OAuthTokenResponse dict) to ``path`` for ``host``. - Deep-merges the grant's ``config`` into the file root (preserving other hosts - and root keys), then writes the host's ``apiKey`` and ``oauth`` block. - ``apply_config=False`` stores tokens only.""" + """Apply a fresh OAuth grant (an OAuthTokenResponse dict) to ``path`` for ``host``: deep-merge the + grant's ``config`` into the file root (preserving other hosts and root keys), then write the host's + ``apiKey`` and ``oauth`` block. ``apply_config=False`` stores tokens only.""" now = time.time() if now is None else now cred = OAuthCredential.from_token_response(grant, now=now, client_id=client_id, token_endpoint=token_endpoint) raw = _read_config(path) @@ -509,11 +372,9 @@ def install_grant( _persist_credential(path, host, cred, raw) return cred - def apply_token_to_client(client: Any, token: str) -> bool: - """Rotate the live Honcho client's Bearer in place. The SDK builds its auth - header per request from ``_http.api_key``, so mutating it rotates every - holder of the singleton. Returns False on an SDK shape change (caller resets).""" + """Rotate the live Honcho client's Bearer in place. The SDK builds its auth header per request from + ``_http.api_key``, so mutating it rotates every holder of the singleton. False on an SDK shape change.""" http = getattr(client, "_http", None) if http is None or not hasattr(http, "api_key"): return False diff --git a/plugins/memory/honcho/oauth_flow.py b/plugins/memory/honcho/oauth_flow.py index 90b784a36a..f29ec7240a 100644 --- a/plugins/memory/honcho/oauth_flow.py +++ b/plugins/memory/honcho/oauth_flow.py @@ -23,41 +23,35 @@ from typing import Callable from urllib.parse import parse_qs, urlencode, urlparse from plugins.memory.honcho import oauth -from plugins.memory.honcho.client import resolve_active_host, resolve_config_path +from plugins.memory.honcho.client import HonchoClientConfig, resolve_active_host, resolve_config_path logger = logging.getLogger(__name__) -# Loopback redirect registered for the Hermes OAuth client. IP-literal so the -# browser can't resolve the advertised host to ::1 and miss the IPv4 bind. +# Loopback redirect registered for the Hermes OAuth client. IP-literal so the browser can't resolve the +# advertised host to ::1 and miss the IPv4 bind. LOOPBACK_HOST = "127.0.0.1" LOOPBACK_PORT = 8765 LOOPBACK_REDIRECT_URI = f"http://{LOOPBACK_HOST}:{LOOPBACK_PORT}/callback" - -# Pending authorizations are keyed by the CSRF ``state`` so a forged callback -# can't complete a grant; stale entries are swept after this TTL. +# Pending authorizations are keyed by the CSRF ``state`` so a forged callback can't complete a grant; +# stale entries are swept after this TTL. _PENDING_TTL_SECONDS = 600 - -# Cloud (production) hosts; dashboard serves /authorize, API serves /oauth/token. +# Dashboard serves /authorize, API serves /oauth/token. _CLOUD_DASHBOARD = "https://app.honcho.dev" _CLOUD_TOKEN_URL = "https://api.honcho.dev/oauth/token" _LOCAL_DASHBOARD = "http://localhost:3000" _LOCAL_TOKEN_URL = "http://localhost:8000/oauth/token" - -# One OAuth client for every surface (consent branding varies via ``source``), so -# there is a single grant identity to refresh — no clientId/refresh-token desync. +# One OAuth client for every surface (consent branding varies via ``source``), so there is a single +# grant identity to refresh — no clientId/refresh-token desync. _DEFAULT_CLIENT_ID = "hermes-agent" - def _display_config_path(path: object) -> str: - """Home-relative display string for the consent screen (never the write path). - A path outside ``$HOME`` yields the bare filename rather than leaking it.""" + """Home-relative display string for the consent screen (never the write path); outside ``$HOME``, the bare name.""" p = Path(str(path)) try: return "~/" + str(p.relative_to(Path.home())) except ValueError: return p.name - @dataclass(frozen=True) class OAuthEndpoints: """Resolved authorization-server URLs and client identity.""" @@ -68,35 +62,25 @@ class OAuthEndpoints: scope: str device_authorization_url: str = "" # API /oauth/device_authorization - -def _is_loopback_url(url: str | None) -> bool: - return bool(url) and any(h in url for h in ("localhost", "127.0.0.1", "::1")) - - def resolve_endpoints(environment: str | None = None, base_url: str | None = None) -> OAuthEndpoints: - """Resolve OAuth endpoints, zero-config by default: the host's honcho - ``environment`` picks cloud vs localhost, a self-hosted ``base_url`` derives - the token endpoint from the API host, and env vars override every field.""" + """Resolve OAuth endpoints, zero-config by default: the host's honcho ``environment`` picks cloud vs + localhost, a self-hosted ``base_url`` derives the token endpoint from the API host, env vars override all.""" if environment is None or base_url is None: try: - from plugins.memory.honcho.client import HonchoClientConfig - cfg = HonchoClientConfig.from_global_config() environment = environment or cfg.environment base_url = base_url if base_url is not None else cfg.base_url except Exception: environment = environment or "production" - is_local = (environment or "").lower() == "local" or _is_loopback_url(base_url) - default_dashboard = _LOCAL_DASHBOARD if is_local else _CLOUD_DASHBOARD + is_loopback = bool(base_url) and any(h in base_url for h in ("localhost", "127.0.0.1", "::1")) + is_local = (environment or "").lower() == "local" or is_loopback default_token = _LOCAL_TOKEN_URL if is_local else _CLOUD_TOKEN_URL if base_url and not is_local: # self-hosted API: token rides the same host default_token = f"{base_url.rstrip('/')}/oauth/token" - - dashboard = os.environ.get("HONCHO_OAUTH_DASHBOARD", default_dashboard).rstrip("/") + dashboard = os.environ.get("HONCHO_OAUTH_DASHBOARD", _LOCAL_DASHBOARD if is_local else _CLOUD_DASHBOARD).rstrip("/") token_url = os.environ.get("HONCHO_OAUTH_TOKEN_URL", default_token) - # Device authorization rides the token endpoint's origin. - default_device = f"{token_url.rsplit('/', 1)[0]}/device_authorization" + default_device = f"{token_url.rsplit('/', 1)[0]}/device_authorization" # rides the token endpoint's origin return OAuthEndpoints( authorize_url=os.environ.get("HONCHO_OAUTH_AUTHORIZE_URL", f"{dashboard}/authorize"), token_url=token_url, @@ -105,126 +89,77 @@ def resolve_endpoints(environment: str | None = None, base_url: str | None = Non device_authorization_url=os.environ.get("HONCHO_OAUTH_DEVICE_AUTH_URL", default_device), ) - -@dataclass -class _Pending: - verifier: str - redirect_uri: str - created_at: float - - -_pending: dict[str, _Pending] = {} +_pending: dict[str, tuple[str, str, float]] = {} # state -> (verifier, redirect_uri, created_at) _pending_lock = threading.Lock() - -def _pkce() -> tuple[str, str]: - """Return (verifier, S256 challenge) for an authorization-code request.""" - verifier = secrets.token_urlsafe(64) - challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() - return verifier, challenge - - def begin_authorization( - endpoints: OAuthEndpoints, - redirect_uri: str = LOOPBACK_REDIRECT_URI, - *, - source: str | None = None, - config_path: str | None = None, - now: float | None = None, + endpoints: OAuthEndpoints, redirect_uri: str = LOOPBACK_REDIRECT_URI, *, + source: str | None = None, config_path: str | None = None, now: float | None = None, ) -> tuple[str, str]: - """Start an authorization: return ``(authorize_url, state)`` and stash PKCE. - ``source`` tags the initiating surface for consent branding; ``config_path`` - is the home-relative *display* string (the real write path goes to - ``complete_authorization``).""" + """Start an authorization: return ``(authorize_url, state)`` and stash PKCE. ``source`` tags the initiating + surface for consent branding; ``config_path`` is the home-relative *display* string (the real write path + goes to ``complete_authorization``).""" now = time.time() if now is None else now - verifier, challenge = _pkce() + verifier = secrets.token_urlsafe(64) # PKCE: S256 challenge of a fresh verifier + challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() state = secrets.token_urlsafe(32) with _pending_lock: - for stale in [s for s, p in _pending.items() if now - p.created_at > _PENDING_TTL_SECONDS]: + for stale in [s for s, p in _pending.items() if now - p[2] > _PENDING_TTL_SECONDS]: _pending.pop(stale, None) - _pending[state] = _Pending(verifier=verifier, redirect_uri=redirect_uri, created_at=now) + _pending[state] = (verifier, redirect_uri, now) params = { "client_id": endpoints.client_id, "redirect_uri": redirect_uri, "scope": endpoints.scope, "code_challenge": challenge, "code_challenge_method": "S256", "response_type": "code", "state": state, } - if source: - params["source"] = source - if config_path: - params["config_path"] = config_path + params.update({k: v for k, v in (("source", source), ("config_path", config_path)) if v}) return f"{endpoints.authorize_url}?{urlencode(params)}", state - def _install( - endpoints: OAuthEndpoints, - grant: dict, - *, - path: Path | None, - host: str | None, - apply_config: bool, - now: float | None, - kind: str, + endpoints: OAuthEndpoints, grant: dict, *, path: Path | None, host: str | None, + apply_config: bool, now: float | None, kind: str, ) -> oauth.OAuthCredential: - """Persist ``grant`` for the target host; drop the cached client so the next - acquisition uses the new token.""" + """Persist ``grant`` for the target host; drop the cached client so the next acquisition uses the new token.""" target_host = host or resolve_active_host() cred = oauth.install_grant( - path or resolve_config_path(), - target_host, - grant, - client_id=endpoints.client_id, - token_endpoint=endpoints.token_url, - apply_config=apply_config, - now=now, + path or resolve_config_path(), target_host, grant, + client_id=endpoints.client_id, token_endpoint=endpoints.token_url, apply_config=apply_config, now=now, ) from plugins.memory.honcho.client import reset_honcho_client reset_honcho_client() logger.info("Honcho OAuth %sgrant installed for host %s", kind, target_host) return cred - def complete_authorization( - endpoints: OAuthEndpoints, - code: str, - state: str, - *, - config_path: Path | None = None, - host: str | None = None, - apply_config: bool = True, - now: float | None = None, + endpoints: OAuthEndpoints, code: str, state: str, *, + config_path: Path | None = None, host: str | None = None, apply_config: bool = True, now: float | None = None, ) -> oauth.OAuthCredential: - """Exchange ``code`` for a grant and persist it. Raises on bad state/exchange. - ``apply_config=False`` stores tokens only (CLI path: settings stay wizard-owned).""" + """Exchange ``code`` for a grant and persist it. Raises on bad state/exchange. ``apply_config=False`` + stores tokens only (CLI path: settings stay wizard-owned).""" with _pending_lock: pending = _pending.pop(state, None) if pending is None: raise ValueError("unknown or expired authorization state") - - _status_code, grant = oauth._http_json( - "POST", endpoints.token_url, timeout=oauth._REFRESH_TIMEOUT_SECONDS, - data={"grant_type": "authorization_code", "client_id": endpoints.client_id, "code": code, - "redirect_uri": pending.redirect_uri, "code_verifier": pending.verifier}, - ) + verifier, redirect_uri, _ = pending + form = {"grant_type": "authorization_code", "client_id": endpoints.client_id, "code": code, + "redirect_uri": redirect_uri, "code_verifier": verifier} + _, grant = oauth._http_json("POST", endpoints.token_url, timeout=oauth._REFRESH_TIMEOUT_SECONDS, data=form) return _install(endpoints, grant, path=config_path, host=host, apply_config=apply_config, now=now, kind="") - -_CALLBACK_STYLE = ( +_CALLBACK_PAGE = ( + "{title}" "" + "display:flex;align-items:center;justify-content:center;height:100vh;margin:0'>
{body}
" ) -_CALLBACK_HTML = ( - "Honcho connected" + _CALLBACK_STYLE - + "
Connected to Honcho. You can close this tab and return to Hermes.
" +_CALLBACK_HTML = _CALLBACK_PAGE.format( + title="Honcho connected", body="Connected to Honcho. You can close this tab and return to Hermes." ).encode() -_CALLBACK_ERROR_HTML = ( - "Honcho sign-in failed" + _CALLBACK_STYLE - + "
Sign-in was not completed ({error}). You can close this tab and re-run setup.
" +_CALLBACK_ERROR_HTML = _CALLBACK_PAGE.format( # ``{error}`` is filled per request + title="Honcho sign-in failed", body="Sign-in was not completed ({error}). You can close this tab and re-run setup." ) - def _bind_loopback_server() -> tuple[HTTPServer, dict[str, str]]: - """Bind the one-shot callback server, returning it and its capture dict. - Prefers :8765, else an OS-assigned port (the AS relaxes the port for loopback - redirect URIs; the caller advertises the bound port).""" + """Bind the one-shot callback server, returning it and its capture dict. Prefers :8765, else an + OS-assigned port (the AS relaxes the port for loopback redirect URIs; the caller advertises the bound port).""" captured: dict[str, str] = {} class _Handler(BaseHTTPRequestHandler): @@ -240,115 +175,88 @@ def _bind_loopback_server() -> tuple[HTTPServer, dict[str, str]]: self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() - if captured["error"]: - self.wfile.write(_CALLBACK_ERROR_HTML.format(error=html.escape(captured["error"])).encode("utf-8")) - else: - self.wfile.write(_CALLBACK_HTML) + error = captured["error"] + self.wfile.write(_CALLBACK_ERROR_HTML.format(error=html.escape(error)).encode() if error else _CALLBACK_HTML) def log_message(self, *args): # silence stdlib request logging return try: - server = HTTPServer((LOOPBACK_HOST, LOOPBACK_PORT), _Handler) + return HTTPServer((LOOPBACK_HOST, LOOPBACK_PORT), _Handler), captured except OSError: - server = HTTPServer((LOOPBACK_HOST, 0), _Handler) - return server, captured - + return HTTPServer((LOOPBACK_HOST, 0), _Handler), captured def capture_loopback_code(server: HTTPServer, captured: dict[str, str], *, timeout: float = 300.0) -> tuple[str, str]: - """Serve ``/callback`` until our code lands; return ``(code, state)``. - Loops so a stray probe to another path doesn't end the wait; raises - ``TimeoutError`` if nothing arrives within ``timeout``.""" + """Serve ``/callback`` until our code lands; return ``(code, state)``. Loops so a stray probe to another + path doesn't end the wait; raises ``TimeoutError`` if nothing arrives within ``timeout``.""" server.timeout = timeout + deadline = time.monotonic() + timeout try: - deadline = time.monotonic() + timeout while "code" not in captured and time.monotonic() < deadline: server.handle_request() finally: server.server_close() - if captured.get("error"): + if error := captured.get("error"): detail = captured.get("error_description") - raise ValueError(f"authorization denied: {captured['error']}{f' ({detail})' if detail else ''}") + raise ValueError(f"authorization denied: {error}{f' ({detail})' if detail else ''}") if "code" not in captured: raise TimeoutError("no OAuth callback received before timeout") return captured["code"], captured.get("state", "") - def authorize_via_loopback( - *, - config_path: Path | None = None, - host: str | None = None, - source: str | None = None, - apply_config: bool = True, - open_url: Callable[[str], None] | None = None, - timeout: float = 300.0, + *, config_path: Path | None = None, host: str | None = None, source: str | None = None, + apply_config: bool = True, open_url: Callable[[str], None] | None = None, timeout: float = 300.0, ) -> oauth.OAuthCredential: - """Full loopback flow: open browser → capture code → exchange → persist. - ``open_url`` (default: system browser) always receives the authorize URL, so - a CLI caller can print it for browserless setups.""" + """Full loopback flow: open browser → capture code → exchange → persist. ``open_url`` (default: system + browser) always receives the authorize URL, so a CLI caller can print it for browserless setups.""" # Bind first so the advertised redirect_uri carries the actual bound port. server, captured = _bind_loopback_server() redirect_uri = f"http://{LOOPBACK_HOST}:{server.server_address[1]}/callback" - endpoints = resolve_endpoints() path = config_path or resolve_config_path() - authorize_url, state = begin_authorization( - endpoints, redirect_uri, source=source, config_path=_display_config_path(path) - ) - + authorize_url, state = begin_authorization(endpoints, redirect_uri, source=source, + config_path=_display_config_path(path)) if open_url is None: import webbrowser open_url = webbrowser.open - # Socket is already bound, so a fast redirect can't beat the browser thread. threading.Thread(target=lambda: open_url(authorize_url), daemon=True).start() - code, returned_state = capture_loopback_code(server, captured, timeout=timeout) if returned_state != state: raise ValueError("OAuth state mismatch — possible CSRF, aborting") - return complete_authorization( - endpoints, code, returned_state, config_path=path, host=host, apply_config=apply_config - ) + return complete_authorization(endpoints, code, returned_state, config_path=path, host=host, + apply_config=apply_config) # — Device authorization grant (RFC 8628), for headless / remote-VM clients — DEVICE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:device_code" - # RFC 8628 §3.5: slow_down adds 5s per response; cap matches the server's DEVICE_POLL_INTERVAL_MAX. _SLOW_DOWN_STEP = 5 _POLL_INTERVAL_CAP = 60 - # RFC 8414 metadata; advertising the device grant marks a host as device-login capable. _AS_METADATA_PATH = "/.well-known/oauth-authorization-server" - class DeviceFlowError(RuntimeError): """A device-flow request failed. ``error`` is the RFC error code when known.""" def __init__(self, error: str, description: str | None = None): - self.error = error - self.description = description + self.error, self.description = error, description super().__init__(f"{error}: {description}" if description else error) - class AccessDenied(DeviceFlowError): """The user denied the authorization request.""" - class DeviceCodeExpired(DeviceFlowError): """The device code expired before the user approved it.""" - class AuthorizationTimeout(DeviceFlowError): """Polling ran past the device code's lifetime with no decision.""" - # Terminal server outcomes for a device-token poll (RFC 8628 §3.5). _DEVICE_POLL_ERRORS = {"access_denied": AccessDenied, "expired_token": DeviceCodeExpired} - @dataclass(frozen=True) class DeviceCode: """RFC 8628 §3.2 device authorization response.""" @@ -360,10 +268,8 @@ class DeviceCode: expires_in: int interval: int - def supports_device_login(endpoints: OAuthEndpoints, *, timeout: float = 5.0) -> bool: - """Whether the host advertises the device grant in its RFC 8414 metadata. - Fails closed on any error / non-200 / missing capability.""" + """Whether the host advertises the device grant in its RFC 8414 metadata. Fails closed on any error.""" origin = endpoints.token_url.rsplit("/oauth/", 1)[0] try: body = oauth._http_json("GET", f"{origin}{_AS_METADATA_PATH}", timeout=timeout)[1] @@ -372,45 +278,35 @@ def supports_device_login(endpoints: OAuthEndpoints, *, timeout: float = 5.0) -> grants = body.get("grant_types_supported") if isinstance(body, dict) else None return isinstance(grants, list) and DEVICE_GRANT_TYPE in grants - def request_device_code(endpoints: OAuthEndpoints, *, source: str | None = None) -> DeviceCode: """Request a device + user code pair (RFC 8628 §3.1).""" if not endpoints.device_authorization_url: raise ValueError("no device authorization endpoint resolved") - data = {"client_id": endpoints.client_id, "scope": endpoints.scope} - if source: - data["source"] = source - status, body = oauth._http_post_form_status(endpoints.device_authorization_url, data, oauth._REFRESH_TIMEOUT_SECONDS) + data = {"client_id": endpoints.client_id, "scope": endpoints.scope, **({"source": source} if source else {})} + url = endpoints.device_authorization_url + status, body = oauth._http_post_form_status(url, data, oauth._REFRESH_TIMEOUT_SECONDS) if status != 200: raise DeviceFlowError(str(body.get("error") or f"http_{status}"), body.get("error_description")) try: - verification_uri = body["verification_uri"] + uri = body["verification_uri"] return DeviceCode( - device_code=body["device_code"], - user_code=body["user_code"], - verification_uri=verification_uri, - verification_uri_complete=body.get("verification_uri_complete", f"{verification_uri}?user_code={body['user_code']}"), - expires_in=int(body["expires_in"]), - interval=int(body.get("interval", 5)), # §3.2: optional, default 5s + device_code=body["device_code"], user_code=body["user_code"], verification_uri=uri, + verification_uri_complete=body.get("verification_uri_complete", f"{uri}?user_code={body['user_code']}"), + expires_in=int(body["expires_in"]), interval=int(body.get("interval", 5)), # §3.2: default 5s ) except (KeyError, TypeError, ValueError) as e: raise DeviceFlowError("invalid_response", f"malformed device authorization response: {e}") from e - def poll_for_token( - endpoints: OAuthEndpoints, - device: DeviceCode, - *, - on_poll: Callable[[], None] | None = None, - sleep: Callable[[float], None] = time.sleep, - monotonic: Callable[[], float] = time.monotonic, + endpoints: OAuthEndpoints, device: DeviceCode, *, on_poll: Callable[[], None] | None = None, + sleep: Callable[[float], None] = time.sleep, monotonic: Callable[[], float] = time.monotonic, ) -> dict[str, object]: - """Poll the token endpoint until approved (RFC 8628 §3.4/§3.5). - Sleeps ``interval`` before each poll, bumping it on ``slow_down``. Raises - ``AccessDenied`` / ``DeviceCodeExpired`` on terminal outcomes and + """Poll the token endpoint until approved (RFC 8628 §3.4/§3.5). Sleeps ``interval`` before each poll, + bumping it on ``slow_down``. Raises ``AccessDenied`` / ``DeviceCodeExpired`` on terminal outcomes and ``AuthorizationTimeout`` when ``expires_in`` elapses with no decision.""" import httpx + form = {"grant_type": DEVICE_GRANT_TYPE, "device_code": device.device_code, "client_id": endpoints.client_id} interval = max(1, min(device.interval, _POLL_INTERVAL_CAP)) deadline = monotonic() + max(1, device.expires_in) while True: @@ -420,11 +316,7 @@ def poll_for_token( if on_poll: on_poll() try: - status, body = oauth._http_post_form_status( - endpoints.token_url, - {"grant_type": DEVICE_GRANT_TYPE, "device_code": device.device_code, "client_id": endpoints.client_id}, - oauth._REFRESH_TIMEOUT_SECONDS, - ) + status, body = oauth._http_post_form_status(endpoints.token_url, form, oauth._REFRESH_TIMEOUT_SECONDS) except httpx.TransportError as e: # a network blip mid-poll shouldn't kill a 10-minute wait logger.debug("device token poll transport error, retrying: %s", e) continue @@ -434,28 +326,19 @@ def poll_for_token( raise DeviceFlowError("invalid_response", "token response missing access_token") return body error = str(body.get("error") or f"http_{status}") - if error == "authorization_pending": - continue if error == "slow_down": interval = min(interval + _SLOW_DOWN_STEP, _POLL_INTERVAL_CAP) - continue - raise _DEVICE_POLL_ERRORS.get(error, DeviceFlowError)(error, body.get("error_description")) - + elif error != "authorization_pending": + raise _DEVICE_POLL_ERRORS.get(error, DeviceFlowError)(error, body.get("error_description")) def authorize_via_device_code( - *, - config_path: Path | None = None, - host: str | None = None, - source: str | None = None, - apply_config: bool = True, - display: Callable[[DeviceCode], None] | None = None, - open_url: Callable[[str], None] | None = None, - on_poll: Callable[[], None] | None = None, + *, config_path: Path | None = None, host: str | None = None, source: str | None = None, + apply_config: bool = True, display: Callable[[DeviceCode], None] | None = None, + open_url: Callable[[str], None] | None = None, on_poll: Callable[[], None] | None = None, sleep: Callable[[float], None] = time.sleep, ) -> oauth.OAuthCredential: - """Full device flow: request codes → show user code → poll → persist. - ``open_url`` (if given) receives ``verification_uri_complete``; no default - browser open, since the approving browser may be on another machine.""" + """Full device flow: request codes → show user code → poll → persist. ``open_url`` (if given) receives + ``verification_uri_complete``; no default browser open, since the approving browser may be on another machine.""" endpoints = resolve_endpoints() path = config_path or resolve_config_path() # resolve NOW so a later ambient lookup can't drift target_host = host or resolve_active_host() @@ -465,41 +348,32 @@ def authorize_via_device_code( display(device) if open_url: open_url(device.verification_uri_complete) - grant = poll_for_token(endpoints, device, on_poll=on_poll, sleep=sleep) - return _install(endpoints, grant, path=path, host=target_host, apply_config=apply_config, now=None, kind="device ") + return _install(endpoints, grant, path=path, host=target_host, apply_config=apply_config, now=None, + kind="device ") # — Background launcher + status, for the desktop "Connect" button — the flow # blocks on a browser round-trip, so web_server runs it in a thread and the UI polls. - @dataclass class FlowStatus: state: str = "idle" # idle | pending | connected | error detail: str = "" - _status = FlowStatus() _status_lock = threading.Lock() _flow_thread: threading.Thread | None = None - def _detect_connection() -> tuple[bool, str | None]: """Report whether a credential is already stored: 'oauth', 'apikey', or none.""" try: - from plugins.memory.honcho.client import HonchoClientConfig - cfg = HonchoClientConfig.from_global_config() block = (cfg.raw.get("hosts") or {}).get(cfg.host) or {} - if oauth.OAuthCredential.from_host_block(block) is not None: - return True, "oauth" - if cfg.api_key: - return True, "apikey" + auth = "oauth" if oauth.OAuthCredential.from_host_block(block) is not None else "apikey" if cfg.api_key else None except Exception: - pass - return False, None - + auth = None + return auth is not None, auth def get_flow_status() -> dict[str, object]: with _status_lock: @@ -507,24 +381,18 @@ def get_flow_status() -> dict[str, object]: connected, auth = _detect_connection() return {"state": state, "detail": detail, "connected": connected, "auth": auth} - def _set_status(state: str, detail: str = "") -> None: with _status_lock: _status.state, _status.detail = state, detail - def start_loopback_flow_background( - *, - config_path: Path | None = None, - host: str | None = None, - source: str = "hermes-desktop", + *, config_path: Path | None = None, host: str | None = None, source: str = "hermes-desktop", timeout: float = 300.0, ) -> dict[str, str]: """Launch the loopback flow in a daemon thread; returns the initial status. Idempotent while pending, so a double-click can't open two tabs / bind :8765 twice.""" global _flow_thread - # Resolve under the caller's profile scope NOW — a context-local HERMES_HOME - # override can't reach the worker thread. + # Resolve under the caller's profile scope NOW — a context-local HERMES_HOME override can't reach the worker. config_path = config_path or resolve_config_path() host = host or resolve_active_host() with _status_lock: