(self, request)
| 627 | self._expires = expires |
| 628 | |
| 629 | def _modify_request_before_signing(self, request): |
| 630 | # We automatically set this header, so if it's the auto-set value we |
| 631 | # want to get rid of it since it doesn't make sense for presigned urls. |
| 632 | content_type = request.headers.get('content-type') |
| 633 | blocklisted_content_type = ( |
| 634 | 'application/x-www-form-urlencoded; charset=utf-8' |
| 635 | ) |
| 636 | if content_type == blocklisted_content_type: |
| 637 | del request.headers['content-type'] |
| 638 | |
| 639 | # Note that we're not including X-Amz-Signature. |
| 640 | # From the docs: "The Canonical Query String must include all the query |
| 641 | # parameters from the preceding table except for X-Amz-Signature. |
| 642 | signed_headers = self.signed_headers(self.headers_to_sign(request)) |
| 643 | |
| 644 | auth_params = { |
| 645 | 'X-Amz-Algorithm': 'AWS4-HMAC-SHA256', |
| 646 | 'X-Amz-Credential': self.scope(request), |
| 647 | 'X-Amz-Date': request.context['timestamp'], |
| 648 | 'X-Amz-Expires': self._expires, |
| 649 | 'X-Amz-SignedHeaders': signed_headers, |
| 650 | } |
| 651 | if self.credentials.token is not None: |
| 652 | auth_params['X-Amz-S3session-Token'] = self.credentials.token |
| 653 | # Now parse the original query string to a dict, inject our new query |
| 654 | # params, and serialize back to a query string. |
| 655 | url_parts = urlsplit(request.url) |
| 656 | # parse_qs makes each value a list, but in our case we know we won't |
| 657 | # have repeated keys so we know we have single element lists which we |
| 658 | # can convert back to scalar values. |
| 659 | query_string_parts = parse_qs(url_parts.query, keep_blank_values=True) |
| 660 | query_dict = {k: v[0] for k, v in query_string_parts.items()} |
| 661 | |
| 662 | if request.params: |
| 663 | query_dict.update(request.params) |
| 664 | request.params = {} |
| 665 | # The spec is particular about this. It *has* to be: |
| 666 | # https://<endpoint>?<operation params>&<auth params> |
| 667 | # You can't mix the two types of params together, i.e just keep doing |
| 668 | # new_query_params.update(op_params) |
| 669 | # new_query_params.update(auth_params) |
| 670 | # percent_encode_sequence(new_query_params) |
| 671 | operation_params = '' |
| 672 | if request.data: |
| 673 | # We also need to move the body params into the query string. To |
| 674 | # do this, we first have to convert it to a dict. |
| 675 | query_dict.update(_get_body_as_dict(request)) |
| 676 | request.data = '' |
| 677 | if query_dict: |
| 678 | operation_params = percent_encode_sequence(query_dict) + '&' |
| 679 | new_query_string = ( |
| 680 | f"{operation_params}{percent_encode_sequence(auth_params)}" |
| 681 | ) |
| 682 | # url_parts is a tuple (and therefore immutable) so we need to create |
| 683 | # a new url_parts with the new query string. |
| 684 | # <part> - <index> |
| 685 | # scheme - 0 |
| 686 | # netloc - 1 |
nothing calls this directly
no test coverage detected