feat(auth): make xAI Grok OAuth device-code-only, drop loopback login

Replace the loopback/PKCE-callback server and manual-paste fallback with
the RFC 8628 device-code flow as the only xAI Grok OAuth login path. The
flow works in headless/SSH/container sessions with no 127.0.0.1 listener,
shrinking the local attack surface.

- Poll the token endpoint with server-provided interval, honoring
  slow_down and expires_in; store tokens with auth_mode
  oauth_device_code.
- Adaptive proactive refresh skew for short-lived device-code JWTs;
  rotated tokens sync back to auth.json, the global root store, and the
  credential pool (no refresh-token replay).
- Clear source suppression on successful re-login (CLI + dashboard) and
  drop the duplicate dashboard pool entry so exactly one seeded
  device_code entry exists.
- Use the shared device_code source name for consistency with the
  nous/codex device-code providers.
- Desktop: remove the loopback OAuth flow states and dead type variants;
  pkce providers' sign-in URL selection is unchanged.
- Docs (EN + zh-Hans) rewritten for device-code login; drop the deleted
  --manual-paste flag from documented commands.
This commit is contained in:
Jaaneek 2026-07-02 15:35:06 +00:00 committed by Teknium
parent 472d75193f
commit 5ef0b8acb0
38 changed files with 733 additions and 2904 deletions

View file

@ -4207,7 +4207,7 @@ def resolve_provider_client(
return (_to_async_client(client, final_model, is_vision=is_vision) if async_mode
else (client, final_model))
# ── xAI Grok OAuth (loopback PKCE → Responses API) ───────────────
# ── xAI Grok OAuth (device code → Responses API) ───────────────
# Without this branch, an xai-oauth main provider falls through to the
# generic ``oauth_external`` arm below and returns ``(None, None)``,
# silently re-routing every auxiliary task (compression, web extract,

View file

@ -22,7 +22,7 @@ _PERSISTABLE_PROVIDER_SOURCES = frozenset({
("minimax-oauth", "oauth"),
("nous", "device_code"),
("openai-codex", "device_code"),
("xai-oauth", "loopback_pkce"),
("xai-oauth", "device_code"),
})
_SAFE_SECRETISH_METADATA_KEYS = frozenset({

View file

@ -82,7 +82,7 @@ _TERMINAL_AUTH_REASONS = frozenset({
# without losing recoverability — the user always has the option to re-add
# via ``hermes auth add``.
#
# Singleton-seeded entries (``device_code``, ``loopback_pkce``, ``claude_code``)
# Singleton-seeded entries (``device_code``, ``claude_code``)
# are NOT pruned because ``_seed_from_singletons`` would just re-create them
# on the next ``load_pool()`` with the same stale singleton tokens, defeating
# the cleanup. They remain in the pool marked DEAD until an explicit re-auth
@ -724,11 +724,11 @@ class CredentialPool:
keeps the consumed refresh_token and the next ``_refresh_entry`` call
would replay it and get a ``refresh_token_reused``-style 4xx.
Only applies to entries seeded from the singleton (``loopback_pkce``);
manually added entries (``manual:xai_pkce``) are independent
credentials with their own refresh-token lifecycle.
Only applies to entries seeded from the singleton (``device_code``);
manually added entries are independent credentials with their own
refresh-token lifecycle.
"""
if self.provider != "xai-oauth" or entry.source != "loopback_pkce":
if self.provider != "xai-oauth" or entry.source != "device_code":
return entry
try:
with _auth_store_lock():
@ -868,8 +868,9 @@ class CredentialPool:
"""
# Only sync entries that were seeded *from* a singleton. Manually
# added pool entries (source="manual:*") are independent credentials
# and must not write back to the singleton.
if entry.source not in {"device_code", "loopback_pkce"}:
# and must not write back to the singleton. All singleton-seeded
# device-code sources (nous, openai-codex, xAI) use ``device_code``.
if entry.source != "device_code":
return
try:
with _auth_store_lock():
@ -1112,8 +1113,8 @@ class CredentialPool:
# consumed the refresh token between our proactive sync and the
# HTTP call. Re-check auth.json and adopt the fresh tokens if
# they have rotated since. Only meaningful for singleton-seeded
# (loopback_pkce) entries; manual entries don't share state with
# the singleton.
# (device_code) entries; manual entries don't share
# state with the singleton.
if self.provider == "xai-oauth":
synced = self._sync_xai_oauth_entry_from_auth_store(entry)
if synced.refresh_token != entry.refresh_token:
@ -1135,8 +1136,8 @@ class CredentialPool:
# Terminal error: auth.json has no newer tokens — the stored
# refresh_token is dead. Clear it from auth.json so the next
# session does not re-seed the same revoked credentials, and
# remove all singleton-seeded (loopback_pkce) entries from the
# in-memory pool. Mirrors the Nous quarantine path above.
# remove all singleton-seeded xAI entries from the in-memory
# pool. Mirrors the Nous quarantine path above.
if auth_mod._is_terminal_xai_oauth_refresh_error(exc):
logger.debug(
"xAI OAuth refresh token is terminally invalid; clearing local token state"
@ -1174,7 +1175,7 @@ class CredentialPool:
]
self._entries = [
item for item in self._entries
if item.source != "loopback_pkce"
if item.source != "device_code"
]
if self._current_id == entry.id:
self._current_id = None
@ -1352,7 +1353,7 @@ class CredentialPool:
if self.provider == "xai-oauth":
return auth_mod._xai_access_token_is_expiring(
entry.access_token,
auth_mod.XAI_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
auth_mod._xai_proactive_refresh_skew_seconds(entry.access_token),
)
if self.provider == "nous":
# Nous refresh can require network access and should happen when
@ -1414,7 +1415,7 @@ class CredentialPool:
# tokens that another process (or a fresh `hermes model` ->
# xAI Grok OAuth login) has since rotated in auth.json.
if (self.provider == "xai-oauth"
and entry.source == "loopback_pkce"
and entry.source == "device_code"
and entry.last_status in {STATUS_EXHAUSTED, STATUS_DEAD}):
synced = self._sync_xai_oauth_entry_from_auth_store(entry)
if synced is not entry:
@ -2064,28 +2065,30 @@ def _seed_from_singletons(provider: str, entries: List[PooledCredential]) -> Tup
# (``providers["xai-oauth"]``). Surface them in the pool too so
# ``hermes auth list`` reflects the logged-in state and so the pool
# is the single source of truth for refresh during runtime resolution.
if _is_suppressed(provider, "loopback_pkce"):
return changed, active_sources
state = _load_provider_state(auth_store, "xai-oauth")
tokens = state.get("tokens") if isinstance(state, dict) else None
if isinstance(tokens, dict) and tokens.get("access_token"):
active_sources.add("loopback_pkce")
# Device code is the only supported xAI OAuth flow; the singleton is
# always surfaced as ``device_code`` (consistent with nous/codex).
source = "device_code"
if _is_suppressed(provider, source):
return changed, active_sources
active_sources.add(source)
from hermes_cli.auth import DEFAULT_XAI_OAUTH_BASE_URL
base_url = DEFAULT_XAI_OAUTH_BASE_URL
changed |= _upsert_entry(
entries,
provider,
"loopback_pkce",
source,
{
"source": "loopback_pkce",
"source": source,
"auth_type": AUTH_TYPE_OAUTH,
"access_token": tokens.get("access_token", ""),
"refresh_token": tokens.get("refresh_token"),
"base_url": base_url,
"last_refresh": state.get("last_refresh"),
"label": label_from_token(tokens.get("access_token", ""), "loopback_pkce"),
"label": label_from_token(tokens.get("access_token", ""), source),
},
)

View file

@ -265,7 +265,7 @@ def _remove_minimax_oauth(provider: str, removed) -> RemovalResult:
return result
def _remove_xai_oauth_loopback_pkce(provider: str, removed) -> RemovalResult:
def _remove_xai_oauth_device_code(provider: str, removed) -> RemovalResult:
"""xAI OAuth tokens live in auth.json providers.xai-oauth — clear them.
Without this step, ``hermes auth remove xai-oauth <N>`` silently undoes
@ -275,11 +275,6 @@ def _remove_xai_oauth_loopback_pkce(provider: str, removed) -> RemovalResult:
entry from the still-present singleton credentials reappear with no
user feedback. Clearing the singleton in step with the suppression set
by the central dispatcher makes the removal stick.
Belt-and-braces against the manual entry path: ``hermes auth add
xai-oauth`` produces a ``manual:xai_pkce`` entry whose removal step
falls through to "unregistered → nothing to clean up" (correct
manual entries are pool-only).
"""
result = RemovalResult()
if _clear_auth_store_provider(provider):
@ -423,8 +418,8 @@ def _register_all_sources() -> None:
description="auth.json providers.openai-codex + ~/.codex/auth.json",
))
register(RemovalStep(
provider="xai-oauth", source_id="loopback_pkce",
remove_fn=_remove_xai_oauth_loopback_pkce,
provider="xai-oauth", source_id="device_code",
remove_fn=_remove_xai_oauth_device_code,
description="auth.json providers.xai-oauth",
))
register(RemovalStep(

View file

@ -96,21 +96,6 @@ export function FlowPanel({
)
}
if (flow.status === 'awaiting_browser') {
return (
<Step title={t.onboarding.signInWith(title)}>
<p className="text-sm text-muted-foreground">{t.onboarding.autoBrowser(title)}</p>
<FlowFooter left={<DocsLink href={flow.start.auth_url}>{t.onboarding.reopenSignInPage}</DocsLink>}>
<span className="flex items-center gap-2 text-xs text-muted-foreground">
<Loader2 className="size-3 animate-spin" />
{t.onboarding.waitingAuthorize}
</span>
<CancelBtn size="sm" />
</FlowFooter>
</Step>
)
}
if (flow.status === 'external_pending') {
return (
<Step title={t.onboarding.signInWith(title)}>

View file

@ -1744,7 +1744,6 @@ export const en: Translations = {
flowSubtitles: {
pkce: 'Opens your browser to sign in, then continues here',
device_code: 'Opens a verification page in your browser — Hermes connects automatically',
loopback: 'Opens your browser to sign in — Hermes connects automatically',
external: 'Sign in once in your terminal, then come back to chat'
},
startingSignIn: provider => `Starting sign-in for ${provider}...`,

View file

@ -1852,7 +1852,6 @@ export const ja = defineLocale({
flowSubtitles: {
pkce: 'ブラウザーを開いてサインインし、ここに戻ります',
device_code: 'ブラウザーで確認ページを開きます — Hermes が自動接続します',
loopback: 'サインインのためブラウザーを開きます — Hermes が自動接続します',
external: 'ターミナルで一度サインインして、チャットに戻ります'
},
startingSignIn: provider => `${provider} のサインインを開始中...`,

View file

@ -1793,7 +1793,6 @@ export const zhHant = defineLocale({
flowSubtitles: {
pkce: '開啟瀏覽器登入,然後回到這裡繼續',
device_code: '在瀏覽器中開啟驗證頁面 — Hermes 會自動連線',
loopback: '開啟瀏覽器登入 — Hermes 會自動連線',
external: '先在終端機登入一次,然後回來繼續聊天'
},
startingSignIn: provider => `正在為 ${provider} 啟動登入...`,

View file

@ -1917,7 +1917,6 @@ export const zh: Translations = {
flowSubtitles: {
pkce: '打开浏览器登录,然后回到这里继续',
device_code: '在浏览器中打开验证页面 — Hermes 会自动连接',
loopback: '打开浏览器登录 — Hermes 会自动连接',
external: '先在终端登录一次,然后回来继续对话'
},
startingSignIn: provider => `正在为 ${provider} 启动登录...`,

View file

@ -18,7 +18,6 @@ import type { ModelOptionProvider, OAuthProvider, OAuthStartResponse } from '@/t
type PkceStart = Extract<OAuthStartResponse, { flow: 'pkce' }>
type DeviceStart = Extract<OAuthStartResponse, { flow: 'device_code' }>
type LoopbackStart = Extract<OAuthStartResponse, { flow: 'loopback' }>
export type OnboardingMode = 'apikey' | 'oauth'
@ -27,10 +26,6 @@ export type OnboardingFlow =
| { provider: OAuthProvider; status: 'starting' }
| { code: string; provider: OAuthProvider; start: PkceStart; status: 'awaiting_user' }
| { copied: boolean; provider: OAuthProvider; start: DeviceStart; status: 'polling' }
// Loopback PKCE (xAI Grok): browser opens, the local backend's 127.0.0.1
// listener catches the redirect, and we poll until the worker finishes.
// No code to paste and no user_code to show — just a waiting state.
| { provider: OAuthProvider; start: LoopbackStart; status: 'awaiting_browser' }
| { provider: OAuthProvider; start: OAuthStartResponse; status: 'submitting' }
| { copied: boolean; provider: OAuthProvider; status: 'external_pending' }
| { provider: OAuthProvider; status: 'success' }
@ -593,15 +588,6 @@ export async function startProviderOAuth(provider: OAuthProvider, ctx: Onboardin
return
}
if (start.flow === 'loopback') {
// No code to paste: the redirect lands on the backend's loopback
// listener. Just wait and poll the session until the worker finishes.
setFlow({ status: 'awaiting_browser', provider, start })
pollTimer = window.setInterval(() => void pollSession(provider, start, ctx), POLL_MS)
return
}
setFlow({ status: 'polling', provider, start, copied: false })
pollTimer = window.setInterval(() => void pollSession(provider, start, ctx), POLL_MS)
} catch (error) {
@ -609,10 +595,8 @@ export async function startProviderOAuth(provider: OAuthProvider, ctx: Onboardin
}
}
// Poll a session-backed flow (device_code or loopback) until it resolves.
// Both shapes only need the session_id to poll; the start is threaded
// through to the error flow so the user can retry from the same context.
async function pollSession(provider: OAuthProvider, start: DeviceStart | LoopbackStart, ctx: OnboardingContext) {
// Poll a session-backed device-code flow until it resolves.
async function pollSession(provider: OAuthProvider, start: DeviceStart, ctx: OnboardingContext) {
try {
const { error_message, status } = await pollOAuthSession(provider.id, start.session_id)

View file

@ -53,7 +53,7 @@ export interface OAuthProvider {
disconnect_hint?: null | string
disconnectable?: boolean
docs_url: string
flow: 'device_code' | 'external' | 'loopback' | 'pkce'
flow: 'device_code' | 'external' | 'pkce'
id: string
name: string
status: OAuthProviderStatus
@ -78,12 +78,6 @@ export type OAuthStartResponse =
user_code: string
verification_url: string
}
| {
auth_url: string
expires_in: number
flow: 'loopback'
session_id: string
}
export interface OAuthSubmitResponse {
message?: string

File diff suppressed because it is too large Load diff

View file

@ -345,19 +345,19 @@ def auth_add_command(args) -> None:
return
if provider == "xai-oauth":
creds = auth_mod._xai_oauth_loopback_login(
creds = auth_mod._xai_oauth_device_code_login(
timeout_seconds=getattr(args, "timeout", None) or 20.0,
open_browser=not getattr(args, "no_browser", False),
manual_paste=bool(getattr(args, "manual_paste", False)),
)
auth_mod._save_xai_oauth_tokens(
creds["tokens"],
discovery=creds.get("discovery"),
redirect_uri=creds.get("redirect_uri", ""),
last_refresh=creds.get("last_refresh"),
auth_mode="oauth_device_code",
)
pool = load_pool(provider)
entry = next((e for e in pool.entries() if getattr(e, "source", "") == "loopback_pkce"), None)
entry = next((e for e in pool.entries() if getattr(e, "source", "") == "device_code"), None)
shown_label = entry.label if entry is not None else label_from_token(
creds["tokens"]["access_token"], _oauth_default_label(provider, 1)
)

View file

@ -567,12 +567,7 @@ def _model_flow_xai_oauth(_config, current_model="", *, args=None):
print("Starting a fresh xAI OAuth login...")
print()
try:
# Forward CLI flags from ``hermes model --manual-paste``
# / ``--no-browser`` / ``--timeout`` into the loopback
# login. Without this, browser-only remotes (#26923)
# can't reach the manual-paste path via ``hermes model``.
mock_args = argparse.Namespace(
manual_paste=bool(getattr(args, "manual_paste", False)),
no_browser=bool(getattr(args, "no_browser", False)),
timeout=getattr(args, "timeout", None),
)
@ -594,7 +589,6 @@ def _model_flow_xai_oauth(_config, current_model="", *, args=None):
print()
try:
mock_args = argparse.Namespace(
manual_paste=bool(getattr(args, "manual_paste", False)),
no_browser=bool(getattr(args, "no_browser", False)),
timeout=getattr(args, "timeout", None),
)

View file

@ -881,7 +881,7 @@ def _xai_oauth_logged_in_for_setup() -> bool:
def _run_xai_oauth_login_from_setup() -> bool:
"""Run the xAI Grok OAuth loopback login from inside the setup wizard.
"""Run the xAI Grok OAuth device-code login from inside the setup wizard.
Returns True on success, False on any failure (the caller falls back
to whatever the user picked next, e.g. Edge TTS).
@ -892,7 +892,7 @@ def _run_xai_oauth_login_from_setup() -> bool:
_is_remote_session,
_save_xai_oauth_tokens,
_update_config_for_provider,
_xai_oauth_loopback_login,
_xai_oauth_device_code_login,
)
except Exception as exc:
print_warning(f"xAI Grok OAuth helpers unavailable: {exc}")
@ -902,12 +902,13 @@ def _run_xai_oauth_login_from_setup() -> bool:
print()
print_info("Signing in to xAI Grok OAuth (SuperGrok / Premium+)...")
try:
creds = _xai_oauth_loopback_login(open_browser=open_browser)
creds = _xai_oauth_device_code_login(open_browser=open_browser)
_save_xai_oauth_tokens(
creds["tokens"],
discovery=creds.get("discovery"),
redirect_uri=creds.get("redirect_uri", ""),
last_refresh=creds.get("last_refresh"),
auth_mode="oauth_device_code",
)
_update_config_for_provider(
"xai-oauth", creds.get("base_url", DEFAULT_XAI_OAUTH_BASE_URL)

View file

@ -40,17 +40,6 @@ def build_auth_parser(subparsers, *, cmd_auth: Callable) -> None:
action="store_true",
help="Do not auto-open a browser for OAuth login",
)
auth_add.add_argument(
"--manual-paste",
action="store_true",
help=(
"Skip the loopback callback listener and paste the failed "
"callback URL from your browser instead. Use this on "
"browser-only remotes (GCP Cloud Shell, GitHub Codespaces, "
"EC2 Instance Connect, ...) where 127.0.0.1 on the remote "
"isn't reachable from your laptop. See #26923."
),
)
auth_add.add_argument(
"--timeout", type=float, help="OAuth/network timeout in seconds"
)

View file

@ -45,16 +45,6 @@ def build_model_parser(subparsers, *, cmd_model: Callable) -> None:
action="store_true",
help="Do not attempt to open the browser automatically during Nous login",
)
model_parser.add_argument(
"--manual-paste",
action="store_true",
help=(
"For loopback OAuth providers (xai-oauth, ...): skip the local "
"callback listener and paste the failed callback URL from your "
"browser instead. Use on browser-only remotes (Cloud Shell, "
"Codespaces, EC2 Instance Connect, ...). See #26923."
),
)
model_parser.add_argument(
"--timeout",
type=float,

View file

@ -6427,7 +6427,7 @@ def _copilot_acp_status() -> Dict[str, Any]:
# ``flow`` describes the OAuth shape so the modal can pick the right UI:
# ``pkce`` = open URL + paste callback code, ``device_code`` = show code +
# verification URL + poll, ``external`` = read-only (delegated to a third-party
# CLI like Claude Code or Qwen), ``loopback`` = 127.0.0.1 callback listener.
# CLI like Claude Code or Qwen).
_OAUTH_PROVIDER_CATALOG: tuple[Dict[str, Any], ...] = (
{
"id": "nous",
@ -6469,10 +6469,10 @@ _OAUTH_PROVIDER_CATALOG: tuple[Dict[str, Any], ...] = (
{
"id": "xai-oauth",
"name": "xAI Grok OAuth (SuperGrok / Premium+)",
# Loopback PKCE: the desktop's local backend binds a 127.0.0.1
# callback server, the client opens the browser, and the redirect
# lands back on the loopback listener — no code to copy/paste.
"flow": "loopback",
# Device code is the default because it works in remote shells,
# containers, and desktop installs without requiring a reachable
# 127.0.0.1 callback.
"flow": "device_code",
"cli_command": "hermes auth add xai-oauth",
"docs_url": "https://hermes-agent.nousresearch.com/docs/guides/xai-grok-oauth",
"status_fn": None, # dispatched via auth.get_xai_oauth_auth_status
@ -6696,7 +6696,7 @@ async def list_oauth_providers(profile: Optional[str] = None):
Response shape (per provider):
id stable identifier (used in DELETE path)
name human label
flow "pkce" | "device_code" | "external" | "loopback"
flow "pkce" | "device_code" | "external"
cli_command fallback CLI command for users to run manually
disconnect_command shell command that clears an external provider's
creds (run in the embedded terminal), else null
@ -6831,19 +6831,6 @@ async def disconnect_oauth_provider(
# 4. On "approved" the background thread has already saved creds; UI
# refreshes the providers list.
#
# Loopback PKCE (xAI Grok):
# 1. POST /api/providers/oauth/xai-oauth/start
# → server binds a 127.0.0.1 callback listener, builds the xAI
# authorize URL, spawns a background worker waiting on the redirect
# → returns { session_id, flow: "loopback", auth_url, expires_in }
# 2. UI opens auth_url in the browser. There is NO user_code/code to
# paste — the redirect lands back on the loopback listener.
# 3. UI polls GET /api/providers/oauth/{provider}/poll/{session_id}
# (same endpoint as device_code) until status != "pending".
# 4. The worker exchanges the code, persists creds, sets "approved".
# DELETE /sessions/{id} cancels: the worker bails before persisting
# and the callback server is shut down to free the port immediately.
#
# Sessions are kept in-memory only (single-process FastAPI) and time out
# after 15 minutes. A periodic cleanup runs on each /start call to GC
# expired sessions so the dict doesn't grow without bound.
@ -7103,7 +7090,7 @@ async def _start_device_code_flow(
provider_id: str,
profile: Optional[str] = None,
) -> Dict[str, Any]:
"""Initiate a device-code flow (Nous, OpenAI Codex, or MiniMax).
"""Initiate a device-code flow (Nous, OpenAI Codex, MiniMax, or xAI).
Calls the provider's device-auth endpoint via the existing CLI helpers,
then spawns a background poller. Returns the user-facing display fields
@ -7272,222 +7259,43 @@ async def _start_device_code_flow(
"poll_interval": max(2, (sess["interval_ms"] or 2000) // 1000),
}
raise HTTPException(status_code=400, detail=f"Provider {provider_id} does not support device-code flow")
if provider_id == "xai-oauth":
from hermes_cli.auth import _xai_oauth_request_device_code
import httpx
def _do_xai_device_request():
with httpx.Client(
timeout=httpx.Timeout(20.0),
headers={"Accept": "application/json"},
) as client:
return _xai_oauth_request_device_code(client)
# xAI Grok OAuth uses a loopback-redirect PKCE flow (RFC 8252). Unlike the
# device-code providers there is no user_code to display: the local backend
# binds a 127.0.0.1 callback server, the client opens the authorize URL in
# the browser, and the redirect lands back on the loopback listener. The
# background worker waits for that callback, exchanges the code, and persists
# the tokens exactly like `hermes auth add xai-oauth`.
_XAI_LOOPBACK_TIMEOUT_SECONDS = 300.0
def _start_xai_loopback_flow(profile: Optional[str] = None) -> Dict[str, Any]:
"""Begin the xAI loopback PKCE flow.
Binds the local callback server, builds the authorize URL, and spawns a
background worker that waits for the redirect and finishes the exchange.
Returns the authorize URL for the client to open in the browser.
"""
from hermes_cli import auth as hauth
discovery = hauth._xai_oauth_discovery()
server, thread, callback_result, redirect_uri = hauth._xai_start_callback_server()
try:
hauth._xai_validate_loopback_redirect_uri(redirect_uri)
verifier = hauth._oauth_pkce_code_verifier()
challenge = hauth._oauth_pkce_code_challenge(verifier)
state = secrets.token_hex(16)
nonce = secrets.token_hex(16)
authorize_url = hauth._xai_oauth_build_authorize_url(
authorization_endpoint=discovery["authorization_endpoint"],
redirect_uri=redirect_uri,
code_challenge=challenge,
state=state,
nonce=nonce,
device_data = await asyncio.get_running_loop().run_in_executor(
None, _do_xai_device_request
)
except Exception:
# Binding succeeded but URL construction failed — release the socket
# and join the serving thread so we don't leak a listener (or a
# lingering daemon thread) on the loopback port.
try:
server.shutdown()
server.server_close()
except Exception:
pass
try:
thread.join(timeout=1.0)
except Exception:
pass
raise
sid, sess = _new_oauth_session("xai-oauth", "loopback", profile=profile)
sess["server"] = server
sess["thread"] = thread
sess["callback_result"] = callback_result
sess["redirect_uri"] = redirect_uri
sess["verifier"] = verifier
sess["challenge"] = challenge
sess["state"] = state
sess["token_endpoint"] = discovery["token_endpoint"]
sess["discovery"] = discovery
sess["expires_at"] = time.time() + _XAI_LOOPBACK_TIMEOUT_SECONDS
threading.Thread(
target=_xai_loopback_worker, args=(sid,), daemon=True,
name=f"oauth-xai-{sid[:6]}",
).start()
return {
"session_id": sid,
"flow": "loopback",
"auth_url": authorize_url,
"expires_in": int(_XAI_LOOPBACK_TIMEOUT_SECONDS),
}
def _xai_loopback_worker(session_id: str) -> None:
"""Wait for the xAI loopback callback, exchange the code, persist tokens."""
from datetime import datetime, timezone
from hermes_cli import auth as hauth
with _oauth_sessions_lock:
sess = _oauth_sessions.get(session_id)
if not sess:
return
def _fail(message: str) -> None:
with _oauth_sessions_lock:
s = _oauth_sessions.get(session_id)
if s is not None:
s["status"] = "error"
s["error_message"] = message
def _cancelled() -> bool:
# The session is removed from the registry when the user cancels
# (DELETE /sessions/{id}). If that happened while we were blocked on
# the callback or token exchange, abort instead of persisting tokens
# the user no longer wants.
with _oauth_sessions_lock:
return session_id not in _oauth_sessions
try:
callback = hauth._xai_wait_for_callback(
sess["server"],
sess["thread"],
sess["callback_result"],
timeout_seconds=_XAI_LOOPBACK_TIMEOUT_SECONDS,
)
except Exception as exc:
_fail(f"xAI authorization timed out: {exc}")
return
if _cancelled():
return
if callback.get("error"):
detail = callback.get("error_description") or callback["error"]
_fail(f"xAI authorization failed: {detail}")
return
if callback.get("state") != sess["state"]:
_fail("xAI authorization failed: state mismatch.")
return
code = str(callback.get("code") or "").strip()
if not code:
_fail("xAI authorization failed: missing authorization code.")
return
try:
payload = hauth._xai_oauth_exchange_code_for_tokens(
token_endpoint=sess["token_endpoint"],
code=code,
redirect_uri=sess["redirect_uri"],
code_verifier=sess["verifier"],
code_challenge=sess["challenge"],
)
access_token = str(payload.get("access_token", "") or "").strip()
refresh_token = str(payload.get("refresh_token", "") or "").strip()
if not access_token or not refresh_token:
_fail("xAI token exchange did not return the expected tokens.")
return
base_url = hauth._xai_validate_inference_base_url(
os.getenv("HERMES_XAI_BASE_URL", "").strip().rstrip("/")
or os.getenv("XAI_BASE_URL", "").strip().rstrip("/"),
fallback=hauth.DEFAULT_XAI_OAUTH_BASE_URL,
)
last_refresh = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
tokens = {
"access_token": access_token,
"refresh_token": refresh_token,
"id_token": str(payload.get("id_token", "") or "").strip(),
"expires_in": payload.get("expires_in"),
"token_type": str(payload.get("token_type") or "Bearer").strip() or "Bearer",
sid, sess = _new_oauth_session("xai-oauth", "device_code", profile=profile)
sess["device_code"] = str(device_data["device_code"])
sess["interval"] = int(device_data["interval"])
sess["expires_at"] = time.time() + int(device_data["expires_in"])
threading.Thread(
target=_xai_device_poller,
args=(sid,),
daemon=True,
name=f"oauth-poll-{sid[:6]}",
).start()
return {
"session_id": sid,
"flow": "device_code",
"user_code": str(device_data["user_code"]),
"verification_url": str(
device_data.get("verification_uri_complete")
or device_data["verification_uri"]
),
"expires_in": int(device_data["expires_in"]),
"poll_interval": int(device_data["interval"]),
}
if _cancelled():
return
with _profile_scope(_oauth_session_profile(session_id)):
hauth._save_xai_oauth_tokens(
tokens,
discovery=sess.get("discovery"),
redirect_uri=sess["redirect_uri"],
last_refresh=last_refresh,
)
_add_xai_oauth_pool_entry(access_token, refresh_token, base_url, last_refresh)
except Exception as exc:
_fail(f"xAI token exchange failed: {exc}")
return
with _oauth_sessions_lock:
s = _oauth_sessions.get(session_id)
if s is not None:
s["status"] = "approved"
_log.info("oauth/loopback: xai-oauth login completed (session=%s)", session_id)
def _add_xai_oauth_pool_entry(
access_token: str, refresh_token: str, base_url: str, last_refresh: str
) -> None:
"""Mirror `hermes auth add xai-oauth`'s credential-pool insert.
Best-effort: the auth-store write in _save_xai_oauth_tokens is the source
of truth for runtime resolution; the pool entry only matters for the
rotation strategy.
"""
try:
import uuid
from agent.credential_pool import (
PooledCredential,
load_pool,
AUTH_TYPE_OAUTH,
SOURCE_MANUAL,
)
pool = load_pool("xai-oauth")
existing = [
e for e in pool.entries()
if getattr(e, "source", "").startswith(f"{SOURCE_MANUAL}:dashboard_xai_pkce")
]
for e in existing:
try:
pool.remove_entry(getattr(e, "id", ""))
except Exception:
pass
entry = PooledCredential(
provider="xai-oauth",
id=uuid.uuid4().hex[:6],
label="dashboard PKCE",
auth_type=AUTH_TYPE_OAUTH,
priority=0,
source=f"{SOURCE_MANUAL}:dashboard_xai_pkce",
access_token=access_token,
refresh_token=refresh_token,
base_url=base_url,
last_refresh=last_refresh,
)
pool.add_entry(entry)
except Exception as e:
_log.warning("xai-oauth pool add (dashboard) failed: %s", e)
raise HTTPException(status_code=400, detail=f"Provider {provider_id} does not support device-code flow")
def _nous_poller(session_id: str) -> None:
@ -7638,6 +7446,70 @@ def _minimax_poller(session_id: str) -> None:
sess["error_message"] = str(e)
def _xai_device_poller(session_id: str) -> None:
"""Background poller for xAI's OAuth device-code flow."""
import httpx
from hermes_cli.auth import (
_save_xai_oauth_tokens,
_xai_oauth_discovery,
_xai_oauth_poll_device_token,
unsuppress_credential_source,
)
with _oauth_sessions_lock:
sess = _oauth_sessions.get(session_id)
if not sess:
return
device_code = sess["device_code"]
interval = int(sess["interval"])
expires_in = max(60, int(sess["expires_at"] - time.time()))
try:
discovery = _xai_oauth_discovery(20.0)
with httpx.Client(
timeout=httpx.Timeout(20.0),
headers={"Accept": "application/json"},
) as client:
token_data = _xai_oauth_poll_device_token(
client,
token_endpoint=discovery["token_endpoint"],
device_code=device_code,
expires_in=expires_in,
poll_interval=interval,
)
tokens = {
"access_token": str(token_data.get("access_token", "") or "").strip(),
"refresh_token": str(token_data.get("refresh_token", "") or "").strip(),
"id_token": str(token_data.get("id_token", "") or "").strip(),
"expires_in": token_data.get("expires_in"),
"token_type": str(token_data.get("token_type") or "Bearer").strip() or "Bearer",
}
with _profile_scope(_oauth_session_profile(session_id)):
_save_xai_oauth_tokens(
tokens,
discovery=discovery,
last_refresh=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
auth_mode="oauth_device_code",
)
# The singleton write above is the single source of truth: the
# credential-pool load seeds it as the canonical ``device_code``
# entry. Do NOT also insert a parallel ``manual:dashboard_*`` pool
# entry — that duplicates the single-use refresh token across two
# entries and triggers rotation churn / ``refresh_token_reused``.
# An interactive dashboard login is also an explicit re-enable
# signal, so clear any ``device_code`` suppression left by a
# prior ``hermes auth remove xai-oauth`` (mirrors auth_add_command
# and the ``hermes model`` re-login path in _login_xai_oauth).
unsuppress_credential_source("xai-oauth", "device_code")
with _oauth_sessions_lock:
sess["status"] = "approved"
_log.info("oauth/device: xai login completed (session=%s)", session_id)
except Exception as e:
_log.warning("xai device-code poll failed (session=%s): %s", session_id, e)
with _oauth_sessions_lock:
sess["status"] = "error"
sess["error_message"] = str(e)
def _codex_full_login_worker(session_id: str) -> None:
"""Run the complete OpenAI Codex device-code flow.
@ -7787,10 +7659,6 @@ async def start_oauth_login(
return _start_anthropic_pkce(profile=profile)
if catalog_entry["flow"] == "device_code":
return await _start_device_code_flow(provider_id, profile=profile)
if catalog_entry["flow"] == "loopback" and provider_id == "xai-oauth":
return await asyncio.get_running_loop().run_in_executor(
None, _start_xai_loopback_flow, profile,
)
except HTTPException:
raise
except Exception as e:
@ -7828,10 +7696,9 @@ async def poll_oauth_session(
):
"""Poll a session's status (no auth — read-only state).
Shared by the device-code flows (Nous, OpenAI Codex, MiniMax) and the
loopback flow (xAI Grok). Both surface progress through the same
background-worker-updated ``status`` field, so a single poll endpoint
serves them all.
Shared by the device-code flows (Nous, OpenAI Codex, MiniMax, xAI).
Each surfaces progress through the same background-worker-updated
``status`` field, so a single poll endpoint serves them all.
"""
with _oauth_sessions_lock:
sess = _oauth_sessions.get(session_id)
@ -7859,33 +7726,6 @@ async def cancel_oauth_session(
sess = _oauth_sessions.pop(session_id, None)
if sess is None:
return {"ok": False, "message": "session not found"}
# Loopback sessions own a bound 127.0.0.1 callback server. Without an
# explicit shutdown the worker would keep that port held until
# _xai_wait_for_callback times out (up to 5 min). Free it immediately so
# an orphaned listener can't block a subsequent sign-in attempt.
if sess.get("flow") == "loopback":
# The worker is blocked in _xai_wait_for_callback, which polls
# callback_result rather than the server state. Flag the result as
# cancelled so that loop returns on its next tick instead of spinning
# until the timeout — otherwise repeated cancel/retry piles up daemon
# threads. (_cancelled() in the worker then short-circuits before any
# persist.)
result = sess.get("callback_result")
if isinstance(result, dict):
result["error"] = result.get("error") or "cancelled"
server = sess.get("server")
thread = sess.get("thread")
try:
if server is not None:
server.shutdown()
server.server_close()
except Exception:
pass
try:
if thread is not None:
thread.join(timeout=1.0)
except Exception:
pass
return {"ok": True, "session_id": session_id}

View file

@ -4112,7 +4112,7 @@ class AIAgent:
#
# When an agent is using a non-singleton credential — e.g. a manual
# pool entry (``hermes auth add xai-oauth``) whose tokens belong to
# a different account than the loopback_pkce singleton, or an agent
# a different account than the device_code singleton, or an agent
# constructed with an explicit ``api_key=`` arg — force-refreshing
# the singleton here and adopting its tokens silently re-routes the
# rest of the conversation onto the singleton's account. The

View file

@ -2827,7 +2827,7 @@ def test_xai_oauth_terminal_refresh_clears_auth_json_and_removes_pool_entries(
pool = load_pool("xai-oauth")
selected = pool.select()
assert selected is not None
assert selected.source == "loopback_pkce"
assert selected.source == "device_code"
# Add a manual API-key entry that must survive the quarantine.
pool.add_entry(PooledCredential.from_dict("xai-oauth", {
@ -2868,7 +2868,7 @@ def test_xai_oauth_terminal_refresh_clears_auth_json_and_removes_pool_entries(
assert [entry["id"] for entry in auth_payload["credential_pool"]["xai-oauth"]] == ["manual-key"]
# A second try_refresh_current must not call refresh_xai_oauth_pure again
# (pool is now empty of loopback entries and current is None).
# (pool is now empty of device-code entries and current is None).
assert pool.try_refresh_current() is None
assert refresh_calls["count"] == 1

View file

@ -520,7 +520,7 @@ def test_auth_add_xai_oauth_sets_active_provider(tmp_path, monkeypatch):
_write_auth_store(tmp_path, {"version": 1, "providers": {}})
access_token = "xai-test-access-token"
monkeypatch.setattr(
"hermes_cli.auth._xai_oauth_loopback_login",
"hermes_cli.auth._xai_oauth_device_code_login",
lambda **kwargs: {
"tokens": {
"access_token": access_token,
@ -529,10 +529,10 @@ def test_auth_add_xai_oauth_sets_active_provider(tmp_path, monkeypatch):
"token_type": "Bearer",
},
"discovery": {"token_endpoint": "https://auth.x.ai/token"},
"redirect_uri": "http://127.0.0.1:7777/callback",
"redirect_uri": "",
"base_url": "https://api.x.ai/v1",
"last_refresh": "2026-06-02T10:00:00Z",
"source": "oauth-loopback",
"source": "oauth-device-code",
},
)
@ -545,7 +545,6 @@ def test_auth_add_xai_oauth_sets_active_provider(tmp_path, monkeypatch):
label = None
timeout = None
no_browser = False
manual_paste = False
auth_add_command(_Args())
@ -554,9 +553,10 @@ def test_auth_add_xai_oauth_sets_active_provider(tmp_path, monkeypatch):
assert payload["active_provider"] == "xai-oauth"
# providers singleton written by _save_xai_oauth_tokens
assert payload["providers"]["xai-oauth"]["tokens"]["access_token"] == access_token
assert payload["providers"]["xai-oauth"]["auth_mode"] == "oauth_device_code"
# pool seeded from singleton by _seed_from_singletons("xai-oauth")
entries = payload["credential_pool"]["xai-oauth"]
entry = next(item for item in entries if item["source"] == "loopback_pkce")
entry = next(item for item in entries if item["source"] == "device_code")
assert entry["refresh_token"] == "xai-refresh-token"

View file

@ -1,148 +0,0 @@
"""Unit tests for _print_loopback_ssh_hint() in hermes_cli/auth.py.
The helper exists to warn users that loopback OAuth flows (xAI Grok OAuth,
Spotify) don't work over SSH unless they set up an `ssh -L` port forward
between their laptop's browser and the remote host's loopback listener.
"""
from __future__ import annotations
import io
import contextlib
import socket
from hermes_cli import auth as auth_mod
def _cap(fn):
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
fn()
return buf.getvalue()
def test_loopback_ssh_hint_silent_when_not_remote(monkeypatch):
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: False)
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"http://127.0.0.1:56121/callback", docs_url=auth_mod.XAI_OAUTH_DOCS_URL
))
assert out == ""
def test_loopback_ssh_hint_prints_tunnel_command_on_ssh(monkeypatch):
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"http://127.0.0.1:56121/callback", docs_url=auth_mod.XAI_OAUTH_DOCS_URL
))
# Must include the exact ssh -L command with the port from the redirect URI
assert "ssh -N -L 56121:127.0.0.1:56121" in out
# Must include the provider-specific docs URL
assert auth_mod.XAI_OAUTH_DOCS_URL in out
# Must always include the cross-provider SSH guide
assert auth_mod.OAUTH_OVER_SSH_DOCS_URL in out
def test_loopback_ssh_hint_uses_actual_bound_port(monkeypatch):
"""When the preferred port is busy, _xai_start_callback_server falls back to
an OS-assigned port. The hint must echo whichever port actually got bound,
not the hardcoded constant."""
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"http://127.0.0.1:51234/callback", docs_url=auth_mod.XAI_OAUTH_DOCS_URL
))
assert "ssh -N -L 51234:127.0.0.1:51234" in out
assert "56121" not in out
def test_loopback_ssh_hint_silent_for_non_loopback_uri(monkeypatch):
"""Defense in depth: if a future caller passes a non-loopback redirect URI
by mistake, we don't tell the user to forward an external port."""
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"https://example.com/callback", docs_url=auth_mod.XAI_OAUTH_DOCS_URL
))
assert out == ""
def test_loopback_ssh_hint_silent_for_malformed_uri(monkeypatch):
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"not-a-uri", docs_url=auth_mod.XAI_OAUTH_DOCS_URL
))
assert out == ""
def test_loopback_ssh_hint_works_without_provider_docs_url(monkeypatch):
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"http://127.0.0.1:43827/spotify/callback"
))
assert "ssh -N -L 43827:127.0.0.1:43827" in out
# Generic SSH guide is always present even without a provider-specific URL
assert auth_mod.OAUTH_OVER_SSH_DOCS_URL in out
# Should not falsely show "Provider docs:" when no docs_url was passed
assert "Provider docs:" not in out
def test_loopback_ssh_hint_accepts_localhost_hostname(monkeypatch):
"""The constant is 127.0.0.1, but parsing tolerates `localhost` too in case
a future caller normalizes the URI differently."""
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"http://localhost:56121/callback"
))
assert "ssh -N -L 56121:127.0.0.1:56121" in out
def test_loopback_ssh_hint_includes_user_at_host(monkeypatch):
"""The SSH command should include a detected user@host so the user can
copy-paste it without manually substituting placeholders."""
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
monkeypatch.setattr(auth_mod, "_ssh_user_at_host", lambda: "alice@myserver.lan")
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"http://127.0.0.1:56121/callback"
))
assert "ssh -N -L 56121:127.0.0.1:56121 alice@myserver.lan" in out
def test_loopback_ssh_hint_has_visual_header(monkeypatch):
"""The hint should print a divider and header so it stands out in noisy output."""
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
out = _cap(lambda: auth_mod._print_loopback_ssh_hint(
"http://127.0.0.1:56121/callback"
))
assert "Remote session detected" in out
assert "---" in out # divider is present
class TestSshUserAtHost:
def test_resolves_user_and_hostname(self, monkeypatch):
monkeypatch.setenv("USER", "alice")
monkeypatch.delenv("LOGNAME", raising=False)
monkeypatch.setattr(socket, "gethostname", lambda: "myserver")
assert auth_mod._ssh_user_at_host() == "alice@myserver"
def test_falls_back_to_logname(self, monkeypatch):
monkeypatch.delenv("USER", raising=False)
monkeypatch.setenv("LOGNAME", "bob")
monkeypatch.setattr(socket, "gethostname", lambda: "host1")
assert auth_mod._ssh_user_at_host() == "bob@host1"
def test_placeholder_when_no_env_vars(self, monkeypatch):
monkeypatch.delenv("USER", raising=False)
monkeypatch.delenv("LOGNAME", raising=False)
monkeypatch.setattr(socket, "gethostname", lambda: "host1")
assert auth_mod._ssh_user_at_host() == "<user>@host1"
def test_placeholder_when_socket_raises(self, monkeypatch):
monkeypatch.setenv("USER", "charlie")
def _raise():
raise OSError("no network")
monkeypatch.setattr(socket, "gethostname", _raise)
assert auth_mod._ssh_user_at_host() == "charlie@<this-host>"
def test_placeholder_when_empty_hostname(self, monkeypatch):
monkeypatch.setenv("USER", "dave")
monkeypatch.setattr(socket, "gethostname", lambda: "")
assert auth_mod._ssh_user_at_host() == "dave@<this-host>"

View file

@ -1,684 +0,0 @@
"""Tests for the OAuth manual-paste fallback for browser-only remotes.
Regression coverage for [#26923](https://github.com/NousResearch/hermes-agent/issues/26923):
GCP Cloud Shell, GitHub Codespaces, AWS EC2 Instance Connect and
other browser-only remote consoles can't reach the
``http://127.0.0.1:56121/callback`` loopback listener bound on the
remote VM. The previous SSH-tunnel hint was useless without a real
SSH client, leaving the user with no path forward. This test file
locks in four things:
* ``_is_remote_session`` recognises the cloud-shell / Codespaces
envvars (so the existing hint at least fires).
* ``_parse_pasted_callback`` accepts every form a user might paste
(full URL, ``?code=...&state=...`` fragment, bare ``code=...``,
bare opaque value) and returns the same shape the loopback HTTP
handler does.
* ``_prompt_manual_callback_paste`` reads stdin and produces that
same shape.
* ``_xai_oauth_loopback_login(manual_paste=True)`` skips the HTTP
server entirely, validates ``state``, and goes straight to the
token exchange proving the paste path actually wires up.
"""
from __future__ import annotations
import builtins
import io
import contextlib
import pytest
from hermes_cli import auth as auth_mod
# ---------------------------------------------------------------------------
# _is_remote_session — broadened detection (#26923)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"envvar",
[
"SSH_CLIENT",
"SSH_TTY",
"CLOUD_SHELL",
"CODESPACES",
"CODESPACE_NAME",
"GITPOD_WORKSPACE_ID",
"REPL_ID",
"STACKBLITZ",
],
)
def test_is_remote_session_detects_known_remote_envvar(monkeypatch, envvar):
"""Each documented remote-console env var must trip the check.
The SSH ones preserve historical behaviour; the cloud-shell ones
are what closes #26923. Without these, the SSH hint never fires
and the user has no signal that ``--manual-paste`` exists.
"""
for name in (
"SSH_CLIENT",
"SSH_TTY",
"CLOUD_SHELL",
"CODESPACES",
"CODESPACE_NAME",
"GITPOD_WORKSPACE_ID",
"REPL_ID",
"STACKBLITZ",
):
monkeypatch.delenv(name, raising=False)
monkeypatch.setenv(envvar, "1")
assert auth_mod._is_remote_session() is True
def test_is_remote_session_false_when_no_remote_envvars(monkeypatch):
for name in (
"SSH_CLIENT",
"SSH_TTY",
"CLOUD_SHELL",
"CODESPACES",
"CODESPACE_NAME",
"GITPOD_WORKSPACE_ID",
"REPL_ID",
"STACKBLITZ",
):
monkeypatch.delenv(name, raising=False)
assert auth_mod._is_remote_session() is False
# ---------------------------------------------------------------------------
# _parse_pasted_callback — accept every plausible paste form
# ---------------------------------------------------------------------------
def test_parse_full_callback_url():
out = auth_mod._parse_pasted_callback(
"http://127.0.0.1:56121/callback?code=abc123&state=deadbeef"
)
assert out == {
"code": "abc123",
"state": "deadbeef",
"error": None,
"error_description": None,
}
def test_parse_callback_url_https_and_extra_params():
out = auth_mod._parse_pasted_callback(
"https://127.0.0.1:56121/callback?code=abc&state=xyz&scope=openid"
)
assert out["code"] == "abc"
assert out["state"] == "xyz"
def test_parse_bare_query_string_with_leading_question_mark():
out = auth_mod._parse_pasted_callback("?code=p1&state=s1")
assert out["code"] == "p1"
assert out["state"] == "s1"
def test_parse_bare_query_fragment_no_question_mark():
out = auth_mod._parse_pasted_callback("code=p2&state=s2")
assert out["code"] == "p2"
assert out["state"] == "s2"
def test_parse_bare_opaque_code_value():
"""Some users only copy the ``code`` value itself."""
out = auth_mod._parse_pasted_callback("ABCDEF-the-code-value")
assert out["code"] == "ABCDEF-the-code-value"
assert out["state"] is None
def test_parse_callback_with_error_field():
out = auth_mod._parse_pasted_callback(
"http://127.0.0.1:56121/callback?error=access_denied"
"&error_description=user+rejected"
)
assert out["code"] is None
assert out["error"] == "access_denied"
assert out["error_description"] == "user rejected"
def test_parse_empty_input_returns_all_none():
out = auth_mod._parse_pasted_callback("")
assert out == {
"code": None,
"state": None,
"error": None,
"error_description": None,
}
def test_parse_whitespace_only_returns_all_none():
out = auth_mod._parse_pasted_callback(" \n\t ")
assert out["code"] is None
def test_parse_malformed_url_does_not_crash():
out = auth_mod._parse_pasted_callback("http://[not a url")
# Malformed URLs return all-None rather than raising — the caller
# (state check) will reject the empty payload with a clear error.
assert out["code"] is None
# ---------------------------------------------------------------------------
# _prompt_manual_callback_paste — stdin handling
# ---------------------------------------------------------------------------
def test_prompt_reads_stdin_and_parses(monkeypatch):
monkeypatch.setattr(
builtins, "input",
lambda *_a, **_k: "http://127.0.0.1:56121/callback?code=abc&state=xyz",
)
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
out = auth_mod._prompt_manual_callback_paste(
"http://127.0.0.1:56121/callback"
)
rendered = buf.getvalue()
assert "Manual callback paste" in rendered
assert "127.0.0.1:56121" in rendered
assert out["code"] == "abc"
assert out["state"] == "xyz"
def test_prompt_eof_returns_all_none(monkeypatch):
def _raise_eof(*_a, **_k):
raise EOFError()
monkeypatch.setattr(builtins, "input", _raise_eof)
with contextlib.redirect_stdout(io.StringIO()):
out = auth_mod._prompt_manual_callback_paste(
"http://127.0.0.1:56121/callback"
)
assert out["code"] is None
def test_prompt_keyboard_interrupt_returns_all_none(monkeypatch):
def _raise_kbi(*_a, **_k):
raise KeyboardInterrupt()
monkeypatch.setattr(builtins, "input", _raise_kbi)
with contextlib.redirect_stdout(io.StringIO()):
out = auth_mod._prompt_manual_callback_paste(
"http://127.0.0.1:56121/callback"
)
assert out["code"] is None
# ---------------------------------------------------------------------------
# _xai_oauth_loopback_login(manual_paste=True) — full integration
# ---------------------------------------------------------------------------
class _StubTokenResponse:
status_code = 200
def __init__(self, payload):
self._payload = payload
self.text = ""
def json(self):
return self._payload
def test_xai_loopback_login_manual_paste_skips_http_server(monkeypatch):
"""``manual_paste=True`` must NOT bind a loopback HTTP server.
Direct end-to-end regression for #26923: the whole point is that
the listener is unreachable on browser-only remotes, so the paste
path must avoid it entirely. We assert this by replacing
``_xai_start_callback_server`` with a function that fails if
invoked, then driving the full happy path with a stubbed prompt
+ stubbed token endpoint.
"""
monkeypatch.setattr(
auth_mod, "_xai_oauth_discovery",
lambda *_a, **_k: {
"authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
"token_endpoint": "https://auth.x.ai/oauth2/token",
},
)
def _server_must_not_be_called(*_a, **_k):
raise AssertionError(
"manual_paste=True must skip the loopback HTTP server "
"(regression for #26923)"
)
monkeypatch.setattr(
auth_mod, "_xai_start_callback_server", _server_must_not_be_called
)
captured_state: dict = {}
def _fake_prompt(_redirect_uri):
# Hermes generates state internally; we won't know it ahead of
# time, so capture the state Hermes baked into the authorize
# URL via a sneak peek on ``_xai_oauth_build_authorize_url``.
return {
"code": "fake-auth-code",
"state": captured_state["value"],
"error": None,
"error_description": None,
}
monkeypatch.setattr(
auth_mod, "_prompt_manual_callback_paste", _fake_prompt
)
original_build = auth_mod._xai_oauth_build_authorize_url
def _capture_state(**kwargs):
captured_state["value"] = kwargs["state"]
return original_build(**kwargs)
monkeypatch.setattr(
auth_mod, "_xai_oauth_build_authorize_url", _capture_state
)
def _fake_token_post(*_a, **_k):
return _StubTokenResponse(
{
"access_token": "at",
"refresh_token": "rt",
"id_token": "",
"expires_in": 3600,
"token_type": "Bearer",
}
)
monkeypatch.setattr(auth_mod.httpx, "post", _fake_token_post)
with contextlib.redirect_stdout(io.StringIO()):
creds = auth_mod._xai_oauth_loopback_login(manual_paste=True)
assert creds["tokens"]["access_token"] == "at"
assert creds["tokens"]["refresh_token"] == "rt"
assert "127.0.0.1:56121" in creds["redirect_uri"]
def test_xai_loopback_login_manual_paste_state_mismatch_raises(monkeypatch):
"""A pasted callback with the wrong state must still be rejected.
The HTTP-server path uses the same state check; manual-paste
must not be a CSRF bypass.
"""
monkeypatch.setattr(
auth_mod, "_xai_oauth_discovery",
lambda *_a, **_k: {
"authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
"token_endpoint": "https://auth.x.ai/oauth2/token",
},
)
monkeypatch.setattr(
auth_mod, "_prompt_manual_callback_paste",
lambda _ru: {
"code": "fake",
"state": "WRONG-STATE",
"error": None,
"error_description": None,
},
)
with contextlib.redirect_stdout(io.StringIO()):
with pytest.raises(auth_mod.AuthError) as exc:
auth_mod._xai_oauth_loopback_login(manual_paste=True)
assert exc.value.code == "xai_state_mismatch"
def test_xai_loopback_login_manual_paste_bare_code_succeeds(monkeypatch):
"""Bare-code paste (state=None) must complete login under manual_paste.
xAI's consent page renders the authorization code in-page rather than
redirecting through 127.0.0.1, so on remote/headless setups the only
value the user can obtain is the opaque code with no ``state=``
parameter. ``_parse_pasted_callback`` correctly returns
``state=None`` for that input. The login flow must accept this case
(PKCE still protects the exchange); historically it raised
``xai_state_mismatch``. Regression for the bare-code branch of #26923.
"""
monkeypatch.setattr(
auth_mod, "_xai_oauth_discovery",
lambda *_a, **_k: {
"authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
"token_endpoint": "https://auth.x.ai/oauth2/token",
},
)
monkeypatch.setattr(
auth_mod, "_prompt_manual_callback_paste",
lambda _ru: {
"code": "bare-opaque-code",
"state": None,
"error": None,
"error_description": None,
},
)
def _fake_token_post(*_a, **_k):
return _StubTokenResponse(
{
"access_token": "at",
"refresh_token": "rt",
"id_token": "",
"expires_in": 3600,
"token_type": "Bearer",
}
)
monkeypatch.setattr(auth_mod.httpx, "post", _fake_token_post)
with contextlib.redirect_stdout(io.StringIO()):
creds = auth_mod._xai_oauth_loopback_login(manual_paste=True)
assert creds["tokens"]["access_token"] == "at"
assert creds["tokens"]["refresh_token"] == "rt"
def test_xai_loopback_login_loopback_path_rejects_missing_state(monkeypatch):
"""Loopback (manual_paste=False) must NOT accept ``state=None``.
The bare-code relaxation only applies to the manual-paste path,
where the user demonstrably has no way to supply ``state``. The
HTTP-server path always sees ``state`` populated from the real
callback query string, so missing state there means something is
wrong (a malformed callback, an attacker-supplied request) and
must still raise ``xai_state_mismatch``.
"""
monkeypatch.setattr(
auth_mod, "_xai_oauth_discovery",
lambda *_a, **_k: {
"authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
"token_endpoint": "https://auth.x.ai/oauth2/token",
},
)
class _StubServer:
def shutdown(self):
return None
def server_close(self):
return None
monkeypatch.setattr(
auth_mod, "_xai_start_callback_server",
lambda *_a, **_k: (
_StubServer(),
None,
{"code": "fake", "state": None, "error": None,
"error_description": None},
"http://127.0.0.1:56121/callback",
),
)
monkeypatch.setattr(
auth_mod, "_xai_wait_for_callback",
lambda *_a, **_k: {
"code": "fake",
"state": None,
"error": None,
"error_description": None,
},
)
monkeypatch.setattr(auth_mod, "_xai_validate_loopback_redirect_uri", lambda _u: None)
monkeypatch.setattr(auth_mod, "_print_loopback_ssh_hint", lambda *_a, **_k: None)
with contextlib.redirect_stdout(io.StringIO()):
with pytest.raises(auth_mod.AuthError) as exc:
auth_mod._xai_oauth_loopback_login(manual_paste=False, open_browser=False)
assert exc.value.code == "xai_state_mismatch"
def test_xai_loopback_login_manual_paste_missing_code_raises(monkeypatch):
"""Empty paste must surface as ``xai_code_missing``, not crash."""
monkeypatch.setattr(
auth_mod, "_xai_oauth_discovery",
lambda *_a, **_k: {
"authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
"token_endpoint": "https://auth.x.ai/oauth2/token",
},
)
captured: dict = {"state": None}
original_build = auth_mod._xai_oauth_build_authorize_url
def _capture(**kw):
captured["state"] = kw["state"]
return original_build(**kw)
monkeypatch.setattr(auth_mod, "_xai_oauth_build_authorize_url", _capture)
monkeypatch.setattr(
auth_mod, "_prompt_manual_callback_paste",
lambda _ru: {
"code": None,
"state": captured["state"],
"error": None,
"error_description": None,
},
)
with contextlib.redirect_stdout(io.StringIO()):
with pytest.raises(auth_mod.AuthError) as exc:
auth_mod._xai_oauth_loopback_login(manual_paste=True)
assert exc.value.code == "xai_code_missing"
def test_xai_loopback_login_timeout_falls_back_to_manual_paste(monkeypatch):
"""Loopback timeout should accept a bare Grok Build code paste."""
monkeypatch.setattr(
auth_mod, "_xai_oauth_discovery",
lambda *_a, **_k: {
"authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
"token_endpoint": "https://auth.x.ai/oauth2/token",
},
)
class _StubServer:
def shutdown(self):
return None
def server_close(self):
return None
class _StubThread:
def join(self, timeout=None):
return None
monkeypatch.setattr(
auth_mod,
"_xai_start_callback_server",
lambda: (
_StubServer(),
_StubThread(),
{
"code": None,
"state": None,
"error": None,
"error_description": None,
},
"http://127.0.0.1:56121/callback",
),
)
captured: dict = {"state": None, "prompt_calls": 0}
original_build = auth_mod._xai_oauth_build_authorize_url
def _capture(**kwargs):
captured["state"] = kwargs["state"]
return original_build(**kwargs)
monkeypatch.setattr(auth_mod, "_xai_oauth_build_authorize_url", _capture)
def _raise_timeout(*_a, **_k):
raise auth_mod.AuthError(
"xAI authorization timed out waiting for the local callback.",
provider="xai-oauth",
code="xai_callback_timeout",
)
monkeypatch.setattr(auth_mod, "_xai_wait_for_callback", _raise_timeout)
def _fake_prompt(_redirect_uri):
captured["prompt_calls"] += 1
return {
"code": "manual-auth-code",
"state": None,
"error": None,
"error_description": None,
}
monkeypatch.setattr(auth_mod, "_prompt_manual_callback_paste", _fake_prompt)
monkeypatch.setattr(
auth_mod.sys, "stdin", type("StubStdin", (), {"isatty": lambda self: True})()
)
monkeypatch.setattr(
auth_mod.httpx,
"post",
lambda *_a, **_k: _StubTokenResponse(
{
"access_token": "at-timeout",
"refresh_token": "rt-timeout",
"id_token": "",
"expires_in": 3600,
"token_type": "Bearer",
}
),
)
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
creds = auth_mod._xai_oauth_loopback_login(manual_paste=False)
rendered = buf.getvalue()
assert "xAI loopback callback timed out." in rendered
assert "--manual-paste" in rendered
assert captured["prompt_calls"] == 1
assert creds["tokens"]["access_token"] == "at-timeout"
assert creds["tokens"]["refresh_token"] == "rt-timeout"
def test_xai_wait_for_callback_accepts_ready_stdin_code(monkeypatch):
"""Users can paste the Grok Build code while Hermes is still waiting."""
class _StubServer:
shutdown_called = False
close_called = False
def shutdown(self):
self.shutdown_called = True
def server_close(self):
self.close_called = True
class _StubThread:
joined = False
def join(self, timeout=None):
self.joined = True
server = _StubServer()
thread = _StubThread()
monkeypatch.setattr(
auth_mod,
"_read_ready_stdin_line",
lambda: "ready-grok-build-code\n",
)
out = auth_mod._xai_wait_for_callback(
server,
thread,
{"code": None, "error": None},
timeout_seconds=5,
manual_paste_redirect_uri="http://127.0.0.1:56121/callback",
)
assert out["code"] == "ready-grok-build-code"
assert out["state"] is None
assert out["_manual_paste"] is True
assert server.shutdown_called is True
assert server.close_called is True
assert thread.joined is True
def test_xai_loopback_login_timeout_noninteractive_reraises(monkeypatch):
"""Non-interactive stdin must keep the original timeout error."""
monkeypatch.setattr(
auth_mod, "_xai_oauth_discovery",
lambda *_a, **_k: {
"authorization_endpoint": "https://auth.x.ai/oauth2/authorize",
"token_endpoint": "https://auth.x.ai/oauth2/token",
},
)
class _StubServer:
def shutdown(self):
return None
def server_close(self):
return None
class _StubThread:
def join(self, timeout=None):
return None
monkeypatch.setattr(
auth_mod,
"_xai_start_callback_server",
lambda: (
_StubServer(),
_StubThread(),
{
"code": None,
"state": None,
"error": None,
"error_description": None,
},
"http://127.0.0.1:56121/callback",
),
)
monkeypatch.setattr(
auth_mod,
"_xai_wait_for_callback",
lambda *_a, **_k: (_ for _ in ()).throw(
auth_mod.AuthError(
"xAI authorization timed out waiting for the local callback.",
provider="xai-oauth",
code="xai_callback_timeout",
)
),
)
monkeypatch.setattr(
auth_mod.sys, "stdin", type("StubStdin", (), {"isatty": lambda self: False})()
)
monkeypatch.setattr(
auth_mod,
"_prompt_manual_callback_paste",
lambda *_a, **_k: pytest.fail("manual-paste fallback should not run"),
)
with contextlib.redirect_stdout(io.StringIO()):
with pytest.raises(auth_mod.AuthError) as exc:
auth_mod._xai_oauth_loopback_login(manual_paste=False)
assert exc.value.code == "xai_callback_timeout"
# ---------------------------------------------------------------------------
# _print_loopback_ssh_hint — now also mentions --manual-paste
# ---------------------------------------------------------------------------
def test_ssh_hint_mentions_manual_paste_for_non_ssh_remotes(monkeypatch):
"""Users on Cloud Shell / Codespaces have no real SSH client; the
hint must point them at the new ``--manual-paste`` flag instead
of leaving them stuck on the ``ssh -L`` recipe."""
monkeypatch.setattr(auth_mod, "_is_remote_session", lambda: True)
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
auth_mod._print_loopback_ssh_hint(
"http://127.0.0.1:56121/callback",
docs_url=auth_mod.XAI_OAUTH_DOCS_URL,
)
rendered = buf.getvalue()
assert "--manual-paste" in rendered
assert "Cloud Shell" in rendered or "Codespaces" in rendered

View file

@ -2,9 +2,7 @@
import base64
import json
import socket
import time
import urllib.request
from pathlib import Path
import pytest
@ -14,17 +12,13 @@ from hermes_cli.auth import (
DEFAULT_XAI_OAUTH_BASE_URL,
PROVIDER_REGISTRY,
XAI_OAUTH_CLIENT_ID,
XAI_OAUTH_REDIRECT_HOST,
XAI_OAUTH_REDIRECT_PATH,
XAI_OAUTH_SCOPE,
_read_xai_oauth_tokens,
_save_xai_oauth_tokens,
_xai_access_token_is_expiring,
_xai_callback_cors_origin,
_xai_oauth_build_authorize_url,
_xai_start_callback_server,
_xai_oauth_poll_device_token,
_xai_oauth_request_device_code,
_xai_validate_inference_base_url,
_xai_validate_loopback_redirect_uri,
format_auth_error,
get_xai_oauth_auth_status,
refresh_xai_oauth_pure,
@ -44,6 +38,7 @@ def _setup_hermes_auth(
access_token: str = "access",
refresh_token: str = "refresh",
discovery: dict | None = None,
auth_mode: str = "oauth_pkce",
):
"""Write xAI OAuth tokens into the Hermes auth store at the given root."""
hermes_home.mkdir(parents=True, exist_ok=True)
@ -56,7 +51,7 @@ def _setup_hermes_auth(
"token_type": "Bearer",
},
"last_refresh": "2026-05-14T00:00:00Z",
"auth_mode": "oauth_pkce",
"auth_mode": auth_mode,
}
if discovery is not None:
state["discovery"] = discovery
@ -177,233 +172,84 @@ def test_xai_access_token_is_expiring_returns_false_for_jwt_without_exp():
# ---------------------------------------------------------------------------
# Loopback redirect URI validation
# Device-code flow
# ---------------------------------------------------------------------------
def test_xai_validate_loopback_redirect_uri_accepts_localhost_with_port():
host, port, path = _xai_validate_loopback_redirect_uri(
"http://127.0.0.1:56121/callback"
def test_xai_oauth_request_device_code_returns_display_fields():
response = _StubHTTPResponse(
200,
{
"device_code": "device-code",
"user_code": "ABCD-EFGH",
"verification_uri": "https://accounts.x.ai/oauth2/device",
"verification_uri_complete": "https://accounts.x.ai/oauth2/device?user_code=ABCD-EFGH",
"expires_in": 1800,
"interval": 5,
},
)
assert host == XAI_OAUTH_REDIRECT_HOST
assert port == 56121
assert path == XAI_OAUTH_REDIRECT_PATH
client = _StubHTTPClient(response)
payload = _xai_oauth_request_device_code(client)
assert payload["user_code"] == "ABCD-EFGH"
method, args, kwargs = client.last_call
assert method == "post"
assert args[0] == "https://auth.x.ai/oauth2/device/code"
assert kwargs["data"]["client_id"] == XAI_OAUTH_CLIENT_ID
assert kwargs["data"]["scope"] == XAI_OAUTH_SCOPE
def test_xai_validate_loopback_redirect_uri_rejects_https():
def test_xai_oauth_request_device_code_rejects_missing_fields():
client = _StubHTTPClient(_StubHTTPResponse(200, {"device_code": "d"}))
with pytest.raises(AuthError) as exc:
_xai_validate_loopback_redirect_uri("https://127.0.0.1:56121/callback")
assert exc.value.code == "xai_redirect_invalid"
_xai_oauth_request_device_code(client)
assert exc.value.code == "device_code_invalid"
def test_xai_validate_loopback_redirect_uri_rejects_non_loopback():
with pytest.raises(AuthError) as exc:
_xai_validate_loopback_redirect_uri("http://example.com:56121/callback")
assert exc.value.code == "xai_redirect_invalid"
def test_xai_oauth_poll_device_token_waits_until_authorized(monkeypatch):
class _SequenceClient:
def __init__(self):
self.calls = []
self.responses = [
_StubHTTPResponse(
400,
{
"error": "authorization_pending",
"error_description": "User has not yet authorized",
},
),
_StubHTTPResponse(
200,
{
"access_token": "xai-access",
"refresh_token": "xai-refresh",
"expires_in": 3600,
"token_type": "Bearer",
},
),
]
def post(self, *args, **kwargs):
self.calls.append((args, kwargs))
return self.responses.pop(0)
def test_xai_validate_loopback_redirect_uri_rejects_missing_port():
with pytest.raises(AuthError) as exc:
_xai_validate_loopback_redirect_uri("http://127.0.0.1/callback")
assert exc.value.code == "xai_redirect_invalid"
monkeypatch.setattr("hermes_cli.auth.time.sleep", lambda _: None)
client = _SequenceClient()
# ---------------------------------------------------------------------------
# Authorize URL construction
# ---------------------------------------------------------------------------
def _parse_authorize_url(url: str) -> dict:
from urllib.parse import urlparse, parse_qs
parsed = urlparse(url)
return {k: v[0] for k, v in parse_qs(parsed.query).items()}
def test_xai_oauth_authorize_url_includes_plan_generic():
"""Regression: accounts.x.ai requires `plan=generic` for loopback OAuth on
non-allowlisted clients. Must always be present on the authorize URL."""
url = _xai_oauth_build_authorize_url(
authorization_endpoint="https://auth.x.ai/oauth2/authorize",
redirect_uri="http://127.0.0.1:56121/callback",
code_challenge="challenge-xyz",
state="state-abc",
nonce="nonce-def",
payload = _xai_oauth_poll_device_token(
client,
token_endpoint="https://auth.x.ai/oauth2/token",
device_code="device-code",
expires_in=30,
poll_interval=1,
)
params = _parse_authorize_url(url)
assert params["plan"] == "generic"
def test_xai_oauth_authorize_url_includes_referrer_hermes_agent():
"""Attribution: xAI's OAuth server can identify Hermes-originated logins
via the referrer query param. Must always be present on the authorize URL."""
url = _xai_oauth_build_authorize_url(
authorization_endpoint="https://auth.x.ai/oauth2/authorize",
redirect_uri="http://127.0.0.1:56121/callback",
code_challenge="challenge-xyz",
state="state-abc",
nonce="nonce-def",
)
params = _parse_authorize_url(url)
assert params["referrer"] == "hermes-agent"
def test_xai_oauth_authorize_url_includes_pkce_and_oidc_params():
url = _xai_oauth_build_authorize_url(
authorization_endpoint="https://auth.x.ai/oauth2/authorize",
redirect_uri="http://127.0.0.1:56121/callback",
code_challenge="challenge-xyz",
state="state-abc",
nonce="nonce-def",
)
params = _parse_authorize_url(url)
assert params["response_type"] == "code"
assert params["client_id"] == XAI_OAUTH_CLIENT_ID
assert params["redirect_uri"] == "http://127.0.0.1:56121/callback"
assert params["scope"] == XAI_OAUTH_SCOPE
assert params["code_challenge"] == "challenge-xyz"
assert params["code_challenge_method"] == "S256"
assert params["state"] == "state-abc"
assert params["nonce"] == "nonce-def"
# ---------------------------------------------------------------------------
# CORS allowlist
# ---------------------------------------------------------------------------
def test_xai_callback_cors_origin_allowlist():
assert _xai_callback_cors_origin("https://accounts.x.ai") == "https://accounts.x.ai"
assert _xai_callback_cors_origin("https://auth.x.ai") == "https://auth.x.ai"
def test_xai_callback_cors_origin_rejects_unknown_origin():
assert _xai_callback_cors_origin("https://attacker.example.com") == ""
assert _xai_callback_cors_origin(None) == ""
assert _xai_callback_cors_origin("") == ""
def test_xai_callback_server_accepts_fallback_code_while_browser_connection_is_stuck():
"""Regression: Chrome/xAI can leave a loopback connection open after
showing the Grok Build fallback code. A single-threaded callback server then
blocks forever and cannot accept the manual fallback callback.
"""
server, thread, result, redirect_uri = _xai_start_callback_server(preferred_port=0)
stuck = socket.create_connection((XAI_OAUTH_REDIRECT_HOST, server.server_address[1]), timeout=2)
try:
stuck.sendall(b"GET /callback?code=stuck")
callback_url = f"{redirect_uri}?code=fallback-code&state=state-123"
with urllib.request.urlopen(callback_url, timeout=2) as response:
body = response.read().decode("utf-8")
assert response.status == 200
assert "xAI authorization received" in body
assert result["code"] == "fallback-code"
assert result["state"] == "state-123"
finally:
stuck.close()
server.shutdown()
server.server_close()
thread.join(timeout=1.0)
def test_xai_callback_server_latches_first_terminal_callback_result():
server, thread, result, redirect_uri = _xai_start_callback_server(preferred_port=0)
try:
with urllib.request.urlopen(f"{redirect_uri}?code=first-code&state=state-1", timeout=2) as response:
assert response.status == 200
with urllib.request.urlopen(
f"{redirect_uri}?error=access_denied&error_description=late&state=state-2",
timeout=2,
) as response:
body = response.read().decode("utf-8")
assert response.status == 200
assert "xAI authorization failed" in body
assert result["code"] == "first-code"
assert result["state"] == "state-1"
assert result["error"] is None
assert result["error_description"] is None
finally:
server.shutdown()
server.server_close()
thread.join(timeout=1.0)
# ---------------------------------------------------------------------------
# Loopback callback handler GET responses
# ---------------------------------------------------------------------------
def _get_callback(redirect_uri: str, query: str = "") -> tuple[int, str]:
"""GET the loopback callback URL with an optional query string."""
from urllib.request import Request, urlopen
from urllib.error import HTTPError
target = redirect_uri + (("?" + query) if query else "")
req = Request(target, method="GET")
try:
with urlopen(req, timeout=5.0) as resp:
return resp.getcode(), resp.read().decode("utf-8", "replace")
except HTTPError as exc:
return exc.code, exc.read().decode("utf-8", "replace")
def test_xai_callback_handler_returns_400_when_callback_url_lacks_code_and_error():
"""Bare loopback URL (no code, no error) must not claim authorization received.
Regression for #27385: when xAI's auth backend fails to redirect and the user
manually navigates to http://127.0.0.1:<port>/callback, the handler used to
return 200 "xAI authorization received" while the CLI's wait loop still timed
out leaving the user with a contradictory success page and a CLI error.
"""
server, thread, result, redirect_uri = _xai_start_callback_server(preferred_port=0)
try:
status, body = _get_callback(redirect_uri)
assert status == 400
assert "not received" in body.lower()
assert "hermes auth add xai-oauth" in body
# Wait loop must still see no code/error so it raises a real timeout,
# rather than treating this empty hit as a successful callback.
assert result["code"] is None
assert result["error"] is None
finally:
server.shutdown()
server.server_close()
thread.join(timeout=1.0)
def test_xai_callback_handler_accepts_callback_with_code():
"""A real OAuth redirect (code + state) still records both and shows success."""
server, thread, result, redirect_uri = _xai_start_callback_server(preferred_port=0)
try:
status, body = _get_callback(redirect_uri, query="code=abc&state=xyz")
assert status == 200
assert "xAI authorization received" in body
assert result["code"] == "abc"
assert result["state"] == "xyz"
assert result["error"] is None
finally:
server.shutdown()
server.server_close()
thread.join(timeout=1.0)
def test_xai_callback_handler_records_error_callback():
"""A redirect carrying an `error` param must surface the failure page and capture detail."""
server, thread, result, redirect_uri = _xai_start_callback_server(preferred_port=0)
try:
status, body = _get_callback(
redirect_uri,
query="error=access_denied&error_description=user%20cancelled",
)
assert status == 200
assert "xAI authorization failed" in body
assert result["error"] == "access_denied"
assert result["error_description"] == "user cancelled"
assert result["code"] is None
finally:
server.shutdown()
server.server_close()
thread.join(timeout=1.0)
assert payload["access_token"] == "xai-access"
assert len(client.calls) == 2
assert client.calls[0][1]["data"]["grant_type"] == "urn:ietf:params:oauth:grant-type:device_code"
# ---------------------------------------------------------------------------
@ -487,7 +333,9 @@ def test_resolve_xai_runtime_credentials_returns_singleton_state(tmp_path, monke
assert creds["api_key"] == fresh
assert creds["base_url"] == DEFAULT_XAI_OAUTH_BASE_URL
assert creds["source"] == "hermes-auth-store"
assert creds["auth_mode"] == "oauth_pkce"
# Display/telemetry label is hardcoded to the only supported flow, even
# though this fixture persisted a legacy ``oauth_pkce`` auth_mode.
assert creds["auth_mode"] == "oauth_device_code"
def test_resolve_xai_runtime_credentials_refreshes_expiring_token(tmp_path, monkeypatch):
@ -814,7 +662,9 @@ def test_get_xai_oauth_auth_status_logged_in_via_singleton(tmp_path, monkeypatch
status = get_xai_oauth_auth_status()
assert status["logged_in"] is True
assert status["api_key"] == fresh
assert status["auth_mode"] == "oauth_pkce"
# Display/telemetry label is hardcoded to the only supported flow, even
# though this fixture persisted a legacy ``oauth_pkce`` auth_mode.
assert status["auth_mode"] == "oauth_device_code"
def test_get_xai_oauth_auth_status_logged_out(tmp_path, monkeypatch):
@ -1168,7 +1018,10 @@ def test_xai_oauth_discovery_validates_authorization_endpoint(monkeypatch):
def test_credential_pool_seeds_xai_oauth_from_singleton(tmp_path, monkeypatch):
"""After `hermes model` -> xai-oauth, the singleton holds tokens. load_pool
must surface that as a pool entry so `hermes auth list` reflects truth and
refreshes route through the pool consistently with codex."""
refreshes route through the pool consistently with codex.
Device code is the only supported xAI OAuth flow, so the singleton is
always surfaced as ``device_code``."""
from agent.credential_pool import load_pool
hermes_home = tmp_path / "hermes"
@ -1183,10 +1036,30 @@ def test_credential_pool_seeds_xai_oauth_from_singleton(tmp_path, monkeypatch):
entry = entries[0]
assert entry.access_token == fresh
assert entry.refresh_token == "rt-1"
assert entry.source == "loopback_pkce"
assert entry.source == "device_code"
assert entry.base_url == DEFAULT_XAI_OAUTH_BASE_URL
def test_credential_pool_seeds_xai_oauth_device_code_source(tmp_path, monkeypatch):
"""Device-code xAI logins should show a device_code source in auth list."""
from agent.credential_pool import load_pool
hermes_home = tmp_path / "hermes"
fresh = _jwt_with_exp(int(time.time()) + 2 * 60 * 60)
_setup_hermes_auth(
hermes_home,
access_token=fresh,
refresh_token="rt-1",
auth_mode="oauth_device_code",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
pool = load_pool("xai-oauth")
entry = pool.entries()[0]
assert entry.source == "device_code"
assert entry.access_token == fresh
def test_credential_pool_does_not_seed_when_singleton_missing_access_token(tmp_path, monkeypatch):
from agent.credential_pool import load_pool
@ -1208,20 +1081,20 @@ def test_credential_pool_does_not_seed_when_singleton_missing_access_token(tmp_p
assert not pool.has_credentials()
def test_credential_pool_seed_respects_suppression(tmp_path, monkeypatch):
"""`hermes auth remove xai-oauth <N>` for the seeded entry suppresses
further re-seeding so the removal is stable across load_pool calls."""
def test_credential_pool_device_code_seed_respects_suppression(tmp_path, monkeypatch):
from agent.credential_pool import load_pool
from hermes_cli.auth import suppress_credential_source
hermes_home = tmp_path / "hermes"
fresh = _jwt_with_exp(int(time.time()) + 2 * 60 * 60)
_setup_hermes_auth(hermes_home, access_token=fresh)
_setup_hermes_auth(
hermes_home,
access_token=fresh,
auth_mode="oauth_device_code",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
# Suppress the source — mimic `hermes auth remove`.
from hermes_cli.auth import suppress_credential_source
suppress_credential_source("xai-oauth", "loopback_pkce")
suppress_credential_source("xai-oauth", "device_code")
pool = load_pool("xai-oauth")
assert not pool.has_credentials()
@ -1235,11 +1108,11 @@ def test_auth_remove_xai_oauth_clears_singleton_and_sticks(tmp_path, monkeypatch
the user-facing removal a no-op (the entry reappears on the next
invocation with no warning).
The bug pre-fix: there was no RemovalStep registered for
(xai-oauth, loopback_pkce), so ``find_removal_step`` returned None
The bug pre-fix: there was no RemovalStep registered for the
xai-oauth singleton source, so ``find_removal_step`` returned None
and ``auth_remove_command`` fell through to the "unregistered source —
nothing to clean up" branch. That branch is correct for ``manual``
entries (pool-only) but wrong for singleton-seeded loopback_pkce
entries (pool-only) but wrong for singleton-seeded ``device_code``
entries (auth.json singleton survives the in-memory removal)."""
from agent.credential_pool import load_pool
from hermes_cli.auth_commands import auth_remove_command
@ -1272,11 +1145,82 @@ def test_auth_remove_xai_oauth_clears_singleton_and_sticks(tmp_path, monkeypatch
pool_after = load_pool("xai-oauth")
assert not pool_after.has_credentials(), (
"Removal must stick across load_pool() calls — without the "
"loopback_pkce RemovalStep, the seed function reads the singleton "
"device_code RemovalStep, the seed function reads the singleton "
"and rebuilds the entry on every Hermes invocation."
)
def test_login_xai_oauth_relogin_clears_suppression_and_reseeds(tmp_path, monkeypatch):
"""remove -> ``hermes model`` re-login (``_login_xai_oauth``) must clear the
``device_code`` suppression marker so the singleton seed re-creates the
pool entry.
Pre-fix: ``auth_remove_command`` set ``["device_code"]`` suppression but
only ``auth_add_command`` cleared it the ``hermes model`` re-login path did
not. So after remove -> re-login the seed kept skipping and ``hermes auth
list`` showed no xAI entry even though the agent still worked via the
singleton fallback. The fix calls ``unsuppress_credential_source`` on
explicit interactive login success.
"""
from types import SimpleNamespace
from agent.credential_pool import load_pool
from hermes_cli.auth import (
_login_xai_oauth,
is_source_suppressed,
suppress_credential_source,
)
hermes_home = tmp_path / "hermes"
hermes_home.mkdir(parents=True, exist_ok=True)
(hermes_home / "auth.json").write_text(json.dumps({"version": 1, "providers": {}}))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.delenv("HERMES_XAI_BASE_URL", raising=False)
monkeypatch.delenv("XAI_BASE_URL", raising=False)
# Post-remove state: singleton gone + device_code suppressed, so the
# seed is gated off and the pool is empty.
suppress_credential_source("xai-oauth", "device_code")
assert is_source_suppressed("xai-oauth", "device_code") is True
assert not load_pool("xai-oauth").has_credentials()
new_access = _jwt_with_exp(int(time.time()) + 2 * 60 * 60)
monkeypatch.setattr(
"hermes_cli.auth._xai_oauth_device_code_login",
lambda **kwargs: {
"tokens": {
"access_token": new_access,
"refresh_token": "rt-relogin",
"id_token": "",
"token_type": "Bearer",
},
"discovery": {"token_endpoint": "https://auth.x.ai/token"},
"redirect_uri": "",
"base_url": DEFAULT_XAI_OAUTH_BASE_URL,
"last_refresh": "2026-06-30T10:00:00Z",
},
)
# Don't mutate a real config file during the test.
monkeypatch.setattr(
"hermes_cli.auth._update_config_for_provider",
lambda *args, **kwargs: "config.toml",
)
_login_xai_oauth(
SimpleNamespace(no_browser=True, timeout=3),
None, # pconfig is `del`-eted inside the function
force_new_login=True,
)
# The explicit interactive login cleared the suppression marker...
assert is_source_suppressed("xai-oauth", "device_code") is False
# ...so the singleton seed re-creates the canonical pool entry.
pool = load_pool("xai-oauth")
assert pool.has_credentials()
entry = next(e for e in pool.entries() if e.source == "device_code")
assert entry.access_token == new_access
# ---------------------------------------------------------------------------
# Pool sync-back to singleton after refresh
# ---------------------------------------------------------------------------
@ -1599,7 +1543,7 @@ def test_pool_refresh_marks_entry_exhausted_on_failure(tmp_path, monkeypatch):
def test_pool_seeded_entry_sync_back_after_refresh(tmp_path, monkeypatch):
"""When an entry seeded from the singleton (source='loopback_pkce')
"""When an entry seeded from the singleton (source='device_code')
is refreshed by the pool, the new tokens must be written back so a
fresh process load doesn't re-seed the now-consumed refresh token."""
from agent.credential_pool import load_pool
@ -1756,7 +1700,7 @@ def test_pool_exhausted_xai_entry_recovers_after_singleton_refresh(tmp_path, mon
pool = load_pool("xai-oauth")
seeded = pool.entries()[0]
assert seeded.source == "loopback_pkce"
assert seeded.source == "device_code"
# Park the seeded entry as exhausted with a far-future cooldown so
# without resync it would never be selectable.
@ -1796,7 +1740,7 @@ def test_pool_exhausted_xai_entry_recovers_after_singleton_refresh(tmp_path, mon
def test_pool_manual_xai_entry_not_synced_from_singleton(tmp_path, monkeypatch):
"""Sync from the singleton must apply ONLY to the singleton-seeded
entry (source='loopback_pkce'). Manually added entries (e.g. via
entry (source='device_code'). Manually added entries (e.g. via
``hermes auth add xai-oauth``) own their own refresh-token lifecycle
and must not be silently overwritten when the user logs in via
``hermes model``."""

View file

@ -460,13 +460,13 @@ def test_anthropic_pkce_branch_still_works():
assert "claude.ai" in body["auth_url"]
def test_xai_oauth_listed_as_loopback_flow():
"""xAI Grok OAuth must surface in the catalog as a first-class loopback flow."""
def test_xai_oauth_listed_as_device_code_flow():
"""xAI Grok OAuth must surface in the catalog as a device-code flow."""
resp = client.get("/api/providers/oauth", headers=HEADERS)
assert resp.status_code == 200, resp.text
providers = {p["id"]: p for p in resp.json()["providers"]}
assert "xai-oauth" in providers
assert providers["xai-oauth"]["flow"] == "loopback"
assert providers["xai-oauth"]["flow"] == "device_code"
assert "grok" in providers["xai-oauth"]["name"].lower()
@ -557,247 +557,117 @@ def test_env_sourced_oauth_status_is_not_disconnectable(monkeypatch):
assert "Settings" in delete_resp.text
def test_xai_loopback_start_returns_authorize_url(monkeypatch):
"""Start MUST bind the loopback listener and hand back an xAI authorize URL."""
def test_xai_oauth_device_code_start_returns_user_code(monkeypatch):
"""Start MUST hand back xAI's verification URL and user code."""
from hermes_cli import auth as auth_mod
from hermes_cli import web_server as ws
class _FakeServer:
def shutdown(self):
pass
def server_close(self):
pass
class _FakeThread:
def join(self, timeout=None):
pass
redirect_uri = (
f"http://{auth_mod.XAI_OAUTH_REDIRECT_HOST}:{auth_mod.XAI_OAUTH_REDIRECT_PORT}"
f"{auth_mod.XAI_OAUTH_REDIRECT_PATH}"
)
monkeypatch.setattr(
auth_mod,
"_xai_oauth_discovery",
"_xai_oauth_request_device_code",
lambda *a, **k: {
"authorization_endpoint": "https://auth.x.ai/oauth2/auth",
"token_endpoint": "https://auth.x.ai/oauth2/token",
"device_code": "device-code",
"user_code": "ABCD-EFGH",
"verification_uri": "https://accounts.x.ai/oauth2/device",
"verification_uri_complete": "https://accounts.x.ai/oauth2/device?user_code=ABCD-EFGH",
"expires_in": 1800,
"interval": 5,
},
)
monkeypatch.setattr(
auth_mod,
"_xai_start_callback_server",
lambda *a, **k: (_FakeServer(), _FakeThread(), {"code": None, "error": None}, redirect_uri),
)
# Don't let the background worker run a real callback wait/exchange.
monkeypatch.setattr(ws, "_xai_loopback_worker", lambda sid: None)
# Don't let the background poller hit the real token endpoint.
monkeypatch.setattr(ws, "_xai_device_poller", lambda sid: None)
resp = client.post("/api/providers/oauth/xai-oauth/start", headers=HEADERS)
assert resp.status_code == 200, resp.text
body = resp.json()
try:
assert body["flow"] == "loopback"
assert "user_code" not in body # loopback has nothing to paste/show
assert body["auth_url"].startswith("https://auth.x.ai/oauth2/auth?")
assert "code_challenge" in body["auth_url"]
assert body["flow"] == "device_code"
assert body["user_code"] == "ABCD-EFGH"
assert body["verification_url"].startswith("https://accounts.x.ai/oauth2/device")
sess = ws._oauth_sessions[body["session_id"]]
assert sess["provider"] == "xai-oauth"
assert sess["flow"] == "loopback"
assert sess["flow"] == "device_code"
assert sess["device_code"] == "device-code"
finally:
ws._oauth_sessions.pop(body["session_id"], None)
def test_xai_loopback_worker_persists_tokens_on_success(monkeypatch):
"""The worker exchanges the callback code and marks the session approved."""
def test_xai_dashboard_poller_seeds_single_entry_and_clears_suppression(tmp_path, monkeypatch):
"""The dashboard device-code poller must leave exactly ONE pool entry — the
singleton-seeded ``device_code`` source and must NOT create a parallel
``manual:dashboard_*`` entry.
Dedupe: a parallel dashboard entry would share the singleton's single-use
refresh token, and two entries racing the same rotation ->
``refresh_token_reused`` (on main, the dashboard login inserted exactly
such a duplicate alongside the singleton seed). The poller writes the
singleton only; the seed is the single source of truth.
Suppression: an interactive dashboard login must also clear any
``device_code`` suppression left by a prior ``hermes auth remove
xai-oauth``.
"""
from hermes_cli import auth as auth_mod
from hermes_cli import web_server as ws
from agent.credential_pool import load_pool
saved = {}
session_id = "xai-loopback-success-test"
ws._oauth_sessions[session_id] = {
"session_id": session_id,
"provider": "xai-oauth",
"flow": "loopback",
"created_at": time.time(),
"status": "pending",
"error_message": None,
"server": object(),
"thread": object(),
"callback_result": {"code": "auth-code", "state": "st"},
"redirect_uri": "http://127.0.0.1:56121/callback",
"verifier": "verifier",
"challenge": "challenge",
"state": "st",
"token_endpoint": "https://auth.x.ai/oauth2/token",
"discovery": {"token_endpoint": "https://auth.x.ai/oauth2/token"},
}
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.delenv("HERMES_XAI_BASE_URL", raising=False)
monkeypatch.delenv("XAI_BASE_URL", raising=False)
# Prior `hermes auth remove xai-oauth` left the source suppressed.
auth_mod.suppress_credential_source("xai-oauth", "device_code")
assert auth_mod.is_source_suppressed("xai-oauth", "device_code") is True
monkeypatch.setattr(
auth_mod,
"_xai_wait_for_callback",
lambda *a, **k: {"code": "auth-code", "state": "st"},
"_xai_oauth_discovery",
lambda *a, **k: {"token_endpoint": "https://auth.x.ai/token"},
)
monkeypatch.setattr(
auth_mod,
"_xai_oauth_exchange_code_for_tokens",
lambda **k: {
"access_token": "xai-access",
"refresh_token": "xai-refresh",
"_xai_oauth_poll_device_token",
lambda client, **kwargs: {
"access_token": "xai-dashboard-access",
"refresh_token": "rt-dashboard",
"id_token": "",
"expires_in": 3600,
"token_type": "Bearer",
},
)
monkeypatch.setattr(
auth_mod,
"_save_xai_oauth_tokens",
lambda tokens, **k: saved.update(tokens),
)
monkeypatch.setattr(ws, "_add_xai_oauth_pool_entry", lambda *a, **k: None)
session_id = "xai-dashboard-dedupe-test"
ws._oauth_sessions[session_id] = {
"session_id": session_id,
"provider": "xai-oauth",
"flow": "device_code",
"created_at": time.time(),
"status": "pending",
"error_message": None,
"device_code": "device-code",
"interval": 5,
"expires_at": time.time() + 600,
}
try:
ws._xai_loopback_worker(session_id)
ws._xai_device_poller(session_id)
assert ws._oauth_sessions[session_id]["status"] == "approved"
assert saved["access_token"] == "xai-access"
assert saved["refresh_token"] == "xai-refresh"
finally:
ws._oauth_sessions.pop(session_id, None)
# The interactive dashboard login cleared the suppression marker.
assert auth_mod.is_source_suppressed("xai-oauth", "device_code") is False
def test_xai_loopback_worker_fails_on_state_mismatch(monkeypatch):
"""A mismatched OAuth state must fail the session, not persist tokens."""
from hermes_cli import auth as auth_mod
from hermes_cli import web_server as ws
session_id = "xai-loopback-state-test"
ws._oauth_sessions[session_id] = {
"session_id": session_id,
"provider": "xai-oauth",
"flow": "loopback",
"created_at": time.time(),
"status": "pending",
"error_message": None,
"server": object(),
"thread": object(),
"callback_result": {},
"redirect_uri": "http://127.0.0.1:56121/callback",
"verifier": "verifier",
"challenge": "challenge",
"state": "expected-state",
"token_endpoint": "https://auth.x.ai/oauth2/token",
"discovery": {},
}
monkeypatch.setattr(
auth_mod,
"_xai_wait_for_callback",
lambda *a, **k: {"code": "auth-code", "state": "ATTACKER-state"},
# The credential pool has exactly one entry, seeded from the
# singleton as ``device_code`` — no parallel ``manual:dashboard_*``
# duplicate sharing the single-use refresh token.
entries = load_pool("xai-oauth").entries()
assert len(entries) == 1
assert entries[0].source == "device_code"
assert entries[0].refresh_token == "rt-dashboard"
assert not any(
getattr(e, "source", "").startswith("manual:dashboard") for e in entries
)
def _boom(**kwargs):
raise AssertionError("token exchange must not run on state mismatch")
monkeypatch.setattr(auth_mod, "_xai_oauth_exchange_code_for_tokens", _boom)
try:
ws._xai_loopback_worker(session_id)
sess = ws._oauth_sessions[session_id]
assert sess["status"] == "error"
assert "state mismatch" in sess["error_message"].lower()
finally:
ws._oauth_sessions.pop(session_id, None)
def test_xai_loopback_worker_skips_persist_when_cancelled(monkeypatch):
"""If the session is cancelled while waiting, the worker must not persist."""
from hermes_cli import auth as auth_mod
from hermes_cli import web_server as ws
session_id = "xai-loopback-cancel-test"
ws._oauth_sessions[session_id] = {
"session_id": session_id,
"provider": "xai-oauth",
"flow": "loopback",
"created_at": time.time(),
"status": "pending",
"error_message": None,
"server": object(),
"thread": object(),
"callback_result": {},
"redirect_uri": "http://127.0.0.1:56121/callback",
"verifier": "verifier",
"challenge": "challenge",
"state": "st",
"token_endpoint": "https://auth.x.ai/oauth2/token",
"discovery": {},
}
def _wait_then_cancel(*args, **kwargs):
# Simulate the user cancelling (DELETE /sessions/{id}) while we were
# blocked on the callback: the session vanishes, then a valid code
# arrives. The worker must notice and bail before persisting.
ws._oauth_sessions.pop(session_id, None)
return {"code": "auth-code", "state": "st"}
monkeypatch.setattr(auth_mod, "_xai_wait_for_callback", _wait_then_cancel)
def _must_not_persist(*args, **kwargs):
raise AssertionError("tokens must not be persisted for a cancelled session")
monkeypatch.setattr(auth_mod, "_save_xai_oauth_tokens", _must_not_persist)
monkeypatch.setattr(ws, "_add_xai_oauth_pool_entry", _must_not_persist)
# Should return cleanly without raising and without persisting.
ws._xai_loopback_worker(session_id)
assert session_id not in ws._oauth_sessions
def test_cancel_loopback_session_shuts_down_callback_server():
"""Cancelling a loopback session must free the bound callback port now."""
from hermes_cli import web_server as ws
shutdown_calls = {"shutdown": 0, "close": 0, "join": 0}
class _FakeServer:
def shutdown(self):
shutdown_calls["shutdown"] += 1
def server_close(self):
shutdown_calls["close"] += 1
class _FakeThread:
def join(self, timeout=None):
shutdown_calls["join"] += 1
# callback_result is the dict the worker's _xai_wait_for_callback polls.
callback_result = {"code": None, "error": None}
session_id = "xai-loopback-cancel-shutdown-test"
ws._oauth_sessions[session_id] = {
"session_id": session_id,
"provider": "xai-oauth",
"flow": "loopback",
"created_at": time.time(),
"status": "pending",
"server": _FakeServer(),
"thread": _FakeThread(),
"callback_result": callback_result,
}
try:
resp = client.delete(
f"/api/providers/oauth/sessions/{session_id}", headers=HEADERS
)
assert resp.status_code == 200, resp.text
assert resp.json()["ok"] is True
assert shutdown_calls == {"shutdown": 1, "close": 1, "join": 1}
# The waiting worker must be signalled so it returns promptly instead
# of spinning until the timeout.
assert callback_result["error"] == "cancelled"
assert session_id not in ws._oauth_sessions
finally:
ws._oauth_sessions.pop(session_id, None)
def test_unknown_pkce_provider_rejected_cleanly():
"""A future PKCE provider without an explicit branch must NOT silently route to Anthropic.

View file

@ -33,12 +33,11 @@ def test_xai_model_flow_reauth_uses_standard_radio_prompt(monkeypatch):
main_mod._model_flow_xai_oauth(
{},
current_model="grok-build-0.1",
args=argparse.Namespace(manual_paste=True, no_browser=True, timeout=3),
args=argparse.Namespace(no_browser=True, timeout=3),
)
assert captured["login_calls"] == 1
assert captured["force_new_login"] is True
assert captured["args"].manual_paste is True
assert captured["args"].no_browser is True
assert captured["args"].timeout == 3

View file

@ -1,359 +0,0 @@
"""Regression coverage for xAI OAuth PKCE token exchange (issue #26990).
Issue [#26990] reported that ``hermes auth add xai-oauth`` succeeds at the
browser-side authorize step but fails at the token endpoint with
``code_challenge is required`` the symptom of an OAuth server that
re-validates PKCE at the token step instead of relying purely on
state captured during the authorize redirect.
The fix in ``hermes_cli/auth.py`` extracts the token POST into
:func:`_xai_oauth_exchange_code_for_tokens` and:
* Sends ``code_verifier`` (RFC 7636 §4.5 requirement).
* **Also** echoes ``code_challenge`` and ``code_challenge_method``
in the request body as defense-in-depth strictly compliant
servers ignore extras at the token endpoint, but xAI's server
needs them.
* Refuses to fire the POST locally when ``code_verifier`` is empty
(avoids leaking the auth code to a server that can't redeem it).
* Surfaces the HTTP status code prominently in the error message so
users / maintainers can tell a 400 (bad request) from a 403
(entitlement denied) at a glance.
These tests pin all three behaviors so the fix can't silently regress.
"""
from __future__ import annotations
from typing import Any, Dict, List
from urllib.parse import parse_qs
import httpx
import pytest
from hermes_cli.auth import (
AuthError,
XAI_OAUTH_CLIENT_ID,
_xai_oauth_exchange_code_for_tokens,
)
# ---------------------------------------------------------------------------
# httpx.post recorder
# ---------------------------------------------------------------------------
class _PostRecorder:
"""Capture every ``httpx.post`` call without touching the network."""
def __init__(self, response: httpx.Response) -> None:
self.response = response
self.calls: List[Dict[str, Any]] = []
def __call__(self, url, *, headers=None, data=None, timeout=None, **kw):
self.calls.append(
{"url": url, "headers": headers or {}, "data": data or {},
"timeout": timeout, "extra": kw}
)
return self.response
def _ok_response(payload: dict) -> httpx.Response:
return httpx.Response(200, json=payload)
def _err_response(status: int, body: str) -> httpx.Response:
return httpx.Response(status, text=body)
@pytest.fixture
def post_recorder(monkeypatch):
"""Default: 200 response with a full xAI token payload."""
recorder = _PostRecorder(
_ok_response(
{
"access_token": "AT-fresh",
"refresh_token": "RT-fresh",
"id_token": "ID",
"expires_in": 3600,
"token_type": "Bearer",
}
)
)
monkeypatch.setattr("hermes_cli.auth.httpx.post", recorder)
return recorder
# ---------------------------------------------------------------------------
# Core contract: which fields go on the wire?
# ---------------------------------------------------------------------------
def test_token_exchange_includes_code_verifier(post_recorder):
"""RFC 7636 §4.5 — ``code_verifier`` MUST be sent."""
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="theVerifier_43_to_128_chars_____________________",
code_challenge="aBcDeF",
)
sent = post_recorder.calls[-1]["data"]
assert sent["code_verifier"] == "theVerifier_43_to_128_chars_____________________"
def test_token_exchange_also_echoes_code_challenge_for_xai(post_recorder):
"""Defense-in-depth for #26990 — xAI re-validates the challenge
at the token endpoint, not just at authorize. Without this echo
we get ``code_challenge is required`` even though we send a valid
``code_verifier``."""
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="aBcDeF",
)
sent = post_recorder.calls[-1]["data"]
assert sent["code_challenge"] == "aBcDeF"
assert sent["code_challenge_method"] == "S256"
def test_token_exchange_uses_correct_grant_and_client(post_recorder):
"""Lock the static fields too — a future refactor must not flip
these to ``client_credentials`` or drop ``client_id``."""
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
)
sent = post_recorder.calls[-1]["data"]
assert sent["grant_type"] == "authorization_code"
assert sent["code"] == "AUTHCODE"
assert sent["redirect_uri"] == "http://127.0.0.1:56121/callback"
assert sent["client_id"] == XAI_OAUTH_CLIENT_ID
def test_token_exchange_uses_form_urlencoded_content_type(post_recorder):
"""xAI's token endpoint expects ``application/x-www-form-urlencoded``."""
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
)
headers = post_recorder.calls[-1]["headers"]
assert headers["Content-Type"] == "application/x-www-form-urlencoded"
assert headers["Accept"] == "application/json"
def test_token_exchange_targets_the_supplied_endpoint(post_recorder):
"""Some test fixtures sniff the discovered token endpoint dynamically.
We must POST to the URL the caller passed, not a hard-coded constant."""
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/some/other/token/path",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
)
assert post_recorder.calls[-1]["url"] == "https://auth.x.ai/some/other/token/path"
def test_token_exchange_passes_timeout_through(post_recorder):
"""Operators on slow networks pass a higher ``timeout_seconds``;
the helper must forward it (and bump the floor to 20s)."""
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
timeout_seconds=45.0,
)
assert post_recorder.calls[-1]["timeout"] == 45.0
def test_token_exchange_floor_timeout_is_20s(post_recorder):
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
timeout_seconds=2.0,
)
assert post_recorder.calls[-1]["timeout"] == 20.0
# ---------------------------------------------------------------------------
# Sanity guard: refuse to POST with an empty code_verifier
# ---------------------------------------------------------------------------
def test_empty_code_verifier_raises_without_posting(post_recorder):
"""If ``code_verifier`` is somehow lost upstream, we must refuse to
send the request leaking an authorization code to xAI without a
verifier is worse than failing locally with an actionable error."""
with pytest.raises(AuthError) as exc_info:
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="",
code_challenge="c" * 43,
)
assert exc_info.value.code == "xai_pkce_verifier_missing"
assert "26990" in str(exc_info.value)
# And critically: nothing was sent.
assert post_recorder.calls == []
def test_missing_code_challenge_omits_echo_but_still_sends_verifier(post_recorder):
"""``code_challenge`` is defensive — if a caller doesn't have it
handy, we must still send the standards-compliant request rather
than refusing. This keeps RFC-compliant servers happy."""
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="",
)
sent = post_recorder.calls[-1]["data"]
assert sent["code_verifier"] == "v" * 64
assert "code_challenge" not in sent
assert "code_challenge_method" not in sent
# ---------------------------------------------------------------------------
# Error surfacing
# ---------------------------------------------------------------------------
def test_non_200_response_surfaces_status_and_body(monkeypatch):
"""When xAI returns a 4xx, the operator needs both the HTTP status
code (to tell 400 from 401 from 403 at a glance) and the response
body (the actual server-side reason)."""
recorder = _PostRecorder(
_err_response(400, '{"error":"invalid_grant","error_description":"code_challenge is required"}')
)
monkeypatch.setattr("hermes_cli.auth.httpx.post", recorder)
with pytest.raises(AuthError) as exc_info:
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
)
msg = str(exc_info.value)
assert "HTTP 400" in msg, (
"Status code must be in the error so callers can disambiguate "
"tier-denied (403) from bad-request (400) without inspecting "
"exc.code."
)
assert "code_challenge is required" in msg
assert exc_info.value.code == "xai_token_exchange_failed"
def test_transport_error_wraps_as_auth_error(monkeypatch):
"""A connection failure must come back as ``AuthError`` so the
surrounding ``format_auth_error`` UI mapping fires correctly."""
def _boom(*args, **kwargs):
raise httpx.ConnectError("dns failure")
monkeypatch.setattr("hermes_cli.auth.httpx.post", _boom)
with pytest.raises(AuthError) as exc_info:
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
)
assert exc_info.value.code == "xai_token_exchange_failed"
assert "dns failure" in str(exc_info.value)
def test_non_dict_payload_raises_invalid_json(monkeypatch):
"""xAI returning ``[]`` or a string at 200 is a server bug — fail
with a precise error rather than crashing later in token storage."""
recorder = _PostRecorder(_ok_response([1, 2, 3])) # type: ignore[arg-type]
monkeypatch.setattr("hermes_cli.auth.httpx.post", recorder)
with pytest.raises(AuthError) as exc_info:
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
)
assert exc_info.value.code == "xai_token_exchange_invalid"
def test_success_returns_full_payload_dict(post_recorder):
"""200 happy path: the parsed JSON dict comes back verbatim so the
caller can pluck ``access_token`` / ``refresh_token`` etc."""
out = _xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="v" * 64,
code_challenge="c" * 43,
)
assert out["access_token"] == "AT-fresh"
assert out["refresh_token"] == "RT-fresh"
# ---------------------------------------------------------------------------
# Wire-format guard: httpx must serialise ``data`` as form-urlencoded
# ---------------------------------------------------------------------------
def test_wire_format_is_form_urlencoded_with_all_pkce_fields(monkeypatch):
"""End-to-end check on the actual bytes httpx puts on the wire.
If anyone ever swaps ``data=`` for ``json=`` or refactors the dict,
xAI will start rejecting again this catches it locally."""
captured: Dict[str, Any] = {}
class _Transport(httpx.BaseTransport):
def handle_request(self, request):
captured["body"] = bytes(request.read())
captured["content_type"] = request.headers.get("content-type", "")
return httpx.Response(
200,
json={"access_token": "AT", "refresh_token": "RT",
"id_token": "", "expires_in": 60, "token_type": "Bearer"},
)
real_post = httpx.post
def _post(*args, **kwargs):
with httpx.Client(transport=_Transport()) as c:
return c.post(*args, **kwargs)
monkeypatch.setattr("hermes_cli.auth.httpx.post", _post)
_xai_oauth_exchange_code_for_tokens(
token_endpoint="https://auth.x.ai/oauth2/token",
code="AUTHCODE",
redirect_uri="http://127.0.0.1:56121/callback",
code_verifier="theVerifier_43+",
code_challenge="theChallenge_43+",
)
assert "application/x-www-form-urlencoded" in captured["content_type"]
parsed = parse_qs(captured["body"].decode())
assert parsed["grant_type"] == ["authorization_code"]
assert parsed["code"] == ["AUTHCODE"]
assert parsed["redirect_uri"] == ["http://127.0.0.1:56121/callback"]
assert parsed["client_id"] == [XAI_OAUTH_CLIENT_ID]
assert parsed["code_verifier"] == ["theVerifier_43+"]
assert parsed["code_challenge"] == ["theChallenge_43+"]
assert parsed["code_challenge_method"] == ["S256"]

View file

@ -41,3 +41,17 @@ def test_xai_oauth_token_not_expiring_beyond_one_hour_skew() -> None:
token,
auth.XAI_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
)
def test_xai_proactive_refresh_skew_short_lived_token() -> None:
token = _jwt_with_exp(int(time.time()) + 15 * 60)
skew = auth._xai_proactive_refresh_skew_seconds(token)
assert skew == 120
assert not auth._xai_access_token_is_expiring(token, skew)
def test_xai_proactive_refresh_skew_long_lived_token() -> None:
token = _jwt_with_exp(int(time.time()) + 5 * 60 * 60)
assert auth._xai_proactive_refresh_skew_seconds(token) == auth.XAI_ACCESS_TOKEN_REFRESH_SKEW_SECONDS

View file

@ -958,7 +958,7 @@ def test_try_refresh_codex_client_credentials_handles_xai_oauth(monkeypatch):
def test_try_refresh_codex_client_credentials_skips_xai_oauth_when_singleton_differs(monkeypatch):
"""An xai-oauth agent constructed with a non-singleton credential
(e.g. a manual pool entry whose tokens belong to a different account
than the loopback_pkce singleton, or an explicit ``api_key=`` arg)
than the device_code singleton, or an explicit ``api_key=`` arg)
MUST NOT silently adopt the singleton's tokens on a 401 reactive
refresh. Otherwise a 401 mid-conversation would re-route the rest
of the conversation onto a different account, with no user feedback.

View file

@ -1,56 +1,41 @@
---
sidebar_position: 17
title: "OAuth over SSH / Remote Hosts"
description: "How to complete browser-based OAuth (xAI, Spotify, MCP servers) when Hermes runs on a remote machine, container, or behind a jump box"
description: "How to complete browser-based OAuth (Spotify, MCP servers) when Hermes runs on a remote machine, container, or behind a jump box"
---
# OAuth over SSH / Remote Hosts
Some Hermes providers — **xAI Grok OAuth**, **Spotify**, and **remote MCP servers** (Linear, Sentry, Atlassian, Asana, Figma, …) — use a *loopback redirect* OAuth flow. The auth server redirects your browser to `http://127.0.0.1:<port>/callback` so a tiny HTTP listener started by Hermes can grab the authorization code.
Some Hermes providers — **Spotify** and **remote MCP servers** (Linear, Sentry, Atlassian, Asana, Figma, …) — use a *loopback redirect* OAuth flow. The auth server redirects your browser to `http://127.0.0.1:<port>/callback` so a tiny HTTP listener started by Hermes can grab the authorization code.
This works perfectly when Hermes and your browser are on the same machine. It breaks the moment they aren't: your laptop's browser tries to reach `127.0.0.1` on **your laptop**, but the listener is bound to `127.0.0.1` on **the remote server**.
The fix is a one-line SSH local-forward — **or**, when you don't have a real SSH client (GCP Cloud Shell, GitHub Codespaces, EC2 Instance Connect, Gitpod, browser-based web IDEs), the new `--manual-paste` flag introduced in [#26923](https://github.com/NousResearch/hermes-agent/issues/26923).
The fix is a one-line SSH local-forward. For MCP servers on an interactive terminal, you can often paste the redirect URL back instead (no tunnel).
**xAI Grok OAuth (`xai-oauth`) uses OAuth device code**, not a loopback callback — open the printed verification URL in any browser and Hermes polls until approval. No SSH tunnel is required. See [xAI Grok OAuth](./xai-grok-oauth.md).
## TL;DR
```bash
# On your local machine (laptop), in a separate terminal:
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
ssh -N -L 43827:127.0.0.1:43827 user@remote-host
# In your existing SSH session on the remote machine:
hermes auth add xai-oauth --no-browser
hermes auth add spotify --no-browser
# → Hermes prints an authorize URL. Open it in a browser on your laptop.
# → Your browser redirects to 127.0.0.1:56121/callback, the tunnel forwards
# → Your browser redirects to 127.0.0.1:43827/callback, the tunnel forwards
# the request to the remote listener, login completes.
```
Port `56121` is what xAI OAuth uses. For Spotify, replace it with `43827`. Hermes prints the exact port it bound to on the `Waiting for callback on ...` line — copy it from there.
## Browser-only remote (Cloud Shell / Codespaces / EC2 Instance Connect)
If you don't have a regular SSH client — for example because you're running Hermes inside GCP Cloud Shell, GitHub Codespaces, AWS EC2 Instance Connect, Gitpod, or another browser-based console — the SSH tunnel above isn't available. Use `--manual-paste` instead:
```bash
hermes auth add xai-oauth --manual-paste
# → Hermes prints an authorize URL. Open it in a browser on your laptop.
# → Approve in the browser. The redirect to 127.0.0.1:56121/callback fails
# to load — that's expected.
# → Copy the FULL URL from the failed page's address bar.
# → Paste it back into the terminal at the "Callback URL:" prompt.
```
The same flag works on `hermes model --manual-paste` for the integrated model picker. Hermes accepts three callback paste forms interchangeably: the full URL, a bare `?code=...&state=...` query fragment, or — when the upstream consent page renders the authorization code in-page instead of redirecting (xAI's current behavior on browser-based consoles) — just the bare code value on its own.
Hermes uses the **same PKCE verifier, state and nonce** for both paths, so the upstream OAuth flow is byte-identical — `--manual-paste` is purely a transport change for the callback hop and is not a security downgrade.
Hermes prints the exact port it bound to on the `Waiting for callback on ...` line — copy it from there. Spotify defaults to port `43827`.
## Which Providers Need This
| Provider | Loopback port | Tunnel needed? |
|----------|---------------|----------------|
| `xai-oauth` (Grok SuperGrok) | `56121` | Yes, when Hermes is remote |
| Spotify | `43827` | Yes, when Hermes is remote |
| MCP servers (`auth: oauth`) | auto-picked per server | Yes, when Hermes is remote |
| Spotify | `43827` (default) | Yes, when Hermes is remote |
| MCP servers (`auth: oauth`) | auto-picked per server | Yes, when Hermes is remote (or paste redirect URL) |
| `xai-oauth` (Grok SuperGrok) | n/a | No — device code flow |
| `anthropic` (Claude Pro/Max) | n/a | No — paste-the-code flow |
| `openai-codex` (ChatGPT Plus/Pro) | n/a | No — device code flow |
| `minimax`, `nous-portal` | n/a | No — device code flow |
@ -78,7 +63,7 @@ You have two ways to complete it from a remote host:
A bare `?code=...&state=...` query string is accepted too. This works for any MCP server with `auth: oauth` and requires no SSH config changes.
**Option 2 — SSH port forward (same as xAI / Spotify).** Hermes prints the exact port it bound to in the SSH-session hint. Open a separate terminal on your laptop:
**Option 2 — SSH port forward (same as Spotify).** Hermes prints the exact port it bound to in the SSH-session hint. Open a separate terminal on your laptop:
```bash
ssh -N -L <port>:127.0.0.1:<port> user@remote-host
@ -90,17 +75,14 @@ Then open the authorize URL in your browser as normal; the redirect tunnels thro
## Why the listener can't just bind 0.0.0.0
xAI and Spotify both validate the `redirect_uri` parameter against an allowlist. Both require the loopback form (`http://127.0.0.1:<exact-port>/callback`). Binding the listener to `0.0.0.0` or a different port would cause the auth server to reject the request as a redirect_uri mismatch. The SSH tunnel keeps the loopback URI intact end-to-end.
Spotify and most MCP OAuth servers validate the `redirect_uri` parameter against an allowlist. Both require the loopback form (`http://127.0.0.1:<exact-port>/callback`). Binding the listener to `0.0.0.0` or a different port would cause the auth server to reject the request as a redirect_uri mismatch. The SSH tunnel keeps the loopback URI intact end-to-end.
## Step-by-step: single SSH hop
### 1. Start the tunnel from your local machine
```bash
# xAI Grok OAuth (port 56121)
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
# Or for Spotify (port 43827)
# Spotify (port 43827)
ssh -N -L 43827:127.0.0.1:43827 user@remote-host
```
@ -110,9 +92,7 @@ ssh -N -L 43827:127.0.0.1:43827 user@remote-host
```bash
ssh user@remote-host
hermes auth add xai-oauth --no-browser
# or for Spotify:
# hermes auth add spotify --no-browser
hermes auth add spotify --no-browser
```
Hermes detects the SSH session, skips the browser auto-open, and prints an authorize URL plus a `Waiting for callback on http://127.0.0.1:<port>/callback` line.
@ -128,17 +108,17 @@ You can tear down the tunnel (Ctrl+C in the first terminal) once you see the suc
If you reach Hermes through a bastion / jump host, use SSH's built-in `-J` (ProxyJump):
```bash
ssh -N -L 56121:127.0.0.1:56121 -J jump-user@jump-host user@final-host
ssh -N -L 43827:127.0.0.1:43827 -J jump-user@jump-host user@final-host
```
This chains a SSH connection through the jump host without putting the loopback port on the jump box itself. The local `127.0.0.1:56121` on your laptop tunnels straight through to `127.0.0.1:56121` on the final remote host.
This chains a SSH connection through the jump host without putting the loopback port on the jump box itself. The local `127.0.0.1:43827` on your laptop tunnels straight through to `127.0.0.1:43827` on the final remote host.
For older OpenSSH that doesn't support `-J`, the long form is:
```bash
ssh -N \
-o "ProxyCommand=ssh -W %h:%p jump-user@jump-host" \
-L 56121:127.0.0.1:56121 \
-L 43827:127.0.0.1:43827 \
user@final-host
```
@ -150,30 +130,26 @@ If you use `ssh -o ControlMaster=auto`, port forwards on a multiplexed connectio
```bash
ssh -O exit user@remote-host
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
ssh -N -L 43827:127.0.0.1:43827 user@remote-host
```
## Troubleshooting
### `bind [127.0.0.1]:56121: Address already in use`
### `bind [127.0.0.1]:43827: Address already in use`
Something on your laptop is already using that port. Either the previous tunnel didn't shut down cleanly, or a local Hermes is also listening on it. Find and kill the offender:
```bash
# macOS / Linux
lsof -iTCP:56121 -sTCP:LISTEN
lsof -iTCP:43827 -sTCP:LISTEN
kill <PID>
```
Then retry the `ssh -L` command.
### "Could not establish connection. We couldn't reach your app." (xAI)
### Authorization timed out waiting for the local callback
xAI's authorize page shows this when its redirect to `127.0.0.1:<port>/callback` doesn't reach a listener. Either the tunnel isn't running, the port is wrong, or you're using the port Hermes printed in a previous run (the port can be auto-bumped if the preferred one is busy — always read the latest `Waiting for callback on ...` line).
### `xAI authorization timed out waiting for the local callback`
Same root cause as above — the redirect never made it back. Check the tunnel is still alive (`ssh -N` doesn't show output, so look at the terminal you started it from), restart it if needed, and re-run `hermes auth add xai-oauth --no-browser`.
The redirect never made it back to the remote listener. Check the tunnel is still alive (`ssh -N` doesn't show output, so look at the terminal you started it from), confirm you used the port from the latest `Waiting for callback on ...` line (Hermes may auto-bump if the preferred port is busy), restart the tunnel if needed, and re-run the auth command.
### Tokens land in the wrong `~/.hermes`
@ -181,7 +157,7 @@ The tokens are written under the Linux user that ran `hermes auth add ...`. If y
## See Also
- [xAI Grok OAuth](./xai-grok-oauth.md)
- [xAI Grok OAuth](./xai-grok-oauth.md) — device code; no SSH tunnel
- [Spotify (`Running over SSH`)](../user-guide/features/spotify.md#running-over-ssh--in-a-headless-environment)
- [Native MCP client (OAuth section)](../user-guide/features/mcp.md#oauth-authenticated-http-servers)
- [SSH `-J` / ProxyJump (man page)](https://man.openbsd.org/ssh#J)

View file

@ -47,8 +47,8 @@ OAuth needs a browser, but the loopback callback runs on the machine where Herme
ssh -N -L 8642:127.0.0.1:8642 user@remote-host # in a local terminal
hermes setup --portal # on the remote, open the printed URL in your local browser
# Option B: manual paste (for Cloud Shell, Codespaces, EC2 Instance Connect)
hermes auth add nous --type oauth --manual-paste
# Option B: device-code login (works from Cloud Shell, Codespaces, EC2 Instance Connect)
hermes auth add nous --type oauth
# Then re-run `hermes setup --portal` to wire the provider + gateway
```
@ -186,7 +186,7 @@ The OAuth flow didn't complete. Re-run it:
hermes portal
```
If your browser doesn't open or the callback fails, you're likely on a remote/headless host — see [OAuth over SSH](/guides/oauth-over-ssh) for the port-forwarding and manual-paste workarounds.
If your browser doesn't open or the callback fails, you're likely on a remote/headless host — see [OAuth over SSH](/guides/oauth-over-ssh) for the port-forwarding workarounds.
### "Model: currently openrouter" (or some other provider) instead of "using Nous as inference provider"

View file

@ -113,7 +113,7 @@ Already set up with another model?
- **Don't see the model in the list?** Make sure you finished the Nous Portal connection and that you're on the **Free** plan. In the CLI, `hermes portal info` confirms you're logged in and routing through Nous.
- **Picked the wrong variant?** Re-select `nvidia/nemotron-3-ultra:free` — the `:free` suffix is required to stay on the no-cost tier.
- **Browser didn't open / you're on a remote host (CLI)?** See [OAuth over SSH / Remote Hosts](/guides/oauth-over-ssh) for port-forwarding and manual-paste workarounds.
- **Browser didn't open / you're on a remote host (CLI)?** See [OAuth over SSH / Remote Hosts](/guides/oauth-over-ssh) for port-forwarding workarounds.
## See also

View file

@ -6,7 +6,7 @@ description: "Sign in with your SuperGrok or X Premium+ subscription to use Grok
# xAI Grok OAuth (SuperGrok / X Premium+)
Hermes Agent supports xAI Grok through a browser-based OAuth login flow against [accounts.x.ai](https://accounts.x.ai), using either a **SuperGrok subscription** ([grok.com](https://x.ai/grok)) or an **X Premium+ subscription** (linked X account). No `XAI_API_KEY` is required — log in once and Hermes automatically refreshes your session in the background.
Hermes Agent supports xAI Grok through a browser-based OAuth device-code login flow against [accounts.x.ai](https://accounts.x.ai), using either a **SuperGrok subscription** ([grok.com](https://x.ai/grok)) or an **X Premium+ subscription** (linked X account). No `XAI_API_KEY` is required — log in once and Hermes automatically refreshes your session in the background.
When you sign in with an X account that has Premium+, xAI automatically links the subscription status to your xAI session, so the OAuth flow works the same as it does for direct SuperGrok subscribers.
@ -20,7 +20,7 @@ The same OAuth bearer token is also reused by every direct-to-xAI surface in Her
|------|-------|
| Provider ID | `xai-oauth` |
| Display name | xAI Grok OAuth (SuperGrok / X Premium+) |
| Auth type | Browser OAuth 2.0 PKCE (loopback callback) |
| Auth type | Browser OAuth 2.0 device code |
| Transport | xAI Responses API (`codex_responses`) |
| Default model | `grok-build-0.1` |
| Endpoint | `https://api.x.ai/v1` |
@ -33,7 +33,7 @@ The same OAuth bearer token is also reused by every direct-to-xAI surface in Her
- Python 3.9+
- Hermes Agent installed
- An active **SuperGrok** subscription on your xAI account, **or** an **X Premium+** subscription on the X account you sign in with (xAI links the subscription automatically)
- A browser available on the local machine (or use `--no-browser` for remote sessions)
- A browser available anywhere you can open the printed verification URL
:::warning xAI may restrict OAuth API access by tier
xAI's backend enforces its own allowlist on the OAuth API surface and has been seen to reject standard SuperGrok subscribers with `HTTP 403` (see issue [#26847](https://github.com/NousResearch/hermes-agent/issues/26847)) even though the in-app subscription is active. If OAuth login succeeds in the browser but inference returns 403, set `XAI_API_KEY` and switch to the API-key path (`provider: xai`) — that surface is not subject to the same gating today.
@ -45,8 +45,8 @@ xAI's backend enforces its own allowlist on the OAuth API surface and has been s
# Launch the provider and model picker
hermes model
# → Select "xAI Grok OAuth (SuperGrok / X Premium+)" from the provider list
# → Hermes opens your browser to accounts.x.ai
# → Approve access in the browser
# → Hermes opens or prints an accounts.x.ai verification URL
# → Enter the displayed code if prompted, then approve access in the browser
# → Pick a model (grok-build-0.1 is at the top)
# → Start chatting
@ -65,42 +65,20 @@ hermes auth add xai-oauth
### Remote / headless sessions
On servers, containers, or SSH sessions where no browser is available, Hermes detects the remote environment and prints the authorization URL instead of opening a browser.
**Important:** the loopback listener still runs on the remote machine at `127.0.0.1:56121`. The xAI redirect needs to reach *that* listener, so opening the URL on your laptop will fail (`Could not establish connection. We couldn't reach your app.`) unless you forward the port:
On servers, containers, browser-only consoles (Cloud Shell, Codespaces, EC2 Instance Connect), or SSH sessions where Hermes cannot open a browser locally, Hermes prints the xAI verification URL and user code. Open the URL in any browser on your laptop or in the cloud console, enter the code if prompted, and Hermes will keep polling until xAI approves the login. No SSH tunnel or local callback listener is required.
```bash
# In a separate terminal on your local machine:
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
# Then in your SSH session on the remote machine:
hermes auth add xai-oauth --no-browser
# Open the printed authorize URL in your local browser.
# Open the printed verification URL in your browser.
```
Through a jump box / bastion: add `-J jump-user@jump-host`.
See [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md) for the full step-by-step, including ProxyJump chains, mosh/tmux, and ControlMaster gotchas.
### Browser-only remotes (Cloud Shell, Codespaces, EC2 Instance Connect)
If you don't have a regular SSH client (e.g. you're running Hermes inside GCP Cloud Shell, GitHub Codespaces, AWS EC2 Instance Connect, Gitpod, or another browser-based console), the `ssh -L` recipe above isn't available. Use `--manual-paste` instead — Hermes skips the loopback listener and lets you paste the failed callback URL straight from your browser:
```bash
hermes auth add xai-oauth --manual-paste
# Or via the model picker:
hermes model --manual-paste
```
See [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md#browser-only-remote-cloud-shell--codespaces--ec2-instance-connect) for the full walkthrough. Regression fix for [#26923](https://github.com/NousResearch/hermes-agent/issues/26923).
If the consent page renders the authorization code directly on the page (xAI's current behavior on browser-based consoles) instead of redirecting to your `127.0.0.1:56121/callback`, paste **just the bare code value** at the `Callback URL:` prompt — Hermes accepts the full URL, a bare `?code=...&state=...` query fragment, or a bare code interchangeably.
The same device-code flow applies when you sign in from the web dashboard or the desktop app: Hermes shows the verification URL and user code, then polls in the background until you approve access.
## How the Login Works
1. Hermes opens your browser to `accounts.x.ai`.
2. You sign in (or confirm your existing session) and approve access.
3. xAI redirects back to Hermes and the tokens are saved to `~/.hermes/auth.json`.
1. Hermes requests a device code from `auth.x.ai`.
2. You open the verification URL, sign in, enter the displayed code if prompted, and approve access.
3. Hermes polls xAI until approval, then saves tokens to `~/.hermes/auth.json`.
4. From then on, Hermes refreshes the access token in the background — you stay signed in until you `hermes auth logout xai-oauth` or revoke access from your xAI account settings.
## Checking Login Status
@ -209,29 +187,19 @@ When the refresh failure is terminal (HTTP 4xx, `invalid_grant`, revoked grant,
### Authorization timed out
The loopback listener has a finite expiry window (default 180 s). If you don't approve the login in time, Hermes raises a timeout error.
Device-code approval has a finite expiry window (xAI sets `expires_in` on the device-code response, typically on the order of tens of minutes). If you do not approve the login in time, Hermes raises a timeout error.
**Fix:** re-run `hermes auth add xai-oauth` (or `hermes model`). The flow starts fresh.
### State mismatch (possible CSRF)
Hermes detected that the `state` value returned by the authorization server doesn't match what it sent.
**Fix:** re-run the login. If it persists, check for a proxy or redirect that is modifying the OAuth response.
### Logging in from a remote server
On SSH or container sessions Hermes prints the authorization URL instead of opening a browser. The loopback callback listener still binds `127.0.0.1:56121` on the remote host — your laptop's browser can't reach it without an SSH local-forward:
On SSH or container sessions Hermes prints the verification URL and user code instead of opening a browser. Open that URL in a browser on your laptop or in a cloud console — no SSH port forward is needed for xAI Grok OAuth.
```bash
# Local machine, separate terminal:
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
# Remote machine:
hermes auth add xai-oauth --no-browser
```
Full walkthrough (jump boxes, mosh/tmux, port conflicts): [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md).
For loopback-redirect providers (Spotify, MCP servers), see [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md).
### HTTP 403 after a successful login (tier / entitlement)
@ -266,7 +234,7 @@ This clears both the singleton OAuth entry in `auth.json` and any credential-poo
## See Also
- [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md) — required reading if Hermes is on a different machine than your browser
- [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md) — SSH tunnels for loopback-redirect providers (Spotify, MCP); xAI uses device code and does not need a tunnel
- [AI Providers reference](../integrations/providers.md)
- [Environment Variables](../reference/environment-variables.md)
- [Configuration](../user-guide/configuration.md)

View file

@ -120,7 +120,7 @@ Your existing providers stay configured. You can switch between them with `/mode
### Headless / SSH / remote setup
OAuth needs a browser, but the loopback callback runs on the machine where Hermes is running. For remote hosts, see [OAuth over SSH / Remote Hosts](/guides/oauth-over-ssh) — the same patterns work for the Portal as for any other OAuth-based provider (`ssh -L` port forwarding, `--manual-paste` for browser-only environments like Cloud Shell / Codespaces).
OAuth needs a browser, but the loopback callback runs on the machine where Hermes is running. For remote hosts, see [OAuth over SSH / Remote Hosts](/guides/oauth-over-ssh) — the same patterns work for the Portal as for any other OAuth-based provider (`ssh -L` port forwarding).
### Profile setup

View file

@ -1,55 +1,40 @@
---
sidebar_position: 17
title: "SSH / 远程主机上的 OAuth"
description: "当 Hermes 运行在远程机器、容器或跳板机后面时,如何完成基于浏览器的 OAuthxAI、Spotify"
description: "当 Hermes 运行在远程机器、容器或跳板机后面时,如何完成基于浏览器的 OAuthSpotify、MCP 服务器"
---
# SSH / 远程主机上的 OAuth
部分 Hermes 提供商——目前是 **xAI Grok OAuth****Spotify**——使用*回环重定向loopback redirect* OAuth 流程。认证服务器xAI、Spotify将浏览器重定向到 `http://127.0.0.1:<port>/callback`,由 `hermes auth ...` 命令启动的一个小型 HTTP 监听器来获取授权码。
部分 Hermes 提供商——**Spotify** 和 **远程 MCP 服务器**Linear、Sentry、Atlassian、Asana、Figma 等)——使用*回环重定向loopback redirect* OAuth 流程。认证服务器将浏览器重定向到 `http://127.0.0.1:<port>/callback`,由 Hermes 启动的小型 HTTP 监听器获取授权码。
当 Hermes 和浏览器在同一台机器上时,这一切运行正常。一旦两者不在同一台机器上就会出问题:你笔记本上的浏览器试图访问**你笔记本**上的 `127.0.0.1`,但监听器绑定的是**远程服务器**上的 `127.0.0.1`
解决方法是一行 SSH 本地端口转发——**或者**,当你没有真正的 SSH 客户端时GCP Cloud Shell、GitHub Codespaces、EC2 Instance Connect、Gitpod、基于浏览器的 Web IDE使用 [#26923](https://github.com/NousResearch/hermes-agent/issues/26923) 中引入的新 `--manual-paste` 标志。
解决方法是一行 SSH 本地端口转发。对于交互式终端上的 MCP 服务器,通常也可以直接粘贴重定向 URL无需隧道
**xAI Grok OAuth`xai-oauth`)使用 OAuth 设备代码**,不是回环回调——在任意浏览器中打开打印的验证 URLHermes 轮询直到批准即可,无需 SSH 隧道。请参阅 [xAI Grok OAuth](./xai-grok-oauth.md)。
## 快速概览
```bash
# 在你的本地机器(笔记本)上,另开一个终端:
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
ssh -N -L 43827:127.0.0.1:43827 user@remote-host
# 在远程机器的现有 SSH 会话中:
hermes auth add xai-oauth --no-browser
# → Hermes 打印一个授权 URL在笔记本的浏览器中打开它。
# → 浏览器重定向到 127.0.0.1:56121/callback隧道将请求转发
# 到远程监听器,登录完成。
hermes auth add spotify --no-browser
# → Hermes 打印授权 URL在笔记本的浏览器中打开。
# → 浏览器重定向到 127.0.0.1:43827/callback隧道转发到远程监听器登录完成。
```
`56121` 是 xAI OAuth 使用的端口。Spotify 请将其替换为 `43827`。Hermes 会在 `Waiting for callback on ...` 这一行打印它实际绑定的端口——从那里复制。
## 仅限浏览器的远程环境Cloud Shell / Codespaces / EC2 Instance Connect
如果你没有常规的 SSH 客户端——例如你在 GCP Cloud Shell、GitHub Codespaces、AWS EC2 Instance Connect、Gitpod 或其他基于浏览器的控制台中运行 Hermes——上述 SSH 隧道不可用。请改用 `--manual-paste`
```bash
hermes auth add xai-oauth --manual-paste
# → Hermes 打印一个授权 URL在笔记本的浏览器中打开它。
# → 在浏览器中批准。重定向到 127.0.0.1:56121/callback 会加载失败
# ——这是预期行为。
# → 从失败页面的地址栏复制完整 URL。
# → 在终端的 "Callback URL:" 提示处粘贴。
```
同样的标志也适用于集成模型选择器的 `hermes model --manual-paste`。如果不想粘贴完整 URL也可以只接受裸的 `?code=...&state=...` 查询片段。
Hermes 对两种路径使用**相同的 PKCE verifier、state 和 nonce**,因此上游 OAuth 流程在字节层面完全一致——`--manual-paste` 纯粹是回调跳转的传输方式变更,不会降低安全性。
Hermes 会在 `Waiting for callback on ...` 一行打印实际绑定的端口——从那里复制。Spotify 默认端口为 `43827`
## 哪些提供商需要此操作
| 提供商 | 回环端口 | 需要隧道? |
|----------|---------------|----------------|
| `xai-oauth`Grok SuperGrok | `56121` | 是,当 Hermes 在远程时 |
| Spotify | `43827` | 是,当 Hermes 在远程时 |
| Spotify | `43827`(默认) | 是,当 Hermes 在远程时 |
| MCP 服务器(`auth: oauth` | 每台服务器自动选择 | 是(或粘贴重定向 URL |
| `xai-oauth`Grok SuperGrok | 不适用 | 否——设备代码流程 |
| `anthropic`Claude Pro/Max | 不适用 | 否——粘贴代码流程 |
| `openai-codex`ChatGPT Plus/Pro | 不适用 | 否——设备码流程 |
| `minimax``nous-portal` | 不适用 | 否——设备码流程 |
@ -58,97 +43,54 @@ Hermes 对两种路径使用**相同的 PKCE verifier、state 和 nonce**,因
## 为什么监听器不能直接绑定 0.0.0.0
xAI 和 Spotify 都会根据白名单验证 `redirect_uri` 参数。两者都要求回环形式(`http://127.0.0.1:<exact-port>/callback`)。将监听器绑定到 `0.0.0.0` 或不同端口会导致认证服务器以 redirect_uri 不匹配为由拒绝请求。SSH 隧道可以端到端保持回环 URI 不变。
Spotify 和大多数 MCP OAuth 服务器会根据白名单验证 `redirect_uri` 参数,并要求回环形式(`http://127.0.0.1:<精确端口>/callback`)。将监听器绑定到 `0.0.0.0`使用不同端口会导致认证服务器以 redirect_uri 不匹配为由拒绝请求。SSH 隧道可以端到端保持回环 URI 不变。
## 分步说明:单跳 SSH
## 分步操作:单次 SSH 跳转
### 1. 从本地机器启动隧道
```bash
# xAI Grok OAuth端口 56121
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
# 或 Spotify端口 43827
# Spotify端口 43827
ssh -N -L 43827:127.0.0.1:43827 user@remote-host
```
`-N` 表示"不打开远程 shell只保持隧道开启"。在登录期间保持此终端运行。
`-N` 表示「不打开远程 shell仅保持隧道」。登录期间保持此终端运行。
### 2. 在另一个 SSH 会话中运行认证命令
```bash
ssh user@remote-host
hermes auth add xai-oauth --no-browser
# 或 Spotify
# hermes auth add spotify --no-browser
hermes auth add spotify --no-browser
```
Hermes 检测到 SSH 会话,跳过自动打开浏览器,打印授权 URL 以及 `Waiting for callback on http://127.0.0.1:<port>/callback` 这一行
Hermes 检测到 SSH 会话,跳过自动打开浏览器,打印授权 URL 以及 `Waiting for callback on http://127.0.0.1:<port>/callback`
### 3. 在本地浏览器中打开 URL
从远程终端复制授权 URL粘贴到笔记本的浏览器中。批准同意页面。认证服务器重定向到 `http://127.0.0.1:<port>/callback`。浏览器访问隧道,请求转发到远程监听器Hermes 打印 `Login successful!`
从远程终端复制授权 URL粘贴到笔记本的浏览器中。批准同意后,认证服务器重定向到 `http://127.0.0.1:<port>/callback`。浏览器经隧道访问请求转发到远程监听器Hermes 打印 `Login successful!`
看到成功提示后,可以关闭隧道(在第一个终端按 Ctrl+C
看到成功提示后即可关闭隧道(在第一个终端按 Ctrl+C
## 分步说明:通过跳板机
## 通过跳板机
如果通过堡垒机 / 跳板机访问 Hermes使用 SSH 内置的 `-J`ProxyJump
如果通过堡垒机 / 跳板机访问 Hermes使用 SSH 内置的 `-J`ProxyJump
```bash
ssh -N -L 56121:127.0.0.1:56121 -J jump-user@jump-host user@final-host
ssh -N -L 43827:127.0.0.1:43827 -J jump-user@jump-host user@final-host
```
这会通过跳板机链式建立 SSH 连接,而不会将回环端口暴露在跳板机上。你笔记本上的本地 `127.0.0.1:56121` 直接隧道到最终远程主机上的 `127.0.0.1:56121`
## 故障排除
对于不支持 `-J` 的旧版 OpenSSH完整写法为
### `bind [127.0.0.1]:43827: Address already in use`
```bash
ssh -N \
-o "ProxyCommand=ssh -W %h:%p jump-user@jump-host" \
-L 56121:127.0.0.1:56121 \
user@final-host
```
笔记本上已有进程占用该端口。结束占用进程后重试 `ssh -L`
## Mosh、tmux、ssh ControlMaster
### 等待本地回调超时
隧道是底层 SSH 连接的属性。如果你在 mosh 会话中的 `tmux` 里运行 Hermesmosh 的漫游不会携带 `-L` 转发。**单独**开一个普通 SSH 会话**仅用于** `-L` 隧道——这个连接必须在整个认证流程期间保持存活。你的交互式 mosh/tmux 会话可以继续正常运行 Hermes。
如果你使用 `ssh -o ControlMaster=auto`,多路复用连接上的端口转发共享主连接的生命周期。如果隧道未能建立,重启主连接:
```bash
ssh -O exit user@remote-host
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
```
## 故障排查
### `bind [127.0.0.1]:56121: Address already in use`
你笔记本上已有某个程序占用了该端口。可能是上一个隧道没有正常关闭,或者本地也有一个 Hermes 在监听。找到并终止占用进程:
```bash
# macOS / Linux
lsof -iTCP:56121 -sTCP:LISTEN
kill <PID>
```
然后重试 `ssh -L` 命令。
### "Could not establish connection. We couldn't reach your app."xAI
当 xAI 重定向到 `127.0.0.1:<port>/callback` 未能到达监听器时xAI 的授权页面会显示此错误。可能是隧道未运行、端口错误,或者你使用的是 Hermes 上一次运行时打印的端口(如果首选端口被占用,端口可能会自动递增——始终以最新的 `Waiting for callback on ...` 行为准)。
### `xAI authorization timed out waiting for the local callback`
与上述原因相同——重定向从未返回。检查隧道是否仍然存活(`ssh -N` 不显示输出,查看启动它的终端),必要时重启,然后重新运行 `hermes auth add xai-oauth --no-browser`
### Token 写入了错误的 `~/.hermes`
Token 写入运行 `hermes auth add ...` 的 Linux 用户目录下。如果你的网关 / systemd 服务以不同用户(如 `root` 或专用的 `hermes` 用户)运行,请以**该**用户身份进行认证,使 token 写入其 `~/.hermes/auth.json`。使用 `sudo -u hermes -i` 或等效命令。
重定向未到达远程监听器。确认隧道仍在运行,并使用最新一次 `Waiting for callback on ...` 中的端口(首选端口被占用时 Hermes 可能自动递增)。
## 另请参阅
- [xAI Grok OAuth](./xai-grok-oauth.md)
- [Spotify`通过 SSH 运行`](../user-guide/features/spotify.md#running-over-ssh--in-a-headless-environment)
- [SSH `-J` / ProxyJumpman 手册)](https://man.openbsd.org/ssh#J)
- [xAI Grok OAuth](./xai-grok-oauth.md)——设备代码;无需 SSH 隧道
- [SpotifySSH 上运行)](../user-guide/features/spotify.md#running-over-ssh--in-a-headless-environment)
- [原生 MCP 客户端OAuth 部分)](../user-guide/features/mcp.md#oauth-authenticated-http-servers)

View file

@ -47,8 +47,8 @@ OAuth 需要浏览器,但 loopback 回调运行在 Hermes 所在的机器上
ssh -N -L 8642:127.0.0.1:8642 user@remote-host # 在本地终端执行
hermes setup --portal # 在远程机器上执行,在本地浏览器中打开打印出的 URL
# 方案 B手动粘贴(适用于 Cloud Shell、Codespaces、EC2 Instance Connect
hermes auth add nous --type oauth --manual-paste
# 方案 B设备码登录(适用于 Cloud Shell、Codespaces、EC2 Instance Connect
hermes auth add nous --type oauth
# 然后重新运行 `hermes setup --portal` 以连接 provider + gateway
```
@ -183,7 +183,7 @@ OAuth 流程未完成。重新运行:
hermes portal
```
如果浏览器未打开或回调失败,你可能在远程/无头主机上——参见 [OAuth over SSH](/guides/oauth-over-ssh) 了解端口转发和手动粘贴的解决方案。
如果浏览器未打开或回调失败,你可能在远程/无头主机上——参见 [OAuth over SSH](/guides/oauth-over-ssh) 了解端口转发的解决方案。
### "Model: currently openrouter"(或其他 provider而非"using Nous as inference provider"

View file

@ -20,7 +20,7 @@ Hermes Agent 通过基于浏览器的 OAuth 登录流程支持 xAI Grok认证
|------|-------|
| Provider ID | `xai-oauth` |
| 显示名称 | xAI Grok OAuth (SuperGrok / X Premium+) |
| 认证类型 | 浏览器 OAuth 2.0 PKCE回环回调 |
| 认证类型 | 浏览器 OAuth 2.0 设备代码 |
| 传输层 | xAI Responses API`codex_responses` |
| 默认模型 | `grok-build-0.1` |
| 端点 | `https://api.x.ai/v1` |
@ -33,7 +33,7 @@ Hermes Agent 通过基于浏览器的 OAuth 登录流程支持 xAI Grok认证
- Python 3.9+
- 已安装 Hermes Agent
- 你的 xAI 账号拥有有效的 **SuperGrok** 订阅,**或**你登录所用的 X 账号拥有 **X Premium+** 订阅xAI 会自动关联订阅)
- 本地机器上有可用的浏览器(远程会话可使用 `--no-browser`
- 任意可打开打印出的验证 URL 的浏览器
:::warning xAI 可能按套餐限制 OAuth API 访问
xAI 的后端对 OAuth API 接口维护自己的白名单,已有记录显示即使应用内订阅处于激活状态,标准 SuperGrok 订阅者也会收到 `HTTP 403`(见 issue [#26847](https://github.com/NousResearch/hermes-agent/issues/26847))。如果浏览器中 OAuth 登录成功但推理返回 403请设置 `XAI_API_KEY` 并切换到 API 密钥路径(`provider: xai`)——该接口目前不受相同限制。
@ -45,8 +45,8 @@ xAI 的后端对 OAuth API 接口维护自己的白名单,已有记录显示
# 启动 provider 和模型选择器
hermes model
# → 从 provider 列表中选择 "xAI Grok OAuth (SuperGrok / X Premium+)"
# → Hermes 在浏览器中打开 accounts.x.ai
# → 在浏览器中批准访问
# → Hermes 打开或打印 accounts.x.ai 验证 URL
# → 如有提示,输入显示的代码,然后在浏览器中批准访问
# → 选择模型grok-build-0.1 在列表顶部)
# → 开始对话
@ -65,40 +65,20 @@ hermes auth add xai-oauth
### 远程 / 无头会话
在没有浏览器的服务器、容器或 SSH 会话中Hermes 会检测到远程环境并打印授权 URL而不是打开浏览器。
**重要:** 回环监听器仍在远程机器的 `127.0.0.1:56121` 上运行。xAI 的重定向需要到达*该*监听器,因此在你的笔记本上打开 URL 会失败(`Could not establish connection. We couldn't reach your app.`),除非你转发端口:
在没有浏览器的服务器、容器、仅限浏览器的远程控制台Cloud Shell、Codespaces、EC2 Instance Connect或 SSH 会话中Hermes 会打印 xAI 验证 URL 和用户代码。在笔记本电脑或云控制台的任意浏览器中打开该 URL如有提示则输入代码Hermes 会持续轮询直到 xAI 批准登录。无需 SSH 隧道或本地回调监听器。
```bash
# 在本地机器的另一个终端中:
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
# 然后在远程机器的 SSH 会话中:
hermes auth add xai-oauth --no-browser
# 在本地浏览器中打开打印出的授权 URL。
# 在浏览器中打开打印出的验证 URL。
```
通过跳板机 / 堡垒机:添加 `-J jump-user@jump-host`
完整步骤(包括 ProxyJump 链、mosh/tmux 和 ControlMaster 注意事项)请参阅 [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md)。
### 仅限浏览器的远程环境Cloud Shell、Codespaces、EC2 Instance Connect
如果你没有常规 SSH 客户端(例如在 GCP Cloud Shell、GitHub Codespaces、AWS EC2 Instance Connect、Gitpod 或其他基于浏览器的控制台中运行 Hermes上述 `ssh -L` 方案不可用。请改用 `--manual-paste`——Hermes 跳过回环监听器,让你直接从浏览器粘贴失败的回调 URL
```bash
hermes auth add xai-oauth --manual-paste
# 或通过模型选择器:
hermes model --manual-paste
```
完整操作说明请参阅 [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md#browser-only-remote-cloud-shell--codespaces--ec2-instance-connect)。此为 [#26923](https://github.com/NousResearch/hermes-agent/issues/26923) 的回归修复。
Web 仪表盘和桌面应用使用相同的设备代码流程:显示验证 URL 和用户代码,并在你批准访问后在后台轮询。
## 登录流程说明
1. Hermes 在浏览器中打开 `accounts.x.ai`
2. 你登录(或确认现有会话)并批准访问。
3. xAI 重定向回 Hermestoken 保存到 `~/.hermes/auth.json`
1. Hermes 向 `auth.x.ai` 请求设备代码。
2. 你打开验证 URL登录如有提示则输入显示的代码并批准访问。
3. Hermes 轮询 xAI 直到批准,然后将 token 保存到 `~/.hermes/auth.json`
4. 此后Hermes 在后台刷新 access token——你将保持登录状态直到执行 `hermes auth logout xai-oauth` 或在 xAI 账号设置中撤销访问。
## 检查登录状态
@ -207,29 +187,19 @@ Hermes 在每次会话前刷新 token并在收到 401 时响应式地再次
### 授权超时
回环监听器有有限的过期窗口(默认 180 秒。如果你未在时限内批准登录Hermes 会抛出超时错误。
设备代码批准有有限的过期窗口xAI 在设备代码响应中设置 `expires_in`,通常为数十分钟量级。如果你未在时限内批准登录Hermes 会抛出超时错误。
**修复方法:** 重新运行 `hermes auth add xai-oauth`(或 `hermes model`)。流程重新开始。
### State 不匹配(可能的 CSRF
Hermes 检测到授权服务器返回的 `state` 值与发送的不匹配。
**修复方法:** 重新运行登录。如果问题持续,检查是否有代理或重定向在修改 OAuth 响应。
### 从远程服务器登录
在 SSH 或容器会话中Hermes 打印授权 URL 而不是打开浏览器。回环回调监听器仍绑定在远程主机的 `127.0.0.1:56121`——你笔记本上的浏览器无法访问它,除非进行 SSH 本地端口转发:
在 SSH 或容器会话中Hermes 打印验证 URL 和用户代码,而不是打开浏览器。在笔记本电脑或云控制台的浏览器中打开该 URL——xAI Grok OAuth 无需 SSH 端口转发。
```bash
# 本地机器,另一个终端:
ssh -N -L 56121:127.0.0.1:56121 user@remote-host
# 远程机器:
hermes auth add xai-oauth --no-browser
```
完整操作说明跳板机、mosh/tmux、端口冲突[OAuth over SSH / Remote Hosts](./oauth-over-ssh.md)。
回环重定向类 providerSpotify、MCP 服务器)请参阅 [OAuth over SSH / Remote Hosts](./oauth-over-ssh.md)。
### 登录成功后 HTTP 403套餐 / 权限问题)

View file

@ -116,7 +116,7 @@ hermes model
### 无头环境 / SSH / 远程配置
OAuth 需要浏览器,但回调的 loopback 运行在 Hermes 所在的机器上。对于远程主机,请参阅 [OAuth over SSH / 远程主机](/guides/oauth-over-ssh)——与其他基于 OAuth 的提供商相同的方式同样适用于 Portal`ssh -L` 端口转发,或在 Cloud Shell / Codespaces 等纯浏览器环境中使用 `--manual-paste`)。
OAuth 需要浏览器,但回调的 loopback 运行在 Hermes 所在的机器上。对于远程主机,请参阅 [OAuth over SSH / 远程主机](/guides/oauth-over-ssh)——与其他基于 OAuth 的提供商相同的方式同样适用于 Portal`ssh -L` 端口转发)。
### Profile 配置