diff --git a/src/openai/auth/_workload.py b/src/openai/auth/_workload.py index 4f9797f092..dbae7f6137 100644 --- a/src/openai/auth/_workload.py +++ b/src/openai/auth/_workload.py @@ -1,5 +1,6 @@ from __future__ import annotations +import math import time import threading from typing import Any, Generic, TypeVar, Callable, TypedDict, cast @@ -305,9 +306,20 @@ def _handle_token_response(self, response: httpx2.Response) -> dict[str, Any]: ) def _validate_expires_in(self, expires_in: object) -> float: - if not isinstance(expires_in, (int, float)): + # `bool` is a subclass of `int`, so guard against it explicitly to avoid + # treating `true`/`false` as a `1`/`0` second lifetime. + if isinstance(expires_in, bool) or not isinstance(expires_in, (int, float)): raise OpenAIError("Token exchange response did not include a valid expires_in") - return float(expires_in) + try: + expires_in_value = float(expires_in) + except OverflowError: + raise OpenAIError("Token exchange response did not include a valid expires_in") from None + # A non-positive lifetime yields an already-expired token, while a + # non-finite one (e.g. `inf`) would never trigger a refresh, leaving the + # client to reuse a token indefinitely after it has expired server-side. + if not math.isfinite(expires_in_value) or expires_in_value <= 0: + raise OpenAIError("Token exchange response did not include a valid expires_in") + return expires_in_value def _token_unusable(self) -> bool: return self._cached_token is None or self._token_expired() diff --git a/tests/test_auth.py b/tests/test_auth.py index fc717973e6..0d352c7961 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -7,7 +7,7 @@ from inline_snapshot import snapshot from tests import respx2 -from openai import OpenAI, OAuthError +from openai import OpenAI, OAuthError, OpenAIError from openai.auth import WorkloadIdentity, WorkloadIdentityAuth, SubjectTokenWorkloadIdentity from tests.respx2.models import Call from openai.auth._workload import ( @@ -166,6 +166,43 @@ def test_workload_identity_exchange_error() -> None: assert api_route.call_count == 0 +@respx2.mock +@pytest.mark.parametrize("expires_in", [0, -1, True, "3600", None, 10**400]) +def test_workload_identity_rejects_nonpositive_or_nonnumeric_expiration(expires_in: object) -> None: + exchange_route = respx2.post("https://auth.openai.com/oauth/token").mock( + return_value=httpx2.Response( + 200, + json={ + "access_token": "fake_access_token", + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + "token_type": "Bearer", + "expires_in": expires_in, + }, + ) + ) + api_route = respx2.get("https://api.openai.com/v1/models").mock( + return_value=httpx2.Response(200, json={"data": [], "object": "list"}) + ) + + client = OpenAI( + max_retries=0, + workload_identity={ + "identity_provider_id": "idp_123", + "service_account_id": "sa_123", + "provider": { + "get_token": lambda: "fake_subject_token", + "token_type": "jwt", + }, + }, + ) + + with pytest.raises(OpenAIError, match="expires_in"): + client.models.list() + + assert exchange_route.call_count == 1 + assert api_route.call_count == 0 + + def test_k8s_service_account_token_provider(tmp_path: Path) -> None: token_file = tmp_path / "token" token_file.write_text("my-k8s-token")