in alibabacloud_credentials/client.py [0:0]
def get_credentials(config):
if config.type == ac.ACCESS_KEY:
provider = StaticAKCredentialsProvider(
access_key_id=config.access_key_id,
access_key_secret=config.access_key_secret,
)
return _CredentialsProviderWrap(type_name='access_key', provider=provider)
elif config.type == ac.STS:
provider = StaticSTSCredentialsProvider(
access_key_id=config.access_key_id,
access_key_secret=config.access_key_secret,
security_token=config.security_token,
)
return _CredentialsProviderWrap(type_name='sts', provider=provider)
elif config.type == ac.BEARER:
return credentials.BearerTokenCredential(config.bearer_token)
elif config.type == ac.ECS_RAM_ROLE:
provider = EcsRamRoleCredentialsProvider(
role_name=config.role_name,
disable_imds_v1=config.disable_imds_v1,
http_options=HttpOptions(
read_timeout=config.timeout,
connect_timeout=config.connect_timeout,
proxy=config.proxy,
),
)
return _CredentialsProviderWrap(type_name='ecs_ram_role', provider=provider)
elif config.type == ac.CREDENTIALS_URI:
provider = URLCredentialsProvider(
uri=config.credentials_uri,
http_options=HttpOptions(
read_timeout=config.timeout,
connect_timeout=config.connect_timeout,
proxy=config.proxy,
),
)
return _CredentialsProviderWrap(type_name='credentials_uri', provider=provider)
elif config.type == ac.RAM_ROLE_ARN:
if config.security_token is not None and config.security_token != '':
previous_provider = StaticSTSCredentialsProvider(
access_key_id=config.access_key_id,
access_key_secret=config.access_key_secret,
security_token=config.security_token,
)
else:
previous_provider = StaticAKCredentialsProvider(
access_key_id=config.access_key_id,
access_key_secret=config.access_key_secret,
)
provider = RamRoleArnCredentialsProvider(
credentials_provider=previous_provider,
role_arn=config.role_arn,
role_session_name=config.role_session_name,
duration_seconds=config.role_session_expiration,
policy=config.policy,
external_id=config.external_id,
sts_endpoint=config.sts_endpoint,
http_options=HttpOptions(
read_timeout=config.timeout,
connect_timeout=config.connect_timeout,
proxy=config.proxy,
),
)
return _CredentialsProviderWrap(type_name='ram_role_arn', provider=provider)
elif config.type == ac.RSA_KEY_PAIR:
provider = RsaKeyPairCredentialsProvider(
public_key_id=config.public_key_id,
private_key_file=config.private_key_file,
duration_seconds=config.role_session_expiration,
sts_endpoint=config.sts_endpoint,
http_options=HttpOptions(
read_timeout=config.timeout,
connect_timeout=config.connect_timeout,
proxy=config.proxy,
),
)
return _CredentialsProviderWrap(type_name='rsa_key_pair', provider=provider)
elif config.type == ac.OIDC_ROLE_ARN:
provider = OIDCRoleArnCredentialsProvider(
role_arn=config.role_arn,
oidc_provider_arn=config.oidc_provider_arn,
oidc_token_file_path=config.oidc_token_file_path,
role_session_name=config.role_session_name,
duration_seconds=config.role_session_expiration,
policy=config.policy,
sts_endpoint=config.sts_endpoint,
http_options=HttpOptions(
read_timeout=config.timeout,
connect_timeout=config.connect_timeout,
proxy=config.proxy,
),
)
return _CredentialsProviderWrap(type_name='oidc_role_arn', provider=provider)
raise CredentialException(
'invalid type option, support: access_key, sts, bearer, ecs_ram_role, ram_role_arn, rsa_key_pair, oidc_role_arn, credentials_uri')