diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/botocore/credentials.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/botocore/credentials.py | 2391 |
1 files changed, 2391 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/botocore/credentials.py b/.venv/lib/python3.12/site-packages/botocore/credentials.py new file mode 100644 index 00000000..a425cb79 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/botocore/credentials.py @@ -0,0 +1,2391 @@ +# Copyright (c) 2012-2013 Mitch Garnaat http://garnaat.org/ +# Copyright 2012-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). You +# may not use this file except in compliance with the License. A copy of +# the License is located at +# +# http://aws.amazon.com/apache2.0/ +# +# or in the "license" file accompanying this file. This file is +# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF +# ANY KIND, either express or implied. See the License for the specific +# language governing permissions and limitations under the License. +import datetime +import getpass +import json +import logging +import os +import subprocess +import threading +import time +from collections import namedtuple +from copy import deepcopy +from hashlib import sha1 + +from dateutil.parser import parse +from dateutil.tz import tzlocal, tzutc + +import botocore.compat +import botocore.configloader +from botocore import UNSIGNED +from botocore.compat import compat_shell_split, total_seconds +from botocore.config import Config +from botocore.exceptions import ( + ConfigNotFound, + CredentialRetrievalError, + InfiniteLoopConfigError, + InvalidConfigError, + MetadataRetrievalError, + PartialCredentialsError, + RefreshWithMFAUnsupportedError, + UnauthorizedSSOTokenError, + UnknownCredentialError, +) +from botocore.tokens import SSOTokenProvider +from botocore.utils import ( + ArnParser, + ContainerMetadataFetcher, + FileWebIdentityTokenLoader, + InstanceMetadataFetcher, + JSONFileCache, + SSOTokenLoader, + parse_key_val_file, + resolve_imds_endpoint_mode, +) + +logger = logging.getLogger(__name__) +ReadOnlyCredentials = namedtuple( + 'ReadOnlyCredentials', + ['access_key', 'secret_key', 'token', 'account_id'], + defaults=(None,), +) + +_DEFAULT_MANDATORY_REFRESH_TIMEOUT = 10 * 60 # 10 min +_DEFAULT_ADVISORY_REFRESH_TIMEOUT = 15 * 60 # 15 min + + +def create_credential_resolver(session, cache=None, region_name=None): + """Create a default credential resolver. + + This creates a pre-configured credential resolver + that includes the default lookup chain for + credentials. + + """ + profile_name = session.get_config_variable('profile') or 'default' + metadata_timeout = session.get_config_variable('metadata_service_timeout') + num_attempts = session.get_config_variable('metadata_service_num_attempts') + disable_env_vars = session.instance_variables().get('profile') is not None + + imds_config = { + 'ec2_metadata_service_endpoint': session.get_config_variable( + 'ec2_metadata_service_endpoint' + ), + 'ec2_metadata_service_endpoint_mode': resolve_imds_endpoint_mode( + session + ), + 'ec2_credential_refresh_window': _DEFAULT_ADVISORY_REFRESH_TIMEOUT, + 'ec2_metadata_v1_disabled': session.get_config_variable( + 'ec2_metadata_v1_disabled' + ), + } + + if cache is None: + cache = {} + + env_provider = EnvProvider() + container_provider = ContainerProvider() + instance_metadata_provider = InstanceMetadataProvider( + iam_role_fetcher=InstanceMetadataFetcher( + timeout=metadata_timeout, + num_attempts=num_attempts, + user_agent=session.user_agent(), + config=imds_config, + ) + ) + + profile_provider_builder = ProfileProviderBuilder( + session, cache=cache, region_name=region_name + ) + assume_role_provider = AssumeRoleProvider( + load_config=lambda: session.full_config, + client_creator=_get_client_creator(session, region_name), + cache=cache, + profile_name=profile_name, + credential_sourcer=CanonicalNameCredentialSourcer( + [env_provider, container_provider, instance_metadata_provider] + ), + profile_provider_builder=profile_provider_builder, + ) + + pre_profile = [ + env_provider, + assume_role_provider, + ] + profile_providers = profile_provider_builder.providers( + profile_name=profile_name, + disable_env_vars=disable_env_vars, + ) + post_profile = [ + OriginalEC2Provider(), + BotoProvider(), + container_provider, + instance_metadata_provider, + ] + providers = pre_profile + profile_providers + post_profile + + if disable_env_vars: + # An explicitly provided profile will negate an EnvProvider. + # We will defer to providers that understand the "profile" + # concept to retrieve credentials. + # The one edge case if is all three values are provided via + # env vars: + # export AWS_ACCESS_KEY_ID=foo + # export AWS_SECRET_ACCESS_KEY=bar + # export AWS_PROFILE=baz + # Then, just like our client() calls, the explicit credentials + # will take precedence. + # + # This precedence is enforced by leaving the EnvProvider in the chain. + # This means that the only way a "profile" would win is if the + # EnvProvider does not return credentials, which is what we want + # in this scenario. + providers.remove(env_provider) + logger.debug( + 'Skipping environment variable credential check' + ' because profile name was explicitly set.' + ) + + resolver = CredentialResolver(providers=providers) + return resolver + + +class ProfileProviderBuilder: + """This class handles the creation of profile based providers. + + NOTE: This class is only intended for internal use. + + This class handles the creation and ordering of the various credential + providers that primarly source their configuration from the shared config. + This is needed to enable sharing between the default credential chain and + the source profile chain created by the assume role provider. + """ + + def __init__( + self, session, cache=None, region_name=None, sso_token_cache=None + ): + self._session = session + self._cache = cache + self._region_name = region_name + self._sso_token_cache = sso_token_cache + + def providers(self, profile_name, disable_env_vars=False): + return [ + self._create_web_identity_provider( + profile_name, + disable_env_vars, + ), + self._create_sso_provider(profile_name), + self._create_shared_credential_provider(profile_name), + self._create_process_provider(profile_name), + self._create_config_provider(profile_name), + ] + + def _create_process_provider(self, profile_name): + return ProcessProvider( + profile_name=profile_name, + load_config=lambda: self._session.full_config, + ) + + def _create_shared_credential_provider(self, profile_name): + credential_file = self._session.get_config_variable('credentials_file') + return SharedCredentialProvider( + profile_name=profile_name, + creds_filename=credential_file, + ) + + def _create_config_provider(self, profile_name): + config_file = self._session.get_config_variable('config_file') + return ConfigProvider( + profile_name=profile_name, + config_filename=config_file, + ) + + def _create_web_identity_provider(self, profile_name, disable_env_vars): + return AssumeRoleWithWebIdentityProvider( + load_config=lambda: self._session.full_config, + client_creator=_get_client_creator( + self._session, self._region_name + ), + cache=self._cache, + profile_name=profile_name, + disable_env_vars=disable_env_vars, + ) + + def _create_sso_provider(self, profile_name): + return SSOProvider( + load_config=lambda: self._session.full_config, + client_creator=self._session.create_client, + profile_name=profile_name, + cache=self._cache, + token_cache=self._sso_token_cache, + token_provider=SSOTokenProvider( + self._session, + cache=self._sso_token_cache, + profile_name=profile_name, + ), + ) + + +def get_credentials(session): + resolver = create_credential_resolver(session) + return resolver.load_credentials() + + +def _local_now(): + return datetime.datetime.now(tzlocal()) + + +def _parse_if_needed(value): + if isinstance(value, datetime.datetime): + return value + return parse(value) + + +def _serialize_if_needed(value, iso=False): + if isinstance(value, datetime.datetime): + if iso: + return value.isoformat() + return value.strftime('%Y-%m-%dT%H:%M:%S%Z') + return value + + +def _get_client_creator(session, region_name): + def client_creator(service_name, **kwargs): + create_client_kwargs = {'region_name': region_name} + create_client_kwargs.update(**kwargs) + return session.create_client(service_name, **create_client_kwargs) + + return client_creator + + +def create_assume_role_refresher(client, params): + def refresh(): + response = client.assume_role(**params) + credentials = response['Credentials'] + # We need to normalize the credential names to + # the values expected by the refresh creds. + return { + 'access_key': credentials['AccessKeyId'], + 'secret_key': credentials['SecretAccessKey'], + 'token': credentials['SessionToken'], + 'expiry_time': _serialize_if_needed(credentials['Expiration']), + } + + return refresh + + +def create_mfa_serial_refresher(actual_refresh): + class _Refresher: + def __init__(self, refresh): + self._refresh = refresh + self._has_been_called = False + + def __call__(self): + if self._has_been_called: + # We can explore an option in the future to support + # reprompting for MFA, but for now we just error out + # when the temp creds expire. + raise RefreshWithMFAUnsupportedError() + self._has_been_called = True + return self._refresh() + + return _Refresher(actual_refresh) + + +class Credentials: + """ + Holds the credentials needed to authenticate requests. + + :param str access_key: The access key part of the credentials. + :param str secret_key: The secret key part of the credentials. + :param str token: The security token, valid only for session credentials. + :param str method: A string which identifies where the credentials + were found. + :param str account_id: (optional) An account ID associated with the credentials. + """ + + def __init__( + self, access_key, secret_key, token=None, method=None, account_id=None + ): + self.access_key = access_key + self.secret_key = secret_key + self.token = token + + if method is None: + method = 'explicit' + self.method = method + self.account_id = account_id + + self._normalize() + + def _normalize(self): + # Keys would sometimes (accidentally) contain non-ascii characters. + # It would cause a confusing UnicodeDecodeError in Python 2. + # We explicitly convert them into unicode to avoid such error. + # + # Eventually the service will decide whether to accept the credential. + # This also complies with the behavior in Python 3. + self.access_key = botocore.compat.ensure_unicode(self.access_key) + self.secret_key = botocore.compat.ensure_unicode(self.secret_key) + + def get_frozen_credentials(self): + return ReadOnlyCredentials( + self.access_key, self.secret_key, self.token, self.account_id + ) + + def get_deferred_property(self, property_name): + def get_property(): + return getattr(self, property_name, None) + + return get_property + + +class RefreshableCredentials(Credentials): + """ + Holds the credentials needed to authenticate requests. In addition, it + knows how to refresh itself. + + :param str access_key: The access key part of the credentials. + :param str secret_key: The secret key part of the credentials. + :param str token: The security token, valid only for session credentials. + :param datetime expiry_time: The expiration time of the credentials. + :param function refresh_using: Callback function to refresh the credentials. + :param str method: A string which identifies where the credentials + were found. + :param function time_fetcher: Callback function to retrieve current time. + """ + + # The time at which we'll attempt to refresh, but not + # block if someone else is refreshing. + _advisory_refresh_timeout = _DEFAULT_ADVISORY_REFRESH_TIMEOUT + # The time at which all threads will block waiting for + # refreshed credentials. + _mandatory_refresh_timeout = _DEFAULT_MANDATORY_REFRESH_TIMEOUT + + def __init__( + self, + access_key, + secret_key, + token, + expiry_time, + refresh_using, + method, + time_fetcher=_local_now, + advisory_timeout=None, + mandatory_timeout=None, + account_id=None, + ): + self._refresh_using = refresh_using + self._access_key = access_key + self._secret_key = secret_key + self._token = token + self._account_id = account_id + self._expiry_time = expiry_time + self._time_fetcher = time_fetcher + self._refresh_lock = threading.Lock() + self.method = method + self._frozen_credentials = ReadOnlyCredentials( + access_key, secret_key, token, account_id + ) + self._normalize() + if advisory_timeout is not None: + self._advisory_refresh_timeout = advisory_timeout + if mandatory_timeout is not None: + self._mandatory_refresh_timeout = mandatory_timeout + + def _normalize(self): + self._access_key = botocore.compat.ensure_unicode(self._access_key) + self._secret_key = botocore.compat.ensure_unicode(self._secret_key) + + @classmethod + def create_from_metadata( + cls, + metadata, + refresh_using, + method, + advisory_timeout=None, + mandatory_timeout=None, + ): + kwargs = {} + if advisory_timeout is not None: + kwargs['advisory_timeout'] = advisory_timeout + if mandatory_timeout is not None: + kwargs['mandatory_timeout'] = mandatory_timeout + + instance = cls( + access_key=metadata['access_key'], + secret_key=metadata['secret_key'], + token=metadata['token'], + expiry_time=cls._expiry_datetime(metadata['expiry_time']), + method=method, + refresh_using=refresh_using, + account_id=metadata.get('account_id'), + **kwargs, + ) + return instance + + @property + def access_key(self): + """Warning: Using this property can lead to race conditions if you + access another property subsequently along the refresh boundary. + Please use get_frozen_credentials instead. + """ + self._refresh() + return self._access_key + + @access_key.setter + def access_key(self, value): + self._access_key = value + + @property + def secret_key(self): + """Warning: Using this property can lead to race conditions if you + access another property subsequently along the refresh boundary. + Please use get_frozen_credentials instead. + """ + self._refresh() + return self._secret_key + + @secret_key.setter + def secret_key(self, value): + self._secret_key = value + + @property + def token(self): + """Warning: Using this property can lead to race conditions if you + access another property subsequently along the refresh boundary. + Please use get_frozen_credentials instead. + """ + self._refresh() + return self._token + + @token.setter + def token(self, value): + self._token = value + + @property + def account_id(self): + """Warning: Using this property can lead to race conditions if you + access another property subsequently along the refresh boundary. + Please use get_frozen_credentials instead. + """ + self._refresh() + return self._account_id + + @account_id.setter + def account_id(self, value): + self._account_id = value + + def _seconds_remaining(self): + delta = self._expiry_time - self._time_fetcher() + return total_seconds(delta) + + def refresh_needed(self, refresh_in=None): + """Check if a refresh is needed. + + A refresh is needed if the expiry time associated + with the temporary credentials is less than the + provided ``refresh_in``. If ``time_delta`` is not + provided, ``self.advisory_refresh_needed`` will be used. + + For example, if your temporary credentials expire + in 10 minutes and the provided ``refresh_in`` is + ``15 * 60``, then this function will return ``True``. + + :type refresh_in: int + :param refresh_in: The number of seconds before the + credentials expire in which refresh attempts should + be made. + + :return: True if refresh needed, False otherwise. + + """ + if self._expiry_time is None: + # No expiration, so assume we don't need to refresh. + return False + + if refresh_in is None: + refresh_in = self._advisory_refresh_timeout + # The credentials should be refreshed if they're going to expire + # in less than 5 minutes. + if self._seconds_remaining() >= refresh_in: + # There's enough time left. Don't refresh. + return False + logger.debug("Credentials need to be refreshed.") + return True + + def _is_expired(self): + # Checks if the current credentials are expired. + return self.refresh_needed(refresh_in=0) + + def _refresh(self): + # In the common case where we don't need a refresh, we + # can immediately exit and not require acquiring the + # refresh lock. + if not self.refresh_needed(self._advisory_refresh_timeout): + return + + # acquire() doesn't accept kwargs, but False is indicating + # that we should not block if we can't acquire the lock. + # If we aren't able to acquire the lock, we'll trigger + # the else clause. + if self._refresh_lock.acquire(False): + try: + if not self.refresh_needed(self._advisory_refresh_timeout): + return + is_mandatory_refresh = self.refresh_needed( + self._mandatory_refresh_timeout + ) + self._protected_refresh(is_mandatory=is_mandatory_refresh) + return + finally: + self._refresh_lock.release() + elif self.refresh_needed(self._mandatory_refresh_timeout): + # If we're within the mandatory refresh window, + # we must block until we get refreshed credentials. + with self._refresh_lock: + if not self.refresh_needed(self._mandatory_refresh_timeout): + return + self._protected_refresh(is_mandatory=True) + + def _protected_refresh(self, is_mandatory): + # precondition: this method should only be called if you've acquired + # the self._refresh_lock. + try: + metadata = self._refresh_using() + except Exception: + period_name = 'mandatory' if is_mandatory else 'advisory' + logger.warning( + "Refreshing temporary credentials failed " + "during %s refresh period.", + period_name, + exc_info=True, + ) + if is_mandatory: + # If this is a mandatory refresh, then + # all errors that occur when we attempt to refresh + # credentials are propagated back to the user. + raise + # Otherwise we'll just return. + # The end result will be that we'll use the current + # set of temporary credentials we have. + return + self._set_from_data(metadata) + self._frozen_credentials = ReadOnlyCredentials( + self._access_key, self._secret_key, self._token, self._account_id + ) + if self._is_expired(): + # We successfully refreshed credentials but for whatever + # reason, our refreshing function returned credentials + # that are still expired. In this scenario, the only + # thing we can do is let the user know and raise + # an exception. + msg = ( + "Credentials were refreshed, but the " + "refreshed credentials are still expired." + ) + logger.warning(msg) + raise RuntimeError(msg) + + @staticmethod + def _expiry_datetime(time_str): + return parse(time_str) + + def _set_from_data(self, data): + expected_keys = ['access_key', 'secret_key', 'token', 'expiry_time'] + if not data: + missing_keys = expected_keys + else: + missing_keys = [k for k in expected_keys if k not in data] + + if missing_keys: + message = "Credential refresh failed, response did not contain: %s" + raise CredentialRetrievalError( + provider=self.method, + error_msg=message % ', '.join(missing_keys), + ) + + self.access_key = data['access_key'] + self.secret_key = data['secret_key'] + self.token = data['token'] + self._expiry_time = parse(data['expiry_time']) + self.account_id = data.get('account_id') + logger.debug( + "Retrieved credentials will expire at: %s", self._expiry_time + ) + self._normalize() + + def get_frozen_credentials(self): + """Return immutable credentials. + + The ``access_key``, ``secret_key``, and ``token`` properties + on this class will always check and refresh credentials if + needed before returning the particular credentials. + + This has an edge case where you can get inconsistent + credentials. Imagine this: + + # Current creds are "t1" + tmp.access_key ---> expired? no, so return t1.access_key + # ---- time is now expired, creds need refreshing to "t2" ---- + tmp.secret_key ---> expired? yes, refresh and return t2.secret_key + + This means we're using the access key from t1 with the secret key + from t2. To fix this issue, you can request a frozen credential object + which is guaranteed not to change. + + The frozen credentials returned from this method should be used + immediately and then discarded. The typical usage pattern would + be:: + + creds = RefreshableCredentials(...) + some_code = SomeSignerObject() + # I'm about to sign the request. + # The frozen credentials are only used for the + # duration of generate_presigned_url and will be + # immediately thrown away. + request = some_code.sign_some_request( + with_credentials=creds.get_frozen_credentials()) + print("Signed request:", request) + + """ + self._refresh() + return self._frozen_credentials + + +class DeferredRefreshableCredentials(RefreshableCredentials): + """Refreshable credentials that don't require initial credentials. + + refresh_using will be called upon first access. + """ + + def __init__(self, refresh_using, method, time_fetcher=_local_now): + self._refresh_using = refresh_using + self._access_key = None + self._secret_key = None + self._token = None + self._account_id = None + self._expiry_time = None + self._time_fetcher = time_fetcher + self._refresh_lock = threading.Lock() + self.method = method + self._frozen_credentials = None + + def refresh_needed(self, refresh_in=None): + if self._frozen_credentials is None: + return True + return super().refresh_needed(refresh_in) + + +class CachedCredentialFetcher: + DEFAULT_EXPIRY_WINDOW_SECONDS = 60 * 15 + + def __init__(self, cache=None, expiry_window_seconds=None): + if cache is None: + cache = {} + self._cache = cache + self._cache_key = self._create_cache_key() + if expiry_window_seconds is None: + expiry_window_seconds = self.DEFAULT_EXPIRY_WINDOW_SECONDS + self._expiry_window_seconds = expiry_window_seconds + + def _create_cache_key(self): + raise NotImplementedError('_create_cache_key()') + + def _make_file_safe(self, filename): + # Replace :, path sep, and / to make it the string filename safe. + filename = filename.replace(':', '_').replace(os.sep, '_') + return filename.replace('/', '_') + + def _get_credentials(self): + raise NotImplementedError('_get_credentials()') + + def fetch_credentials(self): + return self._get_cached_credentials() + + def _get_cached_credentials(self): + """Get up-to-date credentials. + + This will check the cache for up-to-date credentials, calling assume + role if none are available. + """ + response = self._load_from_cache() + if response is None: + response = self._get_credentials() + self._write_to_cache(response) + else: + logger.debug("Credentials for role retrieved from cache.") + + creds = response['Credentials'] + expiration = _serialize_if_needed(creds['Expiration'], iso=True) + credentials = { + 'access_key': creds['AccessKeyId'], + 'secret_key': creds['SecretAccessKey'], + 'token': creds['SessionToken'], + 'expiry_time': expiration, + 'account_id': creds.get('AccountId'), + } + + return credentials + + def _load_from_cache(self): + if self._cache_key in self._cache: + creds = deepcopy(self._cache[self._cache_key]) + if not self._is_expired(creds): + return creds + else: + logger.debug( + "Credentials were found in cache, but they are expired." + ) + return None + + def _write_to_cache(self, response): + self._cache[self._cache_key] = deepcopy(response) + + def _is_expired(self, credentials): + """Check if credentials are expired.""" + end_time = _parse_if_needed(credentials['Credentials']['Expiration']) + seconds = total_seconds(end_time - _local_now()) + return seconds < self._expiry_window_seconds + + +class BaseAssumeRoleCredentialFetcher(CachedCredentialFetcher): + def __init__( + self, + client_creator, + role_arn, + extra_args=None, + cache=None, + expiry_window_seconds=None, + ): + self._client_creator = client_creator + self._role_arn = role_arn + + if extra_args is None: + self._assume_kwargs = {} + else: + self._assume_kwargs = deepcopy(extra_args) + self._assume_kwargs['RoleArn'] = self._role_arn + + self._role_session_name = self._assume_kwargs.get('RoleSessionName') + self._using_default_session_name = False + if not self._role_session_name: + self._generate_assume_role_name() + + super().__init__(cache, expiry_window_seconds) + + def _generate_assume_role_name(self): + self._role_session_name = f'botocore-session-{int(time.time())}' + self._assume_kwargs['RoleSessionName'] = self._role_session_name + self._using_default_session_name = True + + def _create_cache_key(self): + """Create a predictable cache key for the current configuration. + + The cache key is intended to be compatible with file names. + """ + args = deepcopy(self._assume_kwargs) + + # The role session name gets randomly generated, so we don't want it + # in the hash. + if self._using_default_session_name: + del args['RoleSessionName'] + + if 'Policy' in args: + # To have a predictable hash, the keys of the policy must be + # sorted, so we have to load it here to make sure it gets sorted + # later on. + args['Policy'] = json.loads(args['Policy']) + + args = json.dumps(args, sort_keys=True) + argument_hash = sha1(args.encode('utf-8')).hexdigest() + return self._make_file_safe(argument_hash) + + def _add_account_id_to_response(self, response): + role_arn = response.get('AssumedRoleUser', {}).get('Arn') + if ArnParser.is_arn(role_arn): + arn_parser = ArnParser() + account_id = arn_parser.parse_arn(role_arn)['account'] + response['Credentials']['AccountId'] = account_id + else: + logger.debug(f"Unable to extract account ID from Arn: {role_arn}") + + +class AssumeRoleCredentialFetcher(BaseAssumeRoleCredentialFetcher): + def __init__( + self, + client_creator, + source_credentials, + role_arn, + extra_args=None, + mfa_prompter=None, + cache=None, + expiry_window_seconds=None, + ): + """ + :type client_creator: callable + :param client_creator: A callable that creates a client taking + arguments like ``Session.create_client``. + + :type source_credentials: Credentials + :param source_credentials: The credentials to use to create the + client for the call to AssumeRole. + + :type role_arn: str + :param role_arn: The ARN of the role to be assumed. + + :type extra_args: dict + :param extra_args: Any additional arguments to add to the assume + role request using the format of the botocore operation. + Possible keys include, but may not be limited to, + DurationSeconds, Policy, SerialNumber, ExternalId and + RoleSessionName. + + :type mfa_prompter: callable + :param mfa_prompter: A callable that returns input provided by the + user (i.e raw_input, getpass.getpass, etc.). + + :type cache: dict + :param cache: An object that supports ``__getitem__``, + ``__setitem__``, and ``__contains__``. An example of this is + the ``JSONFileCache`` class in aws-cli. + + :type expiry_window_seconds: int + :param expiry_window_seconds: The amount of time, in seconds, + """ + self._source_credentials = source_credentials + self._mfa_prompter = mfa_prompter + if self._mfa_prompter is None: + self._mfa_prompter = getpass.getpass + + super().__init__( + client_creator, + role_arn, + extra_args=extra_args, + cache=cache, + expiry_window_seconds=expiry_window_seconds, + ) + + def _get_credentials(self): + """Get credentials by calling assume role.""" + kwargs = self._assume_role_kwargs() + client = self._create_client() + response = client.assume_role(**kwargs) + self._add_account_id_to_response(response) + return response + + def _assume_role_kwargs(self): + """Get the arguments for assume role based on current configuration.""" + assume_role_kwargs = deepcopy(self._assume_kwargs) + + mfa_serial = assume_role_kwargs.get('SerialNumber') + + if mfa_serial is not None: + prompt = f'Enter MFA code for {mfa_serial}: ' + token_code = self._mfa_prompter(prompt) + assume_role_kwargs['TokenCode'] = token_code + + duration_seconds = assume_role_kwargs.get('DurationSeconds') + + if duration_seconds is not None: + assume_role_kwargs['DurationSeconds'] = duration_seconds + + return assume_role_kwargs + + def _create_client(self): + """Create an STS client using the source credentials.""" + frozen_credentials = self._source_credentials.get_frozen_credentials() + return self._client_creator( + 'sts', + aws_access_key_id=frozen_credentials.access_key, + aws_secret_access_key=frozen_credentials.secret_key, + aws_session_token=frozen_credentials.token, + ) + + +class AssumeRoleWithWebIdentityCredentialFetcher( + BaseAssumeRoleCredentialFetcher +): + def __init__( + self, + client_creator, + web_identity_token_loader, + role_arn, + extra_args=None, + cache=None, + expiry_window_seconds=None, + ): + """ + :type client_creator: callable + :param client_creator: A callable that creates a client taking + arguments like ``Session.create_client``. + + :type web_identity_token_loader: callable + :param web_identity_token_loader: A callable that takes no arguments + and returns a web identity token str. + + :type role_arn: str + :param role_arn: The ARN of the role to be assumed. + + :type extra_args: dict + :param extra_args: Any additional arguments to add to the assume + role request using the format of the botocore operation. + Possible keys include, but may not be limited to, + DurationSeconds, Policy, SerialNumber, ExternalId and + RoleSessionName. + + :type cache: dict + :param cache: An object that supports ``__getitem__``, + ``__setitem__``, and ``__contains__``. An example of this is + the ``JSONFileCache`` class in aws-cli. + + :type expiry_window_seconds: int + :param expiry_window_seconds: The amount of time, in seconds, + """ + self._web_identity_token_loader = web_identity_token_loader + + super().__init__( + client_creator, + role_arn, + extra_args=extra_args, + cache=cache, + expiry_window_seconds=expiry_window_seconds, + ) + + def _get_credentials(self): + """Get credentials by calling assume role.""" + kwargs = self._assume_role_kwargs() + # Assume role with web identity does not require credentials other than + # the token, explicitly configure the client to not sign requests. + config = Config(signature_version=UNSIGNED) + client = self._client_creator('sts', config=config) + response = client.assume_role_with_web_identity(**kwargs) + self._add_account_id_to_response(response) + return response + + def _assume_role_kwargs(self): + """Get the arguments for assume role based on current configuration.""" + assume_role_kwargs = deepcopy(self._assume_kwargs) + identity_token = self._web_identity_token_loader() + assume_role_kwargs['WebIdentityToken'] = identity_token + + return assume_role_kwargs + + +class CredentialProvider: + # A short name to identify the provider within botocore. + METHOD = None + + # A name to identify the provider for use in cross-sdk features like + # assume role's `credential_source` configuration option. These names + # are to be treated in a case-insensitive way. NOTE: any providers not + # implemented in botocore MUST prefix their canonical names with + # 'custom' or we DO NOT guarantee that it will work with any features + # that this provides. + CANONICAL_NAME = None + + def __init__(self, session=None): + self.session = session + + def load(self): + """ + Loads the credentials from their source & sets them on the object. + + Subclasses should implement this method (by reading from disk, the + environment, the network or wherever), returning ``True`` if they were + found & loaded. + + If not found, this method should return ``False``, indictating that the + ``CredentialResolver`` should fall back to the next available method. + + The default implementation does nothing, assuming the user has set the + ``access_key/secret_key/token`` themselves. + + :returns: Whether credentials were found & set + :rtype: Credentials + """ + return True + + def _extract_creds_from_mapping(self, mapping, *key_names): + found = [] + for key_name in key_names: + try: + found.append(mapping[key_name]) + except KeyError: + raise PartialCredentialsError( + provider=self.METHOD, cred_var=key_name + ) + return found + + +class ProcessProvider(CredentialProvider): + METHOD = 'custom-process' + + def __init__(self, profile_name, load_config, popen=subprocess.Popen): + self._profile_name = profile_name + self._load_config = load_config + self._loaded_config = None + self._popen = popen + + def load(self): + credential_process = self._credential_process + if credential_process is None: + return + + creds_dict = self._retrieve_credentials_using(credential_process) + if creds_dict.get('expiry_time') is not None: + return RefreshableCredentials.create_from_metadata( + creds_dict, + lambda: self._retrieve_credentials_using(credential_process), + self.METHOD, + ) + + return Credentials( + access_key=creds_dict['access_key'], + secret_key=creds_dict['secret_key'], + token=creds_dict.get('token'), + method=self.METHOD, + account_id=creds_dict.get('account_id'), + ) + + def _retrieve_credentials_using(self, credential_process): + # We're not using shell=True, so we need to pass the + # command and all arguments as a list. + process_list = compat_shell_split(credential_process) + p = self._popen( + process_list, stdout=subprocess.PIPE, stderr=subprocess.PIPE + ) + stdout, stderr = p.communicate() + if p.returncode != 0: + raise CredentialRetrievalError( + provider=self.METHOD, error_msg=stderr.decode('utf-8') + ) + parsed = botocore.compat.json.loads(stdout.decode('utf-8')) + version = parsed.get('Version', '<Version key not provided>') + if version != 1: + raise CredentialRetrievalError( + provider=self.METHOD, + error_msg=( + f"Unsupported version '{version}' for credential process " + f"provider, supported versions: 1" + ), + ) + try: + return { + 'access_key': parsed['AccessKeyId'], + 'secret_key': parsed['SecretAccessKey'], + 'token': parsed.get('SessionToken'), + 'expiry_time': parsed.get('Expiration'), + 'account_id': self._get_account_id(parsed), + } + except KeyError as e: + raise CredentialRetrievalError( + provider=self.METHOD, + error_msg=f"Missing required key in response: {e}", + ) + + @property + def _credential_process(self): + return self.profile_config.get('credential_process') + + @property + def profile_config(self): + if self._loaded_config is None: + self._loaded_config = self._load_config() + profile_config = self._loaded_config.get('profiles', {}).get( + self._profile_name, {} + ) + return profile_config + + def _get_account_id(self, parsed): + account_id = parsed.get('AccountId') + return account_id or self.profile_config.get('aws_account_id') + + +class InstanceMetadataProvider(CredentialProvider): + METHOD = 'iam-role' + CANONICAL_NAME = 'Ec2InstanceMetadata' + + def __init__(self, iam_role_fetcher): + self._role_fetcher = iam_role_fetcher + + def load(self): + fetcher = self._role_fetcher + # We do the first request, to see if we get useful data back. + # If not, we'll pass & move on to whatever's next in the credential + # chain. + metadata = fetcher.retrieve_iam_role_credentials() + if not metadata: + return None + logger.info( + 'Found credentials from IAM Role: %s', metadata['role_name'] + ) + # We manually set the data here, since we already made the request & + # have it. When the expiry is hit, the credentials will auto-refresh + # themselves. + creds = RefreshableCredentials.create_from_metadata( + metadata, + method=self.METHOD, + refresh_using=fetcher.retrieve_iam_role_credentials, + ) + return creds + + +class EnvProvider(CredentialProvider): + METHOD = 'env' + CANONICAL_NAME = 'Environment' + ACCESS_KEY = 'AWS_ACCESS_KEY_ID' + SECRET_KEY = 'AWS_SECRET_ACCESS_KEY' + # The token can come from either of these env var. + # AWS_SESSION_TOKEN is what other AWS SDKs have standardized on. + TOKENS = ['AWS_SECURITY_TOKEN', 'AWS_SESSION_TOKEN'] + EXPIRY_TIME = 'AWS_CREDENTIAL_EXPIRATION' + ACCOUNT_ID = 'AWS_ACCOUNT_ID' + + def __init__(self, environ=None, mapping=None): + """ + + :param environ: The environment variables (defaults to + ``os.environ`` if no value is provided). + :param mapping: An optional mapping of variable names to + environment variable names. Use this if you want to + change the mapping of access_key->AWS_ACCESS_KEY_ID, etc. + The dict can have up to 5 keys: + * ``access_key`` + * ``secret_key`` + * ``token`` + * ``expiry_time`` + * ``account_id`` + """ + if environ is None: + environ = os.environ + self.environ = environ + self._mapping = self._build_mapping(mapping) + + def _build_mapping(self, mapping): + # Mapping of variable name to env var name. + var_mapping = {} + if mapping is None: + # Use the class var default. + var_mapping['access_key'] = self.ACCESS_KEY + var_mapping['secret_key'] = self.SECRET_KEY + var_mapping['token'] = self.TOKENS + var_mapping['expiry_time'] = self.EXPIRY_TIME + var_mapping['account_id'] = self.ACCOUNT_ID + else: + var_mapping['access_key'] = mapping.get( + 'access_key', self.ACCESS_KEY + ) + var_mapping['secret_key'] = mapping.get( + 'secret_key', self.SECRET_KEY + ) + var_mapping['token'] = mapping.get('token', self.TOKENS) + if not isinstance(var_mapping['token'], list): + var_mapping['token'] = [var_mapping['token']] + var_mapping['expiry_time'] = mapping.get( + 'expiry_time', self.EXPIRY_TIME + ) + var_mapping['account_id'] = mapping.get( + 'account_id', self.ACCOUNT_ID + ) + return var_mapping + + def load(self): + """ + Search for credentials in explicit environment variables. + """ + + access_key = self.environ.get(self._mapping['access_key'], '') + + if access_key: + logger.info('Found credentials in environment variables.') + fetcher = self._create_credentials_fetcher() + credentials = fetcher(require_expiry=False) + + expiry_time = credentials['expiry_time'] + if expiry_time is not None: + expiry_time = parse(expiry_time) + return RefreshableCredentials( + credentials['access_key'], + credentials['secret_key'], + credentials['token'], + expiry_time, + refresh_using=fetcher, + method=self.METHOD, + account_id=credentials['account_id'], + ) + + return Credentials( + credentials['access_key'], + credentials['secret_key'], + credentials['token'], + method=self.METHOD, + account_id=credentials['account_id'], + ) + else: + return None + + def _create_credentials_fetcher(self): + mapping = self._mapping + method = self.METHOD + environ = self.environ + + def fetch_credentials(require_expiry=True): + credentials = {} + + access_key = environ.get(mapping['access_key'], '') + if not access_key: + raise PartialCredentialsError( + provider=method, cred_var=mapping['access_key'] + ) + credentials['access_key'] = access_key + + secret_key = environ.get(mapping['secret_key'], '') + if not secret_key: + raise PartialCredentialsError( + provider=method, cred_var=mapping['secret_key'] + ) + credentials['secret_key'] = secret_key + + credentials['token'] = None + for token_env_var in mapping['token']: + token = environ.get(token_env_var, '') + if token: + credentials['token'] = token + break + + credentials['expiry_time'] = None + expiry_time = environ.get(mapping['expiry_time'], '') + if expiry_time: + credentials['expiry_time'] = expiry_time + if require_expiry and not expiry_time: + raise PartialCredentialsError( + provider=method, cred_var=mapping['expiry_time'] + ) + + credentials['account_id'] = None + account_id = environ.get(mapping['account_id'], '') + if account_id: + credentials['account_id'] = account_id + + return credentials + + return fetch_credentials + + +class OriginalEC2Provider(CredentialProvider): + METHOD = 'ec2-credentials-file' + CANONICAL_NAME = 'Ec2Config' + + CRED_FILE_ENV = 'AWS_CREDENTIAL_FILE' + ACCESS_KEY = 'AWSAccessKeyId' + SECRET_KEY = 'AWSSecretKey' + + def __init__(self, environ=None, parser=None): + if environ is None: + environ = os.environ + if parser is None: + parser = parse_key_val_file + self._environ = environ + self._parser = parser + + def load(self): + """ + Search for a credential file used by original EC2 CLI tools. + """ + if 'AWS_CREDENTIAL_FILE' in self._environ: + full_path = os.path.expanduser( + self._environ['AWS_CREDENTIAL_FILE'] + ) + creds = self._parser(full_path) + if self.ACCESS_KEY in creds: + logger.info('Found credentials in AWS_CREDENTIAL_FILE.') + access_key = creds[self.ACCESS_KEY] + secret_key = creds[self.SECRET_KEY] + # EC2 creds file doesn't support session tokens. + return Credentials(access_key, secret_key, method=self.METHOD) + else: + return None + + +class SharedCredentialProvider(CredentialProvider): + METHOD = 'shared-credentials-file' + CANONICAL_NAME = 'SharedCredentials' + + ACCESS_KEY = 'aws_access_key_id' + SECRET_KEY = 'aws_secret_access_key' + # Same deal as the EnvProvider above. Botocore originally supported + # aws_security_token, but the SDKs are standardizing on aws_session_token + # so we support both. + TOKENS = ['aws_security_token', 'aws_session_token'] + ACCOUNT_ID = 'aws_account_id' + + def __init__(self, creds_filename, profile_name=None, ini_parser=None): + self._creds_filename = creds_filename + if profile_name is None: + profile_name = 'default' + self._profile_name = profile_name + if ini_parser is None: + ini_parser = botocore.configloader.raw_config_parse + self._ini_parser = ini_parser + + def load(self): + try: + available_creds = self._ini_parser(self._creds_filename) + except ConfigNotFound: + return None + if self._profile_name in available_creds: + config = available_creds[self._profile_name] + if self.ACCESS_KEY in config: + logger.info( + "Found credentials in shared credentials file: %s", + self._creds_filename, + ) + access_key, secret_key = self._extract_creds_from_mapping( + config, self.ACCESS_KEY, self.SECRET_KEY + ) + token = self._get_session_token(config) + account_id = self._get_account_id(config) + return Credentials( + access_key, + secret_key, + token, + method=self.METHOD, + account_id=account_id, + ) + + def _get_session_token(self, config): + for token_envvar in self.TOKENS: + if token_envvar in config: + return config[token_envvar] + + def _get_account_id(self, config): + return config.get(self.ACCOUNT_ID) + + +class ConfigProvider(CredentialProvider): + """INI based config provider with profile sections.""" + + METHOD = 'config-file' + CANONICAL_NAME = 'SharedConfig' + + ACCESS_KEY = 'aws_access_key_id' + SECRET_KEY = 'aws_secret_access_key' + # Same deal as the EnvProvider above. Botocore originally supported + # aws_security_token, but the SDKs are standardizing on aws_session_token + # so we support both. + TOKENS = ['aws_security_token', 'aws_session_token'] + ACCOUNT_ID = 'aws_account_id' + + def __init__(self, config_filename, profile_name, config_parser=None): + """ + + :param config_filename: The session configuration scoped to the current + profile. This is available via ``session.config``. + :param profile_name: The name of the current profile. + :param config_parser: A config parser callable. + + """ + self._config_filename = config_filename + self._profile_name = profile_name + if config_parser is None: + config_parser = botocore.configloader.load_config + self._config_parser = config_parser + + def load(self): + """ + If there is are credentials in the configuration associated with + the session, use those. + """ + try: + full_config = self._config_parser(self._config_filename) + except ConfigNotFound: + return None + if self._profile_name in full_config['profiles']: + profile_config = full_config['profiles'][self._profile_name] + if self.ACCESS_KEY in profile_config: + logger.info( + "Credentials found in config file: %s", + self._config_filename, + ) + access_key, secret_key = self._extract_creds_from_mapping( + profile_config, self.ACCESS_KEY, self.SECRET_KEY + ) + token = self._get_session_token(profile_config) + account_id = self._get_account_id(profile_config) + return Credentials( + access_key, + secret_key, + token, + method=self.METHOD, + account_id=account_id, + ) + else: + return None + + def _get_session_token(self, profile_config): + for token_name in self.TOKENS: + if token_name in profile_config: + return profile_config[token_name] + + def _get_account_id(self, config): + return config.get(self.ACCOUNT_ID) + + +class BotoProvider(CredentialProvider): + METHOD = 'boto-config' + CANONICAL_NAME = 'Boto2Config' + + BOTO_CONFIG_ENV = 'BOTO_CONFIG' + DEFAULT_CONFIG_FILENAMES = ['/etc/boto.cfg', '~/.boto'] + ACCESS_KEY = 'aws_access_key_id' + SECRET_KEY = 'aws_secret_access_key' + + def __init__(self, environ=None, ini_parser=None): + if environ is None: + environ = os.environ + if ini_parser is None: + ini_parser = botocore.configloader.raw_config_parse + self._environ = environ + self._ini_parser = ini_parser + + def load(self): + """ + Look for credentials in boto config file. + """ + if self.BOTO_CONFIG_ENV in self._environ: + potential_locations = [self._environ[self.BOTO_CONFIG_ENV]] + else: + potential_locations = self.DEFAULT_CONFIG_FILENAMES + for filename in potential_locations: + try: + config = self._ini_parser(filename) + except ConfigNotFound: + # Move on to the next potential config file name. + continue + if 'Credentials' in config: + credentials = config['Credentials'] + if self.ACCESS_KEY in credentials: + logger.info( + "Found credentials in boto config file: %s", filename + ) + access_key, secret_key = self._extract_creds_from_mapping( + credentials, self.ACCESS_KEY, self.SECRET_KEY + ) + return Credentials( + access_key, secret_key, method=self.METHOD + ) + + +class AssumeRoleProvider(CredentialProvider): + METHOD = 'assume-role' + # The AssumeRole provider is logically part of the SharedConfig and + # SharedCredentials providers. Since the purpose of the canonical name + # is to provide cross-sdk compatibility, calling code will need to be + # aware that either of those providers should be tied to the AssumeRole + # provider as much as possible. + CANONICAL_NAME = None + ROLE_CONFIG_VAR = 'role_arn' + WEB_IDENTITY_TOKE_FILE_VAR = 'web_identity_token_file' + # Credentials are considered expired (and will be refreshed) once the total + # remaining time left until the credentials expires is less than the + # EXPIRY_WINDOW. + EXPIRY_WINDOW_SECONDS = 60 * 15 + + def __init__( + self, + load_config, + client_creator, + cache, + profile_name, + prompter=getpass.getpass, + credential_sourcer=None, + profile_provider_builder=None, + ): + """ + :type load_config: callable + :param load_config: A function that accepts no arguments, and + when called, will return the full configuration dictionary + for the session (``session.full_config``). + + :type client_creator: callable + :param client_creator: A factory function that will create + a client when called. Has the same interface as + ``botocore.session.Session.create_client``. + + :type cache: dict + :param cache: An object that supports ``__getitem__``, + ``__setitem__``, and ``__contains__``. An example + of this is the ``JSONFileCache`` class in the CLI. + + :type profile_name: str + :param profile_name: The name of the profile. + + :type prompter: callable + :param prompter: A callable that returns input provided + by the user (i.e raw_input, getpass.getpass, etc.). + + :type credential_sourcer: CanonicalNameCredentialSourcer + :param credential_sourcer: A credential provider that takes a + configuration, which is used to provide the source credentials + for the STS call. + """ + #: The cache used to first check for assumed credentials. + #: This is checked before making the AssumeRole API + #: calls and can be useful if you have short lived + #: scripts and you'd like to avoid calling AssumeRole + #: until the credentials are expired. + self.cache = cache + self._load_config = load_config + # client_creator is a callable that creates function. + # It's basically session.create_client + self._client_creator = client_creator + self._profile_name = profile_name + self._prompter = prompter + # The _loaded_config attribute will be populated from the + # load_config() function once the configuration is actually + # loaded. The reason we go through all this instead of just + # requiring that the loaded_config be passed to us is to that + # we can defer configuration loaded until we actually try + # to load credentials (as opposed to when the object is + # instantiated). + self._loaded_config = {} + self._credential_sourcer = credential_sourcer + self._profile_provider_builder = profile_provider_builder + self._visited_profiles = [self._profile_name] + + def load(self): + self._loaded_config = self._load_config() + profiles = self._loaded_config.get('profiles', {}) + profile = profiles.get(self._profile_name, {}) + if self._has_assume_role_config_vars(profile): + return self._load_creds_via_assume_role(self._profile_name) + + def _has_assume_role_config_vars(self, profile): + return ( + self.ROLE_CONFIG_VAR in profile + and + # We need to ensure this provider doesn't look at a profile when + # the profile has configuration for web identity. Simply relying on + # the order in the credential chain is insufficient as it doesn't + # prevent the case when we're doing an assume role chain. + self.WEB_IDENTITY_TOKE_FILE_VAR not in profile + ) + + def _load_creds_via_assume_role(self, profile_name): + role_config = self._get_role_config(profile_name) + source_credentials = self._resolve_source_credentials( + role_config, profile_name + ) + + extra_args = {} + role_session_name = role_config.get('role_session_name') + if role_session_name is not None: + extra_args['RoleSessionName'] = role_session_name + + external_id = role_config.get('external_id') + if external_id is not None: + extra_args['ExternalId'] = external_id + + mfa_serial = role_config.get('mfa_serial') + if mfa_serial is not None: + extra_args['SerialNumber'] = mfa_serial + + duration_seconds = role_config.get('duration_seconds') + if duration_seconds is not None: + extra_args['DurationSeconds'] = duration_seconds + + fetcher = AssumeRoleCredentialFetcher( + client_creator=self._client_creator, + source_credentials=source_credentials, + role_arn=role_config['role_arn'], + extra_args=extra_args, + mfa_prompter=self._prompter, + cache=self.cache, + ) + refresher = fetcher.fetch_credentials + if mfa_serial is not None: + refresher = create_mfa_serial_refresher(refresher) + + # The initial credentials are empty and the expiration time is set + # to now so that we can delay the call to assume role until it is + # strictly needed. + return DeferredRefreshableCredentials( + method=self.METHOD, + refresh_using=refresher, + time_fetcher=_local_now, + ) + + def _get_role_config(self, profile_name): + """Retrieves and validates the role configuration for the profile.""" + profiles = self._loaded_config.get('profiles', {}) + + profile = profiles[profile_name] + source_profile = profile.get('source_profile') + role_arn = profile['role_arn'] + credential_source = profile.get('credential_source') + mfa_serial = profile.get('mfa_serial') + external_id = profile.get('external_id') + role_session_name = profile.get('role_session_name') + duration_seconds = profile.get('duration_seconds') + + role_config = { + 'role_arn': role_arn, + 'external_id': external_id, + 'mfa_serial': mfa_serial, + 'role_session_name': role_session_name, + 'source_profile': source_profile, + 'credential_source': credential_source, + } + + if duration_seconds is not None: + try: + role_config['duration_seconds'] = int(duration_seconds) + except ValueError: + pass + + # Either the credential source or the source profile must be + # specified, but not both. + if credential_source is not None and source_profile is not None: + raise InvalidConfigError( + error_msg=( + f'The profile "{profile_name}" contains both ' + 'source_profile and credential_source.' + ) + ) + elif credential_source is None and source_profile is None: + raise PartialCredentialsError( + provider=self.METHOD, + cred_var='source_profile or credential_source', + ) + elif credential_source is not None: + self._validate_credential_source(profile_name, credential_source) + else: + self._validate_source_profile(profile_name, source_profile) + + return role_config + + def _validate_credential_source(self, parent_profile, credential_source): + if self._credential_sourcer is None: + raise InvalidConfigError( + error_msg=( + f"The credential_source \"{credential_source}\" is specified " + f"in profile \"{parent_profile}\", " + f"but no source provider was configured." + ) + ) + if not self._credential_sourcer.is_supported(credential_source): + raise InvalidConfigError( + error_msg=( + f"The credential source \"{credential_source}\" referenced " + f"in profile \"{parent_profile}\" is not valid." + ) + ) + + def _source_profile_has_credentials(self, profile): + return any( + [ + self._has_static_credentials(profile), + self._has_assume_role_config_vars(profile), + ] + ) + + def _validate_source_profile( + self, parent_profile_name, source_profile_name + ): + profiles = self._loaded_config.get('profiles', {}) + if source_profile_name not in profiles: + raise InvalidConfigError( + error_msg=( + f"The source_profile \"{source_profile_name}\" referenced in " + f"the profile \"{parent_profile_name}\" does not exist." + ) + ) + + source_profile = profiles[source_profile_name] + + # Make sure we aren't going into an infinite loop. If we haven't + # visited the profile yet, we're good. + if source_profile_name not in self._visited_profiles: + return + + # If we have visited the profile and the profile isn't simply + # referencing itself, that's an infinite loop. + if source_profile_name != parent_profile_name: + raise InfiniteLoopConfigError( + source_profile=source_profile_name, + visited_profiles=self._visited_profiles, + ) + + # A profile is allowed to reference itself so that it can source + # static credentials and have configuration all in the same + # profile. This will only ever work for the top level assume + # role because the static credentials will otherwise take + # precedence. + if not self._has_static_credentials(source_profile): + raise InfiniteLoopConfigError( + source_profile=source_profile_name, + visited_profiles=self._visited_profiles, + ) + + def _has_static_credentials(self, profile): + static_keys = ['aws_secret_access_key', 'aws_access_key_id'] + return any(static_key in profile for static_key in static_keys) + + def _resolve_source_credentials(self, role_config, profile_name): + credential_source = role_config.get('credential_source') + if credential_source is not None: + return self._resolve_credentials_from_source( + credential_source, profile_name + ) + + source_profile = role_config['source_profile'] + self._visited_profiles.append(source_profile) + return self._resolve_credentials_from_profile(source_profile) + + def _resolve_credentials_from_profile(self, profile_name): + profiles = self._loaded_config.get('profiles', {}) + profile = profiles[profile_name] + + if ( + self._has_static_credentials(profile) + and not self._profile_provider_builder + ): + # This is only here for backwards compatibility. If this provider + # isn't given a profile provider builder we still want to be able + # to handle the basic static credential case as we would before the + # profile provider builder parameter was added. + return self._resolve_static_credentials_from_profile(profile) + elif self._has_static_credentials( + profile + ) or not self._has_assume_role_config_vars(profile): + profile_providers = self._profile_provider_builder.providers( + profile_name=profile_name, + disable_env_vars=True, + ) + profile_chain = CredentialResolver(profile_providers) + credentials = profile_chain.load_credentials() + if credentials is None: + error_message = ( + 'The source profile "%s" must have credentials.' + ) + raise InvalidConfigError( + error_msg=error_message % profile_name, + ) + return credentials + + return self._load_creds_via_assume_role(profile_name) + + def _resolve_static_credentials_from_profile(self, profile): + try: + return Credentials( + access_key=profile['aws_access_key_id'], + secret_key=profile['aws_secret_access_key'], + token=profile.get('aws_session_token'), + ) + except KeyError as e: + raise PartialCredentialsError( + provider=self.METHOD, cred_var=str(e) + ) + + def _resolve_credentials_from_source( + self, credential_source, profile_name + ): + credentials = self._credential_sourcer.source_credentials( + credential_source + ) + if credentials is None: + raise CredentialRetrievalError( + provider=credential_source, + error_msg=( + 'No credentials found in credential_source referenced ' + f'in profile {profile_name}' + ), + ) + return credentials + + +class AssumeRoleWithWebIdentityProvider(CredentialProvider): + METHOD = 'assume-role-with-web-identity' + CANONICAL_NAME = None + _CONFIG_TO_ENV_VAR = { + 'web_identity_token_file': 'AWS_WEB_IDENTITY_TOKEN_FILE', + 'role_session_name': 'AWS_ROLE_SESSION_NAME', + 'role_arn': 'AWS_ROLE_ARN', + } + + def __init__( + self, + load_config, + client_creator, + profile_name, + cache=None, + disable_env_vars=False, + token_loader_cls=None, + ): + self.cache = cache + self._load_config = load_config + self._client_creator = client_creator + self._profile_name = profile_name + self._profile_config = None + self._disable_env_vars = disable_env_vars + if token_loader_cls is None: + token_loader_cls = FileWebIdentityTokenLoader + self._token_loader_cls = token_loader_cls + + def load(self): + return self._assume_role_with_web_identity() + + def _get_profile_config(self, key): + if self._profile_config is None: + loaded_config = self._load_config() + profiles = loaded_config.get('profiles', {}) + self._profile_config = profiles.get(self._profile_name, {}) + return self._profile_config.get(key) + + def _get_env_config(self, key): + if self._disable_env_vars: + return None + env_key = self._CONFIG_TO_ENV_VAR.get(key) + if env_key and env_key in os.environ: + return os.environ[env_key] + return None + + def _get_config(self, key): + env_value = self._get_env_config(key) + if env_value is not None: + return env_value + return self._get_profile_config(key) + + def _assume_role_with_web_identity(self): + token_path = self._get_config('web_identity_token_file') + if not token_path: + return None + token_loader = self._token_loader_cls(token_path) + + role_arn = self._get_config('role_arn') + if not role_arn: + error_msg = ( + 'The provided profile or the current environment is ' + 'configured to assume role with web identity but has no ' + 'role ARN configured. Ensure that the profile has the role_arn' + 'configuration set or the AWS_ROLE_ARN env var is set.' + ) + raise InvalidConfigError(error_msg=error_msg) + + extra_args = {} + role_session_name = self._get_config('role_session_name') + if role_session_name is not None: + extra_args['RoleSessionName'] = role_session_name + + fetcher = AssumeRoleWithWebIdentityCredentialFetcher( + client_creator=self._client_creator, + web_identity_token_loader=token_loader, + role_arn=role_arn, + extra_args=extra_args, + cache=self.cache, + ) + # The initial credentials are empty and the expiration time is set + # to now so that we can delay the call to assume role until it is + # strictly needed. + return DeferredRefreshableCredentials( + method=self.METHOD, + refresh_using=fetcher.fetch_credentials, + ) + + +class CanonicalNameCredentialSourcer: + def __init__(self, providers): + self._providers = providers + + def is_supported(self, source_name): + """Validates a given source name. + + :type source_name: str + :param source_name: The value of credential_source in the config + file. This is the canonical name of the credential provider. + + :rtype: bool + :returns: True if the credential provider is supported, + False otherwise. + """ + return source_name in [p.CANONICAL_NAME for p in self._providers] + + def source_credentials(self, source_name): + """Loads source credentials based on the provided configuration. + + :type source_name: str + :param source_name: The value of credential_source in the config + file. This is the canonical name of the credential provider. + + :rtype: Credentials + """ + source = self._get_provider(source_name) + if isinstance(source, CredentialResolver): + return source.load_credentials() + return source.load() + + def _get_provider(self, canonical_name): + """Return a credential provider by its canonical name. + + :type canonical_name: str + :param canonical_name: The canonical name of the provider. + + :raises UnknownCredentialError: Raised if no + credential provider by the provided name + is found. + """ + provider = self._get_provider_by_canonical_name(canonical_name) + + # The AssumeRole provider should really be part of the SharedConfig + # provider rather than being its own thing, but it is not. It is + # effectively part of both the SharedConfig provider and the + # SharedCredentials provider now due to the way it behaves. + # Therefore if we want either of those providers we should return + # the AssumeRole provider with it. + if canonical_name.lower() in ['sharedconfig', 'sharedcredentials']: + assume_role_provider = self._get_provider_by_method('assume-role') + if assume_role_provider is not None: + # The SharedConfig or SharedCredentials provider may not be + # present if it was removed for some reason, but the + # AssumeRole provider could still be present. In that case, + # return the assume role provider by itself. + if provider is None: + return assume_role_provider + + # If both are present, return them both as a + # CredentialResolver so that calling code can treat them as + # a single entity. + return CredentialResolver([assume_role_provider, provider]) + + if provider is None: + raise UnknownCredentialError(name=canonical_name) + + return provider + + def _get_provider_by_canonical_name(self, canonical_name): + """Return a credential provider by its canonical name. + + This function is strict, it does not attempt to address + compatibility issues. + """ + for provider in self._providers: + name = provider.CANONICAL_NAME + # Canonical names are case-insensitive + if name and name.lower() == canonical_name.lower(): + return provider + + def _get_provider_by_method(self, method): + """Return a credential provider by its METHOD name.""" + for provider in self._providers: + if provider.METHOD == method: + return provider + + +class ContainerProvider(CredentialProvider): + METHOD = 'container-role' + CANONICAL_NAME = 'EcsContainer' + ENV_VAR = 'AWS_CONTAINER_CREDENTIALS_RELATIVE_URI' + ENV_VAR_FULL = 'AWS_CONTAINER_CREDENTIALS_FULL_URI' + ENV_VAR_AUTH_TOKEN = 'AWS_CONTAINER_AUTHORIZATION_TOKEN' + ENV_VAR_AUTH_TOKEN_FILE = 'AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE' + + def __init__(self, environ=None, fetcher=None): + if environ is None: + environ = os.environ + if fetcher is None: + fetcher = ContainerMetadataFetcher() + self._environ = environ + self._fetcher = fetcher + + def load(self): + # This cred provider is only triggered if the self.ENV_VAR is set, + # which only happens if you opt into this feature. + if self.ENV_VAR in self._environ or self.ENV_VAR_FULL in self._environ: + return self._retrieve_or_fail() + + def _retrieve_or_fail(self): + if self._provided_relative_uri(): + full_uri = self._fetcher.full_url(self._environ[self.ENV_VAR]) + else: + full_uri = self._environ[self.ENV_VAR_FULL] + fetcher = self._create_fetcher(full_uri) + creds = fetcher() + return RefreshableCredentials( + access_key=creds['access_key'], + secret_key=creds['secret_key'], + token=creds['token'], + method=self.METHOD, + expiry_time=_parse_if_needed(creds['expiry_time']), + refresh_using=fetcher, + account_id=creds.get('account_id'), + ) + + def _build_headers(self): + auth_token = None + if self.ENV_VAR_AUTH_TOKEN_FILE in self._environ: + auth_token_file_path = self._environ[self.ENV_VAR_AUTH_TOKEN_FILE] + with open(auth_token_file_path) as token_file: + auth_token = token_file.read() + elif self.ENV_VAR_AUTH_TOKEN in self._environ: + auth_token = self._environ[self.ENV_VAR_AUTH_TOKEN] + if auth_token is not None: + self._validate_auth_token(auth_token) + return {'Authorization': auth_token} + + def _validate_auth_token(self, auth_token): + if "\r" in auth_token or "\n" in auth_token: + raise ValueError("Auth token value is not a legal header value") + + def _create_fetcher(self, full_uri, *args, **kwargs): + def fetch_creds(): + try: + headers = self._build_headers() + response = self._fetcher.retrieve_full_uri( + full_uri, headers=headers + ) + except MetadataRetrievalError as e: + logger.debug( + "Error retrieving container metadata: %s", e, exc_info=True + ) + raise CredentialRetrievalError( + provider=self.METHOD, error_msg=str(e) + ) + return { + 'access_key': response['AccessKeyId'], + 'secret_key': response['SecretAccessKey'], + 'token': response['Token'], + 'expiry_time': response['Expiration'], + 'account_id': response.get('AccountId'), + } + + return fetch_creds + + def _provided_relative_uri(self): + return self.ENV_VAR in self._environ + + +class CredentialResolver: + def __init__(self, providers): + """ + + :param providers: A list of ``CredentialProvider`` instances. + + """ + self.providers = providers + + def insert_before(self, name, credential_provider): + """ + Inserts a new instance of ``CredentialProvider`` into the chain that + will be tried before an existing one. + + :param name: The short name of the credentials you'd like to insert the + new credentials before. (ex. ``env`` or ``config``). Existing names + & ordering can be discovered via ``self.available_methods``. + :type name: string + + :param cred_instance: An instance of the new ``Credentials`` object + you'd like to add to the chain. + :type cred_instance: A subclass of ``Credentials`` + """ + try: + offset = [p.METHOD for p in self.providers].index(name) + except ValueError: + raise UnknownCredentialError(name=name) + self.providers.insert(offset, credential_provider) + + def insert_after(self, name, credential_provider): + """ + Inserts a new type of ``Credentials`` instance into the chain that will + be tried after an existing one. + + :param name: The short name of the credentials you'd like to insert the + new credentials after. (ex. ``env`` or ``config``). Existing names + & ordering can be discovered via ``self.available_methods``. + :type name: string + + :param cred_instance: An instance of the new ``Credentials`` object + you'd like to add to the chain. + :type cred_instance: A subclass of ``Credentials`` + """ + offset = self._get_provider_offset(name) + self.providers.insert(offset + 1, credential_provider) + + def remove(self, name): + """ + Removes a given ``Credentials`` instance from the chain. + + :param name: The short name of the credentials instance to remove. + :type name: string + """ + available_methods = [p.METHOD for p in self.providers] + if name not in available_methods: + # It's not present. Fail silently. + return + + offset = available_methods.index(name) + self.providers.pop(offset) + + def get_provider(self, name): + """Return a credential provider by name. + + :type name: str + :param name: The name of the provider. + + :raises UnknownCredentialError: Raised if no + credential provider by the provided name + is found. + """ + return self.providers[self._get_provider_offset(name)] + + def _get_provider_offset(self, name): + try: + return [p.METHOD for p in self.providers].index(name) + except ValueError: + raise UnknownCredentialError(name=name) + + def load_credentials(self): + """ + Goes through the credentials chain, returning the first ``Credentials`` + that could be loaded. + """ + # First provider to return a non-None response wins. + for provider in self.providers: + logger.debug("Looking for credentials via: %s", provider.METHOD) + creds = provider.load() + if creds is not None: + return creds + + # If we got here, no credentials could be found. + # This feels like it should be an exception, but historically, ``None`` + # is returned. + # + # +1 + # -js + return None + + +class SSOCredentialFetcher(CachedCredentialFetcher): + _UTC_DATE_FORMAT = '%Y-%m-%dT%H:%M:%SZ' + + def __init__( + self, + start_url, + sso_region, + role_name, + account_id, + client_creator, + token_loader=None, + cache=None, + expiry_window_seconds=None, + token_provider=None, + sso_session_name=None, + ): + self._client_creator = client_creator + self._sso_region = sso_region + self._role_name = role_name + self._account_id = account_id + self._start_url = start_url + self._token_loader = token_loader + self._token_provider = token_provider + self._sso_session_name = sso_session_name + super().__init__(cache, expiry_window_seconds) + + def _create_cache_key(self): + """Create a predictable cache key for the current configuration. + + The cache key is intended to be compatible with file names. + """ + args = { + 'roleName': self._role_name, + 'accountId': self._account_id, + } + if self._sso_session_name: + args['sessionName'] = self._sso_session_name + else: + args['startUrl'] = self._start_url + # NOTE: It would be good to hoist this cache key construction logic + # into the CachedCredentialFetcher class as we should be consistent. + # Unfortunately, the current assume role fetchers that sub class don't + # pass separators resulting in non-minified JSON. In the long term, + # all fetchers should use the below caching scheme. + args = json.dumps(args, sort_keys=True, separators=(',', ':')) + argument_hash = sha1(args.encode('utf-8')).hexdigest() + return self._make_file_safe(argument_hash) + + def _parse_timestamp(self, timestamp_ms): + # fromtimestamp expects seconds so: milliseconds / 1000 = seconds + timestamp_seconds = timestamp_ms / 1000.0 + timestamp = datetime.datetime.fromtimestamp(timestamp_seconds, tzutc()) + return timestamp.strftime(self._UTC_DATE_FORMAT) + + def _get_credentials(self): + """Get credentials by calling SSO get role credentials.""" + config = Config( + signature_version=UNSIGNED, + region_name=self._sso_region, + ) + client = self._client_creator('sso', config=config) + if self._token_provider: + initial_token_data = self._token_provider.load_token() + token = initial_token_data.get_frozen_token().token + else: + token = self._token_loader(self._start_url)['accessToken'] + + kwargs = { + 'roleName': self._role_name, + 'accountId': self._account_id, + 'accessToken': token, + } + try: + response = client.get_role_credentials(**kwargs) + except client.exceptions.UnauthorizedException: + raise UnauthorizedSSOTokenError() + credentials = response['roleCredentials'] + + credentials = { + 'ProviderType': 'sso', + 'Credentials': { + 'AccessKeyId': credentials['accessKeyId'], + 'SecretAccessKey': credentials['secretAccessKey'], + 'SessionToken': credentials['sessionToken'], + 'Expiration': self._parse_timestamp(credentials['expiration']), + 'AccountId': self._account_id, + }, + } + return credentials + + +class SSOProvider(CredentialProvider): + METHOD = 'sso' + + _SSO_TOKEN_CACHE_DIR = os.path.expanduser( + os.path.join('~', '.aws', 'sso', 'cache') + ) + _PROFILE_REQUIRED_CONFIG_VARS = ( + 'sso_role_name', + 'sso_account_id', + ) + _SSO_REQUIRED_CONFIG_VARS = ( + 'sso_start_url', + 'sso_region', + ) + _ALL_REQUIRED_CONFIG_VARS = ( + _PROFILE_REQUIRED_CONFIG_VARS + _SSO_REQUIRED_CONFIG_VARS + ) + + def __init__( + self, + load_config, + client_creator, + profile_name, + cache=None, + token_cache=None, + token_provider=None, + ): + if token_cache is None: + token_cache = JSONFileCache(self._SSO_TOKEN_CACHE_DIR) + self._token_cache = token_cache + self._token_provider = token_provider + if cache is None: + cache = {} + self.cache = cache + self._load_config = load_config + self._client_creator = client_creator + self._profile_name = profile_name + + def _load_sso_config(self): + loaded_config = self._load_config() + profiles = loaded_config.get('profiles', {}) + profile_name = self._profile_name + profile_config = profiles.get(self._profile_name, {}) + sso_sessions = loaded_config.get('sso_sessions', {}) + + # Role name & Account ID indicate the cred provider should be used + if all( + c not in profile_config for c in self._PROFILE_REQUIRED_CONFIG_VARS + ): + return None + + resolved_config, extra_reqs = self._resolve_sso_session_reference( + profile_config, sso_sessions + ) + + config = {} + missing_config_vars = [] + all_required_configs = self._ALL_REQUIRED_CONFIG_VARS + extra_reqs + for config_var in all_required_configs: + if config_var in resolved_config: + config[config_var] = resolved_config[config_var] + else: + missing_config_vars.append(config_var) + + if missing_config_vars: + missing = ', '.join(missing_config_vars) + raise InvalidConfigError( + error_msg=( + f'The profile "{profile_name}" is configured to use SSO ' + f'but is missing required configuration: {missing}' + ) + ) + return config + + def _resolve_sso_session_reference(self, profile_config, sso_sessions): + sso_session_name = profile_config.get('sso_session') + if sso_session_name is None: + # No reference to resolve, proceed with legacy flow + return profile_config, () + + if sso_session_name not in sso_sessions: + error_msg = f'The specified sso-session does not exist: "{sso_session_name}"' + raise InvalidConfigError(error_msg=error_msg) + + config = profile_config.copy() + session = sso_sessions[sso_session_name] + for config_var, val in session.items(): + # Validate any keys referenced in both profile and sso_session match + if config.get(config_var, val) != val: + error_msg = ( + f"The value for {config_var} is inconsistent between " + f"profile ({config[config_var]}) and sso-session ({val})." + ) + raise InvalidConfigError(error_msg=error_msg) + config[config_var] = val + return config, ('sso_session',) + + def load(self): + sso_config = self._load_sso_config() + if not sso_config: + return None + + fetcher_kwargs = { + 'start_url': sso_config['sso_start_url'], + 'sso_region': sso_config['sso_region'], + 'role_name': sso_config['sso_role_name'], + 'account_id': sso_config['sso_account_id'], + 'client_creator': self._client_creator, + 'token_loader': SSOTokenLoader(cache=self._token_cache), + 'cache': self.cache, + } + if 'sso_session' in sso_config: + fetcher_kwargs['sso_session_name'] = sso_config['sso_session'] + fetcher_kwargs['token_provider'] = self._token_provider + + sso_fetcher = SSOCredentialFetcher(**fetcher_kwargs) + + return DeferredRefreshableCredentials( + method=self.METHOD, + refresh_using=sso_fetcher.fetch_credentials, + ) |