fix&feature: establish service accounts prefix filter a primary crite…

…ria when available to filter the list of access control rules (#418)

src/main/java/com/purbon/kafka/topology/AccessControlManager.java

@@ -281,18 +281,17 @@ private boolean matchesManagedPrefixList(TopologyAclBinding topologyAclBinding)
     String resourceName = topologyAclBinding.getResourceName();
     String principle = topologyAclBinding.getPrincipal();
     // For global wild cards ACL's we manage only if we manage the service account/principle,
-    // regardless.
-    if (resourceName.equals("*")) {
+    // regardless. Filtering by service account will always take precedence if defined
+    if (haveServiceAccountPrefixFilters() || resourceName.equals("*")) {
       return matchesServiceAccountPrefixList(principle);
     }
 
     if ("TOPIC".equalsIgnoreCase(topologyAclBinding.getResourceType())) {
       return matchesTopicPrefixList(resourceName);
     } else if ("GROUP".equalsIgnoreCase(topologyAclBinding.getResourceType())) {
       return matchesGroupPrefixList(resourceName);
-    } else {
-      return matchesServiceAccountPrefixList(principle);
     }
+    return true; // should include everything if not properly excluded earlier.
   }
 
   private boolean matchesTopicPrefixList(String topic) {
@@ -307,6 +306,10 @@ private boolean matchesServiceAccountPrefixList(String principal) {
     return matchesPrefix(managedServiceAccountPrefixes, principal, "Principal");
   }
 
+  private boolean haveServiceAccountPrefixFilters() {
+    return managedServiceAccountPrefixes.size() != 0;
+  }
+
   private boolean matchesPrefix(List<String> prefixes, String item, String type) {
     boolean matches = prefixes.size() == 0 || prefixes.stream().anyMatch(item::startsWith);
     LOGGER.debug(String.format("%s %s matches %s with $s", type, item, matches, prefixes));

src/main/java/com/purbon/kafka/topology/api/mds/MDSApiClient.java

@@ -117,6 +117,16 @@ public void bindRequest(TopologyAclBinding binding) throws IOException {
     }
   }
 
+  /**
+   * Create an RBAC resource binding
+   *
+   * @param principal
+   * @param role
+   * @param resource
+   * @param resourceType
+   * @param patternType
+   * @return TopologyAclBinding
+   */
   public TopologyAclBinding bind(
       String principal, String role, String resource, String resourceType, String patternType) {
 

src/test/java/com/purbon/kafka/topology/integration/RBACPRoviderRbacIT.java

@@ -4,6 +4,7 @@
 import static com.purbon.kafka.topology.Constants.*;
 import static com.purbon.kafka.topology.roles.rbac.RBACPredefinedRoles.*;
 import static java.util.Arrays.asList;
+import static java.util.Collections.singleton;
 import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
@@ -497,6 +498,82 @@ public void testRoleDeleteFlow() throws IOException {
     assertThat(bindings).hasSize(2);
   }
 
+  @Test
+  public void deleteRolesShouldBeSkippedIfPrincipalIsNotManaged() throws IOException {
+    BackendController cs = new BackendController();
+    ExecutionPlan plan = ExecutionPlan.init(cs, System.out);
+    RBACProvider rbacProvider = Mockito.spy(new RBACProvider(apiClient));
+    RBACBindingsBuilder bindingsBuilder = new RBACBindingsBuilder(apiClient);
+
+    Properties props = new Properties();
+    props.put(TOPOLOGY_STATE_FROM_CLUSTER, true);
+    props.put(ALLOW_DELETE_TOPICS, true);
+    props.put(ALLOW_DELETE_BINDINGS, true);
+    props.put(SERVICE_ACCOUNT_MANAGED_PREFIXES + ".0", "User:app");
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, "");
+
+    Configuration config = new Configuration(cliOps, props);
+
+    accessControlManager = new AccessControlManager(rbacProvider, bindingsBuilder, config);
+    final List<String> principals = asList("User:Pere", "User:app1b", "User:app2b");
+
+    List<Consumer> consumers = new ArrayList<>();
+    consumers.add(new Consumer("User:app1b"));
+    consumers.add(new Consumer("User:app2b"));
+
+    Topology topology = new TopologyImpl(config);
+    var prefix = "deleteRolesShouldBeSkippedIfPrincipalIsNotManaged-test";
+    topology.setContext(prefix);
+
+    Project project = new ProjectImpl("project");
+    project.setConsumers(consumers);
+    topology.setProjects(Collections.singletonList(project));
+
+    Topic topicA = new Topic("topicA");
+    project.addTopic(topicA);
+
+    accessControlManager.apply(topology, plan);
+    plan.run();
+
+    // should create a new principal outside of the managed ones, before triggering the deletion.
+    var extraBinding = apiClient.bind("User:Pere", RESOURCE_OWNER, "topicA", "Topic", "LITERAL");
+    rbacProvider.createBindings(singleton(extraBinding));
+
+    // two group and three topics, as we have one topic and two principles and one unmanaged
+    // resource binding
+    List<TopologyAclBinding> bindings =
+        getBindings(rbacProvider).stream()
+            .filter(
+                binding ->
+                    principals.contains(binding.getPrincipal())
+                        || binding.getResourceName().startsWith(prefix))
+            .collect(Collectors.toList());
+    assertThat(bindings).hasSize(5);
+    consumers.remove(0); // remove the first consumer
+
+    cs = new BackendController();
+    plan = ExecutionPlan.init(cs, System.out);
+
+    accessControlManager.apply(topology, plan);
+    plan.run();
+
+    bindings =
+        getBindings(rbacProvider).stream()
+            .filter(
+                binding ->
+                    principals.contains(binding.getPrincipal())
+                        || binding.getResourceName().startsWith(prefix))
+            .collect(Collectors.toList());
+    // only one group and one topic as we removed one of principles plus the extra binding
+    assertThat(bindings).hasSize(3);
+    List<String> finalPrincipals = asList("User:Pere", "User:app2b");
+    for (TopologyAclBinding binding : bindings) {
+      assertThat(finalPrincipals).contains(binding.getPrincipal());
+    }
+  }
+
   @Test
   public void testJulieRoleAclCreation() throws IOException {