HDDS-15961. Resolve linked bucket source properties consistently

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

@@ -406,18 +406,19 @@ public Builder toBuilder() {
    */
   public OmBucketInfo withOperationalPropertiesFrom(OmBucketInfo source) {
     return toBuilder()
-        .setDefaultReplicationConfig(source.getDefaultReplicationConfig())
-        .setIsVersionEnabled(source.getIsVersionEnabled())
         .setStorageType(source.getStorageType())
-        .setQuotaInBytes(source.getQuotaInBytes())
-        .setQuotaInNamespace(source.getQuotaInNamespace())
+        .setIsVersionEnabled(source.getIsVersionEnabled())
+        .setBucketEncryptionKey(source.getEncryptionKeyInfo())
         .setUsedBytes(source.getUsedBytes())
         .setUsedNamespace(source.getUsedNamespace())
+        .setQuotaInBytes(source.getQuotaInBytes())
+        .setQuotaInNamespace(source.getQuotaInNamespace())
         .setSnapshotUsedBytes(source.getSnapshotUsedBytes())
         .setSnapshotUsedNamespace(source.getSnapshotUsedNamespace())
-        .addAllMetadata(source.getMetadata())
         .setBucketLayout(source.getBucketLayout())
+        .setDefaultReplicationConfig(source.getDefaultReplicationConfig())
         .setTags(source.getTags())
+        .addAllMetadata(source.getMetadata())
         .build();
   }
 

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmBucketInfo.java

@@ -147,6 +147,27 @@ public void testWithOperationalPropertiesFromPreservesLinkIdentity() {
     assertEquals("linkValue", resolvedLink.getMetadata().get("linkKey"));
   }
 
+  @Test
+  public void testWithOperationalPropertiesFromCopiesEncryptionInfo() {
+    BucketEncryptionKeyInfo encryptionKeyInfo =
+        new BucketEncryptionKeyInfo.Builder().setKeyName("key1").build();
+    OmBucketInfo source = OmBucketInfo.newBuilder()
+        .setVolumeName("vol1")
+        .setBucketName("source")
+        .setBucketEncryptionKey(encryptionKeyInfo)
+        .build();
+    OmBucketInfo link = OmBucketInfo.newBuilder()
+        .setVolumeName("vol1")
+        .setBucketName("link")
+        .setSourceVolume("vol1")
+        .setSourceBucket("source")
+        .build();
+
+    OmBucketInfo resolvedLink = link.withOperationalPropertiesFrom(source);
+
+    assertEquals(encryptionKeyInfo, resolvedLink.getEncryptionKeyInfo());
+  }
+
   @Test
   public void getProtobufMessageEC() {
     OmBucketInfo omBucketInfo =

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

@@ -3062,7 +3062,9 @@ private OmBucketInfo enrichLinkBucketInfo(
       return bucketInfo;
     }
     OmBucketInfo realBucket = getResolvedSourceBucket(resolvedBucket, resolvedSourceCache);
-    return bucketInfo.withOperationalPropertiesFrom(realBucket);
+    return realBucket != null
+        ? bucketInfo.withOperationalPropertiesFrom(realBucket)
+        : bucketInfo;
   }
 
   private OmBucketInfo getResolvedSourceBucket(
@@ -3078,6 +3080,18 @@ private OmBucketInfo getResolvedSourceBucket(
         return cachedSource;
       }
     }
+    if (getAclsEnabled()) {
+      try {
+        omMetadataReader.checkAcls(ResourceType.BUCKET, StoreType.OZONE,
+            ACLType.READ, resolvedBucket.realVolume(),
+            resolvedBucket.realBucket(), null);
+      } catch (OMException e) {
+        if (e.getResult() == PERMISSION_DENIED) {
+          return null;
+        }
+        throw e;
+      }
+    }
     OmBucketInfo realBucket = bucketManager.getBucketInfo(
         resolvedBucket.realVolume(),
         resolvedBucket.realBucket());

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestBucketManagerImpl.java

@@ -18,18 +18,30 @@
 package org.apache.hadoop.ozone.om;
 
 import static java.util.Collections.singletonMap;
+import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.READ;
+import static org.apache.hadoop.ozone.security.acl.OzoneObj.ResourceType.BUCKET;
+import static org.apache.hadoop.ozone.security.acl.OzoneObj.StoreType.OZONE;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import org.apache.hadoop.crypto.key.KeyProvider;
@@ -41,7 +53,9 @@
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.scm.HddsWhiteboxTestUtils;
 import org.apache.hadoop.hdds.server.ServerUtils;
+import org.apache.hadoop.ozone.audit.AuditMessage;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes;
 import org.apache.hadoop.ozone.om.helpers.BucketEncryptionKeyInfo;
@@ -540,4 +554,122 @@ public void testListBucketsResolvesFsoAndObsLinkLayouts() throws Exception {
     assertEquals(volume, listedObsLink.getSourceVolume());
     assertEquals("obs-source", listedObsLink.getSourceBucket());
   }
+
+  @Test
+  void testGetBucketInfoReturnsLinkWithoutSourceReadAccess() throws Exception {
+    String linkVolume = volumeName();
+    String sourceVolume = volumeName();
+    OmBucketInfo link = createLinkBucketInfo(linkVolume, sourceVolume);
+    OmBucketInfo source = createSourceBucketInfo(sourceVolume);
+    BucketManager bucketManager = mock(BucketManager.class);
+    when(bucketManager.getBucketInfo(linkVolume, "link")).thenReturn(link);
+    when(bucketManager.getBucketInfo(sourceVolume, "source")).thenReturn(source);
+    OmMetadataReader metadataReader = mock(OmMetadataReader.class);
+    denySourceRead(metadataReader, sourceVolume);
+    OzoneManager omSpy = createAclEnabledOmSpy(bucketManager, metadataReader);
+
+    OmBucketInfo result = omSpy.getBucketInfo(linkVolume, "link");
+
+    assertThat(result.getMetadata())
+        .containsEntry("linkKey", "linkValue")
+        .doesNotContainKey("sourceKey");
+    verify(metadataReader).checkAcls(BUCKET, OZONE, READ, sourceVolume, "source", null);
+    verify(bucketManager, times(1)).getBucketInfo(sourceVolume, "source");
+  }
+
+  @Test
+  void testListBucketsDoesNotCopySourcePropertiesWithoutReadAccess() throws Exception {
+    String linkVolume = volumeName();
+    String sourceVolume = volumeName();
+    OmBucketInfo link = createLinkBucketInfo(linkVolume, sourceVolume);
+    OmBucketInfo source = createSourceBucketInfo(sourceVolume);
+    OmBucketInfo regular = OmBucketInfo.newBuilder()
+        .setVolumeName(linkVolume)
+        .setBucketName("regular")
+        .build();
+    BucketManager bucketManager = mock(BucketManager.class);
+    when(bucketManager.listBuckets(linkVolume, "", "", 100, false))
+        .thenReturn(new ArrayList<>(Arrays.asList(link, regular)));
+    when(bucketManager.getBucketInfo(sourceVolume, "source")).thenReturn(source);
+    OmMetadataReader metadataReader = mock(OmMetadataReader.class);
+    denySourceRead(metadataReader, sourceVolume);
+    OzoneManager omSpy = createAclEnabledOmSpy(bucketManager, metadataReader);
+
+    List<OmBucketInfo> result = omSpy.listBuckets(linkVolume, "", "", 100, false);
+
+    assertThat(result.get(0).getMetadata())
+        .containsEntry("linkKey", "linkValue")
+        .doesNotContainKey("sourceKey");
+    assertSame(regular, result.get(1));
+    verify(metadataReader).checkAcls(BUCKET, OZONE, READ, sourceVolume, "source", null);
+    verify(bucketManager, times(1)).getBucketInfo(sourceVolume, "source");
+  }
+
+  @Test
+  void testListBucketsChecksSourceReadAccessOncePerSource() throws Exception {
+    String linkVolume = volumeName();
+    String sourceVolume = volumeName();
+    OmBucketInfo firstLink = createLinkBucketInfo(linkVolume, sourceVolume)
+        .toBuilder()
+        .setBucketName("link1")
+        .build();
+    OmBucketInfo secondLink = createLinkBucketInfo(linkVolume, sourceVolume)
+        .toBuilder()
+        .setBucketName("link2")
+        .build();
+    OmBucketInfo source = createSourceBucketInfo(sourceVolume);
+    BucketManager bucketManager = mock(BucketManager.class);
+    when(bucketManager.listBuckets(linkVolume, "", "", 100, false))
+        .thenReturn(new ArrayList<>(Arrays.asList(firstLink, secondLink)));
+    when(bucketManager.getBucketInfo(sourceVolume, "source")).thenReturn(source);
+    OmMetadataReader metadataReader = mock(OmMetadataReader.class);
+    OzoneManager omSpy = createAclEnabledOmSpy(bucketManager, metadataReader);
+
+    List<OmBucketInfo> result = omSpy.listBuckets(linkVolume, "", "", 100, false);
+
+    assertEquals("sourceValue", result.get(0).getMetadata().get("sourceKey"));
+    assertEquals("sourceValue", result.get(1).getMetadata().get("sourceKey"));
+    verify(metadataReader, times(1))
+        .checkAcls(BUCKET, OZONE, READ, sourceVolume, "source", null);
+  }
+
+  private static OmBucketInfo createLinkBucketInfo(String linkVolume, String sourceVolume) {
+    return OmBucketInfo.newBuilder()
+        .setVolumeName(linkVolume)
+        .setBucketName("link")
+        .setSourceVolume(sourceVolume)
+        .setSourceBucket("source")
+        .addAllMetadata(singletonMap("linkKey", "linkValue"))
+        .build();
+  }
+
+  private static OmBucketInfo createSourceBucketInfo(String sourceVolume) {
+    return OmBucketInfo.newBuilder()
+        .setVolumeName(sourceVolume)
+        .setBucketName("source")
+        .addAllMetadata(singletonMap("sourceKey", "sourceValue"))
+        .build();
+  }
+
+  private static void denySourceRead(OmMetadataReader metadataReader, String sourceVolume) throws IOException {
+    doAnswer(invocation -> {
+      if (sourceVolume.equals(invocation.getArgument(3)) && "source".equals(invocation.getArgument(4))) {
+        throw new OMException("denied", ResultCodes.PERMISSION_DENIED);
+      }
+      return null;
+    }).when(metadataReader).checkAcls(any(), any(), any(), any(), any(), any());
+  }
+
+  private OzoneManager createAclEnabledOmSpy(BucketManager bucketManager, OmMetadataReader metadataReader) {
+    OzoneManager omSpy = spy(omTestManagers.getOzoneManager());
+    HddsWhiteboxTestUtils.setInternalState(omSpy, "bucketManager", bucketManager);
+    HddsWhiteboxTestUtils.setInternalState(omSpy, "omMetadataReader", metadataReader);
+    when(omSpy.getAclsEnabled()).thenReturn(true);
+    AuditMessage auditMessage = mock(AuditMessage.class);
+    lenient().when(auditMessage.getOp()).thenReturn("READ_BUCKET");
+    lenient().doReturn(auditMessage).when(omSpy).buildAuditMessageForSuccess(any(), anyMap());
+    lenient().doReturn(auditMessage).when(omSpy)
+        .buildAuditMessageForFailure(any(), anyMap(), any(Throwable.class));
+    return omSpy;
+  }
 }