Configured endpoint URL resolution

.changes/next-release/bugfix-configprovider-46546.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "configprovider",
+  "description": "Fix bug when deep copying config value store where overrides were not preserved"
+}

.changes/next-release/enhancement-configprovider-27540.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "configprovider",
+  "description": "Always use shallow copy of session config value store for clients"
+}

.changes/next-release/feature-configuration-3829.json

@@ -0,0 +1,5 @@
+{
+  "type": "feature",
+  "category": "configuration",
+  "description": "Configure the endpoint URL in the shared configuration file or via an environment variable for a specific AWS service or all AWS services."
+}

botocore/args.py

@@ -115,7 +115,7 @@ def get_client_args(
         s3_config = final_args['s3_config']
         partition = endpoint_config['metadata'].get('partition', None)
         socket_options = final_args['socket_options']
-
+        configured_endpoint_url = final_args['configured_endpoint_url']
         signing_region = endpoint_config['signing_region']
         endpoint_region_name = endpoint_config['region_name']
 
@@ -160,7 +160,7 @@ def get_client_args(
             service_model,
             endpoint_region_name,
             region_name,
-            endpoint_url,
+            configured_endpoint_url,
             endpoint,
             is_secure,
             endpoint_bridge,
@@ -210,10 +210,16 @@ def compute_client_args(
                 parameter_validation = ensure_boolean(raw_value)
 
         s3_config = self.compute_s3_config(client_config)
+
+        configured_endpoint_url = self._compute_configured_endpoint_url(
+            client_config=client_config,
+            endpoint_url=endpoint_url,
+        )
+
         endpoint_config = self._compute_endpoint_config(
             service_name=service_name,
             region_name=region_name,
-            endpoint_url=endpoint_url,
+            endpoint_url=configured_endpoint_url,
             is_secure=is_secure,
             endpoint_bridge=endpoint_bridge,
             s3_config=s3_config,
@@ -270,6 +276,7 @@ def compute_client_args(
         return {
             'service_name': service_name,
             'parameter_validation': parameter_validation,
+            'configured_endpoint_url': configured_endpoint_url,
             'endpoint_config': endpoint_config,
             'protocol': protocol,
             'config_kwargs': config_kwargs,
@@ -279,6 +286,27 @@ def compute_client_args(
             ),
         }
 
+    def _compute_configured_endpoint_url(self, client_config, endpoint_url):
+        if endpoint_url is not None:
+            return endpoint_url
+
+        if self._ignore_configured_endpoint_urls(client_config):
+            logger.debug("Ignoring configured endpoint URLs.")
+            return endpoint_url
+
+        return self._config_store.get_config_variable('endpoint_url')
+
+    def _ignore_configured_endpoint_urls(self, client_config):
+        if (
+            client_config
+            and client_config.ignore_configured_endpoint_urls is not None
+        ):
+            return client_config.ignore_configured_endpoint_urls
+
+        return self._config_store.get_config_variable(
+            'ignore_configured_endpoint_urls'
+        )
+
     def compute_s3_config(self, client_config):
         s3_configuration = self._config_store.get_config_variable('s3')
 

botocore/config.py

@@ -194,6 +194,13 @@ class Config:
 
         Defaults to None.
 
+    :type ignore_configured_endpoint_urls: bool
+    :param ignore_configured_endpoint_urls: Setting to True disables use
+        of endpoint URLs provided via environment variables and
+        the shared configuration file.
+
+        Defaults to None.
+
     :type tcp_keepalive: bool
     :param tcp_keepalive: Enables the TCP Keep-Alive socket option used when
         creating new connections if set to True.
@@ -221,6 +228,7 @@ class Config:
             ('endpoint_discovery_enabled', None),
             ('use_dualstack_endpoint', None),
             ('use_fips_endpoint', None),
+            ('ignore_configured_endpoint_urls', None),
             ('defaults_mode', None),
             ('tcp_keepalive', None),
         ]

botocore/configloader.py

@@ -200,6 +200,17 @@ def _parse_nested(config_value):
     return parsed
 
 
+def _parse_section(key, values):
+    result = {}
+    try:
+        parts = shlex.split(key)
+    except ValueError:
+        return result
+    if len(parts) == 2:
+        result[parts[1]] = values
+    return result
+
+
 def build_profile_map(parsed_ini_config):
     """Convert the parsed INI config into a profile map.
 
@@ -254,22 +265,15 @@ def build_profile_map(parsed_ini_config):
     parsed_config = copy.deepcopy(parsed_ini_config)
     profiles = {}
     sso_sessions = {}
+    services = {}
     final_config = {}
     for key, values in parsed_config.items():
         if key.startswith("profile"):
-            try:
-                parts = shlex.split(key)
-            except ValueError:
-                continue
-            if len(parts) == 2:
-                profiles[parts[1]] = values
+            profiles.update(_parse_section(key, values))
         elif key.startswith("sso-session"):
-            try:
-                parts = shlex.split(key)
-            except ValueError:
-                continue
-            if len(parts) == 2:
-                sso_sessions[parts[1]] = values
+            sso_sessions.update(_parse_section(key, values))
+        elif key.startswith("services"):
+            services.update(_parse_section(key, values))
         elif key == 'default':
             # default section is special and is considered a profile
             # name but we don't require you use 'profile "default"'
@@ -279,4 +283,5 @@ def build_profile_map(parsed_ini_config):
             final_config[key] = values
     final_config['profiles'] = profiles
     final_config['sso_sessions'] = sso_sessions
+    final_config['services'] = services
     return final_config

botocore/configprovider.py

@@ -18,6 +18,7 @@
 import os
 
 from botocore import utils
+from botocore.exceptions import InvalidConfigError
 
 logger = logging.getLogger(__name__)
 
@@ -108,6 +109,12 @@
         None,
         utils.ensure_boolean,
     ),
+    'ignore_configured_endpoint_urls': (
+        'ignore_configured_endpoint_urls',
+        'AWS_IGNORE_CONFIGURED_ENDPOINT_URLS',
+        None,
+        utils.ensure_boolean,
+    ),
     'parameter_validation': ('parameter_validation', None, True, None),
     # Client side monitoring configurations.
     # Note: These configurations are considered internal to botocore.
@@ -403,7 +410,18 @@ def __init__(self, mapping=None):
                 self.set_config_provider(logical_name, provider)
 
     def __deepcopy__(self, memo):
-        return ConfigValueStore(copy.deepcopy(self._mapping, memo))
+        config_store = ConfigValueStore(copy.deepcopy(self._mapping, memo))
+        for logical_name, override_value in self._overrides.items():
+            config_store.set_config_variable(logical_name, override_value)
+
+        return config_store
+
+    def __copy__(self):
+        config_store = ConfigValueStore(copy.copy(self._mapping))
+        for logical_name, override_value in self._overrides.items():
+            config_store.set_config_variable(logical_name, override_value)
+
+        return config_store
 
     def get_config_variable(self, logical_name):
         """
@@ -543,24 +561,28 @@ def resolve_auto_mode(self, region_name):
         return 'standard'
 
     def _update_provider(self, config_store, variable, value):
-        provider = config_store.get_config_provider(variable)
+        original_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):
+        if isinstance(original_provider, ChainProvider):
+            chain_provider_copy = copy.deepcopy(original_provider)
+            chain_provider_copy.set_default_provider(default_provider)
+            default_provider = chain_provider_copy
+        elif isinstance(original_provider, BaseProvider):
             default_provider = ChainProvider(
-                providers=[provider, default_provider]
+                providers=[original_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(
+        section_provider_copy = copy.deepcopy(
+            config_store.get_config_provider(section_name)
+        )
+        section_provider_copy.set_default_provider(
             variable, ConstantProvider(value)
         )
+        config_store.set_config_provider(section_name, section_provider_copy)
 
     def _set_retryMode(self, config_store, value):
         self._update_provider(config_store, 'retry_mode', value)
@@ -837,3 +859,142 @@ def provide(self):
 
     def __repr__(self):
         return 'ConstantProvider(value=%s)' % self._value
+
+
+class ConfiguredEndpointProvider(BaseProvider):
+    """Lookup an endpoint URL from environment variable or shared config file.
+
+    NOTE: This class is considered private and is subject to abrupt breaking
+    changes or removal without prior announcement. Please do not use it
+    directly.
+    """
+
+    _ENDPOINT_URL_LOOKUP_ORDER = [
+        'environment_service',
+        'environment_global',
+        'config_service',
+        'config_global',
+    ]
+
+    def __init__(
+        self,
+        full_config,
+        scoped_config,
+        client_name,
+        environ=None,
+    ):
+        """Initialize a ConfiguredEndpointProviderChain.
+
+        :type full_config: dict
+        :param full_config: This is the dict representing the full
+            configuration file.
+
+        :type scoped_config: dict
+        :param scoped_config: This is the dict representing the configuration
+            for the current profile for the session.
+
+        :type client_name: str
+        :param client_name: The name used to instantiate a client using
+            botocore.session.Session.create_client.
+
+        :type environ: dict
+        :param environ: A mapping to use for environment variables. If this
+            is not provided it will default to use os.environ.
+        """
+        self._full_config = full_config
+        self._scoped_config = scoped_config
+        self._client_name = client_name
+        self._transformed_service_id = self._get_snake_case_service_id(
+            self._client_name
+        )
+        if environ is None:
+            environ = os.environ
+        self._environ = environ
+
+    def provide(self):
+        """Lookup the configured endpoint URL.
+
+        The order is:
+
+        1. The value provided by a service-specific environment variable.
+        2. The value provided by the global endpoint environment variable
+           (AWS_ENDPOINT_URL).
+        3. The value provided by a service-specific parameter from a services
+           definition section in the shared configuration file.
+        4. The value provided by the global parameter from a services
+           definition section in the shared configuration file.
+        """
+        for location in self._ENDPOINT_URL_LOOKUP_ORDER:
+            logger.debug(
+                'Looking for endpoint for %s via: %s',
+                self._client_name,
+                location,
+            )
+
+            endpoint_url = getattr(self, f'_get_endpoint_url_{location}')()
+
+            if endpoint_url:
+                logger.info(
+                    'Found endpoint for %s via: %s.',
+                    self._client_name,
+                    location,
+                )
+                return endpoint_url
+
+        logger.debug('No configured endpoint found.')
+        return None
+
+    def _get_snake_case_service_id(self, client_name):
+        # Get the service ID without loading the service data file, accounting
+        # for any aliases and standardizing the names with hyphens.
+        client_name = utils.SERVICE_NAME_ALIASES.get(client_name, client_name)
+        hyphenized_service_id = (
+            utils.CLIENT_NAME_TO_HYPHENIZED_SERVICE_ID_OVERRIDES.get(
+                client_name, client_name
+            )
+        )
+        return hyphenized_service_id.replace('-', '_')
+
+    def _get_service_env_var_name(self):
+        transformed_service_id_env = self._transformed_service_id.upper()
+        return f'AWS_ENDPOINT_URL_{transformed_service_id_env}'
+
+    def _get_services_config(self):
+        if 'services' not in self._scoped_config:
+            return {}
+
+        section_name = self._scoped_config['services']
+        services_section = self._full_config.get('services', {}).get(
+            section_name
+        )
+
+        if not services_section:
+            error_msg = (
+                f'The profile is configured to use the services '
+                f'section but the "{section_name}" services '
+                f'configuration does not exist.'
+            )
+            raise InvalidConfigError(error_msg=error_msg)
+
+        return services_section
+
+    def _get_endpoint_url_config_service(self):
+        snakecase_service_id = self._transformed_service_id.lower()
+        return (
+            self._get_services_config()
+            .get(snakecase_service_id, {})
+            .get('endpoint_url')
+        )
+
+    def _get_endpoint_url_config_global(self):
+        return self._scoped_config.get('endpoint_url')
+
+    def _get_endpoint_url_environment_service(self):
+        return EnvironmentProvider(
+            name=self._get_service_env_var_name(), env=self._environ
+        ).provide()
+
+    def _get_endpoint_url_environment_global(self):
+        return EnvironmentProvider(
+            name='AWS_ENDPOINT_URL', env=self._environ
+        ).provide()