def _compute_downstream_headers()

in aws_jupyter_proxy/awsproxy.py [0:0]

• 15 lines of code
• 4 McCabe index (conditional complexity)

    def _compute_downstream_headers(self, downstream_request_path) -> HTTPHeaders:
        """
        1. Copy original headers apart from blacklisted ones
        2. Add the Host header based on the service model
        3. Add a security token header if the current session is using temporary credentials
        4. Add the SigV4 Authorization header.

        :param downstream_request_path: the URL path for the downstream service request
        :return: the headers to pass to the downstream request
        """
        downstream_request_headers = self.upstream_request.headers.copy()
        for blacklisted_request_header in self.BLACKLISTED_REQUEST_HEADERS:
            try:
                del downstream_request_headers[blacklisted_request_header]
            except KeyError:
                pass

        base_service_url = urlparse(self.service_info.endpoint_url)
        downstream_request_headers["Host"] = base_service_url.netloc

        if self.credentials.token:
            downstream_request_headers["X-Amz-Security-Token"] = self.credentials.token

        downstream_request_headers["Authorization"] = self._sigv4_auth_header(
            downstream_request_path
        )
        return downstream_request_headers