Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 14 additions & 2 deletions src/openai/auth/_workload.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

import math
import time
import threading
from typing import Any, Generic, TypeVar, Callable, TypedDict, cast
Expand Down Expand Up @@ -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()
Expand Down
39 changes: 38 additions & 1 deletion tests/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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])

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Cover non-finite expirations on both client paths

The regression matrix neither includes NaN/positive or negative infinity—the non-finite values this change is intended to reject—nor exercises the AsyncOpenAI public path. Consequently, removing the math.isfinite guard or breaking async error propagation would leave this suite passing; add explicit non-finite cases and an asynchronous invalid-expiration test, as authentication changes require focused synchronous and asynchronous regression coverage.

AGENTS.md reference: AGENTS.md:L41-L45

Useful? React with 👍 / 👎.

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")
Expand Down