Merge pull request #215 from Aiven-Open/fdorlandi-add-literal-and-pre…

…fixed-matching

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 Apache 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

@@ -20,6 +20,8 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import io.aiven.kafka.auth.utils.ResourceLiteralWildcardMatcher;
+
 import com.google.gson.annotations.SerializedName;
 
 public class AivenAcl {
@@ -41,6 +43,12 @@ 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;
 
@@ -49,20 +57,26 @@ public class AivenAcl {
 
     private final boolean hidden;
 
-    public AivenAcl(final String principalType,
-                    final String principal,
-                    final String host,
-                    final String operation,
-                    final String resource,
-                    final String resourcePattern,
-                    final AclPermissionType permissionType,
-                    final boolean hidden) {
+    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,
+        final boolean hidden
+    ) {
         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.hidden = hidden;
     }
@@ -92,19 +106,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;
     }
@@ -114,6 +116,22 @@ private boolean hostMatch(final String host) {
             || getHostMatcher().equals(host);
     }
 
+    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) {
@@ -138,7 +156,9 @@ private boolean equalsPrincipal(final AivenAcl aivenAcl) {
 
     private boolean equalsResource(final AivenAcl aivenAcl) {
         return comparePattern(resourceRe, aivenAcl.resourceRe)
-            && Objects.equals(resourceRePattern, aivenAcl.resourceRePattern);
+            && Objects.equals(resourceRePattern, aivenAcl.resourceRePattern)
+            && Objects.equals(resourceLiteral, aivenAcl.resourceLiteral)
+            && Objects.equals(resourcePrefix, aivenAcl.resourcePrefix);
     }
 
     private boolean comparePattern(final Pattern p1, final Pattern p2) {
@@ -157,7 +177,8 @@ private boolean comparePattern(final Pattern p1, final Pattern p2) {
     @Override
     public int hashCode() {
         return Objects.hash(
-            principalType, principalRe, hostMatcher, operationRe, resourceRe, resourceRePattern, getPermissionType()
+            principalType, principalRe, hostMatcher, operationRe, resourceRe,
+            resourceRePattern, resourceLiteral, resourcePrefix, getPermissionType()
         );
     }
 
@@ -167,14 +188,25 @@ public String toString() {
                 + "principalType='" + principalType
                 + "', principalRe=" + principalRe
                 + ", operationRe=" + operationRe
-                + ", resourceRe=" + resourceRe
-                + ", resourceRePattern='" + resourceRePattern
+                + ", resource='" + getResourceString()
                 + "', permissionType=" + getPermissionType()
                 + ", hostMatcher='" + getHostMatcher()
                 + ", hidden=" + hidden
                 + "'}";
     }
 
+    private String getResourceString() {
+        if (resourceRe != null) {
+            return resourceRe.toString();
+        } else if (resourceRePattern != null) {
+            return resourceRePattern;
+        } else if (resourceLiteral != null) {
+            return resourceLiteral + " (LITERAL)";
+        } else {
+            return resourcePrefix + " (PREFIXED)";
+        }
+    }
+
     public boolean isHidden() {
         return hidden;
     }

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

@@ -28,7 +28,7 @@
 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) {
             return result;
         }
 
@@ -43,8 +43,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,6 +19,7 @@
 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;
@@ -28,6 +29,8 @@
 import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.resource.ResourceType;
 
+import io.aiven.kafka.auth.nameformatters.ResourceTypeNameFormatter;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -80,6 +83,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 2024 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());

src/test/java/io/aiven/kafka/auth/json/AivenAclTest.java

@@ -32,7 +32,10 @@ public void testAivenAclEntry() {
             "^(Describe|Read)$", // operation
             "^Topic:p_(.*)_s", // resource,
             null, // resource pattern
-            null, false
+            null, // resource literal
+            null, // resource prefix
+            null, // permission type
+            false // hidden
         );
 
         assertTrue(entry.match("User",  "CN=p_pass_s", "*", "Read", "Topic:p_pass_s"));
@@ -49,7 +52,10 @@ public void testAivenAclEntry() {
             "^(Describe|Read)$", // operation
             "^Topic:p_(.*)_s", // resource
             null, // resource pattern
-            null, false
+            null, // resource literal
+            null, // resource prefix
+            null, // permission type
+            false // hidden
         );
 
         assertTrue(entry.match("User", "CN=p_pass_s", "*", "Read", "Topic:p_pass_s"));
@@ -65,12 +71,53 @@ public void testAivenAclEntry() {
             "^(Describe|Read)$", // operation
             null, // resource
             "^Topic:p_${username}_s\\$", // resource pattern
-            null, false
+            null, // resource literal
+            null, // resource prefix
+            null, // permission type
+            false // hidden
         );
 
         assertTrue(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:p_user1_s"));
         assertTrue(entry.match("User", "CN=p_user2_s", "*", "Read", "Topic:p_user2_s"));
         assertFalse(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:p_user2_s"));
         assertFalse(entry.match("User", "CN=p_user2_s", "*", "Read", "Topic:p_user1_s"));
+
+        // Test resources defined by literal match
+        entry = new AivenAcl(
+                "User", // principal type
+                "^CN=p_(?<username>[a-z0-9]+)_s$", // principal
+                "*", // host
+                "^(Describe|Read)$", // operation
+                null, // resource
+                null, // resource pattern
+                "Topic:^(][", // invalid regex just to show that the match is a literal string
+                null, // resource prefix
+                null, // permission type
+                false // hidden
+        );
+
+        assertTrue(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:^(]["));
+        assertFalse(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:wrong_topic"));
+
+
+        // Test resources defined by prefix match
+        entry = new AivenAcl(
+                "User", // principal type
+                "^CN=p_(?<username>[a-z0-9]+)_s$", // principal
+                "*", // host
+                "^(Describe|Read)$", // operation
+                null, // resource
+                null, // resource pattern
+                null, // resource literal
+                "Topic:organizationA.", // invalid regex just to show that the match is a literal string
+                null, // permission type
+                false // hidden
+        );
+        assertTrue(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:organizationA.topic1"));
+        assertTrue(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:organizationA.topic2"));
+        assertTrue(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:organizationA."));
+        assertFalse(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:organizationB.topic1"));
+        assertFalse(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:organizationA"));
+        assertFalse(entry.match("User", "CN=p_user1_s", "*", "Read", "Topic:AAAorganizationA."));
     }
 }