diff --git a/src/anthropic/lib/credentials/_chain.py b/src/anthropic/lib/credentials/_chain.py index 6e974a7b0..5be104ed1 100644 --- a/src/anthropic/lib/credentials/_chain.py +++ b/src/anthropic/lib/credentials/_chain.py @@ -32,7 +32,7 @@ def _build_federation_result(*, base_url: str) -> Optional[CredentialResult]: isn't fully set.""" federation_rule_id = os.environ.get(ENV_FEDERATION_RULE_ID) organization_id = os.environ.get(ENV_ORGANIZATION_ID) - has_literal_token = ENV_IDENTITY_TOKEN in os.environ + has_literal_token = bool(os.environ.get(ENV_IDENTITY_TOKEN)) identity_token_path = resolve_identity_token_path() if not federation_rule_id or not organization_id: @@ -48,11 +48,11 @@ def _build_federation_result(*, base_url: str) -> Optional[CredentialResult]: # at the next token exchange (don't capture into a closure). def _read_env_token() -> str: value = os.environ.get(ENV_IDENTITY_TOKEN) - if value is None: + if not value: raise CredentialsError( - f"{ENV_IDENTITY_TOKEN} is not set; the workload-identity chain " - f"selected this provider at construction time but the env var " - f"is no longer present." + f"{ENV_IDENTITY_TOKEN} is not set or is empty; the workload-identity chain " + f"selected this provider at construction time but the env var no longer " + f"contains an identity token." ) return value diff --git a/tests/lib/test_empty_workload_identity_token.py b/tests/lib/test_empty_workload_identity_token.py new file mode 100644 index 000000000..f9ecc82c0 --- /dev/null +++ b/tests/lib/test_empty_workload_identity_token.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import pathlib + +import pytest + +from anthropic import CredentialsError +from anthropic.lib.credentials import _chain +from anthropic.lib.credentials._constants import ( + ENV_IDENTITY_TOKEN, + ENV_IDENTITY_TOKEN_FILE, + ENV_ORGANIZATION_ID, + ENV_FEDERATION_RULE_ID, +) + + +def test_empty_literal_identity_token_does_not_select_federation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(ENV_FEDERATION_RULE_ID, "fdrl_test") + monkeypatch.setenv(ENV_ORGANIZATION_ID, "org_test") + monkeypatch.setenv(ENV_IDENTITY_TOKEN, "") + monkeypatch.delenv(ENV_IDENTITY_TOKEN_FILE, raising=False) + + assert _chain._build_federation_result(base_url="https://api.anthropic.com") is None + + +def test_literal_identity_token_cleared_after_discovery_fails_before_exchange( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(ENV_FEDERATION_RULE_ID, "fdrl_test") + monkeypatch.setenv(ENV_ORGANIZATION_ID, "org_test") + monkeypatch.setenv(ENV_IDENTITY_TOKEN, "initial-jwt") + monkeypatch.delenv(ENV_IDENTITY_TOKEN_FILE, raising=False) + + result = _chain._build_federation_result(base_url="https://api.anthropic.com") + assert result is not None + + monkeypatch.setenv(ENV_IDENTITY_TOKEN, "") + try: + with pytest.raises(CredentialsError, match="not set or is empty"): + result.provider() + finally: + close = getattr(result.provider, "close", None) + if close is not None: + close() + + +def test_empty_literal_token_does_not_mask_identity_token_file( + monkeypatch: pytest.MonkeyPatch, + tmp_path: pathlib.Path, +) -> None: + token_file = tmp_path / "identity-token" + token_file.write_text("file-jwt") + + monkeypatch.setenv(ENV_FEDERATION_RULE_ID, "fdrl_test") + monkeypatch.setenv(ENV_ORGANIZATION_ID, "org_test") + monkeypatch.setenv(ENV_IDENTITY_TOKEN, "") + monkeypatch.setenv(ENV_IDENTITY_TOKEN_FILE, str(token_file)) + + result = _chain._build_federation_result(base_url="https://api.anthropic.com") + assert result is not None + try: + provider = result.provider + identity_provider = getattr(provider, "_identity_token_provider") + assert identity_provider() == "file-jwt" + finally: + close = getattr(result.provider, "close", None) + if close is not None: + close()