HDFS-15053. RBF: Add permission check for safemode operation. Contributed by Xiaoqiao He.

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java

@@ -241,6 +241,13 @@ public class RouterAdminServer extends AbstractService
     return this.adminAddress;
   }
 
+  void checkSuperuserPrivilege() throws AccessControlException {
+    RouterPermissionChecker pc = RouterAdminServer.getPermissionChecker();
+    if (pc != null) {
+      pc.checkSuperuserPrivilege();
+    }
+  }
+
   @Override
   protected void serviceInit(Configuration configuration) throws Exception {
     this.conf = configuration;
@@ -392,6 +399,7 @@ public class RouterAdminServer extends AbstractService
   @Override
   public EnterSafeModeResponse enterSafeMode(EnterSafeModeRequest request)
       throws IOException {
+    checkSuperuserPrivilege();
     boolean success = false;
     RouterSafemodeService safeModeService = this.router.getSafemodeService();
     if (safeModeService != null) {
@@ -412,6 +420,7 @@ public class RouterAdminServer extends AbstractService
   @Override
   public LeaveSafeModeResponse leaveSafeMode(LeaveSafeModeRequest request)
       throws IOException {
+    checkSuperuserPrivilege();
     boolean success = false;
     RouterSafemodeService safeModeService = this.router.getSafemodeService();
     if (safeModeService != null) {
@@ -508,11 +517,7 @@ public class RouterAdminServer extends AbstractService
   @Override
   public DisableNameserviceResponse disableNameservice(
       DisableNameserviceRequest request) throws IOException {
-
+    checkSuperuserPrivilege();
-    RouterPermissionChecker pc = getPermissionChecker();
-    if (pc != null) {
-      pc.checkSuperuserPrivilege();
-    }
 
     String nsId = request.getNameServiceId();
     boolean success = false;
@@ -545,10 +550,7 @@ public class RouterAdminServer extends AbstractService
   @Override
   public EnableNameserviceResponse enableNameservice(
       EnableNameserviceRequest request) throws IOException {
-    RouterPermissionChecker pc = getPermissionChecker();
+    checkSuperuserPrivilege();
-    if (pc != null) {
-      pc.checkSuperuserPrivilege();
-    }
 
     String nsId = request.getNameServiceId();
     DisabledNameserviceStore store = getDisabledNameserviceStore();

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java

@@ -862,6 +862,43 @@ public class TestRouterAdminCLI {
     assertTrue(out.toString().contains("false"));
   }
 
+  @Test
+  public void testSafeModePermission() throws Exception {
+    // ensure the Router become RUNNING state
+    waitState(RouterServiceState.RUNNING);
+    assertFalse(routerContext.getRouter().getSafemodeService().isInSafeMode());
+
+    UserGroupInformation superUser = UserGroupInformation.createRemoteUser(
+        UserGroupInformation.getCurrentUser().getShortUserName());
+    UserGroupInformation remoteUser = UserGroupInformation
+        .createRemoteUser(TEST_USER);
+    try {
+      // use normal user as current user to test
+      UserGroupInformation.setLoginUser(remoteUser);
+      assertEquals(-1,
+          ToolRunner.run(admin, new String[]{"-safemode", "enter"}));
+
+      // set back login user
+      UserGroupInformation.setLoginUser(superUser);
+      assertEquals(0,
+          ToolRunner.run(admin, new String[]{"-safemode", "enter"}));
+
+      // use normal user as current user to test
+      UserGroupInformation.setLoginUser(remoteUser);
+      assertEquals(-1,
+          ToolRunner.run(admin, new String[]{"-safemode", "leave"}));
+
+      // set back login user
+      UserGroupInformation.setLoginUser(superUser);
+      assertEquals(0,
+          ToolRunner.run(admin, new String[]{"-safemode", "leave"}));
+    } finally {
+      // set back login user to make sure it doesn't pollute other unit tests
+      // even this one fails.
+      UserGroupInformation.setLoginUser(superUser);
+    }
+  }
+
   @Test
   public void testCreateInvalidEntry() throws Exception {
     String[] argv = new String[] {