feat: add acl resource literal and prefixed matching

README.md

@@ -11,10 +11,12 @@ Class implementing a single ACL entry verification. Principal, operation and
 resource are expressed as regular expressions.
 
 Alternatively to straight regular expression for resource, AivenAclEntry can
-be given a resource pattern with back references to principal regex. This is
-used internally in Aiven to map project id from certificate subject into
-project specific management topics. We can thus avoid encoding separate rules
-for each project.
+be given a resource pattern with back references to principal regex, a literal
+match or a prefixed match. The first is used internally in Aiven to map project
+id from certificate subject into project specific management topics. We can thus
+avoid encoding separate rules for each project. Literal and prefixed matchers
+work as defined in the kafka documentation. Only one resource matcher can be
+specified per acl.
 
 Permission type allows to define the verification result in case of an ACL match.
 By default, the permission type is `ALLOW`.

src/main/java/io/aiven/kafka/auth/json/AivenAcl.java

@@ -19,6 +19,9 @@
 import java.util.Objects;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import io.aiven.kafka.auth.utils.ResourceLiteralWildcardMatcher;
 
 import com.google.gson.annotations.SerializedName;
 
@@ -41,26 +44,39 @@ public class AivenAcl {
     @SerializedName("resource_pattern")
     public final String resourceRePattern;
 
+    @SerializedName("resource_literal")
+    public final String resourceLiteral;
+
+    @SerializedName("resource_prefix")
+    public final String resourcePrefix;
+
     @SerializedName("permission_type")
     private final AclPermissionType permissionType;
 
     @SerializedName("host")
     private final String hostMatcher;
 
-    public AivenAcl(final String principalType,
-                    final String principal,
-                    final String host,
-                    final String operation,
-                    final String resource,
-                    final String resourcePattern,
-                    final AclPermissionType permissionType) {
+    public AivenAcl(
+        final String principalType,
+        final String principal,
+        final String host,
+        final String operation,
+        final String resource,
+        final String resourcePattern,
+        final String resourceLiteral,
+        final String resourcePrefix,
+        final AclPermissionType permissionType
+    ) {
         this.principalType = principalType;
         this.principalRe = Pattern.compile(principal);
         this.hostMatcher = host;
         this.operationRe = Pattern.compile(operation);
         this.resourceRe = Objects.nonNull(resource) ? Pattern.compile(resource) : null;
         this.resourceRePattern = resourcePattern;
+        this.resourceLiteral = resourceLiteral;
+        this.resourcePrefix = resourcePrefix;
         this.permissionType = Objects.requireNonNullElse(permissionType, AclPermissionType.ALLOW);
+        this.validate();
     }
 
     public AclPermissionType getPermissionType() {
@@ -88,19 +104,7 @@ public Boolean match(final String principalType,
         if (this.principalType == null || this.principalType.equals(principalType)) {
             final Matcher mp = this.principalRe.matcher(principal);
             final Matcher mo = this.operationRe.matcher(operation);
-            if (mp.find() && mo.find() && this.hostMatch(host)) {
-                Matcher mr = null;
-                if (this.resourceRe != null) {
-                    mr = this.resourceRe.matcher(resource);
-                } else if (this.resourceRePattern != null) {
-                    final String resourceReStr = mp.replaceAll(this.resourceRePattern);
-                    final Pattern resourceRe = Pattern.compile(resourceReStr);
-                    mr = resourceRe.matcher(resource);
-                }
-                if (mr != null && mr.find()) {
-                    return true;
-                }
-            }
+            return mp.find() && mo.find() && this.hostMatch(host) && this.resourceMatch(resource, mp);
         }
         return false;
     }
@@ -110,6 +114,23 @@ private boolean hostMatch(final String host) {
             || getHostMatcher().equals(host);
     }
 
+    // split into separate function because of cyclomatic complexity check
+    private boolean resourceMatch(final String resource, final Matcher principalBackreference) {
+        if (this.resourceRe != null) {
+            return this.resourceRe.matcher(resource).find();
+        } else if (this.resourceRePattern != null) {
+            final String resourceReStr = principalBackreference.replaceAll(this.resourceRePattern);
+            final Pattern resourceRe = Pattern.compile(resourceReStr);
+            return resourceRe.matcher(resource).find();
+        } else if (this.resourceLiteral != null) {
+            return ResourceLiteralWildcardMatcher.match(this.resourceLiteral, resource)
+                || this.resourceLiteral.equals(resource);
+        } else if (this.resourcePrefix != null) {
+            return resource != null && resource.startsWith(this.resourcePrefix);
+        }
+        return false;
+    }
+
     @Override
     public boolean equals(final Object o) {
         if (this == o) {
@@ -118,13 +139,19 @@ public boolean equals(final Object o) {
         if (o == null || getClass() != o.getClass()) {
             return false;
         }
-        final AivenAcl aivenAcl = (AivenAcl) o;
+        return equals((AivenAcl) o);
+        // split for cyclomatic complexity check
+    }
+
+    private boolean equals(final AivenAcl aivenAcl) {
         return Objects.equals(principalType, aivenAcl.principalType)
             && comparePattern(principalRe, aivenAcl.principalRe)
             && getHostMatcher().equals(aivenAcl.getHostMatcher())
             && comparePattern(operationRe, aivenAcl.operationRe)
             && comparePattern(resourceRe, aivenAcl.resourceRe)
             && Objects.equals(resourceRePattern, aivenAcl.resourceRePattern)
+            && Objects.equals(resourceLiteral, aivenAcl.resourceLiteral)
+            && Objects.equals(resourcePrefix, aivenAcl.resourcePrefix)
             && getPermissionType() == aivenAcl.getPermissionType(); // always compare permission type using getter
     }
 
@@ -141,10 +168,22 @@ private boolean comparePattern(final Pattern p1, final Pattern p2) {
         return p1.toString().equals(p2.toString());
     }
 
+    public void validate() {
+        // check that only one of resources fields is not null
+        // in the future we can add checks for other fields
+        final long resourceCount = Stream.of(resourceRe, resourceRePattern, resourceLiteral, resourcePrefix)
+                .filter(Objects::nonNull)
+                .count();
+        if (resourceCount != 1) {
+            throw new IllegalArgumentException("Must specify exactly one \"resource\" for " + this);
+        }
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(
-            principalType, principalRe, hostMatcher, operationRe, resourceRe, resourceRePattern, getPermissionType()
+            principalType, principalRe, hostMatcher, operationRe, resourceRe,
+            resourceRePattern, resourceLiteral, resourcePrefix, getPermissionType()
         );
     }
 
@@ -156,6 +195,8 @@ public String toString() {
                 + ", operationRe=" + operationRe
                 + ", resourceRe=" + resourceRe
                 + ", resourceRePattern='" + resourceRePattern
+                + ", resourceLiteral='" + resourceLiteral
+                + ", resourcePrefix='" + resourcePrefix
                 + "', permissionType=" + getPermissionType()
                 + ", hostMatcher='" + getHostMatcher()
                 + "'}";

src/main/java/io/aiven/kafka/auth/json/reader/AclJsonReader.java

@@ -39,7 +39,11 @@ public AclJsonReader(final Path configPath) {
     public List<AivenAcl> read() {
         try (final Reader reader = Files.newBufferedReader(configFile)) {
             final java.lang.reflect.Type t = new TypeToken<Collection<AivenAcl>>() {}.getType();
-            return gsonBuilder.create().fromJson(reader, t);
+            final List<AivenAcl> acls = gsonBuilder.create().fromJson(reader, t);
+            if (acls != null) {
+                acls.forEach(AivenAcl::validate);
+            }
+            return acls;
         } catch (final JsonSyntaxException | JsonIOException | IOException ex) {
             throw new JsonReaderException(
                 String.format(

src/main/java/io/aiven/kafka/auth/nativeacls/AclAivenToNativeConverter.java

@@ -24,11 +24,13 @@
 import org.apache.kafka.common.acl.AclBinding;
 
 import io.aiven.kafka.auth.json.AivenAcl;
+import org.apache.kafka.common.resource.ResourcePattern;
 
 public class AclAivenToNativeConverter {
     public static List<AclBinding> convert(final AivenAcl aivenAcl) {
         final List<AclBinding> result = new ArrayList<>();
-        if (aivenAcl.resourceRe == null) {
+        if (aivenAcl.resourceRePattern != null) {
+            // resourceRePattern are for internal rules, so don't return them
             return result;
         }
 
@@ -43,8 +45,18 @@ public static List<AclBinding> convert(final AivenAcl aivenAcl) {
             for (final var principal : principals) {
                 final var accessControlEntry = new AccessControlEntry(
                     principal, aivenAcl.getHostMatcher(), operation, aivenAcl.getPermissionType().nativeType);
-                for (final var resourcePattern : ResourcePatternParser.parse(aivenAcl.resourceRe.pattern())) {
-                    result.add(new AclBinding(resourcePattern, accessControlEntry));
+                if (aivenAcl.resourceRe != null) {
+                    for (final var resourcePattern : ResourcePatternParser.parse(aivenAcl.resourceRe.pattern())) {
+                        result.add(new AclBinding(resourcePattern, accessControlEntry));
+                    }
+                } else if (aivenAcl.resourceLiteral != null) {
+                    ResourcePatternParser.parseLiteral(aivenAcl.resourceLiteral).ifPresent(
+                            resourcePattern -> result.add(new AclBinding(resourcePattern, accessControlEntry))
+                    );
+                } else if (aivenAcl.resourcePrefix != null) {
+                    ResourcePatternParser.parsePrefixed(aivenAcl.resourcePrefix).ifPresent(
+                            resourcePattern -> result.add(new AclBinding(resourcePattern, accessControlEntry))
+                    );
                 }
             }
         }

src/main/java/io/aiven/kafka/auth/nativeacls/ResourcePatternParser.java

@@ -19,11 +19,13 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import io.aiven.kafka.auth.nameformatters.ResourceTypeNameFormatter;
 import org.apache.kafka.common.resource.PatternType;
 import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.resource.ResourceType;
@@ -80,6 +82,32 @@ static Iterable<ResourcePattern> parse(final String resourcePattern) {
         return result;
     }
 
+    private static Optional<ResourcePattern> parseSerializedResource(
+        final String resourcePattern,
+        final PatternType patternType
+    ) {
+        if (resourcePattern == null) {
+            return Optional.empty();
+        }
+        final String[] parts = resourcePattern.split(":", 2);
+        if (parts.length != 2 || parts[0].isBlank() || parts[1].isBlank()) {
+            LOGGER.debug("Invalid format for resource literal '{}'", resourcePattern);
+            return Optional.empty();
+        }
+        return Optional.of(new ResourcePattern(
+                ResourceTypeNameFormatter.format(parts[0]),
+                parts[1],
+                patternType));
+    }
+
+    public static Optional<ResourcePattern> parseLiteral(final String resourcePattern) {
+        return parseSerializedResource(resourcePattern, PatternType.LITERAL);
+    }
+
+    public static Optional<ResourcePattern> parsePrefixed(final String resourcePattern) {
+        return parseSerializedResource(resourcePattern, PatternType.PREFIXED);
+    }
+
     private static ResourcePattern createResourcePattern(final ResourceType resourceType, final String resource) {
         if (resource.equals(".*") || resource.equals("(.*)")) {
             return new ResourcePattern(resourceType, "*", PatternType.LITERAL);

src/main/java/io/aiven/kafka/auth/utils/ResourceLiteralWildcardMatcher.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2019 Aiven Oy https://aiven.io
+ *
+ * Licensed 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 io.aiven.kafka.auth.utils;
+
+public class ResourceLiteralWildcardMatcher {
+    // Here "pattern" is something like "Topic:topic-1" or "Topic:*", where the second form is the
+    // wildcard matching. The wildcard match is a bit more difficult than comparing for just "*", because
+    // the prefix must be compared with the one of "resource".
+    public static boolean match(final String pattern, final String resource) {
+        if (pattern == null || resource == null) {
+            return false;
+        }
+        final int matchLength = Math.min(pattern.length(), resource.length());
+        for (int i = 0; i < matchLength; i++) {
+            if (pattern.charAt(i) != resource.charAt(i)) {
+                return false;
+            }
+            if (pattern.charAt(i) == ':') {
+                return pattern.length() > i + 1 && pattern.charAt(i + 1) == '*';
+            }
+        }
+        return false;
+    }
+}

src/test/java/io/aiven/kafka/auth/AivenAclAuthorizerV2Test.java

@@ -440,6 +440,30 @@ public void testHostMatcherWildcard() throws IOException {
         );
     }
 
+    @Test
+    public void testResourceLiteralMatcher() throws IOException {
+        Files.copy(this.getClass().getResourceAsStream("/acls_resource_literal_match.json"), configFilePath);
+        startAuthorizer();
+        checkSingleAction(requestCtx("User", "user1"), action(READ_OPERATION, topic("topic-1")), true);
+        checkSingleAction(requestCtx("User", "user1"), action(READ_OPERATION, topic("topic-2")), false);
+        checkSingleAction(requestCtx("User", "user2"), action(READ_OPERATION, topic("topic-2")), false);
+        checkSingleAction(requestCtx("User", "user3"), action(READ_OPERATION, topic("topic-3")), true);
+    }
+
+    @Test
+    public void testResourcePrefixMatcher() throws IOException {
+        Files.copy(this.getClass().getResourceAsStream("/acls_resource_prefix_match.json"), configFilePath);
+        startAuthorizer();
+        checkSingleAction(requestCtx("User", "user1"), action(READ_OPERATION, topic("groupA.topic")), true);
+        checkSingleAction(requestCtx("User", "user1"), action(READ_OPERATION, topic("groupC.topic")), false);
+        checkSingleAction(requestCtx("User", "user2"), action(READ_OPERATION, topic("groupB.topic")), false);
+    }
+
+
+    public ResourcePattern topic(final String name) {
+        return new ResourcePattern(ResourceType.TOPIC, name, PatternType.LITERAL);
+    }
+
     private void startAuthorizer() {
         final AuthorizerServerInfo serverInfo = mock(AuthorizerServerInfo.class);
         when(serverInfo.endpoints()).thenReturn(List.of());