HDDS-4704. Add permission check in OMDBCheckpointServlet

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMDbCheckpointServlet.java

@@ -41,9 +41,12 @@
 
 import org.apache.commons.io.FileUtils;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_ENABLED;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ADMINISTRATORS;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_OPEN_KEY_EXPIRE_THRESHOLD_SECONDS;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_REQUEST_FLUSH;
 import static org.apache.hadoop.ozone.om.OMDBCheckpointServlet.writeOmDBCheckpointToStream;
+
+import org.apache.hadoop.security.UserGroupInformation;
 import org.junit.After;
 import org.junit.Assert;
 import static org.junit.Assert.assertNotNull;
@@ -88,6 +91,8 @@ public void init() throws Exception {
     scmId = UUID.randomUUID().toString();
     omId = UUID.randomUUID().toString();
     conf.setBoolean(OZONE_ACL_ENABLED, true);
+    conf.set(OZONE_ADMINISTRATORS,
+        UserGroupInformation.getCurrentUser().getUserName());
     conf.setInt(OZONE_OPEN_KEY_EXPIRE_THRESHOLD_SECONDS, 2);
     cluster = MiniOzoneCluster.newBuilder(conf)
         .setClusterId(clusterId)
@@ -119,6 +124,9 @@ public void testDoGet() throws ServletException, IOException {
       doCallRealMethod().when(omDbCheckpointServletMock).init();
 
       HttpServletRequest requestMock = mock(HttpServletRequest.class);
+      // Return current user short name when asked
+      when(requestMock.getRemoteUser())
+          .thenReturn(UserGroupInformation.getCurrentUser().getShortUserName());
       HttpServletResponse responseMock = mock(HttpServletResponse.class);
 
       ServletContext servletContextMock = mock(ServletContext.class);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMDBCheckpointServlet.java

@@ -30,6 +30,7 @@
 import java.nio.file.Path;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Collection;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -47,7 +48,10 @@
 import org.apache.commons.compress.compressors.CompressorStreamFactory;
 import org.apache.commons.compress.utils.IOUtils;
 import org.apache.commons.lang3.StringUtils;
+
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ADMINISTRATORS_WILDCARD;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_REQUEST_FLUSH;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -84,11 +88,33 @@ public void init() throws ServletException {
         OMConfigKeys.OZONE_DB_CHECKPOINT_TRANSFER_RATE_KEY,
         OMConfigKeys.OZONE_DB_CHECKPOINT_TRANSFER_RATE_DEFAULT);
 
+    // TODO: Add servlet
+//    getServletContext().addServlet(OMDBCheckpointAuthorizedServlet);
+
     if (transferBandwidth > 0) {
       throttler = new DataTransferThrottler(transferBandwidth);
     }
   }
 
+  private boolean checkAcls(String username) {
+    // Check ACL for dbCheckpoint only when global Ozone ACL is enabled
+    if (om.getAclsEnabled()) {
+      // Only Ozone admins are allowed
+      try {
+        Collection<String> admins = om.getOzoneAdmins(om.getConfiguration());
+        if (admins.contains(OZONE_ADMINISTRATORS_WILDCARD) ||
+            admins.contains(username)) {
+          return true;
+        }
+      } catch (IOException e) {
+        LOG.warn("Error checking permission: {}", e.getMessage());
+      }
+      return false;
+    } else {
+      return true;
+    }
+  }
+
   /**
    * Process a GET request for the Ozone Manager DB checkpoint snapshot.
    *
@@ -106,6 +132,37 @@ public void doGet(HttpServletRequest request, HttpServletResponse response) {
       return;
     }
 
+    // Check ACL for dbCheckpoint only when global Ozone ACL is enable
+    if (om.getAclsEnabled()) {
+      final String remoteUser = request.getRemoteUser();
+      final java.security.Principal userPrincipal = request.getUserPrincipal();
+      if (userPrincipal == null) {
+        // Fallback to checking login user if userPrincipal is null.
+        // Note: In prod, a secure cluster would deploy Kerberos so this case
+        // shouldn't be hit. This is here for UT and dev testing.
+        if (!checkAcls(remoteUser)) {
+          LOG.error("Permission denied: Current login user '{}' has not been "
+              + "authenticated to access /dbCheckpoint.", remoteUser);
+          response.setStatus(HttpServletResponse.SC_FORBIDDEN);
+          return;
+        } else {
+          LOG.info("Granted login user '{}' access to /dbCheckpoint.",
+              remoteUser);
+        }
+      } else {
+        final String userPrincipalName = userPrincipal.getName();
+        if (!checkAcls(userPrincipalName)) {
+          LOG.error("Permission denied: User principal '{}' doesn't have the "
+              + "permission to access /dbCheckpoint.", userPrincipalName);
+          response.setStatus(HttpServletResponse.SC_FORBIDDEN);
+          return;
+        } else {
+          LOG.info("Granted user principal '{}' access to /dbCheckpoint.",
+              userPrincipalName);
+        }
+      }
+    }
+
     DBCheckpoint checkpoint = null;
     try {
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -3551,7 +3551,7 @@ public OzoneDelegationTokenSecretManager getDelegationTokenMgr() {
   /**
    * Return list of OzoneAdministrators.
    */
-  private Collection<String> getOzoneAdmins(OzoneConfiguration conf)
+  Collection<String> getOzoneAdmins(OzoneConfiguration conf)
       throws IOException {
     Collection<String> ozAdmins =
         conf.getTrimmedStringCollection(OZONE_ADMINISTRATORS);