fix: PII permission (#16149)

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java

@@ -477,7 +477,8 @@ public class TableRepository extends EntityRepository<Table> {
         tableProfiles, startTs.toString(), endTs.toString(), tableProfiles.size());
   }
 
-  public ResultList<ColumnProfile> getColumnProfiles(String fqn, Long startTs, Long endTs) {
+  public ResultList<ColumnProfile> getColumnProfiles(
+      String fqn, Long startTs, Long endTs, boolean authorizePII) {
     List<ColumnProfile> columnProfiles;
     columnProfiles =
         JsonUtils.readObjects(
@@ -490,8 +491,15 @@ public class TableRepository extends EntityRepository<Table> {
                     endTs,
                     EntityTimeSeriesDAO.OrderBy.DESC),
             ColumnProfile.class);
-    return new ResultList<>(
+    ResultList<ColumnProfile> columnProfileResultList =
-        columnProfiles, startTs.toString(), endTs.toString(), columnProfiles.size());
+        new ResultList<>(
+            columnProfiles, startTs.toString(), endTs.toString(), columnProfiles.size());
+    if (!authorizePII) {
+      // Mask the PII data
+      columnProfileResultList.setData(
+          PIIMasker.getColumnProfile(fqn, columnProfileResultList.getData()));
+    }
+    return columnProfileResultList;
   }
 
   public ResultList<SystemProfile> getSystemProfiles(String fqn, Long startTs, Long endTs) {

openmetadata-service/src/main/java/org/openmetadata/service/resources/databases/TableResource.java

@@ -75,6 +75,7 @@ import org.openmetadata.service.resources.EntityResource;
 import org.openmetadata.service.security.Authorizer;
 import org.openmetadata.service.security.policyevaluator.OperationContext;
 import org.openmetadata.service.security.policyevaluator.ResourceContext;
+import org.openmetadata.service.util.FullyQualifiedName;
 import org.openmetadata.service.util.JsonUtils;
 import org.openmetadata.service.util.ResultList;
 
@@ -926,8 +927,13 @@ public class TableResource extends EntityResource<Table, TableRepository> {
           Long endTs) {
     OperationContext operationContext =
         new OperationContext(entityType, MetadataOperation.VIEW_DATA_PROFILE);
-    authorizer.authorize(securityContext, operationContext, getResourceContextByName(fqn));
+    String tableFqn =
-    return repository.getColumnProfiles(fqn, startTs, endTs);
+        FullyQualifiedName.getTableFQN(
+            fqn); // get table fqn for the resource context (vs column fqn)
+    ResourceContext<?> resourceContext = getResourceContextByName(tableFqn);
+    authorizer.authorize(securityContext, operationContext, resourceContext);
+    boolean authorizePII = authorizer.authorizePII(securityContext, resourceContext.getOwner());
+    return repository.getColumnProfiles(fqn, startTs, endTs, authorizePII);
   }
 
   @GET

openmetadata-service/src/main/java/org/openmetadata/service/security/mask/PIIMasker.java

@@ -3,6 +3,7 @@ package org.openmetadata.service.security.mask;
 import static org.openmetadata.common.utils.CommonUtil.listOrEmpty;
 import static org.openmetadata.service.jdbi3.TopicRepository.getAllFieldTags;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -20,6 +21,7 @@ import org.openmetadata.schema.entity.data.Topic;
 import org.openmetadata.schema.entity.teams.User;
 import org.openmetadata.schema.tests.TestCase;
 import org.openmetadata.schema.type.Column;
+import org.openmetadata.schema.type.ColumnProfile;
 import org.openmetadata.schema.type.Field;
 import org.openmetadata.schema.type.Include;
 import org.openmetadata.schema.type.TableData;
@@ -30,6 +32,7 @@ import org.openmetadata.service.Entity;
 import org.openmetadata.service.jdbi3.ColumnUtil;
 import org.openmetadata.service.resources.feeds.MessageParser;
 import org.openmetadata.service.security.Authorizer;
+import org.openmetadata.service.util.FullyQualifiedName;
 import org.openmetadata.service.util.ResultList;
 
 public class PIIMasker {
@@ -137,6 +140,22 @@ public class PIIMasker {
     return table;
   }
 
+  public static List<ColumnProfile> getColumnProfile(
+      String fqn, List<ColumnProfile> columnProfiles) {
+    Table table =
+        Entity.getEntityByName(
+            Entity.TABLE, FullyQualifiedName.getTableFQN(fqn), "columns,tags", Include.ALL);
+    Column column =
+        table.getColumns().stream()
+            .filter(c -> c.getFullyQualifiedName().equals(fqn))
+            .findFirst()
+            .orElse(null);
+    if (column != null && hasPiiSensitiveTag(column)) {
+      return Collections.nCopies(columnProfiles.size(), new ColumnProfile());
+    }
+    return columnProfiles;
+  }
+
   private static TestCase getTestCase(Column column, TestCase testCase) {
     if (!hasPiiSensitiveTag(column)) return testCase;
 

openmetadata-service/src/test/java/org/openmetadata/service/resources/databases/TableResourceTest.java

@@ -2484,15 +2484,38 @@ public class TableResourceTest extends EntityResourceTest<Table, CreateTable> {
             createRequest(test, 1).withOwner(USER_TEAM21.getEntityReference()), ADMIN_AUTH_HEADERS);
     putTableProfile(table, table1, ADMIN_AUTH_HEADERS);
 
+    Column c3 = table.getColumns().stream().filter(c -> c.getName().equals(C3)).findFirst().get();
+
     // Owner can read the column profile of C3
     Table tableWithProfileFromOwner =
         getLatestTableProfile(table.getFullyQualifiedName(), authHeaders(USER_TEAM21.getName()));
     assertNotNull(tableWithProfileFromOwner.getColumns().get(2).getProfile());
+    ResultList<ColumnProfile> columnProfiles =
+        getColumnProfiles(
+            c3.getFullyQualifiedName(),
+            TestUtils.dateToTimestamp("2021-09-01"),
+            TestUtils.dateToTimestamp("2021-09-30"),
+            authHeaders(USER_TEAM21.getName()));
+    for (ColumnProfile columnProfile : columnProfiles.getData()) {
+      assertNotNull(columnProfile.getMax());
+      assertNotNull(columnProfile.getMin());
+    }
 
     // Non owners cannot read the column profile of C3
     Table tableWithProfileFromNotOwner =
         getLatestTableProfile(table.getFullyQualifiedName(), authHeaders(USER1_REF.getName()));
     assertNull(tableWithProfileFromNotOwner.getColumns().get(2).getProfile());
+    ResultList<ColumnProfile> maskedColumnProfiles =
+        getColumnProfiles(
+            c3.getFullyQualifiedName(),
+            TestUtils.dateToTimestamp("2021-09-01"),
+            TestUtils.dateToTimestamp("2021-09-30"),
+            authHeaders(USER1_REF.getName()));
+    for (ColumnProfile columnProfile : maskedColumnProfiles.getData()) {
+      assertNull(columnProfile.getMax());
+      assertNull(columnProfile.getMin());
+    }
+    assertEquals(maskedColumnProfiles.getData().size(), columnProfiles.getData().size());
   }
 
   @Test