D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
imh-python
/
lib
/
python3.9
/
site-packages
/
botocore
/
Filename :
args.py
back
Copy
# Copyright 2016 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. """Internal module to help with normalizing botocore client args. This module (and all function/classes within this module) should be considered internal, and *not* a public API. """ import copy import logging import socket import botocore.exceptions import botocore.serialize import botocore.utils from botocore.signers import RequestSigner from botocore.config import Config from botocore.endpoint import EndpointCreator logger = logging.getLogger(__name__) VALID_REGIONAL_ENDPOINTS_CONFIG = [ 'legacy', 'regional', ] LEGACY_GLOBAL_STS_REGIONS = [ 'ap-northeast-1', 'ap-south-1', 'ap-southeast-1', 'ap-southeast-2', 'aws-global', 'ca-central-1', 'eu-central-1', 'eu-north-1', 'eu-west-1', 'eu-west-2', 'eu-west-3', 'sa-east-1', 'us-east-1', 'us-east-2', 'us-west-1', 'us-west-2', ] class ClientArgsCreator(object): def __init__(self, event_emitter, user_agent, response_parser_factory, loader, exceptions_factory, config_store): self._event_emitter = event_emitter self._user_agent = user_agent self._response_parser_factory = response_parser_factory self._loader = loader self._exceptions_factory = exceptions_factory self._config_store = config_store def get_client_args(self, service_model, region_name, is_secure, endpoint_url, verify, credentials, scoped_config, client_config, endpoint_bridge): final_args = self.compute_client_args( service_model, client_config, endpoint_bridge, region_name, endpoint_url, is_secure, scoped_config) service_name = final_args['service_name'] parameter_validation = final_args['parameter_validation'] endpoint_config = final_args['endpoint_config'] protocol = final_args['protocol'] config_kwargs = final_args['config_kwargs'] s3_config = final_args['s3_config'] partition = endpoint_config['metadata'].get('partition', None) socket_options = final_args['socket_options'] signing_region = endpoint_config['signing_region'] endpoint_region_name = endpoint_config['region_name'] event_emitter = copy.copy(self._event_emitter) signer = RequestSigner( service_model.service_id, signing_region, endpoint_config['signing_name'], endpoint_config['signature_version'], credentials, event_emitter ) config_kwargs['s3'] = s3_config new_config = Config(**config_kwargs) endpoint_creator = EndpointCreator(event_emitter) endpoint = endpoint_creator.create_endpoint( service_model, region_name=endpoint_region_name, endpoint_url=endpoint_config['endpoint_url'], verify=verify, response_parser_factory=self._response_parser_factory, max_pool_connections=new_config.max_pool_connections, proxies=new_config.proxies, timeout=(new_config.connect_timeout, new_config.read_timeout), socket_options=socket_options, client_cert=new_config.client_cert, proxies_config=new_config.proxies_config) serializer = botocore.serialize.create_serializer( protocol, parameter_validation) response_parser = botocore.parsers.create_parser(protocol) return { 'serializer': serializer, 'endpoint': endpoint, 'response_parser': response_parser, 'event_emitter': event_emitter, 'request_signer': signer, 'service_model': service_model, 'loader': self._loader, 'client_config': new_config, 'partition': partition, 'exceptions_factory': self._exceptions_factory } def compute_client_args(self, service_model, client_config, endpoint_bridge, region_name, endpoint_url, is_secure, scoped_config): service_name = service_model.endpoint_prefix protocol = service_model.metadata['protocol'] parameter_validation = True if client_config and not client_config.parameter_validation: parameter_validation = False elif scoped_config: raw_value = scoped_config.get('parameter_validation') if raw_value is not None: parameter_validation = botocore.utils.ensure_boolean(raw_value) # Override the user agent if specified in the client config. user_agent = self._user_agent if client_config is not None: if client_config.user_agent is not None: user_agent = client_config.user_agent if client_config.user_agent_extra is not None: user_agent += ' %s' % client_config.user_agent_extra s3_config = self.compute_s3_config(client_config) endpoint_config = self._compute_endpoint_config( service_name=service_name, region_name=region_name, endpoint_url=endpoint_url, is_secure=is_secure, endpoint_bridge=endpoint_bridge, s3_config=s3_config, ) # Create a new client config to be passed to the client based # on the final values. We do not want the user to be able # to try to modify an existing client with a client config. config_kwargs = dict( region_name=endpoint_config['region_name'], signature_version=endpoint_config['signature_version'], user_agent=user_agent) if client_config is not None: config_kwargs.update( connect_timeout=client_config.connect_timeout, read_timeout=client_config.read_timeout, max_pool_connections=client_config.max_pool_connections, proxies=client_config.proxies, proxies_config=client_config.proxies_config, retries=client_config.retries, client_cert=client_config.client_cert, inject_host_prefix=client_config.inject_host_prefix, ) self._compute_retry_config(config_kwargs) s3_config = self.compute_s3_config(client_config) return { 'service_name': service_name, 'parameter_validation': parameter_validation, 'user_agent': user_agent, 'endpoint_config': endpoint_config, 'protocol': protocol, 'config_kwargs': config_kwargs, 's3_config': s3_config, 'socket_options': self._compute_socket_options(scoped_config) } def compute_s3_config(self, client_config): s3_configuration = self._config_store.get_config_variable('s3') # Next specific client config values takes precedence over # specific values in the scoped config. if client_config is not None: if client_config.s3 is not None: if s3_configuration is None: s3_configuration = client_config.s3 else: # The current s3_configuration dictionary may be # from a source that only should be read from so # we want to be safe and just make a copy of it to modify # before it actually gets updated. s3_configuration = s3_configuration.copy() s3_configuration.update(client_config.s3) return s3_configuration def _compute_endpoint_config(self, service_name, region_name, endpoint_url, is_secure, endpoint_bridge, s3_config): resolve_endpoint_kwargs = { 'service_name': service_name, 'region_name': region_name, 'endpoint_url': endpoint_url, 'is_secure': is_secure, 'endpoint_bridge': endpoint_bridge, } if service_name == 's3': return self._compute_s3_endpoint_config( s3_config=s3_config, **resolve_endpoint_kwargs) if service_name == 'sts': return self._compute_sts_endpoint_config(**resolve_endpoint_kwargs) return self._resolve_endpoint(**resolve_endpoint_kwargs) def _compute_s3_endpoint_config(self, s3_config, **resolve_endpoint_kwargs): force_s3_global = self._should_force_s3_global( resolve_endpoint_kwargs['region_name'], s3_config) if force_s3_global: resolve_endpoint_kwargs['region_name'] = None endpoint_config = self._resolve_endpoint(**resolve_endpoint_kwargs) self._set_region_if_custom_s3_endpoint( endpoint_config, resolve_endpoint_kwargs['endpoint_bridge']) # For backwards compatibility reasons, we want to make sure the # client.meta.region_name will remain us-east-1 if we forced the # endpoint to be the global region. Specifically, if this value # changes to aws-global, it breaks logic where a user is checking # for us-east-1 as the global endpoint such as in creating buckets. if force_s3_global and endpoint_config['region_name'] == 'aws-global': endpoint_config['region_name'] = 'us-east-1' return endpoint_config def _should_force_s3_global(self, region_name, s3_config): s3_regional_config = 'legacy' if s3_config and 'us_east_1_regional_endpoint' in s3_config: s3_regional_config = s3_config['us_east_1_regional_endpoint'] self._validate_s3_regional_config(s3_regional_config) return ( s3_regional_config == 'legacy' and region_name in ['us-east-1', None] ) def _validate_s3_regional_config(self, config_val): if config_val not in VALID_REGIONAL_ENDPOINTS_CONFIG: raise botocore.exceptions.\ InvalidS3UsEast1RegionalEndpointConfigError( s3_us_east_1_regional_endpoint_config=config_val) def _set_region_if_custom_s3_endpoint(self, endpoint_config, endpoint_bridge): # If a user is providing a custom URL, the endpoint resolver will # refuse to infer a signing region. If we want to default to s3v4, # we have to account for this. if endpoint_config['signing_region'] is None \ and endpoint_config['region_name'] is None: endpoint = endpoint_bridge.resolve('s3') endpoint_config['signing_region'] = endpoint['signing_region'] endpoint_config['region_name'] = endpoint['region_name'] def _compute_sts_endpoint_config(self, **resolve_endpoint_kwargs): endpoint_config = self._resolve_endpoint(**resolve_endpoint_kwargs) if self._should_set_global_sts_endpoint( resolve_endpoint_kwargs['region_name'], resolve_endpoint_kwargs['endpoint_url']): self._set_global_sts_endpoint( endpoint_config, resolve_endpoint_kwargs['is_secure']) return endpoint_config def _should_set_global_sts_endpoint(self, region_name, endpoint_url): if endpoint_url: return False return ( self._get_sts_regional_endpoints_config() == 'legacy' and region_name in LEGACY_GLOBAL_STS_REGIONS ) def _get_sts_regional_endpoints_config(self): sts_regional_endpoints_config = self._config_store.get_config_variable( 'sts_regional_endpoints') if not sts_regional_endpoints_config: sts_regional_endpoints_config = 'legacy' if sts_regional_endpoints_config not in \ VALID_REGIONAL_ENDPOINTS_CONFIG: raise botocore.exceptions.InvalidSTSRegionalEndpointsConfigError( sts_regional_endpoints_config=sts_regional_endpoints_config) return sts_regional_endpoints_config def _set_global_sts_endpoint(self, endpoint_config, is_secure): scheme = 'https' if is_secure else 'http' endpoint_config['endpoint_url'] = '%s://sts.amazonaws.com' % scheme endpoint_config['signing_region'] = 'us-east-1' def _resolve_endpoint(self, service_name, region_name, endpoint_url, is_secure, endpoint_bridge): return endpoint_bridge.resolve( service_name, region_name, endpoint_url, is_secure) def _compute_socket_options(self, scoped_config): # This disables Nagle's algorithm and is the default socket options # in urllib3. socket_options = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)] if scoped_config: # Enables TCP Keepalive if specified in shared config file. if self._ensure_boolean(scoped_config.get('tcp_keepalive', False)): socket_options.append( (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)) return socket_options def _compute_retry_config(self, config_kwargs): self._compute_retry_max_attempts(config_kwargs) self._compute_retry_mode(config_kwargs) def _compute_retry_max_attempts(self, config_kwargs): # There's a pre-existing max_attempts client config value that actually # means max *retry* attempts. There's also a `max_attempts` we pull # from the config store that means *total attempts*, which includes the # intitial request. We can't change what `max_attempts` means in # client config so we try to normalize everything to a new # "total_max_attempts" variable. We ensure that after this, the only # configuration for "max attempts" is the 'total_max_attempts' key. # An explicitly provided max_attempts in the client config # overrides everything. retries = config_kwargs.get('retries') if retries is not None: if 'total_max_attempts' in retries: retries.pop('max_attempts', None) return if 'max_attempts' in retries: value = retries.pop('max_attempts') # client config max_attempts means total retries so we # have to add one for 'total_max_attempts' to account # for the initial request. retries['total_max_attempts'] = value + 1 return # Otherwise we'll check the config store which checks env vars, # config files, etc. There is no default value for max_attempts # so if this returns None and we don't set a default value here. max_attempts = self._config_store.get_config_variable('max_attempts') if max_attempts is not None: if retries is None: retries = {} config_kwargs['retries'] = retries retries['total_max_attempts'] = max_attempts def _compute_retry_mode(self, config_kwargs): retries = config_kwargs.get('retries') if retries is None: retries = {} config_kwargs['retries'] = retries elif 'mode' in retries: # If there's a retry mode explicitly set in the client config # that overrides everything. return retry_mode = self._config_store.get_config_variable('retry_mode') if retry_mode is None: retry_mode = 'legacy' retries['mode'] = retry_mode def _ensure_boolean(self, val): if isinstance(val, bool): return val else: return val.lower() == 'true'