Repository: boto/botocore
Pull request #744 (2015-12-17 23:30:05)

Move assume role provider from CLI to botocore
This is a direct port with these exceptions:

* The JSONFileCache was kept in the CLI.  You can still
  instantiate or attach a cache to the assume role provider
  if you're writing short lived scripts and want to avoid
  making unnecessary AssumeRole API calls, but you'll need
  to provide your own cache implementation.
* A get_provider() convenience method was added to the
  credential resolver.

This makes it straightforward to still use the JSONFileCache
in the CLI via:

```
c = session.get_component('credential_provider')
provider = c.get_provider('assume-role')
provide.cache = JSONFileCache()
```

Note: there will be a corresponding PR to CLI we'll need to sync up.
Supersedes and closes #721 

cc @kyleknap @mtdowling @rayluo @JordonPhillips 


Patch:
diff --git a/botocore/credentials.py b/botocore/credentials.py
index f4195d217..ab8e02323 100644
--- a/botocore/credentials.py
+++ b/botocore/credentials.py
@@ -11,14 +11,12 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
-
+import time
 import datetime
-import functools
 import logging
 import os
+import getpass
 
-from botocore.compat import six
-from six.moves import configparser
 from dateutil.parser import parse
 from dateutil.tz import tzlocal
 
@@ -28,6 +26,8 @@ from botocore.compat import total_seconds
 from botocore.exceptions import UnknownCredentialError
 from botocore.exceptions import PartialCredentialsError
 from botocore.exceptions import ConfigNotFound
+from botocore.exceptions import InvalidConfigError
+from botocore.exceptions import RefreshWithMFAUnsupportedError
 from botocore.utils import InstanceMetadataFetcher, parse_key_val_file
 
 
@@ -51,6 +51,12 @@ def create_credential_resolver(session):
     env_provider = EnvProvider()
     providers = [
         env_provider,
+        AssumeRoleProvider(
+            load_config=lambda: session.full_config,
+            client_creator=session.create_client,
+            cache={},
+            profile_name=profile_name,
+        ),
         SharedCredentialProvider(
             creds_filename=credential_file,
             profile_name=profile_name
@@ -103,6 +109,30 @@ def _local_now():
     return datetime.datetime.now(tzlocal())
 
 
+def create_assume_role_refresher(client, params):
+    def refresh():
+        response = client.assume_role(**params)
+        credentials = response['Credentials']
+        # We need to normalize the credential names to
+        # the values expected by the refresh creds.
+        return {
+            'access_key': credentials['AccessKeyId'],
+            'secret_key': credentials['SecretAccessKey'],
+            'token': credentials['SessionToken'],
+            'expiry_time': credentials['Expiration'],
+        }
+    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
+
+
 class Credentials(object):
     """
     Holds the credentials needed to authenticate requests.
@@ -544,6 +574,210 @@ class BotoProvider(CredentialProvider):
                                        method=self.METHOD)
 
 
+class AssumeRoleProvider(CredentialProvider):
+
+    METHOD = 'assume-role'
+    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
+    # EXPIRY_WINDOW.
+    EXPIRY_WINDOW_SECONDS = 60 * 15
+
+    def __init__(self, load_config, client_creator, cache, profile_name,
+                 prompter=getpass.getpass):
+        """
+
+        :type load_config: callable
+        :param load_config: A function that accepts no arguments, and
+            when called, will return the full configuration dictionary
+            for the session (``session.full_config``).
+
+        :type client_creator: callable
+        :param client_creator: A factory function that will create
+            a client when called.  Has the same interface as
+            ``botocore.session.Session.create_client``.
+
+        :type cache: JSONFileCache
+        :param cache: An object that supports ``__getitem__``,
+            ``__setitem__``, and ``__contains__``.  An example
+            of this is the ``JSONFileCache`` class.
+
+        :type profile_name: str
+        :param profile_name: The name of the profile.
+
+        :type prompter: callable
+        :param prompter: A callable that returns input provided
+            by the user (i.e raw_input, getpass.getpass, etc.).
+
+        """
+        #: The cache used to first check for assumed credentials.
+        #: This is checked before making the AssumeRole API
+        #: calls and can be useful if you have short lived
+        #: scripts and you'd like to avoid calling AssumeRole
+        #: until the credentials are expired.
+        self.cache = cache
+        self._load_config = load_config
+        # client_creator is a callable that creates function.
+        # It's basically session.create_client
+        self._client_creator = client_creator
+        self._profile_name = profile_name
+        self._prompter = prompter
+        # The _loaded_config attribute will be populated from the
+        # load_config() function once the configuration is actually
+        # loaded.  The reason we go through all this instead of just
+        # requiring that the loaded_config be passed to us is to that
+        # we can defer configuration loaded until we actually try
+        # to load credentials (as opposed to when the object is
+        # instantiated).
+        self._loaded_config = {}
+
+    def load(self):
+        self._loaded_config = self._load_config()
+        if self._has_assume_role_config_vars():
+            return self._load_creds_via_assume_role()
+
+    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(credentials['Credentials']['Expiration'])
+        now = datetime.datetime.now(tzlocal())
+        seconds = total_seconds(end_time - now)
+        return seconds < self.EXPIRY_WINDOW_SECONDS
+
+    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(':', '_')
+        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)
+
+        return cache_key.replace('/', '-')
+
+    def _write_cached_credentials(self, creds, cache_key):
+        self.cache[cache_key] = creds
+
+    def _get_role_config_values(self):
+        # This returns the role related configuration.
+        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
+        }
+
+    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(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)
+        if assume_role_kwargs.get('RoleSessionName') is None:
+            role_session_name = 'AWS-CLI-session-%s' % (int(time.time()))
+            assume_role_kwargs['RoleSessionName'] = role_session_name
+
+        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']
+        return assume_role_kwargs
+
+
 class CredentialResolver(object):
 
     def __init__(self, providers):
@@ -588,11 +822,7 @@ class CredentialResolver(object):
             you'd like to add to the chain.
         :type cred_instance: A subclass of ``Credentials``
         """
-        try:
-            offset = [p.METHOD for p in self.providers].index(name)
-        except ValueError:
-            raise UnknownCredentialError(name=name)
-
+        offset = self._get_provider_offset(name)
         self.providers.insert(offset + 1, credential_provider)
 
     def remove(self, name):
@@ -610,6 +840,24 @@ class CredentialResolver(object):
         offset = available_methods.index(name)
         self.providers.pop(offset)
 
+    def get_provider(self, name):
+        """Return a credential provider by name.
+
+        :type name: str
+        :param name: The name of the provider.
+
+        :raises UnknownCredentialError: Raised if no
+            credential provider by the provided name
+            is found.
+        """
+        return self.providers[self._get_provider_offset(name)]
+
+    def _get_provider_offset(self, name):
+        try:
+            return [p.METHOD for p in self.providers].index(name)
+        except ValueError:
+            raise UnknownCredentialError(name=name)
+
     def load_credentials(self):
         """
         Goes through the credentials chain, returning the first ``Credentials``
diff --git a/botocore/exceptions.py b/botocore/exceptions.py
index 3e9358235..fb3cddf17 100644
--- a/botocore/exceptions.py
+++ b/botocore/exceptions.py
@@ -340,3 +340,11 @@ class InvalidS3AddressingStyleError(BotoCoreError):
         'S3 addressing style {s3_addressing_style} is invaild. Valid options '
         'are: \'auto\', \'virtual\', and \'path\''
     )
+
+
+class InvalidConfigError(BotoCoreError):
+    fmt = '{error_msg}'
+
+
+class RefreshWithMFAUnsupportedError(BotoCoreError):
+    fmt = 'Cannot refresh credentials: MFA token required.'


Test patch:
diff --git a/tests/unit/test_credentials.py b/tests/unit/test_credentials.py
index 019c565f9..36f7894be 100644
--- a/tests/unit/test_credentials.py
+++ b/tests/unit/test_credentials.py
@@ -12,7 +12,7 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
-import datetime
+from datetime import datetime, timedelta
 import mock
 import os
 
@@ -81,7 +81,7 @@ class TestRefreshableCredentials(TestCredentials):
         }
         self.refresher.return_value = self.metadata
         self.expiry_time = \
-            datetime.datetime.now(tzlocal()) - datetime.timedelta(minutes=30)
+            datetime.now(tzlocal()) - timedelta(minutes=30)
         self.mock_time = mock.Mock()
         self.creds = credentials.RefreshableCredentials(
             'ORIGINAL-ACCESS', 'ORIGINAL-SECRET', 'ORIGINAL-TOKEN',
@@ -93,7 +93,7 @@ class TestRefreshableCredentials(TestCredentials):
         # The expiry time was set for 30 minutes ago, so if we
         # say the current time is utcnow(), then we should need
         # a refresh.
-        self.mock_time.return_value = datetime.datetime.now(tzlocal())
+        self.mock_time.return_value = datetime.now(tzlocal())
         self.assertTrue(self.creds.refresh_needed())
         # We should refresh creds, if we try to access "access_key"
         # or any of the cred vars.
@@ -105,7 +105,7 @@ class TestRefreshableCredentials(TestCredentials):
         # 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.
         self.mock_time.return_value = (
-            datetime.datetime.now(tzlocal()) - datetime.timedelta(minutes=60))
+            datetime.now(tzlocal()) - timedelta(minutes=60))
         self.assertTrue(not self.creds.refresh_needed())
 
         self.assertEqual(self.creds.access_key, 'ORIGINAL-ACCESS')
@@ -211,6 +211,7 @@ class TestEnvVar(BaseEnvVar):
         with self.assertRaises(botocore.exceptions.PartialCredentialsError):
             provider.load()
 
+
 class TestSharedCredentialsProvider(BaseEnvVar):
     def setUp(self):
         super(TestSharedCredentialsProvider, self).setUp()
@@ -503,6 +504,16 @@ class CredentialResolverTest(BaseEnvVar):
         self.assertEqual(creds.secret_key, 'b')
         self.assertEqual(creds.token, 'c')
 
+    def test_get_provider_by_name(self):
+        resolver = credentials.CredentialResolver(providers=[self.provider1])
+        result = resolver.get_provider('provider1')
+        self.assertIs(result, self.provider1)
+
+    def test_get_unknown_provider_raises_error(self):
+        resolver = credentials.CredentialResolver(providers=[self.provider1])
+        with self.assertRaises(botocore.exceptions.UnknownCredentialError):
+            resolver.get_provider('unknown-foo')
+
     def test_first_credential_non_none_wins(self):
         self.provider1.load.return_value = None
         self.provider2.load.return_value = self.fake_creds
@@ -638,5 +649,303 @@ class TestCreateCredentialResolver(BaseEnvVar):
             any(isinstance(p, EnvProvider) for p in resolver.providers))
 
 
+class TestAssumeRoleCredentialProvider(unittest.TestCase):
+
+    maxDiff = None
+
+    def setUp(self):
+        self.fake_config = {
+            'profiles': {
+                'development': {
+                    'role_arn': 'myrole',
+                    'source_profile': 'longterm',
+                },
+                'longterm': {
+                    'aws_access_key_id': 'akid',
+                    'aws_secret_access_key': 'skid',
+                }
+            }
+        }
+
+    def create_config_loader(self, with_config=None):
+        if with_config is None:
+            with_config = self.fake_config
+        load_config = mock.Mock()
+        load_config.return_value = with_config
+        return load_config
+
+    def create_client_creator(self, with_response):
+        # Create a mock sts client that returns a specific response
+        # for assume_role.
+        client = mock.Mock()
+        client.assume_role.return_value = with_response
+        return mock.Mock(return_value=client)
+
+    def test_assume_role_with_no_cache(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': datetime.now(tzlocal()).isoformat()
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            client_creator, cache={}, profile_name='development')
+
+        creds = provider.load()
+
+        self.assertEqual(creds.access_key, 'foo')
+        self.assertEqual(creds.secret_key, 'bar')
+        self.assertEqual(creds.token, 'baz')
+
+    def test_assume_role_retrieves_from_cache(self):
+        date_in_future = datetime.utcnow() + timedelta(seconds=1000)
+        utc_timestamp = date_in_future.isoformat() + 'Z'
+        self.fake_config['profiles']['development']['role_arn'] = 'myrole'
+        cache = {
+            'development--myrole': {
+                'Credentials': {
+                    'AccessKeyId': 'foo-cached',
+                    'SecretAccessKey': 'bar-cached',
+                    'SessionToken': 'baz-cached',
+                    'Expiration': utc_timestamp,
+                }
+            }
+        }
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(), mock.Mock(),
+            cache=cache, profile_name='development')
+
+        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()
+            },
+        }
+        cache = {}
+        self.fake_config['profiles']['development']['role_arn'] = (
+            'arn:aws:iam::foo-role')
+
+        client_creator = self.create_client_creator(with_response=response)
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            client_creator, cache=cache, profile_name='development')
+
+        provider.load()
+        # 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)
+
+    def test_cache_key_with_role_session_name(self):
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': datetime.now(tzlocal()).isoformat()
+            },
+        }
+        cache = {}
+        self.fake_config['profiles']['development']['role_arn'] = (
+            'arn:aws:iam::foo-role')
+        self.fake_config['profiles']['development']['role_session_name'] = (
+            'foo_role_session_name')
+
+        client_creator = self.create_client_creator(with_response=response)
+        provider = credentials.AssumeRoleProvider(
+            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)
+
+    def test_assume_role_in_cache_but_expired(self):
+        expired_creds = datetime.utcnow()
+        utc_timestamp = expired_creds.isoformat() + 'Z'
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': utc_timestamp,
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        cache = {
+            'development--myrole': {
+                'Credentials': {
+                    'AccessKeyId': 'foo-cached',
+                    'SecretAccessKey': 'bar-cached',
+                    'SessionToken': 'baz-cached',
+                    'Expiration': utc_timestamp,
+                }
+            }
+        }
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(), client_creator,
+            cache=cache, profile_name='development')
+
+        creds = provider.load()
+
+        self.assertEqual(creds.access_key, 'foo')
+        self.assertEqual(creds.secret_key, 'bar')
+        self.assertEqual(creds.token, 'baz')
+
+    def test_role_session_name_provided(self):
+        dev_profile = self.fake_config['profiles']['development']
+        dev_profile['role_session_name'] = 'myname'
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': datetime.now(tzlocal()).isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            client_creator, cache={}, profile_name='development')
+
+        provider.load()
+
+        client = client_creator.return_value
+        client.assume_role.assert_called_with(
+            RoleArn='myrole', RoleSessionName='myname')
+
+    def test_external_id_provided(self):
+        self.fake_config['profiles']['development']['external_id'] = 'myid'
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': datetime.now(tzlocal()).isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            client_creator, cache={}, profile_name='development')
+
+        provider.load()
+
+        client = client_creator.return_value
+        client.assume_role.assert_called_with(
+            RoleArn='myrole', ExternalId='myid', RoleSessionName=mock.ANY)
+
+    def test_assume_role_with_mfa(self):
+        self.fake_config['profiles']['development']['mfa_serial'] = 'mfa'
+        response = {
+            'Credentials': {
+                'AccessKeyId': 'foo',
+                'SecretAccessKey': 'bar',
+                'SessionToken': 'baz',
+                'Expiration': datetime.now(tzlocal()).isoformat(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        prompter = mock.Mock(return_value='token-code')
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(), client_creator,
+            cache={}, profile_name='development', prompter=prompter)
+
+        provider.load()
+
+        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_assume_role_mfa_cannot_refresh_credentials(self):
+        # Note: we should look into supporting optional behavior
+        # in the future that allows for reprompting for credentials.
+        # 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'
+        response = {
+            '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(),
+            },
+        }
+        client_creator = self.create_client_creator(with_response=response)
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(), client_creator,
+            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
+
+    def test_no_config_is_noop(self):
+        self.fake_config['profiles']['development'] = {
+            'aws_access_key_id': 'foo',
+            'aws_secret_access_key': 'bar',
+        }
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            mock.Mock(), cache={}, profile_name='development')
+
+        # Because a role_arn was not specified, the AssumeRoleProvider
+        # is a noop and will not return credentials (which means we
+        # move on to the next provider).
+        creds = provider.load()
+        self.assertIsNone(creds)
+
+    def test_source_profile_not_provided(self):
+        del self.fake_config['profiles']['development']['source_profile']
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            mock.Mock(), cache={}, profile_name='development')
+
+        # source_profile is required, we shoudl get an error.
+        with self.assertRaises(botocore.exceptions.PartialCredentialsError):
+            provider.load()
+
+    def test_source_profile_does_not_exist(self):
+        dev_profile = self.fake_config['profiles']['development']
+        dev_profile['source_profile'] = 'does-not-exist'
+        provider = credentials.AssumeRoleProvider(
+            self.create_config_loader(),
+            mock.Mock(), cache={}, profile_name='development')
+
+        # source_profile is required, we shoudl get an error.
+        with self.assertRaises(botocore.exceptions.InvalidConfigError):
+            provider.load()
+
+
 if __name__ == "__main__":
     unittest.main()