HDDS-3161. Block illegal characters when creating keys.

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/client/HddsClientUtils.java

@@ -196,6 +196,23 @@ public static void verifyResourceName(String... resourceNames) {
     }
   }
 
+  /**
+   * verifies that key name is a valid name.
+   *
+   * @param keyName key name to be validated
+   *
+   * @throws IllegalArgumentException
+   */
+  public static void verifyKeyName(String keyName) {
+    if (keyName == null) {
+      throw new IllegalArgumentException("Key name is null");
+    }
+    if(!OzoneConsts.KEYNAME_ILLEGAL_CHARACTER_CHECK_REGEX
+            .matcher(keyName).matches()){
+      throw new IllegalArgumentException("Invalid key name: " + keyName);
+    }
+  }
+
   /**
    * Checks that object parameters passed as reference is not null.
    *

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -24,6 +24,8 @@
 import org.apache.ratis.thirdparty.io.grpc.Context;
 import org.apache.ratis.thirdparty.io.grpc.Metadata;
 
+import java.util.regex.Pattern;
+
 import static org.apache.ratis.thirdparty.io.grpc.Metadata.ASCII_STRING_MARSHALLER;
 
 /**
@@ -337,6 +339,21 @@ private OzoneConsts() {
   public static final String GDPR_SECRET = "secret";
   public static final String GDPR_ALGORITHM = "algorithm";
 
+  /**
+   * Block key name as illegal characters
+   *
+   * This regular expression is used to check if key name
+   * contains illegal characters when creating/renaming key.
+   *
+   * Avoid the following characters in a key name:
+   * "\", "{", "}", "^", "<", ">", "#", "|", "%", "`", "[", "]", "~", "?"
+   * and Non-printable ASCII characters (128–255 decimal characters).
+   * https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingMetadata.html
+   */
+  public static final Pattern KEYNAME_ILLEGAL_CHARACTER_CHECK_REGEX  =
+          Pattern.compile("^[^^{}<>^?%~#`\\[\\]\\|\\\\(\\x80-\\xff)]+$");
+
+  public static final String FS_FILE_COPYING_TEMP_SUFFIX= "._COPYING_";
 
   // Transaction Info
   public static final String TRANSACTION_INFO_KEY = "#TRANSACTIONINFO";

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -2490,4 +2490,14 @@
       Timeout for the request submitted directly to Ratis in datanode.
     </description>
   </property>
+  <property>
+    <name>ozone.om.keyname.character.check.enabled</name>
+    <tag>OZONE, OM</tag>
+    <value>false</value>
+    <description>If true, then enable to check if the key name
+      contains illegal characters when creating/renaming key.
+      For the definition of illegal characters, follow the
+      rules in Amazon S3's object key naming guide.
+    </description>
+  </property>
 </configuration>

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -72,6 +72,7 @@
 import org.apache.hadoop.ozone.client.io.OzoneInputStream;
 import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
 import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.BucketEncryptionKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmBucketArgs;
@@ -145,6 +146,7 @@ public class RpcClient implements ClientProtocol {
   private final long retryInterval;
   private Text dtService;
   private final boolean topologyAwareReadEnabled;
+  private final boolean checkKeyNameEnabled;
 
   /**
     * Creates RpcClient instance with the given configuration.
@@ -242,6 +244,9 @@ public RpcClient(ConfigurationSource conf, String omServiceId)
     topologyAwareReadEnabled = conf.getBoolean(
         OzoneConfigKeys.OZONE_NETWORK_TOPOLOGY_AWARE_READ_KEY,
         OzoneConfigKeys.OZONE_NETWORK_TOPOLOGY_AWARE_READ_DEFAULT);
+    checkKeyNameEnabled = conf.getBoolean(
+            OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_KEY,
+            OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_DEFAULT);
   }
 
   @Override
@@ -638,6 +643,9 @@ public OzoneOutputStream createKey(
       throws IOException {
     verifyVolumeName(volumeName);
     verifyBucketName(bucketName);
+    if(checkKeyNameEnabled) {
+      HddsClientUtils.verifyKeyName(keyName);
+    }
     HddsClientUtils.checkNotNull(keyName, type, factor);
     String requestId = UUID.randomUUID().toString();
 
@@ -718,6 +726,9 @@ public void renameKey(String volumeName, String bucketName,
       String fromKeyName, String toKeyName) throws IOException {
     verifyVolumeName(volumeName);
     verifyBucketName(bucketName);
+    if(checkKeyNameEnabled){
+      HddsClientUtils.verifyKeyName(toKeyName);
+    }
     HddsClientUtils.checkNotNull(fromKeyName, toKeyName);
     OmKeyArgs keyArgs = new OmKeyArgs.Builder()
         .setVolumeName(volumeName)
@@ -831,6 +842,9 @@ public OzoneOutputStream createMultipartKey(String volumeName,
       throws IOException {
     verifyVolumeName(volumeName);
     verifyBucketName(bucketName);
+    if(checkKeyNameEnabled) {
+      HddsClientUtils.verifyKeyName(keyName);
+    }
     HddsClientUtils.checkNotNull(keyName, uploadID);
     Preconditions.checkArgument(partNumber > 0 && partNumber <=10000, "Part " +
         "number should be greater than zero and less than or equal to 10000");

hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/TestHddsClientUtils.java

@@ -262,4 +262,26 @@ public void testVerifyResourceName() {
       }
     }
   }
+
+  @Test
+  public void testVerifyKeyName() {
+    List<String> invalidNames = new ArrayList<>();
+    invalidNames.add("#");
+    invalidNames.add("ab^cd");
+    invalidNames.add("test|name~");
+    invalidNames.add("~hi!ozone");
+    invalidNames.add("test<string>");
+    invalidNames.add("10%3=1");
+    invalidNames.add("photo[0201]");
+    invalidNames.add("what?");
+
+    for (String name : invalidNames) {
+      try {
+        HddsClientUtils.verifyKeyName(name);
+        fail("Did not reject invalid string [" + name + "] as a name");
+      } catch (IllegalArgumentException e) {
+        // throwing up on an invalid name. it's working.
+      }
+    }
+  }
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -516,4 +516,17 @@ public static OmKeyInfo prepareKeyForRecover(OmKeyInfo keyInfo,
       return null;
     }
   }
+
+  /**
+   * Verify key name is a valid name.
+   */
+  public static void validateKeyName(String keyName)
+          throws OMException {
+    try {
+      HddsClientUtils.verifyKeyName(keyName);
+    } catch (IllegalArgumentException e) {
+      throw new OMException(e.getMessage(),
+              OMException.ResultCodes.INVALID_KEY_NAME);
+    }
+  }
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java

@@ -234,4 +234,9 @@ private OMConfigKeys() {
   // hadoop-policy.xml, "*" allows all users/groups to access.
   public static final String OZONE_OM_SECURITY_CLIENT_PROTOCOL_ACL =
       "ozone.om.security.client.protocol.acl";
+
+  public static final String OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_KEY =
+          "ozone.om.keyname.character.check.enabled";
+  public static final boolean OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_DEFAULT =
+          false;
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java

@@ -27,7 +27,11 @@
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.exceptions.OMReplayException;
 import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
 import org.apache.hadoop.ozone.om.response.file.OMFileCreateResponse;
@@ -89,6 +93,15 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
 
     KeyArgs keyArgs = createFileRequest.getKeyArgs();
 
+    // Verify key name
+    final boolean checkKeyNameEnabled = ozoneManager.getConfiguration()
+         .getBoolean(OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_KEY,
+                 OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_DEFAULT);
+    if(checkKeyNameEnabled){
+      OmUtils.validateKeyName(StringUtils.removeEnd(keyArgs.getKeyName(),
+              OzoneConsts.FS_FILE_COPYING_TEMP_SUFFIX));
+    }
+
     if (keyArgs.getKeyName().length() == 0) {
       // Check if this is the root of the filesystem.
       // Not throwing exception here, as need to throw exception after

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java

@@ -25,6 +25,10 @@
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.ozone.OmUtils;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
 import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
@@ -85,6 +89,15 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
 
     KeyArgs keyArgs = commitKeyRequest.getKeyArgs();
 
+    // Verify key name
+    final boolean checkKeyNameEnabled = ozoneManager.getConfiguration()
+         .getBoolean(OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_KEY,
+                 OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_DEFAULT);
+    if(checkKeyNameEnabled){
+      OmUtils.validateKeyName(StringUtils.removeEnd(keyArgs.getKeyName(),
+              OzoneConsts.FS_FILE_COPYING_TEMP_SUFFIX));
+    }
+
     KeyArgs.Builder newKeyArgs =
         keyArgs.toBuilder().setModificationTime(Time.now());
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java

@@ -26,6 +26,8 @@
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.ozone.OmUtils;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -83,6 +85,13 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
 
     KeyArgs keyArgs = createKeyRequest.getKeyArgs();
 
+    // Verify key name
+    final boolean checkKeyNameEnabled = ozoneManager.getConfiguration()
+         .getBoolean(OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_KEY,
+                 OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_DEFAULT);
+    if(checkKeyNameEnabled){
+      OmUtils.validateKeyName(keyArgs.getKeyName());
+    }
     // We cannot allocate block for multipart upload part when
     // createMultipartKey is called, as we will not know type and factor with
     // which initiateMultipartUpload has started for this key. When

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRenameRequest.java

@@ -23,7 +23,9 @@
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
 import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
@@ -85,6 +87,14 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
     RenameKeyRequest renameKeyRequest = getOmRequest().getRenameKeyRequest();
     Preconditions.checkNotNull(renameKeyRequest);
 
+    // Verify key name
+    final boolean checkKeyNameEnabled = ozoneManager.getConfiguration()
+         .getBoolean(OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_KEY,
+                 OMConfigKeys.OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_DEFAULT);
+    if(checkKeyNameEnabled){
+      OmUtils.validateKeyName(renameKeyRequest.getToKeyName());
+    }
+
     // Set modification time.
     KeyArgs.Builder newKeyArgs = renameKeyRequest.getKeyArgs().toBuilder()
             .setModificationTime(Time.now());