We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
--query does not work with aws eks get-token
It looks like `aws eks get-token` returns JSON-like output, but not pretty printed, and not working with `--query`, like
```
aws eks get-token --cluster-name myclustername --query status.token
```
still returns the complete output. As well format output cannot be changed.

Tested with
```
aws --version
aws-cli/1.16.218 Python/3.6.8 Linux/4.15.0-1047-aws botocore/1.12.208
```
but others reported the same for `1.16.230`: https://stackoverflow.com/a/57878048/1545325

Thank you!

--- 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 from awscli.customizations.utils import validate_mutually_exclusive
26 
27 AUTH_SERVICE = "sts"
28 AUTH_COMMAND = "GetCallerIdentity"
29 AUTH_API_VERSION = "2011-06-15"
30 AUTH_SIGNING_VERSION = "v4"
31 
32 ALPHA_API = "client.authentication.k8s.io/v1alpha1"
33 BETA_API = "client.authentication.k8s.io/v1beta1"
34 V1_API = "client.authentication.k8s.io/v1"
35 
36 FULLY_SUPPORTED_API_VERSIONS = [
37     V1_API,
38     BETA_API,
39 ]
40 DEPRECATED_API_VERSIONS = [
41     ALPHA_API,
42 ]
43 
44 ERROR_MSG_TPL = (
45     "{0} KUBERNETES_EXEC_INFO, defaulting to {1}. This is likely a "
46     "bug in your Kubernetes client. Please update your Kubernetes "
47     "client."
48 )
49 UNRECOGNIZED_MSG_TPL = (
50     "Unrecognized API version in KUBERNETES_EXEC_INFO, defaulting to "
51     "{0}. This is likely due to an outdated AWS "
52     "CLI. Please update your AWS CLI."
53 )
54 DEPRECATION_MSG_TPL = (
55     "Kubeconfig user entry is using deprecated API version {0}. Run "
56     "'aws eks update-kubeconfig' to update."
57 )
58 
59 # Presigned url timeout in seconds
60 URL_TIMEOUT = 60
61 
62 TOKEN_EXPIRATION_MINS = 14
63 
64 TOKEN_PREFIX = 'k8s-aws-v1.'
65 
66 K8S_AWS_ID_HEADER = 'x-k8s-aws-id'
67 
68 
69 class GetTokenCommand(BasicCommand):
70     NAME = 'get-token'
71 
72     DESCRIPTION = (
73         "Get a token for authentication with an Amazon EKS cluster. "
74         "This can be used as an alternative to the "
75         "aws-iam-authenticator."
76     )
77 
78     ARG_TABLE = [
79         {
80             'name': 'cluster-name',
81             'help_text': (
82                 "Specify the name of the Amazon EKS cluster to create a token for. (Note: for local clusters on AWS Outposts, please use --cluster-id parameter)"
83             ),
84             'required': False,
85         },
86         {
87             'name': 'role-arn',
88             'help_text': (
89                 "Assume this role for credentials when signing the token. "
90                 "Use this optional parameter when the credentials for signing "
91                 "the token differ from that of the current role session. "
92                 "Using this parameter results in new role session credentials "
93                 "that are used to sign the token."
94             ),
95             'required': False,
96         },
97         {
98             'name': 'cluster-id',
99             # When EKS in-region cluster supports cluster-id, we will need to update this help text
100             'help_text': (
101                 "Specify the id of the Amazon EKS cluster to create a token for. (Note: for local clusters on AWS Outposts only)"
102             ),
103             'required': False,
104         },
105     ]
106 
107     def get_expiration_time(self):
108         token_expiration = datetime.utcnow() + timedelta(
109             minutes=TOKEN_EXPIRATION_MINS
110         )
111         return token_expiration.strftime('%Y-%m-%dT%H:%M:%SZ')
112 
113     def _run_main(self, parsed_args, parsed_globals):
114         client_factory = STSClientFactory(self._session)
115         sts_client = client_factory.get_sts_client(
116             region_name=parsed_globals.region, role_arn=parsed_args.role_arn
117         )
118         
119         validate_mutually_exclusive(parsed_args, ['cluster_name'], ['cluster_id'])
120 
121         if parsed_args.cluster_id:
122             identifier = parsed_args.cluster_id
123         elif parsed_args.cluster_name:
124             identifier = parsed_args.cluster_name
125         else:
126             return ValueError("Either parameter --cluster-name or --cluster-id must be specified.")
127 
128         token = TokenGenerator(sts_client).get_token(identifier)
129 
130         # By default STS signs the url for 15 minutes so we are creating a
131         # rfc3339 timestamp with expiration in 14 minutes as part of the token, which
132         # is used by some clients (client-go) who will refresh the token after 14 mins
133         token_expiration = self.get_expiration_time()
134 
135         full_object = {
136             "kind": "ExecCredential",
137             "apiVersion": self.discover_api_version(),
138             "spec": {},
139             "status": {
140                 "expirationTimestamp": token_expiration,
141                 "token": token,
142             },
143         }
144 
145         uni_print(json.dumps(full_object))
146         uni_print('\n')
147         return 0
148 
149     def discover_api_version(self):
150         """
151         Parses the KUBERNETES_EXEC_INFO environment variable and returns the
152         API version. If the environment variable is malformed or invalid,
153         return the v1beta1 response and print a message to stderr.
154 
155         If the v1alpha1 API is specified explicitly, a message is printed to
156         stderr with instructions to update.
157 
158         :return: The client authentication API version
159         :rtype: string
160         """
161         # At the time Kubernetes v1.29 is released upstream (approx Dec 2023),
162         # "v1beta1" will be removed. At or around that time, EKS will likely
163         # support v1.22 through v1.28, in which client API version "v1beta1"
164         # will be supported by all EKS versions.
165         fallback_api_version = BETA_API
166 
167         error_prefixes = {
168             "error": "Error parsing",
169             "empty": "Empty",
170         }
171 
172         exec_info_raw = os.environ.get("KUBERNETES_EXEC_INFO", "")
173         if not exec_info_raw:
174             # All kube clients should be setting this, but client-go clients
175             # (kubectl, kubelet, etc) < 1.20 were not setting this if the API
176             # version defined in the kubeconfig was not v1alpha1.
177             #
178             # This was changed in kubernetes/kubernetes#95489 so that
179             # KUBERNETES_EXEC_INFO is always provided
180             return fallback_api_version
181         try:
182             exec_info = json.loads(exec_info_raw)
183         except json.JSONDecodeError:
184             # The environment variable was malformed
185             uni_print(
186                 ERROR_MSG_TPL.format(
187                     error_prefixes["error"],
188                     fallback_api_version,
189                 ),
190                 sys.stderr,
191             )
192             uni_print("\n", sys.stderr)
193             return fallback_api_version
194 
195         api_version_raw = exec_info.get("apiVersion")
196         if api_version_raw in FULLY_SUPPORTED_API_VERSIONS:
197             return api_version_raw
198         elif api_version_raw in DEPRECATED_API_VERSIONS:
199             uni_print(DEPRECATION_MSG_TPL.format(api_version_raw), sys.stderr)
200             uni_print("\n", sys.stderr)
201             return api_version_raw
202         else:
203             uni_print(
204                 UNRECOGNIZED_MSG_TPL.format(fallback_api_version),
205                 sys.stderr,
206             )
207             uni_print("\n", sys.stderr)
208             return fallback_api_version
209 
210 
211 class TokenGenerator(object):
212     def __init__(self, sts_client):
213         self._sts_client = sts_client
214 
215     def get_token(self, k8s_aws_id):
216         """Generate a presigned url token to pass to kubectl."""
217         url = self._get_presigned_url(k8s_aws_id)
218         token = TOKEN_PREFIX + base64.urlsafe_b64encode(
219             url.encode('utf-8')
220         ).decode('utf-8').rstrip('=')
221         return token
222 
223     def _get_presigned_url(self, k8s_aws_id):
224         return self._sts_client.generate_presigned_url(
225             'get_caller_identity',
226             Params={K8S_AWS_ID_HEADER: k8s_aws_id},
227             ExpiresIn=URL_TIMEOUT,
228             HttpMethod='GET',
229         )
230 
231 
232 class STSClientFactory(object):
233     def __init__(self, session):
234         self._session = session
235 
236     def get_sts_client(self, region_name=None, role_arn=None):
237         client_kwargs = {'region_name': region_name}
238         if role_arn is not None:
239             creds = self._get_role_credentials(region_name, role_arn)
240             client_kwargs['aws_access_key_id'] = creds['AccessKeyId']
241             client_kwargs['aws_secret_access_key'] = creds['SecretAccessKey']
242             client_kwargs['aws_session_token'] = creds['SessionToken']
243         sts = self._session.create_client('sts', **client_kwargs)
244         self._register_k8s_aws_id_handlers(sts)
245         return sts
246 
247     def _get_role_credentials(self, region_name, role_arn):
248         sts = self._session.create_client('sts', region_name)
249         return sts.assume_role(
250             RoleArn=role_arn, RoleSessionName='EKSGetTokenAuth'
251         )['Credentials']
252 
253     def _register_k8s_aws_id_handlers(self, sts_client):
254         sts_client.meta.events.register(
255             'provide-client-params.sts.GetCallerIdentity',
256             self._retrieve_k8s_aws_id,
257         )
258         sts_client.meta.events.register(
259             'before-sign.sts.GetCallerIdentity',
260             self._inject_k8s_aws_id_header,
261         )
262 
263     def _retrieve_k8s_aws_id(self, params, context, **kwargs):
264         if K8S_AWS_ID_HEADER in params:
265             context[K8S_AWS_ID_HEADER] = params.pop(K8S_AWS_ID_HEADER)
266 
267     def _inject_k8s_aws_id_header(self, request, **kwargs):
268         if K8S_AWS_ID_HEADER in request.context:
269             request.headers[K8S_AWS_ID_HEADER] = request.context[K8S_AWS_ID_HEADER]
270
```
--- 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
@@ -20,6 +20,7 @@
 from botocore.signers import RequestSigner
 from botocore.model import ServiceId
 
+from awscli.formatter import get_formatter
 from awscli.customizations.commands import BasicCommand
 from awscli.customizations.utils import uni_print
 from awscli.customizations.utils import validate_mutually_exclusive
@@ -142,7 +143,11 @@
             },
         }
 
-        uni_print(json.dumps(full_object))
+        output = self._session.get_config_variable('output')
+        formatter = get_formatter(output, parsed_globals)
+        formatter.query = parsed_globals.query
+
+        formatter(self.NAME, full_object)
         uni_print('\n')
         return 0

{"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@@ -20,6 +20,7 @@\n from botocore.signers import RequestSigner\n from botocore.model import ServiceId\n \n+from awscli.formatter import get_formatter\n from awscli.customizations.commands import BasicCommand\n from awscli.customizations.utils import uni_print\n from awscli.customizations.utils import validate_mutually_exclusive\n@@ -142,7 +143,11 @@\n             },\n         }\n \n-        uni_print(json.dumps(full_object))\n+        output = self._session.get_config_variable('output')\n+        formatter = get_formatter(output, parsed_globals)\n+        formatter.query = parsed_globals.query\n+\n+        formatter(self.NAME, full_object)\n         uni_print('\\n')\n         return 0\n", "issue": "--query does not work with aws eks get-token\nIt looks like `aws eks get-token` returns JSON-like output, but not pretty printed, and not working with `--query`, like\r\n```\r\naws eks get-token --cluster-name myclustername --query status.token\r\n```\r\nstill returns the complete output. As well format output cannot be changed.\r\n\r\nTested with\r\n```\r\naws --version\r\naws-cli/1.16.218 Python/3.6.8 Linux/4.15.0-1047-aws botocore/1.12.208\r\n```\r\nbut others reported the same for `1.16.230`: https://stackoverflow.com/a/57878048/1545325\r\n\r\nThank you!\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\nfrom awscli.customizations.utils import validate_mutually_exclusive\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\nK8S_AWS_ID_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. (Note: for local clusters on AWS Outposts, please use --cluster-id parameter)\"\n            ),\n            'required': False,\n        },\n        {\n            'name': 'role-arn',\n            'help_text': (\n                \"Assume this role for credentials when signing the token. \"\n                \"Use this optional parameter when the credentials for signing \"\n                \"the token differ from that of the current role session. \"\n                \"Using this parameter results in new role session credentials \"\n                \"that are used to sign the token.\"\n            ),\n            'required': False,\n        },\n        {\n            'name': 'cluster-id',\n            # When EKS in-region cluster supports cluster-id, we will need to update this help text\n            'help_text': (\n                \"Specify the id of the Amazon EKS cluster to create a token for. (Note: for local clusters on AWS Outposts only)\"\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        \n        validate_mutually_exclusive(parsed_args, ['cluster_name'], ['cluster_id'])\n\n        if parsed_args.cluster_id:\n            identifier = parsed_args.cluster_id\n        elif parsed_args.cluster_name:\n            identifier = parsed_args.cluster_name\n        else:\n            return ValueError(\"Either parameter --cluster-name or --cluster-id must be specified.\")\n\n        token = TokenGenerator(sts_client).get_token(identifier)\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, k8s_aws_id):\n        \"\"\"Generate a presigned url token to pass to kubectl.\"\"\"\n        url = self._get_presigned_url(k8s_aws_id)\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, k8s_aws_id):\n        return self._sts_client.generate_presigned_url(\n            'get_caller_identity',\n            Params={K8S_AWS_ID_HEADER: k8s_aws_id},\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_k8s_aws_id_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_k8s_aws_id_handlers(self, sts_client):\n        sts_client.meta.events.register(\n            'provide-client-params.sts.GetCallerIdentity',\n            self._retrieve_k8s_aws_id,\n        )\n        sts_client.meta.events.register(\n            'before-sign.sts.GetCallerIdentity',\n            self._inject_k8s_aws_id_header,\n        )\n\n    def _retrieve_k8s_aws_id(self, params, context, **kwargs):\n        if K8S_AWS_ID_HEADER in params:\n            context[K8S_AWS_ID_HEADER] = params.pop(K8S_AWS_ID_HEADER)\n\n    def _inject_k8s_aws_id_header(self, request, **kwargs):\n        if K8S_AWS_ID_HEADER in request.context:\n            request.headers[K8S_AWS_ID_HEADER] = request.context[K8S_AWS_ID_HEADER]\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.formatter import get_formatter\nfrom awscli.customizations.commands import BasicCommand\nfrom awscli.customizations.utils import uni_print\nfrom awscli.customizations.utils import validate_mutually_exclusive\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\nK8S_AWS_ID_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. (Note: for local clusters on AWS Outposts, please use --cluster-id parameter)\"\n            ),\n            'required': False,\n        },\n        {\n            'name': 'role-arn',\n            'help_text': (\n                \"Assume this role for credentials when signing the token. \"\n                \"Use this optional parameter when the credentials for signing \"\n                \"the token differ from that of the current role session. \"\n                \"Using this parameter results in new role session credentials \"\n                \"that are used to sign the token.\"\n            ),\n            'required': False,\n        },\n        {\n            'name': 'cluster-id',\n            # When EKS in-region cluster supports cluster-id, we will need to update this help text\n            'help_text': (\n                \"Specify the id of the Amazon EKS cluster to create a token for. (Note: for local clusters on AWS Outposts only)\"\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        \n        validate_mutually_exclusive(parsed_args, ['cluster_name'], ['cluster_id'])\n\n        if parsed_args.cluster_id:\n            identifier = parsed_args.cluster_id\n        elif parsed_args.cluster_name:\n            identifier = parsed_args.cluster_name\n        else:\n            return ValueError(\"Either parameter --cluster-name or --cluster-id must be specified.\")\n\n        token = TokenGenerator(sts_client).get_token(identifier)\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        output = self._session.get_config_variable('output')\n        formatter = get_formatter(output, parsed_globals)\n        formatter.query = parsed_globals.query\n\n        formatter(self.NAME, 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, k8s_aws_id):\n        \"\"\"Generate a presigned url token to pass to kubectl.\"\"\"\n        url = self._get_presigned_url(k8s_aws_id)\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, k8s_aws_id):\n        return self._sts_client.generate_presigned_url(\n            'get_caller_identity',\n            Params={K8S_AWS_ID_HEADER: k8s_aws_id},\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_k8s_aws_id_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_k8s_aws_id_handlers(self, sts_client):\n        sts_client.meta.events.register(\n            'provide-client-params.sts.GetCallerIdentity',\n            self._retrieve_k8s_aws_id,\n        )\n        sts_client.meta.events.register(\n            'before-sign.sts.GetCallerIdentity',\n            self._inject_k8s_aws_id_header,\n        )\n\n    def _retrieve_k8s_aws_id(self, params, context, **kwargs):\n        if K8S_AWS_ID_HEADER in params:\n            context[K8S_AWS_ID_HEADER] = params.pop(K8S_AWS_ID_HEADER)\n\n    def _inject_k8s_aws_id_header(self, request, **kwargs):\n        if K8S_AWS_ID_HEADER in request.context:\n            request.headers[K8S_AWS_ID_HEADER] = request.context[K8S_AWS_ID_HEADER]\n", "path": "awscli/customizations/eks/get_token.py"}]}