feat: add cli support for Apache Kafka native ACLs

aiven/client/cli.py

@@ -6176,6 +6176,94 @@ 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(
+        "--resource-name",
+        help=(
+            "The resource to which ACLs should be added, when using LITERAL resource pattern type, "
+            "a name of '*' matches all resources of the selected type"
+        ),
+        required=True,
+    )
+    @arg(
+        "--resource-type",
+        help="Topic resource type to which ACLs should be added",
+        required=False,
+        choices=["Any", "Topic", "Group", "Cluster", "TransactionalId", "DelegationToken"],
+    )
+    @arg(
+        "--resource-pattern-type",
+        help="The type of the resource pattern",
+        required=False,
+        choices=["LITERAL", "PREFIXED"],
+        default="LITERAL",
+    )
+    @arg(
+        "--permission-type",
+        help="The type of the resource pattern",
+        required=True,
+        choices=["ALLOW", "DENY"],
+    )
+    @arg(
+        "--host",
+        help="The host for the ACLs, a value of '*' matched 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"""
+        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=self.args.resource_name,
+            resource_type=self.args.resource_type,
+            resource_pattern_type=self.args.resource_pattern_type,
+            operation=self.args.operation,
+            permission_type=self.args.permission_type,
+        )
+        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", "principal", "operation", "resourceName", "resourceType", "patternType", "permissionType"]
+        self.print_response(acls, 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 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,
+                "resourceName": resource_name,
+                "resourceType": resource_type,
+                "patternType": resource_pattern_type,
+                "operation": operation,
+                "permissionType": 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),
+        )