Fixes #6728 API to get detailed permissions for any user (#6729)

* WIP

* Fixes #6728 API to get detailed permissions for any user

catalog-rest-service/src/main/java/org/openmetadata/catalog/ResourceRegistry.java

@@ -4,6 +4,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
+import org.openmetadata.catalog.exception.CatalogExceptionMessage;
 import org.openmetadata.catalog.type.ResourceDescriptor;
 
 public class ResourceRegistry {
@@ -19,4 +20,13 @@ public class ResourceRegistry {
   public static List<ResourceDescriptor> listResourceDescriptors() {
     return Collections.unmodifiableList(RESOURCE_DESCRIPTORS);
   }
+
+  public static ResourceDescriptor getResourceDescriptor(String resourceType) {
+    ResourceDescriptor rd =
+        RESOURCE_DESCRIPTORS.stream().filter(r -> r.getName().equalsIgnoreCase(resourceType)).findAny().orElse(null);
+    if (rd == null) {
+      throw new IllegalArgumentException(CatalogExceptionMessage.resourceTypeNotFound(resourceType));
+    }
+    return rd;
+  }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/exception/CatalogExceptionMessage.java

@@ -56,6 +56,10 @@ public final class CatalogExceptionMessage {
     return String.format("Entity type %s not found", entityType);
   }
 
+  public static String resourceTypeNotFound(String resourceType) {
+    return String.format("Resource type %s not found", resourceType);
+  }
+
   public static String entityTypeNotSupported(String entityType) {
     return String.format("Entity type %s not supported", entityType);
   }

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/EntityRepository.java

@@ -53,6 +53,7 @@ import java.util.function.BiPredicate;
 import javax.json.JsonPatch;
 import javax.ws.rs.core.Response.Status;
 import javax.ws.rs.core.UriInfo;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.maven.shared.utils.io.IOUtil;
 import org.jdbi.v3.sqlobject.transaction.Transaction;
@@ -132,8 +133,8 @@ public abstract class EntityRepository<T extends EntityInterface> {
   protected final CollectionDAO daoCollection;
   protected final List<String> allowedFields;
   public final boolean supportsSoftDelete;
-  protected final boolean supportsTags;
+  @Getter protected final boolean supportsTags;
-  protected final boolean supportsOwner;
+  @Getter protected final boolean supportsOwner;
   protected final boolean supportsFollower;
   protected boolean allowEdits = false;
 

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/EntityResource.java

@@ -199,12 +199,10 @@ public abstract class EntityResource<T extends EntityInterface, K extends Entity
   }
 
   protected ResourceContext getResourceContextById(UUID id) {
-    String fields = supportsOwner ? FIELD_OWNER : null;
+    return ResourceContext.builder().resource(entityType).entityRepository(dao).id(id).build();
-    return ResourceContext.builder().resource(entityType).entityRepository(dao).id(id).fields(fields).build();
   }
 
   protected ResourceContext getResourceContextByName(String name) {
-    String fields = supportsOwner ? FIELD_OWNER : null;
+    return ResourceContext.builder().resource(entityType).entityRepository(dao).name(name).build();
-    return ResourceContext.builder().resource(entityType).entityRepository(dao).name(name).fields(fields).build();
   }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/lineage/LineageResource.java

@@ -232,6 +232,12 @@ public class LineageResource {
   }
 
   class LineageResourceContext implements ResourceContextInterface {
+
+    @Override
+    public String getResource() {
+      return "lineage";
+    }
+
     @Override
     public EntityReference getOwner() throws IOException {
       return null;

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/permissions/PermissionsResource.java

@@ -15,20 +15,28 @@ package org.openmetadata.catalog.resources.permissions;
 
 import io.swagger.annotations.Api;
 import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.media.Content;
 import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import java.util.List;
+import java.util.UUID;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.SecurityContext;
 import lombok.NonNull;
+import org.openmetadata.catalog.Entity;
+import org.openmetadata.catalog.EntityInterface;
 import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.jdbi3.EntityRepository;
 import org.openmetadata.catalog.resources.Collection;
 import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.security.policyevaluator.ResourceContext;
 import org.openmetadata.catalog.type.ResourcePermission;
 import org.openmetadata.catalog.util.ResultList;
 
@@ -46,6 +54,7 @@ public class PermissionsResource {
 
   @GET
   @Operation(
+      operationId = "getResourcePermissions",
       summary = "Get permissions for logged in user",
       tags = "permission",
       responses = {
@@ -57,8 +66,78 @@ public class PermissionsResource {
                     mediaType = "application/json",
                     schema = @Schema(implementation = ResourcePermissionList.class)))
       })
-  public ResultList<ResourcePermission> getPermissions(@Context SecurityContext securityContext) {
+  public ResultList<ResourcePermission> getPermissions(
-    return new ResourcePermissionList(authorizer.listPermissions(securityContext));
+      @Context SecurityContext securityContext,
+      @Parameter(
+              description =
+                  "Permission for user specified in this query param. If not specified, the user is "
+                      + "defaulted to the logged in user",
+              schema = @Schema(type = "string", example = "john"))
+          @QueryParam("user")
+          String user) {
+    return new ResourcePermissionList(authorizer.listPermissions(securityContext, user));
+  }
+
+  @GET
+  @Path("/{resource}")
+  @Operation(
+      operationId = "getResourceTypePermission",
+      summary = "Get permissions a given resource/entity type for logged in user",
+      tags = "permission",
+      responses = {
+        @ApiResponse(
+            responseCode = "200",
+            description = "Permissions for logged in user",
+            content =
+                @Content(
+                    mediaType = "application/json",
+                    schema = @Schema(implementation = ResourcePermissionList.class)))
+      })
+  public ResourcePermission getPermission(
+      @Context SecurityContext securityContext,
+      @Parameter(
+              description =
+                  "Permission for user specified in this query param. If not specified, the user is "
+                      + "defaulted to the logged in user",
+              schema = @Schema(type = "string", example = "john"))
+          @QueryParam("user")
+          String user,
+      @Parameter(description = "Resource type", schema = @Schema(type = "String")) @PathParam("resource")
+          String resource) {
+    return authorizer.getPermission(securityContext, user, resource);
+  }
+
+  @GET
+  @Path("/{resource}/{id}")
+  @Operation(
+      operationId = "getResourcePermission",
+      summary = "Get permissions for a given entity for a logged in user",
+      tags = "permission",
+      responses = {
+        @ApiResponse(
+            responseCode = "200",
+            description = "Permissions for logged in user",
+            content =
+                @Content(
+                    mediaType = "application/json",
+                    schema = @Schema(implementation = ResourcePermissionList.class)))
+      })
+  public ResourcePermission getPermission(
+      @Context SecurityContext securityContext,
+      @Parameter(
+              description =
+                  "Permission for user specified in this query param. If not specified, the user is "
+                      + "defaulted to the logged in user",
+              schema = @Schema(type = "string", example = "john"))
+          @QueryParam("user")
+          String user,
+      @Parameter(description = "Resource type", schema = @Schema(type = "String")) @PathParam("resource")
+          String resource,
+      @Parameter(description = "Entity Id", schema = @Schema(type = "UUID")) @PathParam("id") UUID id) {
+    EntityRepository<EntityInterface> entityRepository = Entity.getEntityRepository(resource);
+    ResourceContext resourceContext =
+        ResourceContext.builder().resource(resource).id(id).entityRepository(entityRepository).build();
+    return authorizer.getPermission(securityContext, user, resourceContext);
   }
 
   static class ResourcePermissionList extends ResultList<ResourcePermission> {

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/Authorizer.java

@@ -27,12 +27,15 @@ public interface Authorizer {
   /** Initialize the authorizer */
   void init(AuthorizerConfiguration config, Jdbi jdbi);
 
-  /**
+  /** Returns a list of operations that the authenticated user (subject) can perform */
-   * Returns a list of operations that the authenticated user (subject) can perform on the target entity (object).
+  List<ResourcePermission> listPermissions(SecurityContext securityContext, String user);
-   *
+
-   * @return
+  /** Returns a list of operations that the authenticated user (subject) can perform on a given resource type */
-   */
+  ResourcePermission getPermission(SecurityContext securityContext, String user, String resource);
-  List<ResourcePermission> listPermissions(SecurityContext securityContext);
+
+  /** Returns a list of operations that the authenticated user (subject) can perform on a given resource */
+  ResourcePermission getPermission(
+      SecurityContext securityContext, String user, ResourceContextInterface resourceContext);
 
   boolean isOwner(SecurityContext ctx, EntityReference entityReference);
 

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/DefaultAuthorizer.java

@@ -17,7 +17,6 @@ import static org.openmetadata.catalog.exception.CatalogExceptionMessage.notAdmi
 import static org.openmetadata.catalog.security.SecurityUtil.DEFAULT_PRINCIPAL_DOMAIN;
 
 import java.io.IOException;
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -97,19 +96,40 @@ public class DefaultAuthorizer implements Authorizer {
   }
 
   @Override
-  public List<ResourcePermission> listPermissions(SecurityContext securityContext) {
+  public List<ResourcePermission> listPermissions(SecurityContext securityContext, String user) {
-    SubjectContext subjectContext;
+    SubjectContext subjectContext = getSubjectContext(securityContext);
-    try {
+    subjectContext = changeSubjectContext(user, subjectContext);
-      subjectContext = getSubjectContext(securityContext);
-    } catch (EntityNotFoundException ex) {
-      return Collections.emptyList();
-    }
 
     if (subjectContext.isAdmin() || subjectContext.isBot()) {
       // Admins and bots have permissions to do all operations.
       return PolicyEvaluator.getResourcePermissions(Access.ALLOW);
     }
-    return PolicyEvaluator.getAllowedOperations(subjectContext);
+    return PolicyEvaluator.listPermission(subjectContext);
+  }
+
+  @Override
+  public ResourcePermission getPermission(SecurityContext securityContext, String user, String resourceType) {
+    SubjectContext subjectContext = getSubjectContext(securityContext);
+    subjectContext = changeSubjectContext(user, subjectContext);
+
+    if (subjectContext.isAdmin() || subjectContext.isBot()) {
+      // Admins and bots have permissions to do all operations.
+      return PolicyEvaluator.getResourcePermission(resourceType, Access.ALLOW);
+    }
+    return PolicyEvaluator.getPermission(subjectContext, resourceType);
+  }
+
+  @Override
+  public ResourcePermission getPermission(
+      SecurityContext securityContext, String user, ResourceContextInterface resourceContext) {
+    SubjectContext subjectContext = getSubjectContext(securityContext);
+    subjectContext = changeSubjectContext(user, subjectContext);
+
+    if (subjectContext.isAdmin() || subjectContext.isBot()) {
+      // Admins and bots have permissions to do all operations.
+      return PolicyEvaluator.getResourcePermission(resourceContext.getResource(), Access.ALLOW);
+    }
+    return PolicyEvaluator.getPermission(subjectContext, resourceContext);
   }
 
   @Override
@@ -137,10 +157,6 @@ public class DefaultAuthorizer implements Authorizer {
       return;
     }
 
-    //    if (subjectContext.isOwner(resourceContext.getOwner())) {
-    //      return;
-    //    }
-
     // TODO view is currently allowed for everyone
     if (operationContext.getOperations().size() == 1
         && operationContext.getOperations().get(0) == MetadataOperation.VIEW_ALL) {
@@ -174,6 +190,22 @@ public class DefaultAuthorizer implements Authorizer {
     if (securityContext == null || securityContext.getUserPrincipal() == null) {
       throw new AuthenticationException("No principal in security context");
     }
-    return SubjectCache.getInstance().getSubjectContext(SecurityUtil.getUserName(securityContext.getUserPrincipal()));
+    return getSubjectContext(SecurityUtil.getUserName(securityContext.getUserPrincipal()));
+  }
+
+  private SubjectContext getSubjectContext(String userName) {
+    return SubjectCache.getInstance().getSubjectContext(userName);
+  }
+
+  private SubjectContext changeSubjectContext(String user, SubjectContext loggedInUser) {
+    // Asking for some other user's permissions is admin only operation
+    if (user != null && !loggedInUser.getUser().getName().equals(user)) {
+      if (!loggedInUser.isAdmin()) {
+        throw new AuthorizationException(notAdmin(loggedInUser.getUser().getName()));
+      }
+      LOG.debug("Changing subject context from logged-in user to {}", user);
+      return getSubjectContext(user);
+    }
+    return loggedInUser;
   }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/NoopAuthorizer.java

@@ -41,11 +41,22 @@ public class NoopAuthorizer implements Authorizer {
   }
 
   @Override
-  public List<ResourcePermission> listPermissions(SecurityContext securityContext) {
+  public List<ResourcePermission> listPermissions(SecurityContext securityContext, String user) {
     // Return all operations.
     return PolicyEvaluator.getResourcePermissions(Access.ALLOW);
   }
 
+  @Override
+  public ResourcePermission getPermission(SecurityContext securityContext, String user, String resource) {
+    return PolicyEvaluator.getResourcePermission(resource, Access.ALLOW);
+  }
+
+  @Override
+  public ResourcePermission getPermission(
+      SecurityContext securityContext, String user, ResourceContextInterface resourceContext) {
+    return PolicyEvaluator.getResourcePermission(resourceContext.getResource(), Access.ALLOW);
+  }
+
   @Override
   public boolean isOwner(SecurityContext securityContext, EntityReference entityReference) {
     return true;

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/policyevaluator/CompiledRule.java

@@ -5,12 +5,16 @@ import static org.openmetadata.catalog.exception.CatalogExceptionMessage.permiss
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.catalog.entity.policies.accessControl.Rule;
 import org.openmetadata.catalog.exception.CatalogExceptionMessage;
 import org.openmetadata.catalog.security.AuthorizationException;
 import org.openmetadata.catalog.security.policyevaluator.SubjectContext.PolicyContext;
 import org.openmetadata.catalog.type.MetadataOperation;
+import org.openmetadata.catalog.type.Permission;
+import org.openmetadata.catalog.type.Permission.Access;
+import org.openmetadata.catalog.type.ResourcePermission;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
@@ -99,6 +103,13 @@ public class CompiledRule extends Rule {
     }
   }
 
+  private Access getAccess() {
+    if (getCondition() != null) {
+      return getEffect() == Effect.DENY ? Access.CONDITIONAL_DENY : Access.CONDITIONAL_ALLOW;
+    }
+    return getEffect() == Effect.DENY ? Access.DENY : Access.ALLOW;
+  }
+
   public void evaluateAllowRule(
       OperationContext operationContext, SubjectContext subjectContext, ResourceContextInterface resourceContext) {
     if (getEffect() != Effect.ALLOW || !matchResource(operationContext.getResource())) {
@@ -117,8 +128,58 @@ public class CompiledRule extends Rule {
     }
   }
 
-  public boolean matchRuleForPermissions(SubjectContext subjectContext) {
+  public void setPermission(Map<String, ResourcePermission> resourcePermissionMap, PolicyContext policyContext) {
-    return matchResource("all");
+    for (ResourcePermission resourcePermission : resourcePermissionMap.values()) {
+      setPermission(resourcePermission.getResource(), resourcePermission, policyContext);
+    }
+  }
+
+  public void setPermission(String resource, ResourcePermission resourcePermission, PolicyContext policyContext) {
+    if (!matchResource(resource)) {
+      return;
+    }
+    Access access = getAccess();
+    // Walk through all the operations in the rule and set permissions
+    for (MetadataOperation ruleOperation : getOperations()) {
+      for (Permission permission : resourcePermission.getPermissions()) {
+        if (matchOperation(permission.getOperation())) {
+          if (overrideAccess(access, permission.getAccess())) {
+            permission
+                .withAccess(access)
+                .withRole(policyContext.getRoleName())
+                .withPolicy(policyContext.getPolicyName())
+                .withRule(this);
+            LOG.debug("Updated permission {}", permission);
+          }
+        }
+      }
+    }
+  }
+
+  public void setPermission(
+      SubjectContext subjectContext,
+      ResourceContextInterface resourceContext,
+      ResourcePermission resourcePermission,
+      PolicyContext policyContext) {
+    if (!matchResource(resourceContext.getResource())) {
+      return;
+    }
+    // Walk through all the operations in the rule and set permissions
+    for (MetadataOperation ruleOperation : getOperations()) {
+      for (Permission permission : resourcePermission.getPermissions()) {
+        if (matchOperation(permission.getOperation()) && matchExpression(subjectContext, resourceContext)) {
+          Access access = getEffect() == Effect.DENY ? Access.DENY : Access.ALLOW;
+          if (overrideAccess(access, permission.getAccess())) {
+            permission
+                .withAccess(access)
+                .withRole(policyContext.getRoleName())
+                .withPolicy(policyContext.getPolicyName())
+                .withRule(this);
+            LOG.debug("Updated permission {}", permission);
+          }
+        }
+      }
+    }
   }
 
   protected boolean matchResource(String resource) {
@@ -150,4 +211,9 @@ public class CompiledRule extends Rule {
     StandardEvaluationContext evaluationContext = new StandardEvaluationContext(ruleEvaluator);
     return Boolean.TRUE.equals(expression.getValue(evaluationContext, Boolean.class));
   }
+
+  static boolean overrideAccess(Access newAccess, Access currentAccess) {
+    // Lower the ordinal number of access overrides higher ordinal number
+    return currentAccess.ordinal() > newAccess.ordinal();
+  }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/policyevaluator/PolicyEvaluator.java

@@ -24,7 +24,6 @@ import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.catalog.ResourceRegistry;
 import org.openmetadata.catalog.entity.policies.Policy;
 import org.openmetadata.catalog.entity.policies.accessControl.Rule;
-import org.openmetadata.catalog.entity.policies.accessControl.Rule.Effect;
 import org.openmetadata.catalog.exception.CatalogExceptionMessage;
 import org.openmetadata.catalog.security.AuthorizationException;
 import org.openmetadata.catalog.security.policyevaluator.SubjectContext.PolicyContext;
@@ -95,8 +94,8 @@ public class PolicyEvaluator {
             subjectContext.getUser().getName(), operationContext.getOperations()));
   }
 
-  /** Returns a list of operations that a user can perform on the given entity. */
+  /** Returns a list of operations that a user can perform on all the resources. */
-  public static List<ResourcePermission> getAllowedOperations(@NonNull SubjectContext subjectContext) {
+  public static List<ResourcePermission> listPermission(@NonNull SubjectContext subjectContext) {
     Map<String, ResourcePermission> resourcePermissionMap = initResourcePermissions();
 
     Iterator<PolicyContext> policies = subjectContext.getPolicies();
@@ -104,19 +103,46 @@ public class PolicyEvaluator {
       PolicyContext policyContext = policies.next();
       for (CompiledRule rule : policyContext.getRules()) {
         LOG.debug("evaluating {}:{}:{}\n", policyContext.getRoleName(), policyContext.getPolicyName(), rule.getName());
-        // TODO fix this
+        rule.setPermission(resourcePermissionMap, policyContext);
-        if (rule.matchRuleForPermissions(subjectContext)) {
-          if (rule.getResources().contains("all")) {
-            setPermissionForAllResources(resourcePermissionMap, rule, policyContext);
-          } else {
-            setPermissionForResources(resourcePermissionMap, rule, policyContext);
-          }
-        }
       }
     }
     return new ArrayList<>(resourcePermissionMap.values());
   }
 
+  /** Returns a list of operations that a user can perform on the given resource/entity type */
+  public static ResourcePermission getPermission(@NonNull SubjectContext subjectContext, String resourceType) {
+    // Initialize all permissions to NOT_ALLOW
+    ResourcePermission resourcePermission = getResourcePermission(resourceType, Access.NOT_ALLOW);
+
+    // Iterate through policies and set the permissions to DENY, ALLOW, CONDITIONAL_DENY, or CONDITIONAL_ALLOW
+    Iterator<PolicyContext> policies = subjectContext.getPolicies();
+    while (policies.hasNext()) {
+      PolicyContext policyContext = policies.next();
+      for (CompiledRule rule : policyContext.getRules()) {
+        LOG.debug("evaluating {}:{}:{}\n", policyContext.getRoleName(), policyContext.getPolicyName(), rule.getName());
+        rule.setPermission(resourceType, resourcePermission, policyContext);
+      }
+    }
+    return resourcePermission;
+  }
+
+  public static ResourcePermission getPermission(
+      @NonNull SubjectContext subjectContext, ResourceContextInterface resourceContext) {
+    // Initialize all permissions to NOT_ALLOW
+    ResourcePermission resourcePermission = getResourcePermission(resourceContext.getResource(), Access.NOT_ALLOW);
+
+    // Iterate through policies and set the permissions to DENY, ALLOW, CONDITIONAL_DENY, or CONDITIONAL_ALLOW
+    Iterator<PolicyContext> policies = subjectContext.getPolicies();
+    while (policies.hasNext()) {
+      PolicyContext policyContext = policies.next();
+      for (CompiledRule rule : policyContext.getRules()) {
+        LOG.debug("evaluating {}:{}:{}\n", policyContext.getRoleName(), policyContext.getPolicyName(), rule.getName());
+        rule.setPermission(subjectContext, resourceContext, resourcePermission, policyContext);
+      }
+    }
+    return resourcePermission;
+  }
+
   /** Get list of resources with all their permissions set to given Access */
   public static List<ResourcePermission> getResourcePermissions(Access access) {
     List<ResourcePermission> resourcePermissions = new ArrayList<>();
@@ -132,6 +158,16 @@ public class PolicyEvaluator {
     return resourcePermissions;
   }
 
+  /** Get list of resources with all their permissions set to given Access */
+  public static ResourcePermission getResourcePermission(String resource, Access access) {
+    ResourceDescriptor rd = ResourceRegistry.getResourceDescriptor(resource);
+    List<Permission> permissions = new ArrayList<>();
+    for (MetadataOperation operation : rd.getOperations()) {
+      permissions.add(new Permission().withOperation(operation).withAccess(access));
+    }
+    return new ResourcePermission().withResource(rd.getName()).withPermissions(permissions);
+  }
+
   /**
    * Initialize a map of Resource name to ResourcePermission with for each resource permission for all operations set as
    * NOT_ALLOW
@@ -143,51 +179,4 @@ public class PolicyEvaluator {
     resourcePermissions.forEach(rp -> resourcePermissionMap.put(rp.getResource(), rp));
     return resourcePermissionMap;
   }
-
-  public static void setPermissionForAllResources(
-      Map<String, ResourcePermission> resourcePermissionMap, Rule rule, PolicyContext policyContext) {
-    // For all resources, set the permission
-    for (ResourcePermission resourcePermission : resourcePermissionMap.values()) {
-      setPermission(resourcePermission, rule, policyContext);
-    }
-  }
-
-  private static void setPermissionForResources(
-      Map<String, ResourcePermission> resourcePermissionMap, CompiledRule rule, PolicyContext policyContext) {
-    for (String resource : rule.getResources()) {
-      ResourcePermission resourcePermission = resourcePermissionMap.get(resource);
-      setPermission(resourcePermission, rule, policyContext);
-    }
-  }
-
-  private static void setPermission(ResourcePermission resourcePermission, Rule rule, PolicyContext policyContext) {
-    Access access = getAccess(rule);
-    for (MetadataOperation ruleOperation : rule.getOperations()) {
-      for (Permission permission : resourcePermission.getPermissions()) {
-        if (permission.getOperation().equals(ruleOperation)) {
-          if (overrideAccess(access, permission.getAccess())) {
-            permission
-                .withAccess(access)
-                .withRole(policyContext.getRoleName())
-                .withPolicy(policyContext.getPolicyName())
-                .withRule(rule);
-            LOG.debug("Updated permission {}", permission);
-          }
-        }
-      }
-    }
-  }
-
-  private static Access getAccess(Rule rule) {
-    if (rule.getCondition() != null) {
-      return rule.getEffect() == Effect.DENY ? Access.CONDITIONAL_DENY : Access.CONDITIONAL_ALLOW;
-    }
-    return rule.getEffect() == Effect.DENY ? Access.DENY : Access.ALLOW;
-  }
-
-  // Returns true if access1 has precedence over access2
-  public static boolean overrideAccess(Access newAccess, Access currentAccess) {
-    // Lower the ordinal number of access overrides higher ordinal number
-    return currentAccess.ordinal() > newAccess.ordinal();
-  }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/policyevaluator/PostResourceContext.java

@@ -2,6 +2,7 @@ package org.openmetadata.catalog.security.policyevaluator;
 
 import java.io.IOException;
 import java.util.List;
+import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.EntityInterface;
 import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.TagLabel;
@@ -14,6 +15,11 @@ public class PostResourceContext implements ResourceContextInterface {
     this.owner = owner;
   }
 
+  @Override
+  public String getResource() {
+    return Entity.THREAD;
+  }
+
   @Override
   public EntityReference getOwner() throws IOException {
     return owner;

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/policyevaluator/ResourceContext.java

@@ -1,21 +1,30 @@
 package org.openmetadata.catalog.security.policyevaluator;
 
+import static org.openmetadata.common.utils.CommonUtil.listOrEmpty;
+
 import java.io.IOException;
 import java.util.List;
 import java.util.UUID;
 import lombok.Builder;
+import lombok.Getter;
 import lombok.NonNull;
+import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.EntityInterface;
 import org.openmetadata.catalog.jdbi3.EntityRepository;
 import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.TagLabel;
+import org.openmetadata.catalog.util.EntityUtil;
 
-/** Builds ResourceContext lazily. As multiple threads don't access this, the class is not thread-safe by design. */
+/**
+ * Builds ResourceContext lazily. ResourceContext includes all the attributes of a resource a user is trying to access
+ * to be used for evaluating Access Control policies.
+ *
+ * <p>As multiple threads don't access this, the class is not thread-safe by design.
+ */
 @Builder
 public class ResourceContext implements ResourceContextInterface {
-  @NonNull private String resource;
+  @NonNull @Getter private String resource;
   @NonNull private EntityRepository<? extends EntityInterface> entityRepository;
-  private String fields;
   private UUID id;
   private String name;
   private EntityInterface entity; // Will be lazily initialized
@@ -29,7 +38,7 @@ public class ResourceContext implements ResourceContextInterface {
   @Override
   public List<TagLabel> getTags() throws IOException {
     resolveEntity();
-    return entity == null ? null : entity.getTags();
+    return entity == null ? null : listOrEmpty(entity.getTags());
   }
 
   @Override
@@ -39,6 +48,13 @@ public class ResourceContext implements ResourceContextInterface {
 
   private EntityInterface resolveEntity() throws IOException {
     if (entity == null) {
+      String fields = "";
+      if (entityRepository.isSupportsOwner()) {
+        fields = EntityUtil.addField(fields, Entity.FIELD_OWNER);
+      }
+      if (entityRepository.isSupportsTags()) {
+        fields = EntityUtil.addField(fields, Entity.FIELD_TAGS);
+      }
       if (id != null) {
         entity = entityRepository.get(null, id, entityRepository.getFields(fields));
       } else if (name != null) {

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/policyevaluator/ResourceContextInterface.java

@@ -7,6 +7,8 @@ import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.TagLabel;
 
 public interface ResourceContextInterface {
+  String getResource();
+
   // Get owner of a resource. If the resource does not support owner or has no owner, return null
   EntityReference getOwner() throws IOException;
 

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/policyevaluator/ThreadResourceContext.java

@@ -2,18 +2,24 @@ package org.openmetadata.catalog.security.policyevaluator;
 
 import java.io.IOException;
 import java.util.List;
+import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.EntityInterface;
 import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.TagLabel;
 
 /** Conversation threads require special handling */
 public class ThreadResourceContext implements ResourceContextInterface {
-  private EntityReference owner;
+  private final EntityReference owner;
 
   public ThreadResourceContext(EntityReference owner) {
     this.owner = owner;
   }
 
+  @Override
+  public String getResource() {
+    return Entity.THREAD;
+  }
+
   @Override
   public EntityReference getOwner() throws IOException {
     return owner;

catalog-rest-service/src/main/java/org/openmetadata/catalog/util/EntityUtil.java

@@ -423,4 +423,9 @@ public final class EntityUtil {
         .withDescription(tag.getDescription())
         .withSource(TagSource.TAG);
   }
+
+  public static String addField(String fields, String newField) {
+    fields = fields == null ? "" : fields;
+    return fields.isEmpty() ? newField : fields + ", " + newField;
+  }
 }

catalog-rest-service/src/test/java/org/openmetadata/catalog/EnumBackwardCompatibilityTest.java

@@ -16,6 +16,7 @@ package org.openmetadata.catalog;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import org.junit.jupiter.api.Test;
+import org.openmetadata.catalog.type.Permission.Access;
 import org.openmetadata.catalog.type.Relationship;
 import org.openmetadata.catalog.type.TagLabel;
 import org.openmetadata.catalog.type.TagLabel.LabelType;
@@ -67,4 +68,18 @@ class EnumBackwardCompatibilityTest {
     assertEquals(0, TagSource.TAG.ordinal());
     assertEquals(1, TagSource.GLOSSARY.ordinal());
   }
+
+  /**
+   * Any time a new enum is added, this test will fail. Update the test with total number of enums and test the ordinal
+   * number of the last enum. This will help catch new enum inadvertently being added in the middle.
+   */
+  @Test
+  void testAccessCardinality() {
+    // Don't change the ordinal values of the Access
+    assertEquals(Access.DENY.ordinal(), 0);
+    assertEquals(Access.ALLOW.ordinal(), 1);
+    assertEquals(Access.CONDITIONAL_DENY.ordinal(), 2);
+    assertEquals(Access.CONDITIONAL_ALLOW.ordinal(), 3);
+    assertEquals(Access.NOT_ALLOW.ordinal(), 4);
+  }
 }

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/permissions/PermissionsResourceTest.java

@@ -13,42 +13,60 @@
 
 package org.openmetadata.catalog.resources.permissions;
 
+import static javax.ws.rs.core.Response.Status.FORBIDDEN;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.openmetadata.catalog.type.MetadataOperation.ALL;
 import static org.openmetadata.catalog.type.MetadataOperation.EDIT_DESCRIPTION;
 import static org.openmetadata.catalog.type.MetadataOperation.EDIT_DISPLAY_NAME;
 import static org.openmetadata.catalog.type.MetadataOperation.EDIT_LINEAGE;
 import static org.openmetadata.catalog.type.MetadataOperation.EDIT_OWNER;
 import static org.openmetadata.catalog.type.MetadataOperation.EDIT_TAGS;
 import static org.openmetadata.catalog.type.MetadataOperation.VIEW_ALL;
+import static org.openmetadata.catalog.type.MetadataOperation.VIEW_DATA_PROFILE;
+import static org.openmetadata.catalog.type.MetadataOperation.VIEW_QUERIES;
+import static org.openmetadata.catalog.type.MetadataOperation.VIEW_SAMPLE_DATA;
+import static org.openmetadata.catalog.type.MetadataOperation.VIEW_TESTS;
+import static org.openmetadata.catalog.type.MetadataOperation.VIEW_USAGE;
 import static org.openmetadata.catalog.type.Permission.Access.ALLOW;
 import static org.openmetadata.catalog.type.Permission.Access.CONDITIONAL_ALLOW;
+import static org.openmetadata.catalog.type.Permission.Access.NOT_ALLOW;
 import static org.openmetadata.catalog.util.TestUtils.ADMIN_AUTH_HEADERS;
+import static org.openmetadata.catalog.util.TestUtils.assertResponse;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import javax.ws.rs.client.WebTarget;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.http.client.HttpResponseException;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.TestInstance;
 import org.openmetadata.catalog.CatalogApplicationTest;
+import org.openmetadata.catalog.Entity;
+import org.openmetadata.catalog.ResourceRegistry;
+import org.openmetadata.catalog.api.data.CreateTable;
+import org.openmetadata.catalog.entity.data.Table;
 import org.openmetadata.catalog.entity.policies.Policy;
 import org.openmetadata.catalog.entity.policies.accessControl.Rule;
-import org.openmetadata.catalog.entity.teams.Role;
+import org.openmetadata.catalog.entity.teams.User;
+import org.openmetadata.catalog.exception.CatalogExceptionMessage;
+import org.openmetadata.catalog.resources.EntityResourceTest;
+import org.openmetadata.catalog.resources.databases.TableResourceTest;
 import org.openmetadata.catalog.resources.permissions.PermissionsResource.ResourcePermissionList;
 import org.openmetadata.catalog.resources.policies.PolicyResource;
 import org.openmetadata.catalog.resources.policies.PolicyResourceTest;
-import org.openmetadata.catalog.resources.teams.RoleResource;
-import org.openmetadata.catalog.resources.teams.RoleResourceTest;
-import org.openmetadata.catalog.resources.teams.UserResourceTest;
 import org.openmetadata.catalog.security.SecurityUtil;
 import org.openmetadata.catalog.security.policyevaluator.PolicyEvaluator;
 import org.openmetadata.catalog.type.MetadataOperation;
 import org.openmetadata.catalog.type.Permission;
+import org.openmetadata.catalog.type.Permission.Access;
+import org.openmetadata.catalog.type.ResourceDescriptor;
 import org.openmetadata.catalog.type.ResourcePermission;
 import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.TestUtils;
@@ -57,90 +75,157 @@ import org.openmetadata.catalog.util.TestUtils;
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class PermissionsResourceTest extends CatalogApplicationTest {
   private static final String ORG_POLICY_NAME = "OrganizationPolicy";
-  private static Policy ORG_POLICY;
   private static List<Rule> ORG_RULES;
 
   private static final String DATA_STEWARD_ROLE_NAME = "DataSteward";
-  private static Role DATA_STEWARD_ROLE;
   private static final String DATA_STEWARD_POLICY_NAME = "DataStewardPolicy";
-  private static Policy DATA_STEWARD_POLICY;
   private static List<Rule> DATA_STEWARD_RULES;
 
   private static final String DATA_CONSUMER_ROLE_NAME = "DataConsumer";
-  private static Role DATA_CONSUMER_ROLE;
   private static final String DATA_CONSUMER_POLICY_NAME = "DataConsumerPolicy";
-  private static Policy DATA_CONSUMER_POLICY;
   private static List<Rule> DATA_CONSUMER_RULES;
 
   private static final String DATA_STEWARD_USER_NAME = "user-data-steward";
+  private static User DATA_STEWARD_USER;
   private static final String DATA_CONSUMER_USER_NAME = "user-data-consumer";
+  private static User DATA_CONSUMER_USER;
 
   @BeforeAll
-  static void setup() throws IOException {
+  static void setup(TestInfo test) throws IOException, URISyntaxException {
-    RoleResourceTest roleResourceTest = new RoleResourceTest();
+    new TableResourceTest().setup(test);
     PolicyResourceTest policyResourceTest = new PolicyResourceTest();
-    UserResourceTest userResourceTest = new UserResourceTest();
 
-    ORG_POLICY = policyResourceTest.getEntityByName(ORG_POLICY_NAME, null, PolicyResource.FIELDS, ADMIN_AUTH_HEADERS);
+    Policy ORG_POLICY =
+        policyResourceTest.getEntityByName(ORG_POLICY_NAME, null, PolicyResource.FIELDS, ADMIN_AUTH_HEADERS);
     ORG_RULES = EntityUtil.resolveRules(ORG_POLICY.getRules());
 
-    DATA_STEWARD_ROLE =
+    Policy DATA_STEWARD_POLICY =
-        roleResourceTest.getEntityByName(DATA_STEWARD_ROLE_NAME, null, RoleResource.FIELDS, ADMIN_AUTH_HEADERS);
+        policyResourceTest.getEntityByName(DATA_STEWARD_POLICY_NAME, null, PolicyResource.FIELDS, ADMIN_AUTH_HEADERS);
+    DATA_STEWARD_RULES = EntityUtil.resolveRules(DATA_STEWARD_POLICY.getRules());
+
     DATA_STEWARD_POLICY =
         policyResourceTest.getEntityByName(DATA_STEWARD_POLICY_NAME, null, PolicyResource.FIELDS, ADMIN_AUTH_HEADERS);
     DATA_STEWARD_RULES = EntityUtil.resolveRules(DATA_STEWARD_POLICY.getRules());
 
-    DATA_STEWARD_ROLE =
+    DATA_STEWARD_USER = EntityResourceTest.USER_WITH_DATA_STEWARD_ROLE;
-        roleResourceTest.getEntityByName(DATA_STEWARD_ROLE_NAME, null, RoleResource.FIELDS, ADMIN_AUTH_HEADERS);
-    DATA_STEWARD_POLICY =
-        policyResourceTest.getEntityByName(DATA_STEWARD_POLICY_NAME, null, PolicyResource.FIELDS, ADMIN_AUTH_HEADERS);
-    DATA_STEWARD_RULES = EntityUtil.resolveRules(DATA_STEWARD_POLICY.getRules());
 
-    userResourceTest.createEntity(
+    Policy DATA_CONSUMER_POLICY =
-        userResourceTest
-            .createRequest(DATA_STEWARD_USER_NAME, "", "", null)
-            .withRoles(List.of(DATA_STEWARD_ROLE.getId())),
-        ADMIN_AUTH_HEADERS);
-
-    DATA_CONSUMER_ROLE =
-        roleResourceTest.getEntityByName(DATA_CONSUMER_ROLE_NAME, null, RoleResource.FIELDS, ADMIN_AUTH_HEADERS);
-    DATA_CONSUMER_POLICY =
         policyResourceTest.getEntityByName(DATA_CONSUMER_POLICY_NAME, null, PolicyResource.FIELDS, ADMIN_AUTH_HEADERS);
     DATA_CONSUMER_RULES = EntityUtil.resolveRules(DATA_CONSUMER_POLICY.getRules());
 
-    userResourceTest.createEntity(
+    DATA_CONSUMER_USER = EntityResourceTest.USER_WITH_DATA_CONSUMER_ROLE;
-        userResourceTest
-            .createRequest(DATA_CONSUMER_USER_NAME, "", "", null)
-            .withRoles(List.of(DATA_CONSUMER_ROLE.getId())),
-        ADMIN_AUTH_HEADERS);
   }
 
   @Test
-  void get_admin_permissions() throws HttpResponseException, JsonProcessingException {
+  void get_anotherUserPermission_disallowed() {
+    // Only admin can get another user's permission
+    // Getting as a data consumer, data steward's permissions should fail
+    Map<String, String> authHeaders = SecurityUtil.authHeaders(DATA_CONSUMER_USER_NAME + "@open-metadata.org");
+    assertResponse(
+        () -> getPermissions(DATA_STEWARD_USER_NAME, authHeaders),
+        FORBIDDEN,
+        CatalogExceptionMessage.notAdmin(DATA_CONSUMER_USER_NAME));
+  }
+
+  @Test
+  void get_admin_permissions() throws HttpResponseException {
     List<ResourcePermission> actualPermissions = getPermissions(ADMIN_AUTH_HEADERS);
     assertEquals(PolicyEvaluator.getResourcePermissions(ALLOW), actualPermissions);
   }
 
   @Test
   void get_dataConsumer_permissions() throws HttpResponseException {
+    List<MetadataOperation> conditional = List.of(ALL); // All operations are conditionally allowed
+
+    // Validate permissions for all resources as logged-in user
     Map<String, String> authHeaders = SecurityUtil.authHeaders(DATA_CONSUMER_USER_NAME + "@open-metadata.org");
     List<ResourcePermission> actualPermissions = getPermissions(authHeaders);
+    assertDataConsumerPermissions(actualPermissions, conditional);
 
+    // Validate permissions for DATA_CONSUMER_USER as admin and validate it
+    actualPermissions = getPermissions(DATA_CONSUMER_USER_NAME, ADMIN_AUTH_HEADERS);
+    assertDataConsumerPermissions(actualPermissions, conditional);
+
+    // Validate permission as logged-in user for each resource one at a time
+    for (ResourceDescriptor rd : ResourceRegistry.listResourceDescriptors()) {
+      ResourcePermission actualPermission = getPermission(rd.getName(), null, authHeaders);
+      assertDataConsumerPermission(actualPermission, conditional);
+    }
+
+    // Validate permission as admin user for each resource one at a time
+    for (ResourceDescriptor rd : ResourceRegistry.listResourceDescriptors()) {
+      ResourcePermission actualPermission = getPermission(rd.getName(), DATA_CONSUMER_USER_NAME, authHeaders);
+      assertDataConsumerPermission(actualPermission, conditional);
+    }
+
+    //
+    // Test getting permissions for an entity as an owner
+    //
+    // Create an entity with data consumer as owner
+    TableResourceTest tableResourceTest = new TableResourceTest();
+    CreateTable createTable =
+        tableResourceTest.createRequest("permissionTest").withOwner(DATA_CONSUMER_USER.getEntityReference());
+    Table table1 = tableResourceTest.createEntity(createTable, ADMIN_AUTH_HEADERS);
+
+    // Data consumer must have all operations allowed based on Organization policy as an owner
+    ResourcePermission actualPermission = getPermission(Entity.TABLE, table1.getId(), null, authHeaders);
+    assertAllOperationsAllowed(actualPermission);
+
+    // Admin getting permissions for a specific resource on for Data consumer
+    actualPermission = getPermission(Entity.TABLE, table1.getId(), DATA_CONSUMER_USER_NAME, ADMIN_AUTH_HEADERS);
+    assertAllOperationsAllowed(actualPermission);
+
+    // Create another table with a different owner and make sure data consumer does not have permission as non owner
+    createTable = tableResourceTest.createRequest("permissionTest1").withOwner(DATA_STEWARD_USER.getEntityReference());
+    Table table2 = tableResourceTest.createEntity(createTable, ADMIN_AUTH_HEADERS);
+
+    //
+    // Test getting permissions for an entity user does not own
+    //
+    // Data consumer has non-owner permissions
+    actualPermission = getPermission(Entity.TABLE, table2.getId(), null, authHeaders);
+
+    // Note that conditional list is empty. All the required context to resolve is met when requesting permission of
+    // a specific resource (both subject and resource context). Only Deny, Allow, NotAllow permissions are expected.
+    assertDataConsumerPermission(actualPermission, Collections.emptyList());
+  }
+
+  private void assertAllOperationsAllowed(ResourcePermission actualPermission) {
+    assertEquals(Entity.TABLE, actualPermission.getResource());
+    for (Permission permission : actualPermission.getPermissions()) {
+      assertEquals(ALLOW, permission.getAccess());
+      assertTrue(List.of(ORG_POLICY_NAME, DATA_CONSUMER_POLICY_NAME).contains(permission.getPolicy()));
+    }
+  }
+
+  private void assertDataConsumerPermissions(
+      List<ResourcePermission> actualPermissions, List<MetadataOperation> conditional) {
     // Only allowed operations in DataConsumerRole. All other operations are notAllow by default
-    List<MetadataOperation> allowed = List.of(VIEW_ALL, EDIT_DESCRIPTION, EDIT_TAGS);
-    List<MetadataOperation> conditional = List.of(EDIT_OWNER);
     for (ResourcePermission actualPermission : actualPermissions) { // For all resources
-      for (Permission permission : actualPermission.getPermissions()) {
+      assertDataConsumerPermission(actualPermission, conditional);
-        if (allowed.contains(permission.getOperation())) {
+    }
-          assertEquals(ALLOW, permission.getAccess());
+  }
-          assertEquals(DATA_CONSUMER_ROLE_NAME, permission.getRole());
+
-          assertEquals(DATA_CONSUMER_POLICY_NAME, permission.getPolicy());
+  private void assertDataConsumerPermission(ResourcePermission actualPermission, List<MetadataOperation> conditional) {
-          assertEquals(DATA_CONSUMER_RULES.get(0), permission.getRule());
+    List<MetadataOperation> allowed =
-        } else if (conditional.contains(permission.getOperation())) {
+        List.of(
-          assertEquals(CONDITIONAL_ALLOW, permission.getAccess());
+            VIEW_ALL,
-          assertEquals(ORG_POLICY_NAME, permission.getPolicy());
+            VIEW_USAGE,
-          assertTrue(ORG_RULES.get(0).equals(permission.getRule()) || ORG_RULES.get(1).equals(permission.getRule()));
+            VIEW_DATA_PROFILE,
-        }
+            VIEW_SAMPLE_DATA,
+            VIEW_TESTS,
+            VIEW_QUERIES,
+            EDIT_DESCRIPTION,
+            EDIT_TAGS);
+
+    // Only allowed operations in DataConsumerRole. All other operations are conditional allow or not allow
+    for (Permission permission : actualPermission.getPermissions()) {
+      if (allowed.contains(permission.getOperation())) {
+        assertPermission(permission, ALLOW, DATA_CONSUMER_ROLE_NAME, DATA_CONSUMER_POLICY_NAME, DATA_CONSUMER_RULES);
+      } else if (conditional.contains(permission.getOperation()) || conditional.contains(ALL)) {
+        assertPermission(permission, CONDITIONAL_ALLOW, null, ORG_POLICY_NAME, ORG_RULES);
+      } else {
+        assertPermission(permission, NOT_ALLOW, null, null, null);
       }
     }
   }
@@ -150,23 +235,70 @@ class PermissionsResourceTest extends CatalogApplicationTest {
     Map<String, String> authHeaders = SecurityUtil.authHeaders(DATA_STEWARD_USER_NAME + "@open-metadata.org");
     List<ResourcePermission> actualPermissions = getPermissions(authHeaders);
 
-    // Only allowed operations in DataConsumerRole. All other operations are notAllow by default
-    List<MetadataOperation> allowed =
-        List.of(VIEW_ALL, EDIT_OWNER, EDIT_DISPLAY_NAME, EDIT_LINEAGE, EDIT_DESCRIPTION, EDIT_TAGS);
     for (ResourcePermission actualPermission : actualPermissions) { // For all resources
-      for (Permission permission : actualPermission.getPermissions()) {
+      assertDataStewardPermission(actualPermission);
-        if (allowed.contains(permission.getOperation())) {
+    }
-          assertEquals(ALLOW, permission.getAccess());
+  }
-          assertEquals(DATA_STEWARD_ROLE_NAME, permission.getRole());
+
-          assertEquals(DATA_STEWARD_POLICY_NAME, permission.getPolicy());
+  private void assertDataStewardPermission(ResourcePermission actualPermission) {
-          assertEquals(DATA_STEWARD_RULES.get(0), permission.getRule());
+    // Only allowed operations in DataConsumerRole. All other operations are conditionalAllow by default
-        }
+    List<MetadataOperation> allowed =
+        List.of(
+            VIEW_ALL,
+            VIEW_USAGE,
+            VIEW_DATA_PROFILE,
+            VIEW_SAMPLE_DATA,
+            VIEW_TESTS,
+            VIEW_QUERIES,
+            EDIT_OWNER,
+            EDIT_DISPLAY_NAME,
+            EDIT_LINEAGE,
+            EDIT_DESCRIPTION,
+            EDIT_TAGS);
+    for (Permission permission : actualPermission.getPermissions()) {
+      if (allowed.contains(permission.getOperation())) {
+        assertPermission(permission, ALLOW, DATA_STEWARD_ROLE_NAME, DATA_STEWARD_POLICY_NAME, DATA_STEWARD_RULES);
+      } else {
+        assertPermission(permission, CONDITIONAL_ALLOW, null, ORG_POLICY_NAME, ORG_RULES);
       }
     }
   }
 
+  private void assertPermission(
+      Permission permission,
+      Access expectedAccess,
+      String expectedRole,
+      String expectedPolicy,
+      List<Rule> expectedRules) {
+    assertEquals(expectedAccess, permission.getAccess(), permission.toString());
+    assertEquals(expectedRole, permission.getRole(), permission.toString());
+    assertEquals(expectedPolicy, permission.getPolicy(), permission.toString());
+    assertTrue(expectedRules == null || expectedRules.contains(permission.getRule()));
+  }
+
   public List<ResourcePermission> getPermissions(Map<String, String> authHeaders) throws HttpResponseException {
     WebTarget target = getResource("permissions");
     return TestUtils.get(target, ResourcePermissionList.class, authHeaders).getData();
   }
+
+  public List<ResourcePermission> getPermissions(String user, Map<String, String> authHeaders)
+      throws HttpResponseException {
+    WebTarget target = getResource("permissions");
+    target = user != null ? target.queryParam("user", user) : target;
+    return TestUtils.get(target, ResourcePermissionList.class, authHeaders).getData();
+  }
+
+  public ResourcePermission getPermission(String resource, String user, Map<String, String> authHeaders)
+      throws HttpResponseException {
+    WebTarget target = getResource("permissions/" + resource);
+    target = user != null ? target.queryParam("user", user) : target;
+    return TestUtils.get(target, ResourcePermission.class, authHeaders);
+  }
+
+  public ResourcePermission getPermission(String resource, UUID uuid, String user, Map<String, String> authHeaders)
+      throws HttpResponseException {
+    WebTarget target = getResource("permissions/" + resource + "/" + uuid);
+    target = user != null ? target.queryParam("user", user) : target;
+    return TestUtils.get(target, ResourcePermission.class, authHeaders);
+  }
 }

catalog-rest-service/src/test/java/org/openmetadata/catalog/security/policyevaluator/PolicyEvaluatorTest.java

@@ -13,46 +13,46 @@ class PolicyEvaluatorTest {
     //
 
     // newAccess (Deny|Allow|ConditionDeny|ConditionalAllow|NotAllow) and currentAccess Deny takes precedence
-    assertFalse(PolicyEvaluator.overrideAccess(Access.DENY, Access.DENY));
+    assertFalse(CompiledRule.overrideAccess(Access.DENY, Access.DENY));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.ALLOW, Access.DENY));
+    assertFalse(CompiledRule.overrideAccess(Access.ALLOW, Access.DENY));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_DENY, Access.DENY));
+    assertFalse(CompiledRule.overrideAccess(Access.CONDITIONAL_DENY, Access.DENY));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_ALLOW, Access.DENY));
+    assertFalse(CompiledRule.overrideAccess(Access.CONDITIONAL_ALLOW, Access.DENY));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.NOT_ALLOW, Access.DENY));
+    assertFalse(CompiledRule.overrideAccess(Access.NOT_ALLOW, Access.DENY));
 
     // newAccess (Deny) and currentAccess Allow - newAccess Deny takes precedence
-    assertTrue(PolicyEvaluator.overrideAccess(Access.DENY, Access.ALLOW));
+    assertTrue(CompiledRule.overrideAccess(Access.DENY, Access.ALLOW));
 
     // newAccess (Allow|ConditionDeny|ConditionalAllow|NotAllow) and currentAccess Allow takes precedence
-    assertFalse(PolicyEvaluator.overrideAccess(Access.ALLOW, Access.ALLOW));
+    assertFalse(CompiledRule.overrideAccess(Access.ALLOW, Access.ALLOW));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_DENY, Access.ALLOW));
+    assertFalse(CompiledRule.overrideAccess(Access.CONDITIONAL_DENY, Access.ALLOW));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_ALLOW, Access.ALLOW));
+    assertFalse(CompiledRule.overrideAccess(Access.CONDITIONAL_ALLOW, Access.ALLOW));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.NOT_ALLOW, Access.ALLOW));
+    assertFalse(CompiledRule.overrideAccess(Access.NOT_ALLOW, Access.ALLOW));
 
     // newAccess (Deny|Allow) and currentAccess ConditionalDeny - newAccess takes precedence
-    assertTrue(PolicyEvaluator.overrideAccess(Access.DENY, Access.CONDITIONAL_DENY));
+    assertTrue(CompiledRule.overrideAccess(Access.DENY, Access.CONDITIONAL_DENY));
-    assertTrue(PolicyEvaluator.overrideAccess(Access.ALLOW, Access.CONDITIONAL_DENY));
+    assertTrue(CompiledRule.overrideAccess(Access.ALLOW, Access.CONDITIONAL_DENY));
 
     // newAccess (ConditionDeny|ConditionalAllow|NotAllow) and currentAccess ConditionalDeny takes precedence
-    assertFalse(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_DENY, Access.CONDITIONAL_DENY));
+    assertFalse(CompiledRule.overrideAccess(Access.CONDITIONAL_DENY, Access.CONDITIONAL_DENY));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_ALLOW, Access.CONDITIONAL_DENY));
+    assertFalse(CompiledRule.overrideAccess(Access.CONDITIONAL_ALLOW, Access.CONDITIONAL_DENY));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.NOT_ALLOW, Access.CONDITIONAL_DENY));
+    assertFalse(CompiledRule.overrideAccess(Access.NOT_ALLOW, Access.CONDITIONAL_DENY));
 
     // newAccess (Deny|Allow|ConditionalDeny) and currentAccess ConditionalAllow - newAccess takes precedence
-    assertTrue(PolicyEvaluator.overrideAccess(Access.DENY, Access.CONDITIONAL_ALLOW));
+    assertTrue(CompiledRule.overrideAccess(Access.DENY, Access.CONDITIONAL_ALLOW));
-    assertTrue(PolicyEvaluator.overrideAccess(Access.ALLOW, Access.CONDITIONAL_ALLOW));
+    assertTrue(CompiledRule.overrideAccess(Access.ALLOW, Access.CONDITIONAL_ALLOW));
-    assertTrue(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_DENY, Access.CONDITIONAL_ALLOW));
+    assertTrue(CompiledRule.overrideAccess(Access.CONDITIONAL_DENY, Access.CONDITIONAL_ALLOW));
 
     // newAccess (ConditionalAllow|NotAllow) and currentAccess ConditionalDeny takes precedence
-    assertFalse(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_ALLOW, Access.CONDITIONAL_ALLOW));
+    assertFalse(CompiledRule.overrideAccess(Access.CONDITIONAL_ALLOW, Access.CONDITIONAL_ALLOW));
-    assertFalse(PolicyEvaluator.overrideAccess(Access.NOT_ALLOW, Access.CONDITIONAL_ALLOW));
+    assertFalse(CompiledRule.overrideAccess(Access.NOT_ALLOW, Access.CONDITIONAL_ALLOW));
 
     // newAccess (Deny|Allow|ConditionalDeny|ConditionalAllow) and currentAccess notAllow - newAccess takes precedence
-    assertTrue(PolicyEvaluator.overrideAccess(Access.DENY, Access.NOT_ALLOW));
+    assertTrue(CompiledRule.overrideAccess(Access.DENY, Access.NOT_ALLOW));
-    assertTrue(PolicyEvaluator.overrideAccess(Access.ALLOW, Access.NOT_ALLOW));
+    assertTrue(CompiledRule.overrideAccess(Access.ALLOW, Access.NOT_ALLOW));
-    assertTrue(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_DENY, Access.NOT_ALLOW));
+    assertTrue(CompiledRule.overrideAccess(Access.CONDITIONAL_DENY, Access.NOT_ALLOW));
-    assertTrue(PolicyEvaluator.overrideAccess(Access.CONDITIONAL_ALLOW, Access.NOT_ALLOW));
+    assertTrue(CompiledRule.overrideAccess(Access.CONDITIONAL_ALLOW, Access.NOT_ALLOW));
 
     // newAccess (ConditionalAllow|NotAllow) and currentAccess ConditionalDeny takes precedence
-    assertFalse(PolicyEvaluator.overrideAccess(Access.NOT_ALLOW, Access.NOT_ALLOW));
+    assertFalse(CompiledRule.overrideAccess(Access.NOT_ALLOW, Access.NOT_ALLOW));
   }
 }

ingestion-core/src/metadata/_version.py

@@ -7,5 +7,5 @@ Provides metadata version information.
 
 from incremental import Version
 
-__version__ = Version("metadata", 0, 12, 0, dev=13)
+__version__ = Version("metadata", 0, 12, 0, dev=14)
 __all__ = ["__version__"]