HDDS-16672. Support RFC 2047 encoding for S3 user metadata

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

@@ -579,6 +579,25 @@ public void testPutObject() {
     assertEquals("37b51d194a7513e45b56f6524f2d51f2", putObjectResult.getETag());
   }
 
+  @Test
+  public void testPutObjectWithRfc2047Metadata() {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    final String metadataValue = "café";
+    final String encodedMetadataValue = "=?UTF-8?Q?caf=C3=A9?=";
+    byte[] content = "bar".getBytes(StandardCharsets.UTF_8);
+    ObjectMetadata metadata = new ObjectMetadata();
+    metadata.setContentLength(content.length);
+    metadata.addUserMetadata("meta1", metadataValue);
+    s3Client.createBucket(bucketName);
+
+    s3Client.putObject(new PutObjectRequest(bucketName, keyName,
+        new ByteArrayInputStream(content), metadata));
+
+    assertEquals(encodedMetadataValue,
+        s3Client.getObjectMetadata(bucketName, keyName).getUserMetaDataOf("meta1"));
+  }
+
   @Test
   public void testPutObjectWithEmptyContentType() {
     final String bucketName = getBucketName();

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

@@ -308,6 +308,22 @@ public void testPutObject() {
     assertEquals("\"37b51d194a7513e45b56f6524f2d51f2\"", getObjectResponse.eTag());
   }
 
+  @Test
+  public void testPutObjectWithRfc2047Metadata() {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    final String metadataValue = "café";
+    final String encodedMetadataValue = "=?UTF-8?Q?caf=C3=A9?=";
+    s3Client.createBucket(b -> b.bucket(bucketName));
+
+    s3Client.putObject(b -> b.bucket(bucketName).key(keyName)
+            .metadata(Collections.singletonMap("meta1", metadataValue)),
+        RequestBody.fromString("bar"));
+
+    HeadObjectResponse response = s3Client.headObject(b -> b.bucket(bucketName).key(keyName));
+    assertEquals(encodedMetadataValue, response.metadata().get("meta1"));
+  }
+
   @Test
   public void testPutObjectWithEmptyContentType() {
     final String bucketName = getBucketName();

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

@@ -69,6 +69,7 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -87,6 +88,10 @@
 import javax.ws.rs.core.MultivaluedMap;
 import javax.ws.rs.core.Response;
 import net.jcip.annotations.Immutable;
+import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.EncoderException;
+import org.apache.commons.codec.net.BCodec;
+import org.apache.commons.codec.net.QCodec;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
@@ -138,6 +143,9 @@
  */
 public abstract class EndpointBase {
 
+  private static final BCodec RFC_2047_B_CODEC = new BCodec(UTF_8);
+  private static final QCodec RFC_2047_Q_CODEC = createRfc2047QCodec();
+
   protected static final String ETAG_CUSTOM =
       RESERVED_USER_METADATA_KEY_PREFIX + "etag";
   protected static final String CONTENT_TYPE_CUSTOM =
@@ -451,7 +459,9 @@ protected Map<String, String> getCustomMetadataFromHeaders(
           throw ex;
         }
         List<String> values = requestHeaders.get(key);
-        String value = StringUtils.join(values, ",");
+        String value = values.stream()
+            .map(EndpointBase::decodeRfc2047MetadataValue)
+            .collect(Collectors.joining(","));
         sizeInBytes += mapKey.getBytes(UTF_8).length;
         sizeInBytes += value.getBytes(UTF_8).length;
 
@@ -470,6 +480,57 @@ protected Map<String, String> getCustomMetadataFromHeaders(
     return customMetadata;
   }
 
+  private static String decodeRfc2047MetadataValue(String value) {
+    if (!value.startsWith("=?")) {
+      return value;
+    }
+
+    int charsetEnd = value.indexOf('?', 2);
+    int encodingEnd = charsetEnd < 0 ? -1 : value.indexOf('?', charsetEnd + 1);
+    int encodedTextEnd = encodingEnd < 0
+        ? -1 : value.indexOf("?=", encodingEnd + 1);
+    if (encodedTextEnd != value.length() - 2) {
+      return value;
+    }
+
+    String encoding = value.substring(charsetEnd + 1, encodingEnd);
+    try {
+      if (encoding.equalsIgnoreCase("Q")) {
+        return RFC_2047_Q_CODEC.decode(value);
+      }
+      if (encoding.equalsIgnoreCase("B")) {
+        Base64.getDecoder().decode(
+            value.substring(encodingEnd + 1, encodedTextEnd));
+        return RFC_2047_B_CODEC.decode(value);
+      }
+    } catch (DecoderException | IllegalArgumentException ex) {
+      return value;
+    }
+    return value;
+  }
+
+  private static String encodeRfc2047MetadataValue(String value) {
+    if (value.chars().allMatch(EndpointBase::isSafeAsciiMetadataCharacter)) {
+      return value;
+    }
+
+    try {
+      return RFC_2047_Q_CODEC.encode(value);
+    } catch (EncoderException ex) {
+      throw new IllegalStateException("Failed to encode S3 metadata", ex);
+    }
+  }
+
+  private static boolean isSafeAsciiMetadataCharacter(int value) {
+    return value == '\t' || value >= ' ' && value <= '~';
+  }
+
+  private static QCodec createRfc2047QCodec() {
+    QCodec codec = new QCodec(UTF_8);
+    codec.setEncodeBlanks(true);
+    return codec;
+  }
+
   /**
    * Move a user value under {@code headerName} (canonical or lower-case) to
    * {@code customKey} so it does not collide with the system value.
@@ -500,7 +561,7 @@ protected void addCustomMetadataHeaders(
       metadataKey = REBUILT_RESERVED_KEYS.getOrDefault(metadataKey, metadataKey);
       responseBuilder
           .header(CUSTOM_METADATA_HEADER_PREFIX + metadataKey,
-              entry.getValue());
+              encodeRfc2047MetadataValue(entry.getValue()));
     }
   }
 

hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestObjectGet.java

@@ -243,6 +243,75 @@ public void getKeyWithCustomMetadata() throws IOException, OS3Exception {
         response.getHeaderString(CUSTOM_METADATA_HEADER_PREFIX + "meta1"));
   }
 
+  @Test
+  public void getKeyWithQEncodedRfc2047CustomMetadata()
+      throws IOException, OS3Exception {
+    final String keyName = "key-with-rfc2047-meta";
+    final String metaValue = "Hello Worldé";
+    final String encodedMetaValue = "=?UTF-8?Q?Hello_World=C3=A9?=";
+    MultivaluedMap<String, String> requestHeaders = new MultivaluedHashMap<>();
+    requestHeaders.putSingle(CUSTOM_METADATA_HEADER_PREFIX + "meta1",
+        encodedMetaValue);
+    when(headers.getRequestHeaders()).thenReturn(requestHeaders);
+
+    assertSucceeds(() -> put(rest, BUCKET_NAME, keyName, CONTENT));
+
+    assertEquals(metaValue, bucket.getKey(keyName).getMetadata().get("meta1"));
+    Response response = get(rest, BUCKET_NAME, keyName);
+    assertEquals(encodedMetaValue,
+        response.getHeaderString(CUSTOM_METADATA_HEADER_PREFIX + "meta1"));
+  }
+
+  @Test
+  public void getKeyWithBase64Rfc2047CustomMetadata()
+      throws IOException, OS3Exception {
+    final String keyName = "key-with-base64-rfc2047-meta";
+    final String metaValue = "café";
+    final String encodedMetaValue = "=?UTF-8?B?Y2Fmw6k=?=";
+    final String responseMetaValue = "=?UTF-8?Q?caf=C3=A9?=";
+    MultivaluedMap<String, String> requestHeaders = new MultivaluedHashMap<>();
+    requestHeaders.putSingle(CUSTOM_METADATA_HEADER_PREFIX + "meta1",
+        encodedMetaValue);
+    when(headers.getRequestHeaders()).thenReturn(requestHeaders);
+
+    assertSucceeds(() -> put(rest, BUCKET_NAME, keyName, CONTENT));
+
+    assertEquals(metaValue, bucket.getKey(keyName).getMetadata().get("meta1"));
+    Response response = get(rest, BUCKET_NAME, keyName);
+    assertEquals(responseMetaValue,
+        response.getHeaderString(CUSTOM_METADATA_HEADER_PREFIX + "meta1"));
+  }
+
+  @Test
+  public void getKeyPreservesMultipleRfc2047EncodedWords()
+      throws IOException, OS3Exception {
+    final String keyName = "key-with-multiple-rfc2047-words";
+    final String metaValue = "=?UTF-8?Q?a?= =?UTF-8?Q?b?=";
+    MultivaluedMap<String, String> requestHeaders = new MultivaluedHashMap<>();
+    requestHeaders.putSingle(CUSTOM_METADATA_HEADER_PREFIX + "meta1",
+        metaValue);
+    when(headers.getRequestHeaders()).thenReturn(requestHeaders);
+
+    assertSucceeds(() -> put(rest, BUCKET_NAME, keyName, CONTENT));
+
+    assertEquals(metaValue, bucket.getKey(keyName).getMetadata().get("meta1"));
+  }
+
+  @Test
+  public void getKeyPreservesInvalidBase64Rfc2047Metadata()
+      throws IOException, OS3Exception {
+    final String keyName = "key-with-invalid-rfc2047-base64";
+    final String metaValue = "=?UTF-8?B?!!!?=";
+    MultivaluedMap<String, String> requestHeaders = new MultivaluedHashMap<>();
+    requestHeaders.putSingle(CUSTOM_METADATA_HEADER_PREFIX + "meta1",
+        metaValue);
+    when(headers.getRequestHeaders()).thenReturn(requestHeaders);
+
+    assertSucceeds(() -> put(rest, BUCKET_NAME, keyName, CONTENT));
+
+    assertEquals(metaValue, bucket.getKey(keyName).getMetadata().get("meta1"));
+  }
+
   @Test
   public void getKeyWithTag() throws IOException, OS3Exception {
     //WHEN