1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131 |
- # 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 time
- import datetime
- import logging
- import os
- import getpass
- import threading
- import json
- import subprocess
- 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.configloader
- import botocore.compat
- from botocore import UNSIGNED
- from botocore.compat import total_seconds
- from botocore.compat import compat_shell_split
- from botocore.config import Config
- from botocore.exceptions import UnknownCredentialError
- from botocore.exceptions import PartialCredentialsError
- from botocore.exceptions import ConfigNotFound
- from botocore.exceptions import InvalidConfigError
- from botocore.exceptions import InfiniteLoopConfigError
- from botocore.exceptions import RefreshWithMFAUnsupportedError
- from botocore.exceptions import MetadataRetrievalError
- from botocore.exceptions import CredentialRetrievalError
- from botocore.exceptions import UnauthorizedSSOTokenError
- from botocore.utils import InstanceMetadataFetcher, parse_key_val_file
- from botocore.utils import ContainerMetadataFetcher
- from botocore.utils import FileWebIdentityTokenLoader
- from botocore.utils import SSOTokenLoader
- logger = logging.getLogger(__name__)
- ReadOnlyCredentials = namedtuple('ReadOnlyCredentials',
- ['access_key', 'secret_key', 'token'])
- 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'),
- 'imds_use_ipv6': session.get_config_variable('imds_use_ipv6')
- }
- 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(object):
- """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,
- )
- 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(object):
- 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 JSONFileCache(object):
- """JSON file cache.
- This provides a dict like interface that stores JSON serializable
- objects.
- The objects are serialized to JSON and stored in a file. These
- values can be retrieved at a later time.
- """
- CACHE_DIR = os.path.expanduser(os.path.join('~', '.aws', 'boto', 'cache'))
- def __init__(self, working_dir=CACHE_DIR, dumps_func=None):
- self._working_dir = working_dir
- if dumps_func is None:
- dumps_func = self._default_dumps
- self._dumps = dumps_func
- def _default_dumps(self, obj):
- return json.dumps(obj, default=_serialize_if_needed)
- def __contains__(self, cache_key):
- actual_key = self._convert_cache_key(cache_key)
- return os.path.isfile(actual_key)
- def __getitem__(self, cache_key):
- """Retrieve value from a cache key."""
- actual_key = self._convert_cache_key(cache_key)
- try:
- with open(actual_key) as f:
- return json.load(f)
- except (OSError, ValueError, IOError):
- raise KeyError(cache_key)
- def __setitem__(self, cache_key, value):
- full_key = self._convert_cache_key(cache_key)
- try:
- file_content = self._dumps(value)
- except (TypeError, ValueError):
- raise ValueError("Value cannot be cached, must be "
- "JSON serializable: %s" % value)
- if not os.path.isdir(self._working_dir):
- os.makedirs(self._working_dir)
- with os.fdopen(os.open(full_key,
- os.O_WRONLY | os.O_CREAT, 0o600), 'w') as f:
- f.truncate()
- f.write(file_content)
- def _convert_cache_key(self, cache_key):
- full_path = os.path.join(self._working_dir, cache_key + '.json')
- return full_path
- class Credentials(object):
- """
- Holds the credentials needed to authenticate requests.
- :ivar access_key: The access key part of the credentials.
- :ivar secret_key: The secret key part of the credentials.
- :ivar token: The security token, valid only for session credentials.
- :ivar method: A string which identifies where the credentials
- were found.
- """
- def __init__(self, access_key, secret_key, token=None,
- method=None):
- self.access_key = access_key
- self.secret_key = secret_key
- self.token = token
- if method is None:
- method = 'explicit'
- self.method = method
- 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)
- class RefreshableCredentials(Credentials):
- """
- Holds the credentials needed to authenticate requests. In addition, it
- knows how to refresh itself.
- :ivar access_key: The access key part of the credentials.
- :ivar secret_key: The secret key part of the credentials.
- :ivar token: The security token, valid only for session credentials.
- :ivar method: A string which identifies where the credentials
- were found.
- """
- # The time at which we'll attempt to refresh, but not
- # block if someone else is refreshing.
- _advisory_refresh_timeout = 15 * 60
- # The time at which all threads will block waiting for
- # refreshed credentials.
- _mandatory_refresh_timeout = 10 * 60
- def __init__(self, access_key, secret_key, token,
- expiry_time, refresh_using, method,
- time_fetcher=_local_now):
- self._refresh_using = refresh_using
- self._access_key = access_key
- self._secret_key = secret_key
- self._token = token
- 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)
- self._normalize()
- 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):
- 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
- )
- 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
- 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 as e:
- 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)
- 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'])
- 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._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(DeferredRefreshableCredentials, self).refresh_needed(
- refresh_in
- )
- class CachedCredentialFetcher(object):
- 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.path.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)
- return {
- 'access_key': creds['AccessKeyId'],
- 'secret_key': creds['SecretAccessKey'],
- 'token': creds['SessionToken'],
- 'expiry_time': expiration,
- }
- 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(BaseAssumeRoleCredentialFetcher, self).__init__(
- cache, expiry_window_seconds
- )
- def _generate_assume_role_name(self):
- self._role_session_name = 'botocore-session-%s' % (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)
- 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(AssumeRoleCredentialFetcher, self).__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()
- return client.assume_role(**kwargs)
- 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 = 'Enter MFA code for %s: ' % 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(AssumeRoleWithWebIdentityCredentialFetcher, self).__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)
- return client.assume_role_with_web_identity(**kwargs)
- 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(object):
- # 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
- )
- 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=("Unsupported version '%s' for credential process "
- "provider, supported versions: 1" % version))
- try:
- return {
- 'access_key': parsed['AccessKeyId'],
- 'secret_key': parsed['SecretAccessKey'],
- 'token': parsed.get('SessionToken'),
- 'expiry_time': parsed.get('Expiration'),
- }
- except KeyError as e:
- raise CredentialRetrievalError(
- provider=self.METHOD,
- error_msg="Missing required key in response: %s" % e
- )
- @property
- def _credential_process(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.get('credential_process')
- 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.debug('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'
- 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 3 keys: ``access_key``, ``secret_key``,
- ``session_token``.
- """
- 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
- 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)
- 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
- )
- return Credentials(
- credentials['access_key'], credentials['secret_key'],
- credentials['token'], method=self.METHOD
- )
- 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'])
- 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']
- 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)
- return Credentials(access_key, secret_key, token,
- method=self.METHOD)
- def _get_session_token(self, config):
- for token_envvar in self.TOKENS:
- if token_envvar in config:
- return config[token_envvar]
- 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']
- 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)
- return Credentials(access_key, secret_key, token,
- method=self.METHOD)
- 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]
- 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=(
- 'The profile "%s" contains both source_profile and '
- 'credential_source.' % profile_name
- )
- )
- 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=(
- 'The credential_source "%s" is specified in profile "%s", '
- 'but no source provider was configured.' % (
- credential_source, parent_profile)
- ))
- if not self._credential_sourcer.is_supported(credential_source):
- raise InvalidConfigError(error_msg=(
- 'The credential source "%s" referenced in profile "%s" is not '
- 'valid.' % (credential_source, parent_profile)
- ))
- 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=(
- 'The source_profile "%s" referenced in '
- 'the profile "%s" does not exist.' % (
- source_profile_name, parent_profile_name)
- )
- )
- 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
- # handle the basic static credential case as we would before the
- # provile 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 '
- 'in profile %s' % 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(object):
- 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'
- 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]
- headers = self._build_headers()
- fetcher = self._create_fetcher(full_uri, headers)
- 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,
- )
- def _build_headers(self):
- headers = {}
- auth_token = self._environ.get(self.ENV_VAR_AUTH_TOKEN)
- if auth_token is not None:
- return {
- 'Authorization': auth_token
- }
- def _create_fetcher(self, full_uri, headers):
- def fetch_creds():
- try:
- 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'],
- }
- return fetch_creds
- def _provided_relative_uri(self):
- return self.ENV_VAR in self._environ
- class CredentialResolver(object):
- 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):
- 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
- super(SSOCredentialFetcher, self).__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 = {
- 'startUrl': self._start_url,
- 'roleName': self._role_name,
- 'accountId': self._account_id,
- }
- # 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)
- kwargs = {
- 'roleName': self._role_name,
- 'accountId': self._account_id,
- 'accessToken': self._token_loader(self._start_url),
- }
- 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']),
- }
- }
- return credentials
- class SSOProvider(CredentialProvider):
- METHOD = 'sso'
- _SSO_TOKEN_CACHE_DIR = os.path.expanduser(
- os.path.join('~', '.aws', 'sso', 'cache')
- )
- _SSO_CONFIG_VARS = [
- 'sso_start_url',
- 'sso_region',
- 'sso_role_name',
- 'sso_account_id',
- ]
- def __init__(self, load_config, client_creator, profile_name,
- cache=None, token_cache=None):
- if token_cache is None:
- token_cache = JSONFileCache(self._SSO_TOKEN_CACHE_DIR)
- self._token_cache = token_cache
- 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, {})
- if all(c not in profile_config for c in self._SSO_CONFIG_VARS):
- return None
- config = {}
- missing_config_vars = []
- for config_var in self._SSO_CONFIG_VARS:
- if config_var in profile_config:
- config[config_var] = profile_config[config_var]
- else:
- missing_config_vars.append(config_var)
- if missing_config_vars:
- missing = ', '.join(missing_config_vars)
- raise InvalidConfigError(
- error_msg=(
- 'The profile "%s" is configured to use SSO but is missing '
- 'required configuration: %s' % (profile_name, missing)
- )
- )
- return config
- def load(self):
- sso_config = self._load_sso_config()
- if not sso_config:
- return None
- sso_fetcher = SSOCredentialFetcher(
- sso_config['sso_start_url'],
- sso_config['sso_region'],
- sso_config['sso_role_name'],
- sso_config['sso_account_id'],
- self._client_creator,
- token_loader=SSOTokenLoader(cache=self._token_cache),
- cache=self.cache,
- )
- return DeferredRefreshableCredentials(
- method=self.METHOD,
- refresh_using=sso_fetcher.fetch_credentials,
- )
|