Merge pull request #393 from aiven/fdorlandi-add-support-for-kafka-acls

Add cli support for Apache Kafka native ACLs

aiven/client/cli.py

@@ -2954,7 +2954,7 @@ def service__topic_delete(self) -> None:
         required=True,
     )
     def service__acl_add(self) -> None:
-        """Add a Kafka ACL entry"""
+        """Add an Aiven ACL for Kafka entry"""
         response = self.client.add_service_kafka_acl(
             project=self.get_project(),
             service=self.args.service_name,
@@ -2968,7 +2968,7 @@ def service__acl_add(self) -> None:
     @arg.service_name
     @arg("acl_id", help="ID of the ACL entry to delete")
     def service__acl_delete(self) -> None:
-        """Delete a Kafka ACL entry"""
+        """Delete an Aiven ACL for Kafka entry"""
         response = self.client.delete_service_kafka_acl(
             project=self.get_project(), service=self.args.service_name, acl_id=self.args.acl_id
         )
@@ -2978,7 +2978,7 @@ def service__acl_delete(self) -> None:
     @arg.service_name
     @arg.json
     def service__acl_list(self) -> None:
-        """List Kafka ACL entries"""
+        """List Aiven ACL for Kafka entries"""
         service = self.client.get_service(project=self.get_project(), service=self.args.service_name)
 
         layout = ["id", "username", "topic", "permission"]
@@ -6176,6 +6176,143 @@ def service__alloydbomni__google_cloud_private_key__delete(self) -> None:
             layout = ["client_email", "private_key_id"]
             self.print_response(output, json=self.args.json, table_layout=layout)
 
+    @arg.project
+    @arg.service_name
+    @arg(
+        "--operation",
+        help="Operation that is being allowed or denied.",
+        required=True,
+        choices=[
+            "Describe",
+            "DescribeConfigs",
+            "Alter",
+            "IdempotentWrite",
+            "Read",
+            "Delete",
+            "Create",
+            "ClusterAction",
+            "All",
+            "Write",
+            "AlterConfigs",
+            "CreateTokens",
+            "DescribeTokens",
+        ],
+    )
+    @arg(
+        "--topic",
+        help="Topic resource type to which ACL should be added",
+    )
+    @arg(
+        "--group",
+        help="Group resource type to which ACL should be added",
+    )
+    @arg(
+        "--cluster",
+        action="store_const",
+        const="kafka-cluster",
+        help="Group resource type to which ACL should be added",
+    )
+    @arg(
+        "--transactional-id",
+        help="TransactionalId resource type to which ACL should be added",
+    )
+    @arg(
+        "--resource-pattern-type",
+        help="The type of the resource pattern",
+        required=False,
+        choices=["LITERAL", "PREFIXED"],
+        default="LITERAL",
+    )
+    @arg(
+        "--deny",
+        help="Create a DENY rule (default is ALLOW)",
+        action="store_true",
+    )
+    @arg(
+        "--host",
+        help="The host for the ACLs, a value of '*' matches all hosts",
+        required=False,
+        default="*",
+    )
+    @arg(
+        "--principal",
+        help="The principal for the ACLs, must be in the form principalType:name",
+        required=True,
+    )
+    def service__kafka_acl_add(self) -> None:
+        """Add a Kafka-native ACL entry"""
+        mutually_exclusive_args = [
+            self.args.topic,
+            self.args.group,
+            self.args.cluster,
+            self.args.transactional_id,
+        ]
+        count = len(list(filter(lambda x: x is not None, mutually_exclusive_args)))
+        if count == 0:
+            raise argx.UserError("At least one of --topic --group --cluster --transactional-id must be specified")
+        if count > 1:
+            raise argx.UserError("Arguments --topic --group --cluster --transactional-id are mutually exclusive")
+        if self.args.topic is not None:
+            resource_name = self.args.topic
+            resource_type = "Topic"
+        elif self.args.group is not None:
+            resource_name = self.args.group
+            resource_type = "Group"
+        elif self.args.cluster is not None:
+            resource_name = self.args.cluster
+            resource_type = "Cluster"
+        elif self.args.transactional_id is not None:
+            resource_name = self.args.transactional_id
+            resource_type = "TransactionalId"
+
+        response = self.client.service_kafka_native_acl_add(
+            project=self.get_project(),
+            service=self.args.service_name,
+            principal=self.args.principal,
+            host=self.args.host,
+            resource_name=resource_name,
+            resource_type=resource_type,
+            resource_pattern_type=self.args.resource_pattern_type,
+            operation=self.args.operation,
+            permission_type="DENY" if self.args.deny else "ALLOW",
+        )
+        print(response["message"])
+
+    @arg.project
+    @arg.service_name
+    @arg.json
+    def service__kafka_acl_list(self) -> None:
+        """List Kafka-native ACL entries"""
+        response = self.client.service_kafka_native_acl_list(
+            project=self.get_project(),
+            service=self.args.service_name,
+        )
+        acls = response.get("kafka_acl", [])
+        layout = [
+            "id",
+            "permission_type",
+            "principal",
+            "operation",
+            "resource_type",
+            "pattern_type",
+            "resource_name",
+            "host",
+        ]
+        if acls:
+            self.print_response(acls, json=self.args.json, table_layout=layout)
+        else:
+            self.print_response([{k: "" for k in layout}], json=self.args.json, table_layout=layout)
+
+    @arg.project
+    @arg.service_name
+    @arg("acl_id", help="ID of the ACL entry to delete")
+    def service__kafka_acl_delete(self) -> None:
+        """Delete a Kafka-native ACL entry"""
+        response = self.client.service_kafka_native_acl_delete(
+            project=self.get_project(), service=self.args.service_name, acl_id=self.args.acl_id
+        )
+        print(response["message"])
+
 
 if __name__ == "__main__":
     AivenCLI().main()

aiven/client/client.py

@@ -2919,3 +2919,45 @@ def alloydbomni_google_cloud_private_key_show(self, *, project: str, service: st
                 "google_cloud_private_key",
             ),
         )
+
+    def service_kafka_native_acl_add(
+        self,
+        project: str,
+        service: str,
+        principal: str,
+        host: str,
+        resource_name: str,
+        resource_type: str,
+        resource_pattern_type: str,
+        operation: str,
+        permission_type: str,
+    ) -> Mapping:
+        return self.verify(
+            self.post,
+            self.build_path("project", project, "service", service, "kafka", "acl"),
+            body={
+                "principal": principal,
+                "host": host,
+                "resource_name": resource_name,
+                "resource_type": resource_type,
+                "pattern_type": resource_pattern_type,
+                "operation": operation,
+                "permission_type": permission_type,
+            },
+        )
+
+    def service_kafka_native_acl_list(
+        self,
+        project: str,
+        service: str,
+    ) -> dict[str, Any]:
+        return self.verify(
+            self.get,
+            self.build_path("project", project, "service", service, "kafka", "acl"),
+        )
+
+    def service_kafka_native_acl_delete(self, project: str, service: str, acl_id: str) -> Mapping:
+        return self.verify(
+            self.delete,
+            self.build_path("project", project, "service", service, "kafka", "acl", acl_id),
+        )

tests/test_cli.py

@@ -1992,3 +1992,133 @@ def test_byoc_tags_replace() -> None:
             "byoc_resource_tag:byoc_resource_tag:key_3": "byoc_resource_tag:keep-the-whole-value-3",
         },
     )
+
+
+@pytest.mark.parametrize(
+    "res_arg,res_type,res_name",
+    [
+        ("--topic", "Topic", "TopicABC"),
+        ("--group", "Group", "GroupDEF"),
+        ("--cluster", "Cluster", "kafka-cluster"),
+        ("--transactional-id", "TransactionalId", "Id123"),
+    ],
+)
+def test_service__kafka_acl_add_resource(res_arg: str, res_type: str, res_name: str) -> None:
+    aiven_client = mock.Mock(spec_set=AivenClient)
+    aiven_client.service_kafka_native_acl_add.return_value = {"message": "added"}
+    args = [
+        "service",
+        "kafka-acl-add",
+        "kafka-1",
+        "--project=project1",
+        "--principal=User:alice",
+        "--operation=Describe",
+    ]
+    if res_arg == "--cluster":
+        args.append(f"{res_arg}")
+    else:
+        args.append(f"{res_arg}={res_name}")
+    build_aiven_cli(aiven_client).run(args=args)
+    aiven_client.service_kafka_native_acl_add.assert_called_once_with(
+        project="project1",
+        service="kafka-1",
+        principal="User:alice",
+        host="*",
+        resource_name=res_name,
+        resource_type=res_type,
+        resource_pattern_type="LITERAL",
+        operation="Describe",
+        permission_type="ALLOW",
+    )
+
+
+@pytest.mark.parametrize("deny", [True, False])
+def test_service__kafka_acl_add_allow_deny(deny: bool) -> None:
+    aiven_client = mock.Mock(spec_set=AivenClient)
+    aiven_client.service_kafka_native_acl_add.return_value = {"message": "added"}
+    args = [
+        "service",
+        "kafka-acl-add",
+        "kafka-1",
+        "--project=project1",
+        "--principal=User:alice",
+        "--operation=Describe",
+        "--topic=TopicABC",
+    ]
+    if deny:
+        args.append("--deny")
+    build_aiven_cli(aiven_client).run(args=args)
+    aiven_client.service_kafka_native_acl_add.assert_called_once_with(
+        project="project1",
+        service="kafka-1",
+        principal="User:alice",
+        host="*",
+        resource_name="TopicABC",
+        resource_type="Topic",
+        resource_pattern_type="LITERAL",
+        operation="Describe",
+        permission_type="DENY" if deny else "ALLOW",
+    )
+
+
+@pytest.mark.parametrize("prefixed", [True, False])
+def test_service__kafka_acl_add_prefixed(prefixed: bool) -> None:
+    aiven_client = mock.Mock(spec_set=AivenClient)
+    aiven_client.service_kafka_native_acl_add.return_value = {"message": "added"}
+    args = [
+        "service",
+        "kafka-acl-add",
+        "kafka-1",
+        "--project=project1",
+        "--principal=User:alice",
+        "--operation=Describe",
+        "--topic=TopicABC",
+    ]
+    if prefixed:
+        args.append("--resource-pattern-type=PREFIXED")
+    build_aiven_cli(aiven_client).run(args=args)
+    aiven_client.service_kafka_native_acl_add.assert_called_once_with(
+        project="project1",
+        service="kafka-1",
+        principal="User:alice",
+        host="*",
+        resource_name="TopicABC",
+        resource_type="Topic",
+        resource_pattern_type="PREFIXED" if prefixed else "LITERAL",
+        operation="Describe",
+        permission_type="ALLOW",
+    )
+
+
+def test_service__kafka_acl_list() -> None:
+    aiven_client = mock.Mock(spec_set=AivenClient)
+    aiven_client.service_kafka_native_acl_list.return_value = {"kafka_acl": []}
+    args = [
+        "service",
+        "kafka-acl-list",
+        "kafka-1",
+        "--project=project1",
+    ]
+    build_aiven_cli(aiven_client).run(args=args)
+    aiven_client.service_kafka_native_acl_list.assert_called_once_with(
+        project="project1",
+        service="kafka-1",
+    )
+
+
+def test_service__kafka_acl_delete() -> None:
+    aiven_client = mock.Mock(spec_set=AivenClient)
+    aiven_client.service_kafka_native_acl_delete.return_value = {"message": "added"}
+    args = [
+        "service",
+        "kafka-acl-delete",
+        "kafka-1",
+        "acl4f549bfee6a",
+        "--project=project1",
+    ]
+    build_aiven_cli(aiven_client).run(args=args)
+    aiven_client.service_kafka_native_acl_delete.assert_called_once_with(
+        project="project1",
+        service="kafka-1",
+        acl_id="acl4f549bfee6a",
+    )