Fix: RBAC Subject and Cluster level binding

kafka-ops · Jul 9, 2021 · 151cae4 · 151cae4
1 parent 73c18ea
commit 151cae4
Show file tree

Hide file tree

Showing 7 changed files with 92 additions and 22 deletions.
diff --git a/src/main/java/com/purbon/kafka/topology/KSqlArtefactManager.java b/src/main/java/com/purbon/kafka/topology/KSqlArtefactManager.java
@@ -84,15 +84,9 @@ private Collection<? extends Artefact> getClustersState() throws IOException {
         .map(
             artefact -> {
               if (artefact instanceof KsqlStreamArtefact) {
-                return new KsqlStreamArtefact(
-                    artefact.getPath(),
-                    null,
-                    artefact.getName());
+                return new KsqlStreamArtefact(artefact.getPath(), null, artefact.getName());
               } else if (artefact instanceof KsqlTableArtefact) {
-                return new KsqlTableArtefact(
-                    artefact.getPath(),
-                    null,
-                    artefact.getName());
+                return new KsqlTableArtefact(artefact.getPath(), null, artefact.getName());
               } else {
                 LOGGER.error("KSQL Artefact of wrong type " + artefact.getClass());
                 return null;

diff --git a/src/main/java/com/purbon/kafka/topology/api/mds/MDSApiClient.java b/src/main/java/com/purbon/kafka/topology/api/mds/MDSApiClient.java
@@ -70,22 +70,30 @@ public TopologyAclBinding bindClusterRole(
     return binding;
   }
 
-  public void bindRequest(TopologyAclBinding binding) throws IOException {
+  private boolean isBindingWithResources(TopologyAclBinding binding) {
+    return !binding.getScope().getResources().isEmpty();
+  }
 
+  MDSRequest buildRequest(TopologyAclBinding binding) {
     String url = binding.getPrincipal() + "/roles/" + binding.getOperation();
-    if (!binding.getResourceType().equals(ResourceType.CLUSTER.name())) {
+    String jsonEntity;
+
+    if (isBindingWithResources(binding)) {
       url = url + "/bindings";
+      jsonEntity = binding.getScope().asJson();
+    } else {
+      jsonEntity = binding.getScope().clustersAsJson();
     }
+    LOGGER.debug("bind.entity: " + jsonEntity);
 
+    return new MDSRequest(url, jsonEntity);
+  }
+
+  public void bindRequest(TopologyAclBinding binding) throws IOException {
+    MDSRequest mdsRequest = buildRequest(binding);
     try {
-      String jsonEntity;
-      if (binding.getResourceType().equals(ResourceType.CLUSTER.name())) {
-        jsonEntity = binding.getScope().clustersAsJson();
-      } else {
-        jsonEntity = binding.getScope().asJson();
-      }
-      LOGGER.debug("bind.entity: " + jsonEntity);
-      doPost("/security/1.0/principals/" + url, jsonEntity);
+      LOGGER.debug("bind.entity: " + mdsRequest.getJsonEntity());
+      doPost("/security/1.0/principals/" + mdsRequest.getUrl(), mdsRequest.getJsonEntity());
     } catch (IOException e) {
       LOGGER.error(e);
       throw e;

diff --git a/src/main/java/com/purbon/kafka/topology/api/mds/MDSRequest.java b/src/main/java/com/purbon/kafka/topology/api/mds/MDSRequest.java
@@ -0,0 +1,19 @@
+package com.purbon.kafka.topology.api.mds;
+
+public class MDSRequest {
+  private final String url;
+  private final String jsonEntity;
+
+  public MDSRequest(String url, String jsonEntity) {
+    this.url = url;
+    this.jsonEntity = jsonEntity;
+  }
+
+  public String getUrl() {
+    return url;
+  }
+
+  public String getJsonEntity() {
+    return jsonEntity;
+  }
+}
diff --git a/src/main/java/com/purbon/kafka/topology/model/Artefact.java b/src/main/java/com/purbon/kafka/topology/model/Artefact.java
@@ -1,8 +1,6 @@
 package com.purbon.kafka.topology.model;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
-
-import java.util.Locale;
 import java.util.Objects;
 
 @JsonInclude(JsonInclude.Include.NON_NULL)

diff --git a/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java b/src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java
@@ -354,7 +354,12 @@ public List<TopologyAclBinding> setClusterLevelRole(
   @Override
   public List<TopologyAclBinding> setSchemaAuthorization(String principal, List<String> subjects) {
     return subjects.stream()
-        .map(subject -> apiClient.bind(principal, RESOURCE_OWNER).forSchemaSubject(subject).apply())
+        .map(
+            subject ->
+                apiClient
+                    .bind(principal, RESOURCE_OWNER)
+                    .forSchemaSubject(subject)
+                    .apply("Subject", subject))
         .filter(Objects::nonNull)
         .collect(Collectors.toList());
   }

diff --git a/src/test/java/com/purbon/kafka/topology/api/mds/MDSApiClientTest.java b/src/test/java/com/purbon/kafka/topology/api/mds/MDSApiClientTest.java
@@ -0,0 +1,46 @@
+package com.purbon.kafka.topology.api.mds;
+
+import static com.purbon.kafka.topology.roles.rbac.RBACPredefinedRoles.DEVELOPER_READ;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+import com.purbon.kafka.topology.roles.TopologyAclBinding;
+import org.junit.Test;
+
+public class MDSApiClientTest {
+  MDSApiClient apiClient = new MDSApiClient("http://not_used:8090");
+
+  @Test
+  public void testBindSubjectRole() {
+    String principal = "User:foo";
+    String subject = "topic-value";
+
+    TopologyAclBinding binding =
+        apiClient
+            .bind(principal, DEVELOPER_READ)
+            .forSchemaSubject(subject)
+            .apply("Subject", subject);
+
+    MDSRequest mdsRequest = apiClient.buildRequest(binding);
+
+    assertThat(mdsRequest.getUrl()).isEqualTo("User:foo/roles/DeveloperRead/bindings");
+    assertThat(mdsRequest.getJsonEntity())
+        .isEqualTo(
+            "{\"resourcePatterns\":[{\"name\":\"Subject:topic-value\",\"patternType\":\"LITERAL\",\"resourceType\":\"Subject\"}],\"scope\":{\"clusters\":{\"kafka-cluster\":\"\",\"schema-registry-cluster\":\"\"}}}");
+  }
+
+  @Test
+  public void testBindSubjectRoleWithoutResourceType() {
+    String principal = "User:foo";
+    String subject = "topic-value";
+
+    TopologyAclBinding binding =
+        apiClient.bind(principal, DEVELOPER_READ).forSchemaSubject(subject).apply();
+
+    MDSRequest mdsRequest = apiClient.buildRequest(binding);
+
+    assertThat(mdsRequest.getUrl()).isEqualTo("User:foo/roles/DeveloperRead/bindings");
+    assertThat(mdsRequest.getJsonEntity())
+        .isEqualTo(
+            "{\"resourcePatterns\":[{\"name\":\"Subject:topic-value\",\"patternType\":\"LITERAL\",\"resourceType\":\"Subject\"}],\"scope\":{\"clusters\":{\"kafka-cluster\":\"\",\"schema-registry-cluster\":\"\"}}}");
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/KsqlManagerIT.java b/src/test/java/com/purbon/kafka/topology/integration/KsqlManagerIT.java
@@ -73,7 +73,7 @@ public void testCreateAndUpdatePathWithRemoveClusterState() throws IOException {
     props.put(TOPOLOGY_STATE_FROM_CLUSTER, "true");
     props.put(TOPOLOGY_TOPIC_STATE_FROM_CLUSTER, "false");
     props.put(ALLOW_DELETE_KSQL_ARTEFACTS, "true");
-    props.put(PLATFORM_SERVER_KSQL, "http://"+client.getServer());
+    props.put(PLATFORM_SERVER_KSQL, "http://" + client.getServer());
 
     File file = TestUtils.getResourceFile("/descriptor-ksql.yaml");