HDFS-10883. 's behavior is not consistent in DFS after enabling EZ. Contributed by Yuanbo Liu.

(cherry picked from commit 0007360c33)

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -2474,11 +2474,12 @@ public class DistributedFileSystem extends FileSystem {
    */
   @Override
   public Path getTrashRoot(Path path) {
-    if ((path == null) || path.isRoot() || !dfs.isHDFSEncryptionEnabled()) {
+    if ((path == null) || !dfs.isHDFSEncryptionEnabled()) {
       return super.getTrashRoot(path);
     }
 
-    String parentSrc = path.getParent().toUri().getPath();
+    String parentSrc = path.isRoot()?
+        path.toUri().getPath():path.getParent().toUri().getPath();
     try {
       EncryptionZone ez = dfs.getEZForPath(parentSrc);
       if ((ez != null)) {

hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/TransparentEncryption.md

@@ -242,12 +242,14 @@ By default, distcp compares checksums provided by the filesystem to verify that
 <a name="Rename_and_Trash_considerations"></a>Rename and Trash considerations
 ---------------------
 
-HDFS restricts file and directory renames across encryption zone boundaries. This includes renaming an encrypted file / directory into an unencrypted directory (e.g., `hdfs dfs mv /zone/encryptedFile /home/bob`), renaming an unencrypted file / directory into an encryption zone (e.g., `hdfs dfs mv /home/bob/unEncryptedFile /zone`), and renaming between two different encryption zones (e.g., `hdfs dfs mv /home/alice/zone1/foo /home/alice/zone2`). In these examples, `/zone`, `/home/alice/zone1`, and `/home/alice/zone2` are encryption zones, while `/home/bob` is not. A rename is only allowed if the source and destination paths are in the same encryption zone, or both paths are unencrypted (not in any encryption zone).
+HDFS restricts file and directory renames across encryption zone boundaries. This includes renaming an encrypted file / directory into an unencrypted directory (e.g., `hdfs dfs mv /zone/encryptedFile /home/bob`), renaming an unencrypted file or directory into an encryption zone (e.g., `hdfs dfs mv /home/bob/unEncryptedFile /zone`), and renaming between two different encryption zones (e.g., `hdfs dfs mv /home/alice/zone1/foo /home/alice/zone2`). In these examples, `/zone`, `/home/alice/zone1`, and `/home/alice/zone2` are encryption zones, while `/home/bob` is not. A rename is only allowed if the source and destination paths are in the same encryption zone, or both paths are unencrypted (not in any encryption zone).
 
 This restriction enhances security and eases system management significantly. All file EDEKs under an encryption zone are encrypted with the encryption zone key. Therefore, if the encryption zone key is compromised, it is important to identify all vulnerable files and re-encrypt them. This is fundamentally difficult if a file initially created in an encryption zone can be renamed to an arbitrary location in the filesystem.
 
 To comply with the above rule, each encryption zone has its own `.Trash` directory under the "zone directory". E.g., after `hdfs dfs rm /zone/encryptedFile`, `encryptedFile` will be moved to `/zone/.Trash`, instead of the `.Trash` directory under the user's home directory. When the entire encryption zone is deleted, the "zone directory" will be moved to the `.Trash` directory under the user's home directory.
 
+If the encryption zone is the root directory (e.g., `/` directory), the trash path of root directory is `/.Trash`, not the `.Trash` directory under the user's home directory, and the behavior of renaming sub-directories or sub-files in root directory will keep consistent with the behavior in a general encryption zone, such as `/zone` which is mentioned at the top of this section.
+
 The `crypto` command before Hadoop 2.8.0 does not provision the `.Trash` directory automatically. If an encryption zone is created before Hadoop 2.8.0, and then the cluster is upgraded to Hadoop 2.8.0 or above, the trash directory can be provisioned using `-provisionTrash` option (e.g., `hdfs crypto -provisionTrash -path /zone`).
 <a name="Attack_vectors"></a>Attack vectors
 --------------

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestEncryptionZones.java

@@ -1566,7 +1566,8 @@ public class TestEncryptionZones {
   public void testRootDirEZTrash() throws Exception {
     final HdfsAdmin dfsAdmin =
         new HdfsAdmin(FileSystem.getDefaultUri(conf), conf);
-    dfsAdmin.createEncryptionZone(new Path("/"), TEST_KEY, NO_TRASH);
+    final Path rootDir = new Path("/");
+    dfsAdmin.createEncryptionZone(rootDir, TEST_KEY, NO_TRASH);
     final Path encFile = new Path("/encFile");
     final int len = 8192;
     DFSTestUtil.createFile(fs, encFile, len, (short) 1, 0xFEED);
@@ -1574,6 +1575,13 @@ public class TestEncryptionZones {
     clientConf.setLong(FS_TRASH_INTERVAL_KEY, 1);
     FsShell shell = new FsShell(clientConf);
     verifyShellDeleteWithTrash(shell, encFile);
+
+    // Trash path should be consistent
+    // if root path is an encryption zone
+    Path encFileTrash = shell.getCurrentTrashDir(encFile);
+    Path rootDirTrash = shell.getCurrentTrashDir(rootDir);
+    assertEquals("Root trash should be equal with ezFile trash",
+        encFileTrash, rootDirTrash);
   }
 
   @Test(timeout = 120000)

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestNestedEncryptionZones.java

@@ -20,7 +20,9 @@ package org.apache.hadoop.hdfs.server.namenode;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.crypto.key.JavaKeyStoreProvider;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileSystemTestHelper;
+import org.apache.hadoop.fs.FsShell;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
@@ -29,6 +31,8 @@ import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.ToolRunner;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.junit.Before;
@@ -53,16 +57,19 @@ public class TestNestedEncryptionZones {
 
   private final Path rootDir = new Path("/");
   private final Path rawDir = new Path("/.reserved/raw/");
-  private final Path topEZDir = new Path(rootDir, "topEZ");
-  private final Path nestedEZDir = new Path(topEZDir, "nestedEZ");
 
-  private final Path topEZBaseFile = new Path(rootDir, "topEZBaseFile");
+  private Path nestedEZBaseFile = new Path(rootDir, "nestedEZBaseFile");
-  private Path topEZFile = new Path(topEZDir, "file");
+  private Path topEZBaseFile = new Path(rootDir, "topEZBaseFile");
-  private Path topEZRawFile = new Path(rawDir, "topEZ/file");
+
+  private Path topEZDir;
+  private Path nestedEZDir;
+
+  private Path topEZFile;
+  private Path nestedEZFile;
+
+  private Path topEZRawFile;
+  private Path nestedEZRawFile;
 
-  private final Path nestedEZBaseFile = new Path(rootDir, "nestedEZBaseFile");
-  private Path nestedEZFile = new Path(nestedEZDir, "file");
-  private Path nestedEZRawFile = new Path(rawDir, "topEZ/nestedEZ/file");
 
   // File length
   private final int len = 8196;
@@ -92,6 +99,8 @@ public class TestNestedEncryptionZones {
     // Lower the batch size for testing
     conf.setInt(DFSConfigKeys.DFS_NAMENODE_LIST_ENCRYPTION_ZONES_NUM_RESPONSES,
         2);
+    // enable trash for testing
+    conf.setLong(DFSConfigKeys.FS_TRASH_INTERVAL_KEY, 1);
     cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
     Logger.getLogger(EncryptionZoneManager.class).setLevel(Level.TRACE);
     fs = cluster.getFileSystem();
@@ -100,24 +109,17 @@ public class TestNestedEncryptionZones {
     // Create test keys and EZs
     DFSTestUtil.createKey(TOP_EZ_KEY, cluster, conf);
     DFSTestUtil.createKey(NESTED_EZ_KEY, cluster, conf);
-    fs.mkdir(topEZDir, FsPermission.getDirDefault());
-    fs.createEncryptionZone(topEZDir, TOP_EZ_KEY);
-    fs.mkdir(nestedEZDir, FsPermission.getDirDefault());
-    fs.createEncryptionZone(nestedEZDir, NESTED_EZ_KEY);
-
-    DFSTestUtil.createFile(fs, topEZBaseFile, len, (short) 1, 0xFEED);
-    DFSTestUtil.createFile(fs, topEZFile, len, (short) 1, 0xFEED);
-    DFSTestUtil.createFile(fs, nestedEZBaseFile, len, (short) 1, 0xFEED);
-    DFSTestUtil.createFile(fs, nestedEZFile, len, (short) 1, 0xFEED);
   }
 
   @Test(timeout = 60000)
   public void testNestedEncryptionZones() throws Exception {
+    initTopEZDirAndNestedEZDir(new Path(rootDir, "topEZ"));
     verifyEncryption();
 
     // Restart NameNode to test if nested EZs can be loaded from edit logs
     cluster.restartNameNodes();
     cluster.waitActive();
+    fs = cluster.getFileSystem();
     verifyEncryption();
 
     // Checkpoint and restart NameNode, to test if nested EZs can be loaded
@@ -127,21 +129,88 @@ public class TestNestedEncryptionZones {
     fs.setSafeMode(HdfsConstants.SafeModeAction.SAFEMODE_LEAVE);
     cluster.restartNameNodes();
     cluster.waitActive();
+    fs = cluster.getFileSystem();
     verifyEncryption();
 
+    renameChildrenOfEZ();
+
+    // Verify that a non-nested EZ cannot be moved into another EZ
+    Path topEZ2Dir = new Path(rootDir, "topEZ2");
+    fs.mkdir(topEZ2Dir, FsPermission.getDirDefault());
+    fs.createEncryptionZone(topEZ2Dir, TOP_EZ_KEY);
+    try {
+      fs.rename(topEZ2Dir, new Path(topEZDir, "topEZ2"));
+      fail("Shouldn't be able to move a non-nested EZ into another " +
+          "existing EZ.");
+    } catch (Exception e){
+      assertTrue(e.getMessage().contains(
+          "can't be moved into an encryption zone"));
+    }
+
+    // Should be able to rename the root dir of an EZ.
+    fs.rename(topEZDir, new Path(rootDir, "newTopEZ"));
+
+    // Should be able to rename the nested EZ dir within the same top EZ.
+    fs.rename(new Path(rootDir, "newTopEZ/nestedEZ"),
+        new Path(rootDir, "newTopEZ/newNestedEZ"));
+  }
+
+  @Test(timeout = 60000)
+  public void testNestedEZWithRoot() throws Exception {
+    initTopEZDirAndNestedEZDir(rootDir);
+    verifyEncryption();
+
+    // test rename file
+    renameChildrenOfEZ();
+
+    final String currentUser =
+        UserGroupInformation.getCurrentUser().getShortUserName();
+    final Path suffixTrashPath = new Path(
+        FileSystem.TRASH_PREFIX, currentUser);
+
+    final Path rootTrash = fs.getTrashRoot(rootDir);
+    final Path topEZTrash = fs.getTrashRoot(topEZFile);
+    final Path nestedEZTrash = fs.getTrashRoot(nestedEZFile);
+
+    final Path expectedTopEZTrash = fs.makeQualified(
+        new Path(topEZDir, suffixTrashPath));
+    final Path expectedNestedEZTrash = fs.makeQualified(
+        new Path(nestedEZDir, suffixTrashPath));
+
+    assertEquals("Top ez trash should be " + expectedTopEZTrash,
+        expectedTopEZTrash, topEZTrash);
+    assertEquals("Root trash should be equal with TopEZFile trash",
+        topEZTrash, rootTrash);
+    assertEquals("Nested ez Trash should be " + expectedNestedEZTrash,
+        expectedNestedEZTrash, nestedEZTrash);
+
+    // delete rename file and test trash
+    FsShell shell = new FsShell(fs.getConf());
+    final Path topTrashFile = new Path(
+        shell.getCurrentTrashDir(topEZFile) + "/" + topEZFile);
+    final Path nestedTrashFile = new Path(
+        shell.getCurrentTrashDir(nestedEZFile) + "/" + nestedEZFile);
+
+    ToolRunner.run(shell, new String[]{"-rm", topEZFile.toString()});
+    ToolRunner.run(shell, new String[]{"-rm", nestedEZFile.toString()});
+
+    assertTrue("File not in trash : " + topTrashFile, fs.exists(topTrashFile));
+    assertTrue(
+        "File not in trash : " + nestedTrashFile, fs.exists(nestedTrashFile));
+  }
+
+  private void renameChildrenOfEZ() throws Exception{
     Path renamedTopEZFile = new Path(topEZDir, "renamedFile");
     Path renamedNestedEZFile = new Path(nestedEZDir, "renamedFile");
-    try {
+
+    //Should be able to rename files within the same EZ.
     fs.rename(topEZFile, renamedTopEZFile);
     fs.rename(nestedEZFile, renamedNestedEZFile);
-    } catch (Exception e) {
-      fail("Should be able to rename files within the same EZ.");
-    }
 
     topEZFile = renamedTopEZFile;
     nestedEZFile = renamedNestedEZFile;
-    topEZRawFile = new Path(rawDir, "topEZ/renamedFile");
+    topEZRawFile = new Path(rawDir + topEZFile.toUri().getPath());
-    nestedEZRawFile = new Path(rawDir, "topEZ/nestedEZ/renamedFile");
+    nestedEZRawFile = new Path(rawDir + nestedEZFile.toUri().getPath());
     verifyEncryption();
 
     // Verify that files in top EZ cannot be moved into the nested EZ, and
@@ -168,36 +237,40 @@ public class TestNestedEncryptionZones {
       fs.rename(nestedEZFile, new Path(rootDir, "movedNestedEZFile"));
       fail("Shouldn't be able to move the nested EZ out of the top EZ.");
     } catch (Exception e) {
-      assertTrue(e.getMessage().contains(
+      String exceptionMsg = e.getMessage();
-          "can't be moved from an encryption zone"));
+      assertTrue(exceptionMsg.contains(
+          "can't be moved from") && exceptionMsg.contains("encryption zone"));
+    }
   }
 
-    // Verify that a non-nested EZ cannot be moved into another EZ
+  private void initTopEZDirAndNestedEZDir(Path topPath) throws Exception {
-    Path topEZ2Dir = new Path(rootDir, "topEZ2");
-    fs.mkdir(topEZ2Dir, FsPermission.getDirDefault());
-    fs.createEncryptionZone(topEZ2Dir, TOP_EZ_KEY);
-    try {
-      fs.rename(topEZ2Dir, new Path(topEZDir, "topEZ2"));
-      fail("Shouldn't be able to move a non-nested EZ into another " +
-          "existing EZ.");
-    } catch (Exception e){
-      assertTrue(e.getMessage().contains(
-          "can't be moved into an encryption zone"));
-    }
 
-    try {
+    // init fs root directory
-      fs.rename(topEZDir, new Path(rootDir, "newTopEZDir"));
+    fs.delete(rootDir, true);
-    } catch (Exception e) {
+
-      fail("Should be able to rename the root dir of an EZ.");
+
-    }
+    // init top and nested path or file
+    topEZDir = topPath;
+    nestedEZDir = new Path(topEZDir, "nestedEZ");
+
+    topEZFile = new Path(topEZDir, "file");
+    nestedEZFile = new Path(nestedEZDir, "file");
+
+    topEZRawFile = new Path(rawDir + topEZFile.toUri().getPath());
+    nestedEZRawFile = new Path(rawDir + nestedEZFile.toUri().getPath());
+
+    // create ez zone
+    fs.mkdir(topEZDir, FsPermission.getDirDefault());
+    fs.createEncryptionZone(topEZDir, TOP_EZ_KEY);
+    fs.mkdir(nestedEZDir, FsPermission.getDirDefault());
+    fs.createEncryptionZone(nestedEZDir, NESTED_EZ_KEY);
+
+    // create files
+    DFSTestUtil.createFile(fs, topEZBaseFile, len, (short) 1, 0xFEED);
+    DFSTestUtil.createFile(fs, topEZFile, len, (short) 1, 0xFEED);
+    DFSTestUtil.createFile(fs, nestedEZBaseFile, len, (short) 1, 0xFEED);
+    DFSTestUtil.createFile(fs, nestedEZFile, len, (short) 1, 0xFEED);
 
-    try {
-      fs.rename(new Path(rootDir, "newTopEZDir/nestedEZDir"),
-          new Path(rootDir, "newTopEZDir/newNestedEZDir"));
-    } catch (Exception e) {
-      fail("Should be able to rename the nested EZ dir within " +
-          "the same top EZ.");
-    }
   }
 
   private void verifyEncryption() throws Exception {