| 219 | |
| 220 | |
| 221 | class TokenGenerator: |
| 222 | def __init__(self, sts_client): |
| 223 | self._sts_client = sts_client |
| 224 | |
| 225 | def get_token(self, k8s_aws_id): |
| 226 | """Generate a presigned url token to pass to kubectl.""" |
| 227 | url = self._get_presigned_url(k8s_aws_id) |
| 228 | token = TOKEN_PREFIX + base64.urlsafe_b64encode( |
| 229 | url.encode('utf-8') |
| 230 | ).decode('utf-8').rstrip('=') |
| 231 | return token |
| 232 | |
| 233 | def _get_presigned_url(self, k8s_aws_id): |
| 234 | return self._sts_client.generate_presigned_url( |
| 235 | 'get_caller_identity', |
| 236 | Params={K8S_AWS_ID_HEADER: k8s_aws_id}, |
| 237 | ExpiresIn=URL_TIMEOUT, |
| 238 | HttpMethod='GET', |
| 239 | ) |
| 240 | |
| 241 | |
| 242 | class STSClientFactory: |
no outgoing calls