fix(authenticator): avoid blocking event loop on interactive device-code auth in headless environments

This commit is contained in:
Soumyajit Ghosh 2026-09-26 02:18:10 +05:30
parent 6b7688869e
commit 612e73b6aa
2 changed files with 56 additions and 0 deletions

View file

@ -150,7 +150,27 @@ class Authenticator:
return account_id
return None
def _can_run_interactive_device_login(self) -> bool:
"""Check if the process can run an interactive device-code login.
In headless or proxy environments (e.g. LiteLLM Proxy, Kubernetes, Docker),
synchronous device code polling must never block the event loop.
"""
import sys
return bool(sys.stdin and hasattr(sys.stdin, "isatty") and sys.stdin.isatty())
def _login_device_code(self) -> dict[str, str]:
if not self._can_run_interactive_device_login():
raise GetAccessTokenError(
message=(
"ChatGPT authentication required, but interactive device-code login "
"cannot run in a non-interactive/headless environment (such as the LiteLLM proxy). "
"Please authenticate beforehand using the CLI or provision the auth token file."
),
status_code=401,
)
cooldown_remaining: Final = self._get_device_code_cooldown_remaining(self._read_auth_file())
if cooldown_remaining > 0:
token: Final = self._wait_for_access_token(cooldown_remaining)
@ -372,6 +392,8 @@ class Authenticator:
self._write_auth_file({**auth_data, "device_code_requested_at": time.time()})
def _wait_for_access_token(self, timeout_seconds: float) -> str | None:
if not self._can_run_interactive_device_login():
return None
deadline: Final = time.time() + timeout_seconds
while time.time() < deadline:
auth_data = self._read_auth_file()

View file

@ -6,6 +6,7 @@ from unittest.mock import mock_open, patch
import pytest
from litellm.llms.chatgpt.authenticator import Authenticator
from litellm.llms.chatgpt.common_utils import GetAccessTokenError
def _make_jwt(payload: dict) -> str:
@ -68,3 +69,36 @@ class TestChatGPTAuthenticator:
assert account_id == "acct-123"
mock_write.assert_called_once()
assert mock_write.call_args[0][0]["account_id"] == "acct-123"
def test_can_run_interactive_device_login(self, authenticator):
with patch("sys.stdin.isatty", return_value=True):
assert authenticator._can_run_interactive_device_login() is True
with patch("sys.stdin.isatty", return_value=False):
assert authenticator._can_run_interactive_device_login() is False
with patch("sys.stdin", None):
assert authenticator._can_run_interactive_device_login() is False
with patch("sys.stdin", object()):
assert authenticator._can_run_interactive_device_login() is False
def test_get_access_token_headless_environment_raises_promptly(self, authenticator):
with (
patch("builtins.open", side_effect=FileNotFoundError),
patch("sys.stdin.isatty", return_value=False),
pytest.raises(GetAccessTokenError) as exc_info,
):
authenticator.get_access_token()
assert exc_info.value.status_code == 401
err_msg = str(exc_info.value)
assert "cannot run in a non-interactive/headless environment" in err_msg
assert "/root" not in err_msg
assert "/home" not in err_msg
assert "token.json" not in err_msg
def test_headless_wait_for_access_token_returns_none_immediately(self, authenticator):
with patch("sys.stdin.isatty", return_value=False):
assert authenticator._wait_for_access_token(timeout_seconds=900) is None