Add KafkaAdmin ACL methods

Author: Ulrik Johansson

kafka/admin/__init__.py

@@ -1,10 +1,13 @@
 from __future__ import absolute_import
 
 from kafka.admin.config_resource import ConfigResource, ConfigResourceType
+from kafka.admin.acl_resource import (AclResource, AclOperation, AclResourceType, AclPermissionType,
+                                      AclResourcePatternType)
 from kafka.admin.kafka import KafkaAdmin
 from kafka.admin.new_topic import NewTopic
 from kafka.admin.new_partitions import NewPartitions
 
 __all__ = [
-    'ConfigResource', 'ConfigResourceType', 'KafkaAdmin', 'NewTopic', 'NewPartitions'
+    'ConfigResource', 'ConfigResourceType', 'KafkaAdmin', 'NewTopic', 'NewPartitions', 'AclResource', 'AclOperation',
+    'AclResourceType', 'AclPermissionType', 'AclResourcePatternType'
 ]

kafka/admin/acl_resource.py

@@ -0,0 +1,84 @@
+from __future__ import absolute_import
+
+# enum in stdlib as of py3.4
+try:
+    from enum import IntEnum  # pylint: disable=import-error
+except ImportError:
+    # vendored backport module
+    from kafka.vendor.enum34 import IntEnum
+
+class AclResourceType(IntEnum):
+    """An enumerated type of config resources"""
+
+    ANY = 1,
+    BROKER = 4,
+    DELEGATION_TOKEN = 6,
+    GROUP = 3,
+    TOPIC = 2,
+    TRANSACTIONAL_ID = 5
+
+class AclOperation(IntEnum):
+    """An enumerated type of acl operations"""
+
+    ANY = 1,
+    ALL = 2,
+    READ = 3,
+    WRITE = 4,
+    CREATE = 5,
+    DELETE = 6,
+    ALTER = 7,
+    DESCRIBE = 8,
+    CLUSTER_ACTION = 9,
+    DESCRIBE_CONFIGS = 10,
+    ALTER_CONFIGS = 11,
+    IDEMPOTENT_WRITE = 12
+
+
+class AclPermissionType(IntEnum):
+    """An enumerated type of permissions"""
+
+    ANY = 1,
+    DENY = 2,
+    ALLOW = 3
+
+class AclResourcePatternType(IntEnum):
+    """An enumerated type of resource patterns"""
+
+    ANY = 1,
+    MATCH = 2,
+    LITERAL = 3,
+    PREFIXED = 4
+
+class AclResource(object):
+    """A class for specifying config resources.
+    Arguments:
+        resource_type (ConfigResourceType): the type of kafka resource
+        name (string): The name of the kafka resource
+        configs ({key : value}): A  maps of config keys to values.
+    """
+
+    def __init__(
+            self,
+            resource_type,
+            operation,
+            permission_type,
+            name=None,
+            principal=None,
+            host=None,
+            pattern_type=AclResourcePatternType.LITERAL
+    ):
+        if not isinstance(resource_type, AclResourceType):
+            resource_type = AclResourceType[str(resource_type).upper()]  # pylint: disable-msg=unsubscriptable-object
+        self.resource_type = resource_type
+        if not isinstance(operation, AclOperation):
+            operation = AclOperation[str(operation).upper()]  # pylint: disable-msg:unsubscriptable-object
+        self.operation = operation
+        if not isinstance(permission_type, AclPermissionType):
+            permission_type = AclPermissionType[str(permission_type).upper()]  # pylint: disable-msg=unsubscriptable-object
+        self.permission_type = permission_type
+        self.name = name
+        self.principal = principal
+        self.host = host
+        if not isinstance(pattern_type, AclResourcePatternType):
+            pattern_type = AclResourcePatternType[str(pattern_type).upper()]  # pylint: disable-msg=unsubscriptable-object
+        self.pattern_type = pattern_type

kafka/admin/kafka.py

@@ -5,12 +5,14 @@
 import socket
 from kafka.client_async import KafkaClient, selectors
 from kafka.errors import (
-    KafkaConfigurationError, UnsupportedVersionError, NodeNotReadyError, NotControllerError, KafkaConnectionError)
+    KafkaConfigurationError, UnsupportedVersionError, NodeNotReadyError, NotControllerError, KafkaConnectionError,
+    IllegalArgumentError)
 from kafka.metrics import MetricConfig, Metrics
 from kafka.protocol.admin import (
     CreateTopicsRequest, DeleteTopicsRequest, DescribeConfigsRequest, AlterConfigsRequest, CreatePartitionsRequest,
-    ListGroupsRequest, DescribeGroupsRequest)
+    ListGroupsRequest, DescribeGroupsRequest, DescribeAclsRequest, CreateAclsRequest, DeleteAclsRequest)
 from kafka.protocol.metadata import MetadataRequest
+from kafka.admin.acl_resource import AclOperation, AclPermissionType
 from kafka.version import __version__
 
 log = logging.getLogger(__name__)
@@ -358,11 +360,137 @@ def delete_topics(self, topics, timeout_ms=None):
 
     # describe cluster functionality is in ClusterMetadata
 
-    # describe_acls protocol not implemented
+    def describe_acls(self, acl_resource):
+        """Describe a set of ACLs
+        """
+
+        version = self._matching_api_version(DescribeAclsRequest)
+        if version == 0:
+            request = DescribeAclsRequest[version](
+                resource_type=acl_resource.resource_type,
+                resource_name=acl_resource.name,
+                principal=acl_resource.principal,
+                host=acl_resource.host,
+                operation=acl_resource.operation,
+                permission_type=acl_resource.permission_type
+            )
+        elif version <= 1:
+            request = DescribeAclsRequest[version](
+                resource_type=acl_resource.resource_type,
+                resource_name=acl_resource.name,
+                resource_pattern_type_filter=acl_resource.pattern_type,
+                principal=acl_resource.principal,
+                host=acl_resource.host,
+                operation=acl_resource.operation,
+                permission_type=acl_resource.permission_type
+
+            )
+        else:
+            raise UnsupportedVersionError(
+                "missing implementation of DescribeAcls for library supported version {}"
+                    .format(version)
+            )
+
+        return self._send(request)
+
+    @staticmethod
+    def _convert_create_acls_resource_request_v0(acl_resource):
+        if acl_resource.operation == AclOperation.ANY:
+            raise IllegalArgumentError("operation must not be ANY")
+        if acl_resource.permission_type == AclPermissionType.ANY:
+            raise IllegalArgumentError("permission_type must not be ANY")
+
+        return (
+            acl_resource.resource_type,
+            acl_resource.name,
+            acl_resource.principal,
+            acl_resource.host,
+            acl_resource.operation,
+            acl_resource.permission_type
+        )
+
+    @staticmethod
+    def _convert_create_acls_resource_request_v1(acl_resource):
+
+        if acl_resource.operation == AclOperation.ANY:
+            raise IllegalArgumentError("operation must not be ANY")
+        if acl_resource.permission_type == AclPermissionType.ANY:
+            raise IllegalArgumentError("permission_type must not be ANY")
+
+        return (
+            acl_resource.resource_type,
+            acl_resource.name,
+            acl_resource.pattern_type,
+            acl_resource.principal,
+            acl_resource.host,
+            acl_resource.operation,
+            acl_resource.permission_type
+        )
+
+    def create_acls(self, acl_resources):
+        """Create a set of ACLs"""
+
+        version = self._matching_api_version(DescribeAclsRequest)
+        if version == 0:
+            request = CreateAclsRequest[version](
+                creations=[self._convert_create_acls_resource_request_v0(acl_resource) for acl_resource in acl_resources]
+            )
+        elif version <= 1:
+            request = CreateAclsRequest[version](
+                creations=[self._convert_create_acls_resource_request_v1(acl_resource) for acl_resource in acl_resources]
+            )
+        else:
+            raise UnsupportedVersionError(
+                "missing implementation of DescribeAcls for library supported version {}"
+                    .format(version)
+            )
+
+
+        return self._send(request)
+
+    @staticmethod
+    def _convert_delete_acls_resource_request_v0(acl_resource):
+        return (
+            acl_resource.resource_type,
+            acl_resource.name,
+            acl_resource.principal,
+            acl_resource.host,
+            acl_resource.operation,
+            acl_resource.permission_type
+        )
+
+    @staticmethod
+    def _convert_delete_acls_resource_request_v1(acl_resource):
+        return (
+            acl_resource.resource_type,
+            acl_resource.name,
+            acl_resource.pattern_type,
+            acl_resource.principal,
+            acl_resource.host,
+            acl_resource.operation,
+            acl_resource.permission_type
+        )
+
+    def delete_acls(self, acl_resources):
+        """Delete a set of ACLSs"""
 
-    # create_acls protocol not implemented
+        version = self._matching_api_version(DescribeAclsRequest)
+
+        if version == 0:
+            request = DeleteAclsRequest[version](
+                filters=[self._convert_delete_acls_resource_request_v0(acl_resource) for acl_resource in acl_resources]
+            )
+        elif version <= 1:
+            request = DeleteAclsRequest[version](
+                filters=[self._convert_delete_acls_resource_request_v1(acl_resource) for acl_resource in acl_resources]
+            )
+        else:
+            raise UnsupportedVersionError(
+                "missing implementation of DescribeAcls for library supported version {}"
+                    .format(version)
+            )
 
-    # delete_acls protocol not implemented
+        return self._send(request)
 
     @staticmethod
     def _convert_describe_config_resource_request(config_resource):