Repository: boto/botocore
Pull request #1313 (2017-11-11 00:53:55)

Support additional sources for assume role
This adds support for credential sources aside from profiles with
hard-coded credentials. Now customers will be able to point to
another assume role profile with `source_profile`. Additionally,
they will be able to specify a new `credential_source` parameter
to load source credentials from other credential providers. This
currently has three valid values:

* `Environment` - This loads source credentials from environment
variables.
* `Ec2InstanceMetadata` - This loads source credentials from EC2's
instance metadata service.
* `EcsContainer` - This loads source credentials from ECS's
container credentials.

Resolves aws/aws-cli#2938
Resolves aws/aws-cli#2590
Resolves aws/aws-cli#1390
Closes #946 
Closes #851 

Addresses some, but not all, of #811 

codecov-io: # [Codecov](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=h1) Report
> Merging [#1313](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=desc) into [develop](https://codecov.io/gh/boto/botocore/commit/48b6ab5b0f82a9253f8f70441536c2ece396ba23?src=pr&el=desc) will **increase** coverage by `0.01%`.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/boto/botocore/pull/1313/graphs/tree.svg?height=150&src=pr&token=QOLXw6XSza&width=650)](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##           develop    #1313      +/-   ##
===========================================
+ Coverage    98.05%   98.07%   +0.01%     
===========================================
  Files           45       45              
  Lines         7421     7538     +117     
===========================================
+ Hits          7277     7393     +116     
- Misses         144      145       +1
```


| [Impacted Files](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [botocore/exceptions.py](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=tree#diff-Ym90b2NvcmUvZXhjZXB0aW9ucy5weQ==) | `100% <100%> (ø)` | :arrow_up: |
| [botocore/credentials.py](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=tree#diff-Ym90b2NvcmUvY3JlZGVudGlhbHMucHk=) | `97.98% <100%> (+0.24%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=footer). Last update [48b6ab5...73d616b](https://codecov.io/gh/boto/botocore/pull/1313?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

JordonPhillips: Cool. I'm gonna leave this up for a couple of days to give people a chance to try it out / leave feedback.

Patch:
diff --git a/.changes/next-release/feature-Credentials-18380.json b/.changes/next-release/feature-Credentials-18380.json
new file mode 100644
index 000000000..5db0dcfd7
--- /dev/null
+++ b/.changes/next-release/feature-Credentials-18380.json
@@ -0,0 +1,5 @@
+{
+  "description": "When creating an assume role profile, you can now specify another assume role profile as the source. This allows for chaining assume role calls.",
+  "type": "feature",
+  "category": "Credentials"
+}
diff --git a/.changes/next-release/feature-Credentials-95494.json b/.changes/next-release/feature-Credentials-95494.json
new file mode 100644
index 000000000..2437fe9f3
--- /dev/null
+++ b/.changes/next-release/feature-Credentials-95494.json
@@ -0,0 +1,5 @@
+{
+  "type": "feature",
+  "category": "Credentials",
+  "description": "When creating an assume role profile, you can now specify a credential source outside of the config file using the `credential_source` key."
+}
diff --git a/botocore/credentials.py b/botocore/credentials.py
index fb0198773..b53707dda 100644
--- a/botocore/credentials.py
+++ b/botocore/credentials.py
@@ -18,6 +18,9 @@ import os
 import getpass
 import threading
 from collections import namedtuple
+from copy import deepcopy
+from hashlib import sha256
+import json
 
 from dateutil.parser import parse
 from dateutil.tz import tzlocal
@@ -29,6 +32,7 @@ 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
@@ -56,14 +60,24 @@ def create_credential_resolver(session):
     num_attempts = session.get_config_variable('metadata_service_num_attempts')
 
     env_provider = EnvProvider()
+    container_provider = ContainerProvider()
+    instance_metadata_provider = InstanceMetadataProvider(
+        iam_role_fetcher=InstanceMetadataFetcher(
+            timeout=metadata_timeout,
+            num_attempts=num_attempts)
+    )
+    assume_role_provider = AssumeRoleProvider(
+        load_config=lambda: session.full_config,
+        client_creator=session.create_client,
+        cache={},
+        profile_name=profile_name,
+        credential_sourcer=CanonicalNameCredentialSourcer([
+            env_provider, container_provider, instance_metadata_provider
+        ])
+    )
     providers = [
         env_provider,
-        AssumeRoleProvider(
-            load_config=lambda: session.full_config,
-            client_creator=session.create_client,
-            cache={},
-            profile_name=profile_name,
-        ),
+        assume_role_provider,
         SharedCredentialProvider(
             creds_filename=credential_file,
             profile_name=profile_name
@@ -73,12 +87,8 @@ def create_credential_resolver(session):
         ConfigProvider(config_filename=config_file, profile_name=profile_name),
         OriginalEC2Provider(),
         BotoProvider(),
-        ContainerProvider(),
-        InstanceMetadataProvider(
-            iam_role_fetcher=InstanceMetadataFetcher(
-                timeout=metadata_timeout,
-                num_attempts=num_attempts)
-        )
+        container_provider,
+        instance_metadata_provider
     ]
 
     explicit_profile = session.get_config_variable('profile',
@@ -143,13 +153,23 @@ def create_assume_role_refresher(client, params):
     return refresh
 
 
-def create_mfa_serial_refresher():
-    def _refresher():
-        # 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()
-    return _refresher
+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 Credentials(object):
@@ -196,8 +216,6 @@ class RefreshableCredentials(Credentials):
     Holds the credentials needed to authenticate requests. In addition, it
     knows how to refresh itself.
 
-    :ivar refresh_timeout: How long a given set of credentials are valid for.
-        Useful for credentials fetched over the network.
     :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.
@@ -436,11 +454,220 @@ class RefreshableCredentials(Credentials):
         return self._frozen_credentials
 
 
-class CredentialProvider(object):
+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 any(part is None for part in [self._access_key, self._secret_key]):
+            return True
+        return super(DeferredRefreshableCredentials, self).refresh_needed(
+            refresh_in
+        )
+
+
+class CachedCredentialFetcher(object):
+    def __init__(self, cache=None, expiry_window_seconds=60 * 15):
+        if cache is None:
+            cache = {}
+        self._cache = cache
+        self._cache_key = self._create_cache_key()
+        self._expiry_window_seconds = expiry_window_seconds
+
+    def _create_cache_key(self):
+        raise NotImplementedError('_create_cache_key()')
+
+    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'])
+        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
+
 
-    # Implementations must provide a method.
+class AssumeRoleCredentialFetcher(CachedCredentialFetcher):
+    def __init__(self, client_creator, source_credentials, role_arn,
+                 extra_args=None, mfa_prompter=None, cache=None,
+                 expiry_window_seconds=60 * 15):
+        """
+        :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._client_creator = client_creator
+        self._source_credentials = source_credentials
+        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._role_session_name = 'botocore-session-%s' % (
+                int(time.time()))
+            self._assume_kwargs['RoleSessionName'] = self._role_session_name
+            self._using_default_session_name = True
+
+        self._mfa_prompter = mfa_prompter
+        if self._mfa_prompter is None:
+            self._mfa_prompter = getpass.getpass
+
+        super(AssumeRoleCredentialFetcher, 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 = 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 = sha256(args.encode('utf-8')).hexdigest()
+
+        # Prefix the cache key with the role arn to aid human-readability.
+        # Replace : and path sep to make it file safe.
+        arn = self._role_arn.replace(':', '_').replace(os.path.sep, '_')
+        return '%s--%s' % (arn, argument_hash)
+
+    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 = 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 = deepcopy(assume_role_kwargs)
+            assume_role_kwargs['TokenCode'] = token_code
+
+        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 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
 
@@ -459,7 +686,7 @@ class CredentialProvider(object):
         ``access_key/secret_key/token`` themselves.
 
         :returns: Whether credentials were found & set
-        :rtype: boolean
+        :rtype: Credentials
         """
         return True
 
@@ -476,6 +703,7 @@ class CredentialProvider(object):
 
 class InstanceMetadataProvider(CredentialProvider):
     METHOD = 'iam-role'
+    CANONICAL_NAME = 'Ec2InstanceMetadata'
 
     def __init__(self, iam_role_fetcher):
         self._role_fetcher = iam_role_fetcher
@@ -489,7 +717,7 @@ class InstanceMetadataProvider(CredentialProvider):
         if not metadata:
             return None
         logger.debug('Found credentials from IAM Role: %s',
-                    metadata['role_name'])
+                     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.
@@ -503,6 +731,7 @@ class InstanceMetadataProvider(CredentialProvider):
 
 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.
@@ -613,6 +842,7 @@ class EnvProvider(CredentialProvider):
 
 class OriginalEC2Provider(CredentialProvider):
     METHOD = 'ec2-credentials-file'
+    CANONICAL_NAME = 'Ec2Config'
 
     CRED_FILE_ENV = 'AWS_CREDENTIAL_FILE'
     ACCESS_KEY = 'AWSAccessKeyId'
@@ -631,7 +861,8 @@ class OriginalEC2Provider(CredentialProvider):
         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'])
+            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.')
@@ -645,6 +876,7 @@ class OriginalEC2Provider(CredentialProvider):
 
 class SharedCredentialProvider(CredentialProvider):
     METHOD = 'shared-credentials-file'
+    CANONICAL_NAME = 'SharedCredentials'
 
     ACCESS_KEY = 'aws_access_key_id'
     SECRET_KEY = 'aws_secret_access_key'
@@ -687,6 +919,7 @@ class SharedCredentialProvider(CredentialProvider):
 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'
@@ -740,6 +973,7 @@ class ConfigProvider(CredentialProvider):
 
 class BotoProvider(CredentialProvider):
     METHOD = 'boto-config'
+    CANONICAL_NAME = 'Boto2Config'
 
     BOTO_CONFIG_ENV = 'BOTO_CONFIG'
     DEFAULT_CONFIG_FILENAMES = ['/etc/boto.cfg', '~/.boto']
@@ -780,8 +1014,13 @@ class BotoProvider(CredentialProvider):
 
 
 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'
     # Credentials are considered expired (and will be refreshed) once the total
     # remaining time left until the credentials expires is less than the
@@ -789,9 +1028,8 @@ class AssumeRoleProvider(CredentialProvider):
     EXPIRY_WINDOW_SECONDS = 60 * 15
 
     def __init__(self, load_config, client_creator, cache, profile_name,
-                 prompter=getpass.getpass):
+                 prompter=getpass.getpass, credential_sourcer=None):
         """
-
         :type load_config: callable
         :param load_config: A function that accepts no arguments, and
             when called, will return the full configuration dictionary
@@ -802,10 +1040,10 @@ class AssumeRoleProvider(CredentialProvider):
             a client when called.  Has the same interface as
             ``botocore.session.Session.create_client``.
 
-        :type cache: JSONFileCache
+        :type cache: dict
         :param cache: An object that supports ``__getitem__``,
             ``__setitem__``, and ``__contains__``.  An example
-            of this is the ``JSONFileCache`` class.
+            of this is the ``JSONFileCache`` class in the CLI.
 
         :type profile_name: str
         :param profile_name: The name of the profile.
@@ -814,6 +1052,10 @@ class AssumeRoleProvider(CredentialProvider):
         :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
@@ -835,161 +1077,290 @@ class AssumeRoleProvider(CredentialProvider):
         # to load credentials (as opposed to when the object is
         # instantiated).
         self._loaded_config = {}
+        self._credential_sourcer = credential_sourcer
+        self._visited_profiles = [self._profile_name]
 
     def load(self):
         self._loaded_config = self._load_config()
         if self._has_assume_role_config_vars():
-            return self._load_creds_via_assume_role()
+            return self._load_creds_via_assume_role(self._profile_name)
 
     def _has_assume_role_config_vars(self):
         profiles = self._loaded_config.get('profiles', {})
         return self.ROLE_CONFIG_VAR in profiles.get(self._profile_name, {})
 
-    def _load_creds_via_assume_role(self):
-        # We can get creds in one of two ways:
-        # * It can either be cached on disk from an pre-existing session
-        # * Cache doesn't have the creds (or is expired) so we need to make
-        #   an assume role call to get temporary creds, which we then cache
-        #   for subsequent requests.
-        creds = self._load_creds_from_cache()
-        if creds is not None:
-            logger.debug("Credentials for role retrieved from cache.")
-            return creds
-        else:
-            # We get the Credential used by botocore as well
-            # as the original parsed response from the server.
-            creds, response = self._retrieve_temp_credentials()
-            cache_key = self._create_cache_key()
-            self._write_cached_credentials(response, cache_key)
-            return creds
-
-    def _load_creds_from_cache(self):
-        cache_key = self._create_cache_key()
-        try:
-            from_cache = self.cache[cache_key]
-            if self._is_expired(from_cache):
-                # Don't need to delete the cache entry,
-                # when we refresh via AssumeRole, we'll
-                # update the cache with the new entry.
-                logger.debug(
-                    "Credentials were found in cache, but they are expired.")
-                return None
-            else:
-                return self._create_creds_from_response(from_cache)
-        except KeyError:
-            return None
-
-    def _is_expired(self, credentials):
-        end_time = _parse_if_needed(credentials['Credentials']['Expiration'])
-        now = datetime.datetime.now(tzlocal())
-        seconds = total_seconds(end_time - now)
-        return seconds < self.EXPIRY_WINDOW_SECONDS
+    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
+        )
 
-    def _create_cache_key(self):
-        role_config = self._get_role_config_values()
-        # On windows, ':' is not allowed in filenames, so we'll
-        # replace them with '_' instead.
-        role_arn = role_config['role_arn'].replace(':', '_')
+        extra_args = {}
         role_session_name = role_config.get('role_session_name')
-        if role_session_name:
-            cache_key = '%s--%s--%s' % (self._profile_name, role_arn,
-                                        role_session_name)
-        else:
-            cache_key = '%s--%s' % (self._profile_name, role_arn)
+        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
+
+        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
+        )
 
-        return cache_key.replace('/', '-')
+    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')
 
-    def _write_cached_credentials(self, creds, cache_key):
-        self.cache[cache_key] = creds
+        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
+        }
 
-    def _get_role_config_values(self):
-        # This returns the role related configuration.
+        # 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 _validate_source_profile(self, parent_profile, source_profile):
         profiles = self._loaded_config.get('profiles', {})
-        try:
-            source_profile = profiles[self._profile_name]['source_profile']
-            role_arn = profiles[self._profile_name]['role_arn']
-            mfa_serial = profiles[self._profile_name].get('mfa_serial')
-        except KeyError as e:
-            raise PartialCredentialsError(provider=self.METHOD,
-                                          cred_var=str(e))
-        external_id = profiles[self._profile_name].get('external_id')
-        role_session_name = \
-            profiles[self._profile_name].get('role_session_name')
         if source_profile not in profiles:
             raise InvalidConfigError(
                 error_msg=(
                     'The source_profile "%s" referenced in '
                     'the profile "%s" does not exist.' % (
-                        source_profile, self._profile_name)))
-        source_cred_values = profiles[source_profile]
-        return {
-            'role_arn': role_arn,
-            'external_id': external_id,
-            'source_profile': source_profile,
-            'mfa_serial': mfa_serial,
-            'source_cred_values': source_cred_values,
-            'role_session_name': role_session_name
-        }
+                        source_profile, parent_profile)
+                )
+            )
 
-    def _create_creds_from_response(self, response):
-        config = self._get_role_config_values()
-        if config.get('mfa_serial') is not None:
-            # MFA would require getting a new TokenCode which would require
-            # prompting the user for a new token, so we use a different
-            # refresh_func.
-            refresh_func = create_mfa_serial_refresher()
-        else:
-            refresh_func = create_assume_role_refresher(
-                self._create_client_from_config(config),
-                self._assume_role_base_kwargs(config))
-        return RefreshableCredentials(
-            access_key=response['Credentials']['AccessKeyId'],
-            secret_key=response['Credentials']['SecretAccessKey'],
-            token=response['Credentials']['SessionToken'],
-            method=self.METHOD,
-            expiry_time=_parse_if_needed(
-                response['Credentials']['Expiration']),
-            refresh_using=refresh_func)
-
-    def _create_client_from_config(self, config):
-        source_cred_values = config['source_cred_values']
-        client = self._client_creator(
-            'sts', aws_access_key_id=source_cred_values['aws_access_key_id'],
-            aws_secret_access_key=source_cred_values['aws_secret_access_key'],
-            aws_session_token=source_cred_values.get('aws_session_token'),
-        )
-        return client
-
-    def _retrieve_temp_credentials(self):
-        logger.debug("Retrieving credentials via AssumeRole.")
-        config = self._get_role_config_values()
-        client = self._create_client_from_config(config)
-
-        assume_role_kwargs = self._assume_role_base_kwargs(config)
-
-        response = client.assume_role(**assume_role_kwargs)
-        creds = self._create_creds_from_response(response)
-        return creds, response
-
-    def _assume_role_base_kwargs(self, config):
-        assume_role_kwargs = {'RoleArn': config['role_arn']}
-        if config['external_id'] is not None:
-            assume_role_kwargs['ExternalId'] = config['external_id']
-        if config['mfa_serial'] is not None:
-            token_code = self._prompter("Enter MFA code: ")
-            assume_role_kwargs['SerialNumber'] = config['mfa_serial']
-            assume_role_kwargs['TokenCode'] = token_code
-        if config['role_session_name'] is not None:
-            assume_role_kwargs['RoleSessionName'] = config['role_session_name']
-        else:
-            role_session_name = 'AWS-CLI-session-%s' % (int(time.time()))
-            assume_role_kwargs['RoleSessionName'] = role_session_name
-        return assume_role_kwargs
+        # Make sure we aren't going into an infinite loop. If we haven't
+        # visited the profile yet, we're good.
+        if source_profile 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 != parent_profile:
+            raise InfiniteLoopConfigError(
+                source_profile=source_profile,
+                visited_profiles=self._visited_profiles
+            )
 
-class ContainerProvider(CredentialProvider):
+        # 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(profiles[source_profile]):
+            raise InfiniteLoopConfigError(
+                source_profile=source_profile,
+                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):
+            return self._resolve_static_credentials_from_profile(profile)
+
+        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 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'
@@ -1049,6 +1420,7 @@ class ContainerProvider(CredentialProvider):
                 'token': response['Token'],
                 'expiry_time': response['Expiration'],
             }
+
         return fetch_creds
 
     def _provided_relative_uri(self):
@@ -1056,7 +1428,6 @@ class ContainerProvider(CredentialProvider):
 
 
 class CredentialResolver(object):
-
     def __init__(self, providers):
         """
 
diff --git a/botocore/exceptions.py b/botocore/exceptions.py
index d28b3bb36..5bd2498b0 100644
--- a/botocore/exceptions.py
+++ b/botocore/exceptions.py
@@ -427,6 +427,14 @@ class InvalidConfigError(BotoCoreError):
     fmt = '{error_msg}'
 
 
+class InfiniteLoopConfigError(InvalidConfigError):
+    fmt = (
+        'Infinite loop in credential configuration detected. Attempting to '
+        'load from profile {source_profile} which has already been visited. '
+        'Visited profiles: {visited_profiles}'
+    )
+
+
 class RefreshWithMFAUnsupportedError(BotoCoreError):
     fmt = 'Cannot refresh credentials: MFA token required.'
 


Test patch:
diff --git a/tests/functional/test_credentials.py b/tests/functional/test_credentials.py
index 5093346e8..4a0d3c59e 100644
--- a/tests/functional/test_credentials.py
+++ b/tests/functional/test_credentials.py
@@ -10,10 +10,27 @@
 # 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.
-from tests import unittest, IntegerRefresher
 import threading
+import os
 import math
 import time
+import mock
+import tempfile
+import shutil
+from datetime import datetime, timedelta
+
+from botocore.vendored import requests
+from dateutil.tz import tzlocal
+
+from tests import unittest, IntegerRefresher, BaseEnvVar, random_chars
+from botocore.credentials import EnvProvider, ContainerProvider
+from botocore.credentials import InstanceMetadataProvider
+from botocore.credentials import Credentials, ReadOnlyCredentials
+from botocore.credentials import AssumeRoleProvider
+from botocore.credentials import CanonicalNameCredentialSourcer
+from botocore.session import Session
+from botocore.exceptions import InvalidConfigError, InfiniteLoopConfigError
+from botocore.stub import Stubber
 
 
 class TestCredentialRefreshRaces(unittest.TestCase):
@@ -89,3 +106,297 @@ class TestCredentialRefreshRaces(unittest.TestCase):
                                   frozen.secret_key,
                                   frozen.token))
         self.assert_consistent_credentials_seen(creds, _run_in_thread)
+
+
+class TestAssumeRole(BaseEnvVar):
+    def setUp(self):
+        super(TestAssumeRole, self).setUp()
+        self.tempdir = tempfile.mkdtemp()
+        self.config_file = os.path.join(self.tempdir, 'config')
+        self.environ['AWS_CONFIG_FILE'] = self.config_file
+        self.environ['AWS_ACCESS_KEY_ID'] = 'access_key'
+        self.environ['AWS_SECRET_ACCESS_KEY'] = 'secret_key'
+
+        self.metadata_provider = self.mock_provider(InstanceMetadataProvider)
+        self.env_provider = self.mock_provider(EnvProvider)
+        self.container_provider = self.mock_provider(ContainerProvider)
+
+    def mock_provider(self, provider_cls):
+        mock_instance = mock.Mock(spec=provider_cls)
+        mock_instance.load.return_value = None
+        mock_instance.METHOD = provider_cls.METHOD
+        mock_instance.CANONICAL_NAME = provider_cls.CANONICAL_NAME
+        return mock_instance
+
+    def tearDown(self):
+        shutil.rmtree(self.tempdir)
+
+    def create_session(self, profile=None):
+        session = Session(profile=profile)
+
+        # We have to set bogus credentials here or otherwise we'll trigger
+        # an early credential chain resolution.
+        sts = session.create_client(
+            'sts',
+            aws_access_key_id='spam',
+            aws_secret_access_key='eggs',
+        )
+        stubber = Stubber(sts)
+        stubber.activate()
+        assume_role_provider = AssumeRoleProvider(
+            load_config=lambda: session.full_config,
+            client_creator=lambda *args, **kwargs: sts,
+            cache={},
+            profile_name=profile,
+            credential_sourcer=CanonicalNameCredentialSourcer([
+                self.env_provider, self.container_provider,
+                self.metadata_provider
+            ])
+        )
+
+        component_name = 'credential_provider'
+        resolver = session.get_component(component_name)
+        available_methods = [p.METHOD for p in resolver.providers]
+        replacements = {
+            'env': self.env_provider,
+            'iam-role': self.metadata_provider,
+            'container-role': self.container_provider,
+            'assume-role': assume_role_provider
+        }
+        for name, provider in replacements.items():
+            try:
+                index = available_methods.index(name)
+            except ValueError:
+                # The provider isn't in the session
+                continue
+
+            resolver.providers[index] = provider
+
+        session.register_component(
+            'credential_provider', resolver
+        )
+        return session, stubber
+
+    def create_assume_role_response(self, credentials, expiration=None):
+        if expiration is None:
+            expiration = self.some_future_time()
+
+        response = {
+            'Credentials': {
+                'AccessKeyId': credentials.access_key,
+                'SecretAccessKey': credentials.secret_key,
+                'SessionToken': credentials.token,
+                'Expiration': expiration
+            },
+            'AssumedRoleUser': {
+                'AssumedRoleId': 'myroleid',
+                'Arn': 'arn:aws:iam::1234567890:user/myuser'
+            }
+        }
+
+        return response
+
+    def create_random_credentials(self):
+        return Credentials(
+            'fake-%s' % random_chars(15),
+            'fake-%s' % random_chars(35),
+            'fake-%s' % random_chars(45)
+        )
+
+    def some_future_time(self):
+        timeobj = datetime.now(tzlocal())
+        return timeobj + timedelta(hours=24)
+
+    def write_config(self, config):
+        with open(self.config_file, 'w') as f:
+            f.write(config)
+
+    def assert_creds_equal(self, c1, c2):
+        c1_frozen = c1
+        if not isinstance(c1_frozen, ReadOnlyCredentials):
+            c1_frozen = c1.get_frozen_credentials()
+        c2_frozen = c2
+        if not isinstance(c2_frozen, ReadOnlyCredentials):
+            c2_frozen = c2.get_frozen_credentials()
+        self.assertEqual(c1_frozen, c2_frozen)
+
+    def test_assume_role(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'source_profile = B\n\n'
+            '[profile B]\n'
+            'aws_access_key_id = abc123\n'
+            'aws_secret_access_key = def456\n'
+        )
+        self.write_config(config)
+
+        expected_creds = self.create_random_credentials()
+        response = self.create_assume_role_response(expected_creds)
+        session, stubber = self.create_session(profile='A')
+        stubber.add_response('assume_role', response)
+
+        actual_creds = session.get_credentials()
+        self.assert_creds_equal(actual_creds, expected_creds)
+        stubber.assert_no_pending_responses()
+
+    def test_environment_credential_source(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'credential_source = Environment\n'
+        )
+        self.write_config(config)
+
+        environment_creds = self.create_random_credentials()
+        self.env_provider.load.return_value = environment_creds
+
+        expected_creds = self.create_random_credentials()
+        response = self.create_assume_role_response(expected_creds)
+        session, stubber = self.create_session(profile='A')
+        stubber.add_response('assume_role', response)
+
+        actual_creds = session.get_credentials()
+        self.assert_creds_equal(actual_creds, expected_creds)
+
+        stubber.assert_no_pending_responses()
+        self.assertEqual(self.env_provider.load.call_count, 1)
+
+    def test_instance_metadata_credential_source(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'credential_source = Ec2InstanceMetadata\n'
+        )
+        self.write_config(config)
+
+        metadata_creds = self.create_random_credentials()
+        self.metadata_provider.load.return_value = metadata_creds
+
+        expected_creds = self.create_random_credentials()
+        response = self.create_assume_role_response(expected_creds)
+        session, stubber = self.create_session(profile='A')
+        stubber.add_response('assume_role', response)
+
+        actual_creds = session.get_credentials()
+        self.assert_creds_equal(actual_creds, expected_creds)
+
+        stubber.assert_no_pending_responses()
+        self.assertEqual(self.metadata_provider.load.call_count, 1)
+
+    def test_container_credential_source(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'credential_source = EcsContainer\n'
+        )
+        self.write_config(config)
+
+        container_creds = self.create_random_credentials()
+        self.container_provider.load.return_value = container_creds
+
+        expected_creds = self.create_random_credentials()
+        response = self.create_assume_role_response(expected_creds)
+        session, stubber = self.create_session(profile='A')
+        stubber.add_response('assume_role', response)
+
+        actual_creds = session.get_credentials()
+        self.assert_creds_equal(actual_creds, expected_creds)
+
+        stubber.assert_no_pending_responses()
+        self.assertEqual(self.container_provider.load.call_count, 1)
+
+    def test_invalid_credential_source(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'credential_source = CustomInvalidProvider\n'
+        )
+        self.write_config(config)
+
+        with self.assertRaises(InvalidConfigError):
+            session, _ = self.create_session(profile='A')
+            session.get_credentials()
+
+    def test_recursive_assume_role(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'source_profile = B\n\n'
+            '[profile B]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleB\n'
+            'source_profile = C\n\n'
+            '[profile C]\n'
+            'aws_access_key_id = abc123\n'
+            'aws_secret_access_key = def456\n'
+        )
+        self.write_config(config)
+
+        profile_b_creds = self.create_random_credentials()
+        profile_b_response = self.create_assume_role_response(profile_b_creds)
+        profile_a_creds = self.create_random_credentials()
+        profile_a_response = self.create_assume_role_response(profile_a_creds)
+
+        session, stubber = self.create_session(profile='A')
+        stubber.add_response('assume_role', profile_b_response)
+        stubber.add_response('assume_role', profile_a_response)
+
+        actual_creds = session.get_credentials()
+        self.assert_creds_equal(actual_creds, profile_a_creds)
+        stubber.assert_no_pending_responses()
+
+    def test_recursive_assume_role_stops_at_static_creds(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'source_profile = B\n\n'
+            '[profile B]\n'
+            'aws_access_key_id = abc123\n'
+            'aws_secret_access_key = def456\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleB\n'
+            'source_profile = C\n\n'
+            '[profile C]\n'
+            'aws_access_key_id = abc123\n'
+            'aws_secret_access_key = def456\n'
+        )
+        self.write_config(config)
+
+        profile_a_creds = self.create_random_credentials()
+        profile_a_response = self.create_assume_role_response(profile_a_creds)
+        session, stubber = self.create_session(profile='A')
+        stubber.add_response('assume_role', profile_a_response)
+
+        actual_creds = session.get_credentials()
+        self.assert_creds_equal(actual_creds, profile_a_creds)
+        stubber.assert_no_pending_responses()
+
+    def test_infinitely_recursive_assume_role(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'source_profile = A\n'
+        )
+        self.write_config(config)
+
+        with self.assertRaises(InfiniteLoopConfigError):
+            session, _ = self.create_session(profile='A')
+            session.get_credentials()
+
+    def test_self_referential_profile(self):
+        config = (
+            '[profile A]\n'
+            'role_arn = arn:aws:iam::123456789:role/RoleA\n'
+            'source_profile = A\n'
+            'aws_access_key_id = abc123\n'
+            'aws_secret_access_key = def456\n'
+        )
+        self.write_config(config)
+
+        expected_creds = self.create_random_credentials()
+        response = self.create_assume_role_response(expected_creds)
+        session, stubber = self.create_session(profile='A')
+        stubber.add_response('assume_role', response)
+
+        actual_creds = session.get_credentials()
+        self.assert_creds_equal(actual_creds, expected_creds)
+        stubber.assert_no_pending_responses()
diff --git a/tests/integration/test_credentials.py b/tests/integration/test_credentials.py
index 5198c92bb..7e45f2163 100644
--- a/tests/integration/test_credentials.py
+++ b/tests/integration/test_credentials.py
@@ -10,12 +10,19 @@
 # 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 os
 import mock
+import tempfile
+import shutil
+import json
+import time
 
 from botocore.session import Session
-from tests import BaseEnvVar, temporary_file
+from botocore.exceptions import ClientError
+from tests import BaseEnvVar, temporary_file, random_chars
+
+
+S3_READ_POLICY_ARN = 'arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess'
 
 
 class TestCredentialPrecedence(BaseEnvVar):
@@ -121,3 +128,227 @@ class TestCredentialPrecedence(BaseEnvVar):
 
             self.assertEqual(credentials.access_key, 'custom1')
             self.assertEqual(credentials.secret_key, 'custom2')
+
+
+class TestAssumeRoleCredentials(BaseEnvVar):
+    def setUp(self):
+        super(TestAssumeRoleCredentials, self).setUp()
+        self.parent_session = Session()
+        self.iam = self.parent_session.create_client('iam')
+        self.sts = self.parent_session.create_client('sts')
+        self.tempdir = tempfile.mkdtemp()
+        self.config_file = os.path.join(self.tempdir, 'config')
+
+        # A role trust policy that allows the current account to call assume
+        # role on itself.
+        account_id = self.sts.get_caller_identity()['Account']
+        self.role_policy = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {
+                        "AWS": "arn:aws:iam::%s:root" % account_id
+                    },
+                    "Action": "sts:AssumeRole"
+                }
+            ]
+        }
+
+    def tearDown(self):
+        super(TestAssumeRoleCredentials, self).tearDown()
+        shutil.rmtree(self.tempdir)
+
+    def random_name(self):
+        return 'botocoretest-' + random_chars(10)
+
+    def create_role(self, policy_document, policy_arn=None):
+        name = self.random_name()
+        response = self.iam.create_role(
+            RoleName=name,
+            AssumeRolePolicyDocument=json.dumps(policy_document)
+        )
+        self.addCleanup(self.iam.delete_role, RoleName=name)
+        if policy_arn:
+            self.iam.attach_role_policy(RoleName=name, PolicyArn=policy_arn)
+            self.addCleanup(
+                self.iam.detach_role_policy, RoleName=name,
+                PolicyArn=policy_arn
+            )
+        return response['Role']
+
+    def create_user(self, policy_arns):
+        name = self.random_name()
+        user = self.iam.create_user(UserName=name)['User']
+        self.addCleanup(self.iam.delete_user, UserName=name)
+
+        for arn in policy_arns:
+            self.iam.attach_user_policy(
+                UserName=name,
+                PolicyArn=arn
+            )
+            self.addCleanup(
+                self.iam.detach_user_policy,
+                UserName=name, PolicyArn=arn
+            )
+
+        return user
+
+    def create_creds(self, user_name):
+        creds = self.iam.create_access_key(UserName=user_name)['AccessKey']
+        self.addCleanup(
+            self.iam.delete_access_key,
+            UserName=user_name, AccessKeyId=creds['AccessKeyId']
+        )
+        return creds
+
+    def wait_for_assume_role(self, role_arn, access_key, secret_key,
+                             token=None, attempts=5, delay=5):
+        # "Why not use the policy simulator?" you might ask. The answer is
+        # that the policy simulator will return success far before you can
+        # actually make the calls.
+        client = self.parent_session.create_client(
+            'sts', aws_access_key_id=access_key,
+            aws_secret_access_key=secret_key, aws_session_token=token
+        )
+        attempts_remaining = attempts
+        role_session_name = random_chars(10)
+        while attempts_remaining > 0:
+            attempts_remaining -= 1
+            try:
+                result = client.assume_role(
+                    RoleArn=role_arn, RoleSessionName=role_session_name)
+                return result['Credentials']
+            except ClientError as e:
+                code = e.response.get('Error', {}).get('Code')
+                if code == "InvalidClientTokenId":
+                    time.sleep(delay)
+                else:
+                    raise
+
+        raise Exception("Unable to assume role %s" % role_arn)
+
+    def create_assume_policy(self, role_arn):
+        policy_document = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Resource": role_arn,
+                    "Action": "sts:AssumeRole"
+                }
+            ]
+        }
+        name = self.random_name()
+        response = self.iam.create_policy(
+            PolicyName=name,
+            PolicyDocument=json.dumps(policy_document)
+        )
+        self.addCleanup(
+            self.iam.delete_policy, PolicyArn=response['Policy']['Arn']
+        )
+        return response['Policy']['Arn']
+
+    def assert_s3_read_only_session(self, session):
+        # Calls to S3 should succeed
+        s3 = session.create_client('s3')
+        s3.list_buckets()
+
+        # Calls to other services should not
+        iam = session.create_client('iam')
+        try:
+            iam.list_groups()
+            self.fail("Expected call to list_groups to fail, but it passed.")
+        except ClientError as e:
+            code = e.response.get('Error', {}).get('Code')
+            if code != 'AccessDenied':
+                raise
+
+    def test_recursive_assume_role(self):
+        # Create the final role, the one that will actually have access to s3
+        final_role = self.create_role(self.role_policy, S3_READ_POLICY_ARN)
+
+        # Create the role that can assume the final role
+        middle_policy_arn = self.create_assume_policy(final_role['Arn'])
+        middle_role = self.create_role(self.role_policy, middle_policy_arn)
+
+        # Create a user that can only assume the middle-man role, and then get
+        # static credentials for it.
+        user_policy_arn = self.create_assume_policy(middle_role['Arn'])
+        user = self.create_user([user_policy_arn])
+        user_creds = self.create_creds(user['UserName'])
+
+        # Setup the config file with the profiles we'll be using. For
+        # convenience static credentials are placed here instead of putting
+        # them in the credentials file.
+        config = (
+            '[default]\n'
+            'aws_access_key_id = %s\n'
+            'aws_secret_access_key = %s\n'
+            '[profile middle]\n'
+            'source_profile = default\n'
+            'role_arn = %s\n'
+            '[profile final]\n'
+            'source_profile = middle\n'
+            'role_arn = %s\n'
+        )
+        config = config % (
+            user_creds['AccessKeyId'], user_creds['SecretAccessKey'],
+            middle_role['Arn'], final_role['Arn']
+        )
+        with open(self.config_file, 'w') as f:
+            f.write(config)
+
+        # Wait for IAM permissions to propagate
+        middle_creds = self.wait_for_assume_role(
+            role_arn=middle_role['Arn'],
+            access_key=user_creds['AccessKeyId'],
+            secret_key=user_creds['SecretAccessKey'],
+        )
+        self.wait_for_assume_role(
+            role_arn=final_role['Arn'],
+            access_key=middle_creds['AccessKeyId'],
+            secret_key=middle_creds['SecretAccessKey'],
+            token=middle_creds['SessionToken'],
+        )
+
+        # Configure our credentials file to be THE credentials file
+        self.environ['AWS_CONFIG_FILE'] = self.config_file
+
+        self.assert_s3_read_only_session(Session(profile='final'))
+
+    def test_assume_role_with_credential_source(self):
+        # Create a role with read access to S3
+        role = self.create_role(self.role_policy, S3_READ_POLICY_ARN)
+
+        # Create a user that can assume the role and get static credentials
+        # for it.
+        user_policy_arn = self.create_assume_policy(role['Arn'])
+        user = self.create_user([user_policy_arn])
+        user_creds = self.create_creds(user['UserName'])
+
+        # Setup the config file with the profile we'll be using.
+        config = (
+            '[profile assume]\n'
+            'role_arn = %s\n'
+            'credential_source = Environment\n'
+        )
+        config = config % role['Arn']
+        with open(self.config_file, 'w') as f:
+            f.write(config)
+
+        # Wait for IAM permissions to propagate
+        self.wait_for_assume_role(
+            role_arn=role['Arn'],
+            access_key=user_creds['AccessKeyId'],
+            secret_key=user_creds['SecretAccessKey'],
+        )
+
+        # Setup the environment so that our new config file is THE config
+        # file and add the expected credentials since we're using the
+        # environment as our credential source.
+        self.environ['AWS_CONFIG_FILE'] = self.config_file
+        self.environ['AWS_SECRET_ACCESS_KEY'] = user_creds['SecretAccessKey']
+        self.environ['AWS_ACCESS_KEY_ID'] = user_creds['AccessKeyId']
+
+        self.assert_s3_read_only_session(Session(profile='assume'))
diff --git a/tests/unit/test_credentials.py b/tests/unit/test_credentials.py
index 86d391654..c8951d6f3 100644
--- a/tests/unit/test_credentials.py
+++ b/tests/unit/test_credentials.py
@@ -14,12 +14,17 @@
 from datetime import datetime, timedelta
 import mock
 import os
+import json
+import copy
 
 from dateutil.tz import tzlocal, tzutc
 
 from botocore import credentials
 from botocore.utils import ContainerMetadataFetcher
 from botocore.credentials import EnvProvider, create_assume_role_refresher
+from botocore.credentials import CredentialProvider, AssumeRoleProvider
+from botocore.credentials import ConfigProvider, SharedCredentialProvider
+from botocore.credentials import Credentials
 import botocore.exceptions
 import botocore.session
 from tests import unittest, BaseEnvVar, IntegerRefresher
@@ -102,6 +107,13 @@ class TestRefreshableCredentials(TestCredentials):
         self.assertEqual(self.creds.secret_key, 'NEW-SECRET')
         self.assertEqual(self.creds.token, 'NEW-TOKEN')
 
+    def test_no_expiration(self):
+        creds = credentials.RefreshableCredentials(
+            'ORIGINAL-ACCESS', 'ORIGINAL-SECRET', 'ORIGINAL-TOKEN',
+            None, self.refresher, 'iam-role', time_fetcher=self.mock_time
+        )
+        self.assertFalse(creds.refresh_needed())
+
     def test_no_refresh_needed(self):
         # The expiry time was 30 minutes ago, let's say it's an hour
         # ago currently.  That would mean we don't need a refresh.
@@ -109,20 +121,500 @@ class TestRefreshableCredentials(TestCredentials):
             datetime.now(tzlocal()) - timedelta(minutes=60))
         self.assertTrue(not self.creds.refresh_needed())
 
-        self.assertEqual(self.creds.access_key, 'ORIGINAL-ACCESS')
-        self.assertEqual(self.creds.secret_key, 'ORIGINAL-SECRET')
-        self.assertEqual(self.creds.token, 'ORIGINAL-TOKEN')
+        self.assertEqual(self.creds.access_key, 'ORIGINAL-ACCESS')
+        self.assertEqual(self.creds.secret_key, 'ORIGINAL-SECRET')
+        self.assertEqual(self.creds.token, 'ORIGINAL-TOKEN')
+
+    def test_get_credentials_set(self):
+        # We need to return a consistent set of credentials to use during the
+        # signing process.
+        self.mock_time.return_value = (
+            datetime.now(tzlocal()) - timedelta(minutes=60))
+        self.assertTrue(not self.creds.refresh_needed())
+        credential_set = self.creds.get_frozen_credentials()
+        self.assertEqual(credential_set.access_key, 'ORIGINAL-ACCESS')
+        self.assertEqual(credential_set.secret_key, 'ORIGINAL-SECRET')
+        self.assertEqual(credential_set.token, 'ORIGINAL-TOKEN')
+
+
+class TestDeferredRefreshableCredentials(unittest.TestCase):
+    def setUp(self):
+        self.refresher = mock.Mock()
+        self.future_time = datetime.now(tzlocal()) + timedelta(hours=24)
+        self.metadata = {
+            'access_key': 'NEW-ACCESS',
+            'secret_key': 'NEW-SECRET',
+            'token': 'NEW-TOKEN',
+            'expiry_time': self.future_time.isoformat(),
+            'role_name': 'rolename',
+        }
+        self.refresher.return_value = self.metadata
+        self.mock_time = mock.Mock()
+        self.mock_time.return_value = datetime.now(tzlocal())
+
+    def test_refresh_using_called_on_first_access(self):
+        creds = credentials.DeferredRefreshableCredentials(
+            self.refresher, 'iam-role', self.mock_time
+        )
+
+        # The credentials haven't been accessed, so there should be no calls.
+        self.refresher.assert_not_called()
+
+        # Now that the object has been accessed, it should have called the
+        # refresher
+        creds.get_frozen_credentials()
+        self.assertEqual(self.refresher.call_count, 1)
+
+    def test_refresh_only_called_once(self):
+        creds = credentials.DeferredRefreshableCredentials(
+            self.refresher, 'iam-role', self.mock_time
+        )
+
+        for _ in range(5):
+            creds.get_frozen_credentials()
+
+        # The credentials were accessed several times in a row, but only
+        # should call refresh once.
+        self.assertEqual(self.refresher.call_count, 1)
+
+
+class TestAssumeRoleCredentialRefresher(BaseEnvVar):
+    def setUp(self):
+        super(TestAssumeRoleCredentialRefresher, self).setUp()
+        self.source_creds = credentials.Credentials('a', 'b', 'c')
+        self.role_arn = 'myrole'
+
+    def create_client_creator(self, with_response):
+        # Create a mock sts client that returns a specific response
+        # for assume_role.
+        client = mock.Mock()
+        if isinstance(with_response, list):
+            client.assume_role.side_effect = with_response
+        else:
+            client.assume_role.return_value = with_response
+        return mock.Mock(return_value=client)
+
+    def get_expected_creds_from_response(self, response):
+        expiration = response['Credentials']['Expiration']
+        if isinstance(expiration, datetime):
+            expiration = expiration.strftime('%Y-%m-%dT%H:%M:%S%Z')
+        return {
+            'access_key': response['Credentials']['AccessKeyId'],
+            'secret_key': response['Credentials']['SecretAccessKey'],
+            'token': response['Credentials']['SessionToken'],
+            'expiry_time': expiration
+        }
+
+    def some_future_time(self):
+        timeobj = datetime.now(tzlocal())
+        return timeobj + timedelta(hours=24)
+
+    def test_no_cache(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat()
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn
+        )
+
+        expected_response = self.get_expected_creds_from_response(response)
+        response = refresher.fetch_credentials()
+
+        self.assertEqual(response, expected_response)
+
+    def test_expiration_in_datetime_format(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                # Note the lack of isoformat(), we're using
+                # a datetime.datetime type.  This will ensure
+                # we test both parsing as well as serializing
+                # from a given datetime because the credentials
+                # are immediately expired.
+                'Expiration': self.some_future_time()
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn
+        )
+
+        expected_response = self.get_expected_creds_from_response(response)
+        response = refresher.fetch_credentials()
+
+        self.assertEqual(response, expected_response)
+
+    def test_retrieves_from_cache(self):
+        date_in_future = datetime.utcnow() + timedelta(seconds=1000)
+        utc_timestamp = date_in_future.isoformat() + 'Z'
+        cache_key = (
+            'myrole--'
+            'c317144bac72cf0af5958944e842acc49158de1532778e43569566b76c826f1f'
+        )
+        cache = {
+            cache_key: {
+                'Credentials': {
+                    'AccessKeyId': 'foo-cached',
+                    'SecretAccessKey': 'bar-cached',
+                    'SessionToken': 'baz-cached',
+                    'Expiration': utc_timestamp,
+                }
+            }
+        }
+        client_creator = mock.Mock()
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn, cache=cache
+        )
+
+        expected_response = self.get_expected_creds_from_response(
+            cache[cache_key]
+        )
+        response = refresher.fetch_credentials()
+
+        self.assertEqual(response, expected_response)
+        client_creator.assert_not_called()
+
+    def test_cache_key_is_windows_safe(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat()
+            },
+        }
+        cache = {}
+        client_creator = self.create_client_creator(with_response=response)
+
+        role_arn = 'arn:aws:iam::foo-role'
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, role_arn, cache=cache
+        )
+
+        refresher.fetch_credentials()
+
+        # On windows, you cannot use a a ':' in the filename, so
+        # we need to make sure that it doesn't make it into the cache key.
+        cache_key = (
+            'arn_aws_iam__foo-role--'
+            '2658f5f4e4ba3dce28e1c7eaf23dc91ccf512a3cface8cf9b48fed7dfc81a2db'
+        )
+        self.assertIn(cache_key, cache)
+        self.assertEqual(cache[cache_key], response)
+
+    def test_cache_key_with_role_session_name(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat()
+            },
+        }
+        cache = {}
+        client_creator = self.create_client_creator(with_response=response)
+        role_session_name = 'my_session_name'
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn, cache=cache,
+            extra_args={'RoleSessionName': role_session_name}
+        )
+        refresher.fetch_credentials()
+
+        # This is the sha256 hex digest of the expected assume role args.
+        cache_key = (
+            'myrole--'
+            'e1d8b3cff6b4198df3db001d5719fa40842e623bed271eb93afd3b46faaad847'
+        )
+        self.assertIn(cache_key, cache)
+        self.assertEqual(cache[cache_key], response)
+
+    def test_cache_key_with_policy(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat()
+            },
+        }
+        cache = {}
+        client_creator = self.create_client_creator(with_response=response)
+        policy = json.dumps({
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Action": "*",
+                    "Resource": "*"
+                }
+            ]
+        })
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn, cache=cache,
+            extra_args={'Policy': policy}
+        )
+        refresher.fetch_credentials()
+
+        # This is the sha256 hex digest of the expected assume role args.
+        cache_key = (
+            'myrole--'
+            'f56c4c34a846dcb116b7ab180767960a583b0697cd7d6f736696b5ec01291321'
+        )
+        self.assertIn(cache_key, cache)
+        self.assertEqual(cache[cache_key], response)
+
+    def test_assume_role_in_cache_but_expired(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        cache = {
+            'development--myrole': {
+                'Credentials': {
+                    'AccessKeyId': 'foo-cached',
+                    'SecretAccessKey': 'bar-cached',
+                    'SessionToken': 'baz-cached',
+                    'Expiration': datetime.now(tzlocal()),
+                }
+            }
+        }
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn, cache=cache
+        )
+        expected = self.get_expected_creds_from_response(response)
+        response = refresher.fetch_credentials()
+
+        self.assertEqual(response, expected)
+
+    def test_role_session_name_can_be_provided(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        role_session_name = 'myname'
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn,
+            extra_args={'RoleSessionName': role_session_name}
+        )
+        refresher.fetch_credentials()
+
+        client = client_creator.return_value
+        client.assume_role.assert_called_with(
+            RoleArn=self.role_arn, RoleSessionName=role_session_name)
+
+    def test_external_id_can_be_provided(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        external_id = 'my_external_id'
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn,
+            extra_args={'ExternalId': external_id}
+        )
+        refresher.fetch_credentials()
+
+        client = client_creator.return_value
+        client.assume_role.assert_called_with(
+            RoleArn=self.role_arn, ExternalId=external_id,
+            RoleSessionName=mock.ANY)
+
+    def test_policy_can_be_provided(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        policy = json.dumps({
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Action": "*",
+                    "Resource": "*"
+                }
+            ]
+        })
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn,
+            extra_args={'Policy': policy}
+        )
+        refresher.fetch_credentials()
+
+        client = client_creator.return_value
+        client.assume_role.assert_called_with(
+            RoleArn=self.role_arn, Policy=policy,
+            RoleSessionName=mock.ANY)
+
+    def test_duration_seconds_can_be_provided(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        duration = 1234
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn,
+            extra_args={'DurationSeconds': duration}
+        )
+        refresher.fetch_credentials()
+
+        client = client_creator.return_value
+        client.assume_role.assert_called_with(
+            RoleArn=self.role_arn, DurationSeconds=duration,
+            RoleSessionName=mock.ANY)
+
+    def test_mfa(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        prompter = mock.Mock(return_value='token-code')
+        mfa_serial = 'mfa'
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn,
+            extra_args={'SerialNumber': mfa_serial}, mfa_prompter=prompter
+        )
+        refresher.fetch_credentials()
+
+        client = client_creator.return_value
+        # In addition to the normal assume role args, we should also
+        # inject the serial number from the config as well as the
+        # token code that comes from prompting the user (the prompter
+        # object).
+        client.assume_role.assert_called_with(
+            RoleArn='myrole', RoleSessionName=mock.ANY, SerialNumber='mfa',
+            TokenCode='token-code')
+
+    def test_refreshes(self):
+        responses = [{
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                # We're creating an expiry time in the past so as
+                # soon as we try to access the credentials, the
+                # refresh behavior will be triggered.
+                'Expiration': (
+                    datetime.now(tzlocal()) -
+                    timedelta(seconds=100)).isoformat(),
+            },
+        }, {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat(),
+            }
+        }]
+        client_creator = self.create_client_creator(with_response=responses)
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn
+        )
+
+        # The first call will simply use whatever credentials it is given.
+        # The second will check the cache, and only make a call if the
+        # cached credentials are expired.
+        refresher.fetch_credentials()
+        refresher.fetch_credentials()
+
+        client = client_creator.return_value
+        assume_role_calls = client.assume_role.call_args_list
+        self.assertEqual(len(assume_role_calls), 2, assume_role_calls)
 
-    def test_get_credentials_set(self):
-        # We need to return a consistent set of credentials to use during the
-        # signing process.
-        self.mock_time.return_value = (
-            datetime.now(tzlocal()) - timedelta(minutes=60))
-        self.assertTrue(not self.creds.refresh_needed())
-        credential_set = self.creds.get_frozen_credentials()
-        self.assertEqual(credential_set.access_key, 'ORIGINAL-ACCESS')
-        self.assertEqual(credential_set.secret_key, 'ORIGINAL-SECRET')
-        self.assertEqual(credential_set.token, 'ORIGINAL-TOKEN')
+    def test_mfa_refresh_enabled(self):
+        responses = [{
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                # We're creating an expiry time in the past so as
+                # soon as we try to access the credentials, the
+                # refresh behavior will be triggered.
+                'Expiration': (
+                    datetime.now(tzlocal()) -
+                    timedelta(seconds=100)).isoformat(),
+            },
+        }, {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat(),
+            }
+        }]
+        client_creator = self.create_client_creator(with_response=responses)
+
+        token_code = 'token-code-1'
+        prompter = mock.Mock(side_effect=[token_code])
+        mfa_serial = 'mfa'
+
+        refresher = credentials.AssumeRoleCredentialFetcher(
+            client_creator, self.source_creds, self.role_arn,
+            extra_args={'SerialNumber': mfa_serial}, mfa_prompter=prompter
+        )
+
+        # This is will refresh credentials if they're expired. Because
+        # we set the expiry time to something in the past, this will
+        # trigger the refresh behavior.
+        refresher.fetch_credentials()
+
+        assume_role = client_creator.return_value.assume_role
+        calls = [c[1] for c in assume_role.call_args_list]
+        expected_calls = [
+            {
+                'RoleArn': self.role_arn,
+                'RoleSessionName': mock.ANY,
+                'SerialNumber': mfa_serial,
+                'TokenCode': token_code
+            }
+        ]
+        self.assertEqual(calls, expected_calls)
 
 
 class TestEnvVar(BaseEnvVar):
@@ -669,8 +1161,10 @@ class CredentialResolverTest(BaseEnvVar):
         super(CredentialResolverTest, self).setUp()
         self.provider1 = mock.Mock()
         self.provider1.METHOD = 'provider1'
+        self.provider1.CANONICAL_NAME = 'CustomProvider1'
         self.provider2 = mock.Mock()
         self.provider2.METHOD = 'provider2'
+        self.provider2.CANONICAL_NAME = 'CustomProvider2'
         self.fake_creds = credentials.Credentials('a', 'b', 'c')
 
     def test_load_credentials_single_provider(self):
@@ -826,6 +1320,140 @@ class TestCreateCredentialResolver(BaseEnvVar):
             any(isinstance(p, EnvProvider) for p in resolver.providers))
 
 
+class TestCanonicalNameSourceProvider(BaseEnvVar):
+    def setUp(self):
+        super(TestCanonicalNameSourceProvider, self).setUp()
+        self.custom_provider1 = mock.Mock(spec=CredentialProvider)
+        self.custom_provider1.METHOD = 'provider1'
+        self.custom_provider1.CANONICAL_NAME = 'CustomProvider1'
+        self.custom_provider2 = mock.Mock(spec=CredentialProvider)
+        self.custom_provider2.METHOD = 'provider2'
+        self.custom_provider2.CANONICAL_NAME = 'CustomProvider2'
+        self.fake_creds = credentials.Credentials('a', 'b', 'c')
+
+    def test_load_source_credentials(self):
+        provider = credentials.CanonicalNameCredentialSourcer(providers=[
+            self.custom_provider1, self.custom_provider2
+        ])
+        self.custom_provider1.load.return_value = self.fake_creds
+        result = provider.source_credentials('CustomProvider1')
+        self.assertIs(result, self.fake_creds)
+
+    def test_load_source_credentials_case_insensitive(self):
+        provider = credentials.CanonicalNameCredentialSourcer(providers=[
+            self.custom_provider1, self.custom_provider2
+        ])
+        self.custom_provider1.load.return_value = self.fake_creds
+        result = provider.source_credentials('cUsToMpRoViDeR1')
+        self.assertIs(result, self.fake_creds)
+
+    def test_load_unknown_canonical_name_raises_error(self):
+        provider = credentials.CanonicalNameCredentialSourcer(providers=[
+            self.custom_provider1])
+        with self.assertRaises(botocore.exceptions.UnknownCredentialError):
+            provider.source_credentials('CustomUnknown')
+
+    def _assert_assume_role_creds_returned_with_shared_file(self, provider):
+        assume_role_provider = mock.Mock(spec=AssumeRoleProvider)
+        assume_role_provider.METHOD = 'assume-role'
+        assume_role_provider.CANONICAL_NAME = None
+
+        source = credentials.CanonicalNameCredentialSourcer(providers=[
+            assume_role_provider, provider
+        ])
+
+        # If the assume role provider returns credentials, those should be
+        # what is returned.
+        assume_role_provider.load.return_value = self.fake_creds
+        provider.load.return_value = credentials.Credentials(
+            'd', 'e', 'f'
+        )
+
+        creds = source.source_credentials(provider.CANONICAL_NAME)
+        self.assertIsNotNone(creds)
+        self.assertEqual(creds.access_key, 'a')
+        self.assertEqual(creds.secret_key, 'b')
+        self.assertEqual(creds.token, 'c')
+        self.assertFalse(provider.load.called)
+
+    def _assert_returns_creds_if_assume_role_not_used(self, provider):
+        assume_role_provider = mock.Mock(spec=AssumeRoleProvider)
+        assume_role_provider.METHOD = 'assume-role'
+        assume_role_provider.CANONICAL_NAME = None
+
+        source = credentials.CanonicalNameCredentialSourcer(providers=[
+            assume_role_provider, provider
+        ])
+
+        # If the assume role provider returns nothing, then whatever is in
+        # the config provider should be returned.
+        assume_role_provider.load.return_value = None
+        provider.load.return_value = credentials.Credentials(
+            'd', 'e', 'f'
+        )
+
+        creds = source.source_credentials(provider.CANONICAL_NAME)
+        self.assertIsNotNone(creds)
+        self.assertEqual(creds.access_key, 'd')
+        self.assertEqual(creds.secret_key, 'e')
+        self.assertEqual(creds.token, 'f')
+        self.assertTrue(assume_role_provider.load.called)
+
+    def test_assume_role_creds_returned_with_config_file(self):
+        provider = mock.Mock(spec=ConfigProvider)
+        provider.METHOD = 'config-file'
+        provider.CANONICAL_NAME = 'SharedConfig'
+        self._assert_assume_role_creds_returned_with_shared_file(provider)
+
+    def test_config_file_returns_creds_if_assume_role_not_used(self):
+        provider = mock.Mock(spec=ConfigProvider)
+        provider.METHOD = 'config-file'
+        provider.CANONICAL_NAME = 'SharedConfig'
+        self._assert_returns_creds_if_assume_role_not_used(provider)
+
+    def test_assume_role_creds_returned_with_cred_file(self):
+        provider = mock.Mock(spec=SharedCredentialProvider)
+        provider.METHOD = 'credentials-file'
+        provider.CANONICAL_NAME = 'SharedCredentials'
+        self._assert_assume_role_creds_returned_with_shared_file(provider)
+
+    def test_creds_file_returns_creds_if_assume_role_not_used(self):
+        provider = mock.Mock(spec=SharedCredentialProvider)
+        provider.METHOD = 'credentials-file'
+        provider.CANONICAL_NAME = 'SharedCredentials'
+        self._assert_returns_creds_if_assume_role_not_used(provider)
+
+    def test_get_canonical_assume_role_without_shared_files(self):
+        assume_role_provider = mock.Mock(spec=AssumeRoleProvider)
+        assume_role_provider.METHOD = 'assume-role'
+        assume_role_provider.CANONICAL_NAME = None
+        assume_role_provider.load.return_value = self.fake_creds
+
+        provider = credentials.CanonicalNameCredentialSourcer(providers=[
+            assume_role_provider
+        ])
+
+        creds = provider.source_credentials('SharedConfig')
+        self.assertIsNotNone(creds)
+        self.assertEqual(creds.access_key, 'a')
+        self.assertEqual(creds.secret_key, 'b')
+        self.assertEqual(creds.token, 'c')
+
+        creds = provider.source_credentials('SharedCredentials')
+        self.assertIsNotNone(creds)
+        self.assertEqual(creds.access_key, 'a')
+        self.assertEqual(creds.secret_key, 'b')
+        self.assertEqual(creds.token, 'c')
+
+    def test_get_canonical_shared_files_without_assume_role(self):
+        provider = credentials.CanonicalNameCredentialSourcer(
+            providers=[self.custom_provider1])
+        with self.assertRaises(botocore.exceptions.UnknownCredentialError):
+            provider.source_credentials('SharedConfig')
+        with self.assertRaises(botocore.exceptions.UnknownCredentialError):
+            provider.source_credentials('SharedCredentials')
+
+
 class TestAssumeRoleCredentialProvider(unittest.TestCase):
 
     maxDiff = None
@@ -840,6 +1468,14 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
                 'longterm': {
                     'aws_access_key_id': 'akid',
                     'aws_secret_access_key': 'skid',
+                },
+                'non-static': {
+                    'role_arn': 'myrole',
+                    'credential_source': 'Environment'
+                },
+                'chained': {
+                    'role_arn': 'chained-role',
+                    'source_profile': 'development'
                 }
             }
         }
@@ -931,8 +1567,13 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
         date_in_future = datetime.utcnow() + timedelta(seconds=1000)
         utc_timestamp = date_in_future.isoformat() + 'Z'
         self.fake_config['profiles']['development']['role_arn'] = 'myrole'
+
+        cache_key = (
+            'myrole--'
+            'c317144bac72cf0af5958944e842acc49158de1532778e43569566b76c826f1f'
+        )
         cache = {
-            'development--myrole': {
+            cache_key: {
                 'Credentials': {
                     'AccessKeyId': 'foo-cached',
                     'SecretAccessKey': 'bar-cached',
@@ -951,13 +1592,49 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
         self.assertEqual(creds.secret_key, 'bar-cached')
         self.assertEqual(creds.token, 'baz-cached')
 
+    def test_chain_prefers_cache(self):
+        date_in_future = datetime.utcnow() + timedelta(seconds=1000)
+        utc_timestamp = date_in_future.isoformat() + 'Z'
+
+        # The profile we will be using has a cache entry, but the profile it
+        # is sourcing from does not. This should result in the cached
+        # credentials being used, and the source profile not being called.
+        cache_key = (
+            'chained-role--'
+            '3c84bd9fee47cf6be518796a56e77b9ed2a407ecbff0ebf322cd025cfff83abf'
+        )
+        cache = {
+            cache_key: {
+                'Credentials': {
+                    'AccessKeyId': 'foo-cached',
+                    'SecretAccessKey': 'bar-cached',
+                    'SessionToken': 'baz-cached',
+                    'Expiration': utc_timestamp,
+                }
+            }
+        }
+
+        client_creator = self.create_client_creator([
+            Exception("Attempted to call assume role when not needed.")
+        ])
+
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(), client_creator,
+            cache=cache, profile_name='chained')
+
+        creds = provider.load()
+
+        self.assertEqual(creds.access_key, 'foo-cached')
+        self.assertEqual(creds.secret_key, 'bar-cached')
+        self.assertEqual(creds.token, 'baz-cached')
+
     def test_cache_key_is_windows_safe(self):
         response = {
             'Credentials': {
                 'AccessKeyId': 'foo',
                 'SecretAccessKey': 'bar',
                 'SessionToken': 'baz',
-                'Expiration': datetime.now(tzlocal()).isoformat()
+                'Expiration': self.some_future_time().isoformat()
             },
         }
         cache = {}
@@ -969,12 +1646,15 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
             self.create_config_loader(),
             client_creator, cache=cache, profile_name='development')
 
-        provider.load()
+        provider.load().get_frozen_credentials()
         # On windows, you cannot use a a ':' in the filename, so
-        # we need to do some small transformations on the filename
-        # to replace any ':' that come up.
-        self.assertEqual(cache['development--arn_aws_iam__foo-role'],
-                         response)
+        # we need to make sure it doesn't come up in the cache key.
+        cache_key = (
+            'arn_aws_iam__foo-role--'
+            '2658f5f4e4ba3dce28e1c7eaf23dc91ccf512a3cface8cf9b48fed7dfc81a2db'
+        )
+        self.assertIn(cache_key, cache)
+        self.assertEqual(cache[cache_key], response)
 
     def test_cache_key_with_role_session_name(self):
         response = {
@@ -982,7 +1662,7 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
                 'AccessKeyId': 'foo',
                 'SecretAccessKey': 'bar',
                 'SessionToken': 'baz',
-                'Expiration': datetime.now(tzlocal()).isoformat()
+                'Expiration': self.some_future_time().isoformat()
             },
         }
         cache = {}
@@ -996,10 +1676,15 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
             self.create_config_loader(),
             client_creator, cache=cache, profile_name='development')
 
-        provider.load()
-        self.assertEqual(
-            cache['development--arn_aws_iam__foo-role--foo_role_session_name'],
-            response)
+        # The credentials won't actually be assumed until they're requested.
+        provider.load().get_frozen_credentials()
+
+        cache_key = (
+            'arn_aws_iam__foo-role--'
+            '3c29738019009f3a0bec23a6ade81f946d016400572f5c2b399084ed68432a60'
+        )
+        self.assertIn(cache_key, cache)
+        self.assertEqual(cache[cache_key], response)
 
     def test_assume_role_in_cache_but_expired(self):
         expired_creds = datetime.now(tzlocal())
@@ -1041,7 +1726,7 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
                 'AccessKeyId': 'foo',
                 'SecretAccessKey': 'bar',
                 'SessionToken': 'baz',
-                'Expiration': datetime.now(tzlocal()).isoformat(),
+                'Expiration': self.some_future_time().isoformat(),
             },
         }
         client_creator = self.create_client_creator(with_response=response)
@@ -1049,7 +1734,8 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
             self.create_config_loader(),
             client_creator, cache={}, profile_name='development')
 
-        provider.load()
+        # The credentials won't actually be assumed until they're requested.
+        provider.load().get_frozen_credentials()
 
         client = client_creator.return_value
         client.assume_role.assert_called_with(
@@ -1062,7 +1748,7 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
                 'AccessKeyId': 'foo',
                 'SecretAccessKey': 'bar',
                 'SessionToken': 'baz',
-                'Expiration': datetime.now(tzlocal()).isoformat(),
+                'Expiration': self.some_future_time().isoformat(),
             },
         }
         client_creator = self.create_client_creator(with_response=response)
@@ -1070,7 +1756,8 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
             self.create_config_loader(),
             client_creator, cache={}, profile_name='development')
 
-        provider.load()
+        # The credentials won't actually be assumed until they're requested.
+        provider.load().get_frozen_credentials()
 
         client = client_creator.return_value
         client.assume_role.assert_called_with(
@@ -1083,7 +1770,7 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
                 'AccessKeyId': 'foo',
                 'SecretAccessKey': 'bar',
                 'SessionToken': 'baz',
-                'Expiration': datetime.now(tzlocal()).isoformat(),
+                'Expiration': self.some_future_time().isoformat(),
             },
         }
         client_creator = self.create_client_creator(with_response=response)
@@ -1092,7 +1779,8 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
             self.create_config_loader(), client_creator,
             cache={}, profile_name='development', prompter=prompter)
 
-        provider.load()
+        # The credentials won't actually be assumed until they're requested.
+        provider.load().get_frozen_credentials()
 
         client = client_creator.return_value
         # In addition to the normal assume role args, we should also
@@ -1104,6 +1792,8 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
             TokenCode='token-code')
 
     def test_assume_role_populates_session_name_on_refresh(self):
+        expiration_time = self.some_future_time()
+        next_expiration_time = expiration_time + timedelta(hours=4)
         responses = [{
             'Credentials': {
                 'AccessKeyId': 'foo',
@@ -1112,18 +1802,14 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
                 # We're creating an expiry time in the past so as
                 # soon as we try to access the credentials, the
                 # refresh behavior will be triggered.
-                'Expiration': (
-                    datetime.now(tzlocal()) -
-                    timedelta(seconds=100)).isoformat(),
+                'Expiration': expiration_time.isoformat(),
             },
         }, {
             'Credentials': {
                 'AccessKeyId': 'foo',
                 'SecretAccessKey': 'bar',
                 'SessionToken': 'baz',
-                'Expiration': (
-                    datetime.now(tzlocal()) + timedelta(seconds=100)
-                ).isoformat(),
+                'Expiration': next_expiration_time.isoformat(),
             }
         }]
         client_creator = self.create_client_creator(with_response=responses)
@@ -1132,12 +1818,18 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
             cache={}, profile_name='development',
             prompter=mock.Mock(return_value='token-code'))
 
-        # This will trigger the first assume_role() call.  It returns
-        # credentials that are expired and will trigger a refresh.
-        creds = provider.load()
-        # This will trigger the second assume_role() call because
-        # a refresh is needed.
-        creds.get_frozen_credentials()
+        local_now = mock.Mock(return_value=datetime.now(tzlocal()))
+        with mock.patch('botocore.credentials._local_now', local_now):
+            # This will trigger the first assume_role() call.  It returns
+            # credentials that are expired and will trigger a refresh.
+            creds = provider.load()
+            creds.get_frozen_credentials()
+
+            # This will trigger the second assume_role() call because
+            # a refresh is needed.
+            local_now.return_value = expiration_time
+            creds.get_frozen_credentials()
+
         client = client_creator.return_value
         assume_role_calls = client.assume_role.call_args_list
         self.assertEqual(len(assume_role_calls), 2, assume_role_calls)
@@ -1152,6 +1844,7 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
         # But for now, if we get temp creds with MFA then when those
         # creds expire, we can't refresh the credentials.
         self.fake_config['profiles']['development']['mfa_serial'] = 'mfa'
+        expiration_time = self.some_future_time()
         response = {
             'Credentials': {
                 'AccessKeyId': 'foo',
@@ -1160,9 +1853,7 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
                 # We're creating an expiry time in the past so as
                 # soon as we try to access the credentials, the
                 # refresh behavior will be triggered.
-                'Expiration': (
-                    datetime.now(tzlocal()) -
-                    timedelta(seconds=100)).isoformat(),
+                'Expiration': expiration_time.isoformat(),
             },
         }
         client_creator = self.create_client_creator(with_response=response)
@@ -1171,13 +1862,19 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
             cache={}, profile_name='development',
             prompter=mock.Mock(return_value='token-code'))
 
-        creds = provider.load()
-        with self.assertRaises(credentials.RefreshWithMFAUnsupportedError):
-            # access_key is a property that will refresh credentials
-            # if they're expired.  Because we set the expiry time to
-            # something in the past, this will trigger the refresh
-            # behavior, with with MFA will currently raise an exception.
-            creds.access_key
+        local_now = mock.Mock(return_value=datetime.now(tzlocal()))
+        with mock.patch('botocore.credentials._local_now', local_now):
+            # Loads the credentials, resulting in the first assume role call.
+            creds = provider.load()
+            creds.get_frozen_credentials()
+
+            local_now.return_value = expiration_time
+            with self.assertRaises(credentials.RefreshWithMFAUnsupportedError):
+                # access_key is a property that will refresh credentials
+                # if they're expired.  Because we set the expiry time to
+                # something in the past, this will trigger the refresh
+                # behavior, with with MFA will currently raise an exception.
+                creds.access_key
 
     def test_no_config_is_noop(self):
         self.fake_config['profiles']['development'] = {
@@ -1215,6 +1912,243 @@ class TestAssumeRoleCredentialProvider(unittest.TestCase):
         with self.assertRaises(botocore.exceptions.InvalidConfigError):
             provider.load()
 
+    def test_incomplete_source_credentials_raises_error(self):
+        del self.fake_config['profiles']['longterm']['aws_access_key_id']
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            mock.Mock(), cache={}, profile_name='development')
+
+        with self.assertRaises(botocore.exceptions.PartialCredentialsError):
+            provider.load()
+
+    def test_source_profile_and_credential_source_provided(self):
+        profile = self.fake_config['profiles']['development']
+        profile['credential_source'] = 'SomeCredentialProvider'
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            mock.Mock(), cache={}, profile_name='development')
+
+        with self.assertRaises(botocore.exceptions.InvalidConfigError):
+            provider.load()
+
+    def test_credential_source_with_no_resolver_configured(self):
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            mock.Mock(), cache={}, profile_name='non-static')
+
+        with self.assertRaises(botocore.exceptions.InvalidConfigError):
+            provider.load()
+
+    def test_credential_source_with_no_providers_configured(self):
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            mock.Mock(), cache={}, profile_name='non-static',
+            credential_sourcer=credentials.CanonicalNameCredentialSourcer([])
+        )
+
+        with self.assertRaises(botocore.exceptions.InvalidConfigError):
+            provider.load()
+
+    def test_credential_source_not_among_providers(self):
+        fake_provider = mock.Mock()
+        fake_provider.CANONICAL_NAME = 'CustomFakeProvider'
+
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            mock.Mock(), cache={}, profile_name='non-static',
+            credential_sourcer=credentials.CanonicalNameCredentialSourcer(
+                [fake_provider])
+        )
+
+        # We configured the assume role provider with a single fake source
+        # provider, CustomFakeProvider. The profile we are attempting to use
+        # calls for the Environment credential provider as the credentials
+        # source. Since that isn't one of the configured source providers,
+        # an error is thrown.
+        with self.assertRaises(botocore.exceptions.InvalidConfigError):
+            provider.load()
+
+    def test_assume_role_with_credential_source(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat()
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+
+        config = {
+            'profiles': {
+                'sourced': {
+                    'role_arn': 'myrole',
+                    'credential_source': 'CustomMockProvider'
+                }
+            }
+        }
+        config_loader = self.create_config_loader(with_config=config)
+
+        fake_provider = mock.Mock()
+        fake_provider.CANONICAL_NAME = 'CustomMockProvider'
+        fake_creds = credentials.Credentials(
+            'akid', 'skid', 'token'
+        )
+        fake_provider.load.return_value = fake_creds
+
+        provider = credentials.AssumeRoleProvider(
+            config_loader, client_creator, cache={}, profile_name='sourced',
+            credential_sourcer=credentials.CanonicalNameCredentialSourcer(
+                [fake_provider])
+        )
+
+        creds = provider.load()
+        self.assertEqual(creds.access_key, 'foo')
+        self.assertEqual(creds.secret_key, 'bar')
+        self.assertEqual(creds.token, 'baz')
+        client_creator.assert_called_with(
+            'sts', aws_access_key_id=fake_creds.access_key,
+            aws_secret_access_key=fake_creds.secret_key,
+            aws_session_token=fake_creds.token
+        )
+
+    def test_credential_source_returns_none(self):
+        config = {
+            'profiles': {
+                'sourced': {
+                    'role_arn': 'myrole',
+                    'credential_source': 'CustomMockProvider'
+                }
+            }
+        }
+        config_loader = self.create_config_loader(with_config=config)
+
+        fake_provider = mock.Mock()
+        fake_provider.CANONICAL_NAME = 'CustomMockProvider'
+        fake_provider.load.return_value = None
+
+        provider = credentials.AssumeRoleProvider(
+            config_loader, mock.Mock(), cache={}, profile_name='sourced',
+            credential_sourcer=credentials.CanonicalNameCredentialSourcer(
+                [fake_provider])
+        )
+
+        with self.assertRaises(botocore.exceptions.CredentialRetrievalError):
+            provider.load()
+
+    def test_source_profile_can_reference_self(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': self.some_future_time().isoformat()
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+
+        config = {
+            'profiles': {
+                'self-referencial': {
+                    'aws_access_key_id': 'akid',
+                    'aws_secret_access_key': 'skid',
+                    'role_arn': 'myrole',
+                    'source_profile': 'self-referencial'
+                }
+            }
+        }
+
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(config),
+            client_creator, cache={}, profile_name='self-referencial'
+        )
+
+        creds = provider.load()
+        self.assertEqual(creds.access_key, 'foo')
+        self.assertEqual(creds.secret_key, 'bar')
+        self.assertEqual(creds.token, 'baz')
+
+    def test_infinite_looping_profiles_raises_error(self):
+        config = {
+            'profiles': {
+                'first': {
+                    'role_arn': 'first',
+                    'source_profile': 'second'
+                },
+                'second': {
+                    'role_arn': 'second',
+                    'source_profile': 'first'
+                }
+            }
+        }
+
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(config),
+            mock.Mock(), cache={}, profile_name='first'
+        )
+
+        with self.assertRaises(botocore.credentials.InfiniteLoopConfigError):
+            provider.load()
+
+    def test_recursive_assume_role(self):
+        assume_responses = [
+            Credentials('foo', 'bar', 'baz'),
+            Credentials('spam', 'eggs', 'spamandegss'),
+        ]
+        responses = []
+        for credential_set in assume_responses:
+            responses.append({
+                'Credentials': {
+                    'AccessKeyId': credential_set.access_key,
+                    'SecretAccessKey': credential_set.secret_key,
+                    'SessionToken': credential_set.token,
+                    'Expiration': self.some_future_time().isoformat()
+                }
+            })
+        client_creator = self.create_client_creator(with_response=responses)
+
+        static_credentials = Credentials('akid', 'skid')
+        config = {
+            'profiles': {
+                'first': {
+                    'role_arn': 'first',
+                    'source_profile': 'second'
+                },
+                'second': {
+                    'role_arn': 'second',
+                    'source_profile': 'third'
+                },
+                'third': {
+                    'aws_access_key_id': static_credentials.access_key,
+                    'aws_secret_access_key': static_credentials.secret_key,
+                }
+            }
+        }
+
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(config),
+            client_creator, cache={}, profile_name='first'
+        )
+
+        creds = provider.load()
+        expected_creds = assume_responses[-1]
+        self.assertEqual(creds.access_key, expected_creds.access_key)
+        self.assertEqual(creds.secret_key, expected_creds.secret_key)
+        self.assertEqual(creds.token, expected_creds.token)
+
+        client_creator.assert_has_calls([
+            mock.call(
+                'sts', aws_access_key_id=static_credentials.access_key,
+                aws_secret_access_key=static_credentials.secret_key,
+                aws_session_token=static_credentials.token
+            ),
+            mock.call(
+                'sts', aws_access_key_id=assume_responses[0].access_key,
+                aws_secret_access_key=assume_responses[0].secret_key,
+                aws_session_token=assume_responses[0].token
+            ),
+        ])
+
 
 class TestRefreshLogic(unittest.TestCase):
     def test_mandatory_refresh_needed(self):