feat(cli): add device code oauth login for honcho servers (#891)
* feat(cli): add device code oauth login for honcho servers * fix(cli): harden device-auth input validation and transport errors Reject zero/negative auth-method choices instead of letting Python negative indexing wrap to the tail of the options list, and wrap httpx transport failures in the OAuth POST helpers as OAuthFlowError so connection errors surface through existing caller handling rather than escaping as an uncaught traceback. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * feat: add HONCHO_CONFIG_DIR env var for cli * refactor(cli): address review nits on device-auth PR - split `init` into manual-key and interactive helpers - document that access_valid checks persisted expiry, not the token - rename single-letter local in redacted() - cover 500 alongside 404 in supports_device_login metadata probe - add config edge-case tests: stale apiKey drop, garbage/string accessExpiresAt, empty-env-var popping, refresh-rotation fallback, missing-token access_valid Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(cli): stop deleting shared apiKey on device login; bind oauth grant to its host apiKey is shared with sibling tools that read it from the same config file, so device login must not remove it. Precedence flips to a live OAuth token over apiKey; a dead grant now degrades to the saved key with a warning instead of aborting. The oauth block records the host it was minted against and is ignored (no use, no refresh) when base_url points elsewhere, so a staging grant is never sent to prod. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Aakash Kattelu
GitHub
parent
68c46cde81
commit
055d73b580
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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 ``***<last4>`` — 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 ``***<last4>`` 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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -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)
|
||||
|
|
@ -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"
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
Loading…
Reference in New Issue