R/custom_s3.R

In paws.common: Paws Low-Level Amazon Web Services API

#' @include service.R
#' @include util.R
#' @include error.R
#' @include head_bucket.R
NULL

VERSION_ID_SUFFIX <- "?versionId="
ACCESSPOINT_ARN <- paste0(
  "^arn:(aws).*:(s3|s3-object-lambda):[a-z\\-0-9]*:[0-9]{12}:accesspoint[/:]",
  "[a-zA-Z0-9\\-.]{1,63}$"
)
OUTPOST_ARN <- paste0(
  "^arn:(aws).*:s3-outposts:[a-z\\-0-9]+:[0-9]{12}:outpost[/:]",
  "[a-zA-Z0-9\\-]{1,63}[/:]accesspoint[/:][a-zA-Z0-9\\-]{1,63}$"
)

VALID_S3_ARN <- paste(ACCESSPOINT_ARN, OUTPOST_ARN, sep = "|")

################################################################################

convert_file_to_raw <- function(request) {
  operation_name <- request$operation$name
  if (operation_name != "PutObject") {
    return(request)
  }

  request_params <- request$params
  content_body <- request_params["Body"][[1]]
  if (!is.character(content_body)) {
    return(request)
  }

  file_name <- content_body[[1]]
  if (!file.exists(file_name)) {
    stopf("Unable to find file: %s", file_name)
  }
  file_connection <- file(file_name, "rb")
  raw_body <- readBin(file_connection, "raw", n = file.size(file_name))
  close(file_connection)

  attributes(raw_body) <- attributes(content_body)
  request$params["Body"][[1]] <- raw_body
  return(request)
}

################################################################################
# host_compatible_bucket_name returns true if the request should
# put the bucket in the host. This is false if S3ForcePathStyle is
# explicitly set or if the bucket is not DNS compatible.
host_compatible_bucket_name <- function(url, bucket) {
  # Bucket might be DNS compatible but dots in the hostname will fail
  # certificate validation, so do not use host-style.
  if (url$scheme == "https" && grepl(".", bucket, fixed = TRUE)) {
    return(FALSE)
  }
  return(dns_compatible_bucket_name(bucket))
}

# dns_compatible_bucket_name returns true if the bucket name is DNS compatible.
# Buckets created outside of the classic region MUST be DNS compatible.
dns_compatible_bucket_name <- function(bucket) {
  domain <- "^[a-z0-9][a-z0-9\\.\\-]{1,61}[a-z0-9]$"
  ip_address <- "^(\\d+\\.){3}\\d+$"
  return(
    grepl(domain, bucket) &&
      !grepl(ip_address, bucket) &&
      !grepl("..", bucket, fixed = TRUE)
  )
}

move_bucket_to_host <- function(url, bucket) {
  url[["host"]] <- paste0(bucket, ".", url[["host"]])
  url[["path"]] <- gsub("/\\{Bucket\\}", "", url$path)

  if (url[["path"]] == "") {
    url[["path"]] <- "/"
  }
  return(url)
}

# Check if a given bucket name is an S3 access point.
is_access_point <- function(bucket_name) {
  regex <- "^(.+):(.+):(.+):(.+):(.+):(.+)$"
  return(grepl(regex, bucket_name))
}

# Parse the S3 access point ARN and return the corresponding endpoint.
# See https://docs.aws.amazon.com/AmazonS3/latest/userguide/using-access-points.html
#
# ARN format: arn:aws:s3:{region}:{account-id}:accesspoint/{accesspoint-name}
# Access point endpoint format: {accesspoint-name}-{account-id}.s3-accesspoint.{region}.{dns-suffix}
# Example:
#   In: arn:aws:s3:us-west-2:123456789012:accesspoint/test
#   Out: test-123456789012.s3-accesspoint.us-west-2.amazonaws.com
get_access_point_endpoint <- function(access_point) {
  part <- strsplit(access_point, ":|/")[[1]]
  region <- part[4]
  account <- part[5]
  name <- part[7]
  endpoint <- sprintf("%s-%s.s3-accesspoint.%s.amazonaws.com", name, account, region)
  return(endpoint)
}

remove_bucket_from_url <- function(url) {
  url$path <- gsub("\\{Bucket+?\\}", "", url$path)
  if (url$path == "") {
    url$path <- "/"
  }
  return(url)
}

update_endpoint_for_s3_config <- function(request) {
  if (is.null(bucket_name <- request$params[["Bucket"]])) {
    return(request)
  }

  if (is_access_point(bucket_name)) {
    request$http_request$url[["host"]] <- get_access_point_endpoint(bucket_name)
    request$http_request$url <- remove_bucket_from_url(request$http_request$url)
    return(request)
  }

  if (!host_compatible_bucket_name(request$http_request$url, bucket_name)) {
    return(request)
  }

  if (request$operation$name %in% c("GetBucketLocation")) {
    return(request)
  }

  use_virtual_host_style <- TRUE
  if (request$config$s3_force_path_style) {
    use_virtual_host_style <- FALSE
  }
  if (request$client_info$custom_endpoint && isFALSE(request$config$s3_virtual_address)) {
    use_virtual_host_style <- FALSE
  }

  if (use_virtual_host_style) {
    request$http_request$url <- move_bucket_to_host(request$http_request$url, bucket_name)
  }

  return(request)
}

################################################################################

populate_location_constraint <- function(request) {
  operation_name <- request$operation$name

  if (operation_name != "CreateBucket") {
    return(request)
  }

  request_params <- request$params
  location <- request_params$CreateBucketConfiguration$LocationConstraint

  if (length(location) == 0) {
    request$params[["CreateBucketConfiguration"]] <- NULL
  }

  return(request)
}

################################################################################
# developed from boto3:
# https://github.com/boto/botocore/blob/54a09c7d025181b8221d0046eb6dd6def9ace338/botocore/handlers.py#L287-L321

sse_md5_build <- function(request) {
  request$params <- sse_md5(request$params)
  request$params <- copy_source_sse_md5(request$params)
  return(request)
}

# S3 server-side encryption requires the encryption key to be sent to the
# server base64 encoded, as well as a base64-encoded MD5 hash of the
# encryption key. This handler does both if the MD5 has not been set by
# the caller.
sse_md5 <- function(params) {
  return(.sse_md5(params, "SSECustomer"))
}

# S3 server-side encryption requires the encryption key to be sent to the
# server base64 encoded, as well as a base64-encoded MD5 hash of the
# encryption key. This handler does both if the MD5 has not been set by
# the caller specifically if the parameter is for the copy-source sse-c key.
copy_source_sse_md5 <- function(params) {
  return(.sse_md5(params, "CopySourceSSECustomer"))
}

.sse_md5 <- function(params, sse_member_prefix = "SSECustomer") {
  if (!.needs_s3_sse_customization(params, sse_member_prefix)) {
    return(params)
  }
  sse_key_member <- paste0(sse_member_prefix, "Key")
  sse_md5_member <- paste0(sse_member_prefix, "KeyMD5")
  key_md5_str <- base64enc::base64encode(digest::digest(
    params[[sse_key_member]],
    serialize = FALSE,
    raw = TRUE
  ))
  attributes(key_md5_str) <- attributes(params[[sse_md5_member]])
  params[[sse_md5_member]] <- key_md5_str
  return(params)
}

.needs_s3_sse_customization <- function(params, sse_member_prefix) {
  return(
    !is_empty(params[[paste0(sse_member_prefix, "Key")]]) &
      is_empty(params[[paste0(sse_member_prefix, "KeyMD5")]])
  )
}

################################################################################

content_md5 <- function(request) {
  operation_name <- request$operation$name
  if (
    !(operation_name %in%
      c(
        "PutBucketCors",
        "PutBucketLifecycle",
        "PutBucketPolicy",
        "PutBucketTagging",
        "DeleteObjects",
        "PutBucketLifecycleConfiguration",
        "PutBucketReplication",
        "PutObject",
        "UploadPart"
      ))
  ) {
    return(request)
  }
  # Create Content-MD5 header if missing.
  # https://github.com/aws/aws-sdk-go/blob/e2d6cb448883e4f4fcc5246650f89bde349041ec/private/checksum/content_md5.go#L18
  if (is.null(request$http_request$header[["Content-MD5"]])) {
    body <- request$body
    if (length(body) == 0) {
      body <- raw(0)
    }
    hash <- digest::digest(body, serialize = FALSE, raw = TRUE)
    base64_hash <- base64enc::base64encode(hash)
    request$http_request$header$`Content-Md5` <- base64_hash
  }
  return(request)
}

################################################################################

s3_unmarshal_get_bucket_location <- function(request) {
  if (request$operation$name != "GetBucketLocation") {
    return(request)
  }
  response <- decode_xml(request$http_response$body)
  data <- request$data
  location <- response$LocationConstraint
  if (length(location) == 0) {
    location <- "us-east-1"
  } else {
    location <- location[[1]]
  }
  if (location == "EU") {
    location <- "eu-west-1"
  }
  data$LocationConstraint <- location
  request$data <- data
  return(request)
}

################################################################################

s3_unmarshal_error <- function(request) {
  request$http_response$body <- get_connection_error(
    request$http_response$body,
    request$operation$stream_api
  )
  data <- tryCatch(decode_xml(request$http_response$body), error = function(e) NULL)
  # Bucket exists in a different region, and request needs
  # to be made to the correct region.
  if (request$http_response$status_code == 301) {
    error_msg <- list()
    error_msg[[1]] <- sprintf(
      "incorrect region, the bucket is not in '%s' region at endpoint '%s'",
      request$config$region,
      request$config$endpoint
    )
    if (nzchar(v <- request$http_response$header[["x-amz-bucket-region"]] %||% "")) {
      error_msg[[2]] <- sprintf(", bucket is in '%s' region", v)
    }
    request$error <- Error(
      "BucketRegionError",
      paste(error_msg, collapse = ""),
      request$http_response$status_code,
      request$request_id
    )
    return(request)
  }
  if (is.null(data)) {
    request$error <- serialization_error(request)
    return(request)
  }

  error_response <- lapply(data$Error, unlist)
  code <- error_response$Code
  message <- error_response$Message

  if (is.null(message) && is.null(code)) {
    request$error <- serialization_error(request)
    return(request)
  }

  request$error <- Error(code, message, request$http_response$status_code, error_response)
  return(request)
}

################################################################################
# Developed from botocore S3RegionRedirectorv2:
# https://github.com/boto/botocore/blob/de6dfccdb68e70005ed2a73dc5d04bc1e97f0541/botocore/utils.py#L1509

# An S3 request sent to the wrong region will return an error that
# contains the endpoint the request should be sent to. This handler
# will add the redirect information to the signing context and then
# redirect the request.
s3_redirect_from_error <- function(request) {
  if (is.null(request$http_response)) {
    return(request)
  }
  if (isTRUE(request$context$s3_redirect)) {
    log_debug("S3 request was previously redirected, not redirecting.")
    return(request)
  }
  error_code <- request$http_response$status_code
  # Exit s3_redirect_from_error function if initial request is successful
  # https://docs.aws.amazon.com/waf/latest/developerguide/customizing-the-response-status-codes.html
  http_success_code <- c(200, 201, 202, 204, 206)
  if (error_code %in% http_success_code) {
    return(request)
  }
  request$http_response$body <- get_connection_error(
    request$http_response$body,
    request$operation$stream_api
  )
  error <- decode_xml(request$http_response$body)$Error
  if (!can_be_redirected(request, error_code, error)) {
    return(request)
  }
  bucket_name <- request$params[["Bucket"]]
  new_region <- s3_get_bucket_region(request, error, bucket_name)
  if (is.null(new_region)) {
    log_debug(
      paste(
        "S3 client configured for region %s but the bucket %s is not",
        "in that region and the proper region could not be",
        "automatically determined."
      ),
      request$client_info$signing_region,
      bucket_name
    )
    return(request)
  }
  log_debug(
    paste(
      "S3 client configured for region %s but the bucket %s is in region",
      "%s; Please configure the proper region to avoid multiple",
      "unnecessary redirects and signing attempts."
    ),
    request$client_info$signing_region,
    bucket_name,
    new_region
  )
  # Update client_info for redirect
  request$client_info$signing_region <- new_region
  # Re-resolve endpoint with new region and modify request_dict with
  # the new URL, auth scheme, and signing context.
  ep_info <- resolver_endpoint(
    service = "s3",
    region = new_region,
    endpoints = .s3_endpoint()
  )
  request$client_info$endpoint <- set_request_url(
    request$client_info$endpoint,
    ep_info$endpoint
  )
  request$http_request$url <- parse_url(paste0(
    request$client_info$endpoint,
    request$operation$http_path
  ))
  request$built <- FALSE
  request$context$s3_redirect <- TRUE
  # re-sign redirect request
  request <- sign(request)
  # re-send redirect request
  request <- send(request)

  return(request)
}

can_be_redirected <- function(request, error_code, error) {
  # We have to account for 400 responses because
  # if we sign a Head* request with the wrong region,
  # we'll get a 400 Bad Request but we won't get a
  # body saying it's an "AuthorizationHeaderMalformed".
  is_special_head_object <- (error_code %in%
    c("301", "400") &
    request$operation$name == "HeadObject")
  is_special_head_bucket <- (error_code %in%
    c("301", "400") &
    request$operation$name == "HeadBucket" &
    "x-amz-bucket-region" %in% names(request$http_response$header))
  is_wrong_signing_region <- (error$Code == "AuthorizationHeaderMalformed" &
    "Region" %in% names(error))
  is_redirect_status <- request$http_response$status_code %in% c(301, 302, 307)
  is_permanent_redirect <- error$Code == "PermanentRedirect"

  return(any(c(
    is_special_head_object,
    is_wrong_signing_region,
    is_permanent_redirect,
    is_special_head_bucket,
    is_redirect_status
  )))
}

# There are multiple potential sources for the new region to redirect to,
# but they aren't all universally available for use. This will try to
# find region from response elements, but will fall back to calling
# HEAD on the bucket if all else fails.
# param response: HttpResponse
# param error: Error
s3_get_bucket_region <- function(request, error, bucket) {
  # First try to source the region from the headers.
  response_headers <- request$http_response$header
  if (!is.null(region <- response_headers[["x-amz-bucket-region"]])) {
    return(unlist(region))
  }
  if (!is.null(region <- unlist(error$Region))) {
    return(region)
  }

  # Finally, HEAD the bucket. No other choice sadly.
  resp <- s3(request$config)$head_bucket(Bucket = bucket)
  return(resp$BucketRegion)
}

# Splice a new endpoint into an existing URL. Note that some endpoints
# from the endpoint provider have a path component which will be
# discarded by this function.
set_request_url <- function(original_endpoint, new_endpoint, use_new_scheme = TRUE) {
  new_endpoint_components <- parse_url(new_endpoint)
  final_endpoint_components <- parse_url(original_endpoint)
  scheme <- final_endpoint_components$scheme
  if (use_new_scheme) {
    scheme <- new_endpoint_components[["scheme"]]
  }
  path <- (if (final_endpoint_components[["path"]] == "/") {
    ""
  } else {
    final_endpoint_components[["path"]]
  })
  final_endpoint_components[["host"]] <- new_endpoint_components$host
  final_endpoint_components[["scheme"]] <- scheme
  final_endpoint_components[["path"]] <- path
  return(build_url(final_endpoint_components))
}

################################################################################
handle_copy_source_param <- function(request) {
  if (
    !(request$operation$name %in% c("CopyObject", "CopyPart")) |
      isTRUE(request$context$s3_redirect)
  ) {
    return(request)
  }
  source <- request$params$CopySource
  tags <- tag_get_all(source)
  if (is.character(source)) {
    request$params$CopySource <- quote_source_header(source, tags)
  } else if (is.list(source)) {
    request$params$CopySource <- quote_source_header_from_list(source, tags)
  }
  return(request)
}

quote_source_header <- function(source, tags) {
  result <- strsplit(source, VERSION_ID_SUFFIX, fixed = TRUE)[[1]]
  if (is.na(result[2])) {
    return(tag_add(paws_url_encoder(result[1], "/"), tags))
  } else {
    return(tag_add(
      paste0(paws_url_encoder(result[1], "/"), VERSION_ID_SUFFIX, result[2]),
      tags
    ))
  }
}

quote_source_header_from_list <- function(source, tags) {
  if (is.null(bucket <- source[["Bucket"]])) {
    stopf("CopySource list is missing required parameter: Bucket")
  }
  if (is.null(key <- source[["Key"]])) {
    stopf("CopySource list is missing required parameter: Key")
  }
  if (grepl(VALID_S3_ARN, bucket, perl = TRUE)) {
    final <- sprintf("%s/object/%s", bucket, key)
  } else {
    final <- sprintf("%s/%s", bucket, key)
  }
  final <- paws_url_encoder(final, "/")
  if (!is.null(version_id <- source[["VersionId"]])) {
    final <- paste0(final, VERSION_ID_SUFFIX, version_id)
  }
  return(tag_add(final, tags))
}

################################################################################

customizations$s3 <- function(handlers) {
  handlers$build <- handlers_add_front(handlers$build, update_endpoint_for_s3_config)
  handlers$build <- handlers_add_front(handlers$build, populate_location_constraint)
  handlers$build <- handlers_add_front(handlers$build, convert_file_to_raw)
  handlers$build <- handlers_add_front(handlers$build, handle_copy_source_param)
  handlers$build <- handlers_add_front(handlers$build, sse_md5_build)
  handlers$build <- handlers_add_back(handlers$build, content_md5)
  handlers$send <- handlers_add_back(handlers$send, s3_redirect_from_error)
  handlers$unmarshal <- handlers_add_back(
    handlers$unmarshal,
    s3_unmarshal_get_bucket_location
  )
  handlers$unmarshal_error <- handlers_set(s3_unmarshal_error)
  handlers
}