Add ACL api to KafkaAdminClient

kafka/admin/__init__.py

@@ -2,9 +2,13 @@
 
 from kafka.admin.config_resource import ConfigResource, ConfigResourceType
 from kafka.admin.client import KafkaAdminClient
+from kafka.admin.acl_resource import (ACL, ACLFilter, ResourcePattern, ResourcePatternFilter, ACLOperation,
+                                      ResourceType, ACLPermissionType, ACLResourcePatternType)
 from kafka.admin.new_topic import NewTopic
 from kafka.admin.new_partitions import NewPartitions
 
 __all__ = [
-    'ConfigResource', 'ConfigResourceType', 'KafkaAdminClient', 'NewTopic', 'NewPartitions'
+    'ConfigResource', 'ConfigResourceType', 'KafkaAdminClient', 'NewTopic', 'NewPartitions', 'ACL', 'ACLFilter',
+    'ResourcePattern', 'ResourcePatternFilter', 'ACLOperation', 'ResourceType', 'ACLPermissionType',
+    'ACLResourcePatternType'
 ]

kafka/admin/acl_resource.py

@@ -0,0 +1,212 @@
+from __future__ import absolute_import
+from kafka.errors import IllegalArgumentError
+
+# 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 ResourceType(IntEnum):
+    """Type of kafka resource to set ACL for
+
+    The ANY value is only valid in a filter context
+    """
+
+    UNKNOWN = 0,
+    ANY = 1,
+    CLUSTER = 4,
+    DELEGATION_TOKEN = 6,
+    GROUP = 3,
+    TOPIC = 2,
+    TRANSACTIONAL_ID = 5
+
+
+class ACLOperation(IntEnum):
+    """Type of operation
+
+    The ANY value is only valid in a filter context
+    """
+
+    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
+
+    The ANY value is only valid in a filter context
+    """
+
+    ANY = 1,
+    DENY = 2,
+    ALLOW = 3
+
+
+class ACLResourcePatternType(IntEnum):
+    """An enumerated type of resource patterns
+
+    More details on the pattern types and how they work
+    can be found in KIP-290 (Support for prefixed ACLs)
+    https://cwiki.apache.org/confluence/display/KAFKA/KIP-290%3A+Support+for+Prefixed+ACLs
+    """
+
+    ANY = 1,
+    MATCH = 2,
+    LITERAL = 3,
+    PREFIXED = 4
+
+
+class ACLFilter(object):
+    """Represents a filter to use with describing and deleting ACLs
+
+    The difference between this class and the ACL class is mainly that
+    we allow using ANY with the operation, permission, and resource type objects
+    to fetch ALCs matching any of the properties.
+
+    To make a filter matching any principal, set principal to None
+    """
+
+    def __init__(
+        self,
+        principal,
+        host,
+        operation,
+        permission_type,
+        resource_pattern
+    ):
+        self.principal = principal
+        self.host = host
+        self.operation = operation
+        self.permission_type = permission_type
+        self.resource_pattern = resource_pattern
+
+        self.validate()
+
+    def validate(self):
+        if not isinstance(self.operation, ACLOperation):
+            raise IllegalArgumentError("operation must be an ACLOperation object, and cannot be ANY")
+        if not isinstance(self.permission_type, ACLPermissionType):
+            raise IllegalArgumentError("permission_type must be an ACLPermissionType object, and cannot be ANY")
+        if not isinstance(self.resource_pattern, ResourcePatternFilter):
+            raise IllegalArgumentError("resource_pattern must be a ResourcePatternFilter object")
+
+    def __repr__(self):
+        return "<ACL principal={principal}, resource={resource}, operation={operation}, type={type}, host={host}>".format(
+            principal=self.principal,
+            host=self.host,
+            operation=self.operation.name,
+            type=self.permission_type.name,
+            resource=self.resource_pattern
+        )
+
+
+class ACL(ACLFilter):
+    """Represents a concrete ACL for a specific ResourcePattern
+
+    In kafka an ACL is a 4-tuple of (principal, host, operation, permission_type)
+    that limits who can do what on a specific resource (or since KIP-290 a resource pattern)
+
+    Terminology:
+    Principal -> This is the identifier for the user. Depending on the authorization method used (SSL, SASL etc)
+        the principal will look different. See http://kafka.apache.org/documentation/#security_authz for details.
+        The principal must be on the format "User:<name>" or kafka will treat it as invalid. It's possible to use
+        other principal types than "User" if using a custom authorizer for the cluster.
+    Host -> This must currently be an IP address. It cannot be a range, and it cannot be a domain name.
+        It can be set to "*", which is special cased in kafka to mean "any host"
+    Operation -> Which client operation this ACL refers to. Has different meaning depending
+        on the resource type the ACL refers to. See https://docs.confluent.io/current/kafka/authorization.html#acl-format
+        for a list of which combinations of resource/operation that unlocks which kafka APIs
+    Permission Type: Whether this ACL is allowing or denying access
+    Resource Pattern -> This is a representation of the resource or resource pattern that the ACL
+        refers to. See the ResourcePattern class for details.
+
+    """
+
+    def __init__(
+            self,
+            principal,
+            host,
+            operation,
+            permission_type,
+            resource_pattern
+    ):
+        super(ACL, self).__init__(principal, host, operation, permission_type, resource_pattern)
+        self.validate()
+
+    def validate(self):
+        if self.operation == ACLOperation.ANY:
+            raise IllegalArgumentError("operation cannot be ANY")
+        if self.permission_type == ACLPermissionType.ANY:
+            raise IllegalArgumentError("permission_type cannot be ANY")
+        if not isinstance(self.resource_pattern, ResourcePattern):
+            raise IllegalArgumentError("resource_pattern must be a ResourcePattern object")
+
+
+class ResourcePatternFilter(object):
+    def __init__(
+            self,
+            resource_type,
+            resource_name,
+            pattern_type
+    ):
+        self.resource_type = resource_type
+        self.resource_name = resource_name
+        self.pattern_type = pattern_type
+
+        self.validate()
+
+    def validate(self):
+        if not isinstance(self.resource_type, ResourceType):
+            raise IllegalArgumentError("resource_type must be a ResourceType object")
+        if not isinstance(self.pattern_type, ACLResourcePatternType):
+            raise IllegalArgumentError("pattern_type must be an ACLResourcePatternType object")
+
+    def __repr__(self):
+        return "<ResourcePattern type={}, name={}, pattern={}>".format(
+            self.resource_type.name,
+            self.resource_name,
+            self.pattern_type.name
+        )
+
+
+class ResourcePattern(ResourcePatternFilter):
+    """A resource pattern to apply the ACL to
+
+    Resource patterns are used to be able to specify which resources an ACL
+    describes in a more flexible way than just pointing to a literal topic name for example.
+    Since KIP-290 (kafka 2.0) it's possible to set an ACL for a prefixed resource name, which
+    can cut down considerably on the number of ACLs needed when the number of topics and
+    consumer groups start to grow.
+    The default pattern_type is LITERAL, and it describes a specific resource. This is also how
+    ACLs worked before the introduction of prefixed ACLs
+    """
+
+    def __init__(
+            self,
+            resource_type,
+            resource_name,
+            pattern_type=ACLResourcePatternType.LITERAL
+    ):
+        super(ResourcePattern, self).__init__(resource_type, resource_name, pattern_type)
+        self.validate()
+
+    def validate(self):
+        if self.resource_type == ResourceType.ANY:
+            raise IllegalArgumentError("resource_type cannot be ANY")
+        if self.pattern_type in [ACLResourcePatternType.ANY, ACLResourcePatternType.MATCH]:
+            raise IllegalArgumentError(
+                "pattern_type cannot be {} on a concrete ResourcePattern".format(self.pattern_type.name)
+            )