Repository: boto/botocore
Pull request #2591 (2022-01-14 23:56:03)

Adding smart default support
Adding support for `defaults_mode` client configuration option. The `defaults_mode` will be used to determine how certain default configuration options are resolved in the SDK. Configurations values impacted include: `retry_mode`, `sts_regional_endpoints` , s3 `us_east_1_regional_endpoint` and `connect_timeout`.

codecov-commenter: # [Codecov](https://codecov.io/gh/boto/botocore/pull/2591?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) Report
> Merging [#2591](https://codecov.io/gh/boto/botocore/pull/2591?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) (18464cd) into [develop](https://codecov.io/gh/boto/botocore/commit/b0923b4d64c48cb600c61bf993e6e28643bf54a4?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) (b0923b4) will **decrease** coverage by `0.05%`.
> The diff coverage is `95.52%`.

[![Impacted file tree graph](https://codecov.io/gh/boto/botocore/pull/2591/graphs/tree.svg?width=650&height=150&src=pr&token=QOLXw6XSza&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto)](https://codecov.io/gh/boto/botocore/pull/2591?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto)

```diff
@@             Coverage Diff             @@
##           develop    #2591      +/-   ##
===========================================
- Coverage    95.41%   95.36%   -0.06%     
===========================================
  Files           59       59              
  Lines        11631    11826     +195     
===========================================
+ Hits         11098    11278     +180     
- Misses         533      548      +15     
```


| [Impacted Files](https://codecov.io/gh/boto/botocore/pull/2591?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) | Coverage Δ | |
|---|---|---|
| [botocore/utils.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvdXRpbHMucHk=) | `95.40% <89.74%> (-0.35%)` | :arrow_down: |
| [botocore/configprovider.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvY29uZmlncHJvdmlkZXIucHk=) | `94.40% <95.65%> (+0.93%)` | :arrow_up: |
| [botocore/args.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvYXJncy5weQ==) | `100.00% <100.00%> (ø)` | |
| [botocore/config.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvY29uZmlnLnB5) | `100.00% <100.00%> (ø)` | |
| [botocore/exceptions.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvZXhjZXB0aW9ucy5weQ==) | `99.53% <100.00%> (+<0.01%)` | :arrow_up: |
| [botocore/session.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvc2Vzc2lvbi5weQ==) | `96.88% <100.00%> (+0.25%)` | :arrow_up: |
| [botocore/compat.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvY29tcGF0LnB5) | `72.13% <0.00%> (-1.10%)` | :arrow_down: |
| [botocore/httpsession.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvaHR0cHNlc3Npb24ucHk=) | `90.21% <0.00%> (-0.43%)` | :arrow_down: |
| [botocore/hooks.py](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto#diff-Ym90b2NvcmUvaG9va3MucHk=) | `95.27% <0.00%> (-0.40%)` | :arrow_down: |
| ... and [4 more](https://codecov.io/gh/boto/botocore/pull/2591/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/boto/botocore/pull/2591?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/boto/botocore/pull/2591?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto). Last update [b0923b4...18464cd](https://codecov.io/gh/boto/botocore/pull/2591?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=boto).


Patch:
diff --git a/botocore/args.py b/botocore/args.py
index 2e197dcab..d3457ae32 100644
--- a/botocore/args.py
+++ b/botocore/args.py
@@ -177,6 +177,7 @@ class ClientArgsCreator(object):
                 inject_host_prefix=client_config.inject_host_prefix,
             )
         self._compute_retry_config(config_kwargs)
+        self._compute_connect_timeout(config_kwargs)
         s3_config = self.compute_s3_config(client_config)
 
         is_s3_service = service_name in ['s3', 's3-control']
@@ -380,6 +381,18 @@ class ClientArgsCreator(object):
             retry_mode = 'legacy'
         retries['mode'] = retry_mode
 
+    def _compute_connect_timeout(self, config_kwargs):
+        # Checking if connect_timeout is set on the client config.
+        # If it is not, we check the config_store in case a
+        # non legacy default mode has been configured.
+        connect_timeout = config_kwargs.get('connect_timeout')
+        if connect_timeout is not None:
+            return
+        connect_timeout = self._config_store.get_config_variable(
+            'connect_timeout')
+        if connect_timeout:
+            config_kwargs['connect_timeout'] = connect_timeout
+
     def _ensure_boolean(self, val):
         if isinstance(val, bool):
             return val
diff --git a/botocore/config.py b/botocore/config.py
index 80eca4c00..996464c58 100644
--- a/botocore/config.py
+++ b/botocore/config.py
@@ -200,14 +200,23 @@ class Config(object):
         ('endpoint_discovery_enabled', None),
         ('use_dualstack_endpoint', None),
         ('use_fips_endpoint', None),
+        ('defaults_mode', None)
     ])
 
+    NON_LEGACY_OPTION_DEFAULTS = {
+        'connect_timeout': None,
+    }
+
     def __init__(self, *args, **kwargs):
         self._user_provided_options = self._record_user_provided_options(
             args, kwargs)
 
         # Merge the user_provided options onto the default options
         config_vars = copy.copy(self.OPTION_DEFAULTS)
+        defaults_mode = self._user_provided_options.get(
+            'defaults_mode', 'legacy')
+        if defaults_mode != 'legacy':
+            config_vars.update(self.NON_LEGACY_OPTION_DEFAULTS)
         config_vars.update(self._user_provided_options)
 
         # Set the attributes based on the config_vars
diff --git a/botocore/configprovider.py b/botocore/configprovider.py
index 91819d303..6bd170243 100644
--- a/botocore/configprovider.py
+++ b/botocore/configprovider.py
@@ -15,6 +15,7 @@ is loaded.
 """
 import logging
 import os
+import copy
 
 from botocore import utils
 
@@ -109,6 +110,7 @@ BOTOCORE_DEFAUT_SESSION_VARIABLES = {
         None
     ),
     'retry_mode': ('retry_mode', 'AWS_RETRY_MODE', 'legacy', None),
+    'defaults_mode': ('defaults_mode', 'AWS_DEFAULTS_MODE', 'legacy', None),
     # We can't have a default here for v1 because we need to defer to
     # whatever the defaults are in _retry.json.
     'max_attempts': ('max_attempts', 'AWS_MAX_ATTEMPTS', None, int),
@@ -182,6 +184,43 @@ def _create_config_chain_mapping(chain_builder, config_variables):
     return mapping
 
 
+class DefaultConfigResolver:
+
+    def __init__(self, default_config_data):
+        self._base_default_config = default_config_data['base']
+        self._modes = default_config_data['modes']
+        self._resolved_default_configurations = {}
+
+    def _resolve_default_values_by_mode(self, mode):
+        default_config = self._base_default_config.copy()
+        modifications = self._modes.get(mode)
+
+        for config_var in modifications:
+            default_value = default_config[config_var]
+            modification_dict = modifications[config_var]
+            modification = list(modification_dict.keys())[0]
+            modification_value = modification_dict[modification]
+            if modification == 'multiply':
+                default_value *= modification_value
+            elif modification == 'add':
+                default_value += modification_value
+            elif modification == 'override':
+                default_value = modification_value
+            default_config[config_var] = default_value
+        return default_config
+
+    def get_default_modes(self):
+        default_modes = ['legacy', 'auto']
+        default_modes.extend(self._modes.keys())
+        return default_modes
+
+    def get_default_config_values(self, mode):
+        if mode not in self._resolved_default_configurations:
+            defaults = self._resolve_default_values_by_mode(mode)
+            self._resolved_default_configurations[mode] = defaults
+        return self._resolved_default_configurations[mode]
+
+
 class ConfigChainFactory(object):
     """Factory class to create our most common configuration chain case.
 
@@ -308,6 +347,11 @@ class ConfigValueStore(object):
             for logical_name, provider in mapping.items():
                 self.set_config_provider(logical_name, provider)
 
+    def __deepcopy__(self, memo):
+        return ConfigValueStore(
+            copy.deepcopy(self._mapping, memo)
+        )
+
     def get_config_variable(self, logical_name):
         """
         Retrieve the value associeated with the specified logical_name
@@ -329,6 +373,24 @@ class ConfigValueStore(object):
         provider = self._mapping[logical_name]
         return provider.provide()
 
+    def get_config_provider(self, logical_name):
+        """
+        Retrieve the provider associated with the specified logical_name.
+        If no provider is found None will be returned.
+
+        :type logical_name: str
+        :param logical_name: The logical name of the session variable
+            you want to retrieve.  This name will be mapped to the
+            appropriate environment variable name for this session as
+            well as the appropriate config file entry.
+
+        :returns: configuration provider or None if not defined.
+        """
+        if logical_name in self._overrides or logical_name not in self._mapping:
+            return None
+        provider = self._mapping[logical_name]
+        return provider
+
     def set_config_variable(self, logical_name, value):
         """Set a configuration variable to a specific value.
 
@@ -382,6 +444,78 @@ class ConfigValueStore(object):
         self._mapping[logical_name] = provider
 
 
+class SmartDefaultsConfigStoreFactory:
+
+    def __init__(self, default_config_resolver, imds_region_provider):
+        self._default_config_resolver = default_config_resolver
+        self._imds_region_provider = imds_region_provider
+        # Initializing _instance_metadata_region as None so we
+        # can fetch region in a lazy fashion only when needed.
+        self._instance_metadata_region = None
+
+    def merge_smart_defaults(self, config_store, mode, region_name):
+        if mode == 'auto':
+            mode = self.resolve_auto_mode(region_name)
+        default_configs = self._default_config_resolver.get_default_config_values(
+            mode)
+        for config_var in default_configs:
+            config_value = default_configs[config_var]
+            method = getattr(self, f'_set_{config_var}', None)
+            if method:
+                method(config_store, config_value)
+
+    def resolve_auto_mode(self, region_name):
+        current_region = None
+        if os.environ.get('AWS_EXECUTION_ENV'):
+            default_region = os.environ.get('AWS_DEFAULT_REGION')
+            current_region = os.environ.get('AWS_REGION', default_region)
+        if not current_region:
+            if self._instance_metadata_region:
+                current_region = self._instance_metadata_region
+            else:
+                try:
+                    current_region = \
+                        self._imds_region_provider.provide()
+                    self._instance_metadata_region = current_region
+                except Exception:
+                    pass
+
+        if current_region:
+            if region_name == current_region:
+                return 'in-region'
+            else:
+                return 'cross-region'
+        return 'standard'
+
+    def _update_provider(self, config_store, variable, value):
+        provider = config_store.get_config_provider(variable)
+        default_provider = ConstantProvider(value)
+        if isinstance(provider, ChainProvider):
+            provider.set_default_provider(default_provider)
+            return
+        elif isinstance(provider, BaseProvider):
+            default_provider = ChainProvider(providers=[provider, default_provider])
+        config_store.set_config_provider(variable, default_provider)
+
+    def _update_section_provider(self, config_store, section_name, variable,
+                                 value):
+        section_provider = config_store.get_config_provider(section_name)
+        section_provider.set_default_provider(variable, ConstantProvider(value))
+
+    def _set_retryMode(self, config_store, value):
+        self._update_provider(config_store, 'retry_mode', value)
+
+    def _set_stsRegionalEndpoints(self, config_store, value):
+        self._update_provider(config_store, 'sts_regional_endpoints', value)
+
+    def _set_s3UsEast1RegionalEndpoints(self, config_store, value):
+        self._update_section_provider(
+            config_store, 's3', 'us_east_1_regional_endpoint', value)
+
+    def _set_connectTimeoutInMillis(self, config_store, value):
+        self._update_provider(config_store, 'connect_timeout', value/1000)
+
+
 class BaseProvider(object):
     """Base class for configuration value providers.
 
@@ -416,6 +550,12 @@ class ChainProvider(BaseProvider):
         self._providers = providers
         self._conversion_func = conversion_func
 
+    def __deepcopy__(self, memo):
+        return ChainProvider(
+            copy.deepcopy(self._providers, memo),
+            self._conversion_func
+        )
+
     def provide(self):
         """Provide the value from the first provider to return non-None.
 
@@ -429,6 +569,21 @@ class ChainProvider(BaseProvider):
                 return self._convert_type(value)
         return None
 
+    def set_default_provider(self, default_provider):
+        if self._providers and isinstance(self._providers[-1], ConstantProvider):
+            self._providers[-1] = default_provider
+        else:
+            self._providers.append(default_provider)
+
+        num_of_constants = sum(isinstance(
+            provider, ConstantProvider
+        ) for provider in self._providers)
+        if num_of_constants > 1:
+            logger.info(
+                'ChainProvider object contains multiple '
+                'instances of ConstantProvider objects'
+            )
+
     def _convert_type(self, value):
         if self._conversion_func is not None:
             return self._conversion_func(value)
@@ -453,6 +608,12 @@ class InstanceVarProvider(BaseProvider):
         self._instance_var = instance_var
         self._session = session
 
+    def __deepcopy__(self, memo):
+        return InstanceVarProvider(
+            copy.deepcopy(self._instance_var, memo),
+            self._session
+        )
+
     def provide(self):
         """Provide a config value from the session instance vars."""
         instance_vars = self._session.instance_variables()
@@ -483,6 +644,12 @@ class ScopedConfigProvider(BaseProvider):
         self._config_var_name = config_var_name
         self._session = session
 
+    def __deepcopy__(self, memo):
+        return ScopedConfigProvider(
+            copy.deepcopy(self._config_var_name, memo),
+            self._session
+        )
+
     def provide(self):
         """Provide a value from a config file property."""
         scoped_config = self._session.get_scoped_config()
@@ -514,6 +681,12 @@ class EnvironmentProvider(BaseProvider):
         self._name = name
         self._env = env
 
+    def __deepcopy__(self, memo):
+        return EnvironmentProvider(
+            copy.deepcopy(self._name, memo),
+            copy.deepcopy(self._env, memo)
+        )
+
     def provide(self):
         """Provide a config value from a source dictionary."""
         if self._name in self._env:
@@ -539,6 +712,13 @@ class SectionConfigProvider(BaseProvider):
         if self._override_providers is None:
             self._override_providers = {}
 
+    def __deepcopy__(self, memo):
+        return SectionConfigProvider(
+            copy.deepcopy(self._section_name, memo),
+            self._session,
+            copy.deepcopy(self._override_providers, memo)
+        )
+
     def provide(self):
         section_config = self._scoped_config_provider.provide()
         if section_config and not isinstance(section_config, dict):
@@ -554,6 +734,15 @@ class SectionConfigProvider(BaseProvider):
                 section_config[section_config_var] = provider_val
         return section_config
 
+    def set_default_provider(self, key, default_provider):
+        provider = self._override_providers.get(key)
+        if isinstance(provider, ChainProvider):
+            provider.set_default_provider(default_provider)
+            return
+        elif isinstance(provider, BaseProvider):
+            default_provider = ChainProvider(providers=[provider, default_provider])
+        self._override_providers[key] = default_provider
+
     def __repr__(self):
         return (
             'SectionConfigProvider(section_name=%s, '
@@ -569,6 +758,9 @@ class ConstantProvider(BaseProvider):
     def __init__(self, value):
         self._value = value
 
+    def __deepcopy__(self, memo):
+        return ConstantProvider(copy.deepcopy(self._value, memo))
+
     def provide(self):
         """Provide the constant value given during initialization."""
         return self._value
diff --git a/botocore/data/sdk-default-configuration.json b/botocore/data/sdk-default-configuration.json
new file mode 100644
index 000000000..3db13b26c
--- /dev/null
+++ b/botocore/data/sdk-default-configuration.json
@@ -0,0 +1,55 @@
+{
+  "version": 1,
+  "base": {
+    "retryMode": "standard",
+    "stsRegionalEndpoints": "regional",
+    "s3UsEast1RegionalEndpoints": "regional",
+    "connectTimeoutInMillis": 1100,
+    "tlsNegotiationTimeoutInMillis": 1100
+  },
+  "modes": {
+    "standard": {
+      "connectTimeoutInMillis": {
+        "override": 3100
+      },
+      "tlsNegotiationTimeoutInMillis": {
+        "override": 3100
+      }
+    },
+    "in-region": {
+    },
+    "cross-region": {
+      "connectTimeoutInMillis": {
+        "override": 3100
+      },
+      "tlsNegotiationTimeoutInMillis": {
+        "override": 3100
+      }
+    },
+    "mobile": {
+      "connectTimeoutInMillis": {
+        "override": 30000
+      },
+      "tlsNegotiationTimeoutInMillis": {
+        "override": 30000
+      }
+    }
+  },
+  "documentation": {
+    "modes": {
+      "standard": "<p>The STANDARD mode provides the latest recommended default values that should be safe to run in most scenarios</p><p>Note that the default values vended from this mode might change as best practices may evolve. As a result, it is encouraged to perform tests when upgrading the SDK</p>",
+      "in-region": "<p>The IN_REGION mode builds on the standard mode and includes optimization tailored for applications which call AWS services from within the same AWS region</p><p>Note that the default values vended from this mode might change as best practices may evolve. As a result, it is encouraged to perform tests when upgrading the SDK</p>",
+      "cross-region": "<p>The CROSS_REGION mode builds on the standard mode and includes optimization tailored for applications which call AWS services in a different region</p><p>Note that the default values vended from this mode might change as best practices may evolve. As a result, it is encouraged to perform tests when upgrading the SDK</p>",
+      "mobile": "<p>The MOBILE mode builds on the standard mode and includes optimization tailored for mobile applications</p><p>Note that the default values vended from this mode might change as best practices may evolve. As a result, it is encouraged to perform tests when upgrading the SDK</p>",
+      "auto": "<p>The AUTO mode is an experimental mode that builds on the standard mode. The SDK will attempt to discover the execution environment to determine the appropriate settings automatically.</p><p>Note that the auto detection is heuristics-based and does not guarantee 100% accuracy. STANDARD mode will be used if the execution environment cannot be determined. The auto detection might query <a href=\"https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html\">EC2 Instance Metadata service</a>, which might introduce latency. Therefore we recommend choosing an explicit defaults_mode instead if startup latency is critical to your application</p>",
+      "legacy": "<p>The LEGACY mode provides default settings that vary per SDK and were used prior to establishment of defaults_mode</p>"
+    },
+    "configuration": {
+      "retryMode": "<p>A retry mode specifies how the SDK attempts retries. See <a href=\"https://docs.aws.amazon.com/sdkref/latest/guide/setting-global-retry_mode.html\">Retry Mode</a></p>",
+      "stsRegionalEndpoints": "<p>Specifies how the SDK determines the AWS service endpoint that it uses to talk to the AWS Security Token Service (AWS STS). See <a href=\"https://docs.aws.amazon.com/sdkref/latest/guide/setting-global-sts_regional_endpoints.html\">Setting STS Regional endpoints</a></p>",
+      "s3UsEast1RegionalEndpoints": "<p>Specifies how the SDK determines the AWS service endpoint that it uses to talk to the Amazon S3 for the us-east-1 region</p>",
+      "connectTimeoutInMillis": "<p>The amount of time after making an initial connection attempt on a socket, where if the client does not receive a completion of the connect handshake, the client gives up and fails the operation</p>",
+      "tlsNegotiationTimeoutInMillis": "<p>The maximum amount of time that a TLS handshake is allowed to take from the time the CLIENT HELLO message is sent to ethe time the client and server have fully negotiated ciphers and exchanged keys</p>"
+    }
+  }
+}
\ No newline at end of file
diff --git a/botocore/exceptions.py b/botocore/exceptions.py
index 65784270c..32f568fab 100644
--- a/botocore/exceptions.py
+++ b/botocore/exceptions.py
@@ -702,3 +702,10 @@ class InvalidProxiesConfigError(BotoCoreError):
     fmt = (
         'Invalid configuration value(s) provided for proxies_config.'
     )
+
+
+class InvalidDefaultsMode(BotoCoreError):
+    fmt = (
+        'Client configured with invalid defaults mode: {mode}. '
+        'Valid defaults modes include: {valid_modes}.'
+    )
diff --git a/botocore/session.py b/botocore/session.py
index 8b803c57c..0a2334491 100644
--- a/botocore/session.py
+++ b/botocore/session.py
@@ -32,8 +32,10 @@ from botocore.configprovider import ConfigValueStore
 from botocore.configprovider import ConfigChainFactory
 from botocore.configprovider import create_botocore_default_config_mapping
 from botocore.configprovider import BOTOCORE_DEFAUT_SESSION_VARIABLES
+from botocore.configprovider import DefaultConfigResolver
+from botocore.configprovider import SmartDefaultsConfigStoreFactory
 from botocore.exceptions import (
-    ConfigNotFound, ProfileNotFound, UnknownServiceError,
+    ConfigNotFound, ProfileNotFound, UnknownServiceError, InvalidDefaultsMode,
     PartialCredentialsError,
 )
 from botocore.errorfactory import ClientExceptionsFactory
@@ -48,7 +50,7 @@ from botocore import monitoring
 from botocore import paginate
 from botocore import waiter
 from botocore import retryhandler, translate
-from botocore.utils import EVENT_ALIASES, validate_region_name
+from botocore.utils import EVENT_ALIASES, validate_region_name, IMDSRegionProvider
 from botocore.compat import MutableMapping, HAS_CRT
 
 
@@ -137,6 +139,8 @@ class Session(object):
         self._register_exceptions_factory()
         self._register_config_store()
         self._register_monitor()
+        self._register_default_config_resolver()
+        self._register_smart_defaults_factory()
 
     def _register_event_emitter(self):
         self._components.register_component('event_emitter', self._events)
@@ -163,6 +167,24 @@ class Session(object):
         self._internal_components.lazy_register_component(
             'endpoint_resolver', create_default_resolver)
 
+    def _register_default_config_resolver(self):
+        def create_default_config_resolver():
+            loader = self.get_component('data_loader')
+            defaults = loader.load_data('sdk-default-configuration')
+            return DefaultConfigResolver(defaults)
+        self._internal_components.lazy_register_component(
+            'default_config_resolver', create_default_config_resolver)
+
+    def _register_smart_defaults_factory(self):
+        def create_smart_defaults_factory():
+            default_config_resolver = self._get_internal_component(
+                'default_config_resolver')
+            imds_region_provider = IMDSRegionProvider(session=self)
+            return SmartDefaultsConfigStoreFactory(
+                default_config_resolver, imds_region_provider)
+        self._internal_components.lazy_register_component(
+            'smart_defaults_factory', create_smart_defaults_factory)
+
     def _register_response_parser_factory(self):
         self._components.register_component('response_parser_factory',
                                             ResponseParserFactory())
@@ -834,6 +856,13 @@ class Session(object):
         endpoint_resolver = self._get_internal_component('endpoint_resolver')
         exceptions_factory = self._get_internal_component('exceptions_factory')
         config_store = self.get_component('config_store')
+        defaults_mode = self._resolve_defaults_mode(config, config_store)
+        if defaults_mode != 'legacy':
+            smart_defaults_factory = self._get_internal_component(
+                'smart_defaults_factory')
+            config_store = copy.deepcopy(config_store)
+            smart_defaults_factory.merge_smart_defaults(
+                config_store, defaults_mode, region_name)
         client_creator = botocore.client.ClientCreator(
             loader, endpoint_resolver, self.user_agent(), event_emitter,
             retryhandler, translate, response_parser_factory,
@@ -870,6 +899,24 @@ class Session(object):
         self._last_client_region_used = region_name
         return region_name
 
+    def _resolve_defaults_mode(self, client_config, config_store):
+        mode = config_store.get_config_variable('defaults_mode')
+
+        if client_config and client_config.defaults_mode:
+            mode = client_config.defaults_mode
+
+        default_config_resolver = self._get_internal_component(
+            'default_config_resolver')
+        default_modes = default_config_resolver.get_default_modes()
+        lmode = mode.lower()
+        if lmode not in default_modes:
+            raise InvalidDefaultsMode(
+                mode=mode,
+                valid_modes=', '.join(default_modes)
+            )
+
+        return lmode
+
     def _missing_cred_vars(self, access_key, secret_key):
         if access_key is not None and secret_key is None:
             return 'aws_secret_access_key'
diff --git a/botocore/utils.py b/botocore/utils.py
index c7a4d75aa..8a32cf6ea 100644
--- a/botocore/utils.py
+++ b/botocore/utils.py
@@ -611,6 +611,102 @@ class InstanceMetadataFetcher(IMDSFetcher):
         return True
 
 
+class IMDSRegionProvider(object):
+    def __init__(self, session, environ=None, fetcher=None):
+        """Initialize IMDSRegionProvider.
+        :type session: :class:`botocore.session.Session`
+        :param session: The session is needed to look up configuration for
+            how to contact the instance metadata service. Specifically the
+            whether or not it should use the IMDS region at all, and if so how
+            to configure the timeout and number of attempts to reach the
+            service.
+        :type environ: None or dict
+        :param environ: A dictionary of environment variables to use. If
+            ``None`` is the argument then ``os.environ`` will be used by
+            default.
+        :type fecther: :class:`botocore.utils.InstanceMetadataRegionFetcher`
+        :param fetcher: The class to actually handle the fetching of the region
+            from the IMDS. If not provided a default one will be created.
+        """
+        self._session = session
+        if environ is None:
+            environ = os.environ
+        self._environ = environ
+        self._fetcher = fetcher
+
+    def provide(self):
+        """Provide the region value from IMDS."""
+        instance_region = self._get_instance_metadata_region()
+        return instance_region
+
+    def _get_instance_metadata_region(self):
+        fetcher = self._get_fetcher()
+        region = fetcher.retrieve_region()
+        return region
+
+    def _get_fetcher(self):
+        if self._fetcher is None:
+            self._fetcher = self._create_fetcher()
+        return self._fetcher
+
+    def _create_fetcher(self):
+        metadata_timeout = self._session.get_config_variable(
+            'metadata_service_timeout')
+        metadata_num_attempts = self._session.get_config_variable(
+            'metadata_service_num_attempts')
+        imds_config = {
+            'ec2_metadata_service_endpoint': self._session.get_config_variable(
+                'ec2_metadata_service_endpoint'),
+            'ec2_metadata_service_endpoint_mode': resolve_imds_endpoint_mode(
+                self._session
+            )
+        }
+        fetcher = InstanceMetadataRegionFetcher(
+            timeout=metadata_timeout,
+            num_attempts=metadata_num_attempts,
+            env=self._environ,
+            user_agent=self._session.user_agent(),
+            config=imds_config,
+        )
+        return fetcher
+
+
+class InstanceMetadataRegionFetcher(IMDSFetcher):
+    _URL_PATH = 'latest/meta-data/placement/availability-zone/'
+
+    def retrieve_region(self):
+        """Get the current region from the instance metadata service.
+        :rvalue: str
+        :returns: The region the current instance is running in or None
+            if the instance metadata service cannot be contacted or does not
+            give a valid response.
+        :rtype: None or str
+        :returns: Returns the region as a string if it is configured to use
+            IMDS as a region source. Otherwise returns ``None``. It will also
+            return ``None`` if it fails to get the region from IMDS due to
+            exhausting its retries or not being able to connect.
+        """
+        try:
+            region = self._get_region()
+            return region
+        except self._RETRIES_EXCEEDED_ERROR_CLS:
+            logger.debug("Max number of attempts exceeded (%s) when "
+                         "attempting to retrieve data from metadata service.",
+                         self._num_attempts)
+        return None
+
+    def _get_region(self):
+        token = self._fetch_metadata_token()
+        response = self._get_request(
+            url_path=self._URL_PATH,
+            retry_func=self._default_retry,
+            token=token
+        )
+        availability_zone = response.text
+        region = availability_zone[:-1]
+        return region
+
+
 def merge_dicts(dict1, dict2, append_lists=False):
     """Given two dict, merge the second dict into the first.
 


Test patch:
diff --git a/tests/functional/test_config_provider.py b/tests/functional/test_config_provider.py
new file mode 100644
index 000000000..716eb1dda
--- /dev/null
+++ b/tests/functional/test_config_provider.py
@@ -0,0 +1,47 @@
+# Copyright 2022 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+import pytest
+
+from botocore.session import get_session
+from botocore.config import Config
+
+_SDK_DEFAULT_CONFIGURATION_VALUES_ALLOWLIST = (
+    'retryMode', 'stsRegionalEndpoints', 's3UsEast1RegionalEndpoints',
+    'connectTimeoutInMillis', 'tlsNegotiationTimeoutInMillis'
+)
+
+session = get_session()
+loader = session.get_component('data_loader')
+sdk_default_configuration = loader.load_data('sdk-default-configuration')
+
+
+@pytest.mark.parametrize("mode", sdk_default_configuration['base'])
+def test_no_new_sdk_default_configuration_values(mode):
+    err_msg = (
+        f'New default configuration value {mode} introduced to '
+        f'sdk-default-configuration.json. Support for setting {mode} must be '
+        'considered and added to the DefaulConfigResolver. In addition, '
+        'must add value to _SDK_DEFAULT_CONFIGURATION_VALUES_ALLOWLIST.'
+    )
+    assert mode in _SDK_DEFAULT_CONFIGURATION_VALUES_ALLOWLIST, err_msg
+
+
+def test_default_configurations_resolve_correctly():
+    session = get_session()
+    config = Config(defaults_mode='standard')
+    client = session.create_client(
+        'sts', config=config, region_name='us-west-2')
+    assert client.meta.config.s3['us_east_1_regional_endpoint'] == 'regional'
+    assert client.meta.config.connect_timeout == 3.1
+    assert client.meta.endpoint_url == 'https://sts.us-west-2.amazonaws.com'
+    assert client.meta.config.retries['mode'] == 'standard'
diff --git a/tests/unit/test_args.py b/tests/unit/test_args.py
index dd53d1f51..b0a9171ed 100644
--- a/tests/unit/test_args.py
+++ b/tests/unit/test_args.py
@@ -409,6 +409,23 @@ class TestCreateClientArgs(unittest.TestCase):
         )['client_config']
         self.assertEqual(config.retries['mode'], 'standard')
 
+    def test_connect_timeout_set_on_config_store(self):
+        self.config_store.set_config_variable('connect_timeout', 10)
+        config = self.call_get_client_args(
+            client_config=Config(defaults_mode='standard')
+        )['client_config']
+        self.assertEqual(config.connect_timeout, 10)
+
+    def test_connnect_timeout_set_on_client_config(self):
+        config = self.call_get_client_args(
+            client_config=Config(connect_timeout=10)
+        )['client_config']
+        self.assertEqual(config.connect_timeout, 10)
+
+    def test_connnect_timeout_set_to_client_config_default(self):
+        config = self.call_get_client_args()['client_config']
+        self.assertEqual(config.connect_timeout, 60)
+
     def test_client_config_beats_config_store(self):
         self.config_store.set_config_variable('retry_mode', 'adaptive')
         config = self.call_get_client_args(
diff --git a/tests/unit/test_config_provider.py b/tests/unit/test_config_provider.py
index 4e058b246..b25739bcb 100644
--- a/tests/unit/test_config_provider.py
+++ b/tests/unit/test_config_provider.py
@@ -10,12 +10,15 @@
 # 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 copy
 import pytest
+import os
 
 from tests import mock
 from tests import unittest
 
 import botocore.session as session
+import botocore.configprovider
 from botocore.configprovider import ConfigValueStore
 from botocore.configprovider import BaseProvider
 from botocore.configprovider import InstanceVarProvider
@@ -25,6 +28,10 @@ from botocore.configprovider import SectionConfigProvider
 from botocore.configprovider import ConstantProvider
 from botocore.configprovider import ChainProvider
 from botocore.configprovider import ConfigChainFactory
+from botocore.configprovider import SmartDefaultsConfigStoreFactory
+from botocore.configprovider import DefaultConfigResolver
+from botocore.utils import IMDSRegionProvider
+from botocore.exceptions import ConnectTimeoutError
 
 
 class TestConfigChainFactory(unittest.TestCase):
@@ -327,6 +334,28 @@ class TestConfigValueStore(unittest.TestCase):
         value = provider.get_config_variable('fake_variable')
         self.assertEqual(value, 'bar')
 
+    def test_can_get_config_provider(self):
+        chain_provider = ChainProvider(
+            providers=[ConstantProvider(value='bar')]
+        )
+        config_value_store = ConfigValueStore(mapping={
+            'fake_variable': chain_provider,
+        })
+        provider = config_value_store.get_config_provider('fake_variable')
+        value = config_value_store.get_config_variable('fake_variable')
+        self.assertIsInstance(provider, ChainProvider)
+        self.assertEqual(value, 'bar')
+
+    def test_can_get_config_provider_non_chain_provider(self):
+        constant_provider = ConstantProvider(value='bar')
+        config_value_store = ConfigValueStore(mapping={
+            'fake_variable': constant_provider,
+        })
+        provider = config_value_store.get_config_provider('fake_variable')
+        value = config_value_store.get_config_variable('fake_variable')
+        self.assertIsInstance(provider, ConstantProvider)
+        self.assertEqual(value, 'bar')
+
 
 class TestInstanceVarProvider(unittest.TestCase):
     def assert_provides_value(self, name, instance_map, expected_value):
@@ -556,3 +585,224 @@ class TestSectionConfigProvider(unittest.TestCase):
                 'override_var': 'override',
             }
         )
+
+
+class TestSmartDefaults:
+
+    def _template(self):
+        return {
+            "base": {
+                "retryMode": "standard",
+                "stsRegionalEndpoints": "regional",
+                "s3UsEast1RegionalEndpoints": "regional",
+                "connectTimeoutInMillis": 1000,
+                "tlsNegotiationTimeoutInMillis": 1000
+            },
+            "modes": {
+                "standard": {
+                    "connectTimeoutInMillis": {
+                        "multiply": 2
+                    },
+                    "tlsNegotiationTimeoutInMillis": {
+                        "multiply": 2
+                    }
+                },
+                "in-region": {
+                    "connectTimeoutInMillis": {
+                        "multiply": 1
+                    },
+                    "tlsNegotiationTimeoutInMillis": {
+                        "multiply": 1
+                    }
+                },
+                "cross-region": {
+                    "connectTimeoutInMillis": {
+                        "multiply": 2.8
+                    },
+                    "tlsNegotiationTimeoutInMillis": {
+                        "multiply": 2.8
+                    }
+                },
+                "mobile": {
+                    "connectTimeoutInMillis": {
+                        "override": 10000
+                    },
+                    "tlsNegotiationTimeoutInMillis": {
+                        "add": 10000
+                    },
+                    "retryMode": {
+                        "override": "adaptive"
+                    }
+                }
+            }
+        }
+
+    def _create_default_config_resolver(self):
+        return DefaultConfigResolver(self._template())
+
+    @pytest.fixture
+    def smart_defaults_factory(self):
+        fake_session = mock.Mock(spec=session.Session)
+        fake_session.get_scoped_config.return_value = {}
+        default_config_resolver = self._create_default_config_resolver()
+        return SmartDefaultsConfigStoreFactory(
+            default_config_resolver, imds_region_provider=mock.Mock()
+        )
+
+    @pytest.fixture
+    def fake_session(self):
+        fake_session = mock.Mock(spec=session.Session)
+        fake_session.get_scoped_config.return_value = {}
+        return fake_session
+
+    def _create_config_value_store(self, s3_mapping={}, **override_kwargs):
+        provider_foo = ConstantProvider(value='foo')
+        environment_provider_foo = EnvironmentProvider(
+            name='AWS_RETRY_MODE',
+            env={'AWS_RETRY_MODE': None}
+        )
+        fake_session = mock.Mock(spec=session.Session)
+        fake_session.get_scoped_config.return_value = {}
+        # Testing with three different providers to validate
+        # SmartDefaultsConfigStoreFactory._get_new_chain_provider
+        mapping = {
+            'sts_regional_endpoints': ChainProvider(providers=[provider_foo]),
+            'retry_mode': ChainProvider(providers=[environment_provider_foo]),
+            's3': SectionConfigProvider('s3', fake_session, s3_mapping)
+        }
+        mapping.update(**override_kwargs)
+        config_store = ConfigValueStore(mapping=mapping)
+        return config_store
+
+    def _create_os_environ_patcher(self):
+        return mock.patch.object(
+            botocore.configprovider.os, 'environ',
+            mock.Mock(wraps=os.environ)
+        )
+
+    def test_config_store_deepcopy(self):
+        config_store = ConfigValueStore()
+        config_store.set_config_provider('foo', ConstantProvider('bar'))
+        config_store_copy = copy.deepcopy(config_store)
+        config_store_copy.set_config_provider('fizz', ConstantProvider('buzz'))
+        assert config_store.get_config_variable('fizz') is None
+        assert config_store_copy.get_config_variable('foo') == 'bar'
+
+    @pytest.mark.parametrize(
+        'defaults_mode, retry_mode, sts_regional_endpoints,'
+        ' us_east_1_regional_endpoint, connect_timeout',
+        [
+            ('standard', 'standard', 'regional', 'regional', 2000),
+            ('in-region', 'standard', 'regional', 'regional', 1000),
+            ('cross-region', 'standard', 'regional', 'regional', 2800),
+            ('mobile', 'adaptive', 'regional', 'regional', 10000),
+        ]
+    )
+    def test_get_defualt_config_values(self, defaults_mode, retry_mode,
+                                       sts_regional_endpoints,
+                                       us_east_1_regional_endpoint,
+                                       connect_timeout):
+        default_config_resolver = self._create_default_config_resolver()
+        default_values = default_config_resolver.get_default_config_values(
+            defaults_mode
+        )
+        assert default_values['retryMode'] == retry_mode
+        assert default_values['stsRegionalEndpoints'] == sts_regional_endpoints
+        assert default_values['s3UsEast1RegionalEndpoints'] == us_east_1_regional_endpoint
+        assert default_values['connectTimeoutInMillis'] == connect_timeout
+
+    def test_resolve_default_values_on_config(self, smart_defaults_factory,
+                                              fake_session):
+        config_store = self._create_config_value_store()
+        smart_defaults_factory.merge_smart_defaults(
+            config_store, 'standard', 'foo')
+        s3_config = config_store.get_config_variable('s3')
+        assert s3_config['us_east_1_regional_endpoint'] == 'regional'
+        assert config_store.get_config_variable('retry_mode') == 'standard'
+        assert config_store.get_config_variable('sts_regional_endpoints') == 'regional'
+        assert config_store.get_config_variable('connect_timeout') == 2
+
+    def test_no_resolve_default_s3_values_on_config(self,
+                                                    smart_defaults_factory,
+                                                    fake_session):
+        environment_provider = EnvironmentProvider(
+            name='AWS_S3_US_EAST_1_REGIONAL_ENDPOINT',
+            env={'AWS_S3_US_EAST_1_REGIONAL_ENDPOINT': 'legacy'}
+        )
+        s3_mapping = {
+            'us_east_1_regional_endpoint': ChainProvider(
+                providers=[environment_provider])
+        }
+        config_store = self._create_config_value_store(s3_mapping=s3_mapping)
+        smart_defaults_factory.merge_smart_defaults(
+            config_store, 'standard', 'foo')
+        s3_config = config_store.get_config_variable('s3')
+        assert s3_config['us_east_1_regional_endpoint'] == 'legacy'
+        assert config_store.get_config_variable('retry_mode') == 'standard'
+        assert config_store.get_config_variable('sts_regional_endpoints') == 'regional'
+        assert config_store.get_config_variable('connect_timeout') == 2
+
+    def test_resolve_default_s3_values_on_config(self, smart_defaults_factory,
+                                                 fake_session):
+        s3_mapping = {
+            'use_arn_region': ChainProvider(
+                providers=[ConstantProvider(value=False)])
+        }
+        config_store = self._create_config_value_store(s3_mapping=s3_mapping)
+        smart_defaults_factory.merge_smart_defaults(
+            config_store, 'standard', 'foo')
+        s3_config = config_store.get_config_variable('s3')
+        assert s3_config['us_east_1_regional_endpoint'] == 'regional'
+        assert config_store.get_config_variable('retry_mode') == 'standard'
+        assert config_store.get_config_variable('sts_regional_endpoints') == 'regional'
+        assert config_store.get_config_variable('connect_timeout') == 2
+
+    @pytest.mark.parametrize(
+        'execution_env_var, region_env_var, default_region_env_var, '
+        'imds_region, client_region, resolved_mode',
+        [
+            ('AWS_Lambda_python3.6', 'us-east-1', None,
+             None, 'us-east-1', 'in-region'),
+            ('AWS_Lambda_python3.6', 'us-west-2', 'us-west-2',
+             None, 'us-east-1', 'cross-region'),
+            ('AWS_Lambda_python3.6', None, None,
+             'us-west-2', 'us-east-1', 'cross-region'),
+            (None, None, 'us-east-1',
+             'us-east-1', 'us-east-1', 'in-region'),
+            (None, None, None,
+             'us-west-2', 'us-east-1', 'cross-region'),
+            (None, None, None,
+             None, 'us-west-2', 'standard'),
+        ]
+    )
+    def test_resolve_auto_mode(self, execution_env_var, region_env_var,
+                               default_region_env_var,
+                               imds_region, client_region, resolved_mode):
+        imds_region_provider = mock.Mock(spec=IMDSRegionProvider)
+        imds_region_provider.provide.return_value = imds_region
+        default_config_resolver = mock.Mock()
+        with mock.patch.object(
+            botocore.configprovider.os, 'environ',
+            mock.Mock(wraps=os.environ)
+        ) as os_environ_patcher:
+            os_environ_patcher.get.side_effect = [
+                execution_env_var, default_region_env_var, region_env_var]
+            smart_defaults_factory = SmartDefaultsConfigStoreFactory(
+                default_config_resolver, imds_region_provider)
+            mode = smart_defaults_factory.resolve_auto_mode(client_region)
+            assert mode == resolved_mode
+
+    def test_resolve_auto_mode_imds_region_provider_connect_timeout(self):
+        imds_region_provider = mock.Mock(spec=IMDSRegionProvider)
+        imds_region_provider.provide.side_effect = ConnectTimeoutError(
+            endpoint_url='foo')
+        default_config_resolver = mock.Mock()
+        with mock.patch.object(
+            botocore.configprovider.os, 'environ',
+            mock.Mock(wraps=os.environ)
+        ) as os_environ_patcher:
+            os_environ_patcher.get.side_effect = [None] * 3
+            smart_defaults_factory = SmartDefaultsConfigStoreFactory(
+                default_config_resolver, imds_region_provider)
+            mode = smart_defaults_factory.resolve_auto_mode('us-west-2')
+            assert mode == 'standard'
diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py
index 65b80fd93..703ad235e 100644
--- a/tests/unit/test_session.py
+++ b/tests/unit/test_session.py
@@ -728,6 +728,33 @@ class TestCreateClient(BaseSessionTest):
                 create_client.call_args[1]
             self.assertEqual(call_kwargs['api_version'], override_api_version)
 
+    @mock.patch('botocore.client.ClientCreator')
+    def test_defaults_mode_resolved_from_config_store(self, client_creator):
+        config_store = self.session.get_component('config_store')
+        config_store.set_config_variable('defaults_mode', 'standard')
+        self.session.create_client('sts', 'us-west-2')
+        self.assertIsNot(client_creator.call_args[0][-1], config_store)
+
+    @mock.patch('botocore.client.ClientCreator')
+    def test_defaults_mode_resolved_from_client_config(self, client_creator):
+        config_store = self.session.get_component('config_store')
+        config = botocore.config.Config(defaults_mode='standard')
+        self.session.create_client('sts', 'us-west-2', config=config)
+        self.assertIsNot(client_creator.call_args[0][-1], config_store)
+
+    @mock.patch('botocore.client.ClientCreator')
+    def test_defaults_mode_resolved_invalid_mode_exception(self,
+                                                           client_creator):
+        with self.assertRaises(botocore.exceptions.InvalidDefaultsMode):
+            config = botocore.config.Config(defaults_mode='foo')
+            self.session.create_client('sts', 'us-west-2', config=config)
+
+    @mock.patch('botocore.client.ClientCreator')
+    def test_defaults_mode_resolved_legacy(self, client_creator):
+        config_store = self.session.get_component('config_store')
+        self.session.create_client('sts', 'us-west-2')
+        self.assertIs(client_creator.call_args[0][-1], config_store)
+
 
 class TestSessionComponent(BaseSessionTest):
     def test_internal_component(self):
diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py
index 5a64bea09..c021cce0e 100644
--- a/tests/unit/test_utils.py
+++ b/tests/unit/test_utils.py
@@ -73,6 +73,8 @@ from botocore.utils import S3ArnParamHandler
 from botocore.utils import S3EndpointSetter
 from botocore.utils import ContainerMetadataFetcher
 from botocore.utils import InstanceMetadataFetcher
+from botocore.utils import InstanceMetadataRegionFetcher
+from botocore.utils import IMDSRegionProvider
 from botocore.utils import SSOTokenLoader
 from botocore.utils import is_valid_uri, is_valid_ipv6_endpoint_url
 from botocore.utils import has_header
@@ -80,6 +82,7 @@ from botocore.exceptions import SSOTokenLoadError
 from botocore.model import DenormalizedStructureBuilder
 from botocore.model import ShapeResolver
 from botocore.config import Config
+from botocore.session import Session
 
 
 class TestEnsureBoolean(unittest.TestCase):
@@ -2696,6 +2699,106 @@ class TestInstanceMetadataFetcher(unittest.TestCase):
         self.assertEqual(result, {})
 
 
+class TestIMDSRegionProvider(unittest.TestCase):
+    def setUp(self):
+        self.environ = {}
+        self.environ_patch = mock.patch('os.environ', self.environ)
+        self.environ_patch.start()
+
+    def tearDown(self):
+        self.environ_patch.stop()
+
+    def assert_does_provide_expected_value(self, fetcher_region=None,
+                                           expected_result=None,):
+        fake_session = mock.Mock(spec=Session)
+        fake_fetcher = mock.Mock(spec=InstanceMetadataRegionFetcher)
+        fake_fetcher.retrieve_region.return_value = fetcher_region
+        provider = IMDSRegionProvider(fake_session, fetcher=fake_fetcher)
+        value = provider.provide()
+        self.assertEqual(value, expected_result)
+
+    def test_does_provide_region_when_present(self):
+        self.assert_does_provide_expected_value(
+            fetcher_region='us-mars-2',
+            expected_result='us-mars-2',
+        )
+
+    def test_does_provide_none(self):
+        self.assert_does_provide_expected_value(
+            fetcher_region=None,
+            expected_result=None,
+        )
+
+    @mock.patch('botocore.httpsession.URLLib3Session.send')
+    def test_use_truncated_user_agent(self, send):
+        session = Session()
+        session = Session()
+        session.user_agent_version = '3.0'
+        provider = IMDSRegionProvider(session)
+        provider.provide()
+        args, _ = send.call_args
+        self.assertIn('Botocore/3.0', args[0].headers['User-Agent'])
+
+    @mock.patch('botocore.httpsession.URLLib3Session.send')
+    def test_can_use_ipv6(self, send):
+        session = Session()
+        session.set_config_variable('imds_use_ipv6', True)
+        provider = IMDSRegionProvider(session)
+        provider.provide()
+        args, _ = send.call_args
+        self.assertIn('[fd00:ec2::254]', args[0].url)
+
+    @mock.patch('botocore.httpsession.URLLib3Session.send')
+    def test_use_ipv4_by_default(self, send):
+        session = Session()
+        provider = IMDSRegionProvider(session)
+        provider.provide()
+        args, _ = send.call_args
+        self.assertIn('169.254.169.254', args[0].url)
+
+    @mock.patch('botocore.httpsession.URLLib3Session.send')
+    def test_can_set_imds_endpoint_mode_to_ipv4(self, send):
+        session = Session()
+        session.set_config_variable(
+            'ec2_metadata_service_endpoint_mode', 'ipv4')
+        provider = IMDSRegionProvider(session)
+        provider.provide()
+        args, _ = send.call_args
+        self.assertIn('169.254.169.254', args[0].url)
+
+    @mock.patch('botocore.httpsession.URLLib3Session.send')
+    def test_can_set_imds_endpoint_mode_to_ipv6(self, send):
+        session = Session()
+        session.set_config_variable(
+            'ec2_metadata_service_endpoint_mode', 'ipv6')
+        provider = IMDSRegionProvider(session)
+        provider.provide()
+        args, _ = send.call_args
+        self.assertIn('[fd00:ec2::254]', args[0].url)
+
+    @mock.patch('botocore.httpsession.URLLib3Session.send')
+    def test_can_set_imds_service_endpoint(self, send):
+        session = Session()
+        session.set_config_variable(
+            'ec2_metadata_service_endpoint', 'http://myendpoint/')
+        provider = IMDSRegionProvider(session)
+        provider.provide()
+        args, _ = send.call_args
+        self.assertIn('http://myendpoint/', args[0].url)
+
+    @mock.patch('botocore.httpsession.URLLib3Session.send')
+    def test_imds_service_endpoint_overrides_ipv6_endpoint(self, send):
+        session = Session()
+        session.set_config_variable(
+            'ec2_metadata_service_endpoint_mode', 'ipv6')
+        session.set_config_variable(
+            'ec2_metadata_service_endpoint', 'http://myendpoint/')
+        provider = IMDSRegionProvider(session)
+        provider.provide()
+        args, _ = send.call_args
+        self.assertIn('http://myendpoint/', args[0].url)
+
+
 class TestSSOTokenLoader(unittest.TestCase):
     def setUp(self):
         super(TestSSOTokenLoader, self).setUp()