Function resolve_request_checksum_algorithm

awscli/botocore/httpchecksum.py:304–371

(
    request,
    operation_model,
    params,
    supported_algorithms=None,
)

def resolve_request_checksum_algorithm(
    request,
    operation_model,
    params,
    supported_algorithms=None,
):
    # If the header is already set by the customer, skip calculation
    if has_checksum_header(request):
        return

    checksum_context = request["context"].get("checksum", {})
    request_checksum_calculation = request["context"][
        "client_config"
    ].request_checksum_calculation
    http_checksum = operation_model.http_checksum
    request_checksum_required = (
        operation_model.http_checksum_required
        or http_checksum.get("requestChecksumRequired")
    )
    algorithm_member = http_checksum.get("requestAlgorithmMember")
    if algorithm_member and algorithm_member in params:
        # If the client has opted into using flexible checksums and the
        # request supports it, use that instead of checksum required
        if supported_algorithms is None:
            supported_algorithms = _SUPPORTED_CHECKSUM_ALGORITHMS

        algorithm_name = params[algorithm_member].lower()
        if algorithm_name not in supported_algorithms:
            raise FlexibleChecksumError(
                error_msg=f"Unsupported checksum algorithm: {algorithm_name}"
            )
    elif request_checksum_required or (
        algorithm_member and request_checksum_calculation == "when_supported"
    ):
        # Don't use a default checksum for presigned requests.
        if request["context"].get("is_presign_request"):
            return
        algorithm_name = DEFAULT_CHECKSUM_ALGORITHM.lower()
        algorithm_member_header = _get_request_algorithm_member_header(
            operation_model, request, algorithm_member
        )
        if algorithm_member_header is not None:
            checksum_context["request_algorithm_header"] = {
                "name": algorithm_member_header,
                "value": DEFAULT_CHECKSUM_ALGORITHM,
            }
    else:
        return

    location_type = "header"
    if (
        operation_model.has_streaming_input
        and urlparse(request["url"]).scheme == "https"
    ):
        if request["context"]["client_config"].signature_version != 's3':
            # Operations with streaming input must support trailers.
            # We only support unsigned trailer checksums currently. As this
            # disables payload signing we'll only use trailers over TLS.