feat(cli): add device code oauth login for honcho servers
This commit is contained in:
Aakash Kattelu
parent
0cb0c9abf0
commit
41be8496d8
|
|
@ -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,19 +120,124 @@ def init(
|
|||
_console.print()
|
||||
_console.print()
|
||||
|
||||
final_key = _prompt_api_key(key_val)
|
||||
# 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:
|
||||
final_key = _prompt_api_key(key_val)
|
||||
final_url = _prompt_url(url_val)
|
||||
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})
|
||||
return
|
||||
|
||||
# Interactive: 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)
|
||||
|
||||
# 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():
|
||||
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)
|
||||
|
||||
if use_json():
|
||||
print_result({"apiKey": _redact(final_key), "baseUrl": final_url})
|
||||
|
||||
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:
|
||||
return options[int(choice) - 1]
|
||||
except (ValueError, IndexError):
|
||||
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
|
||||
)
|
||||
|
||||
|
||||
def _prompt_api_key(value: str) -> str:
|
||||
|
|
@ -206,6 +314,18 @@ 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."""
|
||||
if config.api_key:
|
||||
return "API key"
|
||||
if config.oauth and config.oauth.access_token:
|
||||
if config.oauth.access_valid():
|
||||
secs = max(int(config.oauth.access_expires_at - time.time()), 0)
|
||||
return f"OAuth device token (expires in {secs // 60}m)"
|
||||
return "OAuth device token (expired — will refresh)"
|
||||
return "missing — run `honcho init`"
|
||||
|
||||
|
||||
def doctor(
|
||||
json_output: bool = typer.Option(False, "--json", help="Force JSON output"),
|
||||
) -> None:
|
||||
|
|
@ -230,23 +350,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 +407,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,41 @@ 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 a manual api_key is set or the access token is still valid. On
|
||||
refresh failure the grant is gone — prints a re-login hint and exits.
|
||||
"""
|
||||
if config.api_key: # manual key takes precedence; nothing to refresh
|
||||
return
|
||||
tokens = config.oauth
|
||||
if tokens is None or tokens.access_valid():
|
||||
return
|
||||
if not tokens.refresh_token:
|
||||
print_error("SESSION_EXPIRED", "OAuth session expired. Run `honcho init` to log in again.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
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:
|
||||
print_error("SESSION_EXPIRED", "OAuth session expired. Run `honcho init` to log in again.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
# 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,
|
||||
)
|
||||
config.save()
|
||||
|
||||
|
||||
def get_client(*, require_workspace: bool = True):
|
||||
"""Create a Honcho client from resolved config.
|
||||
|
||||
|
|
@ -65,6 +102,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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -2,10 +2,12 @@
|
|||
|
||||
Config stored at ``~/.honcho/config.json`` with env var overrides.
|
||||
|
||||
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
|
||||
apiKey -- Honcho admin JWT (manual auth)
|
||||
environmentUrl -- Honcho API URL (full URL, e.g. https://api.honcho.dev)
|
||||
oauth -- OAuth device-grant tokens (accessToken, refreshToken,
|
||||
accessExpiresAt, clientId, scope), written by device login
|
||||
|
||||
All other top-level keys (``hosts``, ``sessions``, ``saveMessages``,
|
||||
``sessionStrategy``, …) are written by sibling Honcho tools and are
|
||||
|
|
@ -20,14 +22,38 @@ from __future__ import annotations
|
|||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from dataclasses import dataclass, fields
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from honcho_cli.oauth import TokenResponse
|
||||
|
||||
CONFIG_DIR = Path.home() / ".honcho"
|
||||
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 +66,43 @@ 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 = ""
|
||||
|
||||
def access_valid(self, skew: int = 60) -> bool:
|
||||
"""True while the access token is present and not within ``skew`` of expiry."""
|
||||
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 = "",
|
||||
) -> 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,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CLIConfig:
|
||||
"""CLI configuration with layered resolution: flag > env > file > default.
|
||||
|
|
@ -54,6 +117,15 @@ class CLIConfig:
|
|||
workspace_id: str = ""
|
||||
peer_id: str = ""
|
||||
session_id: str = ""
|
||||
oauth: OAuthTokens | None = None
|
||||
|
||||
def resolved_api_key(self) -> str:
|
||||
"""The key handed to the SDK: manual apiKey wins, else the OAuth token."""
|
||||
if self.api_key:
|
||||
return self.api_key
|
||||
if self.oauth and self.oauth.access_token:
|
||||
return self.oauth.access_token
|
||||
return ""
|
||||
|
||||
@classmethod
|
||||
def load(cls) -> CLIConfig:
|
||||
|
|
@ -74,6 +146,15 @@ 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", "")),
|
||||
)
|
||||
|
||||
for fld_name, env_var in ENV_MAP.items():
|
||||
val = os.environ.get(env_var)
|
||||
|
|
@ -111,6 +192,17 @@ class CLIConfig:
|
|||
else:
|
||||
data.pop("apiKey", None)
|
||||
|
||||
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,
|
||||
}
|
||||
else:
|
||||
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.
|
||||
try:
|
||||
|
|
@ -132,7 +224,9 @@ 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 "***"
|
||||
d[fld.name] = _redact_token(val)
|
||||
elif fld.name == "oauth":
|
||||
d[fld.name] = _redact_token(val.access_token)
|
||||
else:
|
||||
d[fld.name] = val
|
||||
return d
|
||||
|
|
@ -143,8 +237,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,248 @@
|
|||
"""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 _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 = httpx.post(
|
||||
endpoints.device_auth_url,
|
||||
data={
|
||||
"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 = httpx.post(
|
||||
endpoints.token_url,
|
||||
data={
|
||||
"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 = httpx.post(
|
||||
endpoints.token_url,
|
||||
data={
|
||||
"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,93 @@
|
|||
"""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_manual_key_short_circuits(cfg_path):
|
||||
config = _cfg(time.time() - 100)
|
||||
config.api_key = "manual"
|
||||
with patch("honcho_cli.oauth.refresh_access_token") as refresh:
|
||||
common.maybe_refresh_token(config)
|
||||
refresh.assert_not_called()
|
||||
|
||||
|
||||
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_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)
|
||||
|
|
@ -2,9 +2,10 @@
|
|||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from honcho_cli.config import CLIConfig
|
||||
from honcho_cli.config import CLIConfig, OAuthTokens
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -104,6 +105,60 @@ 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"}
|
||||
|
||||
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_resolved_api_key_prefers_manual_key(self, cfg_path):
|
||||
cfg = CLIConfig(api_key="manual", oauth=self._tokens(9999999999))
|
||||
assert cfg.resolved_api_key() == "manual"
|
||||
|
||||
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_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,212 @@
|
|||
"""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
|
||||
|
||||
def test_false_on_non_200(self):
|
||||
with patch("honcho_cli.oauth.httpx.get", return_value=FakeResponse(404, "")):
|
||||
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"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 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_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"
|
||||
Loading…
Reference in New Issue