We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`Empty KUBERNETES_EXEC_INFO` warning when running `aws eks get-token`
### Describe the bug

This is likely related to https://github.com/kubernetes/kubectl/issues/1210

Using the newest update contained the in above patch, `kubectl` auth commands run as expected. However any requests to `aws eks get-token` that don't come from `kubectl` itself (or a tool using a similar mechanism) result in the following warning:
```shell
Empty KUBERNETES_EXEC_INFO, defaulting to client.authentication.k8s.io/v1alpha1. This is likely a bug in your Kubernetes client. Please update your Kubernetes client.
```
This might just need some documentation to prevent confusion, but I'm already seeing users run into it and ask. And the usual [reddit suspects](https://www.reddit.com/r/kubernetes/comments/um1wa1/what_is_kubernetes_exec_info_and_why_would_it_be/)

### Expected Behavior

- More detailed warning message? 
- Suppression of warnings when an interactive prompt is detected?
- Documentation notes?

### Current Behavior

`aws eks get-token` (with valid credentials)
Output:
```shell
Empty KUBERNETES_EXEC_INFO, defaulting to client.authentication.k8s.io/v1alpha1. This is likely a bug in your Kubernetes client. Please update your Kubernetes client.
<token>
```

### Reproduction Steps

`aws eks get-token`

### Possible Solution

_No response_

### Additional Information/Context

_No response_

### CLI version used

`aws-cli/2.6.3 Python/3.9.12 Darwin/21.4.0 source/arm64 prompt/off`

### Environment details (OS name and version, etc.)

macOS 12.3.1

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `awscli/customizations/eks/get_token.py`
Content:
```
1 # Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License"). You
4 # may not use this file except in compliance with the License. A copy of
5 # the License is located at
6 #
7 # http://aws.amazon.com/apache2.0/
8 #
9 # or in the "license" file accompanying this file. This file is
10 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11 # ANY KIND, either express or implied. See the License for the specific
12 # language governing permissions and limitations under the License.
13 import base64
14 import botocore
15 import json
16 import os
17 import sys
18 
19 from datetime import datetime, timedelta
20 from botocore.signers import RequestSigner
21 from botocore.model import ServiceId
22 
23 from awscli.customizations.commands import BasicCommand
24 from awscli.customizations.utils import uni_print
25 
26 AUTH_SERVICE = "sts"
27 AUTH_COMMAND = "GetCallerIdentity"
28 AUTH_API_VERSION = "2011-06-15"
29 AUTH_SIGNING_VERSION = "v4"
30 
31 ALPHA_API = "client.authentication.k8s.io/v1alpha1"
32 BETA_API = "client.authentication.k8s.io/v1beta1"
33 V1_API = "client.authentication.k8s.io/v1"
34 
35 FULLY_SUPPORTED_API_VERSIONS = [
36     V1_API,
37     BETA_API,
38 ]
39 DEPRECATED_API_VERSIONS = [
40     ALPHA_API,
41 ]
42 
43 ERROR_MSG_TPL = (
44     "{0} KUBERNETES_EXEC_INFO, defaulting to {1}. This is likely a "
45     "bug in your Kubernetes client. Please update your Kubernetes "
46     "client."
47 )
48 UNRECOGNIZED_MSG_TPL = (
49     "Unrecognized API version in KUBERNETES_EXEC_INFO, defaulting to "
50     "{0}. This is likely due to an outdated AWS "
51     "CLI. Please update your AWS CLI."
52 )
53 DEPRECATION_MSG_TPL = (
54     "Kubeconfig user entry is using deprecated API version {0}. Run "
55     "'aws eks update-kubeconfig' to update."
56 )
57 
58 # Presigned url timeout in seconds
59 URL_TIMEOUT = 60
60 
61 TOKEN_EXPIRATION_MINS = 14
62 
63 TOKEN_PREFIX = 'k8s-aws-v1.'
64 
65 CLUSTER_NAME_HEADER = 'x-k8s-aws-id'
66 
67 
68 class GetTokenCommand(BasicCommand):
69     NAME = 'get-token'
70 
71     DESCRIPTION = (
72         "Get a token for authentication with an Amazon EKS cluster. "
73         "This can be used as an alternative to the "
74         "aws-iam-authenticator."
75     )
76 
77     ARG_TABLE = [
78         {
79             'name': 'cluster-name',
80             'help_text': (
81                 "Specify the name of the Amazon EKS cluster to create a token for."
82             ),
83             'required': True,
84         },
85         {
86             'name': 'role-arn',
87             'help_text': (
88                 "Assume this role for credentials when signing the token."
89             ),
90             'required': False,
91         },
92     ]
93 
94     def get_expiration_time(self):
95         token_expiration = datetime.utcnow() + timedelta(
96             minutes=TOKEN_EXPIRATION_MINS
97         )
98         return token_expiration.strftime('%Y-%m-%dT%H:%M:%SZ')
99 
100     def _run_main(self, parsed_args, parsed_globals):
101         client_factory = STSClientFactory(self._session)
102         sts_client = client_factory.get_sts_client(
103             region_name=parsed_globals.region, role_arn=parsed_args.role_arn
104         )
105         token = TokenGenerator(sts_client).get_token(parsed_args.cluster_name)
106 
107         # By default STS signs the url for 15 minutes so we are creating a
108         # rfc3339 timestamp with expiration in 14 minutes as part of the token, which
109         # is used by some clients (client-go) who will refresh the token after 14 mins
110         token_expiration = self.get_expiration_time()
111 
112         full_object = {
113             "kind": "ExecCredential",
114             "apiVersion": self.discover_api_version(),
115             "spec": {},
116             "status": {
117                 "expirationTimestamp": token_expiration,
118                 "token": token,
119             },
120         }
121 
122         uni_print(json.dumps(full_object))
123         uni_print('\n')
124         return 0
125 
126     def discover_api_version(self):
127         """
128         Parses the KUBERNETES_EXEC_INFO environment variable and returns the
129         API version. If the environment variable is empty, malformed, or
130         invalid, return the v1alpha1 response and print a message to stderr.
131 
132         If the v1alpha1 API is specified explicitly, a message is printed to
133         stderr with instructions to update.
134 
135         :return: The client authentication API version
136         :rtype: string
137         """
138         # At the time Kubernetes v1.29 is released upstream (approx Dec 2023),
139         # "v1beta1" will be removed. At or around that time, EKS will likely
140         # support v1.22 through v1.28, in which client API version "v1beta1"
141         # will be supported by all EKS versions.
142         fallback_api_version = ALPHA_API
143 
144         error_prefixes = {
145             "error": "Error parsing",
146             "empty": "Empty",
147         }
148 
149         exec_info_raw = os.environ.get("KUBERNETES_EXEC_INFO", "")
150         if not exec_info_raw:
151             # All kube clients should be setting this. Otherewise, we'll return
152             # the fallback and write an error.
153             uni_print(
154                 ERROR_MSG_TPL.format(
155                     error_prefixes["empty"],
156                     fallback_api_version,
157                 ),
158                 sys.stderr,
159             )
160             uni_print("\n", sys.stderr)
161             return fallback_api_version
162         try:
163             exec_info = json.loads(exec_info_raw)
164         except json.JSONDecodeError:
165             # The environment variable was malformed
166             uni_print(
167                 ERROR_MSG_TPL.format(
168                     error_prefixes["error"],
169                     fallback_api_version,
170                 ),
171                 sys.stderr,
172             )
173             uni_print("\n", sys.stderr)
174             return fallback_api_version
175 
176         api_version_raw = exec_info.get("apiVersion")
177         if api_version_raw in FULLY_SUPPORTED_API_VERSIONS:
178             return api_version_raw
179         elif api_version_raw in DEPRECATED_API_VERSIONS:
180             uni_print(DEPRECATION_MSG_TPL.format(ALPHA_API), sys.stderr)
181             uni_print("\n", sys.stderr)
182             return api_version_raw
183         else:
184             uni_print(
185                 UNRECOGNIZED_MSG_TPL.format(fallback_api_version),
186                 sys.stderr,
187             )
188             uni_print("\n", sys.stderr)
189             return fallback_api_version
190 
191 
192 class TokenGenerator(object):
193     def __init__(self, sts_client):
194         self._sts_client = sts_client
195 
196     def get_token(self, cluster_name):
197         """Generate a presigned url token to pass to kubectl."""
198         url = self._get_presigned_url(cluster_name)
199         token = TOKEN_PREFIX + base64.urlsafe_b64encode(
200             url.encode('utf-8')
201         ).decode('utf-8').rstrip('=')
202         return token
203 
204     def _get_presigned_url(self, cluster_name):
205         return self._sts_client.generate_presigned_url(
206             'get_caller_identity',
207             Params={'ClusterName': cluster_name},
208             ExpiresIn=URL_TIMEOUT,
209             HttpMethod='GET',
210         )
211 
212 
213 class STSClientFactory(object):
214     def __init__(self, session):
215         self._session = session
216 
217     def get_sts_client(self, region_name=None, role_arn=None):
218         client_kwargs = {'region_name': region_name}
219         if role_arn is not None:
220             creds = self._get_role_credentials(region_name, role_arn)
221             client_kwargs['aws_access_key_id'] = creds['AccessKeyId']
222             client_kwargs['aws_secret_access_key'] = creds['SecretAccessKey']
223             client_kwargs['aws_session_token'] = creds['SessionToken']
224         sts = self._session.create_client('sts', **client_kwargs)
225         self._register_cluster_name_handlers(sts)
226         return sts
227 
228     def _get_role_credentials(self, region_name, role_arn):
229         sts = self._session.create_client('sts', region_name)
230         return sts.assume_role(
231             RoleArn=role_arn, RoleSessionName='EKSGetTokenAuth'
232         )['Credentials']
233 
234     def _register_cluster_name_handlers(self, sts_client):
235         sts_client.meta.events.register(
236             'provide-client-params.sts.GetCallerIdentity',
237             self._retrieve_cluster_name,
238         )
239         sts_client.meta.events.register(
240             'before-sign.sts.GetCallerIdentity',
241             self._inject_cluster_name_header,
242         )
243 
244     def _retrieve_cluster_name(self, params, context, **kwargs):
245         if 'ClusterName' in params:
246             context['eks_cluster'] = params.pop('ClusterName')
247 
248     def _inject_cluster_name_header(self, request, **kwargs):
249         if 'eks_cluster' in request.context:
250             request.headers[CLUSTER_NAME_HEADER] = request.context[
251                 'eks_cluster'
252             ]
253
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/awscli/customizations/eks/get_token.py b/awscli/customizations/eks/get_token.py
--- a/awscli/customizations/eks/get_token.py
+++ b/awscli/customizations/eks/get_token.py
@@ -126,8 +126,8 @@
     def discover_api_version(self):
         """
         Parses the KUBERNETES_EXEC_INFO environment variable and returns the
-        API version. If the environment variable is empty, malformed, or
-        invalid, return the v1alpha1 response and print a message to stderr.
+        API version. If the environment variable is malformed or invalid,
+        return the v1beta1 response and print a message to stderr.
 
         If the v1alpha1 API is specified explicitly, a message is printed to
         stderr with instructions to update.
@@ -139,7 +139,7 @@
         # "v1beta1" will be removed. At or around that time, EKS will likely
         # support v1.22 through v1.28, in which client API version "v1beta1"
         # will be supported by all EKS versions.
-        fallback_api_version = ALPHA_API
+        fallback_api_version = BETA_API
 
         error_prefixes = {
             "error": "Error parsing",
@@ -148,16 +148,12 @@
 
         exec_info_raw = os.environ.get("KUBERNETES_EXEC_INFO", "")
         if not exec_info_raw:
-            # All kube clients should be setting this. Otherewise, we'll return
-            # the fallback and write an error.
-            uni_print(
-                ERROR_MSG_TPL.format(
-                    error_prefixes["empty"],
-                    fallback_api_version,
-                ),
-                sys.stderr,
-            )
-            uni_print("\n", sys.stderr)
+            # All kube clients should be setting this, but client-go clients
+            # (kubectl, kubelet, etc) < 1.20 were not setting this if the API
+            # version defined in the kubeconfig was not v1alpha1.
+            #
+            # This was changed in kubernetes/kubernetes#95489 so that
+            # KUBERNETES_EXEC_INFO is always provided
             return fallback_api_version
         try:
             exec_info = json.loads(exec_info_raw)
@@ -177,7 +173,7 @@
         if api_version_raw in FULLY_SUPPORTED_API_VERSIONS:
             return api_version_raw
         elif api_version_raw in DEPRECATED_API_VERSIONS:
-            uni_print(DEPRECATION_MSG_TPL.format(ALPHA_API), sys.stderr)
+            uni_print(DEPRECATION_MSG_TPL.format(api_version_raw), sys.stderr)
             uni_print("\n", sys.stderr)
             return api_version_raw
         else:

{"golden_diff": "diff --git a/awscli/customizations/eks/get_token.py b/awscli/customizations/eks/get_token.py\n--- a/awscli/customizations/eks/get_token.py\n+++ b/awscli/customizations/eks/get_token.py\n@@ -126,8 +126,8 @@\n     def discover_api_version(self):\n         \"\"\"\n         Parses the KUBERNETES_EXEC_INFO environment variable and returns the\n-        API version. If the environment variable is empty, malformed, or\n-        invalid, return the v1alpha1 response and print a message to stderr.\n+        API version. If the environment variable is malformed or invalid,\n+        return the v1beta1 response and print a message to stderr.\n \n         If the v1alpha1 API is specified explicitly, a message is printed to\n         stderr with instructions to update.\n@@ -139,7 +139,7 @@\n         # \"v1beta1\" will be removed. At or around that time, EKS will likely\n         # support v1.22 through v1.28, in which client API version \"v1beta1\"\n         # will be supported by all EKS versions.\n-        fallback_api_version = ALPHA_API\n+        fallback_api_version = BETA_API\n \n         error_prefixes = {\n             \"error\": \"Error parsing\",\n@@ -148,16 +148,12 @@\n \n         exec_info_raw = os.environ.get(\"KUBERNETES_EXEC_INFO\", \"\")\n         if not exec_info_raw:\n-            # All kube clients should be setting this. Otherewise, we'll return\n-            # the fallback and write an error.\n-            uni_print(\n-                ERROR_MSG_TPL.format(\n-                    error_prefixes[\"empty\"],\n-                    fallback_api_version,\n-                ),\n-                sys.stderr,\n-            )\n-            uni_print(\"\\n\", sys.stderr)\n+            # All kube clients should be setting this, but client-go clients\n+            # (kubectl, kubelet, etc) < 1.20 were not setting this if the API\n+            # version defined in the kubeconfig was not v1alpha1.\n+            #\n+            # This was changed in kubernetes/kubernetes#95489 so that\n+            # KUBERNETES_EXEC_INFO is always provided\n             return fallback_api_version\n         try:\n             exec_info = json.loads(exec_info_raw)\n@@ -177,7 +173,7 @@\n         if api_version_raw in FULLY_SUPPORTED_API_VERSIONS:\n             return api_version_raw\n         elif api_version_raw in DEPRECATED_API_VERSIONS:\n-            uni_print(DEPRECATION_MSG_TPL.format(ALPHA_API), sys.stderr)\n+            uni_print(DEPRECATION_MSG_TPL.format(api_version_raw), sys.stderr)\n             uni_print(\"\\n\", sys.stderr)\n             return api_version_raw\n         else:\n", "issue": "`Empty KUBERNETES_EXEC_INFO` warning when running `aws eks get-token`\n### Describe the bug\n\nThis is likely related to https://github.com/kubernetes/kubectl/issues/1210\r\n\r\nUsing the newest update contained the in above patch, `kubectl` auth commands run as expected. However any requests to `aws eks get-token` that don't come from `kubectl` itself (or a tool using a similar mechanism) result in the following warning:\r\n```shell\r\nEmpty KUBERNETES_EXEC_INFO, defaulting to client.authentication.k8s.io/v1alpha1. This is likely a bug in your Kubernetes client. Please update your Kubernetes client.\r\n```\r\nThis might just need some documentation to prevent confusion, but I'm already seeing users run into it and ask. And the usual [reddit suspects](https://www.reddit.com/r/kubernetes/comments/um1wa1/what_is_kubernetes_exec_info_and_why_would_it_be/)\n\n### Expected Behavior\n\n- More detailed warning message? \r\n- Suppression of warnings when an interactive prompt is detected?\r\n- Documentation notes?\n\n### Current Behavior\n\n`aws eks get-token` (with valid credentials)\r\nOutput:\r\n```shell\r\nEmpty KUBERNETES_EXEC_INFO, defaulting to client.authentication.k8s.io/v1alpha1. This is likely a bug in your Kubernetes client. Please update your Kubernetes client.\r\n<token>\r\n```\n\n### Reproduction Steps\n\n`aws eks get-token`\n\n### Possible Solution\n\n_No response_\n\n### Additional Information/Context\n\n_No response_\n\n### CLI version used\n\n`aws-cli/2.6.3 Python/3.9.12 Darwin/21.4.0 source/arm64 prompt/off`\n\n### Environment details (OS name and version, etc.)\n\nmacOS 12.3.1\n", "before_files": [{"content": "# Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"). You\n# may not use this file except in compliance with the License. A copy of\n# the License is located at\n#\n# http://aws.amazon.com/apache2.0/\n#\n# or in the \"license\" file accompanying this file. This file is\n# distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF\n# ANY KIND, either express or implied. See the License for the specific\n# language governing permissions and limitations under the License.\nimport base64\nimport botocore\nimport json\nimport os\nimport sys\n\nfrom datetime import datetime, timedelta\nfrom botocore.signers import RequestSigner\nfrom botocore.model import ServiceId\n\nfrom awscli.customizations.commands import BasicCommand\nfrom awscli.customizations.utils import uni_print\n\nAUTH_SERVICE = \"sts\"\nAUTH_COMMAND = \"GetCallerIdentity\"\nAUTH_API_VERSION = \"2011-06-15\"\nAUTH_SIGNING_VERSION = \"v4\"\n\nALPHA_API = \"client.authentication.k8s.io/v1alpha1\"\nBETA_API = \"client.authentication.k8s.io/v1beta1\"\nV1_API = \"client.authentication.k8s.io/v1\"\n\nFULLY_SUPPORTED_API_VERSIONS = [\n    V1_API,\n    BETA_API,\n]\nDEPRECATED_API_VERSIONS = [\n    ALPHA_API,\n]\n\nERROR_MSG_TPL = (\n    \"{0} KUBERNETES_EXEC_INFO, defaulting to {1}. This is likely a \"\n    \"bug in your Kubernetes client. Please update your Kubernetes \"\n    \"client.\"\n)\nUNRECOGNIZED_MSG_TPL = (\n    \"Unrecognized API version in KUBERNETES_EXEC_INFO, defaulting to \"\n    \"{0}. This is likely due to an outdated AWS \"\n    \"CLI. Please update your AWS CLI.\"\n)\nDEPRECATION_MSG_TPL = (\n    \"Kubeconfig user entry is using deprecated API version {0}. Run \"\n    \"'aws eks update-kubeconfig' to update.\"\n)\n\n# Presigned url timeout in seconds\nURL_TIMEOUT = 60\n\nTOKEN_EXPIRATION_MINS = 14\n\nTOKEN_PREFIX = 'k8s-aws-v1.'\n\nCLUSTER_NAME_HEADER = 'x-k8s-aws-id'\n\n\nclass GetTokenCommand(BasicCommand):\n    NAME = 'get-token'\n\n    DESCRIPTION = (\n        \"Get a token for authentication with an Amazon EKS cluster. \"\n        \"This can be used as an alternative to the \"\n        \"aws-iam-authenticator.\"\n    )\n\n    ARG_TABLE = [\n        {\n            'name': 'cluster-name',\n            'help_text': (\n                \"Specify the name of the Amazon EKS cluster to create a token for.\"\n            ),\n            'required': True,\n        },\n        {\n            'name': 'role-arn',\n            'help_text': (\n                \"Assume this role for credentials when signing the token.\"\n            ),\n            'required': False,\n        },\n    ]\n\n    def get_expiration_time(self):\n        token_expiration = datetime.utcnow() + timedelta(\n            minutes=TOKEN_EXPIRATION_MINS\n        )\n        return token_expiration.strftime('%Y-%m-%dT%H:%M:%SZ')\n\n    def _run_main(self, parsed_args, parsed_globals):\n        client_factory = STSClientFactory(self._session)\n        sts_client = client_factory.get_sts_client(\n            region_name=parsed_globals.region, role_arn=parsed_args.role_arn\n        )\n        token = TokenGenerator(sts_client).get_token(parsed_args.cluster_name)\n\n        # By default STS signs the url for 15 minutes so we are creating a\n        # rfc3339 timestamp with expiration in 14 minutes as part of the token, which\n        # is used by some clients (client-go) who will refresh the token after 14 mins\n        token_expiration = self.get_expiration_time()\n\n        full_object = {\n            \"kind\": \"ExecCredential\",\n            \"apiVersion\": self.discover_api_version(),\n            \"spec\": {},\n            \"status\": {\n                \"expirationTimestamp\": token_expiration,\n                \"token\": token,\n            },\n        }\n\n        uni_print(json.dumps(full_object))\n        uni_print('\\n')\n        return 0\n\n    def discover_api_version(self):\n        \"\"\"\n        Parses the KUBERNETES_EXEC_INFO environment variable and returns the\n        API version. If the environment variable is empty, malformed, or\n        invalid, return the v1alpha1 response and print a message to stderr.\n\n        If the v1alpha1 API is specified explicitly, a message is printed to\n        stderr with instructions to update.\n\n        :return: The client authentication API version\n        :rtype: string\n        \"\"\"\n        # At the time Kubernetes v1.29 is released upstream (approx Dec 2023),\n        # \"v1beta1\" will be removed. At or around that time, EKS will likely\n        # support v1.22 through v1.28, in which client API version \"v1beta1\"\n        # will be supported by all EKS versions.\n        fallback_api_version = ALPHA_API\n\n        error_prefixes = {\n            \"error\": \"Error parsing\",\n            \"empty\": \"Empty\",\n        }\n\n        exec_info_raw = os.environ.get(\"KUBERNETES_EXEC_INFO\", \"\")\n        if not exec_info_raw:\n            # All kube clients should be setting this. Otherewise, we'll return\n            # the fallback and write an error.\n            uni_print(\n                ERROR_MSG_TPL.format(\n                    error_prefixes[\"empty\"],\n                    fallback_api_version,\n                ),\n                sys.stderr,\n            )\n            uni_print(\"\\n\", sys.stderr)\n            return fallback_api_version\n        try:\n            exec_info = json.loads(exec_info_raw)\n        except json.JSONDecodeError:\n            # The environment variable was malformed\n            uni_print(\n                ERROR_MSG_TPL.format(\n                    error_prefixes[\"error\"],\n                    fallback_api_version,\n                ),\n                sys.stderr,\n            )\n            uni_print(\"\\n\", sys.stderr)\n            return fallback_api_version\n\n        api_version_raw = exec_info.get(\"apiVersion\")\n        if api_version_raw in FULLY_SUPPORTED_API_VERSIONS:\n            return api_version_raw\n        elif api_version_raw in DEPRECATED_API_VERSIONS:\n            uni_print(DEPRECATION_MSG_TPL.format(ALPHA_API), sys.stderr)\n            uni_print(\"\\n\", sys.stderr)\n            return api_version_raw\n        else:\n            uni_print(\n                UNRECOGNIZED_MSG_TPL.format(fallback_api_version),\n                sys.stderr,\n            )\n            uni_print(\"\\n\", sys.stderr)\n            return fallback_api_version\n\n\nclass TokenGenerator(object):\n    def __init__(self, sts_client):\n        self._sts_client = sts_client\n\n    def get_token(self, cluster_name):\n        \"\"\"Generate a presigned url token to pass to kubectl.\"\"\"\n        url = self._get_presigned_url(cluster_name)\n        token = TOKEN_PREFIX + base64.urlsafe_b64encode(\n            url.encode('utf-8')\n        ).decode('utf-8').rstrip('=')\n        return token\n\n    def _get_presigned_url(self, cluster_name):\n        return self._sts_client.generate_presigned_url(\n            'get_caller_identity',\n            Params={'ClusterName': cluster_name},\n            ExpiresIn=URL_TIMEOUT,\n            HttpMethod='GET',\n        )\n\n\nclass STSClientFactory(object):\n    def __init__(self, session):\n        self._session = session\n\n    def get_sts_client(self, region_name=None, role_arn=None):\n        client_kwargs = {'region_name': region_name}\n        if role_arn is not None:\n            creds = self._get_role_credentials(region_name, role_arn)\n            client_kwargs['aws_access_key_id'] = creds['AccessKeyId']\n            client_kwargs['aws_secret_access_key'] = creds['SecretAccessKey']\n            client_kwargs['aws_session_token'] = creds['SessionToken']\n        sts = self._session.create_client('sts', **client_kwargs)\n        self._register_cluster_name_handlers(sts)\n        return sts\n\n    def _get_role_credentials(self, region_name, role_arn):\n        sts = self._session.create_client('sts', region_name)\n        return sts.assume_role(\n            RoleArn=role_arn, RoleSessionName='EKSGetTokenAuth'\n        )['Credentials']\n\n    def _register_cluster_name_handlers(self, sts_client):\n        sts_client.meta.events.register(\n            'provide-client-params.sts.GetCallerIdentity',\n            self._retrieve_cluster_name,\n        )\n        sts_client.meta.events.register(\n            'before-sign.sts.GetCallerIdentity',\n            self._inject_cluster_name_header,\n        )\n\n    def _retrieve_cluster_name(self, params, context, **kwargs):\n        if 'ClusterName' in params:\n            context['eks_cluster'] = params.pop('ClusterName')\n\n    def _inject_cluster_name_header(self, request, **kwargs):\n        if 'eks_cluster' in request.context:\n            request.headers[CLUSTER_NAME_HEADER] = request.context[\n                'eks_cluster'\n            ]\n", "path": "awscli/customizations/eks/get_token.py"}], "after_files": [{"content": "# Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\"). You\n# may not use this file except in compliance with the License. A copy of\n# the License is located at\n#\n# http://aws.amazon.com/apache2.0/\n#\n# or in the \"license\" file accompanying this file. This file is\n# distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF\n# ANY KIND, either express or implied. See the License for the specific\n# language governing permissions and limitations under the License.\nimport base64\nimport botocore\nimport json\nimport os\nimport sys\n\nfrom datetime import datetime, timedelta\nfrom botocore.signers import RequestSigner\nfrom botocore.model import ServiceId\n\nfrom awscli.customizations.commands import BasicCommand\nfrom awscli.customizations.utils import uni_print\n\nAUTH_SERVICE = \"sts\"\nAUTH_COMMAND = \"GetCallerIdentity\"\nAUTH_API_VERSION = \"2011-06-15\"\nAUTH_SIGNING_VERSION = \"v4\"\n\nALPHA_API = \"client.authentication.k8s.io/v1alpha1\"\nBETA_API = \"client.authentication.k8s.io/v1beta1\"\nV1_API = \"client.authentication.k8s.io/v1\"\n\nFULLY_SUPPORTED_API_VERSIONS = [\n    V1_API,\n    BETA_API,\n]\nDEPRECATED_API_VERSIONS = [\n    ALPHA_API,\n]\n\nERROR_MSG_TPL = (\n    \"{0} KUBERNETES_EXEC_INFO, defaulting to {1}. This is likely a \"\n    \"bug in your Kubernetes client. Please update your Kubernetes \"\n    \"client.\"\n)\nUNRECOGNIZED_MSG_TPL = (\n    \"Unrecognized API version in KUBERNETES_EXEC_INFO, defaulting to \"\n    \"{0}. This is likely due to an outdated AWS \"\n    \"CLI. Please update your AWS CLI.\"\n)\nDEPRECATION_MSG_TPL = (\n    \"Kubeconfig user entry is using deprecated API version {0}. Run \"\n    \"'aws eks update-kubeconfig' to update.\"\n)\n\n# Presigned url timeout in seconds\nURL_TIMEOUT = 60\n\nTOKEN_EXPIRATION_MINS = 14\n\nTOKEN_PREFIX = 'k8s-aws-v1.'\n\nCLUSTER_NAME_HEADER = 'x-k8s-aws-id'\n\n\nclass GetTokenCommand(BasicCommand):\n    NAME = 'get-token'\n\n    DESCRIPTION = (\n        \"Get a token for authentication with an Amazon EKS cluster. \"\n        \"This can be used as an alternative to the \"\n        \"aws-iam-authenticator.\"\n    )\n\n    ARG_TABLE = [\n        {\n            'name': 'cluster-name',\n            'help_text': (\n                \"Specify the name of the Amazon EKS cluster to create a token for.\"\n            ),\n            'required': True,\n        },\n        {\n            'name': 'role-arn',\n            'help_text': (\n                \"Assume this role for credentials when signing the token.\"\n            ),\n            'required': False,\n        },\n    ]\n\n    def get_expiration_time(self):\n        token_expiration = datetime.utcnow() + timedelta(\n            minutes=TOKEN_EXPIRATION_MINS\n        )\n        return token_expiration.strftime('%Y-%m-%dT%H:%M:%SZ')\n\n    def _run_main(self, parsed_args, parsed_globals):\n        client_factory = STSClientFactory(self._session)\n        sts_client = client_factory.get_sts_client(\n            region_name=parsed_globals.region, role_arn=parsed_args.role_arn\n        )\n        token = TokenGenerator(sts_client).get_token(parsed_args.cluster_name)\n\n        # By default STS signs the url for 15 minutes so we are creating a\n        # rfc3339 timestamp with expiration in 14 minutes as part of the token, which\n        # is used by some clients (client-go) who will refresh the token after 14 mins\n        token_expiration = self.get_expiration_time()\n\n        full_object = {\n            \"kind\": \"ExecCredential\",\n            \"apiVersion\": self.discover_api_version(),\n            \"spec\": {},\n            \"status\": {\n                \"expirationTimestamp\": token_expiration,\n                \"token\": token,\n            },\n        }\n\n        uni_print(json.dumps(full_object))\n        uni_print('\\n')\n        return 0\n\n    def discover_api_version(self):\n        \"\"\"\n        Parses the KUBERNETES_EXEC_INFO environment variable and returns the\n        API version. If the environment variable is malformed or invalid,\n        return the v1beta1 response and print a message to stderr.\n\n        If the v1alpha1 API is specified explicitly, a message is printed to\n        stderr with instructions to update.\n\n        :return: The client authentication API version\n        :rtype: string\n        \"\"\"\n        # At the time Kubernetes v1.29 is released upstream (approx Dec 2023),\n        # \"v1beta1\" will be removed. At or around that time, EKS will likely\n        # support v1.22 through v1.28, in which client API version \"v1beta1\"\n        # will be supported by all EKS versions.\n        fallback_api_version = BETA_API\n\n        error_prefixes = {\n            \"error\": \"Error parsing\",\n            \"empty\": \"Empty\",\n        }\n\n        exec_info_raw = os.environ.get(\"KUBERNETES_EXEC_INFO\", \"\")\n        if not exec_info_raw:\n            # All kube clients should be setting this, but client-go clients\n            # (kubectl, kubelet, etc) < 1.20 were not setting this if the API\n            # version defined in the kubeconfig was not v1alpha1.\n            #\n            # This was changed in kubernetes/kubernetes#95489 so that\n            # KUBERNETES_EXEC_INFO is always provided\n            return fallback_api_version\n        try:\n            exec_info = json.loads(exec_info_raw)\n        except json.JSONDecodeError:\n            # The environment variable was malformed\n            uni_print(\n                ERROR_MSG_TPL.format(\n                    error_prefixes[\"error\"],\n                    fallback_api_version,\n                ),\n                sys.stderr,\n            )\n            uni_print(\"\\n\", sys.stderr)\n            return fallback_api_version\n\n        api_version_raw = exec_info.get(\"apiVersion\")\n        if api_version_raw in FULLY_SUPPORTED_API_VERSIONS:\n            return api_version_raw\n        elif api_version_raw in DEPRECATED_API_VERSIONS:\n            uni_print(DEPRECATION_MSG_TPL.format(api_version_raw), sys.stderr)\n            uni_print(\"\\n\", sys.stderr)\n            return api_version_raw\n        else:\n            uni_print(\n                UNRECOGNIZED_MSG_TPL.format(fallback_api_version),\n                sys.stderr,\n            )\n            uni_print(\"\\n\", sys.stderr)\n            return fallback_api_version\n\n\nclass TokenGenerator(object):\n    def __init__(self, sts_client):\n        self._sts_client = sts_client\n\n    def get_token(self, cluster_name):\n        \"\"\"Generate a presigned url token to pass to kubectl.\"\"\"\n        url = self._get_presigned_url(cluster_name)\n        token = TOKEN_PREFIX + base64.urlsafe_b64encode(\n            url.encode('utf-8')\n        ).decode('utf-8').rstrip('=')\n        return token\n\n    def _get_presigned_url(self, cluster_name):\n        return self._sts_client.generate_presigned_url(\n            'get_caller_identity',\n            Params={'ClusterName': cluster_name},\n            ExpiresIn=URL_TIMEOUT,\n            HttpMethod='GET',\n        )\n\n\nclass STSClientFactory(object):\n    def __init__(self, session):\n        self._session = session\n\n    def get_sts_client(self, region_name=None, role_arn=None):\n        client_kwargs = {'region_name': region_name}\n        if role_arn is not None:\n            creds = self._get_role_credentials(region_name, role_arn)\n            client_kwargs['aws_access_key_id'] = creds['AccessKeyId']\n            client_kwargs['aws_secret_access_key'] = creds['SecretAccessKey']\n            client_kwargs['aws_session_token'] = creds['SessionToken']\n        sts = self._session.create_client('sts', **client_kwargs)\n        self._register_cluster_name_handlers(sts)\n        return sts\n\n    def _get_role_credentials(self, region_name, role_arn):\n        sts = self._session.create_client('sts', region_name)\n        return sts.assume_role(\n            RoleArn=role_arn, RoleSessionName='EKSGetTokenAuth'\n        )['Credentials']\n\n    def _register_cluster_name_handlers(self, sts_client):\n        sts_client.meta.events.register(\n            'provide-client-params.sts.GetCallerIdentity',\n            self._retrieve_cluster_name,\n        )\n        sts_client.meta.events.register(\n            'before-sign.sts.GetCallerIdentity',\n            self._inject_cluster_name_header,\n        )\n\n    def _retrieve_cluster_name(self, params, context, **kwargs):\n        if 'ClusterName' in params:\n            context['eks_cluster'] = params.pop('ClusterName')\n\n    def _inject_cluster_name_header(self, request, **kwargs):\n        if 'eks_cluster' in request.context:\n            request.headers[CLUSTER_NAME_HEADER] = request.context[\n                'eks_cluster'\n            ]\n", "path": "awscli/customizations/eks/get_token.py"}]}