ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(honcho): compact oauth + oauth_flow; drop _atomic_write_config for utils.atomic_json_write

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:07:48 -07:00
parent 58f5d23f78
commit cc67af75c4
2 changed files with 185 additions and 456 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/memory/honcho/oauth.py
+91 -230
View File
@@ -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 (``<config>.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 (``<config>.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
plugins/memory/honcho/oauth_flow.py
+94 -226
View File
@@ -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 = (
"<!doctype html><meta charset=utf-8><title>{title}</title>"
"<body style='font:14px ui-monospace,monospace;background:#0b0e14;color:#c9d1d9;"
"display:flex;align-items:center;justify-content:center;height:100vh;margin:0'>"
"display:flex;align-items:center;justify-content:center;height:100vh;margin:0'><div>{body}</div>"
)
_CALLBACK_HTML = (
"<!doctype html><meta charset=utf-8><title>Honcho connected</title>" + _CALLBACK_STYLE
+ "<div>Connected to Honcho. You can close this tab and return to Hermes.</div>"
_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 = (
"<!doctype html><meta charset=utf-8><title>Honcho sign-in failed</title>" + _CALLBACK_STYLE
+ "<div>Sign-in was not completed ({error}). You can close this tab and re-run setup.</div>"
_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: