HDDS-12935. Support unsigned chunked upload with STREAMING-UNSIGNED-PAYLOAD-TRAILER (apache#8366)

(cherry picked from commit e9e8b30)

Author: ivandika3

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java

@@ -29,6 +29,7 @@
 
 import com.amazonaws.AmazonServiceException;
 import com.amazonaws.AmazonServiceException.ErrorType;
+import com.amazonaws.HttpMethod;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.model.AbortMultipartUploadRequest;
 import com.amazonaws.services.s3.model.AccessControlList;
@@ -37,6 +38,7 @@
 import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
 import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
 import com.amazonaws.services.s3.model.CreateBucketRequest;
+import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest;
 import com.amazonaws.services.s3.model.GetObjectRequest;
 import com.amazonaws.services.s3.model.Grantee;
 import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
@@ -69,17 +71,22 @@
 import com.amazonaws.services.s3.transfer.TransferManagerBuilder;
 import com.amazonaws.services.s3.transfer.Upload;
 import com.amazonaws.services.s3.transfer.model.UploadResult;
+import com.amazonaws.util.IOUtils;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -955,6 +962,42 @@ public void testQuotaExceeded() throws IOException {
     assertEquals("QuotaExceeded", ase.getErrorCode());
   }
 
+  @Test
+  public void testPresignedUrlGet() throws IOException {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    final String content = "bar";
+    s3Client.createBucket(bucketName);
+
+    InputStream is = new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8));
+
+    s3Client.putObject(bucketName, keyName, is, new ObjectMetadata());
+
+    // Set the presigned URL to expire after one hour.
+    Date expiration = Date.from(Instant.now().plusMillis(1000 * 60 * 60));
+
+    // Generate the presigned URL
+    GeneratePresignedUrlRequest generatePresignedUrlRequest =
+        new GeneratePresignedUrlRequest(bucketName, keyName)
+            .withMethod(HttpMethod.GET)
+            .withExpiration(expiration);
+    generatePresignedUrlRequest.addRequestParameter("x-custom-parameter", "custom-value");
+    URL url = s3Client.generatePresignedUrl(generatePresignedUrlRequest);
+
+    // Download the object using HttpUrlConnection (since v1.1)
+    // Capture the response body to a byte array.
+    URL presignedUrl = new URL(url.toExternalForm());
+    HttpURLConnection connection = (HttpURLConnection) presignedUrl.openConnection();
+    connection.setRequestMethod("GET");
+    // Download the result of executing the request.
+    try (InputStream s3is = connection.getInputStream();
+         ByteArrayOutputStream bos = new ByteArrayOutputStream(
+             content.getBytes(StandardCharsets.UTF_8).length)) {
+      IOUtils.copy(s3is, bos);
+      assertEquals(content, bos.toString("UTF-8"));
+    }
+  }
+
   private boolean isBucketEmpty(Bucket bucket) {
     ObjectListing objectListing = s3Client.listObjects(bucket.getName());
     return objectListing.getObjectSummaries().isEmpty();

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java

@@ -23,13 +23,18 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.RandomAccessFile;
+import java.net.HttpURLConnection;
+import java.net.URL;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -48,20 +53,33 @@
 import org.junit.jupiter.api.io.TempDir;
 import software.amazon.awssdk.core.ResponseBytes;
 import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.http.HttpExecuteRequest;
+import software.amazon.awssdk.http.HttpExecuteResponse;
+import software.amazon.awssdk.http.SdkHttpClient;
+import software.amazon.awssdk.http.SdkHttpMethod;
+import software.amazon.awssdk.http.SdkHttpRequest;
+import software.amazon.awssdk.http.apache.ApacheHttpClient;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
 import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
 import software.amazon.awssdk.services.s3.model.CompletedPart;
 import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
 import software.amazon.awssdk.services.s3.model.CopyObjectResponse;
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 import software.amazon.awssdk.services.s3.model.Tag;
 import software.amazon.awssdk.services.s3.model.Tagging;
 import software.amazon.awssdk.services.s3.model.UploadPartRequest;
 import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+import software.amazon.awssdk.services.s3.presigner.S3Presigner;
+import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest;
+import software.amazon.awssdk.services.s3.presigner.model.PresignedGetObjectRequest;
+import software.amazon.awssdk.utils.IoUtils;
 
 /**
  * This is an abstract class to test the AWS Java S3 SDK operations.
@@ -195,6 +213,78 @@ public void testLowLevelMultipartUpload(@TempDir Path tempDir) throws Exception
     assertEquals(userMetadata, headObjectResponse.metadata());
   }
 
+  @Test
+  public void testPresignedUrlGet() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    final String content = "bar";
+    s3Client.createBucket(b -> b.bucket(bucketName));
+
+    s3Client.putObject(b -> b
+            .bucket(bucketName)
+            .key(keyName),
+        RequestBody.fromString(content));
+
+    try (S3Presigner presigner = S3Presigner.builder()
+        // TODO: Find a way to retrieve the path style configuration from S3Client instead
+        .serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build())
+        .endpointOverride(s3Client.serviceClientConfiguration().endpointOverride().get())
+        .region(s3Client.serviceClientConfiguration().region())
+        .credentialsProvider(s3Client.serviceClientConfiguration().credentialsProvider()).build()) {
+      GetObjectRequest objectRequest = GetObjectRequest.builder()
+          .bucket(bucketName)
+          .key(keyName)
+          .build();
+
+      GetObjectPresignRequest presignRequest = GetObjectPresignRequest.builder()
+          .signatureDuration(Duration.ofMinutes(10))  // The URL will expire in 10 minutes.
+          .getObjectRequest(objectRequest)
+          .build();
+
+      PresignedGetObjectRequest presignedRequest = presigner.presignGetObject(presignRequest);
+
+      // Download the object using HttpUrlConnection (since v1.1)
+      // Capture the response body to a byte array.
+      URL presignedUrl = presignedRequest.url();
+      HttpURLConnection connection = (HttpURLConnection) presignedUrl.openConnection();
+      connection.setRequestMethod("GET");
+      // Download the result of executing the request.
+      try (InputStream s3is = connection.getInputStream();
+           ByteArrayOutputStream bos = new ByteArrayOutputStream(
+               content.getBytes(StandardCharsets.UTF_8).length)) {
+        IoUtils.copy(s3is, bos);
+        assertEquals(content, bos.toString("UTF-8"));
+      }
+
+      // Use the AWS SDK for Java SdkHttpClient class to do the download
+      SdkHttpRequest request = SdkHttpRequest.builder()
+          .method(SdkHttpMethod.GET)
+          .uri(presignedUrl.toURI())
+          .build();
+
+      HttpExecuteRequest executeRequest = HttpExecuteRequest.builder()
+          .request(request)
+          .build();
+
+      try (SdkHttpClient sdkHttpClient = ApacheHttpClient.create();
+           ByteArrayOutputStream bos = new ByteArrayOutputStream(
+               content.getBytes(StandardCharsets.UTF_8).length)) {
+        HttpExecuteResponse response = sdkHttpClient.prepareRequest(executeRequest).call();
+        assertTrue(response.responseBody().isPresent(), () -> "The presigned url download request " +
+            "should have a response body");
+        response.responseBody().ifPresent(
+            abortableInputStream -> {
+              try {
+                IoUtils.copy(abortableInputStream, bos);
+              } catch (IOException e) {
+                throw new RuntimeException(e);
+              }
+            });
+        assertEquals(content, bos.toString("UTF-8"));
+      }
+    }
+  }
+
   private String getBucketName() {
     return getBucketName(null);
   }

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/SignedChunksInputStream.java

@@ -17,21 +17,57 @@
 
 package org.apache.hadoop.ozone.s3;
 
+import static org.apache.hadoop.ozone.s3.util.S3Utils.eol;
+
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Objects;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 /**
- * Input stream implementation to read body with chunked signatures. This should also work
+ * Input stream implementation to read body of a signed chunked upload. This should also work
  * with the chunked payloads with trailer.
  *
+ * <p>
+ * Example chunk data:
+ * <pre>
+ * 10000;chunk-signature=b474d8862b1487a5145d686f57f013e54db672cee1c953b3010fb58501ef5aa2\r\n
+ * &lt;65536-bytes&gt;\r\n
+ * 400;chunk-signature=1c1344b170168f8e65b41376b44b20fe354e373826ccbbe2c1d40a8cae51e5c7\r\n
+ * &lt;1024-bytes&gt;\r\n
+ * 0;chunk-signature=b6c6ea8a5354eaf15b3cb7646744f4275b71ea724fed81ceb9323e279d449df9\r\n
+ * x-amz-checksum-crc32c:sOO8/Q==\r\n
+ * x-amz-trailer-signature:63bddb248ad2590c92712055f51b8e78ab024eead08276b24f010b0efd74843f\r\n
+ * </pre>
+ * </p>
+ * For the first chunk 10000 will be read and decoded from base-16 representation to 65536, which is the size of
+ * the first chunk payload. Each chunk upload ends with a zero-byte final additional chunk.
+ * At the end, there might be a trailer checksum payload and signature, depending on whether the x-amz-content-sha256
+ * header value contains "-TRAILER" suffix (e.g. STREAMING-AWS4-HMAC-SHA256-PAYLOAD-TRAILER
+ * and STREAMING-AWS4-ECDSA-P256-SHA256-PAYLOAD-TRAILER) and "x-amz-trailer" is specified (e.g. x-amz-checksum-crc32c).
+ * <p>
+ *
+ * <p>
+ * The logic is similar to {@link UnsignedChunksInputStream}, but there is a "chunk-signature" to parse.
+ * </p>
+ *
+ * <p>
  * Note that there are no actual chunk signature verification taking place. The InputStream only
  * returns the actual chunk payload from chunked signatures format.
+ * </p>
  *
- * See
- * - https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html
- * - https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming-trailers.html
+ * Reference:
+ * <ul>
+ *   <li>
+ *     <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html">
+ *        Signature Calculation: Transfer Payload in Multiple Chunks</a>
+ *   </li>
+ *   <li>
+ *     <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming-trailers.html">
+ *        Signature Calculation: Including Trailing Headers</a>
+ *   </li>
+ * </ul>
  */
 public class SignedChunksInputStream extends InputStream {
 
@@ -46,12 +82,21 @@ public class SignedChunksInputStream extends InputStream {
    */
   private int remainingData = 0;
 
+  /**
+   * Every chunked uploads (multiple chunks) contains an additional final zero-byte
+   * chunk. This can be used as the end-of-file marker.
+   */
+  private boolean isFinalChunkEncountered = false;
+
   public SignedChunksInputStream(InputStream inputStream) {
     originalStream = inputStream;
   }
 
   @Override
   public int read() throws IOException {
+    if (isFinalChunkEncountered) {
+      return -1;
+    }
     if (remainingData > 0) {
       int curr = originalStream.read();
       remainingData--;
@@ -63,7 +108,10 @@ public int read() throws IOException {
       return curr;
     } else {
       remainingData = readContentLengthFromHeader();
-      if (remainingData == -1) {
+      if (remainingData <= 0) {
+        // there is always a final zero byte chunk so we can stop reading
+        // if we encounter this chunk
+        isFinalChunkEncountered = true;
         return -1;
       }
       return read();
@@ -72,12 +120,14 @@ public int read() throws IOException {
 
   @Override
   public int read(byte[] b, int off, int len) throws IOException {
-    if (b == null) {
-      throw new NullPointerException();
-    } else if (off < 0 || len < 0 || len > b.length - off) {
-      throw new IndexOutOfBoundsException();
+    Objects.requireNonNull(b, "b == null");
+    if (off < 0 || len < 0 || len > b.length - off) {
+      throw new IndexOutOfBoundsException("Offset=" + off + " and len="
+          + len + " don't match the array length of " + b.length);
     } else if (len == 0) {
       return 0;
+    } else if (isFinalChunkEncountered) {
+      return -1;
     }
     int currentOff = off;
     int currentLen = len;
@@ -103,7 +153,12 @@ public int read(byte[] b, int off, int len) throws IOException {
         }
       } else {
         remainingData = readContentLengthFromHeader();
-        if (remainingData == -1) {
+        if (remainingData == 0) {
+          // there is always a final zero byte chunk so we can stop reading
+          // if we encounter this chunk
+          isFinalChunkEncountered = true;
+        }
+        if (isFinalChunkEncountered || remainingData == -1) {
           break;
         }
       }
@@ -125,10 +180,9 @@ private int readContentLengthFromHeader() throws IOException {
       prev = curr;
       curr = next;
     }
-    // Example
-    // The chunk data sent:
-    //  10000;chunk-signature=b474d8862b1487a5145d686f57f013e54db672cee1c953b3010fb58501ef5aa2
-    //  <65536-bytes>
+    // Example of a single chunk data:
+    //  10000;chunk-signature=b474d8862b1487a5145d686f57f013e54db672cee1c953b3010fb58501ef5aa2\r\n
+    //  <65536-bytes>\r\n
     //
     // 10000 will be read and decoded from base-16 representation to 65536, which is the size of
     // the subsequent chunk payload.
@@ -145,8 +199,4 @@ private int readContentLengthFromHeader() throws IOException {
       throw new IOException("Invalid signature line: " + signatureLine);
     }
   }
-
-  private boolean eol(int prev, int curr) {
-    return prev == 13 && curr == 10;
-  }
 }