in src/run.py [0:0]
def get_platform_credentials(platform: Platform) -> dict:
"""
Credentials for platform (k8s) only. This should be refactored somehow.
Raises ExecutorException if not credentials are found
:param platform:
:return:
"""
# TODO K8S request a short-lived token here from long-lived in case
# it's possible
# if creds.token:
# _LOG.info('Request a temp token')
# conf = kubernetes.client.Configuration()
# conf.host = creds.endpoint
# conf.api_key['authorization'] = creds.token
# conf.api_key_prefix['authorization'] = 'Bearer'
# conf.ssl_ca_cert = creds.ca_file()
# kubernetes.client.AuthenticationV1TokenRequest()
# with kubernetes.client.ApiClient(conf) as client:
# kubernetes.client.CoreV1Api(client).create_namespaced_service_account_token('readonly-user')
app = SP.modular_client.application_service().get_application_by_id(
platform.parent.application_id
)
token = BSP.credentials_service.get_credentials_from_ssm()
kubeconfig = {}
if app.secret:
kubeconfig = SP.modular_client.assume_role_ssm_service().get_parameter(
app.secret) or {} # noqa
if kubeconfig and token:
_LOG.debug('Kubeconfig and custom token are provided. '
'Combining both')
config = Kubeconfig(kubeconfig)
session = str(int(time.time()))
user = f'user-{session}'
context = f'context-{session}'
cluster = next(config.cluster_names()) # always should be 1 at least
config.add_user(user, token)
config.add_context(context, cluster, user)
config.current_context = context
return {ENV_KUBECONFIG: str(config.to_temp_file())}
elif kubeconfig:
_LOG.debug('Only kubeconfig is provided')
config = Kubeconfig(kubeconfig)
return {ENV_KUBECONFIG: str(config.to_temp_file())}
if platform.type != PlatformType.EKS:
_LOG.warning('No kubeconfig provided and platform is not EKS')
raise ExecutorException(ExecutorError.NO_CREDENTIALS)
_LOG.debug('Kubeconfig and token are not provided. '
'Using management creds for EKS')
tenant = SP.modular_client.tenant_service().get(platform.tenant_name)
parent = SP.modular_client.parent_service().get_linked_parent_by_tenant(
tenant=tenant, type_=ParentType.AWS_MANAGEMENT
)
if not parent:
_LOG.warning('Parent AWS_MANAGEMENT not found')
raise ExecutorException(ExecutorError.NO_CREDENTIALS)
application = SP.modular_client.application_service().get_application_by_id(parent.application_id)
if not application:
_LOG.warning('Management application is not found')
raise ExecutorException(ExecutorError.NO_CREDENTIALS)
creds = SP.modular_client.maestro_credentials_service().get_by_application(
application, tenant
)
if not creds:
_LOG.warning(f'No credentials in '
f'application: {application.application_id}')
raise ExecutorException(ExecutorError.NO_CREDENTIALS)
cluster = EKSClient.factory().from_keys(
aws_access_key_id=creds.AWS_ACCESS_KEY_ID,
aws_secret_access_key=creds.AWS_SECRET_ACCESS_KEY,
aws_session_token=creds.AWS_SESSION_TOKEN,
region_name=platform.region
).describe_cluster(platform.name)
if not cluster:
_LOG.error(f'No cluster with name: {platform.name} '
f'in region: {platform.region}')
raise ExecutorException(ExecutorError.NO_CREDENTIALS)
sts = Boto3ClientFactory('sts').from_keys(
aws_access_key_id=creds.AWS_ACCESS_KEY_ID,
aws_secret_access_key=creds.AWS_SECRET_ACCESS_KEY,
aws_session_token=creds.AWS_SESSION_TOKEN,
region_name=AWS_DEFAULT_REGION
)
token_config = K8STokenKubeconfig(
endpoint=cluster['endpoint'],
ca=cluster['certificateAuthority']['data'],
token=TokenGenerator(sts).get_token(platform.name)
)
return {ENV_KUBECONFIG: str(token_config.to_temp_file())}