codekingpro/portable-devtools

1# Copyright 2022 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7# http://aws.amazon.com/apache2.0/8#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13 14import datetime15from io import BytesIO16 17from botocore.auth import (18    SIGNED_HEADERS_BLACKLIST,19    STREAMING_UNSIGNED_PAYLOAD_TRAILER,20    UNSIGNED_PAYLOAD,21    BaseSigner,22    _get_body_as_dict,23    _host_from_url,24)25from botocore.compat import HTTPHeaders, awscrt, parse_qs, urlsplit, urlunsplit26from botocore.exceptions import NoCredentialsError27from botocore.utils import percent_encode_sequence28 29 30class CrtSigV4Auth(BaseSigner):31    REQUIRES_REGION = True32    _PRESIGNED_HEADERS_BLOCKLIST = [33        'Authorization',34        'X-Amz-Date',35        'X-Amz-Content-SHA256',36        'X-Amz-Security-Token',37    ]38    _SIGNATURE_TYPE = awscrt.auth.AwsSignatureType.HTTP_REQUEST_HEADERS39    _USE_DOUBLE_URI_ENCODE = True40    _SHOULD_NORMALIZE_URI_PATH = True41 42    def __init__(self, credentials, service_name, region_name):43        self.credentials = credentials44        self._service_name = service_name45        self._region_name = region_name46        self._expiration_in_seconds = None47 48    def _is_streaming_checksum_payload(self, request):49        checksum_context = request.context.get('checksum', {})50        algorithm = checksum_context.get('request_algorithm')51        return isinstance(algorithm, dict) and algorithm.get('in') == 'trailer'52 53    def add_auth(self, request):54        if self.credentials is None:55            raise NoCredentialsError()56 57        # Use utcnow() because that's what gets mocked by tests, but set58        # timezone because CRT assumes naive datetime is local time.59        datetime_now = datetime.datetime.utcnow().replace(60            tzinfo=datetime.timezone.utc61        )62 63        # Use existing 'X-Amz-Content-SHA256' header if able64        existing_sha256 = self._get_existing_sha256(request)65 66        self._modify_request_before_signing(request)67 68        credentials_provider = awscrt.auth.AwsCredentialsProvider.new_static(69            access_key_id=self.credentials.access_key,70            secret_access_key=self.credentials.secret_key,71            session_token=self.credentials.token,72        )73 74        if self._is_streaming_checksum_payload(request):75            explicit_payload = STREAMING_UNSIGNED_PAYLOAD_TRAILER76        elif self._should_sha256_sign_payload(request):77            if existing_sha256:78                explicit_payload = existing_sha25679            else:80                explicit_payload = None  # to be calculated during signing81        else:82            explicit_payload = UNSIGNED_PAYLOAD83 84        if self._should_add_content_sha256_header(explicit_payload):85            body_header = (86                awscrt.auth.AwsSignedBodyHeaderType.X_AMZ_CONTENT_SHA_25687            )88        else:89            body_header = awscrt.auth.AwsSignedBodyHeaderType.NONE90 91        signing_config = awscrt.auth.AwsSigningConfig(92            algorithm=awscrt.auth.AwsSigningAlgorithm.V4,93            signature_type=self._SIGNATURE_TYPE,94            credentials_provider=credentials_provider,95            region=self._region_name,96            service=self._service_name,97            date=datetime_now,98            should_sign_header=self._should_sign_header,99            use_double_uri_encode=self._USE_DOUBLE_URI_ENCODE,100            should_normalize_uri_path=self._SHOULD_NORMALIZE_URI_PATH,101            signed_body_value=explicit_payload,102            signed_body_header_type=body_header,103            expiration_in_seconds=self._expiration_in_seconds,104        )105        crt_request = self._crt_request_from_aws_request(request)106        future = awscrt.auth.aws_sign_request(crt_request, signing_config)107        future.result()108        self._apply_signing_changes(request, crt_request)109 110    def _crt_request_from_aws_request(self, aws_request):111        url_parts = urlsplit(aws_request.url)112        crt_path = url_parts.path if url_parts.path else '/'113        if aws_request.params:114            array = []115            for param, value in aws_request.params.items():116                value = str(value)117                array.append(f'{param}={value}')118            crt_path = crt_path + '?' + '&'.join(array)119        elif url_parts.query:120            crt_path = f'{crt_path}?{url_parts.query}'121 122        crt_headers = awscrt.http.HttpHeaders(aws_request.headers.items())123 124        # CRT requires body (if it exists) to be an I/O stream.125        crt_body_stream = None126        if aws_request.body:127            if hasattr(aws_request.body, 'seek'):128                crt_body_stream = aws_request.body129            else:130                crt_body_stream = BytesIO(aws_request.body)131 132        crt_request = awscrt.http.HttpRequest(133            method=aws_request.method,134            path=crt_path,135            headers=crt_headers,136            body_stream=crt_body_stream,137        )138        return crt_request139 140    def _apply_signing_changes(self, aws_request, signed_crt_request):141        # Apply changes from signed CRT request to the AWSRequest142        aws_request.headers = HTTPHeaders.from_pairs(143            list(signed_crt_request.headers)144        )145 146    def _should_sign_header(self, name, **kwargs):147        return name.lower() not in SIGNED_HEADERS_BLACKLIST148 149    def _modify_request_before_signing(self, request):150        # This could be a retry. Make sure the previous151        # authorization headers are removed first.152        for h in self._PRESIGNED_HEADERS_BLOCKLIST:153            if h in request.headers:154                del request.headers[h]155        # If necessary, add the host header156        if 'host' not in request.headers:157            request.headers['host'] = _host_from_url(request.url)158 159    def _get_existing_sha256(self, request):160        return request.headers.get('X-Amz-Content-SHA256')161 162    def _should_sha256_sign_payload(self, request):163        # Payloads will always be signed over insecure connections.164        if not request.url.startswith('https'):165            return True166 167        # Certain operations may have payload signing disabled by default.168        # Since we don't have access to the operation model, we pass in this169        # bit of metadata through the request context.170        return request.context.get('payload_signing_enabled', True)171 172    def _should_add_content_sha256_header(self, explicit_payload):173        # only add X-Amz-Content-SHA256 header if payload is explicitly set174        return explicit_payload is not None175 176 177class CrtS3SigV4Auth(CrtSigV4Auth):178    # For S3, we do not normalize the path.179    _USE_DOUBLE_URI_ENCODE = False180    _SHOULD_NORMALIZE_URI_PATH = False181 182    def _get_existing_sha256(self, request):183        # always recalculate184        return None185 186    def _should_sha256_sign_payload(self, request):187        # S3 allows optional body signing, so to minimize the performance188        # impact, we opt to not SHA256 sign the body on streaming uploads,189        # provided that we're on https.190        client_config = request.context.get('client_config')191        s3_config = getattr(client_config, 's3', None)192 193        # The config could be None if it isn't set, or if the customer sets it194        # to None.195        if s3_config is None:196            s3_config = {}197 198        # The explicit configuration takes precedence over any implicit199        # configuration.200        sign_payload = s3_config.get('payload_signing_enabled', None)201        if sign_payload is not None:202            return sign_payload203 204        # We require that both a checksum be present and https be enabled205        # to implicitly disable body signing. The combination of TLS and206        # a checksum is sufficiently secure and durable for us to be207        # confident in the request without body signing.208        checksum_header = 'Content-MD5'209        checksum_context = request.context.get('checksum', {})210        algorithm = checksum_context.get('request_algorithm')211        if isinstance(algorithm, dict) and algorithm.get('in') == 'header':212            checksum_header = algorithm['name']213        if (214            not request.url.startswith('https')215            or checksum_header not in request.headers216        ):217            return True218 219        # If the input is streaming we disable body signing by default.220        if request.context.get('has_streaming_input', False):221            return False222 223        # If the S3-specific checks had no results, delegate to the generic224        # checks.225        return super()._should_sha256_sign_payload(request)226 227    def _should_add_content_sha256_header(self, explicit_payload):228        # Always add X-Amz-Content-SHA256 header229        return True230 231 232class CrtSigV4AsymAuth(BaseSigner):233    REQUIRES_REGION = True234    _PRESIGNED_HEADERS_BLOCKLIST = [235        'Authorization',236        'X-Amz-Date',237        'X-Amz-Content-SHA256',238        'X-Amz-Security-Token',239    ]240    _SIGNATURE_TYPE = awscrt.auth.AwsSignatureType.HTTP_REQUEST_HEADERS241    _USE_DOUBLE_URI_ENCODE = True242    _SHOULD_NORMALIZE_URI_PATH = True243 244    def __init__(self, credentials, service_name, region_name):245        self.credentials = credentials246        self._service_name = service_name247        self._region_name = region_name248        self._expiration_in_seconds = None249 250    def add_auth(self, request):251        if self.credentials is None:252            raise NoCredentialsError()253 254        # Use utcnow() because that's what gets mocked by tests, but set255        # timezone because CRT assumes naive datetime is local time.256        datetime_now = datetime.datetime.utcnow().replace(257            tzinfo=datetime.timezone.utc258        )259 260        # Use existing 'X-Amz-Content-SHA256' header if able261        existing_sha256 = self._get_existing_sha256(request)262 263        self._modify_request_before_signing(request)264 265        credentials_provider = awscrt.auth.AwsCredentialsProvider.new_static(266            access_key_id=self.credentials.access_key,267            secret_access_key=self.credentials.secret_key,268            session_token=self.credentials.token,269        )270 271        if self._is_streaming_checksum_payload(request):272            explicit_payload = STREAMING_UNSIGNED_PAYLOAD_TRAILER273        elif self._should_sha256_sign_payload(request):274            if existing_sha256:275                explicit_payload = existing_sha256276            else:277                explicit_payload = None  # to be calculated during signing278        else:279            explicit_payload = UNSIGNED_PAYLOAD280 281        if self._should_add_content_sha256_header(explicit_payload):282            body_header = (283                awscrt.auth.AwsSignedBodyHeaderType.X_AMZ_CONTENT_SHA_256284            )285        else:286            body_header = awscrt.auth.AwsSignedBodyHeaderType.NONE287 288        signing_config = awscrt.auth.AwsSigningConfig(289            algorithm=awscrt.auth.AwsSigningAlgorithm.V4_ASYMMETRIC,290            signature_type=self._SIGNATURE_TYPE,291            credentials_provider=credentials_provider,292            region=self._region_name,293            service=self._service_name,294            date=datetime_now,295            should_sign_header=self._should_sign_header,296            use_double_uri_encode=self._USE_DOUBLE_URI_ENCODE,297            should_normalize_uri_path=self._SHOULD_NORMALIZE_URI_PATH,298            signed_body_value=explicit_payload,299            signed_body_header_type=body_header,300            expiration_in_seconds=self._expiration_in_seconds,301        )302        crt_request = self._crt_request_from_aws_request(request)303        future = awscrt.auth.aws_sign_request(crt_request, signing_config)304        future.result()305        self._apply_signing_changes(request, crt_request)306 307    def _crt_request_from_aws_request(self, aws_request):308        url_parts = urlsplit(aws_request.url)309        crt_path = url_parts.path if url_parts.path else '/'310        if aws_request.params:311            array = []312            for param, value in aws_request.params.items():313                value = str(value)314                array.append(f'{param}={value}')315            crt_path = crt_path + '?' + '&'.join(array)316        elif url_parts.query:317            crt_path = f'{crt_path}?{url_parts.query}'318 319        crt_headers = awscrt.http.HttpHeaders(aws_request.headers.items())320 321        # CRT requires body (if it exists) to be an I/O stream.322        crt_body_stream = None323        if aws_request.body:324            if hasattr(aws_request.body, 'seek'):325                crt_body_stream = aws_request.body326            else:327                crt_body_stream = BytesIO(aws_request.body)328 329        crt_request = awscrt.http.HttpRequest(330            method=aws_request.method,331            path=crt_path,332            headers=crt_headers,333            body_stream=crt_body_stream,334        )335        return crt_request336 337    def _apply_signing_changes(self, aws_request, signed_crt_request):338        # Apply changes from signed CRT request to the AWSRequest339        aws_request.headers = HTTPHeaders.from_pairs(340            list(signed_crt_request.headers)341        )342 343    def _should_sign_header(self, name, **kwargs):344        return name.lower() not in SIGNED_HEADERS_BLACKLIST345 346    def _modify_request_before_signing(self, request):347        # This could be a retry. Make sure the previous348        # authorization headers are removed first.349        for h in self._PRESIGNED_HEADERS_BLOCKLIST:350            if h in request.headers:351                del request.headers[h]352        # If necessary, add the host header353        if 'host' not in request.headers:354            request.headers['host'] = _host_from_url(request.url)355 356    def _get_existing_sha256(self, request):357        return request.headers.get('X-Amz-Content-SHA256')358 359    def _is_streaming_checksum_payload(self, request):360        checksum_context = request.context.get('checksum', {})361        algorithm = checksum_context.get('request_algorithm')362        return isinstance(algorithm, dict) and algorithm.get('in') == 'trailer'363 364    def _should_sha256_sign_payload(self, request):365        # Payloads will always be signed over insecure connections.366        if not request.url.startswith('https'):367            return True368 369        # Certain operations may have payload signing disabled by default.370        # Since we don't have access to the operation model, we pass in this371        # bit of metadata through the request context.372        return request.context.get('payload_signing_enabled', True)373 374    def _should_add_content_sha256_header(self, explicit_payload):375        # only add X-Amz-Content-SHA256 header if payload is explicitly set376        return explicit_payload is not None377 378 379class CrtS3SigV4AsymAuth(CrtSigV4AsymAuth):380    # For S3, we do not normalize the path.381    _USE_DOUBLE_URI_ENCODE = False382    _SHOULD_NORMALIZE_URI_PATH = False383 384    def _get_existing_sha256(self, request):385        # always recalculate386        return None387 388    def _should_sha256_sign_payload(self, request):389        # S3 allows optional body signing, so to minimize the performance390        # impact, we opt to not SHA256 sign the body on streaming uploads,391        # provided that we're on https.392        client_config = request.context.get('client_config')393        s3_config = getattr(client_config, 's3', None)394 395        # The config could be None if it isn't set, or if the customer sets it396        # to None.397        if s3_config is None:398            s3_config = {}399 400        # The explicit configuration takes precedence over any implicit401        # configuration.402        sign_payload = s3_config.get('payload_signing_enabled', None)403        if sign_payload is not None:404            return sign_payload405 406        # We require that both content-md5 be present and https be enabled407        # to implicitly disable body signing. The combination of TLS and408        # content-md5 is sufficiently secure and durable for us to be409        # confident in the request without body signing.410        if (411            not request.url.startswith('https')412            or 'Content-MD5' not in request.headers413        ):414            return True415 416        # If the input is streaming we disable body signing by default.417        if request.context.get('has_streaming_input', False):418            return False419 420        # If the S3-specific checks had no results, delegate to the generic421        # checks.422        return super()._should_sha256_sign_payload(request)423 424    def _should_add_content_sha256_header(self, explicit_payload):425        # Always add X-Amz-Content-SHA256 header426        return True427 428 429class CrtSigV4AsymQueryAuth(CrtSigV4AsymAuth):430    DEFAULT_EXPIRES = 3600431    _SIGNATURE_TYPE = awscrt.auth.AwsSignatureType.HTTP_REQUEST_QUERY_PARAMS432 433    def __init__(434        self, credentials, service_name, region_name, expires=DEFAULT_EXPIRES435    ):436        super().__init__(credentials, service_name, region_name)437        self._expiration_in_seconds = expires438 439    def _modify_request_before_signing(self, request):440        super()._modify_request_before_signing(request)441 442        # We automatically set this header, so if it's the auto-set value we443        # want to get rid of it since it doesn't make sense for presigned urls.444        content_type = request.headers.get('content-type')445        if content_type == 'application/x-www-form-urlencoded; charset=utf-8':446            del request.headers['content-type']447 448        # Now parse the original query string to a dict, inject our new query449        # params, and serialize back to a query string.450        url_parts = urlsplit(request.url)451        # parse_qs makes each value a list, but in our case we know we won't452        # have repeated keys so we know we have single element lists which we453        # can convert back to scalar values.454        query_string_parts = parse_qs(url_parts.query, keep_blank_values=True)455        query_dict = {k: v[0] for k, v in query_string_parts.items()}456 457        # The spec is particular about this.  It *has* to be:458        # https://<endpoint>?<operation params>&<auth params>459        # You can't mix the two types of params together, i.e just keep doing460        # new_query_params.update(op_params)461        # new_query_params.update(auth_params)462        # percent_encode_sequence(new_query_params)463        if request.data:464            # We also need to move the body params into the query string. To465            # do this, we first have to convert it to a dict.466            query_dict.update(_get_body_as_dict(request))467            request.data = ''468        new_query_string = percent_encode_sequence(query_dict)469        # url_parts is a tuple (and therefore immutable) so we need to create470        # a new url_parts with the new query string.471        # <part>   - <index>472        # scheme   - 0473        # netloc   - 1474        # path     - 2475        # query    - 3  <-- we're replacing this.476        # fragment - 4477        p = url_parts478        new_url_parts = (p[0], p[1], p[2], new_query_string, p[4])479        request.url = urlunsplit(new_url_parts)480 481    def _apply_signing_changes(self, aws_request, signed_crt_request):482        # Apply changes from signed CRT request to the AWSRequest483        super()._apply_signing_changes(aws_request, signed_crt_request)484 485        signed_query = urlsplit(signed_crt_request.path).query486        p = urlsplit(aws_request.url)487        # urlsplit() returns a tuple (and therefore immutable) so we488        # need to create new url with the new query string.489        # <part>   - <index>490        # scheme   - 0491        # netloc   - 1492        # path     - 2493        # query    - 3  <-- we're replacing this.494        # fragment - 4495        aws_request.url = urlunsplit((p[0], p[1], p[2], signed_query, p[4]))496 497 498class CrtS3SigV4AsymQueryAuth(CrtSigV4AsymQueryAuth):499    """S3 SigV4A auth using query parameters.500    This signer will sign a request using query parameters and signature501    version 4A, i.e a "presigned url" signer.502    """503 504    # For S3, we do not normalize the path.505    _USE_DOUBLE_URI_ENCODE = False506    _SHOULD_NORMALIZE_URI_PATH = False507 508    def _should_sha256_sign_payload(self, request):509        # From the doc link above:510        # "You don't include a payload hash in the Canonical Request, because511        # when you create a presigned URL, you don't know anything about the512        # payload. Instead, you use a constant string "UNSIGNED-PAYLOAD".513        return False514 515    def _should_add_content_sha256_header(self, explicit_payload):516        # Never add X-Amz-Content-SHA256 header517        return False518 519 520class CrtSigV4QueryAuth(CrtSigV4Auth):521    DEFAULT_EXPIRES = 3600522    _SIGNATURE_TYPE = awscrt.auth.AwsSignatureType.HTTP_REQUEST_QUERY_PARAMS523 524    def __init__(525        self, credentials, service_name, region_name, expires=DEFAULT_EXPIRES526    ):527        super().__init__(credentials, service_name, region_name)528        self._expiration_in_seconds = expires529 530    def _modify_request_before_signing(self, request):531        super()._modify_request_before_signing(request)532 533        # We automatically set this header, so if it's the auto-set value we534        # want to get rid of it since it doesn't make sense for presigned urls.535        content_type = request.headers.get('content-type')536        if content_type == 'application/x-www-form-urlencoded; charset=utf-8':537            del request.headers['content-type']538 539        # Now parse the original query string to a dict, inject our new query540        # params, and serialize back to a query string.541        url_parts = urlsplit(request.url)542        # parse_qs makes each value a list, but in our case we know we won't543        # have repeated keys so we know we have single element lists which we544        # can convert back to scalar values.545        query_dict = {546            k: v[0]547            for k, v in parse_qs(548                url_parts.query, keep_blank_values=True549            ).items()550        }551        if request.params:552            query_dict.update(request.params)553            request.params = {}554        # The spec is particular about this.  It *has* to be:555        # https://<endpoint>?<operation params>&<auth params>556        # You can't mix the two types of params together, i.e just keep doing557        # new_query_params.update(op_params)558        # new_query_params.update(auth_params)559        # percent_encode_sequence(new_query_params)560        if request.data:561            # We also need to move the body params into the query string. To562            # do this, we first have to convert it to a dict.563            query_dict.update(_get_body_as_dict(request))564            request.data = ''565        new_query_string = percent_encode_sequence(query_dict)566        # url_parts is a tuple (and therefore immutable) so we need to create567        # a new url_parts with the new query string.568        # <part>   - <index>569        # scheme   - 0570        # netloc   - 1571        # path     - 2572        # query    - 3  <-- we're replacing this.573        # fragment - 4574        p = url_parts575        new_url_parts = (p[0], p[1], p[2], new_query_string, p[4])576        request.url = urlunsplit(new_url_parts)577 578    def _apply_signing_changes(self, aws_request, signed_crt_request):579        # Apply changes from signed CRT request to the AWSRequest580        super()._apply_signing_changes(aws_request, signed_crt_request)581 582        signed_query = urlsplit(signed_crt_request.path).query583        p = urlsplit(aws_request.url)584        # urlsplit() returns a tuple (and therefore immutable) so we585        # need to create new url with the new query string.586        # <part>   - <index>587        # scheme   - 0588        # netloc   - 1589        # path     - 2590        # query    - 3  <-- we're replacing this.591        # fragment - 4592        aws_request.url = urlunsplit((p[0], p[1], p[2], signed_query, p[4]))593 594 595class CrtS3SigV4QueryAuth(CrtSigV4QueryAuth):596    """S3 SigV4 auth using query parameters.597    This signer will sign a request using query parameters and signature598    version 4, i.e a "presigned url" signer.599    Based off of:600    http://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-query-string-auth.html601    """602 603    # For S3, we do not normalize the path.604    _USE_DOUBLE_URI_ENCODE = False605    _SHOULD_NORMALIZE_URI_PATH = False606 607    def _should_sha256_sign_payload(self, request):608        # From the doc link above:609        # "You don't include a payload hash in the Canonical Request, because610        # when you create a presigned URL, you don't know anything about the611        # payload. Instead, you use a constant string "UNSIGNED-PAYLOAD".612        return False613 614    def _should_add_content_sha256_header(self, explicit_payload):615        # Never add X-Amz-Content-SHA256 header616        return False617 618 619# Defined at the bottom of module to ensure all Auth620# classes are defined.621CRT_AUTH_TYPE_MAPS = {622    'v4': CrtSigV4Auth,623    'v4-query': CrtSigV4QueryAuth,624    'v4a': CrtSigV4AsymAuth,625    's3v4': CrtS3SigV4Auth,626    's3v4-query': CrtS3SigV4QueryAuth,627    's3v4a': CrtS3SigV4AsymAuth,628    's3v4a-query': CrtS3SigV4AsymQueryAuth,629}630