HDDS-13826. Move ACL check in OMKeySetTimesRequest

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

@@ -17,11 +17,13 @@
 
 package org.apache.hadoop.ozone.om;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS;
 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.om.exceptions.OMException.ResultCodes.PERMISSION_DENIED;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -31,6 +33,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
+import java.util.concurrent.TimeoutException;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.utils.IOUtils;
@@ -41,15 +44,18 @@
 import org.apache.hadoop.ozone.client.OzoneBucket;
 import org.apache.hadoop.ozone.client.OzoneClient;
 import org.apache.hadoop.ozone.client.OzoneClientFactory;
+import org.apache.hadoop.ozone.client.OzoneKey;
 import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.VolumeArgs;
+import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.ozone.security.acl.OzoneNativeAuthorizer;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.ozone.test.GenericTestUtils;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
 
@@ -76,16 +82,20 @@ public class TestOMHALeaderSpecificACLEnforcement {
   private MiniOzoneHAClusterImpl cluster;
   private OzoneClient client;
   private UserGroupInformation testUserUgi;
+  private UserGroupInformation adminUserUgi;
+  private OzoneManager theLeaderOM;
 
   @BeforeAll
   public void init() throws Exception {
     // Create test user
     testUserUgi = UserGroupInformation.createUserForTesting(TEST_USER, new String[]{"testgroup"});
+    adminUserUgi = UserGroupInformation.getCurrentUser();
 
     // Set up and start the cluster
     setupCluster();
 
     // Create admin volume that will be used for bucket permission testing
+    theLeaderOM = cluster.getOMLeader();
     createAdminVolume();
   }
 
@@ -97,6 +107,22 @@ public void shutdown() {
     }
   }
 
+  @BeforeEach
+  public void restoreLeadership() throws IOException, InterruptedException, TimeoutException {
+    OzoneManager currentLeader = cluster.getOMLeader();
+    if (!currentLeader.getOMNodeId().equals(theLeaderOM.getOMNodeId())) {
+      currentLeader.transferLeadership(theLeaderOM.getOMNodeId());
+      GenericTestUtils.waitFor(() -> {
+        try {
+          OzoneManager currentLeaderCheck = cluster.getOMLeader();
+          return !currentLeaderCheck.getOMNodeId().equals(currentLeader.getOMNodeId());
+        } catch (Exception e) {
+          return false;
+        }
+      }, 1000, 30000);
+    }
+  }
+
   /**
    * Main test method that validates leader-specific ACL enforcement in OM HA.
    * 1. Creates a mini cluster with OM HA
@@ -165,7 +191,7 @@ private OzoneConfiguration createBaseConfiguration() throws IOException {
     conf.setBoolean(OZONE_ACL_ENABLED, true);
 
     // Set current user as initial admin (needed for cluster setup)
-    String currentUser = UserGroupInformation.getCurrentUser().getShortUserName();
+    String currentUser = adminUserUgi.getShortUserName();
     conf.set(OZONE_ADMINISTRATORS, currentUser);
 
     return conf;
@@ -181,7 +207,7 @@ private void createAdminVolume() throws Exception {
 
     // Create volume as admin user
     VolumeArgs volumeArgs = VolumeArgs.newBuilder()
-        .setOwner(UserGroupInformation.getCurrentUser().getShortUserName())
+        .setOwner(adminUserUgi.getShortUserName())
         .build();
 
     adminObjectStore.createVolume(ADMIN_VOLUME, volumeArgs);
@@ -260,7 +286,90 @@ private void testVolumeAndBucketCreationAsUser(boolean shouldSucceed) throws Exc
       }
     } finally {
       // Reset to original user
-      UserGroupInformation.setLoginUser(UserGroupInformation.getCurrentUser());
+      UserGroupInformation.setLoginUser(adminUserUgi);
+    }
+  }
+
+  /**
+   * Tests that setTimes ACL check is enforced in preExecute and is leader-specific.
+   * 1. Creates a key with admin user
+   * 2. Adds test user as admin on the current leader
+   * 3. Verifies that test user (as admin) can setTimes on key owned by someone else
+   * 4. Transfers leadership to another node
+   * 5. Verifies that setTimes fails with PERMISSION_DENIED when test user is no longer admin
+   */
+  @Test
+  public void testKeySetTimesAclEnforcementAfterLeadershipChange() throws Exception {
+    // Step 1: Create a volume, bucket, and key as the admin user
+    ObjectStore adminObjectStore = client.getObjectStore();
+    String keyTestVolume = "keyvol-" +
+        RandomStringUtils.secure().nextAlphabetic(5).toLowerCase(Locale.ROOT);
+    String keyTestBucket = "keybucket-" +
+        RandomStringUtils.secure().nextAlphabetic(5).toLowerCase(Locale.ROOT);
+    String keyName = "testkey-" +
+        RandomStringUtils.secure().nextAlphabetic(5).toLowerCase(Locale.ROOT);
+
+    String adminUser = adminUserUgi.getShortUserName();
+    VolumeArgs volumeArgs = VolumeArgs.newBuilder()
+        .setOwner(adminUser)
+        .build();
+    adminObjectStore.createVolume(keyTestVolume, volumeArgs);
+    OzoneVolume adminVolume = adminObjectStore.getVolume(keyTestVolume);
+
+    BucketArgs bucketArgs = BucketArgs.newBuilder().build();
+    adminVolume.createBucket(keyTestBucket, bucketArgs);
+    OzoneBucket adminBucket = adminVolume.getBucket(keyTestBucket);
+
+    // Create a key as admin (so test user is NOT the owner)
+    try (OzoneOutputStream out = adminBucket.createKey(keyName, 0)) {
+      out.write("test data".getBytes(UTF_8));
+    }
+
+    OzoneKey key = adminBucket.getKey(keyName);
+    assertNotNull(key, "Key should be created successfully");
+    long originalMtime = key.getModificationTime().toEpochMilli();
+
+    // Step 2: Get the current leader and add test user as admin
+    OzoneManager currentLeader = cluster.getOMLeader();
+    String leaderNodeId = currentLeader.getOMNodeId();
+    addAdminToSpecificOM(currentLeader, TEST_USER);
+
+    // Verify admin was added
+    assertTrue(currentLeader.getOmAdminUsernames().contains(TEST_USER),
+        "Test user should be admin on leader OM");
+
+    // Switch to test user and try setTimes as admin (should succeed)
+    UserGroupInformation.setLoginUser(testUserUgi);
+    try (OzoneClient userClient = OzoneClientFactory.getRpcClient(OM_SERVICE_ID, cluster.getConf())) {
+      ObjectStore userObjectStore = userClient.getObjectStore();
+      OzoneVolume userVolume = userObjectStore.getVolume(keyTestVolume);
+      OzoneBucket userBucket = userVolume.getBucket(keyTestBucket);
+
+      long newMtime = System.currentTimeMillis();
+      userBucket.setTimes(keyName, newMtime, -1);
+
+      // Verify the modification time was updated
+      key = userBucket.getKey(keyName);
+      assertEquals(newMtime, key.getModificationTime().toEpochMilli(),
+          "Modification time should be updated by admin user");
+      assertNotEquals(originalMtime, key.getModificationTime().toEpochMilli(),
+          "Modification time should have changed");
+
+      OzoneManager newLeader = transferLeadershipToAnotherNode(currentLeader);
+      assertNotEquals(leaderNodeId, newLeader.getOMNodeId(),
+          "Leadership should have transferred to a different node");
+      assertFalse(newLeader.getOmAdminUsernames().contains(TEST_USER),
+          "Test user should NOT be admin on new leader OM");
+
+      long anotherMtime = System.currentTimeMillis() + 10000;
+      OMException exception = assertThrows(OMException.class, () -> {
+        userBucket.setTimes(keyName, anotherMtime, -1);
+      }, "setTimes should fail for non-admin user on new leader");
+      assertEquals(PERMISSION_DENIED, exception.getResult(),
+          "Should get PERMISSION_DENIED when ACL check fails in preExecute");
+    } finally {
+      // Reset to original user
+      UserGroupInformation.setLoginUser(adminUserUgi);
     }
   }
 

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

@@ -78,6 +78,27 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
 
     OzoneManagerProtocolProtos.KeyArgs newKeyArgs = resolveBucketLink(ozoneManager, keyArgs);
 
+    // ACL check during preExecute
+    if (ozoneManager.getAclsEnabled()) {
+      try {
+        checkAcls(ozoneManager, OzoneObj.ResourceType.KEY,
+            OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.WRITE_ACL,
+            newKeyArgs.getVolumeName(), newKeyArgs.getBucketName(), newKeyArgs.getKeyName());
+      } catch (IOException ex) {
+        // Ensure audit log captures preExecute failures
+        Map<String, String> auditMap = new LinkedHashMap<>();
+        auditMap.put(OzoneConsts.VOLUME, newKeyArgs.getVolumeName());
+        auditMap.put(OzoneConsts.BUCKET, newKeyArgs.getBucketName());
+        auditMap.put(OzoneConsts.KEY, newKeyArgs.getKeyName());
+        auditMap.put(OzoneConsts.MODIFICATION_TIME,
+            String.valueOf(getModificationTime()));
+        markForAudit(ozoneManager.getAuditLogger(),
+            buildAuditMessage(OMAction.SET_TIMES, auditMap, ex,
+                getOmRequest().getUserInfo()));
+        throw ex;
+      }
+    }
+
     return request.toBuilder()
         .setSetTimesRequest(
             setTimesRequest.toBuilder()
@@ -194,12 +215,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       bucket = getBucketName();
       key = getKeyName();
 
-      // check Acl
-      if (ozoneManager.getAclsEnabled()) {
-        checkAcls(ozoneManager, OzoneObj.ResourceType.KEY,
-            OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.WRITE_ACL,
-            volume, bucket, key);
-      }
       mergeOmLockDetails(
           omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volume,
               bucket));

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

@@ -42,8 +42,6 @@
 import org.apache.hadoop.ozone.om.response.key.OMKeySetTimesResponseWithFSO;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
-import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
-import org.apache.hadoop.ozone.security.acl.OzoneObj;
 
 /**
  * Handle set times request for bucket for prefix layout.
@@ -53,6 +51,7 @@ public class OMKeySetTimesRequestWithFSO extends OMKeySetTimesRequest {
   @Override
   public OzoneManagerProtocolProtos.OMRequest preExecute(
       OzoneManager ozoneManager) throws IOException {
+    // The parent class handles ACL checks in preExecute, so just call super
     return super.preExecute(ozoneManager);
   }
 
@@ -82,12 +81,6 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       bucket = getBucketName();
       key = getKeyName();
 
-      // check Acl
-      if (ozoneManager.getAclsEnabled()) {
-        checkAcls(ozoneManager, OzoneObj.ResourceType.KEY,
-            OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.WRITE_ACL,
-            volume, bucket, key);
-      }
       mergeOmLockDetails(omMetadataManager.getLock()
           .acquireWriteLock(BUCKET_LOCK, volume, bucket));
       lockAcquired = getOmLockDetails().isLockAcquired();