Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/botocore/client.py: 20%

1# Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"). You 

4# may not use this file except in compliance with the License. A copy of 

5# the License is located at 

6# 

7# http://aws.amazon.com/apache2.0/ 

8# 

9# or in the "license" file accompanying this file. This file is 

10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF 

11# ANY KIND, either express or implied. See the License for the specific 

12# language governing permissions and limitations under the License. 

13import logging 

14 

15from botocore import ( 

16 UNSIGNED, # noqa: F401 

17 waiter, 

18 xform_name, 

19) 

20from botocore.args import ClientArgsCreator 

21from botocore.auth import AUTH_TYPE_MAPS, resolve_auth_type 

22from botocore.awsrequest import prepare_request_dict 

23from botocore.compress import maybe_compress_request 

24from botocore.config import Config 

25from botocore.context import with_current_context 

26from botocore.credentials import RefreshableCredentials 

27from botocore.discovery import ( 

28 EndpointDiscoveryHandler, 

29 EndpointDiscoveryManager, 

30 block_endpoint_discovery_required_operations, 

31) 

32from botocore.docs.docstring import ClientMethodDocstring, PaginatorDocstring 

33from botocore.exceptions import ( 

34 ClientError, # noqa: F401 

35 DataNotFoundError, 

36 InvalidEndpointDiscoveryConfigurationError, 

37 OperationNotPageableError, 

38 UnknownServiceError, 

39 UnknownSignatureVersionError, 

40) 

41from botocore.history import get_global_history_recorder 

42from botocore.hooks import first_non_none_response 

43from botocore.httpchecksum import ( 

44 apply_request_checksum, 

45 resolve_checksum_context, 

46) 

47from botocore.model import ServiceModel 

48from botocore.paginate import Paginator 

49from botocore.retries import adaptive, standard 

50from botocore.useragent import UserAgentString, register_feature_id 

51from botocore.utils import ( 

52 CachedProperty, 

53 EventbridgeSignerSetter, 

54 S3ArnParamHandler, # noqa: F401 

55 S3ControlArnParamHandler, # noqa: F401 

56 S3ControlArnParamHandlerv2, 

57 S3ControlEndpointSetter, # noqa: F401 

58 S3EndpointSetter, # noqa: F401 

59 S3ExpressIdentityResolver, 

60 S3RegionRedirector, # noqa: F401 

61 S3RegionRedirectorv2, 

62 ensure_boolean, 

63 get_service_module_name, 

64) 

65 

66logger = logging.getLogger(__name__) 

67history_recorder = get_global_history_recorder() 

68 

69 

70class ClientCreator: 

71 """Creates client objects for a service.""" 

72 

73 def __init__( 

74 self, 

75 loader, 

76 endpoint_resolver, 

77 user_agent, 

78 event_emitter, 

79 retry_handler_factory, 

80 retry_config_translator, 

81 response_parser_factory=None, 

82 exceptions_factory=None, 

83 config_store=None, 

84 user_agent_creator=None, 

85 auth_token_resolver=None, 

86 ): 

87 self._loader = loader 

88 self._endpoint_resolver = endpoint_resolver 

89 self._user_agent = user_agent 

90 self._event_emitter = event_emitter 

91 self._retry_handler_factory = retry_handler_factory 

92 self._retry_config_translator = retry_config_translator 

93 self._response_parser_factory = response_parser_factory 

94 self._exceptions_factory = exceptions_factory 

95 # TODO: Migrate things away from scoped_config in favor of the 

96 # config_store. The config store can pull things from both the scoped 

97 # config and environment variables (and potentially more in the 

98 # future). 

99 self._config_store = config_store 

100 self._user_agent_creator = user_agent_creator 

101 self._auth_token_resolver = auth_token_resolver 

102 

103 def create_client( 

104 self, 

105 service_name, 

106 region_name, 

107 is_secure=True, 

108 endpoint_url=None, 

109 verify=None, 

110 credentials=None, 

111 scoped_config=None, 

112 api_version=None, 

113 client_config=None, 

114 auth_token=None, 

115 ): 

116 responses = self._event_emitter.emit( 

117 'choose-service-name', service_name=service_name 

118 ) 

119 service_name = first_non_none_response(responses, default=service_name) 

120 service_model = self._load_service_model(service_name, api_version) 

121 try: 

122 endpoints_ruleset_data = self._load_service_endpoints_ruleset( 

123 service_name, api_version 

124 ) 

125 partition_data = self._loader.load_data('partitions') 

126 except UnknownServiceError: 

127 endpoints_ruleset_data = None 

128 partition_data = None 

129 logger.info( 

130 'No endpoints ruleset found for service %s, falling back to ' 

131 'legacy endpoint routing.', 

132 service_name, 

133 ) 

134 

135 cls = self._create_client_class(service_name, service_model) 

136 region_name, client_config = self._normalize_fips_region( 

137 region_name, client_config 

138 ) 

139 if auth := service_model.metadata.get('auth'): 

140 service_signature_version = resolve_auth_type(auth) 

141 else: 

142 service_signature_version = service_model.metadata.get( 

143 'signatureVersion' 

144 ) 

145 endpoint_bridge = ClientEndpointBridge( 

146 self._endpoint_resolver, 

147 scoped_config, 

148 client_config, 

149 service_signing_name=service_model.metadata.get('signingName'), 

150 config_store=self._config_store, 

151 service_signature_version=service_signature_version, 

152 ) 

153 if token := self._evaluate_client_specific_token( 

154 service_model.signing_name 

155 ): 

156 auth_token = token 

157 client_args = self._get_client_args( 

158 service_model, 

159 region_name, 

160 is_secure, 

161 endpoint_url, 

162 verify, 

163 credentials, 

164 scoped_config, 

165 client_config, 

166 endpoint_bridge, 

167 auth_token, 

168 endpoints_ruleset_data, 

169 partition_data, 

170 ) 

171 service_client = cls(**client_args) 

172 self._register_retries(service_client) 

173 self._register_s3_events( 

174 client=service_client, 

175 endpoint_bridge=None, 

176 endpoint_url=None, 

177 client_config=client_config, 

178 scoped_config=scoped_config, 

179 ) 

180 self._register_s3express_events(client=service_client) 

181 self._register_s3_control_events(client=service_client) 

182 self._register_importexport_events(client=service_client) 

183 self._register_endpoint_discovery( 

184 service_client, endpoint_url, client_config 

185 ) 

186 return service_client 

187 

188 def create_client_class(self, service_name, api_version=None): 

189 service_model = self._load_service_model(service_name, api_version) 

190 return self._create_client_class(service_name, service_model) 

191 

192 def _create_client_class(self, service_name, service_model): 

193 class_attributes = self._create_methods(service_model) 

194 py_name_to_operation_name = self._create_name_mapping(service_model) 

195 class_attributes['_PY_TO_OP_NAME'] = py_name_to_operation_name 

196 bases = [BaseClient] 

197 service_id = service_model.service_id.hyphenize() 

198 self._event_emitter.emit( 

199 f'creating-client-class.{service_id}', 

200 class_attributes=class_attributes, 

201 base_classes=bases, 

202 ) 

203 class_name = get_service_module_name(service_model) 

204 cls = type(str(class_name), tuple(bases), class_attributes) 

205 return cls 

206 

207 def _normalize_fips_region(self, region_name, client_config): 

208 if region_name is not None: 

209 normalized_region_name = region_name.replace('fips-', '').replace( 

210 '-fips', '' 

211 ) 

212 # If region has been transformed then set flag 

213 if normalized_region_name != region_name: 

214 config_use_fips_endpoint = Config(use_fips_endpoint=True) 

215 if client_config: 

216 # Keeping endpoint setting client specific 

217 client_config = client_config.merge( 

218 config_use_fips_endpoint 

219 ) 

220 else: 

221 client_config = config_use_fips_endpoint 

222 logger.warning( 

223 'transforming region from %s to %s and setting ' 

224 'use_fips_endpoint to true. client should not ' 

225 'be configured with a fips psuedo region.', 

226 region_name, 

227 normalized_region_name, 

228 ) 

229 region_name = normalized_region_name 

230 return region_name, client_config 

231 

232 def _load_service_model(self, service_name, api_version=None): 

233 json_model = self._loader.load_service_model( 

234 service_name, 'service-2', api_version=api_version 

235 ) 

236 service_model = ServiceModel(json_model, service_name=service_name) 

237 return service_model 

238 

239 def _load_service_endpoints_ruleset(self, service_name, api_version=None): 

240 return self._loader.load_service_model( 

241 service_name, 'endpoint-rule-set-1', api_version=api_version 

242 ) 

243 

244 def _register_retries(self, client): 

245 retry_mode = client.meta.config.retries['mode'] 

246 if retry_mode == 'standard': 

247 self._register_v2_standard_retries(client) 

248 elif retry_mode == 'adaptive': 

249 self._register_v2_standard_retries(client) 

250 self._register_v2_adaptive_retries(client) 

251 elif retry_mode == 'legacy': 

252 self._register_legacy_retries(client) 

253 else: 

254 return 

255 register_feature_id(f'RETRY_MODE_{retry_mode.upper()}') 

256 

257 def _register_v2_standard_retries(self, client): 

258 max_attempts = client.meta.config.retries.get('total_max_attempts') 

259 kwargs = {'client': client} 

260 if max_attempts is not None: 

261 kwargs['max_attempts'] = max_attempts 

262 standard.register_retry_handler(**kwargs) 

263 

264 def _register_v2_adaptive_retries(self, client): 

265 adaptive.register_retry_handler(client) 

266 

267 def _register_legacy_retries(self, client): 

268 endpoint_prefix = client.meta.service_model.endpoint_prefix 

269 service_id = client.meta.service_model.service_id 

270 service_event_name = service_id.hyphenize() 

271 

272 # First, we load the entire retry config for all services, 

273 # then pull out just the information we need. 

274 original_config = self._loader.load_data('_retry') 

275 if not original_config: 

276 return 

277 

278 retries = self._transform_legacy_retries(client.meta.config.retries) 

279 retry_config = self._retry_config_translator.build_retry_config( 

280 endpoint_prefix, 

281 original_config.get('retry', {}), 

282 original_config.get('definitions', {}), 

283 retries, 

284 ) 

285 

286 logger.debug( 

287 "Registering retry handlers for service: %s", 

288 client.meta.service_model.service_name, 

289 ) 

290 handler = self._retry_handler_factory.create_retry_handler( 

291 retry_config, endpoint_prefix 

292 ) 

293 unique_id = f'retry-config-{service_event_name}' 

294 client.meta.events.register( 

295 f"needs-retry.{service_event_name}", handler, unique_id=unique_id 

296 ) 

297 

298 def _transform_legacy_retries(self, retries): 

299 if retries is None: 

300 return 

301 copied_args = retries.copy() 

302 if 'total_max_attempts' in retries: 

303 copied_args = retries.copy() 

304 copied_args['max_attempts'] = ( 

305 copied_args.pop('total_max_attempts') - 1 

306 ) 

307 return copied_args 

308 

309 def _get_retry_mode(self, client, config_store): 

310 client_retries = client.meta.config.retries 

311 if ( 

312 client_retries is not None 

313 and client_retries.get('mode') is not None 

314 ): 

315 return client_retries['mode'] 

316 return config_store.get_config_variable('retry_mode') or 'legacy' 

317 

318 def _register_endpoint_discovery(self, client, endpoint_url, config): 

319 if endpoint_url is not None: 

320 # Don't register any handlers in the case of a custom endpoint url 

321 return 

322 # Only attach handlers if the service supports discovery 

323 if client.meta.service_model.endpoint_discovery_operation is None: 

324 return 

325 events = client.meta.events 

326 service_id = client.meta.service_model.service_id.hyphenize() 

327 enabled = False 

328 if config and config.endpoint_discovery_enabled is not None: 

329 enabled = config.endpoint_discovery_enabled 

330 elif self._config_store: 

331 enabled = self._config_store.get_config_variable( 

332 'endpoint_discovery_enabled' 

333 ) 

334 

335 enabled = self._normalize_endpoint_discovery_config(enabled) 

336 if enabled and self._requires_endpoint_discovery(client, enabled): 

337 discover = enabled is True 

338 manager = EndpointDiscoveryManager( 

339 client, always_discover=discover 

340 ) 

341 handler = EndpointDiscoveryHandler(manager) 

342 handler.register(events, service_id) 

343 else: 

344 events.register( 

345 'before-parameter-build', 

346 block_endpoint_discovery_required_operations, 

347 ) 

348 

349 def _normalize_endpoint_discovery_config(self, enabled): 

350 """Config must either be a boolean-string or string-literal 'auto'""" 

351 if isinstance(enabled, str): 

352 enabled = enabled.lower().strip() 

353 if enabled == 'auto': 

354 return enabled 

355 elif enabled in ('true', 'false'): 

356 return ensure_boolean(enabled) 

357 elif isinstance(enabled, bool): 

358 return enabled 

359 

360 raise InvalidEndpointDiscoveryConfigurationError(config_value=enabled) 

361 

362 def _requires_endpoint_discovery(self, client, enabled): 

363 if enabled == "auto": 

364 return client.meta.service_model.endpoint_discovery_required 

365 return enabled 

366 

367 def _register_eventbridge_events( 

368 self, client, endpoint_bridge, endpoint_url 

369 ): 

370 if client.meta.service_model.service_name != 'events': 

371 return 

372 EventbridgeSignerSetter( 

373 endpoint_resolver=self._endpoint_resolver, 

374 region=client.meta.region_name, 

375 endpoint_url=endpoint_url, 

376 ).register(client.meta.events) 

377 

378 def _register_s3express_events( 

379 self, 

380 client, 

381 endpoint_bridge=None, 

382 endpoint_url=None, 

383 client_config=None, 

384 scoped_config=None, 

385 ): 

386 if client.meta.service_model.service_name != 's3': 

387 return 

388 S3ExpressIdentityResolver(client, RefreshableCredentials).register() 

389 

390 def _register_s3_events( 

391 self, 

392 client, 

393 endpoint_bridge, 

394 endpoint_url, 

395 client_config, 

396 scoped_config, 

397 ): 

398 if client.meta.service_model.service_name != 's3': 

399 return 

400 S3RegionRedirectorv2(None, client).register() 

401 self._set_s3_presign_signature_version( 

402 client.meta, client_config, scoped_config 

403 ) 

404 client.meta.events.register( 

405 'before-parameter-build.s3', self._inject_s3_input_parameters 

406 ) 

407 

408 def _register_s3_control_events( 

409 self, 

410 client, 

411 endpoint_bridge=None, 

412 endpoint_url=None, 

413 client_config=None, 

414 scoped_config=None, 

415 ): 

416 if client.meta.service_model.service_name != 's3control': 

417 return 

418 S3ControlArnParamHandlerv2().register(client.meta.events) 

419 

420 def _set_s3_presign_signature_version( 

421 self, client_meta, client_config, scoped_config 

422 ): 

423 # This will return the manually configured signature version, or None 

424 # if none was manually set. If a customer manually sets the signature 

425 # version, we always want to use what they set. 

426 provided_signature_version = _get_configured_signature_version( 

427 's3', client_config, scoped_config 

428 ) 

429 if provided_signature_version is not None: 

430 return 

431 

432 # Check to see if the region is a region that we know about. If we 

433 # don't know about a region, then we can safely assume it's a new 

434 # region that is sigv4 only, since all new S3 regions only allow sigv4. 

435 # The only exception is aws-global. This is a pseudo-region for the 

436 # global endpoint, we should respect the signature versions it 

437 # supports, which includes v2. 

438 regions = self._endpoint_resolver.get_available_endpoints( 

439 's3', client_meta.partition 

440 ) 

441 if ( 

442 client_meta.region_name != 'aws-global' 

443 and client_meta.region_name not in regions 

444 ): 

445 return 

446 

447 # If it is a region we know about, we want to default to sigv2, so here 

448 # we check to see if it is available. 

449 endpoint = self._endpoint_resolver.construct_endpoint( 

450 's3', client_meta.region_name 

451 ) 

452 signature_versions = endpoint['signatureVersions'] 

453 if 's3' not in signature_versions: 

454 return 

455 

456 # We now know that we're in a known region that supports sigv2 and 

457 # the customer hasn't set a signature version so we default the 

458 # signature version to sigv2. 

459 client_meta.events.register( 

460 'choose-signer.s3', self._default_s3_presign_to_sigv2 

461 ) 

462 

463 def _inject_s3_input_parameters(self, params, context, **kwargs): 

464 context['input_params'] = {} 

465 inject_parameters = ('Bucket', 'Delete', 'Key', 'Prefix') 

466 for inject_parameter in inject_parameters: 

467 if inject_parameter in params: 

468 context['input_params'][inject_parameter] = params[ 

469 inject_parameter 

470 ] 

471 

472 def _default_s3_presign_to_sigv2(self, signature_version, **kwargs): 

473 """ 

474 Returns the 's3' (sigv2) signer if presigning an s3 request. This is 

475 intended to be used to set the default signature version for the signer 

476 to sigv2. Situations where an asymmetric signature is required are the 

477 exception, for example MRAP needs v4a. 

478 

479 :type signature_version: str 

480 :param signature_version: The current client signature version. 

481 

482 :type signing_name: str 

483 :param signing_name: The signing name of the service. 

484 

485 :return: 's3' if the request is an s3 presign request, None otherwise 

486 """ 

487 if signature_version.startswith('v4a'): 

488 return 

489 

490 if signature_version.startswith('v4-s3express'): 

491 return signature_version 

492 

493 for suffix in ['-query', '-presign-post']: 

494 if signature_version.endswith(suffix): 

495 return f's3{suffix}' 

496 

497 def _register_importexport_events( 

498 self, 

499 client, 

500 endpoint_bridge=None, 

501 endpoint_url=None, 

502 client_config=None, 

503 scoped_config=None, 

504 ): 

505 if client.meta.service_model.service_name != 'importexport': 

506 return 

507 self._set_importexport_signature_version( 

508 client.meta, client_config, scoped_config 

509 ) 

510 

511 def _set_importexport_signature_version( 

512 self, client_meta, client_config, scoped_config 

513 ): 

514 # This will return the manually configured signature version, or None 

515 # if none was manually set. If a customer manually sets the signature 

516 # version, we always want to use what they set. 

517 configured_signature_version = _get_configured_signature_version( 

518 'importexport', client_config, scoped_config 

519 ) 

520 if configured_signature_version is not None: 

521 return 

522 

523 # importexport has a modeled signatureVersion of v2, but we 

524 # previously switched to v4 via endpoint.json before endpoint rulesets. 

525 # Override the model's signatureVersion for backwards compatability. 

526 client_meta.events.register( 

527 'choose-signer.importexport', self._default_signer_to_sigv4 

528 ) 

529 

530 def _default_signer_to_sigv4(self, signature_version, **kwargs): 

531 return 'v4' 

532 

533 def _get_client_args( 

534 self, 

535 service_model, 

536 region_name, 

537 is_secure, 

538 endpoint_url, 

539 verify, 

540 credentials, 

541 scoped_config, 

542 client_config, 

543 endpoint_bridge, 

544 auth_token, 

545 endpoints_ruleset_data, 

546 partition_data, 

547 ): 

548 args_creator = ClientArgsCreator( 

549 self._event_emitter, 

550 self._user_agent, 

551 self._response_parser_factory, 

552 self._loader, 

553 self._exceptions_factory, 

554 config_store=self._config_store, 

555 user_agent_creator=self._user_agent_creator, 

556 ) 

557 return args_creator.get_client_args( 

558 service_model, 

559 region_name, 

560 is_secure, 

561 endpoint_url, 

562 verify, 

563 credentials, 

564 scoped_config, 

565 client_config, 

566 endpoint_bridge, 

567 auth_token, 

568 endpoints_ruleset_data, 

569 partition_data, 

570 ) 

571 

572 def _create_methods(self, service_model): 

573 op_dict = {} 

574 for operation_name in service_model.operation_names: 

575 py_operation_name = xform_name(operation_name) 

576 op_dict[py_operation_name] = self._create_api_method( 

577 py_operation_name, operation_name, service_model 

578 ) 

579 return op_dict 

580 

581 def _create_name_mapping(self, service_model): 

582 # py_name -> OperationName, for every operation available 

583 # for a service. 

584 mapping = {} 

585 for operation_name in service_model.operation_names: 

586 py_operation_name = xform_name(operation_name) 

587 mapping[py_operation_name] = operation_name 

588 return mapping 

589 

590 def _create_api_method( 

591 self, py_operation_name, operation_name, service_model 

592 ): 

593 def _api_call(self, *args, **kwargs): 

594 # We're accepting *args so that we can give a more helpful 

595 # error message than TypeError: _api_call takes exactly 

596 # 1 argument. 

597 if args: 

598 raise TypeError( 

599 f"{py_operation_name}() only accepts keyword arguments." 

600 ) 

601 # The "self" in this scope is referring to the BaseClient. 

602 return self._make_api_call(operation_name, kwargs) 

603 

604 _api_call.__name__ = str(py_operation_name) 

605 

606 # Add the docstring to the client method 

607 operation_model = service_model.operation_model(operation_name) 

608 docstring = ClientMethodDocstring( 

609 operation_model=operation_model, 

610 method_name=operation_name, 

611 event_emitter=self._event_emitter, 

612 method_description=operation_model.documentation, 

613 example_prefix=f'response = client.{py_operation_name}', 

614 include_signature=False, 

615 ) 

616 _api_call.__doc__ = docstring 

617 return _api_call 

618 

619 def _evaluate_client_specific_token(self, signing_name): 

620 # Resolves an auth_token for the given signing_name. 

621 # Returns None if no resolver is set or if resolution fails. 

622 resolver = self._auth_token_resolver 

623 if not resolver or not signing_name: 

624 return None 

625 

626 return resolver(signing_name=signing_name) 

627 

628 

629class ClientEndpointBridge: 

630 """Bridges endpoint data and client creation 

631 

632 This class handles taking out the relevant arguments from the endpoint 

633 resolver and determining which values to use, taking into account any 

634 client configuration options and scope configuration options. 

635 

636 This class also handles determining what, if any, region to use if no 

637 explicit region setting is provided. For example, Amazon S3 client will 

638 utilize "us-east-1" by default if no region can be resolved.""" 

639 

640 DEFAULT_ENDPOINT = '{service}.{region}.amazonaws.com' 

641 _DUALSTACK_CUSTOMIZED_SERVICES = ['s3', 's3-control'] 

642 

643 def __init__( 

644 self, 

645 endpoint_resolver, 

646 scoped_config=None, 

647 client_config=None, 

648 default_endpoint=None, 

649 service_signing_name=None, 

650 config_store=None, 

651 service_signature_version=None, 

652 ): 

653 self.service_signing_name = service_signing_name 

654 self.endpoint_resolver = endpoint_resolver 

655 self.scoped_config = scoped_config 

656 self.client_config = client_config 

657 self.default_endpoint = default_endpoint or self.DEFAULT_ENDPOINT 

658 self.config_store = config_store 

659 self.service_signature_version = service_signature_version 

660 

661 def resolve( 

662 self, service_name, region_name=None, endpoint_url=None, is_secure=True 

663 ): 

664 region_name = self._check_default_region(service_name, region_name) 

665 use_dualstack_endpoint = self._resolve_use_dualstack_endpoint( 

666 service_name 

667 ) 

668 use_fips_endpoint = self._resolve_endpoint_variant_config_var( 

669 'use_fips_endpoint' 

670 ) 

671 resolved = self.endpoint_resolver.construct_endpoint( 

672 service_name, 

673 region_name, 

674 use_dualstack_endpoint=use_dualstack_endpoint, 

675 use_fips_endpoint=use_fips_endpoint, 

676 ) 

677 

678 # If we can't resolve the region, we'll attempt to get a global 

679 # endpoint for non-regionalized services (iam, route53, etc) 

680 if not resolved: 

681 # TODO: fallback partition_name should be configurable in the 

682 # future for users to define as needed. 

683 resolved = self.endpoint_resolver.construct_endpoint( 

684 service_name, 

685 region_name, 

686 partition_name='aws', 

687 use_dualstack_endpoint=use_dualstack_endpoint, 

688 use_fips_endpoint=use_fips_endpoint, 

689 ) 

690 

691 if resolved: 

692 return self._create_endpoint( 

693 resolved, service_name, region_name, endpoint_url, is_secure 

694 ) 

695 else: 

696 return self._assume_endpoint( 

697 service_name, region_name, endpoint_url, is_secure 

698 ) 

699 

700 def resolver_uses_builtin_data(self): 

701 return self.endpoint_resolver.uses_builtin_data 

702 

703 def _check_default_region(self, service_name, region_name): 

704 if region_name is not None: 

705 return region_name 

706 # Use the client_config region if no explicit region was provided. 

707 if self.client_config and self.client_config.region_name is not None: 

708 return self.client_config.region_name 

709 

710 def _create_endpoint( 

711 self, resolved, service_name, region_name, endpoint_url, is_secure 

712 ): 

713 region_name, signing_region = self._pick_region_values( 

714 resolved, region_name, endpoint_url 

715 ) 

716 if endpoint_url is None: 

717 endpoint_url = self._make_url( 

718 resolved.get('hostname'), 

719 is_secure, 

720 resolved.get('protocols', []), 

721 ) 

722 signature_version = self._resolve_signature_version( 

723 service_name, resolved 

724 ) 

725 signing_name = self._resolve_signing_name(service_name, resolved) 

726 return self._create_result( 

727 service_name=service_name, 

728 region_name=region_name, 

729 signing_region=signing_region, 

730 signing_name=signing_name, 

731 endpoint_url=endpoint_url, 

732 metadata=resolved, 

733 signature_version=signature_version, 

734 ) 

735 

736 def _resolve_endpoint_variant_config_var(self, config_var): 

737 client_config = self.client_config 

738 config_val = False 

739 

740 # Client configuration arg has precedence 

741 if client_config and getattr(client_config, config_var) is not None: 

742 return getattr(client_config, config_var) 

743 elif self.config_store is not None: 

744 # Check config store 

745 config_val = self.config_store.get_config_variable(config_var) 

746 return config_val 

747 

748 def _resolve_use_dualstack_endpoint(self, service_name): 

749 s3_dualstack_mode = self._is_s3_dualstack_mode(service_name) 

750 if s3_dualstack_mode is not None: 

751 return s3_dualstack_mode 

752 return self._resolve_endpoint_variant_config_var( 

753 'use_dualstack_endpoint' 

754 ) 

755 

756 def _is_s3_dualstack_mode(self, service_name): 

757 if service_name not in self._DUALSTACK_CUSTOMIZED_SERVICES: 

758 return None 

759 # TODO: This normalization logic is duplicated from the 

760 # ClientArgsCreator class. Consolidate everything to 

761 # ClientArgsCreator. _resolve_signature_version also has similarly 

762 # duplicated logic. 

763 client_config = self.client_config 

764 if ( 

765 client_config is not None 

766 and client_config.s3 is not None 

767 and 'use_dualstack_endpoint' in client_config.s3 

768 ): 

769 # Client config trumps scoped config. 

770 return client_config.s3['use_dualstack_endpoint'] 

771 if self.scoped_config is not None: 

772 enabled = self.scoped_config.get('s3', {}).get( 

773 'use_dualstack_endpoint' 

774 ) 

775 if enabled in [True, 'True', 'true']: 

776 return True 

777 

778 def _assume_endpoint( 

779 self, service_name, region_name, endpoint_url, is_secure 

780 ): 

781 if endpoint_url is None: 

782 # Expand the default hostname URI template. 

783 hostname = self.default_endpoint.format( 

784 service=service_name, region=region_name 

785 ) 

786 endpoint_url = self._make_url( 

787 hostname, is_secure, ['http', 'https'] 

788 ) 

789 logger.debug( 

790 'Assuming an endpoint for %s, %s: %s', 

791 service_name, 

792 region_name, 

793 endpoint_url, 

794 ) 

795 # We still want to allow the user to provide an explicit version. 

796 signature_version = self._resolve_signature_version( 

797 service_name, {'signatureVersions': ['v4']} 

798 ) 

799 signing_name = self._resolve_signing_name(service_name, resolved={}) 

800 return self._create_result( 

801 service_name=service_name, 

802 region_name=region_name, 

803 signing_region=region_name, 

804 signing_name=signing_name, 

805 signature_version=signature_version, 

806 endpoint_url=endpoint_url, 

807 metadata={}, 

808 ) 

809 

810 def _create_result( 

811 self, 

812 service_name, 

813 region_name, 

814 signing_region, 

815 signing_name, 

816 endpoint_url, 

817 signature_version, 

818 metadata, 

819 ): 

820 return { 

821 'service_name': service_name, 

822 'region_name': region_name, 

823 'signing_region': signing_region, 

824 'signing_name': signing_name, 

825 'endpoint_url': endpoint_url, 

826 'signature_version': signature_version, 

827 'metadata': metadata, 

828 } 

829 

830 def _make_url(self, hostname, is_secure, supported_protocols): 

831 if is_secure and 'https' in supported_protocols: 

832 scheme = 'https' 

833 else: 

834 scheme = 'http' 

835 return f'{scheme}://{hostname}' 

836 

837 def _resolve_signing_name(self, service_name, resolved): 

838 # CredentialScope overrides everything else. 

839 if ( 

840 'credentialScope' in resolved 

841 and 'service' in resolved['credentialScope'] 

842 ): 

843 return resolved['credentialScope']['service'] 

844 # Use the signingName from the model if present. 

845 if self.service_signing_name: 

846 return self.service_signing_name 

847 # Just assume is the same as the service name. 

848 return service_name 

849 

850 def _pick_region_values(self, resolved, region_name, endpoint_url): 

851 signing_region = region_name 

852 if endpoint_url is None: 

853 # Do not use the region name or signing name from the resolved 

854 # endpoint if the user explicitly provides an endpoint_url. This 

855 # would happen if we resolve to an endpoint where the service has 

856 # a "defaults" section that overrides all endpoint with a single 

857 # hostname and credentialScope. This has been the case historically 

858 # for how STS has worked. The only way to resolve an STS endpoint 

859 # was to provide a region_name and an endpoint_url. In that case, 

860 # we would still resolve an endpoint, but we would not use the 

861 # resolved endpointName or signingRegion because we want to allow 

862 # custom endpoints. 

863 region_name = resolved['endpointName'] 

864 signing_region = region_name 

865 if ( 

866 'credentialScope' in resolved 

867 and 'region' in resolved['credentialScope'] 

868 ): 

869 signing_region = resolved['credentialScope']['region'] 

870 return region_name, signing_region 

871 

872 def _resolve_signature_version(self, service_name, resolved): 

873 configured_version = _get_configured_signature_version( 

874 service_name, self.client_config, self.scoped_config 

875 ) 

876 if configured_version is not None: 

877 return configured_version 

878 

879 # These have since added the "auth" key to the service model 

880 # with "aws.auth#sigv4", but preserve existing behavior from 

881 # when we preferred endpoints.json over the service models 

882 if service_name in ('s3', 's3-control'): 

883 return 's3v4' 

884 

885 if self.service_signature_version is not None: 

886 # Prefer the service model 

887 potential_versions = [self.service_signature_version] 

888 else: 

889 # Fall back to endpoints.json to preserve existing behavior, which 

890 # may be useful for users who have custom service models 

891 potential_versions = resolved.get('signatureVersions', []) 

892 # This was added for the V2 -> V4 transition, 

893 # for services that added V4 after V2 in endpoints.json 

894 if 'v4' in potential_versions: 

895 return 'v4' 

896 # Now just iterate over the signature versions in order until we 

897 # find the first one that is known to Botocore. 

898 for known in potential_versions: 

899 if known in AUTH_TYPE_MAPS: 

900 return known 

901 

902 raise UnknownSignatureVersionError( 

903 signature_version=potential_versions 

904 ) 

905 

906 

907class BaseClient: 

908 # This is actually reassigned with the py->op_name mapping 

909 # when the client creator creates the subclass. This value is used 

910 # because calls such as client.get_paginator('list_objects') use the 

911 # snake_case name, but we need to know the ListObjects form. 

912 # xform_name() does the ListObjects->list_objects conversion, but 

913 # we need the reverse mapping here. 

914 _PY_TO_OP_NAME = {} 

915 

916 def __init__( 

917 self, 

918 serializer, 

919 endpoint, 

920 response_parser,