From 3860872d694987f2d6d3f47844018d6d3b0dda34 Mon Sep 17 00:00:00 2001 From: moikapy Date: Sun, 24 May 2026 10:58:55 -0400 Subject: [PATCH] fix(auxiliary): detect xAI OAuth 403 bad-credentials as auth error xAI returns HTTP 403 (not 401) with unauthenticated:bad-credentials when an OAuth2 access token has expired or is invalid. The existing _is_auth_error() only checked for 401 status codes, so these tokens were never refreshed and the 403 propagated as a generic permission denied error. Three fixes: 1. _is_auth_error: Recognize xAI's 403+bad-credentials pattern as an auth failure, triggering token refresh instead of silent failure. 2. _refresh_provider_credentials: Add xai-oauth branch with pool-level refresh (try_refresh_current with select to ensure current entry) then fallback to singleton resolver with force_refresh=True. 3. _recoverable_pool_provider: Map api.x.ai host to xai-oauth pool for auto-resolved providers, matching existing pattern for openai-codex/openrouter/nous/anthropic. Includes 14 tests covering the new detection logic, host mapping, and graceful fallback behavior. Signed-off-by: moikapy --- agent/auxiliary_client.py | 31 +++- ...est_auxiliary_client_xai_oauth_recovery.py | 153 ++++++++++++++++++ 2 files changed, 183 insertions(+), 1 deletion(-) create mode 100644 tests/agent/test_auxiliary_client_xai_oauth_recovery.py diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 40044a1062e6a..995daf2c264c9 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -2347,7 +2347,16 @@ def _is_auth_error(exc: Exception) -> bool: if status == 401: return True err_lower = str(exc).lower() - return "error code: 401" in err_lower or "authenticationerror" in type(exc).__name__.lower() + if "error code: 401" in err_lower or "authenticationerror" in type(exc).__name__.lower(): + return True + # xAI returns HTTP 403 with "unauthenticated:bad-credentials" when an OAuth2 + # access token has expired or is invalid — semantically a 401 auth failure, + # even though the status code is 403 (PermissionDenied). + if status == 403 and "bad-credentials" in err_lower: + return True + if "unauthenticated" in err_lower and "bad-credentials" in err_lower: + return True + return False def _is_unsupported_parameter_error(exc: Exception, param: str) -> bool: @@ -2496,6 +2505,8 @@ def _recoverable_pool_provider(resolved_provider: str, client: Any) -> Optional[ return "copilot" if base_url_host_matches(base, "api.kimi.com"): return "kimi-coding" + if base_url_host_matches(base, "api.x.ai"): + return "xai-oauth" return None @@ -2692,6 +2703,24 @@ def _refresh_provider_credentials(provider: str) -> bool: return False _evict_cached_clients(normalized) return True + if normalized == "xai-oauth": + # Preference: pool-level refresh (uses refresh_token from pool entry), + # then fall back to singleton auth-store resolver. + pool = load_pool(normalized) + if pool and pool.has_credentials(): + # Ensure a current entry is selected before trying to refresh. + pool.select() + refreshed = pool.try_refresh_current() + if refreshed is not None and str(getattr(refreshed, "runtime_api_key", "") or "").strip(): + _evict_cached_clients(normalized) + return True + from hermes_cli.auth import resolve_xai_oauth_runtime_credentials + + creds = resolve_xai_oauth_runtime_credentials(force_refresh=True) + if not str(creds.get("api_key", "") or "").strip(): + return False + _evict_cached_clients(normalized) + return True except Exception as exc: logger.debug("Auxiliary provider credential refresh failed for %s: %s", normalized, exc) return False diff --git a/tests/agent/test_auxiliary_client_xai_oauth_recovery.py b/tests/agent/test_auxiliary_client_xai_oauth_recovery.py new file mode 100644 index 0000000000000..3434a68d80ddf --- /dev/null +++ b/tests/agent/test_auxiliary_client_xai_oauth_recovery.py @@ -0,0 +1,153 @@ +"""Tests for xAI OAuth 403 error recovery in auxiliary_client. + +xAI returns HTTP 403 (not 401) with "unauthenticated:bad-credentials" when +an OAuth2 access token has expired. These tests verify the three fixes: + +1. _is_auth_error detects xAI 403 as an auth failure +2. _recoverable_pool_provider maps api.x.ai to xai-oauth +3. _refresh_provider_credentials includes xai-oauth refresh logic +""" + +import pytest + + +# ── _is_auth_error ────────────────────────────────────────────────────────── + +def _import_is_auth_error(): + from agent.auxiliary_client import _is_auth_error + return _is_auth_error + + +class TestIsAuthErrorXaiOauth403: + """Verify _is_auth_error correctly identifies xAI's 403 bad-credentials.""" + + @pytest.fixture(autouse=True) + def _import(self): + self.is_auth_error = _import_is_auth_error() + + def test_xai_403_bad_credentials_is_auth_error(self): + """The exact error xAI returns for expired OAuth tokens.""" + exc = Exception( + "Error code: 403 - {'code': 'The caller does not have permission " + "to execute the specified operation', 'error': 'The OAuth2 access " + "token could not be validated. [WKE=unauthenticated:bad-credentials]'}" + ) + exc.status_code = 403 # openai.PermissionDenied sets this + assert self.is_auth_error(exc) is True + + def test_xai_403_bad_credentials_without_status_code(self): + """Fallback match when status_code attribute is missing.""" + exc = Exception( + "Error code: 403 - unauthenticated:bad-credentials" + ) + # No status_code attribute — should still match via string pattern + assert self.is_auth_error(exc) is True + + def test_generic_403_is_not_auth_error(self): + """A generic 403 (e.g. rate limit, forbidden) should NOT be treated as auth.""" + exc = Exception("Error code: 403 - rate limit exceeded") + exc.status_code = 403 + assert self.is_auth_error(exc) is False + + def test_401_status_code_is_auth_error(self): + """Existing 401 detection still works.""" + exc = Exception("Unauthorized") + exc.status_code = 401 + assert self.is_auth_error(exc) is True + + def test_401_string_is_auth_error(self): + """Existing string-based 401 detection still works.""" + exc = Exception("Error code: 401 - Unauthorized") + assert self.is_auth_error(exc) is True + + def test_authentication_error_class_is_auth_error(self): + """Existing AuthenticationError class detection still works.""" + exc_type = type("AuthenticationError", (Exception,), {}) + exc = exc_type("auth failure") + assert self.is_auth_error(exc) is True + + def test_permission_denied_without_bad_credentials_is_not_auth_error(self): + """403 PermissionDenied without bad-credentials should not be auth.""" + exc = Exception("Error code: 403 - Permission denied") + exc.status_code = 403 + assert self.is_auth_error(exc) is False + + def test_500_is_not_auth_error(self): + """Server errors are not auth errors.""" + exc = Exception("Error code: 500 - Internal server error") + exc.status_code = 500 + assert self.is_auth_error(exc) is False + + def test_unauthenticated_without_bad_credentials_is_not_auth_error(self): + """'unauthenticated' alone (without 'bad-credentials') should not match.""" + exc = Exception("unauthenticated request") + assert self.is_auth_error(exc) is False + + +# ── _recoverable_pool_provider ────────────────────────────────────────────── + +def _import_recoverable_pool_provider(): + from agent.auxiliary_client import _recoverable_pool_provider + return _recoverable_pool_provider + + +class TestRecoverablePoolProviderXaiOAuth: + """Verify _recoverable_pool_provider maps api.x.ai to xai-oauth.""" + + @pytest.fixture(autouse=True) + def _import(self): + self.recover = _import_recoverable_pool_provider() + + def test_explicit_xai_oauth_provider(self): + """Explicit provider name passes through.""" + result = self.recover("xai-oauth", None) + assert result == "xai-oauth" + + def test_api_x_ai_host_match(self): + """api.x.ai base URL maps to xai-oauth pool.""" + class MockClient: + base_url = "https://api.x.ai/v1/" + + result = self.recover("auto", MockClient()) + assert result == "xai-oauth" + + def test_auto_with_unknown_host_returns_none(self): + """auto provider with unknown host returns None.""" + class MockClient: + base_url = "https://unknown.example.com/v1/" + + result = self.recover("auto", MockClient()) + assert result is None + + +# ── _refresh_provider_credentials (structure check) ───────────────────────── + +def _import_refresh_provider_credentials(): + from agent.auxiliary_client import _refresh_provider_credentials + return _refresh_provider_credentials + + +class TestRefreshProviderCredentialsXaiOAuth: + """Verify _refresh_provider_credentials has xai-oauth branch. + + Full integration testing requires live OAuth tokens, so we verify + the branch exists and handles the no-credential case gracefully. + """ + + @pytest.fixture(autouse=True) + def _import(self): + self.refresh = _import_refresh_provider_credentials() + + def test_xai_oauth_no_pool_returns_false(self): + """When no xai-oauth pool exists, refresh returns False gracefully.""" + # This tests that the branch exists and doesn't crash. + # It may return True if the singleton resolver finds tokens, + # or False if neither pool nor singleton has credentials. + # Either way, it should not raise an exception. + result = self.refresh("xai-oauth") + assert isinstance(result, bool) + + def test_unknown_provider_returns_false(self): + """Unknown providers fall through to return False.""" + result = self.refresh("unknown-provider-xyz") + assert result is False \ No newline at end of file