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
5 changes: 4 additions & 1 deletion kubernetes_asyncio/config/kube_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -304,6 +304,8 @@ async def _load_oid_token(self) -> str:
datetime.datetime.utcfromtimestamp(expires)
):
await self._refresh_oidc(provider)
self.token = "Bearer {}".format(provider["config"]["id-token"])
return self.token

self.token = "Bearer {}".format(provider["config"]["id-token"])
return self.token
Expand Down Expand Up @@ -332,7 +334,8 @@ async def _refresh_oidc(self, provider) -> None:
resp = await requestor.refresh_token(provider["config"]["refresh-token"])

provider["config"].value["id-token"] = resp["id_token"]
provider["config"].value["refresh-token"] = resp["refresh_token"]
if "refresh_token" in resp:
provider["config"].value["refresh-token"] = resp["refresh_token"]

if self._config_persister:
self._config_persister(self._config.value)
Expand Down
19 changes: 19 additions & 0 deletions kubernetes_asyncio/config/kube_config_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -725,6 +725,25 @@ async def test_oidc_with_refresh_no_idp_cert_data(self, mock_refresh_token) -> N
await loader._load_authentication()
self.assertEqual("Bearer abc123", loader.token)

@patch("kubernetes_asyncio.config.kube_config.OpenIDRequestor.refresh_token")
async def test_oidc_with_refresh_no_new_refresh_token(self, mock_refresh_token) -> None:
original_refresh_token = "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"
refreshed_id_token = "simple-refreshed-token-123"
mock_refresh_token.return_value = {
"id_token": refreshed_id_token,
}

loader = KubeConfigLoader(
config_dict=self.TEST_KUBE_CONFIG,
active_context="expired_oidc",
)
await loader._load_authentication()
self.assertEqual("Bearer {}".format(refreshed_id_token), loader.token)
self.assertEqual(
original_refresh_token,
loader._user["auth-provider"]["config"]["refresh-token"]
)

async def test_invalid_oidc_configs(self) -> None:
loader = KubeConfigLoader(config_dict=self.TEST_KUBE_CONFIG)

Expand Down
Loading