Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/botocore/utils.py: 20%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2012-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 base64
14import binascii
15import datetime
16import email.message
17import functools
18import hashlib
19import io
20import logging
21import os
22import random
23import re
24import socket
25import tempfile
26import time
27import warnings
28import weakref
29from datetime import datetime as _DatetimeClass
30from ipaddress import ip_address
31from pathlib import Path
32from urllib.request import getproxies, proxy_bypass
34import dateutil.parser
35from dateutil.tz import tzutc
36from urllib3.exceptions import LocationParseError
38import botocore
39import botocore.awsrequest
40import botocore.httpsession
42# IP Regexes retained for backwards compatibility
43from botocore.compat import (
44 HAS_CRT,
45 HEX_PAT, # noqa: F401
46 IPV4_PAT, # noqa: F401
47 IPV4_RE,
48 IPV6_ADDRZ_PAT, # noqa: F401
49 IPV6_ADDRZ_RE,
50 IPV6_PAT, # noqa: F401
51 LS32_PAT, # noqa: F401
52 MD5_AVAILABLE,
53 UNRESERVED_PAT, # noqa: F401
54 UNSAFE_URL_CHARS,
55 ZONE_ID_PAT, # noqa: F401
56 OrderedDict,
57 get_current_datetime,
58 get_md5,
59 get_tzinfo_options,
60 json,
61 quote,
62 urlparse,
63 urlsplit,
64 urlunsplit,
65 zip_longest,
66)
67from botocore.exceptions import (
68 ClientError,
69 ConfigNotFound,
70 ConnectionClosedError,
71 ConnectTimeoutError,
72 EndpointConnectionError,
73 HTTPClientError,
74 InvalidDNSNameError,
75 InvalidEndpointConfigurationError,
76 InvalidExpressionError,
77 InvalidHostLabelError,
78 InvalidIMDSEndpointError,
79 InvalidIMDSEndpointModeError,
80 InvalidRegionError,
81 MetadataRetrievalError,
82 MissingDependencyException,
83 ReadTimeoutError,
84 SSOTokenLoadError,
85 UnsupportedOutpostResourceError,
86 UnsupportedS3AccesspointConfigurationError,
87 UnsupportedS3ArnError,
88 UnsupportedS3ConfigurationError,
89 UnsupportedS3ControlArnError,
90 UnsupportedS3ControlConfigurationError,
91)
92from botocore.plugin import (
93 PluginContext,
94 reset_plugin_context,
95 set_plugin_context,
96)
97from botocore.useragent import register_feature_id
99logger = logging.getLogger(__name__)
100DEFAULT_METADATA_SERVICE_TIMEOUT = 1
101METADATA_BASE_URL = 'http://169.254.169.254/'
102METADATA_BASE_URL_IPv6 = 'http://[fd00:ec2::254]/'
103METADATA_ENDPOINT_MODES = ('ipv4', 'ipv6')
105# These are chars that do not need to be urlencoded.
106# Based on rfc2986, section 2.3
107SAFE_CHARS = '-._~'
108LABEL_RE = re.compile(r'[a-z0-9][a-z0-9\-]*[a-z0-9]')
109RETRYABLE_HTTP_ERRORS = (
110 ReadTimeoutError,
111 EndpointConnectionError,
112 ConnectionClosedError,
113 ConnectTimeoutError,
114)
115S3_ACCELERATE_WHITELIST = ['dualstack']
116# In switching events from using service name / endpoint prefix to service
117# id, we have to preserve compatibility. This maps the instances where either
118# is different than the transformed service id.
119EVENT_ALIASES = {
120 "api.mediatailor": "mediatailor",
121 "api.pricing": "pricing",
122 "api.sagemaker": "sagemaker",
123 "apigateway": "api-gateway",
124 "application-autoscaling": "application-auto-scaling",
125 "appstream2": "appstream",
126 "autoscaling": "auto-scaling",
127 "autoscaling-plans": "auto-scaling-plans",
128 "ce": "cost-explorer",
129 "cloudhsmv2": "cloudhsm-v2",
130 "cloudsearchdomain": "cloudsearch-domain",
131 "cognito-idp": "cognito-identity-provider",
132 "config": "config-service",
133 "cur": "cost-and-usage-report-service",
134 "data.iot": "iot-data-plane",
135 "data.jobs.iot": "iot-jobs-data-plane",
136 "data.mediastore": "mediastore-data",
137 "datapipeline": "data-pipeline",
138 "devicefarm": "device-farm",
139 "directconnect": "direct-connect",
140 "discovery": "application-discovery-service",
141 "dms": "database-migration-service",
142 "ds": "directory-service",
143 "dynamodbstreams": "dynamodb-streams",
144 "elasticbeanstalk": "elastic-beanstalk",
145 "elasticfilesystem": "efs",
146 "elasticloadbalancing": "elastic-load-balancing",
147 "elasticmapreduce": "emr",
148 "elb": "elastic-load-balancing",
149 "elbv2": "elastic-load-balancing-v2",
150 "email": "ses",
151 "entitlement.marketplace": "marketplace-entitlement-service",
152 "es": "elasticsearch-service",
153 "events": "eventbridge",
154 "cloudwatch-events": "eventbridge",
155 "iot-data": "iot-data-plane",
156 "iot-jobs-data": "iot-jobs-data-plane",
157 "kinesisanalytics": "kinesis-analytics",
158 "kinesisvideo": "kinesis-video",
159 "lex-models": "lex-model-building-service",
160 "lex-runtime": "lex-runtime-service",
161 "logs": "cloudwatch-logs",
162 "machinelearning": "machine-learning",
163 "marketplace-entitlement": "marketplace-entitlement-service",
164 "marketplacecommerceanalytics": "marketplace-commerce-analytics",
165 "metering.marketplace": "marketplace-metering",
166 "meteringmarketplace": "marketplace-metering",
167 "mgh": "migration-hub",
168 "models.lex": "lex-model-building-service",
169 "monitoring": "cloudwatch",
170 "mturk-requester": "mturk",
171 "resourcegroupstaggingapi": "resource-groups-tagging-api",
172 "route53": "route-53",
173 "route53domains": "route-53-domains",
174 "runtime.lex": "lex-runtime-service",
175 "runtime.sagemaker": "sagemaker-runtime",
176 "sdb": "simpledb",
177 "secretsmanager": "secrets-manager",
178 "serverlessrepo": "serverlessapplicationrepository",
179 "servicecatalog": "service-catalog",
180 "states": "sfn",
181 "stepfunctions": "sfn",
182 "storagegateway": "storage-gateway",
183 "streams.dynamodb": "dynamodb-streams",
184 "tagging": "resource-groups-tagging-api",
185}
188# This pattern can be used to detect if a header is a flexible checksum header
189CHECKSUM_HEADER_PATTERN = re.compile(
190 r'^X-Amz-Checksum-([a-z0-9]*)$',
191 flags=re.IGNORECASE,
192)
194PRIORITY_ORDERED_SUPPORTED_PROTOCOLS = (
195 'json',
196 'rest-json',
197 'rest-xml',
198 'smithy-rpc-v2-cbor',
199 'query',
200 'ec2',
201)
204def ensure_boolean(val):
205 """Ensures a boolean value if a string or boolean is provided
207 For strings, the value for True/False is case insensitive
208 """
209 if isinstance(val, bool):
210 return val
211 elif isinstance(val, str):
212 return val.lower() == 'true'
213 else:
214 return False
217def resolve_imds_endpoint_mode(session):
218 """Resolving IMDS endpoint mode to either IPv6 or IPv4.
220 ec2_metadata_service_endpoint_mode takes precedence over imds_use_ipv6.
221 """
222 endpoint_mode = session.get_config_variable(
223 'ec2_metadata_service_endpoint_mode'
224 )
225 if endpoint_mode is not None:
226 lendpoint_mode = endpoint_mode.lower()
227 if lendpoint_mode not in METADATA_ENDPOINT_MODES:
228 error_msg_kwargs = {
229 'mode': endpoint_mode,
230 'valid_modes': METADATA_ENDPOINT_MODES,
231 }
232 raise InvalidIMDSEndpointModeError(**error_msg_kwargs)
233 return lendpoint_mode
234 elif session.get_config_variable('imds_use_ipv6'):
235 return 'ipv6'
236 return 'ipv4'
239def is_json_value_header(shape):
240 """Determines if the provided shape is the special header type jsonvalue.
242 :type shape: botocore.shape
243 :param shape: Shape to be inspected for the jsonvalue trait.
245 :return: True if this type is a jsonvalue, False otherwise
246 :rtype: Bool
247 """
248 return (
249 hasattr(shape, 'serialization')
250 and shape.serialization.get('jsonvalue', False)
251 and shape.serialization.get('location') == 'header'
252 and shape.type_name == 'string'
253 )
256def has_header(header_name, headers):
257 """Case-insensitive check for header key."""
258 if header_name is None:
259 return False
260 elif isinstance(headers, botocore.awsrequest.HeadersDict):
261 return header_name in headers
262 else:
263 return header_name.lower() in [key.lower() for key in headers.keys()]
266def get_service_module_name(service_model):
267 """Returns the module name for a service
269 This is the value used in both the documentation and client class name
270 """
271 name = service_model.metadata.get(
272 'serviceAbbreviation',
273 service_model.metadata.get(
274 'serviceFullName', service_model.service_name
275 ),
276 )
277 name = name.replace('Amazon', '')
278 name = name.replace('AWS', '')
279 name = re.sub(r'\W+', '', name)
280 return name
283def normalize_url_path(path):
284 if not path:
285 return '/'
286 return remove_dot_segments(path)
289def normalize_boolean(val):
290 """Returns None if val is None, otherwise ensure value
291 converted to boolean"""
292 if val is None:
293 return val
294 else:
295 return ensure_boolean(val)
298def remove_dot_segments(url):
299 # RFC 3986, section 5.2.4 "Remove Dot Segments"
300 # Also, AWS services require consecutive slashes to be removed,
301 # so that's done here as well
302 if not url:
303 return ''
304 input_url = url.split('/')
305 output_list = []
306 for x in input_url:
307 if x and x != '.':
308 if x == '..':
309 if output_list:
310 output_list.pop()
311 else:
312 output_list.append(x)
314 if url[0] == '/':
315 first = '/'
316 else:
317 first = ''
318 if url[-1] == '/' and output_list:
319 last = '/'
320 else:
321 last = ''
322 return first + '/'.join(output_list) + last
325def validate_jmespath_for_set(expression):
326 # Validates a limited jmespath expression to determine if we can set a
327 # value based on it. Only works with dotted paths.
328 if not expression or expression == '.':
329 raise InvalidExpressionError(expression=expression)
331 for invalid in ['[', ']', '*']:
332 if invalid in expression:
333 raise InvalidExpressionError(expression=expression)
336def set_value_from_jmespath(source, expression, value, is_first=True):
337 # This takes a (limited) jmespath-like expression & can set a value based
338 # on it.
339 # Limitations:
340 # * Only handles dotted lookups
341 # * No offsets/wildcards/slices/etc.
342 if is_first:
343 validate_jmespath_for_set(expression)
345 bits = expression.split('.', 1)
346 current_key, remainder = bits[0], bits[1] if len(bits) > 1 else ''
348 if not current_key:
349 raise InvalidExpressionError(expression=expression)
351 if remainder:
352 if current_key not in source:
353 # We've got something in the expression that's not present in the
354 # source (new key). If there's any more bits, we'll set the key
355 # with an empty dictionary.
356 source[current_key] = {}
358 return set_value_from_jmespath(
359 source[current_key], remainder, value, is_first=False
360 )
362 # If we're down to a single key, set it.
363 source[current_key] = value
366def is_global_accesspoint(context):
367 """Determine if request is intended for an MRAP accesspoint."""
368 s3_accesspoint = context.get('s3_accesspoint', {})
369 is_global = s3_accesspoint.get('region') == ''
370 return is_global
373def create_nested_client(session, service_name, **kwargs):
374 # If a client is created from within a plugin based on the environment variable,
375 # an infinite loop could arise. Any clients created from within another client
376 # must use this method to prevent infinite loops.
377 ctx = PluginContext(plugins="DISABLED")
378 token = set_plugin_context(ctx)
379 try:
380 return session.create_client(service_name, **kwargs)
381 finally:
382 reset_plugin_context(token)
385class _RetriesExceededError(Exception):
386 """Internal exception used when the number of retries are exceeded."""
388 pass
391class BadIMDSRequestError(Exception):
392 def __init__(self, request):
393 self.request = request
396class IMDSFetcher:
397 _RETRIES_EXCEEDED_ERROR_CLS = _RetriesExceededError
398 _TOKEN_PATH = 'latest/api/token'
399 _TOKEN_TTL = '21600'
401 def __init__(
402 self,
403 timeout=DEFAULT_METADATA_SERVICE_TIMEOUT,
404 num_attempts=1,
405 base_url=METADATA_BASE_URL,
406 env=None,
407 user_agent=None,
408 config=None,
409 ):
410 self._timeout = timeout
411 self._num_attempts = num_attempts
412 if config is None:
413 config = {}
414 self._base_url = self._select_base_url(base_url, config)
415 self._config = config
417 if env is None:
418 env = os.environ.copy()
419 self._disabled = (
420 env.get('AWS_EC2_METADATA_DISABLED', 'false').lower() == 'true'
421 )
422 self._imds_v1_disabled = config.get('ec2_metadata_v1_disabled')
423 self._user_agent = user_agent
424 self._session = botocore.httpsession.URLLib3Session(
425 timeout=self._timeout,
426 proxies=get_environ_proxies(self._base_url),
427 )
429 def get_base_url(self):
430 return self._base_url
432 def _select_base_url(self, base_url, config):
433 if config is None:
434 config = {}
436 requires_ipv6 = (
437 config.get('ec2_metadata_service_endpoint_mode') == 'ipv6'
438 )
439 custom_metadata_endpoint = config.get('ec2_metadata_service_endpoint')
441 if requires_ipv6 and custom_metadata_endpoint:
442 logger.warning(
443 "Custom endpoint and IMDS_USE_IPV6 are both set. Using custom endpoint."
444 )
446 chosen_base_url = None
448 if base_url != METADATA_BASE_URL:
449 chosen_base_url = base_url
450 elif custom_metadata_endpoint:
451 chosen_base_url = custom_metadata_endpoint
452 elif requires_ipv6:
453 chosen_base_url = METADATA_BASE_URL_IPv6
454 else:
455 chosen_base_url = METADATA_BASE_URL
457 logger.debug("IMDS ENDPOINT: %s", chosen_base_url)
458 if not is_valid_uri(chosen_base_url):
459 raise InvalidIMDSEndpointError(endpoint=chosen_base_url)
461 return chosen_base_url
463 def _construct_url(self, path):
464 sep = ''
465 if self._base_url and not self._base_url.endswith('/'):
466 sep = '/'
467 return f'{self._base_url}{sep}{path}'
469 def _fetch_metadata_token(self):
470 self._assert_enabled()
471 url = self._construct_url(self._TOKEN_PATH)
472 headers = {
473 'x-aws-ec2-metadata-token-ttl-seconds': self._TOKEN_TTL,
474 }
475 self._add_user_agent(headers)
476 request = botocore.awsrequest.AWSRequest(
477 method='PUT', url=url, headers=headers
478 )
479 for i in range(self._num_attempts):
480 try:
481 response = self._session.send(request.prepare())
482 if response.status_code == 200:
483 return response.text
484 elif response.status_code in (404, 403, 405):
485 return None
486 elif response.status_code in (400,):
487 raise BadIMDSRequestError(request)
488 except ReadTimeoutError:
489 return None
490 except RETRYABLE_HTTP_ERRORS as e:
491 logger.debug(
492 "Caught retryable HTTP exception while making metadata "
493 "service request to %s: %s",
494 url,
495 e,
496 exc_info=True,
497 )
498 except HTTPClientError as e:
499 if isinstance(e.kwargs.get('error'), LocationParseError):
500 raise InvalidIMDSEndpointError(endpoint=url, error=e)
501 else:
502 raise
503 return None
505 def _get_request(self, url_path, retry_func, token=None):
506 """Make a get request to the Instance Metadata Service.
508 :type url_path: str
509 :param url_path: The path component of the URL to make a get request.
510 This arg is appended to the base_url that was provided in the
511 initializer.
513 :type retry_func: callable
514 :param retry_func: A function that takes the response as an argument
515 and determines if it needs to retry. By default empty and non
516 200 OK responses are retried.
518 :type token: str
519 :param token: Metadata token to send along with GET requests to IMDS.
520 """
521 self._assert_enabled()
522 if not token:
523 self._assert_v1_enabled()
524 if retry_func is None:
525 retry_func = self._default_retry
526 url = self._construct_url(url_path)
527 headers = {}
528 if token is not None:
529 headers['x-aws-ec2-metadata-token'] = token
530 self._add_user_agent(headers)
531 for i in range(self._num_attempts):
532 try:
533 request = botocore.awsrequest.AWSRequest(
534 method='GET', url=url, headers=headers
535 )
536 response = self._session.send(request.prepare())
537 if not retry_func(response):
538 return response
539 except RETRYABLE_HTTP_ERRORS as e:
540 logger.debug(
541 "Caught retryable HTTP exception while making metadata "
542 "service request to %s: %s",
543 url,
544 e,
545 exc_info=True,
546 )
547 raise self._RETRIES_EXCEEDED_ERROR_CLS()
549 def _add_user_agent(self, headers):
550 if self._user_agent is not None:
551 headers['User-Agent'] = self._user_agent
553 def _assert_enabled(self):
554 if self._disabled:
555 logger.debug("Access to EC2 metadata has been disabled.")
556 raise self._RETRIES_EXCEEDED_ERROR_CLS()
558 def _assert_v1_enabled(self):
559 if self._imds_v1_disabled:
560 raise MetadataRetrievalError(
561 error_msg="Unable to retrieve token for use in IMDSv2 call and IMDSv1 has been disabled"
562 )
564 def _default_retry(self, response):
565 return self._is_non_ok_response(response) or self._is_empty(response)
567 def _is_non_ok_response(self, response):
568 if response.status_code != 200:
569 self._log_imds_response(response, 'non-200', log_body=True)
570 return True
571 return False
573 def _is_empty(self, response):
574 if not response.content:
575 self._log_imds_response(response, 'no body', log_body=True)
576 return True
577 return False
579 def _log_imds_response(self, response, reason_to_log, log_body=False):
580 statement = (
581 "Metadata service returned %s response "
582 "with status code of %s for url: %s"
583 )
584 logger_args = [reason_to_log, response.status_code, response.url]
585 if log_body:
586 statement += ", content body: %s"
587 logger_args.append(response.content)
588 logger.debug(statement, *logger_args)
591class InstanceMetadataFetcher(IMDSFetcher):
592 _URL_PATH = 'latest/meta-data/iam/security-credentials/'
593 _REQUIRED_CREDENTIAL_FIELDS = [
594 'AccessKeyId',
595 'SecretAccessKey',
596 'Token',
597 'Expiration',
598 ]
600 def retrieve_iam_role_credentials(self):
601 try:
602 token = self._fetch_metadata_token()
603 role_name = self._get_iam_role(token)
604 credentials = self._get_credentials(role_name, token)
605 if self._contains_all_credential_fields(credentials):
606 credentials = {
607 'role_name': role_name,
608 'access_key': credentials['AccessKeyId'],
609 'secret_key': credentials['SecretAccessKey'],
610 'token': credentials['Token'],
611 'expiry_time': credentials['Expiration'],
612 }
613 self._evaluate_expiration(credentials)
614 return credentials
615 else:
616 # IMDS can return a 200 response that has a JSON formatted
617 # error message (i.e. if ec2 is not trusted entity for the
618 # attached role). We do not necessarily want to retry for
619 # these and we also do not necessarily want to raise a key
620 # error. So at least log the problematic response and return
621 # an empty dictionary to signal that it was not able to
622 # retrieve credentials. These error will contain both a
623 # Code and Message key.
624 if 'Code' in credentials and 'Message' in credentials:
625 logger.debug(
626 'Error response received when retrieving'
627 'credentials: %s.',
628 credentials,
629 )
630 return {}
631 except self._RETRIES_EXCEEDED_ERROR_CLS:
632 logger.debug(
633 "Max number of attempts exceeded (%s) when "
634 "attempting to retrieve data from metadata service.",
635 self._num_attempts,
636 )
637 except BadIMDSRequestError as e:
638 logger.debug("Bad IMDS request: %s", e.request)
639 return {}
641 def _get_iam_role(self, token=None):
642 return self._get_request(
643 url_path=self._URL_PATH,
644 retry_func=self._needs_retry_for_role_name,
645 token=token,
646 ).text
648 def _get_credentials(self, role_name, token=None):
649 r = self._get_request(
650 url_path=self._URL_PATH + role_name,
651 retry_func=self._needs_retry_for_credentials,
652 token=token,
653 )
654 return json.loads(r.text)
656 def _is_invalid_json(self, response):
657 try:
658 json.loads(response.text)
659 return False
660 except ValueError:
661 self._log_imds_response(response, 'invalid json')
662 return True
664 def _needs_retry_for_role_name(self, response):
665 return self._is_non_ok_response(response) or self._is_empty(response)
667 def _needs_retry_for_credentials(self, response):
668 return (
669 self._is_non_ok_response(response)
670 or self._is_empty(response)
671 or self._is_invalid_json(response)
672 )
674 def _contains_all_credential_fields(self, credentials):
675 for field in self._REQUIRED_CREDENTIAL_FIELDS:
676 if field not in credentials:
677 logger.debug(
678 'Retrieved credentials is missing required field: %s',
679 field,
680 )
681 return False
682 return True
684 def _evaluate_expiration(self, credentials):
685 expiration = credentials.get("expiry_time")
686 if expiration is None:
687 return
688 try:
689 expiration = datetime.datetime.strptime(
690 expiration, "%Y-%m-%dT%H:%M:%SZ"
691 )
692 refresh_interval = self._config.get(
693 "ec2_credential_refresh_window", 60 * 10
694 )
695 jitter = random.randint(120, 600) # Between 2 to 10 minutes
696 refresh_interval_with_jitter = refresh_interval + jitter
697 current_time = get_current_datetime()
698 refresh_offset = datetime.timedelta(
699 seconds=refresh_interval_with_jitter
700 )
701 extension_time = expiration - refresh_offset
702 if current_time >= extension_time:
703 new_time = current_time + refresh_offset
704 credentials["expiry_time"] = new_time.strftime(
705 "%Y-%m-%dT%H:%M:%SZ"
706 )
707 logger.info(
708 "Attempting credential expiration extension due to a "
709 "credential service availability issue. A refresh of "
710 "these credentials will be attempted again within "
711 "the next %.0f minutes.",
712 refresh_interval_with_jitter / 60,
713 )
714 except ValueError:
715 logger.debug(
716 "Unable to parse expiry_time in %s", credentials['expiry_time']
717 )
720class IMDSRegionProvider:
721 def __init__(self, session, environ=None, fetcher=None):
722 """Initialize IMDSRegionProvider.
723 :type session: :class:`botocore.session.Session`
724 :param session: The session is needed to look up configuration for
725 how to contact the instance metadata service. Specifically the
726 whether or not it should use the IMDS region at all, and if so how
727 to configure the timeout and number of attempts to reach the
728 service.
729 :type environ: None or dict
730 :param environ: A dictionary of environment variables to use. If
731 ``None`` is the argument then ``os.environ`` will be used by
732 default.
733 :type fecther: :class:`botocore.utils.InstanceMetadataRegionFetcher`
734 :param fetcher: The class to actually handle the fetching of the region
735 from the IMDS. If not provided a default one will be created.
736 """
737 self._session = session
738 if environ is None:
739 environ = os.environ
740 self._environ = environ
741 self._fetcher = fetcher
743 def provide(self):
744 """Provide the region value from IMDS."""
745 instance_region = self._get_instance_metadata_region()
746 return instance_region
748 def _get_instance_metadata_region(self):
749 fetcher = self._get_fetcher()
750 region = fetcher.retrieve_region()
751 return region
753 def _get_fetcher(self):
754 if self._fetcher is None:
755 self._fetcher = self._create_fetcher()
756 return self._fetcher
758 def _create_fetcher(self):
759 metadata_timeout = self._session.get_config_variable(
760 'metadata_service_timeout'
761 )
762 metadata_num_attempts = self._session.get_config_variable(
763 'metadata_service_num_attempts'
764 )
765 imds_config = {
766 'ec2_metadata_service_endpoint': self._session.get_config_variable(
767 'ec2_metadata_service_endpoint'
768 ),
769 'ec2_metadata_service_endpoint_mode': resolve_imds_endpoint_mode(
770 self._session
771 ),
772 'ec2_metadata_v1_disabled': self._session.get_config_variable(
773 'ec2_metadata_v1_disabled'
774 ),
775 }
776 fetcher = InstanceMetadataRegionFetcher(
777 timeout=metadata_timeout,
778 num_attempts=metadata_num_attempts,
779 env=self._environ,
780 user_agent=self._session.user_agent(),
781 config=imds_config,
782 )
783 return fetcher
786class InstanceMetadataRegionFetcher(IMDSFetcher):
787 _URL_PATH = 'latest/meta-data/placement/availability-zone/'
789 def retrieve_region(self):
790 """Get the current region from the instance metadata service.
791 :rvalue: str
792 :returns: The region the current instance is running in or None
793 if the instance metadata service cannot be contacted or does not
794 give a valid response.
795 :rtype: None or str
796 :returns: Returns the region as a string if it is configured to use
797 IMDS as a region source. Otherwise returns ``None``. It will also
798 return ``None`` if it fails to get the region from IMDS due to
799 exhausting its retries or not being able to connect.
800 """
801 try:
802 region = self._get_region()
803 return region
804 except self._RETRIES_EXCEEDED_ERROR_CLS:
805 logger.debug(
806 "Max number of attempts exceeded (%s) when "
807 "attempting to retrieve data from metadata service.",
808 self._num_attempts,
809 )
810 return None
812 def _get_region(self):
813 token = self._fetch_metadata_token()
814 response = self._get_request(
815 url_path=self._URL_PATH,
816 retry_func=self._default_retry,
817 token=token,
818 )
819 availability_zone = response.text
820 region = availability_zone[:-1]
821 return region
824def merge_dicts(dict1, dict2, append_lists=False):
825 """Given two dict, merge the second dict into the first.
827 The dicts can have arbitrary nesting.
829 :param append_lists: If true, instead of clobbering a list with the new
830 value, append all of the new values onto the original list.
831 """
832 for key in dict2:
833 if isinstance(dict2[key], dict):
834 if key in dict1 and key in dict2:
835 merge_dicts(dict1[key], dict2[key])
836 else:
837 dict1[key] = dict2[key]
838 # If the value is a list and the ``append_lists`` flag is set,
839 # append the new values onto the original list
840 elif isinstance(dict2[key], list) and append_lists:
841 # The value in dict1 must be a list in order to append new
842 # values onto it.
843 if key in dict1 and isinstance(dict1[key], list):
844 dict1[key].extend(dict2[key])
845 else:
846 dict1[key] = dict2[key]
847 else:
848 # At scalar types, we iterate and merge the
849 # current dict that we're on.
850 dict1[key] = dict2[key]
853def lowercase_dict(original):
854 """Copies the given dictionary ensuring all keys are lowercase strings."""
855 copy = {}
856 for key in original:
857 copy[key.lower()] = original[key]
858 return copy
861def parse_key_val_file(filename, _open=open):
862 try:
863 with _open(filename) as f:
864 contents = f.read()
865 return parse_key_val_file_contents(contents)
866 except OSError:
867 raise ConfigNotFound(path=filename)
870def parse_key_val_file_contents(contents):
871 # This was originally extracted from the EC2 credential provider, which was
872 # fairly lenient in its parsing. We only try to parse key/val pairs if
873 # there's a '=' in the line.
874 final = {}
875 for line in contents.splitlines():
876 if '=' not in line:
877 continue
878 key, val = line.split('=', 1)
879 key = key.strip()
880 val = val.strip()
881 final[key] = val
882 return final
885def percent_encode_sequence(mapping, safe=SAFE_CHARS):
886 """Urlencode a dict or list into a string.
888 This is similar to urllib.urlencode except that:
890 * It uses quote, and not quote_plus
891 * It has a default list of safe chars that don't need
892 to be encoded, which matches what AWS services expect.
894 If any value in the input ``mapping`` is a list type,
895 then each list element wil be serialized. This is the equivalent
896 to ``urlencode``'s ``doseq=True`` argument.
898 This function should be preferred over the stdlib
899 ``urlencode()`` function.
901 :param mapping: Either a dict to urlencode or a list of
902 ``(key, value)`` pairs.
904 """
905 encoded_pairs = []
906 if hasattr(mapping, 'items'):
907 pairs = mapping.items()
908 else:
909 pairs = mapping
910 for key, value in pairs:
911 if isinstance(value, list):
912 for element in value:
913 encoded_pairs.append(
914 f'{percent_encode(key)}={percent_encode(element)}'
915 )
916 else:
917 encoded_pairs.append(
918 f'{percent_encode(key)}={percent_encode(value)}'
919 )
920 return '&'.join(encoded_pairs)
923def percent_encode(input_str, safe=SAFE_CHARS):
924 """Urlencodes a string.
926 Whereas percent_encode_sequence handles taking a dict/sequence and
927 producing a percent encoded string, this function deals only with
928 taking a string (not a dict/sequence) and percent encoding it.
930 If given the binary type, will simply URL encode it. If given the
931 text type, will produce the binary type by UTF-8 encoding the
932 text. If given something else, will convert it to the text type
933 first.
934 """
935 # If its not a binary or text string, make it a text string.
936 if not isinstance(input_str, (bytes, str)):
937 input_str = str(input_str)
938 # If it's not bytes, make it bytes by UTF-8 encoding it.
939 if not isinstance(input_str, bytes):
940 input_str = input_str.encode('utf-8')
941 return quote(input_str, safe=safe)
944def _epoch_seconds_to_datetime(value, tzinfo):
945 """Parse numerical epoch timestamps (seconds since 1970) into a
946 ``datetime.datetime`` in UTC using ``datetime.timedelta``. This is intended
947 as fallback when ``fromtimestamp`` raises ``OverflowError`` or ``OSError``.
949 :type value: float or int
950 :param value: The Unix timestamps as number.
952 :type tzinfo: callable
953 :param tzinfo: A ``datetime.tzinfo`` class or compatible callable.
954 """
955 epoch_zero = datetime.datetime(1970, 1, 1, 0, 0, 0, tzinfo=tzutc())
956 epoch_zero_localized = epoch_zero.astimezone(tzinfo())
957 return epoch_zero_localized + datetime.timedelta(seconds=value)
960def _parse_timestamp_with_tzinfo(value, tzinfo):
961 """Parse timestamp with pluggable tzinfo options."""
962 if isinstance(value, (int, float)):
963 # Possibly an epoch time.
964 return datetime.datetime.fromtimestamp(value, tzinfo())
965 else:
966 try:
967 return datetime.datetime.fromtimestamp(float(value), tzinfo())
968 except (TypeError, ValueError):
969 pass
970 try:
971 # In certain cases, a timestamp marked with GMT can be parsed into a
972 # different time zone, so here we provide a context which will
973 # enforce that GMT == UTC.
974 return dateutil.parser.parse(value, tzinfos={'GMT': tzutc()})
975 except (TypeError, ValueError) as e:
976 raise ValueError(f'Invalid timestamp "{value}": {e}')
979def parse_timestamp(value):
980 """Parse a timestamp into a datetime object.
982 Supported formats:
984 * iso8601
985 * rfc822
986 * epoch (value is an integer)
988 This will return a ``datetime.datetime`` object.
990 """
991 tzinfo_options = get_tzinfo_options()
992 for tzinfo in tzinfo_options:
993 try:
994 return _parse_timestamp_with_tzinfo(value, tzinfo)
995 except (OSError, OverflowError) as e:
996 logger.debug(
997 'Unable to parse timestamp with "%s" timezone info.',
998 tzinfo.__name__,
999 exc_info=e,
1000 )
1001 # For numeric values attempt fallback to using fromtimestamp-free method.
1002 # From Python's ``datetime.datetime.fromtimestamp`` documentation: "This
1003 # may raise ``OverflowError``, if the timestamp is out of the range of
1004 # values supported by the platform C localtime() function, and ``OSError``
1005 # on localtime() failure. It's common for this to be restricted to years
1006 # from 1970 through 2038."
1007 try:
1008 numeric_value = float(value)
1009 except (TypeError, ValueError):
1010 pass
1011 else:
1012 try:
1013 for tzinfo in tzinfo_options:
1014 return _epoch_seconds_to_datetime(numeric_value, tzinfo=tzinfo)
1015 except (OSError, OverflowError) as e:
1016 logger.debug(
1017 'Unable to parse timestamp using fallback method with "%s" '
1018 'timezone info.',
1019 tzinfo.__name__,
1020 exc_info=e,
1021 )
1022 raise RuntimeError(
1023 f'Unable to calculate correct timezone offset for "{value}"'
1024 )
1027def parse_to_aware_datetime(value):
1028 """Converted the passed in value to a datetime object with tzinfo.
1030 This function can be used to normalize all timestamp inputs. This
1031 function accepts a number of different types of inputs, but
1032 will always return a datetime.datetime object with time zone
1033 information.
1035 The input param ``value`` can be one of several types:
1037 * A datetime object (both naive and aware)
1038 * An integer representing the epoch time (can also be a string
1039 of the integer, i.e '0', instead of 0). The epoch time is
1040 considered to be UTC.
1041 * An iso8601 formatted timestamp. This does not need to be
1042 a complete timestamp, it can contain just the date portion
1043 without the time component.
1045 The returned value will be a datetime object that will have tzinfo.
1046 If no timezone info was provided in the input value, then UTC is
1047 assumed, not local time.
1049 """
1050 # This is a general purpose method that handles several cases of
1051 # converting the provided value to a string timestamp suitable to be
1052 # serialized to an http request. It can handle:
1053 # 1) A datetime.datetime object.
1054 if isinstance(value, _DatetimeClass):
1055 datetime_obj = value
1056 else:
1057 # 2) A string object that's formatted as a timestamp.
1058 # We document this as being an iso8601 timestamp, although
1059 # parse_timestamp is a bit more flexible.
1060 datetime_obj = parse_timestamp(value)
1061 if datetime_obj.tzinfo is None:
1062 # I think a case would be made that if no time zone is provided,
1063 # we should use the local time. However, to restore backwards
1064 # compat, the previous behavior was to assume UTC, which is
1065 # what we're going to do here.
1066 datetime_obj = datetime_obj.replace(tzinfo=tzutc())
1067 else:
1068 datetime_obj = datetime_obj.astimezone(tzutc())
1069 return datetime_obj
1072def datetime2timestamp(dt, default_timezone=None):
1073 """Calculate the timestamp based on the given datetime instance.
1075 :type dt: datetime
1076 :param dt: A datetime object to be converted into timestamp
1077 :type default_timezone: tzinfo
1078 :param default_timezone: If it is provided as None, we treat it as tzutc().
1079 But it is only used when dt is a naive datetime.
1080 :returns: The timestamp
1081 """
1082 epoch = datetime.datetime(1970, 1, 1)
1083 if dt.tzinfo is None:
1084 if default_timezone is None:
1085 default_timezone = tzutc()
1086 dt = dt.replace(tzinfo=default_timezone)
1087 d = dt.replace(tzinfo=None) - dt.utcoffset() - epoch
1088 return d.total_seconds()
1091def calculate_sha256(body, as_hex=False):
1092 """Calculate a sha256 checksum.
1094 This method will calculate the sha256 checksum of a file like
1095 object. Note that this method will iterate through the entire
1096 file contents. The caller is responsible for ensuring the proper
1097 starting position of the file and ``seek()``'ing the file back
1098 to its starting location if other consumers need to read from
1099 the file like object.
1101 :param body: Any file like object. The file must be opened
1102 in binary mode such that a ``.read()`` call returns bytes.
1103 :param as_hex: If True, then the hex digest is returned.
1104 If False, then the digest (as binary bytes) is returned.
1106 :returns: The sha256 checksum
1108 """
1109 checksum = hashlib.sha256()
1110 for chunk in iter(lambda: body.read(1024 * 1024), b''):
1111 checksum.update(chunk)
1112 if as_hex:
1113 return checksum.hexdigest()
1114 else:
1115 return checksum.digest()
1118def calculate_tree_hash(body):
1119 """Calculate a tree hash checksum.
1121 For more information see:
1123 http://docs.aws.amazon.com/amazonglacier/latest/dev/checksum-calculations.html
1125 :param body: Any file like object. This has the same constraints as
1126 the ``body`` param in calculate_sha256
1128 :rtype: str
1129 :returns: The hex version of the calculated tree hash
1131 """
1132 chunks = []
1133 required_chunk_size = 1024 * 1024
1134 sha256 = hashlib.sha256
1135 for chunk in iter(lambda: body.read(required_chunk_size), b''):
1136 chunks.append(sha256(chunk).digest())
1137 if not chunks:
1138 return sha256(b'').hexdigest()
1139 while len(chunks) > 1:
1140 new_chunks = []
1141 for first, second in _in_pairs(chunks):
1142 if second is not None:
1143 new_chunks.append(sha256(first + second).digest())
1144 else:
1145 # We're at the end of the list and there's no pair left.
1146 new_chunks.append(first)
1147 chunks = new_chunks
1148 return binascii.hexlify(chunks[0]).decode('ascii')
1151def _in_pairs(iterable):
1152 # Creates iterator that iterates over the list in pairs:
1153 # for a, b in _in_pairs([0, 1, 2, 3, 4]):
1154 # print(a, b)
1155 #
1156 # will print:
1157 # 0, 1
1158 # 2, 3
1159 # 4, None
1160 shared_iter = iter(iterable)
1161 # Note that zip_longest is a compat import that uses
1162 # the itertools izip_longest. This creates an iterator,
1163 # this call below does _not_ immediately create the list
1164 # of pairs.
1165 return zip_longest(shared_iter, shared_iter)
1168class CachedProperty:
1169 """A read only property that caches the initially computed value.
1171 This descriptor will only call the provided ``fget`` function once.
1172 Subsequent access to this property will return the cached value.
1174 """
1176 def __init__(self, fget):
1177 self._fget = fget
1179 def __get__(self, obj, cls):
1180 if obj is None:
1181 return self
1182 else:
1183 computed_value = self._fget(obj)
1184 obj.__dict__[self._fget.__name__] = computed_value
1185 return computed_value
1188class ArgumentGenerator:
1189 """Generate sample input based on a shape model.
1191 This class contains a ``generate_skeleton`` method that will take
1192 an input/output shape (created from ``botocore.model``) and generate
1193 a sample dictionary corresponding to the input/output shape.
1195 The specific values used are place holder values. For strings either an
1196 empty string or the member name can be used, for numbers 0 or 0.0 is used.
1197 The intended usage of this class is to generate the *shape* of the input
1198 structure.
1200 This can be useful for operations that have complex input shapes.
1201 This allows a user to just fill in the necessary data instead of
1202 worrying about the specific structure of the input arguments.
1204 Example usage::
1206 s = botocore.session.get_session()
1207 ddb = s.get_service_model('dynamodb')
1208 arg_gen = ArgumentGenerator()
1209 sample_input = arg_gen.generate_skeleton(
1210 ddb.operation_model('CreateTable').input_shape)
1211 print("Sample input for dynamodb.CreateTable: %s" % sample_input)
1213 """
1215 def __init__(self, use_member_names=False):
1216 self._use_member_names = use_member_names
1218 def generate_skeleton(self, shape):
1219 """Generate a sample input.
1221 :type shape: ``botocore.model.Shape``
1222 :param shape: The input shape.
1224 :return: The generated skeleton input corresponding to the
1225 provided input shape.
1227 """
1228 stack = []
1229 return self._generate_skeleton(shape, stack)
1231 def _generate_skeleton(self, shape, stack, name=''):
1232 stack.append(shape.name)
1233 try:
1234 if shape.type_name == 'structure':
1235 return self._generate_type_structure(shape, stack)
1236 elif shape.type_name == 'list':
1237 return self._generate_type_list(shape, stack)
1238 elif shape.type_name == 'map':
1239 return self._generate_type_map(shape, stack)
1240 elif shape.type_name == 'string':
1241 if self._use_member_names:
1242 return name
1243 if shape.enum:
1244 return random.choice(shape.enum)
1245 return ''
1246 elif shape.type_name in ['integer', 'long']:
1247 return 0
1248 elif shape.type_name in ['float', 'double']:
1249 return 0.0
1250 elif shape.type_name == 'boolean':
1251 return True
1252 elif shape.type_name == 'timestamp':
1253 return datetime.datetime(1970, 1, 1, 0, 0, 0)
1254 finally:
1255 stack.pop()
1257 def _generate_type_structure(self, shape, stack):
1258 if stack.count(shape.name) > 1:
1259 return {}
1260 skeleton = OrderedDict()
1261 for member_name, member_shape in shape.members.items():
1262 skeleton[member_name] = self._generate_skeleton(
1263 member_shape, stack, name=member_name
1264 )
1265 return skeleton
1267 def _generate_type_list(self, shape, stack):
1268 # For list elements we've arbitrarily decided to
1269 # return two elements for the skeleton list.
1270 name = ''
1271 if self._use_member_names:
1272 name = shape.member.name
1273 return [
1274 self._generate_skeleton(shape.member, stack, name),
1275 ]
1277 def _generate_type_map(self, shape, stack):
1278 key_shape = shape.key
1279 value_shape = shape.value
1280 assert key_shape.type_name == 'string'
1281 return OrderedDict(
1282 [
1283 ('KeyName', self._generate_skeleton(value_shape, stack)),
1284 ]
1285 )
1288def is_valid_ipv6_endpoint_url(endpoint_url):
1289 if UNSAFE_URL_CHARS.intersection(endpoint_url):
1290 return False
1291 hostname = f'[{urlparse(endpoint_url).hostname}]'
1292 return IPV6_ADDRZ_RE.match(hostname) is not None
1295def is_valid_ipv4_endpoint_url(endpoint_url):
1296 hostname = urlparse(endpoint_url).hostname
1297 return IPV4_RE.match(hostname) is not None
1300def is_valid_endpoint_url(endpoint_url):
1301 """Verify the endpoint_url is valid.
1303 :type endpoint_url: string
1304 :param endpoint_url: An endpoint_url. Must have at least a scheme
1305 and a hostname.
1307 :return: True if the endpoint url is valid. False otherwise.
1309 """
1310 # post-bpo-43882 urlsplit() strips unsafe characters from URL, causing
1311 # it to pass hostname validation below. Detect them early to fix that.
1312 if UNSAFE_URL_CHARS.intersection(endpoint_url):
1313 return False
1314 parts = urlsplit(endpoint_url)
1315 hostname = parts.hostname
1316 if hostname is None:
1317 return False
1318 if len(hostname) > 255:
1319 return False
1320 if hostname[-1] == ".":
1321 hostname = hostname[:-1]
1322 allowed = re.compile(
1323 r"^((?!-)[A-Z\d-]{1,63}(?<!-)\.)*((?!-)[A-Z\d-]{1,63}(?<!-))$",
1324 re.IGNORECASE,
1325 )
1326 return allowed.match(hostname)
1329def is_valid_uri(endpoint_url):
1330 return is_valid_endpoint_url(endpoint_url) or is_valid_ipv6_endpoint_url(
1331 endpoint_url
1332 )
1335def validate_region_name(region_name):
1336 """Provided region_name must be a valid host label."""
1337 if region_name is None:
1338 return
1339 valid_host_label = re.compile(r'^(?![0-9]+$)(?!-)[a-zA-Z0-9-]{,63}(?<!-)$')
1340 valid = valid_host_label.match(region_name)
1341 if not valid:
1342 raise InvalidRegionError(region_name=region_name)
1345def check_dns_name(bucket_name):
1346 """
1347 Check to see if the ``bucket_name`` complies with the
1348 restricted DNS naming conventions necessary to allow
1349 access via virtual-hosting style.
1351 Even though "." characters are perfectly valid in this DNS
1352 naming scheme, we are going to punt on any name containing a
1353 "." character because these will cause SSL cert validation
1354 problems if we try to use virtual-hosting style addressing.
1355 """
1356 if '.' in bucket_name:
1357 return False
1358 n = len(bucket_name)
1359 if n < 3 or n > 63:
1360 # Wrong length
1361 return False
1362 match = LABEL_RE.match(bucket_name)
1363 if match is None or match.end() != len(bucket_name):
1364 return False
1365 return True
1368def fix_s3_host(
1369 request,
1370 signature_version,
1371 region_name,
1372 default_endpoint_url=None,
1373 **kwargs,
1374):
1375 """
1376 This handler looks at S3 requests just before they are signed.
1377 If there is a bucket name on the path (true for everything except
1378 ListAllBuckets) it checks to see if that bucket name conforms to
1379 the DNS naming conventions. If it does, it alters the request to
1380 use ``virtual hosting`` style addressing rather than ``path-style``
1381 addressing.
1383 """
1384 if request.context.get('use_global_endpoint', False):
1385 default_endpoint_url = 's3.amazonaws.com'
1386 try:
1387 switch_to_virtual_host_style(
1388 request, signature_version, default_endpoint_url
1389 )
1390 except InvalidDNSNameError as e:
1391 bucket_name = e.kwargs['bucket_name']
1392 logger.debug(
1393 'Not changing URI, bucket is not DNS compatible: %s', bucket_name
1394 )
1397def switch_to_virtual_host_style(
1398 request, signature_version, default_endpoint_url=None, **kwargs
1399):
1400 """
1401 This is a handler to force virtual host style s3 addressing no matter
1402 the signature version (which is taken in consideration for the default
1403 case). If the bucket is not DNS compatible an InvalidDNSName is thrown.
1405 :param request: A AWSRequest object that is about to be sent.
1406 :param signature_version: The signature version to sign with
1407 :param default_endpoint_url: The endpoint to use when switching to a
1408 virtual style. If None is supplied, the virtual host will be
1409 constructed from the url of the request.
1410 """
1411 if request.auth_path is not None:
1412 # The auth_path has already been applied (this may be a
1413 # retried request). We don't need to perform this
1414 # customization again.
1415 return
1416 elif _is_get_bucket_location_request(request):
1417 # For the GetBucketLocation response, we should not be using
1418 # the virtual host style addressing so we can avoid any sigv4
1419 # issues.
1420 logger.debug(
1421 "Request is GetBucketLocation operation, not checking "
1422 "for DNS compatibility."
1423 )
1424 return
1425 parts = urlsplit(request.url)
1426 request.auth_path = parts.path
1427 path_parts = parts.path.split('/')
1429 # Retrieve what the endpoint we will be prepending the bucket name to.
1430 if default_endpoint_url is None:
1431 default_endpoint_url = parts.netloc
1433 if len(path_parts) > 1:
1434 bucket_name = path_parts[1]
1435 if not bucket_name:
1436 # If the bucket name is empty we should not be checking for
1437 # dns compatibility.
1438 return
1439 logger.debug('Checking for DNS compatible bucket for: %s', request.url)
1440 if check_dns_name(bucket_name):
1441 # If the operation is on a bucket, the auth_path must be
1442 # terminated with a '/' character.
1443 if len(path_parts) == 2:
1444 if request.auth_path[-1] != '/':
1445 request.auth_path += '/'
1446 path_parts.remove(bucket_name)
1447 # At the very least the path must be a '/', such as with the
1448 # CreateBucket operation when DNS style is being used. If this
1449 # is not used you will get an empty path which is incorrect.
1450 path = '/'.join(path_parts) or '/'
1451 global_endpoint = default_endpoint_url
1452 host = bucket_name + '.' + global_endpoint
1453 new_tuple = (parts.scheme, host, path, parts.query, '')
1454 new_uri = urlunsplit(new_tuple)
1455 request.url = new_uri
1456 logger.debug('URI updated to: %s', new_uri)
1457 else:
1458 raise InvalidDNSNameError(bucket_name=bucket_name)
1461def _is_get_bucket_location_request(request):
1462 return request.url.endswith('?location')
1465def instance_cache(func):
1466 """Method decorator for caching method calls to a single instance.
1468 **This is not a general purpose caching decorator.**
1470 In order to use this, you *must* provide an ``_instance_cache``
1471 attribute on the instance.
1473 This decorator is used to cache method calls. The cache is only
1474 scoped to a single instance though such that multiple instances
1475 will maintain their own cache. In order to keep things simple,
1476 this decorator requires that you provide an ``_instance_cache``
1477 attribute on your instance.
1479 """
1480 func_name = func.__name__
1482 @functools.wraps(func)
1483 def _cache_guard(self, *args, **kwargs):
1484 cache_key = (func_name, args)
1485 if kwargs:
1486 kwarg_items = tuple(sorted(kwargs.items()))
1487 cache_key = (func_name, args, kwarg_items)
1488 result = self._instance_cache.get(cache_key)
1489 if result is not None:
1490 return result
1491 result = func(self, *args, **kwargs)
1492 self._instance_cache[cache_key] = result
1493 return result
1495 return _cache_guard
1498def lru_cache_weakref(*cache_args, **cache_kwargs):
1499 """
1500 Version of functools.lru_cache that stores a weak reference to ``self``.
1502 Serves the same purpose as :py:func:`instance_cache` but uses Python's
1503 functools implementation which offers ``max_size`` and ``typed`` properties.
1505 lru_cache is a global cache even when used on a method. The cache's
1506 reference to ``self`` will prevent garbage collection of the object. This
1507 wrapper around functools.lru_cache replaces the reference to ``self`` with
1508 a weak reference to not interfere with garbage collection.
1509 """
1511 def wrapper(func):
1512 @functools.lru_cache(*cache_args, **cache_kwargs)
1513 def func_with_weakref(weakref_to_self, *args, **kwargs):
1514 return func(weakref_to_self(), *args, **kwargs)
1516 @functools.wraps(func)
1517 def inner(self, *args, **kwargs):
1518 for kwarg_key, kwarg_value in kwargs.items():
1519 if isinstance(kwarg_value, list):
1520 kwargs[kwarg_key] = tuple(kwarg_value)
1521 return func_with_weakref(weakref.ref(self), *args, **kwargs)
1523 inner.cache_info = func_with_weakref.cache_info
1524 return inner
1526 return wrapper
1529def switch_host_s3_accelerate(request, operation_name, **kwargs):
1530 """Switches the current s3 endpoint with an S3 Accelerate endpoint"""
1532 # Note that when registered the switching of the s3 host happens
1533 # before it gets changed to virtual. So we are not concerned with ensuring
1534 # that the bucket name is translated to the virtual style here and we
1535 # can hard code the Accelerate endpoint.
1536 parts = urlsplit(request.url).netloc.split('.')
1537 parts = [p for p in parts if p in S3_ACCELERATE_WHITELIST]
1538 endpoint = 'https://s3-accelerate.'
1539 if len(parts) > 0:
1540 endpoint += '.'.join(parts) + '.'
1541 endpoint += 'amazonaws.com'
1543 if operation_name in ['ListBuckets', 'CreateBucket', 'DeleteBucket']:
1544 return
1545 _switch_hosts(request, endpoint, use_new_scheme=False)
1548def switch_host_with_param(request, param_name):
1549 """Switches the host using a parameter value from a JSON request body"""
1550 request_json = json.loads(request.data.decode('utf-8'))
1551 if request_json.get(param_name):
1552 new_endpoint = request_json[param_name]
1553 _switch_hosts(request, new_endpoint)
1556def _switch_hosts(request, new_endpoint, use_new_scheme=True):
1557 final_endpoint = _get_new_endpoint(
1558 request.url, new_endpoint, use_new_scheme
1559 )
1560 request.url = final_endpoint
1563def _get_new_endpoint(original_endpoint, new_endpoint, use_new_scheme=True):
1564 new_endpoint_components = urlsplit(new_endpoint)
1565 original_endpoint_components = urlsplit(original_endpoint)
1566 scheme = original_endpoint_components.scheme
1567 if use_new_scheme:
1568 scheme = new_endpoint_components.scheme
1569 final_endpoint_components = (
1570 scheme,
1571 new_endpoint_components.netloc,
1572 original_endpoint_components.path,
1573 original_endpoint_components.query,
1574 '',
1575 )
1576 final_endpoint = urlunsplit(final_endpoint_components)
1577 logger.debug(
1578 'Updating URI from %s to %s', original_endpoint, final_endpoint
1579 )
1580 return final_endpoint
1583def deep_merge(base, extra):
1584 """Deeply two dictionaries, overriding existing keys in the base.
1586 :param base: The base dictionary which will be merged into.
1587 :param extra: The dictionary to merge into the base. Keys from this
1588 dictionary will take precedence.
1589 """
1590 for key in extra:
1591 # If the key represents a dict on both given dicts, merge the sub-dicts
1592 if (
1593 key in base
1594 and isinstance(base[key], dict)
1595 and isinstance(extra[key], dict)
1596 ):
1597 deep_merge(base[key], extra[key])
1598 continue
1600 # Otherwise, set the key on the base to be the value of the extra.
1601 base[key] = extra[key]
1604def hyphenize_service_id(service_id):
1605 """Translate the form used for event emitters.
1607 :param service_id: The service_id to convert.
1608 """
1609 return service_id.replace(' ', '-').lower()
1612class IdentityCache:
1613 """Base IdentityCache implementation for storing and retrieving
1614 highly accessed credentials.
1616 This class is not intended to be instantiated in user code.
1617 """
1619 METHOD = "base_identity_cache"
1621 def __init__(self, client, credential_cls):
1622 self._client = client
1623 self._credential_cls = credential_cls
1625 def get_credentials(self, **kwargs):
1626 callback = self.build_refresh_callback(**kwargs)
1627 metadata = callback()
1628 credential_entry = self._credential_cls.create_from_metadata(
1629 metadata=metadata,
1630 refresh_using=callback,
1631 method=self.METHOD,
1632 advisory_timeout=45,
1633 mandatory_timeout=10,
1634 )
1635 return credential_entry
1637 def build_refresh_callback(**kwargs):
1638 """Callback to be implemented by subclasses.
1640 Returns a set of metadata to be converted into a new
1641 credential instance.
1642 """
1643 raise NotImplementedError()
1646class S3ExpressIdentityCache(IdentityCache):
1647 """S3Express IdentityCache for retrieving and storing
1648 credentials from CreateSession calls.
1650 This class is not intended to be instantiated in user code.
1651 """
1653 METHOD = "s3express"
1655 def __init__(self, client, credential_cls):
1656 self._client = client
1657 self._credential_cls = credential_cls
1659 @functools.lru_cache(maxsize=100)
1660 def get_credentials(self, bucket):
1661 return super().get_credentials(bucket=bucket)
1663 def build_refresh_callback(self, bucket):
1664 def refresher():
1665 response = self._client.create_session(Bucket=bucket)
1666 creds = response['Credentials']
1667 expiration = self._serialize_if_needed(
1668 creds['Expiration'], iso=True
1669 )
1670 return {
1671 "access_key": creds['AccessKeyId'],
1672 "secret_key": creds['SecretAccessKey'],
1673 "token": creds['SessionToken'],
1674 "expiry_time": expiration,
1675 }
1677 return refresher
1679 def _serialize_if_needed(self, value, iso=False):
1680 if isinstance(value, _DatetimeClass):
1681 if iso:
1682 return value.isoformat()
1683 return value.strftime('%Y-%m-%dT%H:%M:%S%Z')
1684 return value
1687class S3ExpressIdentityResolver:
1688 def __init__(self, client, credential_cls, cache=None):
1689 self._client = weakref.proxy(client)
1691 if cache is None:
1692 cache = S3ExpressIdentityCache(self._client, credential_cls)
1693 self._cache = cache
1695 def register(self, event_emitter=None):
1696 logger.debug('Registering S3Express Identity Resolver')
1697 emitter = event_emitter or self._client.meta.events
1698 emitter.register('before-call.s3', self.apply_signing_cache_key)
1699 emitter.register('before-sign.s3', self.resolve_s3express_identity)
1701 def apply_signing_cache_key(self, params, context, **kwargs):
1702 endpoint_properties = context.get('endpoint_properties', {})
1703 backend = endpoint_properties.get('backend', None)
1705 # Add cache key if Bucket supplied for s3express request
1706 bucket_name = context.get('input_params', {}).get('Bucket')
1707 if backend == 'S3Express' and bucket_name is not None:
1708 context.setdefault('signing', {})
1709 context['signing']['cache_key'] = bucket_name
1711 def resolve_s3express_identity(
1712 self,
1713 request,
1714 signing_name,
1715 region_name,
1716 signature_version,
1717 request_signer,
1718 operation_name,
1719 **kwargs,
1720 ):
1721 signing_context = request.context.get('signing', {})
1722 signing_name = signing_context.get('signing_name')
1723 if signing_name == 's3express' and signature_version.startswith(
1724 'v4-s3express'
1725 ):
1726 signing_context['identity_cache'] = self._cache
1727 if 'cache_key' not in signing_context:
1728 signing_context['cache_key'] = (
1729 request.context.get('s3_redirect', {})
1730 .get('params', {})
1731 .get('Bucket')
1732 )
1735class S3RegionRedirectorv2:
1736 """Updated version of S3RegionRedirector for use when
1737 EndpointRulesetResolver is in use for endpoint resolution.
1739 This class is considered private and subject to abrupt breaking changes or
1740 removal without prior announcement. Please do not use it directly.
1741 """
1743 def __init__(self, endpoint_bridge, client, cache=None):
1744 self._cache = cache or {}
1745 self._client = weakref.proxy(client)
1747 def register(self, event_emitter=None):
1748 logger.debug('Registering S3 region redirector handler')
1749 emitter = event_emitter or self._client.meta.events
1750 emitter.register('needs-retry.s3', self.redirect_from_error)
1751 emitter.register(
1752 'before-parameter-build.s3', self.annotate_request_context
1753 )
1754 emitter.register(
1755 'before-endpoint-resolution.s3', self.redirect_from_cache
1756 )
1758 def redirect_from_error(self, request_dict, response, operation, **kwargs):
1759 """
1760 An S3 request sent to the wrong region will return an error that
1761 contains the endpoint the request should be sent to. This handler
1762 will add the redirect information to the signing context and then
1763 redirect the request.
1764 """
1765 if response is None:
1766 # This could be none if there was a ConnectionError or other
1767 # transport error.
1768 return
1770 redirect_ctx = request_dict.get('context', {}).get('s3_redirect', {})
1771 if ArnParser.is_arn(redirect_ctx.get('bucket')):
1772 logger.debug(
1773 'S3 request was previously for an Accesspoint ARN, not '
1774 'redirecting.'
1775 )
1776 return
1778 if redirect_ctx.get('redirected'):
1779 logger.debug(
1780 'S3 request was previously redirected, not redirecting.'
1781 )
1782 return
1784 error = response[1].get('Error', {})
1785 error_code = error.get('Code')
1786 response_metadata = response[1].get('ResponseMetadata', {})
1788 # We have to account for 400 responses because
1789 # if we sign a Head* request with the wrong region,
1790 # we'll get a 400 Bad Request but we won't get a
1791 # body saying it's an "AuthorizationHeaderMalformed".
1792 is_special_head_object = (
1793 error_code in ('301', '400') and operation.name == 'HeadObject'
1794 )
1795 is_special_head_bucket = (
1796 error_code in ('301', '400')
1797 and operation.name == 'HeadBucket'
1798 and 'x-amz-bucket-region'
1799 in response_metadata.get('HTTPHeaders', {})
1800 )
1801 is_wrong_signing_region = (
1802 error_code == 'AuthorizationHeaderMalformed' and 'Region' in error
1803 )
1804 is_redirect_status = response[0] is not None and response[
1805 0
1806 ].status_code in (301, 302, 307)
1807 is_permanent_redirect = error_code == 'PermanentRedirect'
1808 is_opt_in_region_redirect = (
1809 error_code == 'IllegalLocationConstraintException'
1810 and operation.name != 'CreateBucket'
1811 )
1812 if not any(
1813 [
1814 is_special_head_object,
1815 is_wrong_signing_region,
1816 is_permanent_redirect,
1817 is_special_head_bucket,
1818 is_redirect_status,
1819 is_opt_in_region_redirect,
1820 ]
1821 ):
1822 return
1824 bucket = request_dict['context']['s3_redirect']['bucket']
1825 client_region = request_dict['context'].get('client_region')
1826 new_region = self.get_bucket_region(bucket, response)
1828 if new_region is None:
1829 logger.debug(
1830 "S3 client configured for region %s but the "
1831 "bucket %s is not in that region and the proper region "
1832 "could not be automatically determined.",
1833 client_region,
1834 bucket,
1835 )
1836 return
1838 logger.debug(
1839 "S3 client configured for region %s but the bucket %s "
1840 "is in region %s; Please configure the proper region to "
1841 "avoid multiple unnecessary redirects and signing attempts.",
1842 client_region,
1843 bucket,
1844 new_region,
1845 )
1846 register_feature_id('S3_REGION_REDIRECT')
1848 # Adding the new region to _cache will make construct_endpoint() to
1849 # use the new region as value for the AWS::Region builtin parameter.
1850 self._cache[bucket] = new_region
1852 # Re-resolve endpoint with new region and modify request_dict with
1853 # the new URL, auth scheme, and signing context.
1854 ep_resolver = self._client._ruleset_resolver
1855 ep_info = ep_resolver.construct_endpoint(
1856 operation_model=operation,
1857 call_args=request_dict['context']['s3_redirect']['params'],
1858 request_context=request_dict['context'],
1859 )
1860 request_dict['url'] = self.set_request_url(
1861 request_dict['url'], ep_info.url
1862 )
1863 request_dict['context']['s3_redirect']['redirected'] = True
1864 auth_schemes = ep_info.properties.get('authSchemes')
1865 if auth_schemes is not None:
1866 auth_info = ep_resolver.auth_schemes_to_signing_ctx(auth_schemes)
1867 auth_type, signing_context = auth_info
1868 request_dict['context']['auth_type'] = auth_type
1869 request_dict['context']['signing'] = {
1870 **request_dict['context'].get('signing', {}),
1871 **signing_context,
1872 }
1874 # Return 0 so it doesn't wait to retry
1875 return 0
1877 def get_bucket_region(self, bucket, response):
1878 """
1879 There are multiple potential sources for the new region to redirect to,
1880 but they aren't all universally available for use. This will try to
1881 find region from response elements, but will fall back to calling
1882 HEAD on the bucket if all else fails.
1884 :param bucket: The bucket to find the region for. This is necessary if
1885 the region is not available in the error response.
1886 :param response: A response representing a service request that failed
1887 due to incorrect region configuration.
1888 """
1889 # First try to source the region from the headers.
1890 service_response = response[1]
1891 response_headers = service_response['ResponseMetadata']['HTTPHeaders']
1892 if 'x-amz-bucket-region' in response_headers:
1893 region = response_headers['x-amz-bucket-region']
1894 # Next, check the error body
1895 elif r := service_response.get('Error', {}).get('Region', None):
1896 region = r
1897 else:
1898 # Finally, HEAD the bucket. No other choice sadly.
1899 try:
1900 response = self._client.head_bucket(Bucket=bucket)
1901 headers = response['ResponseMetadata']['HTTPHeaders']
1902 except ClientError as e:
1903 headers = e.response['ResponseMetadata']['HTTPHeaders']
1904 region = headers.get('x-amz-bucket-region', None)
1905 validate_region_name(region)
1906 return region
1908 def set_request_url(self, old_url, new_endpoint, **kwargs):
1909 """
1910 Splice a new endpoint into an existing URL. Note that some endpoints
1911 from the the endpoint provider have a path component which will be
1912 discarded by this function.
1913 """
1914 return _get_new_endpoint(old_url, new_endpoint, False)
1916 def redirect_from_cache(self, builtins, params, **kwargs):
1917 """
1918 If a bucket name has been redirected before, it is in the cache. This
1919 handler will update the AWS::Region endpoint resolver builtin param
1920 to use the region from cache instead of the client region to avoid the
1921 redirect.
1922 """
1923 bucket = params.get('Bucket')
1924 if bucket is not None and bucket in self._cache:
1925 new_region = self._cache.get(bucket)
1926 builtins['AWS::Region'] = new_region
1927 register_feature_id('S3_REGION_REDIRECT')
1929 def annotate_request_context(self, params, context, **kwargs):
1930 """Store the bucket name in context for later use when redirecting.
1931 The bucket name may be an access point ARN or alias.
1932 """
1933 bucket = params.get('Bucket')
1934 context['s3_redirect'] = {
1935 'redirected': False,
1936 'bucket': bucket,
1937 'params': params,
1938 }
1941class S3RegionRedirector:
1942 """This handler has been replaced by S3RegionRedirectorv2. The original
1943 version remains in place for any third-party libraries that import it.
1944 """
1946 def __init__(self, endpoint_bridge, client, cache=None):
1947 self._endpoint_resolver = endpoint_bridge
1948 self._cache = cache
1949 if self._cache is None:
1950 self._cache = {}
1952 # This needs to be a weak ref in order to prevent memory leaks on
1953 # python 2.6
1954 self._client = weakref.proxy(client)
1956 warnings.warn(
1957 'The S3RegionRedirector class has been deprecated for a new '
1958 'internal replacement. A future version of botocore may remove '
1959 'this class.',
1960 category=FutureWarning,
1961 )
1963 def register(self, event_emitter=None):
1964 emitter = event_emitter or self._client.meta.events
1965 emitter.register('needs-retry.s3', self.redirect_from_error)
1966 emitter.register('before-call.s3', self.set_request_url)
1967 emitter.register('before-parameter-build.s3', self.redirect_from_cache)
1969 def redirect_from_error(self, request_dict, response, operation, **kwargs):
1970 """
1971 An S3 request sent to the wrong region will return an error that
1972 contains the endpoint the request should be sent to. This handler
1973 will add the redirect information to the signing context and then
1974 redirect the request.
1975 """
1976 if response is None:
1977 # This could be none if there was a ConnectionError or other
1978 # transport error.
1979 return
1981 if self._is_s3_accesspoint(request_dict.get('context', {})):
1982 logger.debug(
1983 'S3 request was previously to an accesspoint, not redirecting.'
1984 )
1985 return
1987 if request_dict.get('context', {}).get('s3_redirected'):
1988 logger.debug(
1989 'S3 request was previously redirected, not redirecting.'
1990 )
1991 return
1993 error = response[1].get('Error', {})
1994 error_code = error.get('Code')
1995 response_metadata = response[1].get('ResponseMetadata', {})
1997 # We have to account for 400 responses because
1998 # if we sign a Head* request with the wrong region,
1999 # we'll get a 400 Bad Request but we won't get a
2000 # body saying it's an "AuthorizationHeaderMalformed".
2001 is_special_head_object = (
2002 error_code in ('301', '400') and operation.name == 'HeadObject'
2003 )
2004 is_special_head_bucket = (
2005 error_code in ('301', '400')
2006 and operation.name == 'HeadBucket'
2007 and 'x-amz-bucket-region'
2008 in response_metadata.get('HTTPHeaders', {})
2009 )
2010 is_wrong_signing_region = (
2011 error_code == 'AuthorizationHeaderMalformed' and 'Region' in error
2012 )
2013 is_redirect_status = response[0] is not None and response[
2014 0
2015 ].status_code in (301, 302, 307)
2016 is_permanent_redirect = error_code == 'PermanentRedirect'
2017 if not any(
2018 [
2019 is_special_head_object,
2020 is_wrong_signing_region,
2021 is_permanent_redirect,
2022 is_special_head_bucket,
2023 is_redirect_status,
2024 ]
2025 ):
2026 return
2028 bucket = request_dict['context']['signing']['bucket']
2029 client_region = request_dict['context'].get('client_region')
2030 new_region = self.get_bucket_region(bucket, response)
2032 if new_region is None:
2033 logger.debug(
2034 "S3 client configured for region %s but the bucket %s is not "
2035 "in that region and the proper region could not be "
2036 "automatically determined.",
2037 client_region,
2038 bucket,
2039 )
2040 return
2042 logger.debug(
2043 "S3 client configured for region %s but the bucket %s is in region"
2044 " %s; Please configure the proper region to avoid multiple "
2045 "unnecessary redirects and signing attempts.",
2046 client_region,
2047 bucket,
2048 new_region,
2049 )
2050 endpoint = self._endpoint_resolver.resolve('s3', new_region)
2051 endpoint = endpoint['endpoint_url']
2053 signing_context = {
2054 'region': new_region,
2055 'bucket': bucket,
2056 'endpoint': endpoint,
2057 }
2058 request_dict['context']['signing'] = signing_context
2060 self._cache[bucket] = signing_context
2061 self.set_request_url(request_dict, request_dict['context'])
2063 request_dict['context']['s3_redirected'] = True
2065 # Return 0 so it doesn't wait to retry
2066 return 0
2068 def get_bucket_region(self, bucket, response):
2069 """
2070 There are multiple potential sources for the new region to redirect to,
2071 but they aren't all universally available for use. This will try to
2072 find region from response elements, but will fall back to calling
2073 HEAD on the bucket if all else fails.
2075 :param bucket: The bucket to find the region for. This is necessary if
2076 the region is not available in the error response.
2077 :param response: A response representing a service request that failed
2078 due to incorrect region configuration.
2079 """
2080 # First try to source the region from the headers.
2081 service_response = response[1]
2082 response_headers = service_response['ResponseMetadata']['HTTPHeaders']
2083 if 'x-amz-bucket-region' in response_headers:
2084 return response_headers['x-amz-bucket-region']
2086 # Next, check the error body
2087 region = service_response.get('Error', {}).get('Region', None)
2088 if region is not None:
2089 return region
2091 # Finally, HEAD the bucket. No other choice sadly.
2092 try:
2093 response = self._client.head_bucket(Bucket=bucket)
2094 headers = response['ResponseMetadata']['HTTPHeaders']
2095 except ClientError as e:
2096 headers = e.response['ResponseMetadata']['HTTPHeaders']
2098 region = headers.get('x-amz-bucket-region', None)
2099 return region
2101 def set_request_url(self, params, context, **kwargs):
2102 endpoint = context.get('signing', {}).get('endpoint', None)
2103 if endpoint is not None:
2104 params['url'] = _get_new_endpoint(params['url'], endpoint, False)
2106 def redirect_from_cache(self, params, context, **kwargs):
2107 """
2108 This handler retrieves a given bucket's signing context from the cache
2109 and adds it into the request context.
2110 """
2111 if self._is_s3_accesspoint(context):
2112 return
2113 bucket = params.get('Bucket')
2114 signing_context = self._cache.get(bucket)
2115 if signing_context is not None:
2116 context['signing'] = signing_context
2117 else:
2118 context['signing'] = {'bucket': bucket}
2120 def _is_s3_accesspoint(self, context):
2121 return 's3_accesspoint' in context
2124class InvalidArnException(ValueError):
2125 pass
2128class ArnParser:
2129 def parse_arn(self, arn):
2130 arn_parts = arn.split(':', 5)
2131 if len(arn_parts) < 6:
2132 raise InvalidArnException(
2133 f'Provided ARN: {arn} must be of the format: '
2134 'arn:partition:service:region:account:resource'
2135 )
2136 return {
2137 'partition': arn_parts[1],
2138 'service': arn_parts[2],
2139 'region': arn_parts[3],
2140 'account': arn_parts[4],
2141 'resource': arn_parts[5],
2142 }
2144 @staticmethod
2145 def is_arn(value):
2146 if not isinstance(value, str) or not value.startswith('arn:'):
2147 return False
2148 arn_parser = ArnParser()
2149 try:
2150 arn_parser.parse_arn(value)
2151 return True
2152 except InvalidArnException:
2153 return False
2156class S3ArnParamHandler:
2157 _RESOURCE_REGEX = re.compile(
2158 r'^(?P<resource_type>accesspoint|outpost)[/:](?P<resource_name>.+)$'
2159 )
2160 _OUTPOST_RESOURCE_REGEX = re.compile(
2161 r'^(?P<outpost_name>[a-zA-Z0-9\-]{1,63})[/:]accesspoint[/:]'
2162 r'(?P<accesspoint_name>[a-zA-Z0-9\-]{1,63}$)'
2163 )
2164 _BLACKLISTED_OPERATIONS = ['CreateBucket']
2166 def __init__(self, arn_parser=None):
2167 self._arn_parser = arn_parser
2168 if arn_parser is None:
2169 self._arn_parser = ArnParser()
2171 def register(self, event_emitter):
2172 event_emitter.register('before-parameter-build.s3', self.handle_arn)
2174 def handle_arn(self, params, model, context, **kwargs):
2175 if model.name in self._BLACKLISTED_OPERATIONS:
2176 return
2177 arn_details = self._get_arn_details_from_bucket_param(params)
2178 if arn_details is None:
2179 return
2180 if arn_details['resource_type'] == 'accesspoint':
2181 self._store_accesspoint(params, context, arn_details)
2182 elif arn_details['resource_type'] == 'outpost':
2183 self._store_outpost(params, context, arn_details)
2185 def _get_arn_details_from_bucket_param(self, params):
2186 if 'Bucket' in params:
2187 try:
2188 arn = params['Bucket']
2189 arn_details = self._arn_parser.parse_arn(arn)
2190 self._add_resource_type_and_name(arn, arn_details)
2191 return arn_details
2192 except InvalidArnException:
2193 pass
2194 return None
2196 def _add_resource_type_and_name(self, arn, arn_details):
2197 match = self._RESOURCE_REGEX.match(arn_details['resource'])
2198 if match:
2199 arn_details['resource_type'] = match.group('resource_type')
2200 arn_details['resource_name'] = match.group('resource_name')
2201 else:
2202 raise UnsupportedS3ArnError(arn=arn)
2204 def _store_accesspoint(self, params, context, arn_details):
2205 # Ideally the access-point would be stored as a parameter in the
2206 # request where the serializer would then know how to serialize it,
2207 # but access-points are not modeled in S3 operations so it would fail
2208 # validation. Instead, we set the access-point to the bucket parameter
2209 # to have some value set when serializing the request and additional
2210 # information on the context from the arn to use in forming the
2211 # access-point endpoint.
2212 params['Bucket'] = arn_details['resource_name']
2213 context['s3_accesspoint'] = {
2214 'name': arn_details['resource_name'],
2215 'account': arn_details['account'],
2216 'partition': arn_details['partition'],
2217 'region': arn_details['region'],
2218 'service': arn_details['service'],
2219 }
2221 def _store_outpost(self, params, context, arn_details):
2222 resource_name = arn_details['resource_name']
2223 match = self._OUTPOST_RESOURCE_REGEX.match(resource_name)
2224 if not match:
2225 raise UnsupportedOutpostResourceError(resource_name=resource_name)
2226 # Because we need to set the bucket name to something to pass
2227 # validation we're going to use the access point name to be consistent
2228 # with normal access point arns.
2229 accesspoint_name = match.group('accesspoint_name')
2230 params['Bucket'] = accesspoint_name
2231 context['s3_accesspoint'] = {
2232 'outpost_name': match.group('outpost_name'),
2233 'name': accesspoint_name,
2234 'account': arn_details['account'],
2235 'partition': arn_details['partition'],
2236 'region': arn_details['region'],
2237 'service': arn_details['service'],
2238 }
2241class S3EndpointSetter:
2242 _DEFAULT_PARTITION = 'aws'
2243 _DEFAULT_DNS_SUFFIX = 'amazonaws.com'
2245 def __init__(
2246 self,
2247 endpoint_resolver,
2248 region=None,
2249 s3_config=None,
2250 endpoint_url=None,
2251 partition=None,
2252 use_fips_endpoint=False,
2253 ):
2254 # This is calling the endpoint_resolver in regions.py
2255 self._endpoint_resolver = endpoint_resolver
2256 self._region = region
2257 self._s3_config = s3_config
2258 self._use_fips_endpoint = use_fips_endpoint
2259 if s3_config is None:
2260 self._s3_config = {}
2261 self._endpoint_url = endpoint_url
2262 self._partition = partition
2263 if partition is None:
2264 self._partition = self._DEFAULT_PARTITION
2266 def register(self, event_emitter):
2267 event_emitter.register('before-sign.s3', self.set_endpoint)
2268 event_emitter.register('choose-signer.s3', self.set_signer)
2269 event_emitter.register(
2270 'before-call.s3.WriteGetObjectResponse',
2271 self.update_endpoint_to_s3_object_lambda,
2272 )
2274 def update_endpoint_to_s3_object_lambda(self, params, context, **kwargs):
2275 if self._use_accelerate_endpoint:
2276 raise UnsupportedS3ConfigurationError(
2277 msg='S3 client does not support accelerate endpoints for S3 Object Lambda operations',
2278 )
2280 self._override_signing_name(context, 's3-object-lambda')
2281 if self._endpoint_url:
2282 # Only update the url if an explicit url was not provided
2283 return
2285 resolver = self._endpoint_resolver
2286 # Constructing endpoints as s3-object-lambda as region
2287 resolved = resolver.construct_endpoint(
2288 's3-object-lambda', self._region
2289 )
2291 # Ideally we would be able to replace the endpoint before
2292 # serialization but there's no event to do that currently
2293 # host_prefix is all the arn/bucket specs
2294 new_endpoint = 'https://{host_prefix}{hostname}'.format(
2295 host_prefix=params['host_prefix'],
2296 hostname=resolved['hostname'],
2297 )
2299 params['url'] = _get_new_endpoint(params['url'], new_endpoint, False)
2301 def set_endpoint(self, request, **kwargs):
2302 if self._use_accesspoint_endpoint(request):
2303 self._validate_accesspoint_supported(request)
2304 self._validate_fips_supported(request)
2305 self._validate_global_regions(request)
2306 region_name = self._resolve_region_for_accesspoint_endpoint(
2307 request
2308 )
2309 self._resolve_signing_name_for_accesspoint_endpoint(request)
2310 self._switch_to_accesspoint_endpoint(request, region_name)
2311 return
2312 if self._use_accelerate_endpoint:
2313 if self._use_fips_endpoint:
2314 raise UnsupportedS3ConfigurationError(
2315 msg=(
2316 'Client is configured to use the FIPS psuedo region '
2317 f'for "{self._region}", but S3 Accelerate does not have any FIPS '
2318 'compatible endpoints.'
2319 )
2320 )
2321 switch_host_s3_accelerate(request=request, **kwargs)
2322 if self._s3_addressing_handler:
2323 self._s3_addressing_handler(request=request, **kwargs)
2325 def _use_accesspoint_endpoint(self, request):
2326 return 's3_accesspoint' in request.context
2328 def _validate_fips_supported(self, request):
2329 if not self._use_fips_endpoint:
2330 return
2331 if 'fips' in request.context['s3_accesspoint']['region']:
2332 raise UnsupportedS3AccesspointConfigurationError(
2333 msg={'Invalid ARN, FIPS region not allowed in ARN.'}
2334 )
2335 if 'outpost_name' in request.context['s3_accesspoint']:
2336 raise UnsupportedS3AccesspointConfigurationError(
2337 msg=(
2338 f'Client is configured to use the FIPS psuedo-region "{self._region}", '
2339 'but outpost ARNs do not support FIPS endpoints.'
2340 )
2341 )
2342 # Transforming psuedo region to actual region
2343 accesspoint_region = request.context['s3_accesspoint']['region']
2344 if accesspoint_region != self._region:
2345 if not self._s3_config.get('use_arn_region', True):
2346 # TODO: Update message to reflect use_arn_region
2347 # is not set
2348 raise UnsupportedS3AccesspointConfigurationError(
2349 msg=(
2350 'Client is configured to use the FIPS psuedo-region '
2351 f'for "{self._region}", but the access-point ARN provided is for '
2352 f'the "{accesspoint_region}" region. For clients using a FIPS '
2353 'psuedo-region calls to access-point ARNs in another '
2354 'region are not allowed.'
2355 )
2356 )
2358 def _validate_global_regions(self, request):
2359 if self._s3_config.get('use_arn_region', True):
2360 return
2361 if self._region in ['aws-global', 's3-external-1']:
2362 raise UnsupportedS3AccesspointConfigurationError(
2363 msg=(
2364 'Client is configured to use the global psuedo-region '
2365 f'"{self._region}". When providing access-point ARNs a regional '
2366 'endpoint must be specified.'
2367 )
2368 )
2370 def _validate_accesspoint_supported(self, request):
2371 if self._use_accelerate_endpoint:
2372 raise UnsupportedS3AccesspointConfigurationError(
2373 msg=(
2374 'Client does not support s3 accelerate configuration '
2375 'when an access-point ARN is specified.'
2376 )
2377 )
2378 request_partition = request.context['s3_accesspoint']['partition']
2379 if request_partition != self._partition:
2380 raise UnsupportedS3AccesspointConfigurationError(
2381 msg=(
2382 f'Client is configured for "{self._partition}" partition, but access-point'
2383 f' ARN provided is for "{request_partition}" partition. The client and '
2384 ' access-point partition must be the same.'
2385 )
2386 )
2387 s3_service = request.context['s3_accesspoint'].get('service')
2388 if s3_service == 's3-object-lambda' and self._s3_config.get(
2389 'use_dualstack_endpoint'
2390 ):
2391 raise UnsupportedS3AccesspointConfigurationError(
2392 msg=(
2393 'Client does not support s3 dualstack configuration '
2394 'when an S3 Object Lambda access point ARN is specified.'
2395 )
2396 )
2397 outpost_name = request.context['s3_accesspoint'].get('outpost_name')
2398 if outpost_name and self._s3_config.get('use_dualstack_endpoint'):
2399 raise UnsupportedS3AccesspointConfigurationError(
2400 msg=(
2401 'Client does not support s3 dualstack configuration '
2402 'when an outpost ARN is specified.'
2403 )
2404 )
2405 self._validate_mrap_s3_config(request)
2407 def _validate_mrap_s3_config(self, request):
2408 if not is_global_accesspoint(request.context):
2409 return
2410 if self._s3_config.get('s3_disable_multiregion_access_points'):
2411 raise UnsupportedS3AccesspointConfigurationError(
2412 msg=(
2413 'Invalid configuration, Multi-Region Access Point '
2414 'ARNs are disabled.'
2415 )
2416 )
2417 elif self._s3_config.get('use_dualstack_endpoint'):
2418 raise UnsupportedS3AccesspointConfigurationError(
2419 msg=(
2420 'Client does not support s3 dualstack configuration '
2421 'when a Multi-Region Access Point ARN is specified.'
2422 )
2423 )
2425 def _resolve_region_for_accesspoint_endpoint(self, request):
2426 if is_global_accesspoint(request.context):
2427 # Requests going to MRAP endpoints MUST be set to any (*) region.
2428 self._override_signing_region(request, '*')
2429 elif self._s3_config.get('use_arn_region', True):
2430 accesspoint_region = request.context['s3_accesspoint']['region']
2431 # If we are using the region from the access point,
2432 # we will also want to make sure that we set it as the
2433 # signing region as well
2434 self._override_signing_region(request, accesspoint_region)
2435 return accesspoint_region
2436 return self._region
2438 def set_signer(self, context, **kwargs):
2439 if is_global_accesspoint(context):
2440 if HAS_CRT:
2441 return 's3v4a'
2442 else:
2443 raise MissingDependencyException(
2444 msg="Using S3 with an MRAP arn requires an additional "
2445 "dependency. You will need to pip install "
2446 "botocore[crt] before proceeding."
2447 )
2449 def _resolve_signing_name_for_accesspoint_endpoint(self, request):
2450 accesspoint_service = request.context['s3_accesspoint']['service']
2451 self._override_signing_name(request.context, accesspoint_service)
2453 def _switch_to_accesspoint_endpoint(self, request, region_name):
2454 original_components = urlsplit(request.url)
2455 accesspoint_endpoint = urlunsplit(
2456 (
2457 original_components.scheme,
2458 self._get_netloc(request.context, region_name),
2459 self._get_accesspoint_path(
2460 original_components.path, request.context
2461 ),
2462 original_components.query,
2463 '',
2464 )
2465 )
2466 logger.debug(
2467 'Updating URI from %s to %s', request.url, accesspoint_endpoint
2468 )
2469 request.url = accesspoint_endpoint
2471 def _get_netloc(self, request_context, region_name):
2472 if is_global_accesspoint(request_context):
2473 return self._get_mrap_netloc(request_context)
2474 else:
2475 return self._get_accesspoint_netloc(request_context, region_name)
2477 def _get_mrap_netloc(self, request_context):
2478 s3_accesspoint = request_context['s3_accesspoint']
2479 region_name = 's3-global'
2480 mrap_netloc_components = [s3_accesspoint['name']]
2481 if self._endpoint_url:
2482 endpoint_url_netloc = urlsplit(self._endpoint_url).netloc
2483 mrap_netloc_components.append(endpoint_url_netloc)
2484 else:
2485 partition = s3_accesspoint['partition']
2486 mrap_netloc_components.extend(
2487 [
2488 'accesspoint',
2489 region_name,
2490 self._get_partition_dns_suffix(partition),
2491 ]
2492 )
2493 return '.'.join(mrap_netloc_components)
2495 def _get_accesspoint_netloc(self, request_context, region_name):
2496 s3_accesspoint = request_context['s3_accesspoint']
2497 accesspoint_netloc_components = [
2498 '{}-{}'.format(s3_accesspoint['name'], s3_accesspoint['account']),
2499 ]
2500 outpost_name = s3_accesspoint.get('outpost_name')
2501 if self._endpoint_url:
2502 if outpost_name:
2503 accesspoint_netloc_components.append(outpost_name)
2504 endpoint_url_netloc = urlsplit(self._endpoint_url).netloc
2505 accesspoint_netloc_components.append(endpoint_url_netloc)
2506 else:
2507 if outpost_name:
2508 outpost_host = [outpost_name, 's3-outposts']
2509 accesspoint_netloc_components.extend(outpost_host)
2510 elif s3_accesspoint['service'] == 's3-object-lambda':
2511 component = self._inject_fips_if_needed(
2512 's3-object-lambda', request_context
2513 )
2514 accesspoint_netloc_components.append(component)
2515 else:
2516 component = self._inject_fips_if_needed(
2517 's3-accesspoint', request_context
2518 )
2519 accesspoint_netloc_components.append(component)
2520 if self._s3_config.get('use_dualstack_endpoint'):
2521 accesspoint_netloc_components.append('dualstack')
2522 accesspoint_netloc_components.extend(
2523 [region_name, self._get_dns_suffix(region_name)]
2524 )
2525 return '.'.join(accesspoint_netloc_components)
2527 def _inject_fips_if_needed(self, component, request_context):
2528 if self._use_fips_endpoint:
2529 return f'{component}-fips'
2530 return component
2532 def _get_accesspoint_path(self, original_path, request_context):
2533 # The Bucket parameter was substituted with the access-point name as
2534 # some value was required in serializing the bucket name. Now that
2535 # we are making the request directly to the access point, we will
2536 # want to remove that access-point name from the path.
2537 name = request_context['s3_accesspoint']['name']
2538 # All S3 operations require at least a / in their path.
2539 return original_path.replace('/' + name, '', 1) or '/'
2541 def _get_partition_dns_suffix(self, partition_name):
2542 dns_suffix = self._endpoint_resolver.get_partition_dns_suffix(
2543 partition_name
2544 )
2545 if dns_suffix is None:
2546 dns_suffix = self._DEFAULT_DNS_SUFFIX
2547 return dns_suffix
2549 def _get_dns_suffix(self, region_name):
2550 resolved = self._endpoint_resolver.construct_endpoint(
2551 's3', region_name
2552 )
2553 dns_suffix = self._DEFAULT_DNS_SUFFIX
2554 if resolved and 'dnsSuffix' in resolved:
2555 dns_suffix = resolved['dnsSuffix']
2556 return dns_suffix
2558 def _override_signing_region(self, request, region_name):
2559 signing_context = request.context.get('signing', {})
2560 # S3SigV4Auth will use the context['signing']['region'] value to
2561 # sign with if present. This is used by the Bucket redirector
2562 # as well but we should be fine because the redirector is never
2563 # used in combination with the accesspoint setting logic.
2564 signing_context['region'] = region_name
2565 request.context['signing'] = signing_context
2567 def _override_signing_name(self, context, signing_name):
2568 signing_context = context.get('signing', {})
2569 # S3SigV4Auth will use the context['signing']['signing_name'] value to
2570 # sign with if present. This is used by the Bucket redirector
2571 # as well but we should be fine because the redirector is never
2572 # used in combination with the accesspoint setting logic.
2573 signing_context['signing_name'] = signing_name
2574 context['signing'] = signing_context
2576 @CachedProperty
2577 def _use_accelerate_endpoint(self):
2578 # Enable accelerate if the configuration is set to to true or the
2579 # endpoint being used matches one of the accelerate endpoints.
2581 # Accelerate has been explicitly configured.
2582 if self._s3_config.get('use_accelerate_endpoint'):
2583 return True
2585 # Accelerate mode is turned on automatically if an endpoint url is
2586 # provided that matches the accelerate scheme.
2587 if self._endpoint_url is None:
2588 return False
2590 # Accelerate is only valid for Amazon endpoints.
2591 netloc = urlsplit(self._endpoint_url).netloc
2592 if not netloc.endswith('amazonaws.com'):
2593 return False
2595 # The first part of the url should always be s3-accelerate.
2596 parts = netloc.split('.')
2597 if parts[0] != 's3-accelerate':
2598 return False
2600 # Url parts between 's3-accelerate' and 'amazonaws.com' which
2601 # represent different url features.
2602 feature_parts = parts[1:-2]
2604 # There should be no duplicate url parts.
2605 if len(feature_parts) != len(set(feature_parts)):
2606 return False
2608 # Remaining parts must all be in the whitelist.
2609 return all(p in S3_ACCELERATE_WHITELIST for p in feature_parts)
2611 @CachedProperty
2612 def _addressing_style(self):
2613 # Use virtual host style addressing if accelerate is enabled or if
2614 # the given endpoint url is an accelerate endpoint.
2615 if self._use_accelerate_endpoint:
2616 return 'virtual'
2618 # If a particular addressing style is configured, use it.
2619 configured_addressing_style = self._s3_config.get('addressing_style')
2620 if configured_addressing_style:
2621 return configured_addressing_style
2623 @CachedProperty
2624 def _s3_addressing_handler(self):
2625 # If virtual host style was configured, use it regardless of whether
2626 # or not the bucket looks dns compatible.
2627 if self._addressing_style == 'virtual':
2628 logger.debug("Using S3 virtual host style addressing.")
2629 return switch_to_virtual_host_style
2631 # If path style is configured, no additional steps are needed. If
2632 # endpoint_url was specified, don't default to virtual. We could
2633 # potentially default provided endpoint urls to virtual hosted
2634 # style, but for now it is avoided.
2635 if self._addressing_style == 'path' or self._endpoint_url is not None:
2636 logger.debug("Using S3 path style addressing.")
2637 return None
2639 logger.debug(
2640 "Defaulting to S3 virtual host style addressing with "
2641 "path style addressing fallback."
2642 )
2644 # By default, try to use virtual style with path fallback.
2645 return fix_s3_host
2648class S3ControlEndpointSetter:
2649 _DEFAULT_PARTITION = 'aws'
2650 _DEFAULT_DNS_SUFFIX = 'amazonaws.com'
2651 _HOST_LABEL_REGEX = re.compile(r'^[a-zA-Z0-9\-]{1,63}$')
2653 def __init__(
2654 self,
2655 endpoint_resolver,
2656 region=None,
2657 s3_config=None,
2658 endpoint_url=None,
2659 partition=None,
2660 use_fips_endpoint=False,
2661 ):
2662 self._endpoint_resolver = endpoint_resolver
2663 self._region = region
2664 self._s3_config = s3_config
2665 self._use_fips_endpoint = use_fips_endpoint
2666 if s3_config is None:
2667 self._s3_config = {}
2668 self._endpoint_url = endpoint_url
2669 self._partition = partition
2670 if partition is None:
2671 self._partition = self._DEFAULT_PARTITION
2673 def register(self, event_emitter):
2674 event_emitter.register('before-sign.s3-control', self.set_endpoint)
2676 def set_endpoint(self, request, **kwargs):
2677 if self._use_endpoint_from_arn_details(request):
2678 self._validate_endpoint_from_arn_details_supported(request)
2679 region_name = self._resolve_region_from_arn_details(request)
2680 self._resolve_signing_name_from_arn_details(request)
2681 self._resolve_endpoint_from_arn_details(request, region_name)
2682 self._add_headers_from_arn_details(request)
2683 elif self._use_endpoint_from_outpost_id(request):
2684 self._validate_outpost_redirection_valid(request)
2685 self._override_signing_name(request, 's3-outposts')
2686 new_netloc = self._construct_outpost_endpoint(self._region)
2687 self._update_request_netloc(request, new_netloc)
2689 def _use_endpoint_from_arn_details(self, request):
2690 return 'arn_details' in request.context
2692 def _use_endpoint_from_outpost_id(self, request):
2693 return 'outpost_id' in request.context
2695 def _validate_endpoint_from_arn_details_supported(self, request):
2696 if 'fips' in request.context['arn_details']['region']:
2697 raise UnsupportedS3ControlArnError(
2698 arn=request.context['arn_details']['original'],
2699 msg='Invalid ARN, FIPS region not allowed in ARN.',
2700 )
2701 if not self._s3_config.get('use_arn_region', False):
2702 arn_region = request.context['arn_details']['region']
2703 if arn_region != self._region:
2704 error_msg = (
2705 'The use_arn_region configuration is disabled but '
2706 f'received arn for "{arn_region}" when the client is configured '
2707 f'to use "{self._region}"'
2708 )
2709 raise UnsupportedS3ControlConfigurationError(msg=error_msg)
2710 request_partion = request.context['arn_details']['partition']
2711 if request_partion != self._partition:
2712 raise UnsupportedS3ControlConfigurationError(
2713 msg=(
2714 f'Client is configured for "{self._partition}" partition, but arn '
2715 f'provided is for "{request_partion}" partition. The client and '
2716 'arn partition must be the same.'
2717 )
2718 )
2719 if self._s3_config.get('use_accelerate_endpoint'):
2720 raise UnsupportedS3ControlConfigurationError(
2721 msg='S3 control client does not support accelerate endpoints',
2722 )
2723 if 'outpost_name' in request.context['arn_details']:
2724 self._validate_outpost_redirection_valid(request)
2726 def _validate_outpost_redirection_valid(self, request):
2727 if self._s3_config.get('use_dualstack_endpoint'):
2728 raise UnsupportedS3ControlConfigurationError(
2729 msg=(
2730 'Client does not support s3 dualstack configuration '
2731 'when an outpost is specified.'
2732 )
2733 )
2735 def _resolve_region_from_arn_details(self, request):
2736 if self._s3_config.get('use_arn_region', False):
2737 arn_region = request.context['arn_details']['region']
2738 # If we are using the region from the expanded arn, we will also
2739 # want to make sure that we set it as the signing region as well
2740 self._override_signing_region(request, arn_region)
2741 return arn_region
2742 return self._region
2744 def _resolve_signing_name_from_arn_details(self, request):
2745 arn_service = request.context['arn_details']['service']
2746 self._override_signing_name(request, arn_service)
2747 return arn_service
2749 def _resolve_endpoint_from_arn_details(self, request, region_name):
2750 new_netloc = self._resolve_netloc_from_arn_details(
2751 request, region_name
2752 )
2753 self._update_request_netloc(request, new_netloc)
2755 def _update_request_netloc(self, request, new_netloc):
2756 original_components = urlsplit(request.url)
2757 arn_details_endpoint = urlunsplit(
2758 (
2759 original_components.scheme,
2760 new_netloc,
2761 original_components.path,
2762 original_components.query,
2763 '',
2764 )
2765 )
2766 logger.debug(
2767 'Updating URI from %s to %s', request.url, arn_details_endpoint
2768 )
2769 request.url = arn_details_endpoint
2771 def _resolve_netloc_from_arn_details(self, request, region_name):
2772 arn_details = request.context['arn_details']
2773 if 'outpost_name' in arn_details:
2774 return self._construct_outpost_endpoint(region_name)
2775 account = arn_details['account']
2776 return self._construct_s3_control_endpoint(region_name, account)
2778 def _is_valid_host_label(self, label):
2779 return self._HOST_LABEL_REGEX.match(label)
2781 def _validate_host_labels(self, *labels):
2782 for label in labels:
2783 if not self._is_valid_host_label(label):
2784 raise InvalidHostLabelError(label=label)
2786 def _construct_s3_control_endpoint(self, region_name, account):
2787 self._validate_host_labels(region_name, account)
2788 if self._endpoint_url:
2789 endpoint_url_netloc = urlsplit(self._endpoint_url).netloc
2790 netloc = [account, endpoint_url_netloc]
2791 else:
2792 netloc = [
2793 account,
2794 's3-control',
2795 ]
2796 self._add_dualstack(netloc)
2797 dns_suffix = self._get_dns_suffix(region_name)
2798 netloc.extend([region_name, dns_suffix])
2799 return self._construct_netloc(netloc)
2801 def _construct_outpost_endpoint(self, region_name):
2802 self._validate_host_labels(region_name)
2803 if self._endpoint_url:
2804 return urlsplit(self._endpoint_url).netloc
2805 else:
2806 netloc = [
2807 's3-outposts',
2808 region_name,
2809 self._get_dns_suffix(region_name),
2810 ]
2811 self._add_fips(netloc)
2812 return self._construct_netloc(netloc)
2814 def _construct_netloc(self, netloc):
2815 return '.'.join(netloc)
2817 def _add_fips(self, netloc):
2818 if self._use_fips_endpoint:
2819 netloc[0] = netloc[0] + '-fips'
2821 def _add_dualstack(self, netloc):
2822 if self._s3_config.get('use_dualstack_endpoint'):
2823 netloc.append('dualstack')
2825 def _get_dns_suffix(self, region_name):
2826 resolved = self._endpoint_resolver.construct_endpoint(
2827 's3', region_name
2828 )
2829 dns_suffix = self._DEFAULT_DNS_SUFFIX
2830 if resolved and 'dnsSuffix' in resolved:
2831 dns_suffix = resolved['dnsSuffix']
2832 return dns_suffix
2834 def _override_signing_region(self, request, region_name):
2835 signing_context = request.context.get('signing', {})
2836 # S3SigV4Auth will use the context['signing']['region'] value to
2837 # sign with if present. This is used by the Bucket redirector
2838 # as well but we should be fine because the redirector is never
2839 # used in combination with the accesspoint setting logic.
2840 signing_context['region'] = region_name
2841 request.context['signing'] = signing_context
2843 def _override_signing_name(self, request, signing_name):
2844 signing_context = request.context.get('signing', {})
2845 # S3SigV4Auth will use the context['signing']['signing_name'] value to
2846 # sign with if present. This is used by the Bucket redirector
2847 # as well but we should be fine because the redirector is never
2848 # used in combination with the accesspoint setting logic.
2849 signing_context['signing_name'] = signing_name
2850 request.context['signing'] = signing_context
2852 def _add_headers_from_arn_details(self, request):
2853 arn_details = request.context['arn_details']
2854 outpost_name = arn_details.get('outpost_name')
2855 if outpost_name:
2856 self._add_outpost_id_header(request, outpost_name)
2858 def _add_outpost_id_header(self, request, outpost_name):
2859 request.headers['x-amz-outpost-id'] = outpost_name
2862class S3ControlArnParamHandler:
2863 """This handler has been replaced by S3ControlArnParamHandlerv2. The
2864 original version remains in place for any third-party importers.
2865 """
2867 _RESOURCE_SPLIT_REGEX = re.compile(r'[/:]')
2869 def __init__(self, arn_parser=None):
2870 self._arn_parser = arn_parser
2871 if arn_parser is None:
2872 self._arn_parser = ArnParser()
2873 warnings.warn(
2874 'The S3ControlArnParamHandler class has been deprecated for a new '
2875 'internal replacement. A future version of botocore may remove '
2876 'this class.',
2877 category=FutureWarning,
2878 )
2880 def register(self, event_emitter):
2881 event_emitter.register(
2882 'before-parameter-build.s3-control',
2883 self.handle_arn,
2884 )
2886 def handle_arn(self, params, model, context, **kwargs):
2887 if model.name in ('CreateBucket', 'ListRegionalBuckets'):
2888 # CreateBucket and ListRegionalBuckets are special cases that do
2889 # not obey ARN based redirection but will redirect based off of the
2890 # presence of the OutpostId parameter
2891 self._handle_outpost_id_param(params, model, context)
2892 else:
2893 self._handle_name_param(params, model, context)
2894 self._handle_bucket_param(params, model, context)
2896 def _get_arn_details_from_param(self, params, param_name):
2897 if param_name not in params:
2898 return None
2899 try:
2900 arn = params[param_name]
2901 arn_details = self._arn_parser.parse_arn(arn)
2902 arn_details['original'] = arn
2903 arn_details['resources'] = self._split_resource(arn_details)
2904 return arn_details
2905 except InvalidArnException:
2906 return None
2908 def _split_resource(self, arn_details):
2909 return self._RESOURCE_SPLIT_REGEX.split(arn_details['resource'])
2911 def _override_account_id_param(self, params, arn_details):
2912 account_id = arn_details['account']
2913 if 'AccountId' in params and params['AccountId'] != account_id:
2914 error_msg = (
2915 'Account ID in arn does not match the AccountId parameter '
2916 'provided: "{}"'
2917 ).format(params['AccountId'])
2918 raise UnsupportedS3ControlArnError(
2919 arn=arn_details['original'],
2920 msg=error_msg,
2921 )
2922 params['AccountId'] = account_id
2924 def _handle_outpost_id_param(self, params, model, context):
2925 if 'OutpostId' not in params:
2926 return
2927 context['outpost_id'] = params['OutpostId']
2929 def _handle_name_param(self, params, model, context):
2930 # CreateAccessPoint is a special case that does not expand Name
2931 if model.name == 'CreateAccessPoint':
2932 return
2933 arn_details = self._get_arn_details_from_param(params, 'Name')
2934 if arn_details is None:
2935 return
2936 if self._is_outpost_accesspoint(arn_details):
2937 self._store_outpost_accesspoint(params, context, arn_details)
2938 else:
2939 error_msg = 'The Name parameter does not support the provided ARN'
2940 raise UnsupportedS3ControlArnError(
2941 arn=arn_details['original'],
2942 msg=error_msg,
2943 )
2945 def _is_outpost_accesspoint(self, arn_details):
2946 if arn_details['service'] != 's3-outposts':
2947 return False
2948 resources = arn_details['resources']
2949 if len(resources) != 4:
2950 return False
2951 # Resource must be of the form outpost/op-123/accesspoint/name
2952 return resources[0] == 'outpost' and resources[2] == 'accesspoint'
2954 def _store_outpost_accesspoint(self, params, context, arn_details):
2955 self._override_account_id_param(params, arn_details)
2956 accesspoint_name = arn_details['resources'][3]
2957 params['Name'] = accesspoint_name
2958 arn_details['accesspoint_name'] = accesspoint_name
2959 arn_details['outpost_name'] = arn_details['resources'][1]
2960 context['arn_details'] = arn_details
2962 def _handle_bucket_param(self, params, model, context):
2963 arn_details = self._get_arn_details_from_param(params, 'Bucket')
2964 if arn_details is None:
2965 return
2966 if self._is_outpost_bucket(arn_details):
2967 self._store_outpost_bucket(params, context, arn_details)
2968 else:
2969 error_msg = (
2970 'The Bucket parameter does not support the provided ARN'
2971 )
2972 raise UnsupportedS3ControlArnError(
2973 arn=arn_details['original'],
2974 msg=error_msg,
2975 )
2977 def _is_outpost_bucket(self, arn_details):
2978 if arn_details['service'] != 's3-outposts':
2979 return False
2980 resources = arn_details['resources']
2981 if len(resources) != 4:
2982 return False
2983 # Resource must be of the form outpost/op-123/bucket/name
2984 return resources[0] == 'outpost' and resources[2] == 'bucket'
2986 def _store_outpost_bucket(self, params, context, arn_details):
2987 self._override_account_id_param(params, arn_details)
2988 bucket_name = arn_details['resources'][3]
2989 params['Bucket'] = bucket_name
2990 arn_details['bucket_name'] = bucket_name
2991 arn_details['outpost_name'] = arn_details['resources'][1]
2992 context['arn_details'] = arn_details
2995class S3ControlArnParamHandlerv2(S3ControlArnParamHandler):
2996 """Updated version of S3ControlArnParamHandler for use when
2997 EndpointRulesetResolver is in use for endpoint resolution.
2999 This class is considered private and subject to abrupt breaking changes or
3000 removal without prior announcement. Please do not use it directly.
3001 """
3003 def __init__(self, arn_parser=None):
3004 self._arn_parser = arn_parser
3005 if arn_parser is None:
3006 self._arn_parser = ArnParser()
3008 def register(self, event_emitter):
3009 event_emitter.register(
3010 'before-endpoint-resolution.s3-control',
3011 self.handle_arn,
3012 )
3014 def _handle_name_param(self, params, model, context):
3015 # CreateAccessPoint is a special case that does not expand Name
3016 if model.name == 'CreateAccessPoint':
3017 return
3018 arn_details = self._get_arn_details_from_param(params, 'Name')
3019 if arn_details is None:
3020 return
3021 self._raise_for_fips_pseudo_region(arn_details)
3022 self._raise_for_accelerate_endpoint(context)
3023 if self._is_outpost_accesspoint(arn_details):
3024 self._store_outpost_accesspoint(params, context, arn_details)
3025 else:
3026 error_msg = 'The Name parameter does not support the provided ARN'
3027 raise UnsupportedS3ControlArnError(
3028 arn=arn_details['original'],
3029 msg=error_msg,
3030 )
3032 def _store_outpost_accesspoint(self, params, context, arn_details):
3033 self._override_account_id_param(params, arn_details)
3035 def _handle_bucket_param(self, params, model, context):
3036 arn_details = self._get_arn_details_from_param(params, 'Bucket')
3037 if arn_details is None:
3038 return
3039 self._raise_for_fips_pseudo_region(arn_details)
3040 self._raise_for_accelerate_endpoint(context)
3041 if self._is_outpost_bucket(arn_details):
3042 self._store_outpost_bucket(params, context, arn_details)
3043 else:
3044 error_msg = (
3045 'The Bucket parameter does not support the provided ARN'
3046 )
3047 raise UnsupportedS3ControlArnError(
3048 arn=arn_details['original'],
3049 msg=error_msg,
3050 )
3052 def _store_outpost_bucket(self, params, context, arn_details):
3053 self._override_account_id_param(params, arn_details)
3055 def _raise_for_fips_pseudo_region(self, arn_details):
3056 # FIPS pseudo region names cannot be used in ARNs
3057 arn_region = arn_details['region']
3058 if arn_region.startswith('fips-') or arn_region.endswith('fips-'):
3059 raise UnsupportedS3ControlArnError(
3060 arn=arn_details['original'],
3061 msg='Invalid ARN, FIPS region not allowed in ARN.',
3062 )
3064 def _raise_for_accelerate_endpoint(self, context):
3065 s3_config = context['client_config'].s3 or {}
3066 if s3_config.get('use_accelerate_endpoint'):
3067 raise UnsupportedS3ControlConfigurationError(
3068 msg='S3 control client does not support accelerate endpoints',
3069 )
3072class ContainerMetadataFetcher:
3073 TIMEOUT_SECONDS = 2
3074 RETRY_ATTEMPTS = 3
3075 SLEEP_TIME = 1
3076 IP_ADDRESS = '169.254.170.2'
3077 _ALLOWED_HOSTS = [
3078 IP_ADDRESS,
3079 '169.254.170.23',
3080 'fd00:ec2::23',
3081 'localhost',
3082 ]
3084 def __init__(self, session=None, sleep=time.sleep):
3085 if session is None:
3086 session = botocore.httpsession.URLLib3Session(
3087 timeout=self.TIMEOUT_SECONDS
3088 )
3089 self._session = session
3090 self._sleep = sleep
3092 def retrieve_full_uri(self, full_url, headers=None):
3093 """Retrieve JSON metadata from container metadata.
3095 :type full_url: str
3096 :param full_url: The full URL of the metadata service.
3097 This should include the scheme as well, e.g
3098 "http://localhost:123/foo"
3100 """
3101 self._validate_allowed_url(full_url)
3102 return self._retrieve_credentials(full_url, headers)
3104 def _validate_allowed_url(self, full_url):
3105 parsed = botocore.compat.urlparse(full_url)
3107 if parsed.scheme == 'https':
3108 return
3109 if self._is_loopback_address(parsed.hostname):
3110 return
3111 is_whitelisted_host = self._check_if_whitelisted_host(parsed.hostname)
3112 if not is_whitelisted_host:
3113 raise ValueError(
3114 f"Unsupported host '{parsed.hostname}'. Can only retrieve metadata "
3115 f"from a loopback address or one of these hosts: {', '.join(self._ALLOWED_HOSTS)}"
3116 )
3118 def _is_loopback_address(self, hostname):
3119 try:
3120 ip = ip_address(hostname)
3121 return ip.is_loopback
3122 except ValueError:
3123 return False
3125 def _check_if_whitelisted_host(self, host):
3126 if host in self._ALLOWED_HOSTS:
3127 return True
3128 return False
3130 def retrieve_uri(self, relative_uri):
3131 """Retrieve JSON metadata from container metadata.
3133 :type relative_uri: str
3134 :param relative_uri: A relative URI, e.g "/foo/bar?id=123"
3136 :return: The parsed JSON response.
3138 """
3139 full_url = self.full_url(relative_uri)
3140 return self._retrieve_credentials(full_url)
3142 def _retrieve_credentials(self, full_url, extra_headers=None):
3143 headers = {'Accept': 'application/json'}
3144 if extra_headers is not None:
3145 headers.update(extra_headers)
3146 attempts = 0
3147 while True:
3148 try:
3149 return self._get_response(
3150 full_url, headers, self.TIMEOUT_SECONDS
3151 )
3152 except MetadataRetrievalError as e:
3153 logger.debug(
3154 "Received error when attempting to retrieve "
3155 "container metadata: %s",
3156 e,
3157 exc_info=True,
3158 )
3159 self._sleep(self.SLEEP_TIME)
3160 attempts += 1
3161 if attempts >= self.RETRY_ATTEMPTS:
3162 raise
3164 def _get_response(self, full_url, headers, timeout):
3165 try:
3166 AWSRequest = botocore.awsrequest.AWSRequest
3167 request = AWSRequest(method='GET', url=full_url, headers=headers)
3168 response = self._session.send(request.prepare())
3169 response_text = response.content.decode('utf-8')
3170 if response.status_code != 200:
3171 raise MetadataRetrievalError(
3172 error_msg=(
3173 f"Received non 200 response {response.status_code} "
3174 f"from container metadata: {response_text}"
3175 )
3176 )
3177 try:
3178 return json.loads(response_text)
3179 except ValueError:
3180 error_msg = "Unable to parse JSON returned from container metadata services"
3181 logger.debug('%s:%s', error_msg, response_text)
3182 raise MetadataRetrievalError(error_msg=error_msg)
3183 except RETRYABLE_HTTP_ERRORS as e:
3184 error_msg = (
3185 "Received error when attempting to retrieve "
3186 f"container metadata: {e}"
3187 )
3188 raise MetadataRetrievalError(error_msg=error_msg)
3190 def full_url(self, relative_uri):
3191 return f'http://{self.IP_ADDRESS}{relative_uri}'
3194def get_environ_proxies(url):
3195 if should_bypass_proxies(url):
3196 return {}
3197 else:
3198 return getproxies()
3201def should_bypass_proxies(url):
3202 """
3203 Returns whether we should bypass proxies or not.
3204 """
3205 # NOTE: requests allowed for ip/cidr entries in no_proxy env that we don't
3206 # support current as urllib only checks DNS suffix
3207 # If the system proxy settings indicate that this URL should be bypassed,
3208 # don't proxy.
3209 # The proxy_bypass function is incredibly buggy on OS X in early versions
3210 # of Python 2.6, so allow this call to fail. Only catch the specific
3211 # exceptions we've seen, though: this call failing in other ways can reveal
3212 # legitimate problems.
3213 try:
3214 if proxy_bypass(urlparse(url).netloc):
3215 return True
3216 except (TypeError, socket.gaierror):
3217 pass
3219 return False
3222def determine_content_length(body):
3223 # No body, content length of 0
3224 if not body:
3225 return 0
3227 # Try asking the body for it's length
3228 try:
3229 return len(body)
3230 except (AttributeError, TypeError):
3231 pass
3233 # Try getting the length from a seekable stream
3234 if hasattr(body, 'seek') and hasattr(body, 'tell'):
3235 try:
3236 orig_pos = body.tell()
3237 body.seek(0, 2)
3238 end_file_pos = body.tell()
3239 body.seek(orig_pos)
3240 return end_file_pos - orig_pos
3241 except io.UnsupportedOperation:
3242 # in case when body is, for example, io.BufferedIOBase object
3243 # it has "seek" method which throws "UnsupportedOperation"
3244 # exception in such case we want to fall back to "chunked"
3245 # encoding
3246 pass
3247 # Failed to determine the length
3248 return None
3251def get_encoding_from_headers(headers, default='ISO-8859-1'):
3252 """Returns encodings from given HTTP Header Dict.
3254 :param headers: dictionary to extract encoding from.
3255 :param default: default encoding if the content-type is text
3256 """
3258 content_type = headers.get('content-type')
3260 if not content_type:
3261 return None
3263 message = email.message.Message()
3264 message['content-type'] = content_type
3265 charset = message.get_param("charset")
3267 if charset is not None:
3268 return charset
3270 if 'text' in content_type:
3271 return default
3274def calculate_md5(body, **kwargs):
3275 """This function has been deprecated, but is kept for backwards compatibility."""
3276 if isinstance(body, (bytes, bytearray)):
3277 binary_md5 = _calculate_md5_from_bytes(body)
3278 else:
3279 binary_md5 = _calculate_md5_from_file(body)
3280 return base64.b64encode(binary_md5).decode('ascii')
3283def _calculate_md5_from_bytes(body_bytes):
3284 """This function has been deprecated, but is kept for backwards compatibility."""
3285 md5 = get_md5(body_bytes, usedforsecurity=False)
3286 return md5.digest()
3289def _calculate_md5_from_file(fileobj):
3290 """This function has been deprecated, but is kept for backwards compatibility."""
3291 start_position = fileobj.tell()
3292 md5 = get_md5(usedforsecurity=False)
3293 for chunk in iter(lambda: fileobj.read(1024 * 1024), b''):
3294 md5.update(chunk)
3295 fileobj.seek(start_position)
3296 return md5.digest()
3299def _is_s3express_request(params):
3300 endpoint_properties = params.get('context', {}).get(
3301 'endpoint_properties', {}
3302 )
3303 return endpoint_properties.get('backend') == 'S3Express'
3306def get_checksum_algorithm_headers(params):
3307 """
3308 Returns the a list of header names from the request which start with
3309 "x-amz-checksum-", otherwise returns an empty list.
3311 This function is considered private and subject to abrupt breaking changes or
3312 removal without prior announcement. Please do not use it directly.
3313 """
3314 headers = params['headers']
3315 checksum_headers = []
3317 # If a header matching the x-amz-checksum-* pattern is present, we
3318 # extract and return the algorithm name.
3319 for header in headers:
3320 match = CHECKSUM_HEADER_PATTERN.match(header)
3321 if match:
3322 checksum_headers.append(header)
3323 return checksum_headers
3326def has_checksum_header(params):
3327 """
3328 Checks if a header starting with "x-amz-checksum-" is provided in a request.
3330 This function is considered private and subject to abrupt breaking changes or
3331 removal without prior announcement. Please do not use it directly.
3332 """
3333 return bool(get_checksum_algorithm_headers(params))
3336def conditionally_calculate_checksum(params, **kwargs):
3337 """This function has been deprecated, but is kept for backwards compatibility."""
3338 if not has_checksum_header(params):
3339 conditionally_calculate_md5(params, **kwargs)
3340 conditionally_enable_crc32(params, **kwargs)
3343def conditionally_enable_crc32(params, **kwargs):
3344 """This function has been deprecated, but is kept for backwards compatibility."""
3345 checksum_context = params.get('context', {}).get('checksum', {})
3346 checksum_algorithm = checksum_context.get('request_algorithm')
3347 if (
3348 _is_s3express_request(params)
3349 and params['body'] is not None
3350 and checksum_algorithm in (None, "conditional-md5")
3351 ):
3352 params['context']['checksum'] = {
3353 'request_algorithm': {
3354 'algorithm': 'crc32',
3355 'in': 'header',
3356 'name': 'x-amz-checksum-crc32',
3357 }
3358 }
3361def conditionally_calculate_md5(params, **kwargs):
3362 """Only add a Content-MD5 if the system supports it.
3364 This function has been deprecated, but is kept for backwards compatibility.
3365 """
3366 body = params['body']
3367 checksum_context = params.get('context', {}).get('checksum', {})
3368 checksum_algorithm = checksum_context.get('request_algorithm')
3369 if checksum_algorithm and checksum_algorithm != 'conditional-md5':
3370 # Skip for requests that will have a flexible checksum applied
3371 return
3373 if has_checksum_header(params):
3374 # Don't add a new header if one is already available.
3375 return
3377 if _is_s3express_request(params):
3378 # S3Express doesn't support MD5
3379 return
3381 if MD5_AVAILABLE and body is not None:
3382 md5_digest = calculate_md5(body, **kwargs)
3383 params['headers']['Content-MD5'] = md5_digest
3386class FileWebIdentityTokenLoader:
3387 def __init__(self, web_identity_token_path, _open=open):
3388 self._web_identity_token_path = web_identity_token_path
3389 self._open = _open
3391 def __call__(self):
3392 with self._open(self._web_identity_token_path) as token_file:
3393 return token_file.read()
3396class SSOTokenLoader:
3397 def __init__(self, cache=None):
3398 if cache is None:
3399 cache = {}
3400 self._cache = cache
3402 def _generate_cache_key(self, start_url, session_name):
3403 input_str = start_url
3404 if session_name is not None:
3405 input_str = session_name
3406 return hashlib.sha1(input_str.encode('utf-8')).hexdigest()
3408 def save_token(self, start_url, token, session_name=None):
3409 cache_key = self._generate_cache_key(start_url, session_name)
3410 self._cache[cache_key] = token
3412 def __call__(self, start_url, session_name=None):
3413 cache_key = self._generate_cache_key(start_url, session_name)
3414 logger.debug('Checking for cached token at: %s', cache_key)
3415 if cache_key not in self._cache:
3416 name = start_url
3417 if session_name is not None:
3418 name = session_name
3419 error_msg = f'Token for {name} does not exist'
3420 raise SSOTokenLoadError(error_msg=error_msg)
3422 token = self._cache[cache_key]
3423 if 'accessToken' not in token or 'expiresAt' not in token:
3424 error_msg = f'Token for {start_url} is invalid'
3425 raise SSOTokenLoadError(error_msg=error_msg)
3426 return token
3429class EventbridgeSignerSetter:
3430 _DEFAULT_PARTITION = 'aws'
3431 _DEFAULT_DNS_SUFFIX = 'amazonaws.com'
3433 def __init__(self, endpoint_resolver, region=None, endpoint_url=None):
3434 self._endpoint_resolver = endpoint_resolver
3435 self._region = region
3436 self._endpoint_url = endpoint_url
3438 def register(self, event_emitter):
3439 event_emitter.register(
3440 'before-parameter-build.events.PutEvents',
3441 self.check_for_global_endpoint,
3442 )
3443 event_emitter.register(
3444 'before-call.events.PutEvents', self.set_endpoint_url
3445 )
3447 def set_endpoint_url(self, params, context, **kwargs):
3448 if 'eventbridge_endpoint' in context:
3449 endpoint = context['eventbridge_endpoint']
3450 logger.debug(
3451 "Rewriting URL from %s to %s", params['url'], endpoint
3452 )
3453 params['url'] = endpoint
3455 def check_for_global_endpoint(self, params, context, **kwargs):
3456 endpoint = params.get('EndpointId')
3457 if endpoint is None:
3458 return
3460 if len(endpoint) == 0:
3461 raise InvalidEndpointConfigurationError(
3462 msg='EndpointId must not be a zero length string'
3463 )
3465 if not HAS_CRT:
3466 raise MissingDependencyException(
3467 msg="Using EndpointId requires an additional "
3468 "dependency. You will need to pip install "
3469 "botocore[crt] before proceeding."
3470 )
3472 config = context.get('client_config')
3473 endpoint_variant_tags = None
3474 if config is not None:
3475 if config.use_fips_endpoint:
3476 raise InvalidEndpointConfigurationError(
3477 msg="FIPS is not supported with EventBridge "
3478 "multi-region endpoints."
3479 )
3480 if config.use_dualstack_endpoint:
3481 endpoint_variant_tags = ['dualstack']
3483 if self._endpoint_url is None:
3484 # Validate endpoint is a valid hostname component
3485 parts = urlparse(f'https://{endpoint}')
3486 if parts.hostname != endpoint:
3487 raise InvalidEndpointConfigurationError(
3488 msg='EndpointId is not a valid hostname component.'
3489 )
3490 resolved_endpoint = self._get_global_endpoint(
3491 endpoint, endpoint_variant_tags=endpoint_variant_tags
3492 )
3493 else:
3494 resolved_endpoint = self._endpoint_url
3496 context['eventbridge_endpoint'] = resolved_endpoint
3497 context['auth_type'] = 'v4a'
3499 def _get_global_endpoint(self, endpoint, endpoint_variant_tags=None):
3500 resolver = self._endpoint_resolver
3502 partition = resolver.get_partition_for_region(self._region)
3503 if partition is None:
3504 partition = self._DEFAULT_PARTITION
3505 dns_suffix = resolver.get_partition_dns_suffix(
3506 partition, endpoint_variant_tags=endpoint_variant_tags
3507 )
3508 if dns_suffix is None:
3509 dns_suffix = self._DEFAULT_DNS_SUFFIX
3511 return f"https://{endpoint}.endpoint.events.{dns_suffix}/"
3514def is_s3_accelerate_url(url):
3515 """Does the URL match the S3 Accelerate endpoint scheme?
3517 Virtual host naming style with bucket names in the netloc part of the URL
3518 are not allowed by this function.
3519 """
3520 if url is None:
3521 return False
3523 # Accelerate is only valid for Amazon endpoints.
3524 url_parts = urlsplit(url)
3525 if not url_parts.netloc.endswith(
3526 'amazonaws.com'
3527 ) or url_parts.scheme not in ['https', 'http']:
3528 return False
3530 # The first part of the URL must be s3-accelerate.
3531 parts = url_parts.netloc.split('.')
3532 if parts[0] != 's3-accelerate':
3533 return False
3535 # Url parts between 's3-accelerate' and 'amazonaws.com' which
3536 # represent different url features.
3537 feature_parts = parts[1:-2]
3539 # There should be no duplicate URL parts.
3540 if len(feature_parts) != len(set(feature_parts)):
3541 return False
3543 # Remaining parts must all be in the whitelist.
3544 return all(p in S3_ACCELERATE_WHITELIST for p in feature_parts)
3547class JSONFileCache:
3548 """JSON file cache.
3549 This provides a dict like interface that stores JSON serializable
3550 objects.
3551 The objects are serialized to JSON and stored in a file. These
3552 values can be retrieved at a later time.
3553 """
3555 CACHE_DIR = os.path.expanduser(os.path.join('~', '.aws', 'boto', 'cache'))
3557 def __init__(self, working_dir=CACHE_DIR, dumps_func=None):
3558 self._working_dir = working_dir
3559 if dumps_func is None:
3560 dumps_func = self._default_dumps
3561 self._dumps = dumps_func
3563 def _default_dumps(self, obj):
3564 return json.dumps(obj, default=self._serialize_if_needed)
3566 def __contains__(self, cache_key):
3567 actual_key = self._convert_cache_key(cache_key)
3568 return os.path.isfile(actual_key)
3570 def __getitem__(self, cache_key):
3571 """Retrieve value from a cache key."""
3572 actual_key = self._convert_cache_key(cache_key)
3573 try:
3574 with open(actual_key) as f:
3575 return json.load(f)
3576 except (OSError, ValueError):
3577 raise KeyError(cache_key)
3579 def __delitem__(self, cache_key):
3580 actual_key = self._convert_cache_key(cache_key)
3581 try:
3582 key_path = Path(actual_key)
3583 key_path.unlink()
3584 except FileNotFoundError:
3585 raise KeyError(cache_key)
3587 def __setitem__(self, cache_key, value):
3588 full_key = self._convert_cache_key(cache_key)
3589 try:
3590 file_content = self._dumps(value)
3591 except (TypeError, ValueError):
3592 raise ValueError(
3593 f"Value cannot be cached, must be JSON serializable: {value}"
3594 )
3595 if not os.path.isdir(self._working_dir):
3596 os.makedirs(self._working_dir, exist_ok=True)
3598 temp_fd, temp_path = tempfile.mkstemp(
3599 dir=self._working_dir, suffix='.tmp'
3600 )
3601 with os.fdopen(temp_fd, 'w') as f:
3602 f.write(file_content)
3603 f.flush()
3604 os.fsync(f.fileno())
3606 os.replace(temp_path, full_key)
3608 def _convert_cache_key(self, cache_key):
3609 full_path = os.path.join(self._working_dir, cache_key + '.json')
3610 return full_path
3612 def _serialize_if_needed(self, value, iso=False):
3613 if isinstance(value, _DatetimeClass):
3614 if iso:
3615 return value.isoformat()
3616 return value.strftime('%Y-%m-%dT%H:%M:%S%Z')
3617 return value
3620def generate_login_cache_key(sign_in_session_name):
3621 return hashlib.sha256(sign_in_session_name.encode('utf-8')).hexdigest()
3624def is_s3express_bucket(bucket):
3625 if bucket is None:
3626 return False
3627 return bucket.endswith('--x-s3')
3630def get_token_from_environment(signing_name, environ=None):
3631 if not isinstance(signing_name, str) or not signing_name.strip():
3632 return None
3634 if environ is None:
3635 environ = os.environ
3636 env_var = _get_bearer_env_var_name(signing_name)
3637 return environ.get(env_var)
3640def _get_bearer_env_var_name(signing_name):
3641 bearer_name = signing_name.replace('-', '_').replace(' ', '_').upper()
3642 return f"AWS_BEARER_TOKEN_{bearer_name}"
3645# This parameter is not part of the public interface and is subject to abrupt
3646# breaking changes or removal without prior announcement.
3647# Mapping of services that have been renamed for backwards compatibility reasons.
3648# Keys are the previous name that should be allowed, values are the documented
3649# and preferred client name.
3650SERVICE_NAME_ALIASES = {'runtime.sagemaker': 'sagemaker-runtime'}
3653# This parameter is not part of the public interface and is subject to abrupt
3654# breaking changes or removal without prior announcement.
3655# Mapping to determine the service ID for services that do not use it as the
3656# model data directory name. The keys are the data directory name and the
3657# values are the transformed service IDs (lower case and hyphenated).
3658CLIENT_NAME_TO_HYPHENIZED_SERVICE_ID_OVERRIDES = {
3659 # Actual service name we use -> Allowed computed service name.
3660 'apigateway': 'api-gateway',
3661 'application-autoscaling': 'application-auto-scaling',
3662 'appmesh': 'app-mesh',
3663 'autoscaling': 'auto-scaling',
3664 'autoscaling-plans': 'auto-scaling-plans',
3665 'ce': 'cost-explorer',
3666 'cloudhsmv2': 'cloudhsm-v2',
3667 'cloudsearchdomain': 'cloudsearch-domain',
3668 'cognito-idp': 'cognito-identity-provider',
3669 'config': 'config-service',
3670 'cur': 'cost-and-usage-report-service',
3671 'datapipeline': 'data-pipeline',
3672 'directconnect': 'direct-connect',
3673 'devicefarm': 'device-farm',
3674 'discovery': 'application-discovery-service',
3675 'dms': 'database-migration-service',
3676 'ds': 'directory-service',
3677 'ds-data': 'directory-service-data',
3678 'dynamodbstreams': 'dynamodb-streams',
3679 'elasticbeanstalk': 'elastic-beanstalk',
3680 'elb': 'elastic-load-balancing',
3681 'elbv2': 'elastic-load-balancing-v2',
3682 'es': 'elasticsearch-service',
3683 'events': 'eventbridge',
3684 'globalaccelerator': 'global-accelerator',
3685 'iot-data': 'iot-data-plane',
3686 'iot-jobs-data': 'iot-jobs-data-plane',
3687 'iotwireless': 'iot-wireless',
3688 'kinesisanalytics': 'kinesis-analytics',
3689 'kinesisanalyticsv2': 'kinesis-analytics-v2',
3690 'kinesisvideo': 'kinesis-video',
3691 'lex-models': 'lex-model-building-service',
3692 'lexv2-models': 'lex-models-v2',
3693 'lex-runtime': 'lex-runtime-service',
3694 'lexv2-runtime': 'lex-runtime-v2',
3695 'logs': 'cloudwatch-logs',
3696 'machinelearning': 'machine-learning',
3697 'marketplacecommerceanalytics': 'marketplace-commerce-analytics',
3698 'marketplace-entitlement': 'marketplace-entitlement-service',
3699 'meteringmarketplace': 'marketplace-metering',
3700 'mgh': 'migration-hub',
3701 'sms-voice': 'pinpoint-sms-voice',
3702 'resourcegroupstaggingapi': 'resource-groups-tagging-api',
3703 'route53': 'route-53',
3704 'route53domains': 'route-53-domains',
3705 's3control': 's3-control',
3706 'sdb': 'simpledb',
3707 'secretsmanager': 'secrets-manager',
3708 'serverlessrepo': 'serverlessapplicationrepository',
3709 'servicecatalog': 'service-catalog',
3710 'servicecatalog-appregistry': 'service-catalog-appregistry',
3711 'stepfunctions': 'sfn',
3712 'storagegateway': 'storage-gateway',
3713}
3716def get_login_token_cache_directory():
3717 """Returns which directory contains the login_session token files"""
3718 if 'AWS_LOGIN_CACHE_DIRECTORY' in os.environ:
3719 path = os.path.expandvars(os.environ['AWS_LOGIN_CACHE_DIRECTORY'])
3720 path = os.path.expanduser(path)
3721 return path
3722 else:
3723 return os.path.expanduser(os.path.join('~', '.aws', 'login', 'cache'))
3726class LoginTokenLoader:
3727 """Loads and saves login access tokens to disk"""
3729 def __init__(self, cache=None):
3730 if cache is None:
3731 cache = {}
3732 self._cache = cache
3734 def save_token(self, session_name, token):
3735 cache_key = generate_login_cache_key(session_name)
3736 self._cache[cache_key] = token
3738 def load_token(self, session_name):
3739 cache_key = generate_login_cache_key(session_name)
3740 if cache_key not in self._cache:
3741 return None
3742 return self._cache[cache_key]