diff --git a/honcho-cli/src/honcho_cli/commands/setup.py b/honcho-cli/src/honcho_cli/commands/setup.py index 6073b3e4..85dc4511 100644 --- a/honcho-cli/src/honcho_cli/commands/setup.py +++ b/honcho-cli/src/honcho_cli/commands/setup.py @@ -7,6 +7,8 @@ from __future__ import annotations import json +import time +import webbrowser import typer from honcho import ( @@ -19,13 +21,14 @@ from honcho import ( from rich.console import Console from rich.panel import Panel -from honcho_cli import __version__ +from honcho_cli import __version__, oauth from honcho_cli.branding import BANNER, BRAND, ICON_FAIL, ICON_OK, ICON_RUN -from honcho_cli.common import get_resolved_config +from honcho_cli.common import get_resolved_config, maybe_refresh_token from honcho_cli.config import ( CONFIG_FILE, DEFAULT_BASE_URL, CLIConfig, + OAuthTokens, ) from honcho_cli.output import print_error, print_result, set_json_mode, use_json @@ -117,21 +120,142 @@ def init( _console.print() _console.print() + # Non-interactive (JSON/piped) or an explicit --api-key: manual-key path. + # Device login needs a human at a browser, so it's TTY-only. + if use_json() or api_key: + _init_manual_key(key_val, url_val, file_key, file_url) + else: + _init_interactive(key_val, url_val, file_url) + + +def _init_manual_key(key_val: str, url_val: str, file_key: str, file_url: str) -> None: + """Non-interactive path: confirm/save apiKey + URL, no device login.""" final_key = _prompt_api_key(key_val) final_url = _prompt_url(url_val) - - # Persist if anything changed or if the value came from env/flag. if final_key != file_key or final_url != file_url: CLIConfig(base_url=final_url, api_key=final_key).save() if not use_json(): _console.print(f" {ICON_OK} [dim]Saved to {CONFIG_FILE}[/dim]") - _check_connection(final_url, final_key) - if use_json(): print_result({"apiKey": _redact(final_key), "baseUrl": final_url}) +def _init_interactive(key_val: str, url_val: str, file_url: str) -> None: + """Interactive path: URL first (device flow needs the host), then auth method.""" + final_url = _prompt_url(url_val) + existing = CLIConfig.load() + has_creds = bool(key_val) or bool(existing.oauth and existing.oauth.access_token) + # only offer browser login if the host advertises the device grant (managed) + device_available = oauth.supports_device_login(final_url) + method = _prompt_auth_method(has_creds, device_available) + + if method == "keep": + if final_url != file_url: + existing.base_url = final_url + existing.save() + _console.print(f" {ICON_OK} [dim]Saved to {CONFIG_FILE}[/dim]") + # refresh an expired token so "keep" behaves like every live command; + # a failed refresh surfaces as the connectivity check below, not an abort + try: + maybe_refresh_token(existing) + except typer.Exit: + pass + _check_connection(final_url, existing.resolved_api_key()) + return + + if method == "device": + tokens = _device_login(final_url) + CLIConfig(base_url=final_url, oauth=tokens).save() + _console.print(f" {ICON_OK} [dim]Saved to {CONFIG_FILE}[/dim]") + _check_connection(final_url, tokens.access_token) + return + + # paste a key + final_key = _prompt_api_key("") + CLIConfig(base_url=final_url, api_key=final_key).save() + _console.print(f" {ICON_OK} [dim]Saved to {CONFIG_FILE}[/dim]") + _check_connection(final_url, final_key) + + +def _prompt_auth_method(has_creds: bool, device_available: bool) -> str: + """Ask how to authenticate. Returns ``device`` / ``key`` / ``keep``. + + ``device`` is only offered when the host advertises the device grant; when + it doesn't, pasting a key is the only login path. + """ + _console.print(" [dim]How do you want to authenticate?[/dim]") + options: list[str] = [] + if device_available: + options.append("device") + _console.print(f" [dim]({len(options)})[/dim] Log in with your browser (device code)") + options.append("key") + _console.print(f" [dim]({len(options)})[/dim] Paste an API key") + if has_creds: + options.append("keep") + _console.print(f" [dim]({len(options)})[/dim] Keep current credentials") + # default to keeping existing creds so a returning user pressing Enter doesn't + # get dropped into an unwanted browser login that overwrites them + default = str(options.index("keep") + 1) if "keep" in options else "1" + choice = typer.prompt(" Choice", default=default, show_default=True, prompt_suffix=": ").strip() + try: + idx = int(choice) + except ValueError: + return options[0] + # explicit 1..len bounds — bare `options[idx - 1]` would let "0"/negatives + # wrap to the tail of the list via Python's negative indexing + if 1 <= idx <= len(options): + return options[idx - 1] + return options[0] + + +def _device_login(base_url: str) -> OAuthTokens: + """Run the device-authorization flow and return the minted tokens. + + Prints the user code + verification URL, opens the browser best-effort, and + blocks on the poll loop until the user approves. Exits non-zero on denial, + expiry, or interrupt. + """ + endpoints = oauth.resolve_endpoints(base_url) + try: + device = oauth.request_device_code(endpoints) + except oauth.OAuthFlowError as e: + _console.print(f" {ICON_FAIL} [red]Could not start device login[/red]: {e}") + raise typer.Exit(1) + + _console.print() + _console.print(f" Enter this code to authorize: [bold {BRAND}]{device.user_code}[/bold {BRAND}]") + _console.print(f" [dim]at[/dim] {device.verification_uri}") + _console.print() + try: + webbrowser.open(device.verification_uri_complete) + except Exception: + pass # headless is expected — the URL is printed above + + try: + with _console.status("Waiting for approval…", spinner="dots"): + tokens = oauth.poll_for_token(endpoints, device) + except oauth.AccessDenied: + _console.print(f" {ICON_FAIL} [red]Authorization denied[/red]") + raise typer.Exit(1) + except (oauth.DeviceCodeExpired, oauth.AuthorizationTimeout): + _console.print(f" {ICON_FAIL} [red]Code expired[/red] — run `honcho init` to try again") + raise typer.Exit(1) + except oauth.OAuthFlowError as e: + _console.print(f" {ICON_FAIL} [red]Login failed[/red]: {e}") + raise typer.Exit(1) + except KeyboardInterrupt: + _console.print(f" {ICON_FAIL} [red]Cancelled[/red]") + raise typer.Exit(1) + + return OAuthTokens.from_response( + tokens, + client_id=endpoints.client_id, + scope_fallback=endpoints.scope, + host=base_url, + ) + + def _prompt_api_key(value: str) -> str: """Prompt for API key. @@ -206,6 +330,21 @@ def _check_connection(base_url: str, api_key: str) -> None: # --------------------------------------------------------------------------- # # honcho doctor +def _auth_mode_detail(config: CLIConfig) -> str: + """Human summary of which credential the CLI will use.""" + tokens = config.usable_oauth() + if tokens is not None: + if tokens.access_valid(): + secs = max(int(tokens.access_expires_at - time.time()), 0) + return f"OAuth device token (expires in {secs // 60}m)" + if config.api_key: + return "API key (OAuth token expired)" + return "OAuth device token (expired — will refresh)" + if config.api_key: + return "API key" + return "missing — run `honcho init`" + + def doctor( json_output: bool = typer.Option(False, "--json", help="Force JSON output"), ) -> None: @@ -230,23 +369,30 @@ def doctor( _console.print(f"\n[bold {BRAND}]Honcho Doctor[/bold {BRAND}]\n") config = get_resolved_config() + # Refresh an expired OAuth token if we can; a failure surfaces as a failed + # connectivity check below rather than aborting the diagnostic. + try: + maybe_refresh_token(config) + except typer.Exit: + pass + key = config.resolved_api_key() + _add("Config file", CONFIG_FILE.exists(), str(CONFIG_FILE) if CONFIG_FILE.exists() else f"{CONFIG_FILE} not found") - _add("API key configured", bool(config.api_key), - "set" if config.api_key else "missing — run `honcho init`") + _add("Credentials configured", bool(key), _auth_mode_detail(config)) - if config.base_url and config.api_key: - _add("API connectivity", *_test_connection(config.base_url, config.api_key)) + if config.base_url and key: + _add("API connectivity", *_test_connection(config.base_url, key)) else: - _add("API connectivity", False, "skipped — no base_url or api_key") + _add("API connectivity", False, "skipped — no base_url or credentials") # Workspace / peer / queue run only when scoped via -w / -p. ws_ok, client = False, None - if config.workspace_id and config.api_key: + if config.workspace_id and key: try: - client = Honcho(base_url=config.base_url, api_key=config.api_key, workspace_id=config.workspace_id) + client = Honcho(base_url=config.base_url, api_key=key, workspace_id=config.workspace_id) client.get_configuration() ws_ok = True _add("Workspace reachable", True, config.workspace_id) @@ -280,7 +426,7 @@ def doctor( _console.print(f"\n [{color}]{passed}/{total}[/{color}] checks passed{hint}\n") # Config file + API connectivity are hard requirements. - critical = {"Config file", "API key configured", "API connectivity"} + critical = {"Config file", "Credentials configured", "API connectivity"} if config.workspace_id: critical.add("Workspace reachable") if any(not c["ok"] for c in checks if c["check"] in critical): diff --git a/honcho-cli/src/honcho_cli/common.py b/honcho-cli/src/honcho_cli/common.py index 2680a869..d87a4be7 100644 --- a/honcho-cli/src/honcho_cli/common.py +++ b/honcho-cli/src/honcho_cli/common.py @@ -12,13 +12,15 @@ no-op if the same flag was already set at an outer level. from __future__ import annotations +from dataclasses import replace from typing import Optional import typer from honcho import Honcho -from honcho_cli.config import CLIConfig, get_client_kwargs +from honcho_cli import oauth +from honcho_cli.config import CLIConfig, OAuthTokens, get_client_kwargs from honcho_cli.output import print_error, set_json_mode from honcho_cli.validation import validate_resource_id @@ -50,6 +52,50 @@ def get_resolved_config(): return config +def maybe_refresh_token(config: CLIConfig) -> None: + """Refresh an expired OAuth access token in place and persist it. + + No-op when there is no grant for the current host or the token is still + valid. A dead grant degrades to the saved apiKey with a warning; exits + only when nothing is left to authenticate with. + """ + tokens = config.usable_oauth() + if tokens is None or tokens.access_valid(): + return + + if tokens.refresh_token: + endpoints = oauth.resolve_endpoints(config.base_url) + if tokens.client_id: + endpoints = replace(endpoints, client_id=tokens.client_id) + try: + refreshed = oauth.refresh_access_token(endpoints, tokens.refresh_token) + except oauth.OAuthFlowError: + refreshed = None + if refreshed is not None: + # rotation-safe: persist the (possibly new) refresh token before + # it's reused; keep the old one if the server didn't rotate + # (refresh_token is optional) + config.oauth = OAuthTokens.from_response( + refreshed, + client_id=tokens.client_id, + scope_fallback=tokens.scope, + refresh_fallback=tokens.refresh_token, + host=tokens.host, + ) + config.save() + return + + if config.api_key: + typer.echo( + "OAuth session expired; using the saved API key. " + "Run `honcho init` to log in again.", + err=True, + ) + return + print_error("SESSION_EXPIRED", "OAuth session expired. Run `honcho init` to log in again.") + raise typer.Exit(1) + + def get_client(*, require_workspace: bool = True): """Create a Honcho client from resolved config. @@ -65,6 +111,7 @@ def get_client(*, require_workspace: bool = True): "No workspace scoped. Pass --workspace/-w or set HONCHO_WORKSPACE_ID.", ) raise typer.Exit(1) + maybe_refresh_token(config) return Honcho(**get_client_kwargs(config)), config diff --git a/honcho-cli/src/honcho_cli/config.py b/honcho-cli/src/honcho_cli/config.py index a0c64864..7cd103bd 100644 --- a/honcho-cli/src/honcho_cli/config.py +++ b/honcho-cli/src/honcho_cli/config.py @@ -1,11 +1,18 @@ """Configuration management for Honcho CLI. -Config stored at ``~/.honcho/config.json`` with env var overrides. +Config stored at ``~/.honcho/config.json`` with env var overrides. The config +directory defaults to ``~/.honcho`` and can be relocated with `HONCHO_CONFIG_DIR` -The CLI owns exactly two top-level keys in that file: +The CLI owns these top-level keys in that file: - apiKey -- Honcho admin JWT environmentUrl -- Honcho API URL (full URL, e.g. https://api.honcho.dev) + oauth -- OAuth device-grant tokens (accessToken, refreshToken, + accessExpiresAt, clientId, scope, host), written by + device login + +``apiKey`` (manual admin JWT) is shared with sibling tools: the CLI writes it +on paste-key login and reads it as a fallback, but never deletes it. A live +OAuth token takes precedence over ``apiKey`` for the CLI's own calls. All other top-level keys (``hosts``, ``sessions``, ``saveMessages``, ``sessionStrategy``, …) are written by sibling Honcho tools and are @@ -20,14 +27,44 @@ from __future__ import annotations import json import os +import time from dataclasses import dataclass, fields from pathlib import Path +from typing import TYPE_CHECKING -CONFIG_DIR = Path.home() / ".honcho" +if TYPE_CHECKING: + from honcho_cli.oauth import TokenResponse + +def _config_dir() -> Path: + """Config directory: ``$HONCHO_CONFIG_DIR`` if set, else ``~/.honcho``.""" + override = os.environ.get("HONCHO_CONFIG_DIR") + return Path(override).expanduser() if override else Path.home() / ".honcho" + + +CONFIG_DIR = _config_dir() CONFIG_FILE = CONFIG_DIR / "config.json" DEFAULT_BASE_URL = "https://api.honcho.dev" + +def _redact_token(token: str) -> str: + """Show ``***`` — enough to compare tokens without leaking the body.""" + if not token: + return "" + return "***" + token[-4:] if len(token) > 4 else "***" + + +def _coerce_epoch(value: object) -> float: + """Parse a persisted epoch-seconds value, treating garbage as expired (0).""" + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str): + try: + return float(value) + except ValueError: + return 0.0 + return 0.0 + # Env var mapping for runtime overrides. # # Resolution order: flag > env var > config file > default. @@ -40,6 +77,59 @@ ENV_MAP: dict[str, str] = { } +@dataclass +class OAuthTokens: + """Device-grant tokens persisted under the config ``oauth`` key.""" + + access_token: str = "" + refresh_token: str = "" + access_expires_at: float = 0.0 # epoch seconds + client_id: str = "" + scope: str = "" + host: str = "" # base_url the grant was minted against + + def matches_host(self, base_url: str) -> bool: + """True when the grant belongs to ``base_url``. + + Tokens are host-scoped — a staging grant must not be sent to prod. + Legacy blocks with no recorded host are trusted. + """ + return not self.host or self.host.rstrip("/") == base_url.rstrip("/") + + def access_valid(self, skew: int = 60) -> bool: + """True while the access token is present and not within ``skew`` of expiry. + + Checks the expiry timestamp recorded at mint time, not the token + itself — the server is the real authority, so a wrong answer here + costs at most an extra refresh or a 401. + """ + return bool(self.access_token) and time.time() < self.access_expires_at - skew + + @classmethod + def from_response( + cls, + resp: TokenResponse, + *, + client_id: str, + scope_fallback: str = "", + refresh_fallback: str = "", + host: str = "", + ) -> OAuthTokens: + """Build persisted tokens from a token response. + + ``refresh_fallback`` keeps the prior refresh token when the server + doesn't rotate one (optional on the refresh grant, RFC 6749 §5.1). + """ + return cls( + access_token=resp.access_token, + refresh_token=resp.refresh_token or refresh_fallback, + access_expires_at=time.time() + resp.expires_in, + client_id=client_id, + scope=resp.scope or scope_fallback, + host=host, + ) + + @dataclass class CLIConfig: """CLI configuration with layered resolution: flag > env > file > default. @@ -54,6 +144,33 @@ class CLIConfig: workspace_id: str = "" peer_id: str = "" session_id: str = "" + oauth: OAuthTokens | None = None + + def usable_oauth(self) -> OAuthTokens | None: + """The OAuth grant, if present and bound to the current host.""" + if ( + self.oauth + and self.oauth.access_token + and self.oauth.matches_host(self.base_url) + ): + return self.oauth + return None + + def resolved_api_key(self) -> str: + """The key handed to the SDK: a live OAuth token wins, else apiKey. + + An expired grant loses to a saved apiKey (a dead grant degrades to the + shared key) but still wins over nothing, since the server is the final + judge. + """ + tokens = self.usable_oauth() + if tokens and tokens.access_valid(): + return tokens.access_token + if self.api_key: + return self.api_key + if tokens: + return tokens.access_token + return "" @classmethod def load(cls) -> CLIConfig: @@ -74,6 +191,16 @@ class CLIConfig: key = data.get("apiKey") if isinstance(key, str): config.api_key = key + oauth = data.get("oauth") + if isinstance(oauth, dict) and oauth.get("accessToken"): + config.oauth = OAuthTokens( + access_token=str(oauth.get("accessToken", "")), + refresh_token=str(oauth.get("refreshToken", "")), + access_expires_at=_coerce_epoch(oauth.get("accessExpiresAt")), + client_id=str(oauth.get("clientId", "")), + scope=str(oauth.get("scope", "")), + host=str(oauth.get("host", "")), + ) for fld_name, env_var in ENV_MAP.items(): val = os.environ.get(env_var) @@ -88,10 +215,12 @@ class CLIConfig: return config def save(self) -> None: - """Write ``apiKey`` + ``environmentUrl`` to config.json. + """Write ``environmentUrl`` + credentials to config.json. Preserves unrelated top-level keys (``hosts``, ``sessions``, ``saveMessages``, ``sessionStrategy``, …) that other tools write. + ``apiKey`` is written when set but never removed — sibling tools read + it. The ``oauth`` block is CLI-owned and dropped when empty. """ CONFIG_DIR.mkdir(parents=True, exist_ok=True) @@ -108,8 +237,18 @@ class CLIConfig: data["environmentUrl"] = self.base_url if self.api_key: data["apiKey"] = self.api_key + + if self.oauth and self.oauth.access_token: + data["oauth"] = { + "accessToken": self.oauth.access_token, + "refreshToken": self.oauth.refresh_token, + "accessExpiresAt": self.oauth.access_expires_at, + "clientId": self.oauth.client_id, + "scope": self.oauth.scope, + "host": self.oauth.host, + } else: - data.pop("apiKey", None) + data.pop("oauth", None) CONFIG_FILE.write_text(json.dumps(data, indent=2) + "\n") # API key in plaintext — restrict to the owner on multi-user hosts. @@ -124,7 +263,7 @@ class CLIConfig: Only includes fields that have a value set — per-command fields (workspace_id, peer_id, session_id) are omitted when empty. """ - d: dict[str, str] = {} + result: dict[str, str] = {} for fld in fields(self): val = getattr(self, fld.name) if not val: @@ -132,10 +271,12 @@ class CLIConfig: if fld.name == "api_key": # Show ``***`` only — enough to compare keys without # leaking the header or body of the JWT. - d[fld.name] = "***" + val[-4:] if len(val) > 4 else "***" + result[fld.name] = _redact_token(val) + elif fld.name == "oauth": + result[fld.name] = _redact_token(val.access_token) else: - d[fld.name] = val - return d + result[fld.name] = val + return result def get_client_kwargs(config: CLIConfig) -> dict: @@ -143,8 +284,9 @@ def get_client_kwargs(config: CLIConfig) -> dict: kwargs: dict = {} if config.base_url: kwargs["base_url"] = config.base_url - if config.api_key: - kwargs["api_key"] = config.api_key + api_key = config.resolved_api_key() + if api_key: + kwargs["api_key"] = api_key if config.workspace_id: kwargs["workspace_id"] = config.workspace_id return kwargs diff --git a/honcho-cli/src/honcho_cli/oauth.py b/honcho-cli/src/honcho_cli/oauth.py new file mode 100644 index 00000000..37e50d36 --- /dev/null +++ b/honcho-cli/src/honcho_cli/oauth.py @@ -0,0 +1,260 @@ +"""OAuth 2.0 Device Authorization Grant (RFC 8628) client for the CLI. + +Transport-only: HTTP calls plus the poll loop, no Typer or config writes, so it +can be unit-tested by mocking httpx. +""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any + +import httpx + +DEVICE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:device_code" + +DEFAULT_CLIENT_ID = "honcho-cli" +DEFAULT_SCOPE = "write" + +# self-declared requesting surface; tells the consent screen not to offer config +# delivery (a CLI has nowhere to write it) +DEVICE_SOURCE = "honcho-cli" + +# extra seconds added to the poll interval on a slow_down response (RFC 8628 §3.5) +SLOW_DOWN_STEP = 5 + + +class OAuthFlowError(Exception): + """A device-flow request failed. ``error`` is the RFC error code when known.""" + + def __init__(self, error: str, description: str | None = None): + self.error: str = error + self.description: str | None = description + super().__init__(description or error) + + +class AccessDenied(OAuthFlowError): + """The user denied the authorization request.""" + + +class DeviceCodeExpired(OAuthFlowError): + """The device code expired before the user approved it.""" + + +class AuthorizationTimeout(OAuthFlowError): + """Polling ran past the device code's lifetime with no decision.""" + + +@dataclass(frozen=True) +class Endpoints: + """Resolved authorization-server URLs and client identity.""" + + device_auth_url: str + token_url: str + client_id: str + scope: str + + +@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 + + +@dataclass(frozen=True) +class TokenResponse: + """An access/refresh token pair minted for a grant.""" + + access_token: str + refresh_token: str + expires_in: int + scope: str + config: dict[str, Any] = field(default_factory=dict) + + +def resolve_endpoints(base_url: str) -> Endpoints: + """Derive OAuth endpoints and client identity from the API ``base_url``.""" + host = base_url.rstrip("/") + return Endpoints( + device_auth_url=f"{host}/oauth/device_authorization", + token_url=f"{host}/oauth/token", + client_id=DEFAULT_CLIENT_ID, + scope=DEFAULT_SCOPE, + ) + + +# RFC 8414 authorization-server metadata; presence of the device grant tells us +# whether this host can do browser login at all (managed only, not core) +AUTH_SERVER_METADATA_PATH = "/.well-known/oauth-authorization-server" + + +def supports_device_login(base_url: str, *, timeout: float = 5.0) -> bool: + """Whether the host advertises the device grant in its RFC 8414 metadata. + + Fails closed: any connection error, non-200, unparseable body, or missing + capability returns False, so self-hosted / non-managed instances simply + don't offer device login. + """ + host = base_url.rstrip("/") + try: + resp = httpx.get(f"{host}{AUTH_SERVER_METADATA_PATH}", timeout=timeout) + except httpx.HTTPError: + return False + if resp.status_code != 200: + return False + try: + body = resp.json() + except ValueError: + return False + grants = body.get("grant_types_supported") if isinstance(body, dict) else None + return isinstance(grants, list) and DEVICE_GRANT_TYPE in grants + + +def _post(url: str, data: dict[str, str]) -> httpx.Response: + """POST form data, surfacing transport failures as ``OAuthFlowError``. + + Connection refusals, DNS failures, and timeouts would otherwise escape as + raw ``httpx.HTTPError`` past callers that only catch ``OAuthFlowError``. + """ + try: + return httpx.post(url, data=data) + except httpx.HTTPError as e: + raise OAuthFlowError("connection_error", f"could not reach {url}: {e}") from e + + +def _error_from_response(resp: httpx.Response) -> tuple[str, str | None]: + """Pull ``(error, error_description)`` out of an OAuth error body.""" + try: + body = resp.json() + except ValueError: + return "invalid_response", resp.text[:200] or None + if isinstance(body, dict) and body.get("error"): + return str(body["error"]), body.get("error_description") + return "invalid_response", None + + +def request_device_code(endpoints: Endpoints) -> DeviceCode: + """Request a device + user code pair (RFC 8628 §3.1).""" + resp = _post( + endpoints.device_auth_url, + { + "client_id": endpoints.client_id, + "scope": endpoints.scope, + "source": DEVICE_SOURCE, + }, + ) + if resp.status_code != 200: + error, desc = _error_from_response(resp) + raise OAuthFlowError(error, desc) + try: + body = resp.json() + return DeviceCode( + device_code=body["device_code"], + user_code=body["user_code"], + verification_uri=body["verification_uri"], + verification_uri_complete=body.get( + "verification_uri_complete", body["verification_uri"] + ), + expires_in=int(body["expires_in"]), + interval=int(body["interval"]), + ) + except (KeyError, TypeError, ValueError) as e: + raise OAuthFlowError( + "invalid_response", f"malformed device authorization response: {e}" + ) from e + + +def _token_from_body(body: dict[str, Any]) -> TokenResponse: + # refresh_token is optional on the refresh grant (RFC 6749 §5.1); a + # malformed/missing field is a server fault, surfaced as OAuthFlowError so + # callers' existing handling catches it instead of a raw KeyError/ValueError + try: + return TokenResponse( + access_token=body["access_token"], + refresh_token=body.get("refresh_token", ""), + expires_in=int(body["expires_in"]), + scope=body.get("scope", ""), + config=body.get("config") or {}, + ) + except (KeyError, TypeError, ValueError) as e: + raise OAuthFlowError("invalid_response", f"malformed token response: {e}") from e + + +def poll_for_token( + endpoints: Endpoints, + device: DeviceCode, + *, + sleep: Callable[[float], None] = time.sleep, + monotonic: Callable[[], float] = time.monotonic, +) -> TokenResponse: + """Poll the token endpoint until the grant is approved (RFC 8628 §3.4/§3.5). + + Sleeps ``interval`` between polls, bumping it on ``slow_down``. Raises + ``AccessDenied`` / ``DeviceCodeExpired`` / ``AuthorizationTimeout`` on the + terminal outcomes. ``sleep`` / ``monotonic`` are injectable for tests. + """ + interval = device.interval + deadline = monotonic() + device.expires_in + while True: + if monotonic() >= deadline: + raise AuthorizationTimeout("expired_token", "Timed out waiting for approval") + sleep(interval) + resp = _post( + endpoints.token_url, + { + "grant_type": DEVICE_GRANT_TYPE, + "device_code": device.device_code, + "client_id": endpoints.client_id, + }, + ) + if resp.status_code == 200: + try: + body = resp.json() + except ValueError as e: + raise OAuthFlowError("invalid_response", "non-JSON token response") from e + return _token_from_body(body) + + error, desc = _error_from_response(resp) + if error == "authorization_pending": + continue + if error == "slow_down": + interval += SLOW_DOWN_STEP + continue + if error == "access_denied": + raise AccessDenied(error, desc) + if error == "expired_token": + raise DeviceCodeExpired(error, desc) + raise OAuthFlowError(error, desc) + + +def refresh_access_token(endpoints: Endpoints, refresh_token: str) -> TokenResponse: + """Exchange a refresh token for a fresh access/refresh pair. + + The response may rotate the refresh token; the caller must persist the + returned ``refresh_token`` before reusing it — replaying a superseded one + revokes the grant. + """ + resp = _post( + endpoints.token_url, + { + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "client_id": endpoints.client_id, + }, + ) + if resp.status_code != 200: + error, desc = _error_from_response(resp) + raise OAuthFlowError(error, desc) + try: + body = resp.json() + except ValueError as e: + raise OAuthFlowError("invalid_response", "non-JSON token response") from e + return _token_from_body(body) diff --git a/honcho-cli/tests/conftest.py b/honcho-cli/tests/conftest.py new file mode 100644 index 00000000..5f121748 --- /dev/null +++ b/honcho-cli/tests/conftest.py @@ -0,0 +1,21 @@ +"""Shared test fixtures.""" + +from __future__ import annotations + +import pytest + +from honcho_cli import common +from honcho_cli.output import set_json_mode + + +@pytest.fixture(autouse=True) +def _reset_cli_globals(): + """Reset process-global CLI state between tests. + + ``_global_overrides`` (set by ``-w``/``-p``/``-s`` flags) and the JSON-mode + flag are module globals that leak across tests otherwise — a workspace set + by one test would silently satisfy the next test's workspace check. + """ + yield + common._global_overrides.update(workspace=None, peer=None, session=None) + set_json_mode(False) diff --git a/honcho-cli/tests/test_common.py b/honcho-cli/tests/test_common.py new file mode 100644 index 00000000..f5b2ec27 --- /dev/null +++ b/honcho-cli/tests/test_common.py @@ -0,0 +1,126 @@ +"""Tests for the client factory's transparent OAuth refresh.""" + +from __future__ import annotations + +import json +import os +import time +from unittest.mock import patch + +import pytest +import typer +from honcho_cli import common +from honcho_cli.config import CLIConfig, OAuthTokens +from honcho_cli.oauth import OAuthFlowError, TokenResponse + + +@pytest.fixture +def cfg_path(tmp_path, monkeypatch): + f = tmp_path / "config.json" + monkeypatch.setattr("honcho_cli.config.CONFIG_FILE", f) + monkeypatch.setattr("honcho_cli.config.CONFIG_DIR", tmp_path) + for k in [k for k in os.environ if k.startswith("HONCHO_")]: + monkeypatch.delenv(k) + return f + + +def _cfg(expires_at: float) -> CLIConfig: + return CLIConfig( + base_url="http://localhost:8000", + oauth=OAuthTokens( + access_token="old-at", + refresh_token="old-rt", + access_expires_at=expires_at, + client_id="honcho-cli", + scope="write", + ), + ) + + +def test_valid_token_is_not_refreshed(cfg_path): + config = _cfg(time.time() + 3600) + with patch("honcho_cli.oauth.refresh_access_token") as refresh: + common.maybe_refresh_token(config) + refresh.assert_not_called() + + +def test_expired_token_refreshes_despite_manual_key(cfg_path): + """OAuth wins over apiKey now, so the grant is kept alive even with a key set.""" + config = _cfg(time.time() - 100) + config.api_key = "manual" + rotated = TokenResponse( + access_token="new-at", refresh_token="new-rt", expires_in=3600, scope="write" + ) + with patch("honcho_cli.oauth.refresh_access_token", return_value=rotated) as refresh: + common.maybe_refresh_token(config) + refresh.assert_called_once() + assert config.resolved_api_key() == "new-at" + + +def test_host_mismatch_skips_refresh(cfg_path): + """A grant minted for another host is ignored — no refresh, apiKey covers this one.""" + config = _cfg(time.time() - 100) + config.oauth.host = "https://staging.example.com" + config.api_key = "manual" + with patch("honcho_cli.oauth.refresh_access_token") as refresh: + common.maybe_refresh_token(config) + refresh.assert_not_called() + assert config.resolved_api_key() == "manual" + + +def test_expired_token_refreshes_and_persists(cfg_path): + config = _cfg(time.time() - 100) + rotated = TokenResponse( + access_token="new-at", + refresh_token="new-rt", + expires_in=3600, + scope="write", + ) + with patch("honcho_cli.oauth.refresh_access_token", return_value=rotated) as refresh: + common.maybe_refresh_token(config) + + # used the stored refresh token + client_id + _endpoints, sent_rt = refresh.call_args.args + assert sent_rt == "old-rt" + assert _endpoints.client_id == "honcho-cli" + + # in-memory config updated with the rotated pair + assert config.oauth.access_token == "new-at" + assert config.oauth.refresh_token == "new-rt" + assert config.oauth.access_valid() + + # rotation persisted to disk before reuse + on_disk = json.loads(cfg_path.read_text())["oauth"] + assert on_disk["accessToken"] == "new-at" + assert on_disk["refreshToken"] == "new-rt" + + +def test_refresh_failure_exits(cfg_path): + config = _cfg(time.time() - 100) + with patch("honcho_cli.oauth.refresh_access_token", side_effect=OAuthFlowError("invalid_grant")): + with pytest.raises(typer.Exit): + common.maybe_refresh_token(config) + + +def test_refresh_failure_falls_back_to_api_key(cfg_path): + """A dead grant degrades to the saved apiKey instead of aborting.""" + config = _cfg(time.time() - 100) + config.api_key = "manual" + with patch("honcho_cli.oauth.refresh_access_token", side_effect=OAuthFlowError("invalid_grant")): + common.maybe_refresh_token(config) # must not raise + assert config.resolved_api_key() == "manual" + + +def test_missing_refresh_token_exits(cfg_path): + config = _cfg(time.time() - 100) + config.oauth.refresh_token = "" + with pytest.raises(typer.Exit): + common.maybe_refresh_token(config) + + +def test_missing_refresh_token_falls_back_to_api_key(cfg_path): + config = _cfg(time.time() - 100) + config.oauth.refresh_token = "" + config.api_key = "manual" + common.maybe_refresh_token(config) # must not raise + assert config.resolved_api_key() == "manual" diff --git a/honcho-cli/tests/test_config.py b/honcho-cli/tests/test_config.py index bdac2bc4..5b90087d 100644 --- a/honcho-cli/tests/test_config.py +++ b/honcho-cli/tests/test_config.py @@ -2,9 +2,12 @@ import json import os +import time +from pathlib import Path import pytest -from honcho_cli.config import CLIConfig +from honcho_cli.config import CLIConfig, OAuthTokens, _config_dir +from honcho_cli.oauth import TokenResponse @pytest.fixture @@ -18,6 +21,20 @@ def cfg_path(tmp_path, monkeypatch): return f +class TestConfigDir: + def test_defaults_to_dot_honcho(self, monkeypatch): + monkeypatch.delenv("HONCHO_CONFIG_DIR", raising=False) + assert _config_dir() == Path.home() / ".honcho" + + def test_honcho_config_dir_override(self, monkeypatch, tmp_path): + monkeypatch.setenv("HONCHO_CONFIG_DIR", str(tmp_path / "profile")) + assert _config_dir() == tmp_path / "profile" + + def test_expands_user_in_override(self, monkeypatch): + monkeypatch.setenv("HONCHO_CONFIG_DIR", "~/.honcho-test") + assert _config_dir() == Path.home() / ".honcho-test" + + class TestLoad: def test_defaults_when_no_file(self, cfg_path): loaded = CLIConfig.load() @@ -44,6 +61,31 @@ class TestLoad: assert loaded.api_key == "env-key" assert loaded.base_url == "http://localhost:8000" + def test_empty_env_var_popped_from_environ(self, cfg_path, monkeypatch): + """Empty HONCHO_* vars are removed so the SDK doesn't crash on them.""" + cfg_path.write_text(json.dumps({"apiKey": "file-key"})) + monkeypatch.setenv("HONCHO_API_KEY", "") + loaded = CLIConfig.load() + assert "HONCHO_API_KEY" not in os.environ + assert loaded.api_key == "file-key" + + def test_garbage_access_expires_at_treated_as_expired(self, cfg_path): + """Hand-edited/corrupt expiry degrades to the refresh path, not a crash.""" + cfg_path.write_text(json.dumps( + {"oauth": {"accessToken": "x", "accessExpiresAt": "not-a-number"}} + )) + loaded = CLIConfig.load() + assert loaded.oauth is not None + assert loaded.oauth.access_valid() is False + + def test_numeric_string_access_expires_at_parses(self, cfg_path): + cfg_path.write_text(json.dumps( + {"oauth": {"accessToken": "x", "accessExpiresAt": "12345"}} + )) + loaded = CLIConfig.load() + assert loaded.oauth is not None + assert loaded.oauth.access_expires_at == 12345.0 + class TestSave: def test_writes_only_cli_owned_keys(self, cfg_path): @@ -104,6 +146,117 @@ def test_api_key_redaction_empty_omitted(): assert "api_key" not in CLIConfig(api_key="").redacted() +class TestOAuth: + def _tokens(self, expires_at: float) -> OAuthTokens: + return OAuthTokens( + access_token="hch-at-x", + refresh_token="hch-rt-x", + access_expires_at=expires_at, + client_id="honcho-cli", + scope="write", + ) + + def test_round_trips_oauth_block(self, cfg_path): + CLIConfig(base_url="http://localhost:8000", oauth=self._tokens(9999999999)).save() + loaded = CLIConfig.load() + assert loaded.oauth is not None + assert loaded.oauth.access_token == "hch-at-x" + assert loaded.oauth.refresh_token == "hch-rt-x" + assert loaded.oauth.client_id == "honcho-cli" + + def test_oauth_persists_camelcase_keys(self, cfg_path): + CLIConfig(base_url="http://localhost:8000", oauth=self._tokens(1234)).save() + on_disk = json.loads(cfg_path.read_text())["oauth"] + assert set(on_disk) == {"accessToken", "refreshToken", "accessExpiresAt", "clientId", "scope", "host"} + + def test_save_preserves_foreign_keys_with_oauth(self, cfg_path): + cfg_path.write_text(json.dumps({"hosts": {"claude_code": {"peerName": "u"}}})) + CLIConfig(base_url="http://localhost:8000", oauth=self._tokens(1234)).save() + on_disk = json.loads(cfg_path.read_text()) + assert on_disk["hosts"] == {"claude_code": {"peerName": "u"}} + assert "oauth" in on_disk + + def test_empty_oauth_is_dropped(self, cfg_path): + cfg_path.write_text(json.dumps({"oauth": {"accessToken": "old"}})) + CLIConfig(base_url="http://localhost:8000").save() + assert "oauth" not in json.loads(cfg_path.read_text()) + + def test_api_key_preserved_on_device_login(self, cfg_path): + """apiKey is shared with sibling tools — device login must not delete it.""" + cfg_path.write_text(json.dumps({"apiKey": "shared-key"})) + CLIConfig(base_url="http://localhost:8000", oauth=self._tokens(9999999999)).save() + on_disk = json.loads(cfg_path.read_text()) + assert on_disk["apiKey"] == "shared-key" + assert on_disk["oauth"]["accessToken"] == "hch-at-x" + + def test_resolved_api_key_prefers_live_oauth(self, cfg_path): + cfg = CLIConfig(api_key="manual", oauth=self._tokens(9999999999)) + assert cfg.resolved_api_key() == "hch-at-x" + + def test_resolved_api_key_expired_oauth_falls_back_to_api_key(self, cfg_path): + cfg = CLIConfig(api_key="manual", oauth=self._tokens(time.time() - 100)) + assert cfg.resolved_api_key() == "manual" + + def test_resolved_api_key_host_mismatch_falls_back_to_api_key(self, cfg_path): + tokens = self._tokens(9999999999) + tokens.host = "https://staging.example.com" + cfg = CLIConfig( + base_url="https://api.honcho.dev", api_key="manual", oauth=tokens + ) + assert cfg.resolved_api_key() == "manual" + + def test_resolved_api_key_expired_oauth_wins_over_nothing(self, cfg_path): + cfg = CLIConfig(oauth=self._tokens(time.time() - 100)) + assert cfg.resolved_api_key() == "hch-at-x" + + def test_resolved_api_key_falls_back_to_oauth(self, cfg_path): + cfg = CLIConfig(oauth=self._tokens(9999999999)) + assert cfg.resolved_api_key() == "hch-at-x" + + def test_access_valid_expiry_and_skew(self): + assert self._tokens(time.time() + 3600).access_valid() + assert not self._tokens(time.time() - 10).access_valid() + # inside the default 60s skew window → treated as invalid + assert not self._tokens(time.time() + 30).access_valid() + + def test_access_valid_false_without_token(self): + """A missing token is invalid even with a far-future expiry.""" + tokens = OAuthTokens(access_token="", access_expires_at=time.time() + 3600) + assert tokens.access_valid() is False + + def test_from_response_keeps_prior_refresh_token_when_not_rotated(self): + """Refresh-token rotation is optional (RFC 6749 §5.1) — keep the old one.""" + resp = TokenResponse( + access_token="new-at", refresh_token="", expires_in=3600, scope="" + ) + tokens = OAuthTokens.from_response( + resp, + client_id="honcho-cli", + scope_fallback="write", + refresh_fallback="prior-rt", + host="https://staging.example.com", + ) + assert tokens.refresh_token == "prior-rt" + assert tokens.scope == "write" + assert tokens.host == "https://staging.example.com" + + def test_host_round_trips_and_legacy_matches_all(self, cfg_path): + tokens = self._tokens(9999999999) + tokens.host = "https://staging.example.com" + CLIConfig(base_url="https://staging.example.com", oauth=tokens).save() + loaded = CLIConfig.load() + assert loaded.oauth is not None + assert loaded.oauth.host == "https://staging.example.com" + # trailing-slash normalization + legacy blocks (no host) trust any host + assert loaded.oauth.matches_host("https://staging.example.com/") + assert not loaded.oauth.matches_host("https://api.honcho.dev") + assert OAuthTokens(access_token="x").matches_host("https://anything.dev") + + def test_redacted_masks_oauth_token(self): + red = CLIConfig(oauth=self._tokens(1234)).redacted() + assert red["oauth"] == "***at-x" + + def test_save_sets_600_permissions(cfg_path): """Config with plaintext API key must be owner-readable only on POSIX.""" import stat diff --git a/honcho-cli/tests/test_oauth.py b/honcho-cli/tests/test_oauth.py new file mode 100644 index 00000000..45121130 --- /dev/null +++ b/honcho-cli/tests/test_oauth.py @@ -0,0 +1,232 @@ +"""Tests for the device-authorization OAuth engine (transport-only).""" + +from __future__ import annotations + +from unittest.mock import patch + +import httpx +import pytest +from honcho_cli import oauth +from honcho_cli.oauth import ( + AccessDenied, + AuthorizationTimeout, + DeviceCode, + DeviceCodeExpired, + Endpoints, + OAuthFlowError, +) + + +class FakeResponse: + def __init__(self, status_code: int, body): + self.status_code = status_code + self._body = body + self.text = str(body) + + def json(self): + if isinstance(self._body, Exception): + raise self._body + return self._body + + +def _endpoints() -> Endpoints: + return Endpoints( + device_auth_url="https://api.honcho.dev/oauth/device_authorization", + token_url="https://api.honcho.dev/oauth/token", + client_id="honcho-cli", + scope="write", + ) + + +DEVICE = DeviceCode( + device_code="dev-abc", + user_code="WXYZ-1234", + verification_uri="https://app.honcho.dev/device", + verification_uri_complete="https://app.honcho.dev/device?user_code=WXYZ-1234", + expires_in=600, + interval=5, +) + + +# --------------------------------------------------------------------------- # +# resolve_endpoints + +class TestResolveEndpoints: + def test_derives_urls_from_base_url(self): + ep = oauth.resolve_endpoints("https://api.honcho.dev") + assert ep.device_auth_url == "https://api.honcho.dev/oauth/device_authorization" + assert ep.token_url == "https://api.honcho.dev/oauth/token" + assert ep.client_id == "honcho-cli" + assert ep.scope == "write" + + def test_strips_trailing_slash(self): + ep = oauth.resolve_endpoints("http://localhost:8000/") + assert ep.token_url == "http://localhost:8000/oauth/token" + + +# --------------------------------------------------------------------------- # +# supports_device_login + +class TestSupportsDeviceLogin: + def test_true_when_device_grant_advertised(self): + body = {"grant_types_supported": ["authorization_code", oauth.DEVICE_GRANT_TYPE]} + with patch("honcho_cli.oauth.httpx.get", return_value=FakeResponse(200, body)): + assert oauth.supports_device_login("https://api.honcho.dev") is True + + def test_false_when_device_grant_absent(self): + body = {"grant_types_supported": ["authorization_code", "refresh_token"]} + with patch("honcho_cli.oauth.httpx.get", return_value=FakeResponse(200, body)): + assert oauth.supports_device_login("https://api.honcho.dev") is False + + @pytest.mark.parametrize("status", [404, 500]) + def test_false_on_non_200(self, status): + with patch("honcho_cli.oauth.httpx.get", return_value=FakeResponse(status, "")): + assert oauth.supports_device_login("http://localhost:8000") is False + + def test_false_on_connection_error(self): + with patch("honcho_cli.oauth.httpx.get", side_effect=httpx.ConnectError("no route")): + assert oauth.supports_device_login("http://localhost:8000") is False + + def test_false_on_unparseable_body(self): + with patch("honcho_cli.oauth.httpx.get", return_value=FakeResponse(200, ValueError())): + assert oauth.supports_device_login("https://api.honcho.dev") is False + + +# --------------------------------------------------------------------------- # +# request_device_code + +class TestRequestDeviceCode: + def test_success(self): + body = { + "device_code": "dev-abc", + "user_code": "WXYZ-1234", + "verification_uri": "https://app.honcho.dev/device", + "verification_uri_complete": "https://app.honcho.dev/device?user_code=WXYZ-1234", + "expires_in": 600, + "interval": 5, + } + with patch("honcho_cli.oauth.httpx.post", return_value=FakeResponse(200, body)) as post: + dc = oauth.request_device_code(_endpoints()) + assert dc.device_code == "dev-abc" + assert dc.user_code == "WXYZ-1234" + assert dc.interval == 5 + assert post.call_args.kwargs["data"]["source"] == "honcho-cli" + + def test_error_raises(self): + body = {"error": "invalid_client", "error_description": "unknown client"} + with patch("honcho_cli.oauth.httpx.post", return_value=FakeResponse(401, body)): + with pytest.raises(OAuthFlowError) as exc: + oauth.request_device_code(_endpoints()) + assert exc.value.error == "invalid_client" + + def test_transport_failure_wrapped(self): + with patch("honcho_cli.oauth.httpx.post", side_effect=httpx.ConnectError("no route")): + with pytest.raises(OAuthFlowError) as exc: + oauth.request_device_code(_endpoints()) + assert exc.value.error == "connection_error" + + +# --------------------------------------------------------------------------- # +# poll_for_token + +class TestPollForToken: + def _run(self, responses, monotonic_vals=None): + """Poll with a scripted response sequence, capturing sleep durations.""" + sleeps: list[float] = [] + clock = iter(monotonic_vals or [0.0] * (len(responses) + 2)) + with patch("honcho_cli.oauth.httpx.post", side_effect=responses): + token = oauth.poll_for_token( + _endpoints(), + DEVICE, + sleep=sleeps.append, + monotonic=lambda: next(clock), + ) + return token, sleeps + + def test_pending_then_slowdown_then_success(self): + success = { + "access_token": "hch-at-1", + "refresh_token": "hch-rt-1", + "expires_in": 3600, + "scope": "write", + "config": {"k": "v"}, + } + responses = [ + FakeResponse(400, {"error": "authorization_pending"}), + FakeResponse(400, {"error": "slow_down"}), + FakeResponse(200, success), + ] + token, sleeps = self._run(responses) + assert token.access_token == "hch-at-1" + assert token.refresh_token == "hch-rt-1" + assert token.config == {"k": "v"} + # interval starts at 5, bumps by 5 after slow_down → third sleep is 10 + assert sleeps == [5, 5, 10] + + def test_access_denied(self): + responses = [FakeResponse(400, {"error": "access_denied"})] + with pytest.raises(AccessDenied): + self._run(responses) + + def test_expired_token(self): + responses = [FakeResponse(400, {"error": "expired_token"})] + with pytest.raises(DeviceCodeExpired): + self._run(responses) + + def test_unexpected_error_raises_generic(self): + responses = [FakeResponse(400, {"error": "invalid_grant"})] + with pytest.raises(OAuthFlowError) as exc: + self._run(responses) + assert exc.value.error == "invalid_grant" + + def test_transport_failure_wrapped(self): + # an exception in the side_effect list is raised on that poll + responses = [httpx.ReadTimeout("timed out")] + with pytest.raises(OAuthFlowError) as exc: + self._run(responses) + assert exc.value.error == "connection_error" + + def test_times_out_past_deadline(self): + # monotonic jumps past deadline (0 + expires_in) on the first check + with patch("honcho_cli.oauth.httpx.post") as post: + with pytest.raises(AuthorizationTimeout): + oauth.poll_for_token( + _endpoints(), + DEVICE, + sleep=lambda _s: None, + monotonic=iter([0.0, 9999.0]).__next__, + ) + post.assert_not_called() + + +# --------------------------------------------------------------------------- # +# refresh_access_token + +class TestRefresh: + def test_success_returns_rotated_pair(self): + body = { + "access_token": "hch-at-2", + "refresh_token": "hch-rt-2", + "expires_in": 3600, + "scope": "write", + } + with patch("honcho_cli.oauth.httpx.post", return_value=FakeResponse(200, body)) as post: + token = oauth.refresh_access_token(_endpoints(), "hch-rt-1") + assert token.access_token == "hch-at-2" + assert token.refresh_token == "hch-rt-2" + sent = post.call_args.kwargs["data"] + assert sent["grant_type"] == "refresh_token" + assert sent["refresh_token"] == "hch-rt-1" + + def test_error_raises(self): + body = {"error": "invalid_grant", "error_description": "revoked"} + with patch("honcho_cli.oauth.httpx.post", return_value=FakeResponse(400, body)): + with pytest.raises(OAuthFlowError) as exc: + oauth.refresh_access_token(_endpoints(), "stale") + assert exc.value.error == "invalid_grant" + + def test_transport_failure_wrapped(self): + with patch("honcho_cli.oauth.httpx.post", side_effect=httpx.ConnectError("no route")): + with pytest.raises(OAuthFlowError) as exc: + oauth.refresh_access_token(_endpoints(), "hch-rt-1") + assert exc.value.error == "connection_error"