mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-24 16:54:43 +00:00
feat(honcho): add OAuth device-code login (RFC 8628) for headless environments
Adds a device authorization grant flow alongside the existing loopback OAuth flow, so `hermes setup` can connect to Honcho cloud from SSH and other no-browser environments. - oauth.py: new HTTP seams — _http_post_form_status (non-raising, since RFC 8628 polling reads the OAuth error off a 400) and _http_get_json for the RFC 8414 metadata probe - oauth_flow.py: DeviceCode, request_device_code, poll_for_token with slow_down backoff (+5s, capped at 60s) bounded by expires_in, typed errors (AccessDenied, DeviceCodeExpired, AuthorizationTimeout), and supports_device_login (fail-closed metadata gate); device flow ends in the same install_grant tail as loopback so refresh/status work unchanged - oauth_flow.py: loopback callback now serves a "sign-in was not completed" page on consent cancel instead of the success page - cli.py: cloud menu offers oauth / device / apikey; the device option only appears when the host advertises the grant, and becomes the default when no browser is detected - 18 new tests covering the full flow against a local fake AS, backoff schedule, error mapping, deadline bound, metadata gate, and wizard branches Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Aakash Kattelu
Teknium
parent
a61183b56f
commit
2aa359ea70
7 changed files with 742 additions and 19 deletions
|
|
@ -17,10 +17,12 @@ hermes memory setup honcho # configure Honcho directly (works on a fresh insta
|
|||
hermes memory setup # generic picker, choose Honcho from the list
|
||||
```
|
||||
|
||||
For cloud, the wizard asks **OAuth or API key**. OAuth opens a browser
|
||||
sign-in and stores the grant itself — nothing to copy; tokens refresh
|
||||
automatically. The desktop app offers the same flow as a **Connect** link
|
||||
next to the memory-provider dropdown.
|
||||
For cloud, the wizard asks **OAuth, device code, or API key**. OAuth opens a
|
||||
browser sign-in and stores the grant itself — nothing to copy; tokens refresh
|
||||
automatically. On SSH/headless machines choose **device**: the CLI prints a
|
||||
short code and a link you open in a browser on any other machine; setup
|
||||
completes once you approve there. The desktop app offers the browser flow as
|
||||
a **Connect** link next to the memory-provider dropdown.
|
||||
|
||||
Or manually:
|
||||
```bash
|
||||
|
|
@ -347,6 +349,7 @@ Presets:
|
|||
| `HONCHO_OAUTH_DASHBOARD` | OAuth authorize origin (default: cloud dashboard; local-dev `localhost:3000`) |
|
||||
| `HONCHO_OAUTH_AUTHORIZE_URL` | Full authorize URL (overrides the dashboard origin) |
|
||||
| `HONCHO_OAUTH_TOKEN_URL` | Token endpoint (default: cloud API; local-dev `localhost:8000`) |
|
||||
| `HONCHO_OAUTH_DEVICE_AUTH_URL` | Device-authorization endpoint (default: derived from the token URL) |
|
||||
| `HONCHO_OAUTH_CLIENT_ID` | OAuth client (default `hermes-agent`) |
|
||||
| `HONCHO_OAUTH_SCOPE` | Requested scope (default `write`) |
|
||||
|
||||
|
|
|
|||
|
|
@ -620,7 +620,7 @@ def cmd_setup(args) -> None:
|
|||
print("\n No local JWT set. Local no-auth ready.")
|
||||
use_oauth = False
|
||||
if not is_local:
|
||||
# --- Cloud: OAuth (browser) or API key ---
|
||||
# --- Cloud: OAuth (browser), device code, or API key ---
|
||||
cfg.pop("baseUrl", None) # cloud uses SDK default
|
||||
|
||||
# Detect an existing OAuth grant so re-running setup reflects it instead
|
||||
|
|
@ -628,15 +628,88 @@ def cmd_setup(args) -> None:
|
|||
from plugins.memory.honcho.oauth import OAuthCredential
|
||||
existing_oauth = OAuthCredential.from_host_block(hermes_host)
|
||||
|
||||
device_available = _device_login_available()
|
||||
is_remote, can_browse = _headless()
|
||||
|
||||
print("\n Auth method:")
|
||||
if existing_oauth is not None:
|
||||
print(f" (currently connected via OAuth — client {existing_oauth.client_id})")
|
||||
print(" oauth -- sign in via browser (recommended)")
|
||||
print(" oauth -- sign in via browser on this machine (recommended)")
|
||||
if device_available:
|
||||
print(" device -- device code: approve from a browser on another machine (SSH / headless)")
|
||||
print(" apikey -- paste an API key from https://app.honcho.dev")
|
||||
method = _prompt("OAuth or API key?", default="oauth").strip().lower()
|
||||
use_oauth = method in {"oauth", "o"}
|
||||
|
||||
if use_oauth:
|
||||
default_method = "oauth"
|
||||
if is_remote or not can_browse:
|
||||
if device_available:
|
||||
print(" (no usable local browser detected — device code recommended)")
|
||||
default_method = "device"
|
||||
else:
|
||||
print(" (no usable local browser detected — browser sign-in may need an SSH tunnel to 127.0.0.1:8765)")
|
||||
prompt_label = "oauth, device, or apikey?" if device_available else "OAuth or API key?"
|
||||
method = _prompt(prompt_label, default=default_method).strip().lower()
|
||||
use_oauth = method in {"oauth", "o"}
|
||||
use_device = device_available and method in {"device", "d"}
|
||||
|
||||
if use_device:
|
||||
from plugins.memory.honcho.oauth_flow import (
|
||||
AccessDenied,
|
||||
AuthorizationTimeout,
|
||||
DeviceCode,
|
||||
DeviceCodeExpired,
|
||||
DeviceFlowError,
|
||||
authorize_via_device_code,
|
||||
)
|
||||
|
||||
def _show(device: DeviceCode) -> None:
|
||||
print("\n To connect, on any device with a browser:")
|
||||
print(f"\n 1. Open {device.verification_uri}")
|
||||
print(f" 2. Enter {device.user_code}")
|
||||
print(f"\n Or open directly:\n\n {device.verification_uri_complete}\n")
|
||||
mins = max(1, device.expires_in // 60)
|
||||
print(f" Waiting for approval (expires in {mins} min, Ctrl-C to cancel) ", end="", flush=True)
|
||||
|
||||
def _open_local(url: str) -> None:
|
||||
import webbrowser
|
||||
|
||||
webbrowser.open(url)
|
||||
|
||||
print("\n Requesting device code…")
|
||||
try:
|
||||
cred = authorize_via_device_code(
|
||||
config_path=write_path,
|
||||
source="hermes-cli",
|
||||
apply_config=False,
|
||||
display=_show,
|
||||
open_url=_open_local if can_browse and not is_remote else None,
|
||||
on_poll=lambda: print(".", end="", flush=True),
|
||||
)
|
||||
except KeyboardInterrupt:
|
||||
print("\n Cancelled. Re-run 'hermes honcho setup' to try again.\n")
|
||||
return
|
||||
except (AuthorizationTimeout, DeviceCodeExpired):
|
||||
print("\n Device code expired before approval.")
|
||||
print(" Re-run 'hermes honcho setup' to get a new code.\n")
|
||||
return
|
||||
except AccessDenied:
|
||||
print("\n Sign-in was denied on the approval page.")
|
||||
print(" Re-run 'hermes honcho setup' to retry, or choose an API key instead.\n")
|
||||
return
|
||||
except DeviceFlowError as e:
|
||||
if e.error == "http_429":
|
||||
print("\n Too many device-code requests — wait a minute and re-run setup.\n")
|
||||
else:
|
||||
print(f"\n Device sign-in failed: {e}")
|
||||
print(" Re-run 'hermes honcho setup' to retry, or choose an API key instead.\n")
|
||||
return
|
||||
except Exception as e:
|
||||
print(f"\n Device sign-in failed: {e}")
|
||||
print(" Re-run 'hermes honcho setup' to retry, or choose an API key instead.\n")
|
||||
return
|
||||
print(" approved")
|
||||
_apply_grant_to_host(hermes_host, cred)
|
||||
print(" Authorized — token saved. Let's finish configuring.\n")
|
||||
elif use_oauth:
|
||||
# Sign in now, up front — the browser link is the whole point, so
|
||||
# don't bury it behind the identity prompts. The grant's tokens are
|
||||
# merged into the in-memory cfg so the wizard's final save preserves
|
||||
|
|
@ -661,11 +734,7 @@ def cmd_setup(args) -> None:
|
|||
print(f" OAuth sign-in failed: {e}")
|
||||
print(" Re-run 'hermes honcho setup' to retry, or choose an API key instead.\n")
|
||||
return
|
||||
hermes_host["apiKey"] = cred.access_token
|
||||
hermes_host["oauth"] = cred.oauth_block()
|
||||
# Default the peer prompt to the name entered at consent.
|
||||
if cred.consent_peer_name:
|
||||
hermes_host["peerName"] = cred.consent_peer_name
|
||||
_apply_grant_to_host(hermes_host, cred)
|
||||
print(" Authorized — token saved. Let's finish configuring.\n")
|
||||
else:
|
||||
current_key = cfg.get("apiKey", "")
|
||||
|
|
@ -974,6 +1043,35 @@ def cmd_setup(args) -> None:
|
|||
print(" hermes honcho map <name> -- map this directory to a session name\n")
|
||||
|
||||
|
||||
def _device_login_available() -> bool:
|
||||
"""Whether the resolved host offers the RFC 8628 device grant. Fails closed."""
|
||||
try:
|
||||
from plugins.memory.honcho.oauth_flow import resolve_endpoints, supports_device_login
|
||||
|
||||
return supports_device_login(resolve_endpoints())
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _headless() -> tuple[bool, bool]:
|
||||
"""(is_remote, can_open_browser) — degrades safely if hermes_cli internals move."""
|
||||
try:
|
||||
from hermes_cli.auth import _can_open_graphical_browser, _is_remote_session
|
||||
|
||||
return _is_remote_session(), _can_open_graphical_browser()
|
||||
except Exception:
|
||||
return False, True
|
||||
|
||||
|
||||
def _apply_grant_to_host(hermes_host: dict, cred) -> None:
|
||||
"""Store an OAuth grant on the host block; the wizard's final save persists it."""
|
||||
hermes_host["apiKey"] = cred.access_token
|
||||
hermes_host["oauth"] = cred.oauth_block()
|
||||
# Default the peer prompt to the name entered at consent.
|
||||
if cred.consent_peer_name:
|
||||
hermes_host["peerName"] = cred.consent_peer_name
|
||||
|
||||
|
||||
def _active_profile_name() -> str:
|
||||
"""Return the active Hermes profile name (respects --target-profile override)."""
|
||||
if _profile_override:
|
||||
|
|
|
|||
|
|
@ -175,6 +175,36 @@ def _http_post_form(url: str, data: dict[str, str], timeout: float) -> dict[str,
|
|||
return resp.json()
|
||||
|
||||
|
||||
def _http_post_form_status(
|
||||
url: str, data: dict[str, str], timeout: float
|
||||
) -> tuple[int, dict[str, Any]]:
|
||||
"""POST form-encoded ``data``; return ``(status, parsed JSON body)``.
|
||||
|
||||
Unlike ``_http_post_form``, 4xx does not raise — RFC 8628 polling reads the
|
||||
OAuth error body off a 400. A non-JSON body parses to ``{}``.
|
||||
"""
|
||||
import httpx
|
||||
|
||||
resp = httpx.post(url, data=data, timeout=timeout)
|
||||
try:
|
||||
body = resp.json()
|
||||
except ValueError:
|
||||
body = {}
|
||||
if not isinstance(body, dict):
|
||||
body = {}
|
||||
return resp.status_code, body
|
||||
|
||||
|
||||
def _http_get_json(url: str, timeout: float) -> dict[str, Any]:
|
||||
"""GET ``url`` and return the parsed JSON body. Raises on non-2xx/non-JSON."""
|
||||
import httpx
|
||||
|
||||
resp = httpx.get(url, timeout=timeout)
|
||||
resp.raise_for_status()
|
||||
body = resp.json()
|
||||
return body if isinstance(body, dict) else {}
|
||||
|
||||
|
||||
def _exchange_refresh_token(cred: OAuthCredential, *, now: float) -> OAuthCredential:
|
||||
"""Run the refresh_token grant and return the rotated credential.
|
||||
|
||||
|
|
|
|||
|
|
@ -62,6 +62,7 @@ class OAuthEndpoints:
|
|||
token_url: str # API /oauth/token
|
||||
client_id: str
|
||||
scope: str
|
||||
device_authorization_url: str = "" # API /oauth/device_authorization
|
||||
|
||||
|
||||
# Cloud (production) hosts; dashboard serves /authorize, API serves /oauth/token.
|
||||
|
|
@ -107,11 +108,15 @@ def resolve_endpoints(
|
|||
default_token = f"{base_url.rstrip('/')}/oauth/token"
|
||||
|
||||
dashboard = os.environ.get("HONCHO_OAUTH_DASHBOARD", default_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"
|
||||
return OAuthEndpoints(
|
||||
authorize_url=os.environ.get("HONCHO_OAUTH_AUTHORIZE_URL", f"{dashboard}/authorize"),
|
||||
token_url=os.environ.get("HONCHO_OAUTH_TOKEN_URL", default_token),
|
||||
token_url=token_url,
|
||||
client_id=os.environ.get("HONCHO_OAUTH_CLIENT_ID", _DEFAULT_CLIENT_ID),
|
||||
scope=os.environ.get("HONCHO_OAUTH_SCOPE", "write"),
|
||||
device_authorization_url=os.environ.get("HONCHO_OAUTH_DEVICE_AUTH_URL", default_device),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -240,6 +245,14 @@ _CALLBACK_HTML = (
|
|||
b"<div>Connected to Honcho. You can close this tab and return to Hermes.</div>"
|
||||
)
|
||||
|
||||
_CALLBACK_ERROR_HTML = (
|
||||
"<!doctype html><meta charset=utf-8>"
|
||||
"<title>Honcho sign-in failed</title>"
|
||||
"<body style='font:14px ui-monospace,monospace;background:#0b0e14;color:#c9d1d9;"
|
||||
"display:flex;align-items:center;justify-content:center;height:100vh;margin:0'>"
|
||||
"<div>Sign-in was not completed ({error}). You can close this tab and re-run setup.</div>"
|
||||
)
|
||||
|
||||
|
||||
def _bind_loopback_server() -> tuple[HTTPServer, dict[str, str]]:
|
||||
"""Bind the one-shot callback server, returning it and its capture dict.
|
||||
|
|
@ -262,10 +275,17 @@ def _bind_loopback_server() -> tuple[HTTPServer, dict[str, str]]:
|
|||
captured["code"] = (params.get("code") or [""])[0]
|
||||
captured["state"] = (params.get("state") or [""])[0]
|
||||
captured["error"] = (params.get("error") or [""])[0]
|
||||
captured["error_description"] = (params.get("error_description") or [""])[0]
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.end_headers()
|
||||
self.wfile.write(_CALLBACK_HTML)
|
||||
if captured["error"]:
|
||||
import html as _html
|
||||
|
||||
page = _CALLBACK_ERROR_HTML.format(error=_html.escape(captured["error"]))
|
||||
self.wfile.write(page.encode("utf-8"))
|
||||
else:
|
||||
self.wfile.write(_CALLBACK_HTML)
|
||||
|
||||
def log_message(self, *args): # silence stdlib request logging
|
||||
return
|
||||
|
|
@ -296,7 +316,9 @@ def capture_loopback_code(
|
|||
server.server_close()
|
||||
|
||||
if captured.get("error"):
|
||||
raise ValueError(f"authorization denied: {captured['error']}")
|
||||
detail = captured.get("error_description")
|
||||
suffix = f" ({detail})" if detail else ""
|
||||
raise ValueError(f"authorization denied: {captured['error']}{suffix}")
|
||||
if "code" not in captured:
|
||||
raise TimeoutError("no OAuth callback received before timeout")
|
||||
return captured["code"], captured.get("state", "")
|
||||
|
|
@ -352,6 +374,208 @@ def authorize_via_loopback(
|
|||
)
|
||||
|
||||
|
||||
# — Device authorization grant (RFC 8628), for headless / remote-VM clients —
|
||||
# The loopback flow needs the browser on the same machine; here the CLI prints
|
||||
# a short user code, the user approves from any browser (dashboard /device),
|
||||
# and the device polls the token endpoint until the grant lands.
|
||||
|
||||
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 so a misbehaving clock can't inflate past it.
|
||||
_SLOW_DOWN_STEP = 5
|
||||
_POLL_INTERVAL_CAP = 60
|
||||
|
||||
# RFC 8414 authorization-server metadata; advertising the device grant is what
|
||||
# distinguishes a host that can do device login from one that can't.
|
||||
_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
|
||||
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."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DeviceCode:
|
||||
"""RFC 8628 §3.2 device authorization response."""
|
||||
|
||||
device_code: str
|
||||
user_code: str
|
||||
verification_uri: str
|
||||
verification_uri_complete: str
|
||||
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: any connection error, non-200, or missing capability returns
|
||||
False, so hosts without the device grant simply don't offer the option.
|
||||
"""
|
||||
origin = endpoints.token_url.rsplit("/oauth/", 1)[0]
|
||||
try:
|
||||
body = oauth._http_get_json(f"{origin}{_AS_METADATA_PATH}", timeout)
|
||||
except Exception:
|
||||
return False
|
||||
grants = body.get("grant_types_supported")
|
||||
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
|
||||
)
|
||||
if status != 200:
|
||||
error = str(body.get("error") or f"http_{status}")
|
||||
raise DeviceFlowError(error, body.get("error_description"))
|
||||
try:
|
||||
verification_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["interval"]),
|
||||
)
|
||||
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,
|
||||
) -> dict[str, object]:
|
||||
"""Poll the token endpoint until the grant is approved (RFC 8628 §3.4/§3.5).
|
||||
|
||||
Sleeps ``interval`` before each poll, bumping it on ``slow_down``. Raises
|
||||
``AccessDenied`` / ``DeviceCodeExpired`` on the terminal server outcomes and
|
||||
``AuthorizationTimeout`` when ``expires_in`` elapses with no decision.
|
||||
``sleep`` / ``monotonic`` are injectable for tests.
|
||||
"""
|
||||
import httpx
|
||||
|
||||
interval = max(1, min(device.interval, _POLL_INTERVAL_CAP))
|
||||
deadline = monotonic() + max(1, device.expires_in)
|
||||
while True:
|
||||
if monotonic() + interval >= deadline:
|
||||
raise AuthorizationTimeout(
|
||||
"expired_token", "timed out waiting for approval"
|
||||
)
|
||||
sleep(interval)
|
||||
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,
|
||||
)
|
||||
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
|
||||
|
||||
if status == 200:
|
||||
if not body.get("access_token"):
|
||||
raise DeviceFlowError("invalid_response", "token response missing access_token")
|
||||
return body
|
||||
error = str(body.get("error") or f"http_{status}")
|
||||
description = body.get("error_description")
|
||||
if error == "authorization_pending":
|
||||
continue
|
||||
if error == "slow_down":
|
||||
interval = min(interval + _SLOW_DOWN_STEP, _POLL_INTERVAL_CAP)
|
||||
continue
|
||||
if error == "access_denied":
|
||||
raise AccessDenied(error, description)
|
||||
if error == "expired_token":
|
||||
raise DeviceCodeExpired(error, description)
|
||||
raise DeviceFlowError(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,
|
||||
sleep: Callable[[float], None] = time.sleep,
|
||||
) -> oauth.OAuthCredential:
|
||||
"""Drive the full device flow: request codes → show user code → poll → persist.
|
||||
|
||||
``display`` shows the user code + verification URL. ``open_url`` (if given)
|
||||
receives ``verification_uri_complete`` — there is no default browser open,
|
||||
since the approving browser may be on another machine.
|
||||
"""
|
||||
endpoints = resolve_endpoints()
|
||||
path = config_path or resolve_config_path()
|
||||
target_host = host or resolve_active_host()
|
||||
|
||||
device = request_device_code(endpoints, source=source)
|
||||
if display:
|
||||
display(device)
|
||||
if open_url:
|
||||
open_url(device.verification_uri_complete)
|
||||
|
||||
grant = poll_for_token(endpoints, device, on_poll=on_poll, sleep=sleep)
|
||||
cred = oauth.install_grant(
|
||||
path,
|
||||
target_host,
|
||||
grant,
|
||||
client_id=endpoints.client_id,
|
||||
token_endpoint=endpoints.token_url,
|
||||
apply_config=apply_config,
|
||||
)
|
||||
from plugins.memory.honcho.client import reset_honcho_client
|
||||
|
||||
reset_honcho_client()
|
||||
logger.info("Honcho OAuth device grant installed for host %s", target_host)
|
||||
return cred
|
||||
|
||||
|
||||
# — Background launcher + status, for the desktop "Connect" button —
|
||||
# The flow blocks on a browser round-trip, so the web_server endpoint kicks it
|
||||
# off in a thread and the UI polls status rather than holding the request open.
|
||||
|
|
|
|||
|
|
@ -409,6 +409,9 @@ class TestSetupWizardDeploymentShape:
|
|||
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
||||
monkeypatch.setattr(honcho_cli, "_ensure_sdk_installed", lambda: True)
|
||||
monkeypatch.setattr(honcho_cli, "_write_config", lambda *a, **k: None)
|
||||
# No network probe / environment sniffing in tests.
|
||||
monkeypatch.setattr(honcho_cli, "_device_login_available", lambda: False)
|
||||
monkeypatch.setattr(honcho_cli, "_headless", lambda: (False, True))
|
||||
# Gate detection is mocked so tests control whether the tree runs.
|
||||
# None → undetectable; list (possibly empty) → connected platforms.
|
||||
gw = None if gateway_platforms is None else list(gateway_platforms)
|
||||
|
|
@ -791,3 +794,121 @@ class TestMigratePinKey:
|
|||
block = {"pinUserPeer": True}
|
||||
assert honcho_cli._migrate_pin_key(block) is False
|
||||
assert block == {"pinUserPeer": True}
|
||||
|
||||
|
||||
class TestCmdSetupDeviceFlow:
|
||||
"""The cloud auth-method menu's device-code branch (RFC 8628)."""
|
||||
|
||||
def _run_setup(self, monkeypatch, tmp_path, *, answers, device_available=True,
|
||||
headless=(False, True), device_result=None, device_error=None):
|
||||
"""Run cmd_setup with the device flow stubbed; returns (cfg, calls, prompts)."""
|
||||
import plugins.memory.honcho.cli as honcho_cli
|
||||
import plugins.memory.honcho.oauth_flow as oauth_flow
|
||||
from plugins.memory.honcho.oauth import OAuthCredential
|
||||
|
||||
cfg_path = tmp_path / "config.json"
|
||||
cfg_path.write_text("{}")
|
||||
cfg = {"apiKey": "***"}
|
||||
|
||||
monkeypatch.setattr(honcho_cli, "_read_config", lambda: cfg)
|
||||
monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
|
||||
monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
|
||||
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
||||
monkeypatch.setattr(honcho_cli, "_ensure_sdk_installed", lambda: True)
|
||||
monkeypatch.setattr(honcho_cli, "_write_config", lambda *a, **k: None)
|
||||
monkeypatch.setattr(honcho_cli, "_gateway_platforms", lambda: [])
|
||||
monkeypatch.setattr(honcho_cli, "_device_login_available", lambda: device_available)
|
||||
monkeypatch.setattr(honcho_cli, "_headless", lambda: headless)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config", lambda: {"memory": {}}, raising=False,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.save_config", lambda c: None, raising=False,
|
||||
)
|
||||
|
||||
class _FakeClientCfg:
|
||||
def resolve_session_name(self):
|
||||
return "hermes-test"
|
||||
workspace_id = "hermes"
|
||||
peer_name = "eri"
|
||||
ai_peer = "hermetika"
|
||||
observation_mode = "directional"
|
||||
write_frequency = "async"
|
||||
recall_mode = "hybrid"
|
||||
session_strategy = "per-session"
|
||||
|
||||
monkeypatch.setattr(
|
||||
"plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
|
||||
lambda host=None: _FakeClientCfg(),
|
||||
)
|
||||
monkeypatch.setattr("plugins.memory.honcho.client.reset_honcho_client", lambda: None)
|
||||
monkeypatch.setattr("plugins.memory.honcho.client.get_honcho_client", lambda hcfg: object())
|
||||
|
||||
calls: list[dict] = []
|
||||
cred = OAuthCredential(
|
||||
access_token="hch-at-x", refresh_token="hch-rt-x", expires_at=9_999_999_999,
|
||||
client_id="hermes-agent", token_endpoint="http://x/oauth/token",
|
||||
consent_peer_name="lyra",
|
||||
)
|
||||
|
||||
def fake_device_flow(**kwargs):
|
||||
calls.append(kwargs)
|
||||
if device_error is not None:
|
||||
raise device_error
|
||||
return device_result or cred
|
||||
|
||||
monkeypatch.setattr(oauth_flow, "authorize_via_device_code", fake_device_flow)
|
||||
|
||||
prompts: list[tuple[str, str | None]] = []
|
||||
answer_iter = iter(answers)
|
||||
def _scripted_prompt(label, default=None, secret=False):
|
||||
prompts.append((label, default))
|
||||
try:
|
||||
# Mirror the real _prompt: blank input falls back to the default.
|
||||
return next(answer_iter) or (default or "")
|
||||
except StopIteration:
|
||||
return default if default is not None else ""
|
||||
monkeypatch.setattr(honcho_cli, "_prompt", _scripted_prompt)
|
||||
|
||||
honcho_cli.cmd_setup(SimpleNamespace())
|
||||
return cfg, calls, prompts
|
||||
|
||||
def test_device_choice_runs_flow_and_stores_grant(self, monkeypatch, tmp_path):
|
||||
cfg, calls, _ = self._run_setup(monkeypatch, tmp_path, answers=["cloud", "device"])
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["apply_config"] is False
|
||||
assert calls[0]["source"] == "hermes-cli"
|
||||
host = cfg["hosts"]["hermes"]
|
||||
assert host["apiKey"] == "hch-at-x"
|
||||
assert host["oauth"]["refreshToken"] == "hch-rt-x"
|
||||
assert host["peerName"] == "lyra"
|
||||
|
||||
def test_headless_defaults_to_device(self, monkeypatch, tmp_path):
|
||||
# Blank answer takes the prompt default, which flips to device on a
|
||||
# remote/no-browser environment.
|
||||
cfg, calls, prompts = self._run_setup(
|
||||
monkeypatch, tmp_path, answers=["cloud", ""], headless=(True, False),
|
||||
)
|
||||
method_prompts = [p for p in prompts if "apikey" in p[0]]
|
||||
assert method_prompts[0][1] == "device"
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["open_url"] is None # never auto-open a browser headless
|
||||
assert cfg["hosts"]["hermes"]["apiKey"] == "hch-at-x"
|
||||
|
||||
def test_denied_device_flow_aborts_without_grant(self, monkeypatch, tmp_path):
|
||||
from plugins.memory.honcho.oauth_flow import AccessDenied
|
||||
|
||||
cfg, calls, _ = self._run_setup(
|
||||
monkeypatch, tmp_path, answers=["cloud", "device"],
|
||||
device_error=AccessDenied("access_denied", "user denied"),
|
||||
)
|
||||
assert len(calls) == 1
|
||||
assert "apiKey" not in cfg.get("hosts", {}).get("hermes", {})
|
||||
|
||||
def test_device_option_hidden_when_not_advertised(self, monkeypatch, tmp_path):
|
||||
_, calls, prompts = self._run_setup(
|
||||
monkeypatch, tmp_path, answers=["cloud", "device"], device_available=False,
|
||||
)
|
||||
method_prompts = [p for p in prompts if "OAuth or API key" in p[0]]
|
||||
assert method_prompts and method_prompts[0][1] == "oauth"
|
||||
assert calls == [] # 'device' answer falls through to the api-key path
|
||||
|
|
|
|||
|
|
@ -24,9 +24,34 @@ class _FakeAS(BaseHTTPRequestHandler):
|
|||
|
||||
# Rotation counter shared across requests so refresh returns a new token.
|
||||
issued = {"n": 0}
|
||||
# Scripted outcomes for device-grant token polls; "ok" mints, anything else
|
||||
# is returned as a 400 OAuth error code.
|
||||
device_responses: list[str] = []
|
||||
# Last form posted to /oauth/device_authorization, for assertions.
|
||||
last_device_form: dict = {}
|
||||
# Whether AS metadata advertises the device grant.
|
||||
advertise_device = True
|
||||
|
||||
def _send_json(self, status: int, body: dict) -> None:
|
||||
payload = json.dumps(body).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
def do_GET(self): # noqa: N802
|
||||
parsed = urlparse(self.path)
|
||||
if parsed.path == "/.well-known/oauth-authorization-server":
|
||||
if not _FakeAS.advertise_device:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
self._send_json(200, {
|
||||
"grant_types_supported": [
|
||||
"authorization_code", "refresh_token", oauth_flow.DEVICE_GRANT_TYPE,
|
||||
],
|
||||
})
|
||||
return
|
||||
if parsed.path != "/authorize":
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
|
|
@ -50,13 +75,41 @@ class _FakeAS(BaseHTTPRequestHandler):
|
|||
|
||||
def do_POST(self): # noqa: N802
|
||||
parsed = urlparse(self.path)
|
||||
length = int(self.headers.get("Content-Length", 0))
|
||||
form = parse_qs(self.rfile.read(length).decode())
|
||||
if parsed.path == "/oauth/device_authorization":
|
||||
_FakeAS.last_device_form = {k: v[0] for k, v in form.items()}
|
||||
base = f"http://{self.server.server_address[0]}:{self.server.server_address[1]}"
|
||||
self._send_json(200, {
|
||||
"device_code": "dev-code-1",
|
||||
"user_code": "ABCD-EFGH",
|
||||
"verification_uri": f"{base}/device",
|
||||
"verification_uri_complete": f"{base}/device?user_code=ABCD-EFGH",
|
||||
"expires_in": 600,
|
||||
"interval": 0,
|
||||
})
|
||||
return
|
||||
if parsed.path != "/oauth/token":
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
length = int(self.headers.get("Content-Length", 0))
|
||||
form = parse_qs(self.rfile.read(length).decode())
|
||||
grant_type = form["grant_type"][0]
|
||||
if grant_type == oauth_flow.DEVICE_GRANT_TYPE:
|
||||
outcome = _FakeAS.device_responses.pop(0) if _FakeAS.device_responses else "ok"
|
||||
if outcome != "ok":
|
||||
self._send_json(400, {"error": outcome, "error_description": f"scripted {outcome}"})
|
||||
return
|
||||
self.issued["n"] += 1
|
||||
n = self.issued["n"]
|
||||
self._send_json(200, {
|
||||
"access_token": f"hch-at-{n}",
|
||||
"refresh_token": f"hch-rt-{n}",
|
||||
"token_type": "Bearer",
|
||||
"expires_in": 3600,
|
||||
"scope": "write",
|
||||
"config": {"peerName": "lyra"},
|
||||
})
|
||||
return
|
||||
self.issued["n"] += 1
|
||||
n = self.issued["n"]
|
||||
body = {
|
||||
|
|
@ -85,6 +138,9 @@ class _FakeAS(BaseHTTPRequestHandler):
|
|||
@pytest.fixture
|
||||
def fake_as(monkeypatch):
|
||||
_FakeAS.issued["n"] = 0
|
||||
_FakeAS.device_responses = []
|
||||
_FakeAS.last_device_form = {}
|
||||
_FakeAS.advertise_device = True
|
||||
server = HTTPServer(("127.0.0.1", 0), _FakeAS)
|
||||
port = server.server_address[1]
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
|
|
@ -236,6 +292,195 @@ def test_cli_flow_stores_tokens_without_applying_config(tmp_path, fake_as):
|
|||
assert cred.consent_peer_name == "lyra"
|
||||
|
||||
|
||||
# ── Device authorization grant (RFC 8628): headless / remote-VM path ──
|
||||
|
||||
|
||||
class _FakeClock:
|
||||
"""Injectable sleep + monotonic pair so poll loops run instantly."""
|
||||
|
||||
def __init__(self):
|
||||
self.t = 0.0
|
||||
self.sleeps: list[float] = []
|
||||
|
||||
def sleep(self, seconds: float) -> None:
|
||||
self.sleeps.append(seconds)
|
||||
self.t += seconds
|
||||
|
||||
def monotonic(self) -> float:
|
||||
return self.t
|
||||
|
||||
|
||||
def test_device_endpoint_derived_from_token_url(monkeypatch):
|
||||
monkeypatch.delenv("HONCHO_OAUTH_DEVICE_AUTH_URL", raising=False)
|
||||
monkeypatch.delenv("HONCHO_OAUTH_TOKEN_URL", raising=False)
|
||||
cloud = oauth_flow.resolve_endpoints(environment="production", base_url="https://api.honcho.dev")
|
||||
assert cloud.device_authorization_url == "https://api.honcho.dev/oauth/device_authorization"
|
||||
local = oauth_flow.resolve_endpoints(environment="local", base_url=None)
|
||||
assert local.device_authorization_url == "http://localhost:8000/oauth/device_authorization"
|
||||
|
||||
|
||||
def test_device_endpoint_env_override(monkeypatch):
|
||||
monkeypatch.setenv("HONCHO_OAUTH_DEVICE_AUTH_URL", "https://alt.example/oauth/device_authorization")
|
||||
endpoints = oauth_flow.resolve_endpoints(environment="production", base_url=None)
|
||||
assert endpoints.device_authorization_url == "https://alt.example/oauth/device_authorization"
|
||||
|
||||
|
||||
def test_supports_device_login_from_metadata(fake_as):
|
||||
endpoints = oauth_flow.resolve_endpoints()
|
||||
assert oauth_flow.supports_device_login(endpoints) is True
|
||||
_FakeAS.advertise_device = False
|
||||
assert oauth_flow.supports_device_login(endpoints) is False
|
||||
# Fail closed on an unreachable host.
|
||||
dead = oauth_flow.OAuthEndpoints(
|
||||
authorize_url="http://127.0.0.1:1/authorize",
|
||||
token_url="http://127.0.0.1:1/oauth/token",
|
||||
client_id="hermes-agent",
|
||||
scope="write",
|
||||
)
|
||||
assert oauth_flow.supports_device_login(dead, timeout=0.2) is False
|
||||
|
||||
|
||||
def test_request_device_code_parses_response_and_sends_identity(fake_as):
|
||||
endpoints = oauth_flow.resolve_endpoints()
|
||||
device = oauth_flow.request_device_code(endpoints, source="hermes-cli")
|
||||
assert device.device_code == "dev-code-1"
|
||||
assert device.user_code == "ABCD-EFGH"
|
||||
assert device.verification_uri.endswith("/device")
|
||||
assert device.verification_uri_complete.endswith("?user_code=ABCD-EFGH")
|
||||
assert (device.expires_in, device.interval) == (600, 0)
|
||||
assert _FakeAS.last_device_form["client_id"] == "hermes-desktop"
|
||||
assert _FakeAS.last_device_form["scope"] == "write"
|
||||
assert _FakeAS.last_device_form["source"] == "hermes-cli"
|
||||
|
||||
|
||||
def test_full_device_flow_pending_then_approved(tmp_path, fake_as):
|
||||
_FakeAS.device_responses = ["authorization_pending", "authorization_pending", "ok"]
|
||||
config_path = tmp_path / "honcho.json"
|
||||
config_path.write_text(json.dumps({"hosts": {"hermes": {"saveMessages": False}}}))
|
||||
|
||||
clock = _FakeClock()
|
||||
cred = oauth_flow.authorize_via_device_code(
|
||||
config_path=config_path,
|
||||
host="hermes",
|
||||
source="hermes-cli",
|
||||
apply_config=False,
|
||||
sleep=clock.sleep,
|
||||
)
|
||||
|
||||
saved = json.loads(config_path.read_text())
|
||||
host = saved["hosts"]["hermes"]
|
||||
assert host["apiKey"] == cred.access_token == "hch-at-1"
|
||||
assert host["oauth"]["refreshToken"] == "hch-rt-1"
|
||||
assert host["oauth"]["clientId"] == "hermes-desktop"
|
||||
assert host["oauth"]["tokenEndpoint"] == oauth_flow.resolve_endpoints().token_url
|
||||
# Wizard-owned settings untouched; consent peer name still surfaced.
|
||||
assert host["saveMessages"] is False
|
||||
assert cred.consent_peer_name == "lyra"
|
||||
assert len(clock.sleeps) == 3 # one wait per poll
|
||||
|
||||
|
||||
def test_poll_backs_off_on_slow_down(fake_as):
|
||||
_FakeAS.device_responses = ["slow_down", "slow_down", "ok"]
|
||||
endpoints = oauth_flow.resolve_endpoints()
|
||||
device = oauth_flow.DeviceCode(
|
||||
device_code="dev-code-1", user_code="X", verification_uri="u",
|
||||
verification_uri_complete="u?c", expires_in=600, interval=5,
|
||||
)
|
||||
clock = _FakeClock()
|
||||
grant = oauth_flow.poll_for_token(endpoints, device, sleep=clock.sleep, monotonic=clock.monotonic)
|
||||
assert grant["access_token"] == "hch-at-1"
|
||||
assert clock.sleeps == [5, 10, 15]
|
||||
|
||||
|
||||
def test_slow_down_interval_caps_at_60(fake_as):
|
||||
_FakeAS.device_responses = ["slow_down", "ok"]
|
||||
endpoints = oauth_flow.resolve_endpoints()
|
||||
device = oauth_flow.DeviceCode(
|
||||
device_code="dev-code-1", user_code="X", verification_uri="u",
|
||||
verification_uri_complete="u?c", expires_in=600, interval=58,
|
||||
)
|
||||
clock = _FakeClock()
|
||||
oauth_flow.poll_for_token(endpoints, device, sleep=clock.sleep, monotonic=clock.monotonic)
|
||||
assert clock.sleeps == [58, 60] # 58 + 5 clamps to the 60s cap
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("error", "exc"),
|
||||
[("access_denied", oauth_flow.AccessDenied), ("expired_token", oauth_flow.DeviceCodeExpired)],
|
||||
)
|
||||
def test_poll_raises_typed_errors(fake_as, error, exc):
|
||||
_FakeAS.device_responses = [error]
|
||||
endpoints = oauth_flow.resolve_endpoints()
|
||||
device = oauth_flow.DeviceCode(
|
||||
device_code="dev-code-1", user_code="X", verification_uri="u",
|
||||
verification_uri_complete="u?c", expires_in=600, interval=0,
|
||||
)
|
||||
clock = _FakeClock()
|
||||
with pytest.raises(exc) as e:
|
||||
oauth_flow.poll_for_token(endpoints, device, sleep=clock.sleep, monotonic=clock.monotonic)
|
||||
assert e.value.error == error
|
||||
|
||||
|
||||
def test_poll_times_out_at_deadline(fake_as):
|
||||
_FakeAS.device_responses = ["authorization_pending"] * 10
|
||||
endpoints = oauth_flow.resolve_endpoints()
|
||||
device = oauth_flow.DeviceCode(
|
||||
device_code="dev-code-1", user_code="X", verification_uri="u",
|
||||
verification_uri_complete="u?c", expires_in=3, interval=1,
|
||||
)
|
||||
clock = _FakeClock()
|
||||
with pytest.raises(oauth_flow.AuthorizationTimeout):
|
||||
oauth_flow.poll_for_token(endpoints, device, sleep=clock.sleep, monotonic=clock.monotonic)
|
||||
assert len(clock.sleeps) <= 3 # bounded by the deadline, not the script
|
||||
|
||||
|
||||
def test_device_flow_browser_open_is_caller_opt_in(tmp_path, fake_as):
|
||||
config_path = tmp_path / "honcho.json"
|
||||
config_path.write_text(json.dumps({"hosts": {}}))
|
||||
opened: list[str] = []
|
||||
shown: list[oauth_flow.DeviceCode] = []
|
||||
|
||||
oauth_flow.authorize_via_device_code(
|
||||
config_path=config_path, host="hermes",
|
||||
display=shown.append, open_url=opened.append, sleep=lambda s: None,
|
||||
)
|
||||
assert opened == [shown[0].verification_uri_complete]
|
||||
|
||||
# No open_url → nothing opened; the flow still completes.
|
||||
config_path.write_text(json.dumps({"hosts": {}}))
|
||||
cred = oauth_flow.authorize_via_device_code(
|
||||
config_path=config_path, host="hermes", sleep=lambda s: None,
|
||||
)
|
||||
assert cred.access_token
|
||||
|
||||
|
||||
def test_callback_page_shows_error_on_denied_consent():
|
||||
server, captured = oauth_flow._bind_loopback_server()
|
||||
port = server.server_address[1]
|
||||
result: dict = {}
|
||||
|
||||
def _deny():
|
||||
for _ in range(50):
|
||||
try:
|
||||
result["resp"] = httpx.get(
|
||||
f"http://127.0.0.1:{port}/callback"
|
||||
"?error=access_denied&error_description=user+denied&state=x",
|
||||
timeout=2,
|
||||
)
|
||||
return
|
||||
except httpx.ConnectError:
|
||||
time.sleep(0.05)
|
||||
|
||||
thread = threading.Thread(target=_deny, daemon=True)
|
||||
thread.start()
|
||||
with pytest.raises(ValueError, match="access_denied.*user denied"):
|
||||
oauth_flow.capture_loopback_code(server, captured, timeout=5)
|
||||
thread.join(timeout=5)
|
||||
page = result["resp"].text
|
||||
assert "Connected" not in page
|
||||
assert "not completed" in page and "access_denied" in page
|
||||
|
||||
|
||||
# ── Desktop "Connect" button path: background launcher, status, dispatch ──
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -70,6 +70,8 @@ hermes memory setup # select "honcho" — runs the Honcho-specific post-s
|
|||
|
||||
The legacy `hermes honcho setup` command still works (it now redirects to `hermes memory setup`), but is only registered after Honcho is selected as the active memory provider.
|
||||
|
||||
**Headless / remote machines:** for cloud auth on a box without a browser (SSH, remote VM), pick **device** at the wizard's auth-method prompt. The CLI prints a short code and a verification link; open the link in a browser on any other machine, approve, and setup completes — no API key copy-paste. The wizard defaults to this option automatically when it detects no usable local browser.
|
||||
|
||||
**Config:** `$HERMES_HOME/honcho.json` (profile-local) or `~/.honcho/config.json` (global). Resolution order: `$HERMES_HOME/honcho.json` > `~/.hermes/honcho.json` > `~/.honcho/config.json`. See the [config reference](https://github.com/NousResearch/hermes-agent/blob/main/plugins/memory/honcho/README.md) and the [Honcho integration guide](https://docs.honcho.dev/v3/guides/integrations/hermes).
|
||||
|
||||
<details>
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue