diff --git a/clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java b/clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java index d264ef1a4f627..feba875ea6291 100644 --- a/clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java +++ b/clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java @@ -19,48 +19,60 @@ import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.resource.Resource; +import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePattern; import java.util.Objects; /** - * Represents a binding between a resource and an access control entry. + * Represents a binding between a resource pattern and an access control entry. * * The API for this class is still evolving and we may break compatibility in minor releases, if necessary. */ @InterfaceStability.Evolving public class AclBinding { - private final Resource resource; + private final ResourcePattern pattern; private final AccessControlEntry entry; + /** + * Create an instance of this class with the provided parameters. + * + * @param pattern non-null resource pattern. + * @param entry non-null entry + */ + public AclBinding(ResourcePattern pattern, AccessControlEntry entry) { + this.pattern = Objects.requireNonNull(pattern, "pattern"); + this.entry = Objects.requireNonNull(entry, "entry"); + } + /** * Create an instance of this class with the provided parameters. * * @param resource non-null resource * @param entry non-null entry + * @deprecated Since 2.0. Use {@link #AclBinding(ResourcePattern, AccessControlEntry)} */ + @Deprecated public AclBinding(Resource resource, AccessControlEntry entry) { - Objects.requireNonNull(resource); - this.resource = resource; - Objects.requireNonNull(entry); - this.entry = entry; + this(new ResourcePattern(resource.resourceType(), resource.name(), ResourceNameType.LITERAL), entry); } /** - * Return true if this binding has any UNKNOWN components. + * @return true if this binding has any UNKNOWN components. */ public boolean isUnknown() { - return resource.isUnknown() || entry.isUnknown(); + return pattern.isUnknown() || entry.isUnknown(); } /** - * Return the resource for this binding. + * @return the resource pattern for this binding. */ - public Resource resource() { - return resource; + public ResourcePattern pattern() { + return pattern; } /** - * Return the access control entry for this binding. + * @return the access control entry for this binding. */ public final AccessControlEntry entry() { return entry; @@ -70,24 +82,25 @@ public final AccessControlEntry entry() { * Create a filter which matches only this AclBinding. */ public AclBindingFilter toFilter() { - return new AclBindingFilter(resource.toFilter(), entry.toFilter()); + return new AclBindingFilter(pattern.toFilter(), entry.toFilter()); } @Override public String toString() { - return "(resource=" + resource + ", entry=" + entry + ")"; + return "(pattern=" + pattern + ", entry=" + entry + ")"; } @Override public boolean equals(Object o) { - if (!(o instanceof AclBinding)) - return false; - AclBinding other = (AclBinding) o; - return resource.equals(other.resource) && entry.equals(other.entry); + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + AclBinding that = (AclBinding) o; + return Objects.equals(pattern, that.pattern) && + Objects.equals(entry, that.entry); } @Override public int hashCode() { - return Objects.hash(resource, entry); + return Objects.hash(pattern, entry); } } diff --git a/clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java b/clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java index 5841b5aeb9a0e..9d13fa7ae04d1 100644 --- a/clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java +++ b/clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java @@ -19,6 +19,8 @@ import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.resource.ResourceFilter; +import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePatternFilter; import java.util.Objects; @@ -29,43 +31,53 @@ */ @InterfaceStability.Evolving public class AclBindingFilter { - private final ResourceFilter resourceFilter; + private final ResourcePatternFilter patternFilter; private final AccessControlEntryFilter entryFilter; /** * A filter which matches any ACL binding. */ - public static final AclBindingFilter ANY = new AclBindingFilter(ResourceFilter.ANY, AccessControlEntryFilter.ANY); + public static final AclBindingFilter ANY = new AclBindingFilter(ResourcePatternFilter.ANY, AccessControlEntryFilter.ANY); + + /** + * Create an instance of this filter with the provided parameters. + * + * @param patternFilter non-null pattern filter + * @param entryFilter non-null access control entry filter + */ + public AclBindingFilter(ResourcePatternFilter patternFilter, AccessControlEntryFilter entryFilter) { + this.patternFilter = Objects.requireNonNull(patternFilter, "patternFilter"); + this.entryFilter = Objects.requireNonNull(entryFilter, "entryFilter"); + } /** * Create an instance of this filter with the provided parameters. * * @param resourceFilter non-null resource filter * @param entryFilter non-null access control entry filter + * @deprecated Since 2.0. Use {@link #AclBindingFilter(ResourcePatternFilter, AccessControlEntryFilter)} */ + @Deprecated public AclBindingFilter(ResourceFilter resourceFilter, AccessControlEntryFilter entryFilter) { - Objects.requireNonNull(resourceFilter); - this.resourceFilter = resourceFilter; - Objects.requireNonNull(entryFilter); - this.entryFilter = entryFilter; + this(new ResourcePatternFilter(resourceFilter.resourceType(), resourceFilter.name(), ResourceNameType.LITERAL), entryFilter); } /** - * Return true if this filter has any UNKNOWN components. + * @return {@code true} if this filter has any UNKNOWN components. */ public boolean isUnknown() { - return resourceFilter.isUnknown() || entryFilter.isUnknown(); + return patternFilter.isUnknown() || entryFilter.isUnknown(); } /** - * Return the resource filter. + * @return the resource pattern filter. */ - public ResourceFilter resourceFilter() { - return resourceFilter; + public ResourcePatternFilter patternFilter() { + return patternFilter; } /** - * Return the access control entry filter. + * @return the access control entry filter. */ public final AccessControlEntryFilter entryFilter() { return entryFilter; @@ -73,15 +85,16 @@ public final AccessControlEntryFilter entryFilter() { @Override public String toString() { - return "(resourceFilter=" + resourceFilter + ", entryFilter=" + entryFilter + ")"; + return "(patternFilter=" + patternFilter + ", entryFilter=" + entryFilter + ")"; } @Override public boolean equals(Object o) { - if (!(o instanceof AclBindingFilter)) - return false; - AclBindingFilter other = (AclBindingFilter) o; - return resourceFilter.equals(other.resourceFilter) && entryFilter.equals(other.entryFilter); + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + AclBindingFilter that = (AclBindingFilter) o; + return Objects.equals(patternFilter, that.patternFilter) && + Objects.equals(entryFilter, that.entryFilter); } /** @@ -89,14 +102,14 @@ public boolean equals(Object o) { * there are no ANY or UNKNOWN fields. */ public boolean matchesAtMostOne() { - return resourceFilter.matchesAtMostOne() && entryFilter.matchesAtMostOne(); + return patternFilter.matchesAtMostOne() && entryFilter.matchesAtMostOne(); } /** * Return a string describing an ANY or UNKNOWN field, or null if there is no such field. */ public String findIndefiniteField() { - String indefinite = resourceFilter.findIndefiniteField(); + String indefinite = patternFilter.findIndefiniteField(); if (indefinite != null) return indefinite; return entryFilter.findIndefiniteField(); @@ -106,11 +119,11 @@ public String findIndefiniteField() { * Return true if the resource filter matches the binding's resource and the entry filter matches binding's entry. */ public boolean matches(AclBinding binding) { - return resourceFilter.matches(binding.resource()) && entryFilter.matches(binding.entry()); + return patternFilter.matches(binding.pattern()) && entryFilter.matches(binding.entry()); } @Override public int hashCode() { - return Objects.hash(resourceFilter, entryFilter); + return Objects.hash(patternFilter, entryFilter); } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java index 00f65c6ca1cb2..6782f7820ce60 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java @@ -19,13 +19,13 @@ import org.apache.kafka.common.acl.AccessControlEntry; import org.apache.kafka.common.acl.AclBinding; +import org.apache.kafka.common.resource.ResourcePattern; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.types.ArrayOf; import org.apache.kafka.common.protocol.types.Field; import org.apache.kafka.common.protocol.types.Schema; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.Resource; import org.apache.kafka.common.resource.ResourceNameType; import org.apache.kafka.common.utils.Utils; @@ -81,9 +81,9 @@ public AclCreation(AclBinding acl) { } static AclCreation fromStruct(Struct struct) { - Resource resource = RequestUtils.resourceFromStructFields(struct); + ResourcePattern pattern = RequestUtils.resourcePatternromStructFields(struct); AccessControlEntry entry = RequestUtils.aceFromStructFields(struct); - return new AclCreation(new AclBinding(resource, entry)); + return new AclCreation(new AclBinding(pattern, entry)); } public AclBinding acl() { @@ -91,7 +91,7 @@ public AclBinding acl() { } void setStructFields(Struct struct) { - RequestUtils.resourceSetStructFields(acl.resource(), struct); + RequestUtils.resourcePatternSetStructFields(acl.pattern(), struct); RequestUtils.aceSetStructFields(acl.entry(), struct); } @@ -179,12 +179,19 @@ private void validate(List aclCreations) { if (version() == 0) { final boolean unsupported = aclCreations.stream() .map(AclCreation::acl) - .map(AclBinding::resource) - .map(Resource::nameType) + .map(AclBinding::pattern) + .map(ResourcePattern::nameType) .anyMatch(nameType -> nameType != ResourceNameType.LITERAL); if (unsupported) { throw new UnsupportedVersionException("Version 0 only supports literal resource name types"); } } + + final boolean unknown = aclCreations.stream() + .map(AclCreation::acl) + .anyMatch(AclBinding::isUnknown); + if (unknown) { + throw new IllegalArgumentException("You can not create ACL bindings with unknown elements"); + } } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java index d896bb2c88c3d..9bb15a358f21c 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java @@ -24,8 +24,8 @@ import org.apache.kafka.common.protocol.types.Field; import org.apache.kafka.common.protocol.types.Schema; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.ResourceFilter; import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePatternFilter; import org.apache.kafka.common.utils.Utils; import java.nio.ByteBuffer; @@ -107,7 +107,7 @@ public DeleteAclsRequest(Struct struct, short version) { this.filters = new ArrayList<>(); for (Object filterStructObj : struct.getArray(FILTERS)) { Struct filterStruct = (Struct) filterStructObj; - ResourceFilter resourceFilter = RequestUtils.resourceFilterFromStructFields(filterStruct); + ResourcePatternFilter resourceFilter = RequestUtils.resourcePatternFilterFromStructFields(filterStruct); AccessControlEntryFilter aceFilter = RequestUtils.aceFilterFromStructFields(filterStruct); this.filters.add(new AclBindingFilter(resourceFilter, aceFilter)); } @@ -123,7 +123,7 @@ protected Struct toStruct() { List filterStructs = new ArrayList<>(); for (AclBindingFilter filter : filters) { Struct filterStruct = struct.instance(FILTERS); - RequestUtils.resourceFilterSetStructFields(filter.resourceFilter(), filterStruct); + RequestUtils.resourcePatternFilterSetStructFields(filter.patternFilter(), filterStruct); RequestUtils.aceFilterSetStructFields(filter.entryFilter(), filterStruct); filterStructs.add(filterStruct); } @@ -156,12 +156,17 @@ public static DeleteAclsRequest parse(ByteBuffer buffer, short version) { private void validate(short version, List filters) { if (version == 0) { final boolean unsupported = filters.stream() - .map(AclBindingFilter::resourceFilter) - .map(ResourceFilter::nameType) + .map(AclBindingFilter::patternFilter) + .map(ResourcePatternFilter::nameType) .anyMatch(nameType -> nameType != ResourceNameType.LITERAL); if (unsupported) { throw new UnsupportedVersionException("Version 0 only supports literal resource name types"); } } + + final boolean unknown = filters.stream().anyMatch(AclBindingFilter::isUnknown); + if (unknown) { + throw new IllegalArgumentException("Filters contain UNKNOWN elements"); + } } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java index 1790457f7e125..112c6a34e8fc1 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java @@ -18,6 +18,7 @@ import org.apache.kafka.common.acl.AccessControlEntry; import org.apache.kafka.common.acl.AclBinding; +import org.apache.kafka.common.resource.ResourcePattern; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; @@ -25,7 +26,6 @@ import org.apache.kafka.common.protocol.types.Field; import org.apache.kafka.common.protocol.types.Schema; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.Resource; import org.apache.kafka.common.resource.ResourceNameType; import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; @@ -180,7 +180,7 @@ public DeleteAclsResponse(Struct struct) { Struct matchingAclStruct = (Struct) matchingAclStructObj; ApiError matchError = new ApiError(matchingAclStruct); AccessControlEntry entry = RequestUtils.aceFromStructFields(matchingAclStruct); - Resource resource = RequestUtils.resourceFromStructFields(matchingAclStruct); + ResourcePattern resource = RequestUtils.resourcePatternromStructFields(matchingAclStruct); deletions.add(new AclDeletionResult(matchError, new AclBinding(resource, entry))); } this.responses.add(new AclFilterResponse(error, deletions)); @@ -201,7 +201,7 @@ protected Struct toStruct(short version) { for (AclDeletionResult deletion : response.deletions()) { Struct deletionStruct = responseStruct.instance(MATCHING_ACLS_KEY_NAME); deletion.error.write(deletionStruct); - RequestUtils.resourceSetStructFields(deletion.acl().resource(), deletionStruct); + RequestUtils.resourcePatternSetStructFields(deletion.acl().pattern(), deletionStruct); RequestUtils.aceSetStructFields(deletion.acl().entry(), deletionStruct); deletionStructs.add(deletionStruct); } @@ -247,12 +247,20 @@ private void validate(short version) { final boolean unsupported = responses.stream() .flatMap(r -> r.deletions.stream()) .map(AclDeletionResult::acl) - .map(AclBinding::resource) - .map(Resource::nameType) + .map(AclBinding::pattern) + .map(ResourcePattern::nameType) .anyMatch(nameType -> nameType != ResourceNameType.LITERAL); if (unsupported) { throw new UnsupportedVersionException("Version 0 only supports literal resource name types"); } } + + final boolean unknown = responses.stream() + .flatMap(r -> r.deletions.stream()) + .map(AclDeletionResult::acl) + .anyMatch(AclBinding::isUnknown); + if (unknown) { + throw new IllegalArgumentException("Response contains UNKNOWN elements"); + } } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java index d3a04d09a1349..50963602b28c1 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java @@ -17,14 +17,13 @@ package org.apache.kafka.common.requests; import org.apache.kafka.common.acl.AccessControlEntryFilter; -import org.apache.kafka.common.acl.AclBinding; import org.apache.kafka.common.acl.AclBindingFilter; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.types.Schema; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.ResourceFilter; import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePatternFilter; import java.nio.ByteBuffer; import java.util.Collections; @@ -95,7 +94,7 @@ public String toString() { public DescribeAclsRequest(Struct struct, short version) { super(version); - ResourceFilter resourceFilter = RequestUtils.resourceFilterFromStructFields(struct); + ResourcePatternFilter resourceFilter = RequestUtils.resourcePatternFilterFromStructFields(struct); AccessControlEntryFilter entryFilter = RequestUtils.aceFilterFromStructFields(struct); this.filter = new AclBindingFilter(resourceFilter, entryFilter); } @@ -103,7 +102,7 @@ public DescribeAclsRequest(Struct struct, short version) { @Override protected Struct toStruct() { Struct struct = new Struct(ApiKeys.DESCRIBE_ACLS.requestSchema(version())); - RequestUtils.resourceFilterSetStructFields(filter.resourceFilter(), struct); + RequestUtils.resourcePatternFilterSetStructFields(filter.patternFilter(), struct); RequestUtils.aceFilterSetStructFields(filter.entryFilter(), struct); return struct; } @@ -115,7 +114,7 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable throwable case 0: case 1: return new DescribeAclsResponse(throttleTimeMs, ApiError.fromThrowable(throwable), - Collections.emptySet()); + Collections.emptySet()); default: throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d", versionId, this.getClass().getSimpleName(), ApiKeys.DESCRIBE_ACLS.latestVersion())); @@ -131,8 +130,12 @@ public AclBindingFilter filter() { } private void validate(AclBindingFilter filter, short version) { - if (version == 0 && filter.resourceFilter().nameType() != ResourceNameType.LITERAL) { + if (version == 0 && filter.patternFilter().nameType() != ResourceNameType.LITERAL) { throw new UnsupportedVersionException("Version 0 only supports literal resource name types"); } + + if (filter.isUnknown()) { + throw new IllegalArgumentException("Filter contain UNKNOWN elements"); + } } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java index b6673d9b4875c..66f2895d310c2 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java @@ -19,6 +19,7 @@ import org.apache.kafka.common.acl.AccessControlEntry; import org.apache.kafka.common.acl.AclBinding; +import org.apache.kafka.common.resource.ResourcePattern; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; @@ -26,7 +27,6 @@ import org.apache.kafka.common.protocol.types.Field; import org.apache.kafka.common.protocol.types.Schema; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.Resource; import org.apache.kafka.common.resource.ResourceNameType; import java.nio.ByteBuffer; @@ -113,11 +113,11 @@ public DescribeAclsResponse(Struct struct) { this.acls = new ArrayList<>(); for (Object resourceStructObj : struct.getArray(RESOURCES_KEY_NAME)) { Struct resourceStruct = (Struct) resourceStructObj; - Resource resource = RequestUtils.resourceFromStructFields(resourceStruct); + ResourcePattern pattern = RequestUtils.resourcePatternromStructFields(resourceStruct); for (Object aclDataStructObj : resourceStruct.getArray(ACLS_KEY_NAME)) { Struct aclDataStruct = (Struct) aclDataStructObj; AccessControlEntry entry = RequestUtils.aceFromStructFields(aclDataStruct); - this.acls.add(new AclBinding(resource, entry)); + this.acls.add(new AclBinding(pattern, entry)); } } } @@ -130,21 +130,18 @@ protected Struct toStruct(short version) { struct.set(THROTTLE_TIME_MS, throttleTimeMs); error.write(struct); - Map> resourceToData = new HashMap<>(); + Map> resourceToData = new HashMap<>(); for (AclBinding acl : acls) { - List entry = resourceToData.get(acl.resource()); - if (entry == null) { - entry = new ArrayList<>(); - resourceToData.put(acl.resource(), entry); - } - entry.add(acl.entry()); + resourceToData + .computeIfAbsent(acl.pattern(), k -> new ArrayList<>()) + .add(acl.entry()); } List resourceStructs = new ArrayList<>(); - for (Map.Entry> tuple : resourceToData.entrySet()) { - Resource resource = tuple.getKey(); + for (Map.Entry> tuple : resourceToData.entrySet()) { + ResourcePattern resource = tuple.getKey(); Struct resourceStruct = struct.instance(RESOURCES_KEY_NAME); - RequestUtils.resourceSetStructFields(resource, resourceStruct); + RequestUtils.resourcePatternSetStructFields(resource, resourceStruct); List dataStructs = new ArrayList<>(); for (AccessControlEntry entry : tuple.getValue()) { Struct dataStruct = resourceStruct.instance(ACLS_KEY_NAME); @@ -188,12 +185,17 @@ public boolean shouldClientThrottle(short version) { private void validate(short version) { if (version == 0) { final boolean unsupported = acls.stream() - .map(AclBinding::resource) - .map(Resource::nameType) + .map(AclBinding::pattern) + .map(ResourcePattern::nameType) .anyMatch(nameType -> nameType != ResourceNameType.LITERAL); if (unsupported) { throw new UnsupportedVersionException("Version 0 only supports literal resource name types"); } } + + final boolean unknown = acls.stream().anyMatch(AclBinding::isUnknown); + if (unknown) { + throw new IllegalArgumentException("Contain UNKNOWN elements"); + } } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java b/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java index f4f00a8064134..76b27075253d4 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java @@ -20,11 +20,11 @@ import org.apache.kafka.common.acl.AccessControlEntryFilter; import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.acl.AclPermissionType; +import org.apache.kafka.common.resource.ResourcePattern; +import org.apache.kafka.common.resource.ResourcePatternFilter; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.Resource; -import org.apache.kafka.common.resource.ResourceFilter; -import org.apache.kafka.common.resource.ResourceType; import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourceType; import static org.apache.kafka.common.protocol.CommonFields.HOST; import static org.apache.kafka.common.protocol.CommonFields.HOST_FILTER; @@ -42,39 +42,39 @@ final class RequestUtils { private RequestUtils() {} - static Resource resourceFromStructFields(Struct struct) { + static ResourcePattern resourcePatternromStructFields(Struct struct) { byte resourceType = struct.get(RESOURCE_TYPE); String name = struct.get(RESOURCE_NAME); ResourceNameType resourceNameType = ResourceNameType.LITERAL; if (struct.hasField(RESOURCE_NAME_TYPE)) { resourceNameType = ResourceNameType.fromCode(struct.get(RESOURCE_NAME_TYPE)); } - return new Resource(ResourceType.fromCode(resourceType), name, resourceNameType); + return new ResourcePattern(ResourceType.fromCode(resourceType), name, resourceNameType); } - static void resourceSetStructFields(Resource resource, Struct struct) { - struct.set(RESOURCE_TYPE, resource.resourceType().code()); - struct.set(RESOURCE_NAME, resource.name()); + static void resourcePatternSetStructFields(ResourcePattern pattern, Struct struct) { + struct.set(RESOURCE_TYPE, pattern.resourceType().code()); + struct.set(RESOURCE_NAME, pattern.name()); if (struct.hasField(RESOURCE_NAME_TYPE)) { - struct.set(RESOURCE_NAME_TYPE, resource.nameType().code()); + struct.set(RESOURCE_NAME_TYPE, pattern.nameType().code()); } } - static ResourceFilter resourceFilterFromStructFields(Struct struct) { + static ResourcePatternFilter resourcePatternFilterFromStructFields(Struct struct) { byte resourceType = struct.get(RESOURCE_TYPE); String name = struct.get(RESOURCE_NAME_FILTER); ResourceNameType resourceNameType = ResourceNameType.LITERAL; if (struct.hasField(RESOURCE_NAME_TYPE_FILTER)) { resourceNameType = ResourceNameType.fromCode(struct.get(RESOURCE_NAME_TYPE_FILTER)); } - return new ResourceFilter(ResourceType.fromCode(resourceType), name, resourceNameType); + return new ResourcePatternFilter(ResourceType.fromCode(resourceType), name, resourceNameType); } - static void resourceFilterSetStructFields(ResourceFilter resourceFilter, Struct struct) { - struct.set(RESOURCE_TYPE, resourceFilter.resourceType().code()); - struct.set(RESOURCE_NAME_FILTER, resourceFilter.name()); + static void resourcePatternFilterSetStructFields(ResourcePatternFilter patternFilter, Struct struct) { + struct.set(RESOURCE_TYPE, patternFilter.resourceType().code()); + struct.set(RESOURCE_NAME_FILTER, patternFilter.name()); if (struct.hasField(RESOURCE_NAME_TYPE_FILTER)) { - struct.set(RESOURCE_NAME_TYPE_FILTER, resourceFilter.nameType().code()); + struct.set(RESOURCE_NAME_TYPE_FILTER, patternFilter.nameType().code()); } } diff --git a/clients/src/main/java/org/apache/kafka/common/resource/Resource.java b/clients/src/main/java/org/apache/kafka/common/resource/Resource.java index a4810b27ac4ad..f41f41a04b626 100644 --- a/clients/src/main/java/org/apache/kafka/common/resource/Resource.java +++ b/clients/src/main/java/org/apache/kafka/common/resource/Resource.java @@ -22,20 +22,14 @@ import java.util.Objects; /** - * Represents a cluster resource with a tuple of (type, name, nameType). + * Represents a cluster resource with a tuple of (type, name). * * The API for this class is still evolving and we may break compatibility in minor releases, if necessary. */ @InterfaceStability.Evolving public class Resource { - /** - * A special literal resource name that corresponds to 'all resources of a certain type'. - */ - public static final String WILDCARD_RESOURCE = "*"; - private final ResourceType resourceType; private final String name; - private final ResourceNameType nameType; /** * The name of the CLUSTER resource. @@ -45,32 +39,19 @@ public class Resource { /** * A resource representing the whole cluster. */ - public final static Resource CLUSTER = new Resource(ResourceType.CLUSTER, CLUSTER_NAME, ResourceNameType.LITERAL); - - /** - * Create an instance of this class with the provided parameters. - * - * @param resourceType non-null resource type - * @param name non-null resource name - * @param nameType non-null resource name type - */ - public Resource(ResourceType resourceType, String name, ResourceNameType nameType) { - this.resourceType = Objects.requireNonNull(resourceType, "resourceType"); - this.name = Objects.requireNonNull(name, "name"); - this.nameType = Objects.requireNonNull(nameType, "nameType"); - } + public final static Resource CLUSTER = new Resource(ResourceType.CLUSTER, CLUSTER_NAME); /** * Create an instance of this class with the provided parameters. - * Resource name type would default to ResourceNameType.LITERAL. * * @param resourceType non-null resource type * @param name non-null resource name - * @deprecated Since 2.0. Use {@link #Resource(ResourceType, String, ResourceNameType)} */ - @Deprecated public Resource(ResourceType resourceType, String name) { - this(resourceType, name, ResourceNameType.LITERAL); + Objects.requireNonNull(resourceType); + this.resourceType = resourceType; + Objects.requireNonNull(name); + this.name = name; } /** @@ -80,13 +61,6 @@ public ResourceType resourceType() { return resourceType; } - /** - * Return the resource name type. - */ - public ResourceNameType nameType() { - return nameType; - } - /** * Return the resource name. */ @@ -98,36 +72,31 @@ public String name() { * Create a filter which matches only this Resource. */ public ResourceFilter toFilter() { - return new ResourceFilter(resourceType, name, nameType); + return new ResourceFilter(resourceType, name); } @Override public String toString() { - return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "" : name) + ", nameType=" + nameType + ")"; + return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "" : name) + ")"; } /** * Return true if this Resource has any UNKNOWN components. */ public boolean isUnknown() { - return resourceType.isUnknown() || nameType.isUnknown(); + return resourceType.isUnknown(); } @Override public boolean equals(Object o) { - if (this == o) - return true; - if (o == null || getClass() != o.getClass()) + if (!(o instanceof Resource)) return false; - - final Resource resource = (Resource) o; - return resourceType == resource.resourceType && - Objects.equals(name, resource.name) && - nameType == resource.nameType; + Resource other = (Resource) o; + return resourceType.equals(other.resourceType) && Objects.equals(name, other.name); } @Override public int hashCode() { - return Objects.hash(resourceType, name, nameType); + return Objects.hash(resourceType, name); } } diff --git a/clients/src/main/java/org/apache/kafka/common/resource/ResourceFilter.java b/clients/src/main/java/org/apache/kafka/common/resource/ResourceFilter.java index e197e918c26c9..0a4611f9874b2 100644 --- a/clients/src/main/java/org/apache/kafka/common/resource/ResourceFilter.java +++ b/clients/src/main/java/org/apache/kafka/common/resource/ResourceFilter.java @@ -21,8 +21,6 @@ import java.util.Objects; -import static org.apache.kafka.common.resource.Resource.WILDCARD_RESOURCE; - /** * A filter which matches Resource objects. * @@ -32,54 +30,22 @@ public class ResourceFilter { private final ResourceType resourceType; private final String name; - private final ResourceNameType nameType; /** * Matches any resource. */ - public static final ResourceFilter ANY = new ResourceFilter(ResourceType.ANY, null, ResourceNameType.ANY); + public static final ResourceFilter ANY = new ResourceFilter(ResourceType.ANY, null); /** - * Create a filter that matches {@link ResourceNameType#LITERAL literal} resources of the - * supplied {@code resourceType} and {@code name}. + * Create an instance of this class with the provided parameters. * * @param resourceType non-null resource type - * @param name resource name or {@code null}. - * If {@code null}, the filter will ignore the name of resources. - * @deprecated Since 2.0. Use {@link #ResourceFilter(ResourceType, String, ResourceNameType)} + * @param name resource name or null */ - @Deprecated public ResourceFilter(ResourceType resourceType, String name) { - this(resourceType, name, ResourceNameType.LITERAL); - } - - /** - * Create a filter that matches resources of the supplied {@code resourceType}, {@code name} and - * {@code nameType}. - *

- * If the filter has each three parameters fully supplied, then it will only match a resource that has exactly - * the same values, e.g. a filter of {@code new ResourceFilter(ResourceType.GROUP, "one", ResourceTypeName.PREFIXED)} - * will only match the resource {@code new Resource(ResourceType.GROUP, "one", ResourceTypeName.PREFIXED)}. - *

- * Any of the three parameters can be set to be ignored by the filter: - *

    - *
  • {@code resourceType} can be set to {@link ResourceType#ANY}, - * meaning it will match a resource of any resource type
    • - *
  • {@code name} can be set to {@code null}, meaning it will match a resource of any name.
    • - *
  • {@code nameType} can be set to {@link ResourceNameType#ANY}, - * meaning it will match a resource with any resource name type, including the - * {@link Resource#WILDCARD_RESOURCE wildcard resource}
    • - *
- * - * @param resourceType non-null resource type to filter by. - * @param name resource name to filter by, or {@code null}. - * If {@code null}, the filter will ignore the name of resources. - * @param nameType non-null resource name type to filter by. - */ - public ResourceFilter(ResourceType resourceType, String name, ResourceNameType nameType) { - this.resourceType = Objects.requireNonNull(resourceType, "resourceType"); + Objects.requireNonNull(resourceType); + this.resourceType = resourceType; this.name = name; - this.nameType = Objects.requireNonNull(nameType, "nameType"); } /** @@ -96,76 +62,40 @@ public String name() { return name; } - /** - * Return the resource name type. - */ - public ResourceNameType nameType() { - return nameType; - } - @Override public String toString() { - return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "" : name) + ", nameType=" + nameType + ")"; + return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "" : name) + ")"; } /** * Return true if this ResourceFilter has any UNKNOWN components. */ public boolean isUnknown() { - return resourceType.isUnknown() || nameType.isUnknown(); + return resourceType.isUnknown(); } @Override public boolean equals(Object o) { - if (this == o) - return true; - if (o == null || getClass() != o.getClass()) + if (!(o instanceof ResourceFilter)) return false; - - final ResourceFilter that = (ResourceFilter) o; - return resourceType == that.resourceType && - Objects.equals(name, that.name) && - nameType == that.nameType; + ResourceFilter other = (ResourceFilter) o; + return resourceType.equals(other.resourceType) && Objects.equals(name, other.name); } @Override public int hashCode() { - return Objects.hash(resourceType, name, nameType); + return Objects.hash(resourceType, name); } /** * Return true if this filter matches the given Resource. - * @param other the resource path under which ACLs are stored. */ - public boolean matches(final Resource other) { - throwOnInvalidParams(other); - - if (!resourceType().equals(ResourceType.ANY) && !resourceType().equals(other.resourceType())) { + public boolean matches(Resource other) { + if ((name != null) && (!name.equals(other.name()))) return false; - } - - if (!nameType().equals(ResourceNameType.ANY) && !nameType().equals(other.nameType())) { + if ((resourceType != ResourceType.ANY) && (!resourceType.equals(other.resourceType()))) return false; - } - - if (name() == null) { - return true; - } - - if (nameType().equals(other.nameType())) { - return other.name().equals(name()); - } - - switch (other.nameType()) { - case LITERAL: - return other.name().equals(name()) || other.name().equals(WILDCARD_RESOURCE); - - case PREFIXED: - return name().startsWith(other.name()); - - default: - throw new IllegalArgumentException("Unsupported ResourceNameType: " + other.nameType()); - } + return true; } /** @@ -185,20 +115,6 @@ public String findIndefiniteField() { return "Resource type is UNKNOWN."; if (name == null) return "Resource name is NULL."; - if (nameType == ResourceNameType.ANY) - return "Resource name type is ANY."; - if (nameType == ResourceNameType.UNKNOWN) - return "Resource name type is UNKNOWN."; return null; } - - private static void throwOnInvalidParams(final Resource aclPath) { - if (aclPath.resourceType().equals(ResourceType.ANY) || aclPath.resourceType().equals(ResourceType.UNKNOWN)) { - throw new IllegalArgumentException("Only concrete resource types are supported. Got: " + aclPath.resourceType()); - } - - if (aclPath.nameType().equals(ResourceNameType.ANY) || aclPath.nameType().equals(ResourceNameType.UNKNOWN)) { - throw new IllegalArgumentException("Only concrete resource name types are supported. Got: " + aclPath.nameType()); - } - } } diff --git a/clients/src/main/java/org/apache/kafka/common/resource/ResourcePattern.java b/clients/src/main/java/org/apache/kafka/common/resource/ResourcePattern.java new file mode 100644 index 0000000000000..c6aee913a65db --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/resource/ResourcePattern.java @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.common.resource; + +import org.apache.kafka.common.annotation.InterfaceStability; + +import java.util.Objects; + +/** + * Represents a pattern that is used by ACLs to match zero or more + * {@link org.apache.kafka.common.resource.Resource Resources}. + * + * The API for this class is still evolving and we may break compatibility in minor releases, if necessary. + */ +@InterfaceStability.Evolving +public class ResourcePattern { + /** + * A special literal resource name that corresponds to 'all resources of a certain type'. + */ + public static final String WILDCARD_RESOURCE = "*"; + + private final ResourceType resourceType; + private final String name; + private final ResourceNameType nameType; + + /** + * Create a pattern using the supplied parameters. + * + * @param resourceType non-null, specific, resource type + * @param name non-null resource name, which can be the {@link #WILDCARD_RESOURCE}. + * @param nameType non-null, specific, resource name type, which controls how the pattern will match resource names. + */ + public ResourcePattern(ResourceType resourceType, String name, ResourceNameType nameType) { + this.resourceType = Objects.requireNonNull(resourceType, "resourceType"); + this.name = Objects.requireNonNull(name, "name"); + this.nameType = Objects.requireNonNull(nameType, "nameType"); + + if (resourceType == ResourceType.ANY) { + throw new IllegalArgumentException("resourceType must not be ANY"); + } + + if (nameType == ResourceNameType.ANY) { + throw new IllegalArgumentException("nameType must not be ANY"); + } + } + + /** + * @return the specific resource type this pattern matches + */ + public ResourceType resourceType() { + return resourceType; + } + + /** + * @return the resource name. + */ + public String name() { + return name; + } + + /** + * @return the resource name type. + */ + public ResourceNameType nameType() { + return nameType; + } + + /** + * @return a filter which matches only this pattern. + */ + public ResourcePatternFilter toFilter() { + return new ResourcePatternFilter(resourceType, name, nameType); + } + + @Override + public String toString() { + return "ResourcePattern(resourceType=" + resourceType + ", name=" + ((name == null) ? "" : name) + ", nameType=" + nameType + ")"; + } + + /** + * @return {@code true} if this Resource has any UNKNOWN components. + */ + public boolean isUnknown() { + return resourceType.isUnknown() || nameType.isUnknown(); + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + + final ResourcePattern resource = (ResourcePattern) o; + return resourceType == resource.resourceType && + Objects.equals(name, resource.name) && + nameType == resource.nameType; + } + + @Override + public int hashCode() { + return Objects.hash(resourceType, name, nameType); + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/resource/ResourcePatternFilter.java b/clients/src/main/java/org/apache/kafka/common/resource/ResourcePatternFilter.java new file mode 100644 index 0000000000000..8b4fdc0b8556a --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/resource/ResourcePatternFilter.java @@ -0,0 +1,169 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.common.resource; + +import org.apache.kafka.common.annotation.InterfaceStability; + +import java.util.Objects; + +import static org.apache.kafka.common.resource.ResourcePattern.WILDCARD_RESOURCE; + +/** + * Represents a filter that can match {@link ResourcePattern}. + *

+ * The API for this class is still evolving and we may break compatibility in minor releases, if necessary. + */ +@InterfaceStability.Evolving +public class ResourcePatternFilter { + /** + * Matches any resource pattern. + */ + public static final ResourcePatternFilter ANY = new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.ANY); + + private final ResourceType resourceType; + private final String name; + private final ResourceNameType nameType; + + /** + * Create a filter using the supplied parameters. + * + * @param resourceType non-null resource type. + * If {@link ResourceType#ANY}, the filter will ignore the resource type of the pattern. + * If any other resource type, the filter will match only patterns with the same type. + * @param name resource name or {@code null}. + * If {@code null}, the filter will ignore the name of resources. + * If {@link ResourcePattern#WILDCARD_RESOURCE}, will match only wildcard patterns. + * @param nameType non-null resource name type. + * If {@link ResourceNameType#ANY}, the filter will match patterns that would match any + * {@code ResourceNameType}, and also any wildcards patterns. + * If any other resource name type, the filter will match only patterns with the same type. + */ + public ResourcePatternFilter(ResourceType resourceType, String name, ResourceNameType nameType) { + this.resourceType = Objects.requireNonNull(resourceType, "resourceType"); + this.name = name; + this.nameType = Objects.requireNonNull(nameType, "nameType"); + } + + /** + * @return {@code true} if this filter has any UNKNOWN components. + */ + public boolean isUnknown() { + return resourceType.isUnknown() || nameType.isUnknown(); + } + + /** + * @return the specific resource type this pattern matches + */ + public ResourceType resourceType() { + return resourceType; + } + + /** + * @return the resource name. + */ + public String name() { + return name; + } + + /** + * @return the resource name type. + */ + public ResourceNameType nameType() { + return nameType; + } + + /** + * @return {@code true} if this filter matches the given pattern. + */ + public boolean matches(ResourcePattern pattern) { + if (!resourceType.equals(ResourceType.ANY) && !resourceType.equals(pattern.resourceType())) { + return false; + } + + if (!nameType.equals(ResourceNameType.ANY) && !nameType.equals(pattern.nameType())) { + return false; + } + + if (name == null) { + return true; + } + + if (nameType.equals(pattern.nameType())) { + return name.equals(pattern.name()); + } + + switch (pattern.nameType()) { + case LITERAL: + return name.equals(pattern.name()) || pattern.name().equals(WILDCARD_RESOURCE); + + case PREFIXED: + return name.startsWith(pattern.name()); + + default: + throw new IllegalArgumentException("Unsupported ResourceNameType: " + pattern.nameType()); + } + } + + /** + * @return {@code true} if this filter could only match one pattern. + * In other words, if there are no ANY or UNKNOWN fields. + */ + public boolean matchesAtMostOne() { + return findIndefiniteField() == null; + } + + /** + * @return a string describing any ANY or UNKNOWN field, or null if there is no such field. + */ + public String findIndefiniteField() { + if (resourceType == ResourceType.ANY) + return "Resource type is ANY."; + if (resourceType == ResourceType.UNKNOWN) + return "Resource type is UNKNOWN."; + if (name == null) + return "Resource name is NULL."; + if (nameType == ResourceNameType.ANY) + return "Resource name type is ANY."; + if (nameType == ResourceNameType.UNKNOWN) + return "Resource name type is UNKNOWN."; + return null; + } + + @Override + public String toString() { + return "ResourcePattern(resourceType=" + resourceType + ", name=" + ((name == null) ? "" : name) + ", nameType=" + nameType + ")"; + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + + final ResourcePatternFilter resource = (ResourcePatternFilter) o; + return resourceType == resource.resourceType && + Objects.equals(name, resource.name) && + nameType == resource.nameType; + } + + @Override + public int hashCode() { + return Objects.hash(resourceType, name, nameType); + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java index 3a300dbfb1bfa..5cb6bbc9761fb 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java @@ -34,6 +34,8 @@ import org.apache.kafka.common.acl.AclBindingFilter; import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.acl.AclPermissionType; +import org.apache.kafka.common.resource.ResourcePattern; +import org.apache.kafka.common.resource.ResourcePatternFilter; import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.errors.AuthenticationException; import org.apache.kafka.common.errors.CoordinatorNotAvailableException; @@ -65,8 +67,6 @@ import org.apache.kafka.common.requests.MetadataRequest; import org.apache.kafka.common.requests.MetadataResponse; import org.apache.kafka.common.requests.OffsetFetchResponse; -import org.apache.kafka.common.resource.Resource; -import org.apache.kafka.common.resource.ResourceFilter; import org.apache.kafka.common.resource.ResourceNameType; import org.apache.kafka.common.resource.ResourceType; import org.apache.kafka.common.utils.MockTime; @@ -537,13 +537,13 @@ private void callAdminClientApisAndExpectAnAuthenticationError(AdminClientUnitTe } } - private static final AclBinding ACL1 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), + private static final AclBinding ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW)); - private static final AclBinding ACL2 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic4", ResourceNameType.LITERAL), + private static final AclBinding ACL2 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic4", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.DENY)); - private static final AclBindingFilter FILTER1 = new AclBindingFilter(new ResourceFilter(ResourceType.ANY, null, ResourceNameType.LITERAL), + private static final AclBindingFilter FILTER1 = new AclBindingFilter(new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.LITERAL), new AccessControlEntryFilter("User:ANONYMOUS", null, AclOperation.ANY, AclPermissionType.ANY)); - private static final AclBindingFilter FILTER2 = new AclBindingFilter(new ResourceFilter(ResourceType.ANY, null, ResourceNameType.LITERAL), + private static final AclBindingFilter FILTER2 = new AclBindingFilter(new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.LITERAL), new AccessControlEntryFilter("User:bob", null, AclOperation.ANY, AclPermissionType.ANY)); @Test diff --git a/clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java b/clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java index 6110c48e2b03d..a35faca80ecde 100644 --- a/clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java +++ b/clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java @@ -16,9 +16,9 @@ */ package org.apache.kafka.common.acl; -import org.apache.kafka.common.resource.Resource; -import org.apache.kafka.common.resource.ResourceFilter; import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePattern; +import org.apache.kafka.common.resource.ResourcePatternFilter; import org.apache.kafka.common.resource.ResourceType; import org.junit.Test; @@ -28,38 +28,38 @@ public class AclBindingTest { private static final AclBinding ACL1 = new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "", AclOperation.ALL, AclPermissionType.ALLOW)); private static final AclBinding ACL2 = new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), new AccessControlEntry("User:*", "", AclOperation.READ, AclPermissionType.ALLOW)); private static final AclBinding ACL3 = new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY)); private static final AclBinding UNKNOWN_ACL = new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.UNKNOWN, AclPermissionType.DENY)); private static final AclBindingFilter ANY_ANONYMOUS = new AclBindingFilter( - ResourceFilter.ANY, + ResourcePatternFilter.ANY, new AccessControlEntryFilter("User:ANONYMOUS", null, AclOperation.ANY, AclPermissionType.ANY)); private static final AclBindingFilter ANY_DENY = new AclBindingFilter( - ResourceFilter.ANY, + ResourcePatternFilter.ANY, new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.DENY)); private static final AclBindingFilter ANY_MYTOPIC = new AclBindingFilter( - new ResourceFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), + new ResourcePatternFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY)); @Test public void testMatching() throws Exception { assertTrue(ACL1.equals(ACL1)); final AclBinding acl1Copy = new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "", AclOperation.ALL, AclPermissionType.ALLOW)); assertTrue(ACL1.equals(acl1Copy)); assertTrue(acl1Copy.equals(ACL1)); diff --git a/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternFilterTest.java b/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternFilterTest.java new file mode 100644 index 0000000000000..87b25fcbe083e --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternFilterTest.java @@ -0,0 +1,164 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.common.acl; + +import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePattern; +import org.apache.kafka.common.resource.ResourcePatternFilter; +import org.junit.Test; + +import static org.apache.kafka.common.resource.ResourceNameType.LITERAL; +import static org.apache.kafka.common.resource.ResourceNameType.PREFIXED; +import static org.apache.kafka.common.resource.ResourceType.ANY; +import static org.apache.kafka.common.resource.ResourceType.GROUP; +import static org.apache.kafka.common.resource.ResourceType.TOPIC; +import static org.apache.kafka.common.resource.ResourceType.UNKNOWN; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class ResourcePatternFilterTest { + @Test(expected = IllegalArgumentException.class) + public void shouldThrowIfResourceTypeIsAny() { + new ResourcePatternFilter(ANY, null, ResourceNameType.ANY) + .matches(new ResourcePattern(ANY, "Name", PREFIXED)); + } + + @Test(expected = IllegalArgumentException.class) + public void shouldThrowIfResourceNameTypeIsAny() { + new ResourcePatternFilter(ANY, null, ResourceNameType.ANY) + .matches(new ResourcePattern(GROUP, "Name", ResourceNameType.ANY)); + } + + @Test + public void shouldBeUnknownIfResourceTypeUnknown() { + assertTrue(new ResourcePatternFilter(UNKNOWN, null, ResourceNameType.LITERAL).isUnknown()); + } + + @Test + public void shouldBeUnknownIfResourceNameTypeUnknown() { + assertTrue(new ResourcePatternFilter(GROUP, null, ResourceNameType.UNKNOWN).isUnknown()); + } + + @Test + public void shouldNotMatchIfDifferentResourceType() { + assertFalse(new ResourcePatternFilter(TOPIC, "Name", LITERAL) + .matches(new ResourcePattern(GROUP, "Name", LITERAL))); + } + + @Test + public void shouldNotMatchIfDifferentName() { + assertFalse(new ResourcePatternFilter(TOPIC, "Different", PREFIXED) + .matches(new ResourcePattern(TOPIC, "Name", PREFIXED))); + } + + @Test + public void shouldNotMatchIfDifferentNameCase() { + assertFalse(new ResourcePatternFilter(TOPIC, "NAME", LITERAL) + .matches(new ResourcePattern(TOPIC, "Name", LITERAL))); + } + + @Test + public void shouldNotMatchIfDifferentNameType() { + assertFalse(new ResourcePatternFilter(TOPIC, "Name", LITERAL) + .matches(new ResourcePattern(TOPIC, "Name", PREFIXED))); + } + + @Test + public void shouldMatchWhereResourceTypeIsAny() { + assertTrue(new ResourcePatternFilter(ANY, "Name", PREFIXED) + .matches(new ResourcePattern(TOPIC, "Name", PREFIXED))); + } + + @Test + public void shouldMatchWhereResourceNameIsAny() { + assertTrue(new ResourcePatternFilter(TOPIC, null, PREFIXED) + .matches(new ResourcePattern(TOPIC, "Name", PREFIXED))); + } + + @Test + public void shouldMatchWhereResourceNameTypeIsAny() { + assertTrue(new ResourcePatternFilter(TOPIC, null, ResourceNameType.ANY) + .matches(new ResourcePattern(TOPIC, "Name", PREFIXED))); + } + + @Test + public void shouldMatchLiteralIfExactMatch() { + assertTrue(new ResourcePatternFilter(TOPIC, "Name", LITERAL) + .matches(new ResourcePattern(TOPIC, "Name", LITERAL))); + } + + @Test + public void shouldMatchLiteralIfNameMatchesAndFilterIsOnAnyNameType() { + assertTrue(new ResourcePatternFilter(TOPIC, "Name", ResourceNameType.ANY) + .matches(new ResourcePattern(TOPIC, "Name", LITERAL))); + } + + @Test + public void shouldNotMatchLiteralIfNamePrefixed() { + assertFalse(new ResourcePatternFilter(TOPIC, "Name-something", ResourceNameType.ANY) + .matches(new ResourcePattern(TOPIC, "Name", LITERAL))); + } + + @Test + public void shouldMatchLiteralWildcardIfExactMatch() { + assertTrue(new ResourcePatternFilter(TOPIC, "*", LITERAL) + .matches(new ResourcePattern(TOPIC, "*", LITERAL))); + } + + @Test + public void shouldNotMatchLiteralWildcardAgainstOtherName() { + assertFalse(new ResourcePatternFilter(TOPIC, "Name", LITERAL) + .matches(new ResourcePattern(TOPIC, "*", LITERAL))); + } + + @Test + public void shouldNotMatchLiteralWildcardTheWayAround() { + assertFalse(new ResourcePatternFilter(TOPIC, "*", LITERAL) + .matches(new ResourcePattern(TOPIC, "Name", LITERAL))); + } + + @Test + public void shouldMatchLiteralWildcardIfFilterHasNameTypeOfAny() { + assertTrue(new ResourcePatternFilter(TOPIC, "Name", ResourceNameType.ANY) + .matches(new ResourcePattern(TOPIC, "*", LITERAL))); + } + + @Test + public void shouldMatchPrefixedIfExactMatch() { + assertTrue(new ResourcePatternFilter(TOPIC, "Name", PREFIXED) + .matches(new ResourcePattern(TOPIC, "Name", PREFIXED))); + } + + @Test + public void shouldNotMatchIfBothPrefixedAndFilterIsPrefixOfResource() { + assertFalse(new ResourcePatternFilter(TOPIC, "Name", PREFIXED) + .matches(new ResourcePattern(TOPIC, "Name-something", PREFIXED))); + } + + @Test + public void shouldNotMatchIfBothPrefixedAndResourceIsPrefixOfFilter() { + assertFalse(new ResourcePatternFilter(TOPIC, "Name-something", PREFIXED) + .matches(new ResourcePattern(TOPIC, "Name", PREFIXED))); + } + + @Test + public void shouldMatchPrefixedIfNamePrefixedAnyFilterTypeIsAny() { + assertTrue(new ResourcePatternFilter(TOPIC, "Name-something", ResourceNameType.ANY) + .matches(new ResourcePattern(TOPIC, "Name", PREFIXED))); + } +} \ No newline at end of file diff --git a/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternTest.java b/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternTest.java new file mode 100644 index 0000000000000..d76e213118ff8 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternTest.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.common.acl; + +import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePattern; +import org.apache.kafka.common.resource.ResourceType; +import org.junit.Test; + +public class ResourcePatternTest { + @Test(expected = IllegalArgumentException.class) + public void shouldThrowIfResourceTypeIsAny() { + new ResourcePattern(ResourceType.ANY, "name", ResourceNameType.LITERAL); + } + + @Test(expected = IllegalArgumentException.class) + public void shouldThrowIfResourceNameTypeIsAny() { + new ResourcePattern(ResourceType.TOPIC, "name", ResourceNameType.ANY); + } + + @Test(expected = NullPointerException.class) + public void shouldThrowIfResourceNameIsNull() { + new ResourcePattern(ResourceType.TOPIC, null, ResourceNameType.ANY); + } +} \ No newline at end of file diff --git a/clients/src/test/java/org/apache/kafka/common/requests/CreateAclsRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/CreateAclsRequestTest.java index 748914b493137..1f3c15c35cc17 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/CreateAclsRequestTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/CreateAclsRequestTest.java @@ -24,8 +24,8 @@ import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation; -import org.apache.kafka.common.resource.Resource; import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePattern; import org.apache.kafka.common.resource.ResourceType; import org.junit.Test; @@ -39,13 +39,16 @@ public class CreateAclsRequestTest { private static final short V0 = 0; private static final short V1 = 1; - private static final AclBinding LITERAL_ACL1 = new AclBinding(new Resource(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), + private static final AclBinding LITERAL_ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY)); - private static final AclBinding LITERAL_ACL2 = new AclBinding(new Resource(ResourceType.GROUP, "group", ResourceNameType.LITERAL), + private static final AclBinding LITERAL_ACL2 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "group", ResourceNameType.LITERAL), new AccessControlEntry("User:*", "127.0.0.1", AclOperation.WRITE, AclPermissionType.ALLOW)); - private static final AclBinding PREFIXED_ACL1 = new AclBinding(new Resource(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + private static final AclBinding PREFIXED_ACL1 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + new AccessControlEntry("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); + + private static final AclBinding UNKNOWN_ACL1 = new AclBinding(new ResourcePattern(ResourceType.UNKNOWN, "unknown", ResourceNameType.LITERAL), new AccessControlEntry("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); @Test(expected = UnsupportedVersionException.class) @@ -53,6 +56,11 @@ public void shouldThrowOnV0IfNotLiteral() { new CreateAclsRequest(V0, aclCreations(PREFIXED_ACL1)); } + @Test(expected = IllegalArgumentException.class) + public void shouldThrowOnIfUnknown() { + new CreateAclsRequest(V0, aclCreations(UNKNOWN_ACL1)); + } + @Test public void shouldRoundTripV0() { final CreateAclsRequest original = new CreateAclsRequest(V0, aclCreations(LITERAL_ACL1, LITERAL_ACL2)); diff --git a/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsRequestTest.java index 7761337867148..42b7dcc808f0b 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsRequestTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsRequestTest.java @@ -21,9 +21,9 @@ import org.apache.kafka.common.acl.AclBindingFilter; import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.acl.AclPermissionType; +import org.apache.kafka.common.resource.ResourcePatternFilter; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.ResourceFilter; import org.apache.kafka.common.resource.ResourceNameType; import org.apache.kafka.common.resource.ResourceType; import org.junit.Test; @@ -37,13 +37,16 @@ public class DeleteAclsRequestTest { private static final short V0 = 0; private static final short V1 = 1; - private static final AclBindingFilter LITERAL_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), + private static final AclBindingFilter LITERAL_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), new AccessControlEntryFilter("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY)); - private static final AclBindingFilter PREFIXED_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + private static final AclBindingFilter PREFIXED_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); - private static final AclBindingFilter ANY_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + private static final AclBindingFilter ANY_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); + + private static final AclBindingFilter UNKNOWN_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.UNKNOWN, "prefix", ResourceNameType.PREFIXED), new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); @Test(expected = UnsupportedVersionException.class) @@ -51,6 +54,11 @@ public void shouldThrowOnV0IfNotLiteral() { new DeleteAclsRequest(V0, aclFilters(PREFIXED_FILTER)); } + @Test(expected = IllegalArgumentException.class) + public void shouldThrowOnUnknownElements() { + new DeleteAclsRequest(V1, aclFilters(UNKNOWN_FILTER)); + } + @Test public void shouldRoundTripV0() { final DeleteAclsRequest original = new DeleteAclsRequest(V0, aclFilters(LITERAL_FILTER)); diff --git a/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsResponseTest.java index f8e9148cbd1e6..b6d9f92bf716c 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsResponseTest.java @@ -21,11 +21,11 @@ import org.apache.kafka.common.acl.AclBinding; import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.acl.AclPermissionType; +import org.apache.kafka.common.resource.ResourcePattern; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.types.Struct; import org.apache.kafka.common.requests.DeleteAclsResponse.AclDeletionResult; import org.apache.kafka.common.requests.DeleteAclsResponse.AclFilterResponse; -import org.apache.kafka.common.resource.Resource; import org.apache.kafka.common.resource.ResourceNameType; import org.apache.kafka.common.resource.ResourceType; import org.junit.Test; @@ -41,24 +41,34 @@ public class DeleteAclsResponseTest { private static final short V0 = 0; private static final short V1 = 1; - private static final AclBinding LITERAL_ACL1 = new AclBinding(new org.apache.kafka.common.resource.Resource(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), + private static final AclBinding LITERAL_ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY)); - private static final AclBinding LITERAL_ACL2 = new AclBinding(new org.apache.kafka.common.resource.Resource(ResourceType.GROUP, "group", ResourceNameType.LITERAL), + private static final AclBinding LITERAL_ACL2 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "group", ResourceNameType.LITERAL), new AccessControlEntry("User:*", "127.0.0.1", AclOperation.WRITE, AclPermissionType.ALLOW)); - private static final AclBinding PREFIXED_ACL1 = new AclBinding(new Resource(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + private static final AclBinding PREFIXED_ACL1 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), new AccessControlEntry("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); + private static final AclBinding UNKNOWN_ACL = new AclBinding(new ResourcePattern(ResourceType.UNKNOWN, "group", ResourceNameType.LITERAL), + new AccessControlEntry("User:*", "127.0.0.1", AclOperation.WRITE, AclPermissionType.ALLOW)); + private static final AclFilterResponse LITERAL_RESPONSE = new AclFilterResponse(aclDeletions(LITERAL_ACL1, LITERAL_ACL2)); private static final AclFilterResponse PREFIXED_RESPONSE = new AclFilterResponse(aclDeletions(LITERAL_ACL1, PREFIXED_ACL1)); + private static final AclFilterResponse UNKNOWN_RESPONSE = new AclFilterResponse(aclDeletions(UNKNOWN_ACL)); + @Test(expected = UnsupportedVersionException.class) public void shouldThrowOnV0IfNotLiteral() { new DeleteAclsResponse(10, aclResponses(PREFIXED_RESPONSE)).toStruct(V0); } + @Test(expected = IllegalArgumentException.class) + public void shouldThrowOnIfUnknown() { + new DeleteAclsResponse(10, aclResponses(UNKNOWN_RESPONSE)).toStruct(V1); + } + @Test public void shouldRoundTripV0() { final DeleteAclsResponse original = new DeleteAclsResponse(10, aclResponses(LITERAL_RESPONSE)); diff --git a/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsRequestTest.java index 543cf37576ca7..3e4e531e29daf 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsRequestTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsRequestTest.java @@ -21,9 +21,9 @@ import org.apache.kafka.common.acl.AclBindingFilter; import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.acl.AclPermissionType; +import org.apache.kafka.common.resource.ResourcePatternFilter; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.ResourceFilter; import org.apache.kafka.common.resource.ResourceNameType; import org.apache.kafka.common.resource.ResourceType; import org.junit.Test; @@ -34,20 +34,28 @@ public class DescribeAclsRequestTest { private static final short V0 = 0; private static final short V1 = 1; - private static final AclBindingFilter LITERAL_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), + private static final AclBindingFilter LITERAL_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), new AccessControlEntryFilter("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY)); - private static final AclBindingFilter PREFIXED_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + private static final AclBindingFilter PREFIXED_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); - private static final AclBindingFilter ANY_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + private static final AclBindingFilter ANY_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); + private static final AclBindingFilter UNKNOWN_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.UNKNOWN, "foo", ResourceNameType.LITERAL), + new AccessControlEntryFilter("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY)); + @Test(expected = UnsupportedVersionException.class) public void shouldThrowOnV0IfNotLiteral() { new DescribeAclsRequest(PREFIXED_FILTER, V0); } + @Test(expected = IllegalArgumentException.class) + public void shouldThrowIfUnknown() { + new DescribeAclsRequest(UNKNOWN_FILTER, V0); + } + @Test public void shouldRoundTripV0() { final DescribeAclsRequest original = new DescribeAclsRequest(LITERAL_FILTER, V0); diff --git a/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsResponseTest.java index 81cf518536777..2d3ac8f847638 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsResponseTest.java @@ -23,8 +23,8 @@ import org.apache.kafka.common.acl.AclPermissionType; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.protocol.types.Struct; -import org.apache.kafka.common.resource.Resource; import org.apache.kafka.common.resource.ResourceNameType; +import org.apache.kafka.common.resource.ResourcePattern; import org.apache.kafka.common.resource.ResourceType; import org.junit.Test; @@ -39,20 +39,28 @@ public class DescribeAclsResponseTest { private static final short V0 = 0; private static final short V1 = 1; - private static final AclBinding LITERAL_ACL1 = new AclBinding(new Resource(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), + private static final AclBinding LITERAL_ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY)); - private static final AclBinding LITERAL_ACL2 = new AclBinding(new Resource(ResourceType.GROUP, "group", ResourceNameType.LITERAL), + private static final AclBinding LITERAL_ACL2 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "group", ResourceNameType.LITERAL), new AccessControlEntry("User:*", "127.0.0.1", AclOperation.WRITE, AclPermissionType.ALLOW)); - private static final AclBinding PREFIXED_ACL1 = new AclBinding(new Resource(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), + private static final AclBinding PREFIXED_ACL1 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED), new AccessControlEntry("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW)); + private static final AclBinding UNKNOWN_ACL = new AclBinding(new ResourcePattern(ResourceType.UNKNOWN, "foo", ResourceNameType.LITERAL), + new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY)); + @Test(expected = UnsupportedVersionException.class) public void shouldThrowOnV0IfNotLiteral() { new DescribeAclsResponse(10, ApiError.NONE, aclBindings(PREFIXED_ACL1)).toStruct(V0); } + @Test(expected = IllegalArgumentException.class) + public void shouldThrowIfUnknown() { + new DescribeAclsResponse(10, ApiError.NONE, aclBindings(UNKNOWN_ACL)).toStruct(V0); + } + @Test public void shouldRoundTripV0() { final DescribeAclsResponse original = new DescribeAclsResponse(10, ApiError.NONE, aclBindings(LITERAL_ACL1, LITERAL_ACL2)); diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java index e0c72a26f547b..f537f48f8f428 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java @@ -25,6 +25,8 @@ import org.apache.kafka.common.acl.AclBindingFilter; import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.acl.AclPermissionType; +import org.apache.kafka.common.resource.ResourcePattern; +import org.apache.kafka.common.resource.ResourcePatternFilter; import org.apache.kafka.common.errors.InvalidReplicaAssignmentException; import org.apache.kafka.common.errors.InvalidTopicException; import org.apache.kafka.common.errors.NotCoordinatorException; @@ -45,8 +47,6 @@ import org.apache.kafka.common.requests.CreateAclsResponse.AclCreationResponse; import org.apache.kafka.common.requests.DeleteAclsResponse.AclDeletionResult; import org.apache.kafka.common.requests.DeleteAclsResponse.AclFilterResponse; -import org.apache.kafka.common.resource.Resource; -import org.apache.kafka.common.resource.ResourceFilter; import org.apache.kafka.common.resource.ResourceNameType; import org.apache.kafka.common.resource.ResourceType; import org.apache.kafka.common.security.auth.KafkaPrincipal; @@ -1087,23 +1087,23 @@ private TxnOffsetCommitResponse createTxnOffsetCommitResponse() { private DescribeAclsRequest createListAclsRequest() { return new DescribeAclsRequest.Builder(new AclBindingFilter( - new ResourceFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), + new ResourcePatternFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY))).build(); } private DescribeAclsResponse createDescribeAclsResponse() { return new DescribeAclsResponse(0, ApiError.NONE, Collections.singleton(new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.ALLOW)))); } private CreateAclsRequest createCreateAclsRequest() { List creations = new ArrayList<>(); creations.add(new AclCreation(new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.ALLOW)))); creations.add(new AclCreation(new AclBinding( - new Resource(ResourceType.GROUP, "mygroup", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.GROUP, "mygroup", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.DENY)))); return new CreateAclsRequest.Builder(creations).build(); } @@ -1116,10 +1116,10 @@ private CreateAclsResponse createCreateAclsResponse() { private DeleteAclsRequest createDeleteAclsRequest() { List filters = new ArrayList<>(); filters.add(new AclBindingFilter( - new ResourceFilter(ResourceType.ANY, null, ResourceNameType.LITERAL), + new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.LITERAL), new AccessControlEntryFilter("User:ANONYMOUS", null, AclOperation.ANY, AclPermissionType.ANY))); filters.add(new AclBindingFilter( - new ResourceFilter(ResourceType.ANY, null, ResourceNameType.LITERAL), + new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.LITERAL), new AccessControlEntryFilter("User:bob", null, AclOperation.ANY, AclPermissionType.ANY))); return new DeleteAclsRequest.Builder(filters).build(); } @@ -1128,10 +1128,10 @@ private DeleteAclsResponse createDeleteAclsResponse() { List responses = new ArrayList<>(); responses.add(new AclFilterResponse(Utils.mkSet( new AclDeletionResult(new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))), new AclDeletionResult(new AclBinding( - new Resource(ResourceType.TOPIC, "mytopic4", ResourceNameType.LITERAL), + new ResourcePattern(ResourceType.TOPIC, "mytopic4", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.DENY)))))); responses.add(new AclFilterResponse(new ApiError(Errors.SECURITY_DISABLED, "No security"), Collections.emptySet())); diff --git a/clients/src/test/java/org/apache/kafka/common/resource/ResourceFilterTest.java b/clients/src/test/java/org/apache/kafka/common/resource/ResourceFilterTest.java index 9b2d6d4e696ff..4399744d91d64 100644 --- a/clients/src/test/java/org/apache/kafka/common/resource/ResourceFilterTest.java +++ b/clients/src/test/java/org/apache/kafka/common/resource/ResourceFilterTest.java @@ -19,145 +19,64 @@ import org.junit.Test; -import static org.apache.kafka.common.resource.ResourceNameType.LITERAL; -import static org.apache.kafka.common.resource.ResourceNameType.PREFIXED; import static org.apache.kafka.common.resource.ResourceType.ANY; import static org.apache.kafka.common.resource.ResourceType.GROUP; import static org.apache.kafka.common.resource.ResourceType.TOPIC; -import static org.apache.kafka.common.resource.ResourceType.UNKNOWN; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; public class ResourceFilterTest { - @Test(expected = IllegalArgumentException.class) - public void shouldThrowIfResourceTypeIsAny() { - new ResourceFilter(ANY, null, ResourceNameType.ANY) - .matches(new Resource(ANY, "Name", PREFIXED)); - } - - @Test(expected = IllegalArgumentException.class) - public void shouldThrowIfResourceTypeIsUnknown() { - new ResourceFilter(ANY, null, ResourceNameType.ANY) - .matches(new Resource(UNKNOWN, "Name", LITERAL)); - } - - @Test(expected = IllegalArgumentException.class) - public void shouldThrowIfResourceNameTypeIsAny() { - new ResourceFilter(ANY, null, ResourceNameType.ANY) - .matches(new Resource(GROUP, "Name", ResourceNameType.ANY)); - } - - @Test(expected = IllegalArgumentException.class) - public void shouldThrowIfAclResourceNameTypeIsUnknown() { - new ResourceFilter(ANY, null, ResourceNameType.ANY) - .matches(new Resource(GROUP, "Name", ResourceNameType.UNKNOWN)); - } - @Test public void shouldNotMatchIfDifferentResourceType() { - assertFalse(new ResourceFilter(TOPIC, "Name", LITERAL) - .matches(new Resource(GROUP, "Name", LITERAL))); + assertFalse(new ResourceFilter(TOPIC, "Name") + .matches(new Resource(GROUP, "Name"))); } @Test public void shouldNotMatchIfDifferentName() { - assertFalse(new ResourceFilter(TOPIC, "Different", PREFIXED) - .matches(new Resource(TOPIC, "Name", PREFIXED))); + assertFalse(new ResourceFilter(TOPIC, "Different") + .matches(new Resource(TOPIC, "Name"))); } @Test public void shouldNotMatchIfDifferentNameCase() { - assertFalse(new ResourceFilter(TOPIC, "NAME", LITERAL) - .matches(new Resource(TOPIC, "Name", LITERAL))); - } - - @Test - public void shouldNotMatchIfDifferentNameType() { - assertFalse(new ResourceFilter(TOPIC, "Name", LITERAL) - .matches(new Resource(TOPIC, "Name", PREFIXED))); + assertFalse(new ResourceFilter(TOPIC, "NAME") + .matches(new Resource(TOPIC, "Name"))); } @Test public void shouldMatchWhereResourceTypeIsAny() { - assertTrue(new ResourceFilter(ANY, "Name", PREFIXED) - .matches(new Resource(TOPIC, "Name", PREFIXED))); + assertTrue(new ResourceFilter(ANY, "Name") + .matches(new Resource(TOPIC, "Name"))); } @Test public void shouldMatchWhereResourceNameIsAny() { - assertTrue(new ResourceFilter(TOPIC, null, PREFIXED) - .matches(new Resource(TOPIC, "Name", PREFIXED))); - } - - @Test - public void shouldMatchWhereResourceNameTypeIsAny() { - assertTrue(new ResourceFilter(TOPIC, null, ResourceNameType.ANY) - .matches(new Resource(TOPIC, "Name", PREFIXED))); + assertTrue(new ResourceFilter(TOPIC, null) + .matches(new Resource(TOPIC, "Name"))); } @Test - public void shouldMatchLiteralIfExactMatch() { - assertTrue(new ResourceFilter(TOPIC, "Name", LITERAL) - .matches(new Resource(TOPIC, "Name", LITERAL))); + public void shouldMatchIfExactMatch() { + assertTrue(new ResourceFilter(TOPIC, "Name") + .matches(new Resource(TOPIC, "Name"))); } @Test - public void shouldMatchLiteralIfNameMatchesAndFilterIsOnAnyNameType() { - assertTrue(new ResourceFilter(TOPIC, "Name", ResourceNameType.ANY) - .matches(new Resource(TOPIC, "Name", LITERAL))); + public void shouldMatchWildcardIfExactMatch() { + assertTrue(new ResourceFilter(TOPIC, "*") + .matches(new Resource(TOPIC, "*"))); } @Test - public void shouldNotMatchLiteralIfNamePrefixed() { - assertFalse(new ResourceFilter(TOPIC, "Name-something", ResourceNameType.ANY) - .matches(new Resource(TOPIC, "Name", LITERAL))); - } - - @Test - public void shouldMatchLiteralWildcardIfExactMatch() { - assertTrue(new ResourceFilter(TOPIC, "*", LITERAL) - .matches(new Resource(TOPIC, "*", LITERAL))); - } - - @Test - public void shouldNotMatchLiteralWildcardAgainstOtherName() { - assertFalse(new ResourceFilter(TOPIC, "Name", LITERAL) - .matches(new Resource(TOPIC, "*", LITERAL))); + public void shouldNotMatchWildcardAgainstOtherName() { + assertFalse(new ResourceFilter(TOPIC, "Name") + .matches(new Resource(TOPIC, "*"))); } @Test public void shouldNotMatchLiteralWildcardTheWayAround() { - assertFalse(new ResourceFilter(TOPIC, "*", LITERAL) - .matches(new Resource(TOPIC, "Name", LITERAL))); - } - - @Test - public void shouldMatchLiteralWildcardIfFilterHasNameTypeOfAny() { - assertTrue(new ResourceFilter(TOPIC, "Name", ResourceNameType.ANY) - .matches(new Resource(TOPIC, "*", LITERAL))); - } - - @Test - public void shouldMatchPrefixedIfExactMatch() { - assertTrue(new ResourceFilter(TOPIC, "Name", PREFIXED) - .matches(new Resource(TOPIC, "Name", PREFIXED))); - } - - @Test - public void shouldNotMatchIfBothPrefixedAndFilterIsPrefixOfResource() { - assertFalse(new ResourceFilter(TOPIC, "Name", PREFIXED) - .matches(new Resource(TOPIC, "Name-something", PREFIXED))); - } - - @Test - public void shouldNotMatchIfBothPrefixedAndResourceIsPrefixOfFilter() { - assertFalse(new ResourceFilter(TOPIC, "Name-something", PREFIXED) - .matches(new Resource(TOPIC, "Name", PREFIXED))); - } - - @Test - public void shouldMatchPrefixedIfNamePrefixedAnyFilterTypeIsAny() { - assertTrue(new ResourceFilter(TOPIC, "Name-something", ResourceNameType.ANY) - .matches(new Resource(TOPIC, "Name", PREFIXED))); + assertFalse(new ResourceFilter(TOPIC, "*") + .matches(new Resource(TOPIC, "Name"))); } } \ No newline at end of file diff --git a/core/src/main/scala/kafka/admin/AclCommand.scala b/core/src/main/scala/kafka/admin/AclCommand.scala index 25b630d4159b2..d55e8861df04f 100644 --- a/core/src/main/scala/kafka/admin/AclCommand.scala +++ b/core/src/main/scala/kafka/admin/AclCommand.scala @@ -25,13 +25,13 @@ import kafka.utils._ import org.apache.kafka.common.security.JaasUtils import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.utils.Utils -import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType => JResourceNameType, ResourceType => JResourceType, Resource => JResource} +import org.apache.kafka.common.resource.{ResourcePatternFilter, ResourceNameType => JResourceNameType, ResourceType => JResourceType, Resource => JResource} import scala.collection.JavaConverters._ object AclCommand extends Logging { - val ClusterResourceFilter = new ResourceFilter(JResourceType.CLUSTER, JResource.CLUSTER_NAME, JResourceNameType.LITERAL) + val ClusterResourceFilter = new ResourcePatternFilter(JResourceType.CLUSTER, JResource.CLUSTER_NAME, JResourceNameType.LITERAL) private val Newline = scala.util.Properties.lineSeparator @@ -126,7 +126,7 @@ object AclCommand extends Logging { } } - private def removeAcls(authorizer: Authorizer, acls: Set[Acl], filter: ResourceFilter) { + private def removeAcls(authorizer: Authorizer, acls: Set[Acl], filter: ResourcePatternFilter) { getAcls(authorizer, filter) .keys .foreach(resource => @@ -148,12 +148,12 @@ object AclCommand extends Logging { } } - private def getAcls(authorizer: Authorizer, filter: ResourceFilter): Map[Resource, Set[Acl]] = + private def getAcls(authorizer: Authorizer, filter: ResourcePatternFilter): Map[Resource, Set[Acl]] = authorizer.getAcls() - .filter { case (resource, acl) => filter.matches(resource.toJava) } + .filter { case (resource, acl) => filter.matches(resource.toPattern) } - private def getResourceFilterToAcls(opts: AclCommandOptions): Map[ResourceFilter, Set[Acl]] = { - var resourceToAcls = Map.empty[ResourceFilter, Set[Acl]] + private def getResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[Acl]] = { + var resourceToAcls = Map.empty[ResourcePatternFilter, Set[Acl]] //if none of the --producer or --consumer options are specified , just construct ACLs from CLI options. if (!opts.options.has(opts.producerOpt) && !opts.options.has(opts.consumerOpt)) { @@ -172,11 +172,11 @@ object AclCommand extends Logging { resourceToAcls } - private def getProducerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourceFilter, Set[Acl]] = { + private def getProducerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[Acl]] = { val filters = getResourceFilter(opts) - val topics: Set[ResourceFilter] = filters.filter(_.resourceType == JResourceType.TOPIC) - val transactionalIds: Set[ResourceFilter] = filters.filter(_.resourceType == JResourceType.TRANSACTIONAL_ID) + val topics: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TOPIC) + val transactionalIds: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TRANSACTIONAL_ID) val enableIdempotence = opts.options.has(opts.idempotentOpt) val topicAcls = getAcl(opts, Set(Write, Describe, Create)) @@ -185,27 +185,27 @@ object AclCommand extends Logging { //Write, Describe, Create permission on topics, Write, Describe on transactionalIds topics.map(_ -> topicAcls).toMap ++ transactionalIds.map(_ -> transactionalIdAcls).toMap ++ - (if (enableIdempotence) + (if (enableIdempotence) Map(ClusterResourceFilter -> getAcl(opts, Set(IdempotentWrite))) else Map.empty) } - private def getConsumerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourceFilter, Set[Acl]] = { + private def getConsumerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[Acl]] = { val filters = getResourceFilter(opts) - val topics: Set[ResourceFilter] = filters.filter(_.resourceType == JResourceType.TOPIC) - val groups: Set[ResourceFilter] = filters.filter(_.resourceType == JResourceType.GROUP) + val topics: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TOPIC) + val groups: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.GROUP) //Read, Describe on topic, Read on consumerGroup val acls = getAcl(opts, Set(Read, Describe)) - topics.map(_ -> acls).toMap[ResourceFilter, Set[Acl]] ++ - groups.map(_ -> getAcl(opts, Set(Read))).toMap[ResourceFilter, Set[Acl]] + topics.map(_ -> acls).toMap[ResourcePatternFilter, Set[Acl]] ++ + groups.map(_ -> getAcl(opts, Set(Read))).toMap[ResourcePatternFilter, Set[Acl]] } - private def getCliResourceFilterToAcls(opts: AclCommandOptions): Map[ResourceFilter, Set[Acl]] = { + private def getCliResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[Acl]] = { val acls = getAcl(opts) val filters = getResourceFilter(opts) filters.map(_ -> acls).toMap @@ -261,25 +261,25 @@ object AclCommand extends Logging { Set.empty[KafkaPrincipal] } - private def getResourceFilter(opts: AclCommandOptions, dieIfNoResourceFound: Boolean = true): Set[ResourceFilter] = { + private def getResourceFilter(opts: AclCommandOptions, dieIfNoResourceFound: Boolean = true): Set[ResourcePatternFilter] = { val resourceNameType: JResourceNameType = opts.options.valueOf(opts.resourceNameType) - var resourceFilters = Set.empty[ResourceFilter] + var resourceFilters = Set.empty[ResourcePatternFilter] if (opts.options.has(opts.topicOpt)) - opts.options.valuesOf(opts.topicOpt).asScala.foreach(topic => resourceFilters += new ResourceFilter(JResourceType.TOPIC, topic.trim, resourceNameType)) + opts.options.valuesOf(opts.topicOpt).asScala.foreach(topic => resourceFilters += new ResourcePatternFilter(JResourceType.TOPIC, topic.trim, resourceNameType)) if (resourceNameType == JResourceNameType.LITERAL && (opts.options.has(opts.clusterOpt) || opts.options.has(opts.idempotentOpt))) resourceFilters += ClusterResourceFilter if (opts.options.has(opts.groupOpt)) - opts.options.valuesOf(opts.groupOpt).asScala.foreach(group => resourceFilters += new ResourceFilter(JResourceType.GROUP, group.trim, resourceNameType)) + opts.options.valuesOf(opts.groupOpt).asScala.foreach(group => resourceFilters += new ResourcePatternFilter(JResourceType.GROUP, group.trim, resourceNameType)) if (opts.options.has(opts.transactionalIdOpt)) opts.options.valuesOf(opts.transactionalIdOpt).asScala.foreach(transactionalId => - resourceFilters += new ResourceFilter(JResourceType.TRANSACTIONAL_ID, transactionalId, resourceNameType)) + resourceFilters += new ResourcePatternFilter(JResourceType.TRANSACTIONAL_ID, transactionalId, resourceNameType)) if (opts.options.has(opts.delegationTokenOpt)) - opts.options.valuesOf(opts.delegationTokenOpt).asScala.foreach(token => resourceFilters += new ResourceFilter(JResourceType.DELEGATION_TOKEN, token.trim, resourceNameType)) + opts.options.valuesOf(opts.delegationTokenOpt).asScala.foreach(token => resourceFilters += new ResourcePatternFilter(JResourceType.DELEGATION_TOKEN, token.trim, resourceNameType)) if (resourceFilters.isEmpty && dieIfNoResourceFound) CommandLineUtils.printUsageAndDie(opts.parser, "You must provide at least one resource: --topic or --cluster or --group or --delegation-token ") @@ -294,7 +294,7 @@ object AclCommand extends Logging { Console.readLine().equalsIgnoreCase("y") } - private def validateOperation(opts: AclCommandOptions, resourceToAcls: Map[ResourceFilter, Set[Acl]]): Unit = { + private def validateOperation(opts: AclCommandOptions, resourceToAcls: Map[ResourcePatternFilter, Set[Acl]]): Unit = { for ((resource, acls) <- resourceToAcls) { val validOps = ResourceTypeToValidOperations(resource.resourceType) if ((acls.map(_.operation) -- validOps).nonEmpty) diff --git a/core/src/main/scala/kafka/security/SecurityUtils.scala b/core/src/main/scala/kafka/security/SecurityUtils.scala index 8442ba0a36fe8..7489a3e9b6483 100644 --- a/core/src/main/scala/kafka/security/SecurityUtils.scala +++ b/core/src/main/scala/kafka/security/SecurityUtils.scala @@ -21,7 +21,7 @@ import kafka.security.auth.{Acl, Operation, PermissionType, Resource, ResourceNa import org.apache.kafka.common.acl.{AccessControlEntry, AclBinding, AclBindingFilter} import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.requests.ApiError -import org.apache.kafka.common.resource.{Resource => AdminResource} +import org.apache.kafka.common.resource.ResourcePattern import org.apache.kafka.common.security.auth.KafkaPrincipal import scala.util.{Failure, Success, Try} @@ -31,12 +31,12 @@ object SecurityUtils { def convertToResourceAndAcl(filter: AclBindingFilter): Either[ApiError, (Resource, Acl)] = { (for { - resourceType <- Try(ResourceType.fromJava(filter.resourceFilter.resourceType)) - resourceNameType <- Try(ResourceNameType.fromJava(filter.resourceFilter.nameType)) + resourceType <- Try(ResourceType.fromJava(filter.patternFilter.resourceType)) + resourceNameType <- Try(ResourceNameType.fromJava(filter.patternFilter.nameType)) principal <- Try(KafkaPrincipal.fromString(filter.entryFilter.principal)) operation <- Try(Operation.fromJava(filter.entryFilter.operation)) permissionType <- Try(PermissionType.fromJava(filter.entryFilter.permissionType)) - resource = Resource(resourceType, filter.resourceFilter.name, resourceNameType) + resource = Resource(resourceType, filter.patternFilter.name, resourceNameType) acl = Acl(principal, permissionType, filter.entryFilter.host, operation) } yield (resource, acl)) match { case Failure(throwable) => Left(new ApiError(Errors.INVALID_REQUEST, throwable.getMessage)) @@ -45,10 +45,10 @@ object SecurityUtils { } def convertToAclBinding(resource: Resource, acl: Acl): AclBinding = { - val adminResource = new AdminResource(resource.resourceType.toJava, resource.name, resource.resourceNameType.toJava) + val resourcePattern = new ResourcePattern(resource.resourceType.toJava, resource.name, resource.nameType.toJava) val entry = new AccessControlEntry(acl.principal.toString, acl.host.toString, acl.operation.toJava, acl.permissionType.toJava) - new AclBinding(adminResource, entry) + new AclBinding(resourcePattern, entry) } } diff --git a/core/src/main/scala/kafka/security/auth/Acl.scala b/core/src/main/scala/kafka/security/auth/Acl.scala index 7fa1638cf0d59..bdd8584036061 100644 --- a/core/src/main/scala/kafka/security/auth/Acl.scala +++ b/core/src/main/scala/kafka/security/auth/Acl.scala @@ -18,6 +18,7 @@ package kafka.security.auth import kafka.utils.Json +import org.apache.kafka.common.resource.ResourcePattern import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.utils.SecurityUtils @@ -26,7 +27,7 @@ import scala.collection.JavaConverters._ object Acl { val WildCardPrincipal: KafkaPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "*") val WildCardHost: String = "*" - val WildCardResource: String = org.apache.kafka.common.resource.Resource.WILDCARD_RESOURCE + val WildCardResource: String = ResourcePattern.WILDCARD_RESOURCE val AllowAllAcl = new Acl(WildCardPrincipal, Allow, WildCardHost, All) val PrincipalKey = "principal" val PermissionTypeKey = "permissionType" diff --git a/core/src/main/scala/kafka/security/auth/Resource.scala b/core/src/main/scala/kafka/security/auth/Resource.scala index fa63fccf10f8a..c9b5727c46ced 100644 --- a/core/src/main/scala/kafka/security/auth/Resource.scala +++ b/core/src/main/scala/kafka/security/auth/Resource.scala @@ -16,14 +16,29 @@ */ package kafka.security.auth -import java.util.Objects -import org.apache.kafka.common.resource.{Resource => JResource} +import org.apache.kafka.common.resource.ResourcePattern object Resource { + val Separator = ":" val ClusterResourceName = "kafka-cluster" val ClusterResource = new Resource(Cluster, Resource.ClusterResourceName, Literal) val ProducerIdResourceName = "producer-id" val WildCardResource = "*" + + def fromString(str: String): Resource = { + ResourceNameType.values.find(nameType => str.startsWith(nameType.name)) match { + case Some(nameType) => + str.split(Separator, 3) match { + case Array(_, resourceType, name, _*) => new Resource(ResourceType.fromString(resourceType), name, nameType) + case _ => throw new IllegalArgumentException("expected a string in format ResourceType:ResourceName but got " + str) + } + case _ => + str.split(Separator, 2) match { + case Array(resourceType, name, _*) => new Resource(ResourceType.fromString(resourceType), name, Literal) + case _ => throw new IllegalArgumentException("expected a string in format ResourceType:ResourceName but got " + str) + } + } + } } /** @@ -31,13 +46,9 @@ object Resource { * @param resourceType non-null type of resource. * @param name non-null name of the resource, for topic this will be topic name , for group it will be group name. For cluster type * it will be a constant string kafka-cluster. - * @param resourceNameType non-null type of resource name: literal, prefixed, etc. + * @param nameType non-null type of resource name: literal, prefixed, etc. */ -case class Resource(resourceType: ResourceType, name: String, resourceNameType: ResourceNameType) { - - Objects.requireNonNull(resourceType, "resourceType") - Objects.requireNonNull(name, "name") - Objects.requireNonNull(resourceNameType, "resourceNameType") +case class Resource(resourceType: ResourceType, name: String, nameType: ResourceNameType) { /** * Create an instance of this class with the provided parameters. @@ -52,8 +63,12 @@ case class Resource(resourceType: ResourceType, name: String, resourceNameType: this(resourceType, name, Literal) } - def toJava: JResource = { - new JResource(resourceType.toJava, name, resourceNameType.toJava) + def toPattern: ResourcePattern = { + new ResourcePattern(resourceType.toJava, name, nameType.toJava) + } + + override def toString: String = { + nameType + Resource.Separator + resourceType.name + Resource.Separator + name } } diff --git a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala index c828970cc0d79..0cb2fae2fa26e 100644 --- a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala +++ b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala @@ -27,7 +27,6 @@ import kafka.server.KafkaConfig import kafka.utils.CoreUtils.{inReadLock, inWriteLock} import kafka.utils._ import kafka.zk.{AclChangeNotificationSequenceZNode, KafkaZkClient, ZkAclStore} -import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType => JResourceNameType} import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.utils.{SecurityUtils, Time} @@ -102,8 +101,8 @@ class SimpleAclAuthorizer extends Authorizer with Logging { } override def authorize(session: Session, operation: Operation, resource: Resource): Boolean = { - if (resource.resourceNameType != Literal) { - throw new IllegalArgumentException("Only literal resources are supported. Got: " + resource.resourceNameType) + if (resource.nameType != Literal) { + throw new IllegalArgumentException("Only literal resources are supported. Got: " + resource.nameType) } val principal = session.principal @@ -203,8 +202,6 @@ class SimpleAclAuthorizer extends Authorizer with Logging { } def getMatchingAcls(resourceType: ResourceType, resourceName: String): Set[Acl] = { - val filter = new ResourceFilter(resourceType.toJava, resourceName, JResourceNameType.ANY) - inReadLock(lock) { val wildcard = aclCache.get(Resource(resourceType, Acl.WildCardResource, Literal)) .map(_.acls) @@ -368,7 +365,7 @@ class SimpleAclAuthorizer extends Authorizer with Logging { if (rt != 0) rt else { - val rnt = a.resourceNameType compare b.resourceNameType + val rnt = a.nameType compare b.nameType if (rnt != 0) rnt else diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala index f4b8689f222f9..6d9e3d115b836 100644 --- a/core/src/main/scala/kafka/server/KafkaApis.scala +++ b/core/src/main/scala/kafka/server/KafkaApis.scala @@ -52,7 +52,6 @@ import org.apache.kafka.common.requests.DeleteAclsResponse.{AclDeletionResult, A import org.apache.kafka.common.requests.DescribeLogDirsResponse.LogDirInfo import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse import org.apache.kafka.common.requests.{Resource => RResource, ResourceType => RResourceType, _} -import org.apache.kafka.common.resource.{Resource => AdminResource} import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol} import org.apache.kafka.common.security.token.delegation.{DelegationToken, TokenInformation} import org.apache.kafka.common.utils.{Time, Utils} @@ -63,6 +62,7 @@ import scala.collection._ import scala.collection.mutable.ArrayBuffer import scala.util.{Failure, Success, Try} import org.apache.kafka.common.requests.CreateTopicsRequest.TopicDetails +import org.apache.kafka.common.resource.ResourcePattern /** * Logic to handle the various Kafka requests @@ -1920,7 +1920,7 @@ class KafkaApis(val requestChannel: RequestChannel, val filter = describeAclsRequest.filter() val returnedAcls = auth.getAcls.toSeq.flatMap { case (resource, acls) => acls.flatMap { acl => - val fixture = new AclBinding(new AdminResource(resource.resourceType.toJava, resource.name, resource.resourceNameType.toJava), + val fixture = new AclBinding(new ResourcePattern(resource.resourceType.toJava, resource.name, resource.nameType.toJava), new AccessControlEntry(acl.principal.toString, acl.host.toString, acl.operation.toJava, acl.permissionType.toJava)) if (filter.matches(fixture)) Some(fixture) else None @@ -1994,7 +1994,7 @@ class KafkaApis(val requestChannel: RequestChannel, val filtersWithIndex = filters.zipWithIndex for ((resource, acls) <- aclMap; acl <- acls) { val binding = new AclBinding( - new AdminResource(resource.resourceType.toJava, resource.name, resource.resourceNameType.toJava), + new ResourcePattern(resource.resourceType.toJava, resource.name, resource.nameType.toJava), new AccessControlEntry(acl.principal.toString, acl.host.toString, acl.operation.toJava, acl.permissionType.toJava)) diff --git a/core/src/main/scala/kafka/zk/KafkaZkClient.scala b/core/src/main/scala/kafka/zk/KafkaZkClient.scala index 90f71a1e1d201..20e4b83c842d1 100644 --- a/core/src/main/scala/kafka/zk/KafkaZkClient.scala +++ b/core/src/main/scala/kafka/zk/KafkaZkClient.scala @@ -1008,7 +1008,7 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean * @param resource resource name */ def createAclChangeNotification(resource: Resource): Unit = { - val store = ZkAclStore(resource.resourceNameType) + val store = ZkAclStore(resource.nameType) val path = store.changeSequenceZNode.createPath val createRequest = CreateRequest(path, AclChangeNotificationSequenceZNode.encode(resource), acls(path), CreateMode.PERSISTENT_SEQUENTIAL) val createResponse = retryRequestUntilConnected(createRequest) diff --git a/core/src/main/scala/kafka/zk/ZkData.scala b/core/src/main/scala/kafka/zk/ZkData.scala index 0524b4599c11d..61210354dc414 100644 --- a/core/src/main/scala/kafka/zk/ZkData.scala +++ b/core/src/main/scala/kafka/zk/ZkData.scala @@ -487,7 +487,7 @@ object ZkAclStore { } object ResourceZNode { - def path(resource: Resource): String = ZkAclStore(resource.resourceNameType).path(resource.resourceType, resource.name) + def path(resource: Resource): String = ZkAclStore(resource.nameType).path(resource.resourceType, resource.name) def encode(acls: Set[Acl]): Array[Byte] = Json.encodeAsBytes(Acl.toJsonCompatibleMap(acls).asJava) def decode(bytes: Array[Byte], stat: Stat): VersionedAcls = VersionedAcls(Acl.fromBytes(bytes), stat.getVersion) diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala index 331a4491e0d63..986fa4a366ac5 100644 --- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala @@ -40,7 +40,7 @@ import org.apache.kafka.common.config.ConfigResource import org.apache.kafka.common.errors._ import org.junit.{After, Before, Rule, Test} import org.apache.kafka.common.requests.{DeleteRecordsRequest, MetadataResponse} -import org.apache.kafka.common.resource.{Resource, ResourceNameType, ResourceType} +import org.apache.kafka.common.resource.{ResourceNameType, ResourcePattern, ResourceType} import org.junit.rules.Timeout import org.junit.Assert._ @@ -933,7 +933,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging { checkInvalidAlterConfigs(zkClient, servers, client) } - val ACL1 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), + val ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW)) /** diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala index b40dab7809c45..a3b3233565214 100644 --- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala @@ -40,9 +40,9 @@ import org.apache.kafka.common.record.{CompressionType, MemoryRecords, Records, import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation import org.apache.kafka.common.requests.CreateTopicsRequest.TopicDetails import org.apache.kafka.common.requests.{Resource => RResource, ResourceType => RResourceType, _} -import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType, Resource => AdminResource, ResourceType => AdminResourceType} +import org.apache.kafka.common.resource.{ResourcePattern, ResourcePatternFilter, ResourceType => AdminResourceType} import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol} -import org.apache.kafka.common.{KafkaException, Node, TopicPartition, requests} +import org.apache.kafka.common.{KafkaException, Node, TopicPartition, acl, requests, resource} import org.junit.Assert._ import org.junit.{After, Assert, Before, Test} @@ -378,12 +378,12 @@ class AuthorizerIntegrationTest extends BaseRequestTest { private def createAclsRequest = new CreateAclsRequest.Builder( Collections.singletonList(new AclCreation(new AclBinding( - new AdminResource(AdminResourceType.TOPIC, "mytopic"), + new ResourcePattern(AdminResourceType.TOPIC, "mytopic", resource.ResourceNameType.LITERAL), new AccessControlEntry(userPrincipal.toString, "*", AclOperation.WRITE, AclPermissionType.DENY))))).build() private def deleteAclsRequest = new DeleteAclsRequest.Builder( Collections.singletonList(new AclBindingFilter( - new ResourceFilter(AdminResourceType.TOPIC, null, ResourceNameType.LITERAL), + new ResourcePatternFilter(AdminResourceType.TOPIC, null, resource.ResourceNameType.LITERAL), new AccessControlEntryFilter(userPrincipal.toString, "*", AclOperation.ANY, AclPermissionType.DENY)))).build() private def alterReplicaLogDirsRequest = new AlterReplicaLogDirsRequest.Builder(Collections.singletonMap(tp, logDir)).build() diff --git a/core/src/test/scala/integration/kafka/api/SaslSslAdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/SaslSslAdminClientIntegrationTest.scala index b3572c0665b62..46bf722b3f0b2 100644 --- a/core/src/test/scala/integration/kafka/api/SaslSslAdminClientIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/SaslSslAdminClientIntegrationTest.scala @@ -15,13 +15,13 @@ package kafka.api import java.io.File import java.util -import kafka.security.auth.{All, Allow, Alter, AlterConfigs, Authorizer, ClusterAction, Create, Delete, Deny, Describe, Group, Literal, Operation, PermissionType, SimpleAclAuthorizer, Topic, Prefixed, Acl => AuthAcl, Resource => AuthResource} +import kafka.security.auth.{All, Allow, Alter, AlterConfigs, Authorizer, ClusterAction, Create, Delete, Deny, Describe, Group, Operation, PermissionType, SimpleAclAuthorizer, Topic, Acl => AuthAcl, Resource => AuthResource} import kafka.server.KafkaConfig import kafka.utils.{CoreUtils, JaasTestUtils, TestUtils} import org.apache.kafka.clients.admin.{AdminClient, CreateAclsOptions, DeleteAclsOptions} import org.apache.kafka.common.acl._ import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidRequestException} -import org.apache.kafka.common.resource.{Resource, ResourceFilter, ResourceNameType, ResourceType} +import org.apache.kafka.common.resource.{ResourceNameType, ResourcePattern, ResourcePatternFilter, ResourceType} import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol} import org.junit.Assert.assertEquals import org.junit.{After, Assert, Before, Test} @@ -89,19 +89,19 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with closeSasl() } - val anyAcl = new AclBinding(new Resource(ResourceType.TOPIC, "*", ResourceNameType.LITERAL), + val anyAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "*", ResourceNameType.LITERAL), new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW)) - val acl2 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), + val acl2 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.ALLOW)) - val acl3 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), + val acl3 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW)) - val fooAcl = new AclBinding(new Resource(ResourceType.TOPIC, "foobar", ResourceNameType.LITERAL), + val fooAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "foobar", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW)) - val prefixAcl = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.PREFIXED), + val prefixAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.PREFIXED), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW)) - val transactionalIdAcl = new AclBinding(new Resource(ResourceType.TRANSACTIONAL_ID, "transactional_id", ResourceNameType.LITERAL), + val transactionalIdAcl = new AclBinding(new ResourcePattern(ResourceType.TRANSACTIONAL_ID, "transactional_id", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.ALLOW)) - val groupAcl = new AclBinding(new Resource(ResourceType.GROUP, "*", ResourceNameType.LITERAL), + val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", ResourceNameType.LITERAL), new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW)) @Test @@ -111,7 +111,7 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with val results = client.createAcls(List(acl2, acl3).asJava) assertEquals(Set(acl2, acl3), results.values.keySet().asScala) results.values.values().asScala.foreach(value => value.get) - val aclUnknown = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), + val aclUnknown = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.UNKNOWN, AclPermissionType.ALLOW)) val results2 = client.createAcls(List(aclUnknown).asJava) assertEquals(Set(aclUnknown), results2.values.keySet().asScala) @@ -132,9 +132,9 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with waitForDescribeAcls(client, acl2.toFilter, Set(acl2)) waitForDescribeAcls(client, transactionalIdAcl.toFilter, Set(transactionalIdAcl)) - val filterA = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) - val filterB = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) - val filterC = new AclBindingFilter(new ResourceFilter(ResourceType.TRANSACTIONAL_ID, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val filterA = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val filterB = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val filterC = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TRANSACTIONAL_ID, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) waitForDescribeAcls(client, filterA, Set(groupAcl)) waitForDescribeAcls(client, filterC, Set(transactionalIdAcl)) @@ -154,13 +154,13 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with client = AdminClient.create(createConfig()) ensureAcls(Set(anyAcl, acl2, fooAcl, prefixAcl)) - val allTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY) - val allLiteralTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) - val allPrefixedTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY) - val literalMyTopic2Acls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) - val prefixedMyTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY) - val allMyTopic2Acls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.ANY), AccessControlEntryFilter.ANY) - val allFooTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "foobar", ResourceNameType.ANY), AccessControlEntryFilter.ANY) + val allTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY) + val allLiteralTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val allPrefixedTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY) + val literalMyTopic2Acls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val prefixedMyTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY) + val allMyTopic2Acls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.ANY), AccessControlEntryFilter.ANY) + val allFooTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "foobar", ResourceNameType.ANY), AccessControlEntryFilter.ANY) assertEquals(Set(anyAcl), getAcls(anyAcl.toFilter)) assertEquals(Set(prefixAcl), getAcls(prefixAcl.toFilter)) @@ -181,9 +181,9 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with client = AdminClient.create(createConfig()) ensureAcls(Set(anyAcl, acl2, fooAcl, prefixAcl)) - val allTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY) - val allLiteralTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) - val allPrefixedTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY) + val allTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY) + val allLiteralTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val allPrefixedTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY) // Delete only ACLs on literal 'mytopic2' topic var deleted = client.deleteAcls(List(acl2.toFilter).asJava).all().get().asScala.toSet @@ -231,11 +231,11 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with client = AdminClient.create(createConfig()) ensureAcls(Set(anyAcl, acl2, fooAcl, prefixAcl)) // <-- prefixed exists, but should never be returned. - val allTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY) - val legacyAllTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) - val legacyMyTopic2Acls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) - val legacyAnyTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "*", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) - val legacyFooTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "foobar", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val allTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY) + val legacyAllTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val legacyMyTopic2Acls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val legacyAnyTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "*", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) + val legacyFooTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "foobar", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY) assertEquals(Set(anyAcl, acl2, fooAcl), getAcls(legacyAllTopicAcls)) assertEquals(Set(acl2), getAcls(legacyMyTopic2Acls)) @@ -266,9 +266,9 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with @Test def testAttemptToCreateInvalidAcls(): Unit = { client = AdminClient.create(createConfig()) - val clusterAcl = new AclBinding(new Resource(ResourceType.CLUSTER, "foobar", ResourceNameType.LITERAL), + val clusterAcl = new AclBinding(new ResourcePattern(ResourceType.CLUSTER, "foobar", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW)) - val emptyResourceNameAcl = new AclBinding(new Resource(ResourceType.TOPIC, "", ResourceNameType.LITERAL), + val emptyResourceNameAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW)) val results = client.createAcls(List(clusterAcl, emptyResourceNameAcl).asJava, new CreateAclsOptions()) assertEquals(Set(clusterAcl, emptyResourceNameAcl), results.values.keySet().asScala) @@ -336,7 +336,7 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with private def testAclGet(expectAuth: Boolean): Unit = { TestUtils.waitUntilTrue(() => { - val userAcl = new AclBinding(new Resource(ResourceType.TOPIC, "*", ResourceNameType.LITERAL), + val userAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "*", ResourceNameType.LITERAL), new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW)) val results = client.describeAcls(userAcl.toFilter) if (expectAuth) { diff --git a/core/src/test/scala/kafka/security/auth/ResourceTest.scala b/core/src/test/scala/kafka/security/auth/ResourceTest.scala new file mode 100644 index 0000000000000..66049b416295b --- /dev/null +++ b/core/src/test/scala/kafka/security/auth/ResourceTest.scala @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package kafka.security.auth + +import kafka.common.KafkaException +import org.junit.Test +import org.junit.Assert._ + +class ResourceTest { + @Test(expected = classOf[KafkaException]) + def shouldThrowTwoPartStringWithUnknownResourceType(): Unit = { + Resource.fromString("Unknown:fred") + } + + @Test + def shouldParseOldTwoPartString(): Unit = { + assertEquals(Resource(Group, "fred", Literal), Resource.fromString("Group:fred")) + assertEquals(Resource(Topic, "t", Literal), Resource.fromString("Topic:t")) + } + + @Test + def shouldParseOldTwoPartWithEmbeddedSeparators(): Unit = { + assertEquals(Resource(Group, ":This:is:a:weird:group:name:", Literal), Resource.fromString("Group::This:is:a:weird:group:name:")) + } + + @Test + def shouldParseThreePartString(): Unit = { + assertEquals(Resource(Group, "fred", Prefixed), Resource.fromString("Prefixed:Group:fred")) + assertEquals(Resource(Topic, "t", Literal), Resource.fromString("Literal:Topic:t")) + } + + @Test + def shouldParseThreePartWithEmbeddedSeparators(): Unit = { + assertEquals(Resource(Group, ":This:is:a:weird:group:name:", Prefixed), Resource.fromString("Prefixed:Group::This:is:a:weird:group:name:")) + assertEquals(Resource(Group, ":This:is:a:weird:group:name:", Literal), Resource.fromString("Literal:Group::This:is:a:weird:group:name:")) + } + + @Test + def shouldRoundTripViaString(): Unit = { + val expected = Resource(Group, "fred", Prefixed) + + val actual = Resource.fromString(expected.toString) + + assertEquals(expected, actual) + } +} \ No newline at end of file diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala index 59f543b522a1f..f56b3b4878934 100644 --- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala +++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala @@ -24,12 +24,13 @@ import kafka.security.auth._ import kafka.utils.TestUtils import org.apache.kafka.clients.admin.NewPartitions import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter, AclBinding, AclBindingFilter, AclOperation, AclPermissionType} -import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType, Resource => AdminResource, ResourceType => AdminResourceType} +import org.apache.kafka.common.resource.{ResourcePattern, ResourcePatternFilter, ResourceType => AdminResourceType} import org.apache.kafka.common.{Node, TopicPartition} import org.apache.kafka.common.metrics.{KafkaMetric, Quota, Sensor} import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.protocol.ApiKeys import org.apache.kafka.common.protocol.types.Struct +import org.apache.kafka.common.resource.ResourceNameType import org.apache.kafka.common.record._ import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation import org.apache.kafka.common.requests.{Resource => RResource, ResourceType => RResourceType, _} @@ -312,12 +313,12 @@ class RequestQuotaTest extends BaseRequestTest { case ApiKeys.CREATE_ACLS => new CreateAclsRequest.Builder(Collections.singletonList(new AclCreation(new AclBinding( - new AdminResource(AdminResourceType.TOPIC, "mytopic"), + new ResourcePattern(AdminResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL), new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.DENY))))) case ApiKeys.DELETE_ACLS => new DeleteAclsRequest.Builder(Collections.singletonList(new AclBindingFilter( - new ResourceFilter(AdminResourceType.TOPIC, null, ResourceNameType.LITERAL), + new ResourcePatternFilter(AdminResourceType.TOPIC, null, ResourceNameType.LITERAL), new AccessControlEntryFilter("User:ANONYMOUS", "*", AclOperation.ANY, AclPermissionType.DENY)))) case ApiKeys.DESCRIBE_CONFIGS =>