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>
This commit is contained in:
Aakash Kattelu 2026-07-09 14:38:14 -07:00
parent 41be8496d8
commit 302cf9b71d
3 changed files with 44 additions and 8 deletions

View File

@ -191,9 +191,14 @@ def _prompt_auth_method(has_creds: bool, device_available: bool) -> str:
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):
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:

View File

@ -118,6 +118,18 @@ def supports_device_login(base_url: str, *, timeout: float = 5.0) -> bool:
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:
@ -131,9 +143,9 @@ def _error_from_response(resp: httpx.Response) -> tuple[str, str | None]:
def request_device_code(endpoints: Endpoints) -> DeviceCode:
"""Request a device + user code pair (RFC 8628 §3.1)."""
resp = httpx.post(
resp = _post(
endpoints.device_auth_url,
data={
{
"client_id": endpoints.client_id,
"scope": endpoints.scope,
"source": DEVICE_SOURCE,
@ -195,9 +207,9 @@ def poll_for_token(
if monotonic() >= deadline:
raise AuthorizationTimeout("expired_token", "Timed out waiting for approval")
sleep(interval)
resp = httpx.post(
resp = _post(
endpoints.token_url,
data={
{
"grant_type": DEVICE_GRANT_TYPE,
"device_code": device.device_code,
"client_id": endpoints.client_id,
@ -230,9 +242,9 @@ def refresh_access_token(endpoints: Endpoints, refresh_token: str) -> TokenRespo
returned ``refresh_token`` before reusing it replaying a superseded one
revokes the grant.
"""
resp = httpx.post(
resp = _post(
endpoints.token_url,
data={
{
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": endpoints.client_id,

View File

@ -118,6 +118,12 @@ class TestRequestDeviceCode:
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
@ -172,6 +178,13 @@ class TestPollForToken:
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:
@ -210,3 +223,9 @@ class TestRefresh:
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"