fix(xai): pin oauth side-tool base URLs

This commit is contained in:
dsad 2026-07-11 01:00:29 +03:00 committed by Teknium
parent 0d0ad3f9d9
commit f96eee3a91
4 changed files with 89 additions and 6 deletions

View file

@ -1699,6 +1699,38 @@ class TestTranscribeXAI:
url = call_args[0][0] if call_args[0] else call_args.kwargs.get("url", "")
assert "custom.x.ai" in url
def test_oauth_credentials_ignore_stt_base_url_override(
self,
monkeypatch,
sample_ogg,
mock_xai_http_module,
):
monkeypatch.delenv("XAI_API_KEY", raising=False)
monkeypatch.setenv("XAI_STT_BASE_URL", "https://attacker.example/v1")
mock_xai_http_module.resolve_xai_http_credentials.return_value = {
"provider": "xai-oauth",
"api_key": "oauth-bearer-token",
"base_url": "https://api.x.ai/v1",
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"text": "test", "language": "en", "duration": 1.0}
with patch(
"tools.transcription_tools._load_stt_config",
return_value={"xai": {"base_url": "https://attacker.example/config"}},
), patch("requests.post", return_value=mock_response) as mock_post:
from tools.transcription_tools import _transcribe_xai
result = _transcribe_xai(sample_ogg, "grok-stt")
assert result["success"] is True
call_args = mock_post.call_args
url = call_args[0][0] if call_args[0] else call_args.kwargs.get("url", "")
assert url == "https://api.x.ai/v1/stt"
assert call_args.kwargs["headers"]["Authorization"] == "Bearer oauth-bearer-token"
def test_diarize_sent_when_configured(self, monkeypatch, sample_ogg, mock_xai_http_module):
monkeypatch.setenv("XAI_API_KEY", "xai-test-key")

View file

@ -132,6 +132,48 @@ def test_generate_xai_tts_sends_auxiliary_rewriter_output_to_api(
assert captured["json"]["text"] == rewriter_output
def test_generate_xai_tts_uses_oauth_pinned_base_url(tmp_path, monkeypatch):
"""OAuth bearer tokens must not follow user/env base URL overrides."""
captured = {}
class FakeResponse:
content = b"mp3"
def raise_for_status(self):
pass
def fake_post(url, headers, json, timeout):
captured["url"] = url
captured["headers"] = headers
return FakeResponse()
monkeypatch.setenv("XAI_BASE_URL", "https://attacker.example/v1")
monkeypatch.setattr(
"tools.xai_http.resolve_xai_http_credentials",
lambda: {
"provider": "xai-oauth",
"api_key": "oauth-bearer-token",
"base_url": "https://api.x.ai/v1",
},
)
monkeypatch.setattr("requests.post", fake_post)
out = tmp_path / "out.mp3"
_generate_xai_tts(
"Bonjour.",
str(out),
{
"xai": {
"base_url": "https://attacker.example/config",
"auto_speech_tags": False,
}
},
)
assert captured["url"] == "https://api.x.ai/v1/tts"
assert captured["headers"]["Authorization"] == "Bearer oauth-bearer-token"
def test_auto_speech_tags_calls_auxiliary_rewriter_with_tts_audio_tags_task():
"""When input has no explicit speech tags, the function must call the
auxiliary rewriter with task='tts_audio_tags' and a system prompt

View file

@ -1821,6 +1821,12 @@ def _transcribe_xai(file_path: str, model_name: str) -> Dict[str, Any]:
xai_config = stt_config.get("xai") or {}
def _resolve_base_url(resolved_creds: Dict[str, str]) -> str:
# OAuth bearers are pinned to the resolver-validated xAI origin;
# config/env base URL overrides only apply to API-key credentials.
if resolved_creds.get("provider") == "xai-oauth":
return str(
resolved_creds.get("base_url") or XAI_STT_BASE_URL
).strip().rstrip("/")
return str(
xai_config.get("base_url")
or get_env_value("XAI_STT_BASE_URL")

View file

@ -1481,12 +1481,15 @@ def _generate_xai_tts(text: str, output_path: str, tts_config: Dict[str, Any]) -
optimize_streaming_latency = max(0, min(2, optimize_streaming_latency))
if auto_speech_tags:
text = _apply_xai_auto_speech_tags(text)
base_url = str(
xai_config.get("base_url")
or creds.get("base_url")
or get_env_value("XAI_BASE_URL")
or DEFAULT_XAI_BASE_URL
).strip().rstrip("/")
if creds.get("provider") == "xai-oauth":
base_url = str(creds.get("base_url") or DEFAULT_XAI_BASE_URL).strip().rstrip("/")
else:
base_url = str(
xai_config.get("base_url")
or creds.get("base_url")
or get_env_value("XAI_BASE_URL")
or DEFAULT_XAI_BASE_URL
).strip().rstrip("/")
# Match the documented minimal POST /v1/tts shape by default. Only send
# output_format when Hermes actually needs a non-default format/override.