用于管理 Kafka ACL 的 Python 客户端答案

【问题标题】：Python client for Managing Kafka ACL用于管理 Kafka ACL 的 Python 客户端
【发布时间】：2022-10-30 07:44:04
【问题描述】：

我们有任何用于管理 Kafka ACL 的 python 客户端吗？我知道我们有 Java Admin 客户端，可以使用 Kafka 脚本（kafka-acl.sh）执行相同的操作，但找不到用于 python 客户端的。汇合的python库不支持它。遇到kafka-python，但根据文档，它不支持为现有主题创建 ACL。

【问题讨论】：

您始终可以使用 os.subprocess 调用现有的 Kafka CLI 命令。但我认为将 ACL 外部化到不同的解决方案（如 Apache Ranger 或 openpolicyagent）是一个更好的主意

标签： python apache-kafka kafka-python confluent-kafka-python

【解决方案1】：

使用 kafka-python 库

配置acl，new add，例子：

用户 user1 具有对 topic2 和 group2 组（消费者）的读取权限

用户user2 拥有topic2（生产者）的写权限

from kafka.admin import KafkaAdminClient, ACLPermissionType, ResourcePattern, ResourceType, ACL, ACLOperation, ACLFilter

brokers = 'localhost:9092'

admin = KafkaAdminClient(
    bootstrap_servers=brokers,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanism='SCRAM-SHA-512',
    sasl_plain_username='admin',      # your privileged user
    sasl_plain_password='admin12345'  # password
)


acl1 = ACL(
    principal="User:user1",
    host="*",
    operation=ACLOperation.READ,
    permission_type=ACLPermissionType.ALLOW,
    resource_pattern=ResourcePattern(ResourceType.TOPIC, 'topic2')
)
acl2 = ACL(
    principal="User:user1",
    host="*",
    operation=ACLOperation.READ,
    permission_type=ACLPermissionType.ALLOW,
    resource_pattern=ResourcePattern(ResourceType.GROUP, 'group2')
)
acl3 = ACL(
    principal="User:user2",
    host="*",
    operation=ACLOperation.WRITE,
    permission_type=ACLPermissionType.ALLOW,
    resource_pattern=ResourcePattern(ResourceType.TOPIC, 'topic2')
)


acls_result = admin.create_acls([acl1, acl2, acl3])
print(acls_result)

删除 acl

from kafka.admin import KafkaAdminClient, ACLPermissionType, ResourcePattern, ResourceType, ACL, ACLOperation, ACLFilter

brokers = 'localhost:9092'

admin = KafkaAdminClient(
    bootstrap_servers=brokers,
    security_protocol='SASL_PLAINTEXT',
    sasl_mechanism='SCRAM-SHA-512',
    sasl_plain_username='admin',      # your privileged user
    sasl_plain_password='admin12345'  # password
)

delete_acls = admin.delete_acls(
    [
        ACLFilter(principal='User:user1',
                  host='*',
                  operation=ACLOperation.ANY,
                  permission_type=ACLPermissionType.ANY,
                  resource_pattern=ResourcePattern(ResourceType.TOPIC, 'topic2')),
        ACLFilter(principal='User:user1',
                  host='*',
                  operation=ACLOperation.ANY,
                  permission_type=ACLPermissionType.ANY,
                  resource_pattern=ResourcePattern(ResourceType.GROUP, 'group2'))
    ]

)

print(delete_acls)

admin.delete_acls提交的数组，可以写多个ACLFilter来查找要删除的acls。

描述 acl

acl_filter = ACLFilter(
    principal=None,
    host='*',
    operation=ACLOperation.ANY,
    permission_type=ACLPermissionType.ANY,
    resource_pattern=ResourcePattern(ResourceType.TOPIC,'topic2')
)

result = admin.describe_acls(acl_filter)
print(result)

【讨论】：