def _get_signed_canonical_headers()

in aws_jupyter_proxy/awsproxy.py [0:0]

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

    def _get_signed_canonical_headers(self) -> Tuple[str, str]:
        canonical_headers = {}

        for signed_header in self.upstream_auth_info.signed_headers:
            canonical_headers[signed_header] = self.upstream_request.headers[
                signed_header
            ]

        base_service_url = urlparse(self.service_info.endpoint_url)
        canonical_headers["host"] = base_service_url.netloc
        if self.credentials.token:
            canonical_headers["x-amz-security-token"] = self.credentials.token

        canonical_headers_string = "\n".join(
            [
                f"{canonical_header}:{canonical_headers[canonical_header]}"
                for canonical_header in sorted(canonical_headers)
            ]
        )
        canonical_headers_string += "\n"
        signed_headers = ";".join(sorted(canonical_headers))

        return signed_headers, canonical_headers_string