Merge pull request #218 from Aiven-Open/implicit-describe-operations

Support implicit describe operations

README.md

@@ -7,7 +7,7 @@ Config file is watched for modifications and reloaded as necessary.
 
 ### AivenAclEntry
 
-Class implementing a single ACL entry verification. Principal, operation and
+Class implementing a single ACL entry verification. Principal and
 resource are expressed as regular expressions.
 
 Alternatively to straight regular expression for resource, AivenAclEntry can
@@ -18,6 +18,12 @@ 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.
 
+Operations can be expressed as a list of operation names, or in deprecated mode
+as regular expression in `operation` field.  If both are defined, `operations`
+takes precedence.  For operations listed with operation names, also implicit Decribe
+is supported if Read, Write, Alter, or Delete is allowed, and implicit
+DescribeConfigs if AlterConfigs is allowed.
+
 Permission type allows to define the verification result in case of an ACL match.
 By default, the permission type is `ALLOW`.
 
@@ -27,7 +33,7 @@ A specific ACL entry can be hidden from public listing by setting hidden flag.
 
     [
         {
-            "operation": "^(.*)$",
+            "operations": ["All"],
             "principal": (
                 "^CN=(?<vmname>[a-z0-9-]+),OU=(?<nodeid>n[0-9]+),"
                 "O=00000000-0000-a000-1000-(500000000005|a00000000001|b00000000001|d00000000001),ST=vm$"
@@ -38,6 +44,7 @@ A specific ACL entry can be hidden from public listing by setting hidden flag.
             "hidden": true
         },
         {
+            "operations": ["Describe", "DescribeConfigs", "Read", "Write"],
             "operation": "^(Describe|DescribeConfigs|Read|Write)$",
             "principal": "^CN=(?<vmname>[a-z0-9-]+),OU=(?<nodeid>n[0-9]+),O=(?<projectid>[a-f0-9-]+),ST=vm$",
             "principal_type": "Prune",

build.gradle

@@ -74,7 +74,7 @@ jmh {
 }
 
 ext {
-    kafkaVersion = "2.8.0"
+    kafkaVersion = "3.6.0"
     mockitoVersion = "5.14.2"
 }
 

src/main/java/io/aiven/kafka/auth/AivenAclAuthorizerV2.java

@@ -59,7 +59,6 @@
 import io.aiven.kafka.auth.json.AivenAcl;
 import io.aiven.kafka.auth.json.reader.AclJsonReader;
 import io.aiven.kafka.auth.json.reader.JsonReaderException;
-import io.aiven.kafka.auth.nameformatters.LegacyOperationNameFormatter;
 import io.aiven.kafka.auth.nameformatters.LegacyResourceTypeNameFormatter;
 import io.aiven.kafka.auth.nativeacls.AclAivenToNativeConverter;
 
@@ -195,7 +194,7 @@ public final List<AuthorizationResult> authorize(final AuthorizableRequestContex
             final String host = requestContext.clientAddress().getHostAddress();
             final boolean verdict = cacheReference.get().get(principal,
                                                              host,
-                                                             LegacyOperationNameFormatter.format(operation),
+                                                             operation,
                                                              resourceToCheck);
             final var authResult = verdict ? AuthorizationResult.ALLOWED : AuthorizationResult.DENIED;
 

src/main/java/io/aiven/kafka/auth/VerdictCache.java

@@ -26,6 +26,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.security.auth.KafkaPrincipal;
 
 import io.aiven.kafka.auth.json.AclPermissionType;
@@ -44,7 +45,7 @@ private VerdictCache(@Nonnull final List<AivenAcl> denyAclEntries, @Nonnull fina
     public boolean get(
         final KafkaPrincipal principal,
         final String host,
-        final String operation,
+        final AclOperation operation,
         final String resource
     ) {
         final String principalType = principal.getPrincipalType();

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

@@ -0,0 +1,40 @@
+/*
+ * 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.json;
+
+public enum AclOperationType {
+    Unknown(org.apache.kafka.common.acl.AclOperation.UNKNOWN),
+    All(org.apache.kafka.common.acl.AclOperation.ALL),
+    Read(org.apache.kafka.common.acl.AclOperation.READ),
+    Write(org.apache.kafka.common.acl.AclOperation.WRITE),
+    Create1(org.apache.kafka.common.acl.AclOperation.CREATE),
+    Delete(org.apache.kafka.common.acl.AclOperation.DELETE),
+    Alter(org.apache.kafka.common.acl.AclOperation.ALTER),
+    Describe(org.apache.kafka.common.acl.AclOperation.DESCRIBE),
+    ClusterAction(org.apache.kafka.common.acl.AclOperation.CLUSTER_ACTION),
+    DescribeConfigs(org.apache.kafka.common.acl.AclOperation.DESCRIBE_CONFIGS),
+    AlterConfigs(org.apache.kafka.common.acl.AclOperation.ALTER_CONFIGS),
+    IdempotentWrite(org.apache.kafka.common.acl.AclOperation.IDEMPOTENT_WRITE),
+    CreateTokens(org.apache.kafka.common.acl.AclOperation.CREATE_TOKENS),
+    DescribeTokens(org.apache.kafka.common.acl.AclOperation.DESCRIBE_TOKENS);
+
+    public final org.apache.kafka.common.acl.AclOperation nativeType;
+
+    AclOperationType(final org.apache.kafka.common.acl.AclOperation nativeType) {
+        this.nativeType = nativeType;
+    }
+}

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

@@ -16,10 +16,17 @@
 
 package io.aiven.kafka.auth.json;
 
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import org.apache.kafka.common.acl.AclOperation;
+
+import io.aiven.kafka.auth.nameformatters.LegacyOperationNameFormatter;
 import io.aiven.kafka.auth.utils.ResourceLiteralWildcardMatcher;
 
 import com.google.gson.annotations.SerializedName;
@@ -37,6 +44,8 @@ public class AivenAcl {
     @SerializedName("operation")
     public final Pattern operationRe;
 
+    public final List<AclOperationType> operations;
+
     @SerializedName("resource")
     public final Pattern resourceRe;
 
@@ -73,6 +82,32 @@ public AivenAcl(
         this.principalRe = Pattern.compile(principal);
         this.hostMatcher = host;
         this.operationRe = Pattern.compile(operation);
+        this.operations = null;
+        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;
+    }
+
+    public AivenAcl(
+        final String principalType,
+        final String principal,
+        final String host,
+        final List<AclOperationType> operations,
+        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 = null;
+        this.operations = operations;
         this.resourceRe = Objects.nonNull(resource) ? Pattern.compile(resource) : null;
         this.resourceRePattern = resourcePattern;
         this.resourceLiteral = resourceLiteral;
@@ -101,12 +136,63 @@ public String getHostMatcher() {
     public Boolean match(final String principalType,
                             final String principal,
                             final String host,
-                            final String operation,
+                            final AclOperation operation,
                             final String resource) {
         if (this.principalType == null || this.principalType.equals(principalType)) {
             final Matcher mp = this.principalRe.matcher(principal);
-            final Matcher mo = this.operationRe.matcher(operation);
-            return mp.find() && mo.find() && this.hostMatch(host) && this.resourceMatch(resource, mp);
+            return mp.find() && this.matchOperation(operation) && this.hostMatch(host)
+                    && this.resourceMatch(resource, mp);
+        }
+        return false;
+    }
+
+    /* Some operations implicitly allow describe operation, but with allow rules only */
+    private static final Set<AclOperation> IMPLIES_DESCRIBE = Collections.unmodifiableSet(
+        EnumSet.of(AclOperation.DESCRIBE, AclOperation.READ, AclOperation.WRITE,
+            AclOperation.DELETE, AclOperation.ALTER));
+
+    private static final Set<AclOperation> IMPLIES_DESCRIBE_CONFIGS = Collections.unmodifiableSet(
+        EnumSet.of(AclOperation.DESCRIBE_CONFIGS, AclOperation.ALTER_CONFIGS));
+
+
+    private boolean matchOperation(final AclOperation operation) {
+        if (this.operations != null) {
+            for (final var mo : this.operations) {
+                if (matchSingleOperation(operation, mo)) {
+                    return true;
+                }
+            }
+            return false;
+        } else {
+            final Matcher mo = this.operationRe.matcher(LegacyOperationNameFormatter.format(operation));
+            return mo.find();
+        }
+    }
+
+    private boolean matchSingleOperation(final AclOperation operation, final AclOperationType mo) {
+        if (mo.nativeType == AclOperation.ALL) {
+            return true;
+        } else if (getPermissionType() == AclPermissionType.ALLOW) {
+            switch (operation) {
+                case DESCRIBE:
+                    if (IMPLIES_DESCRIBE.contains(mo.nativeType)) {
+                        return true;
+                    }
+                    break;
+                case DESCRIBE_CONFIGS:
+                    if (IMPLIES_DESCRIBE_CONFIGS.contains(mo.nativeType)) {
+                        return true;
+                    }
+                    break;
+                default:
+                    if (operation == mo.nativeType) {
+                        return true;
+                    }
+            }
+        } else {
+            if (operation == mo.nativeType) {
+                return true;
+            }
         }
         return false;
     }
@@ -177,7 +263,7 @@ private boolean comparePattern(final Pattern p1, final Pattern p2) {
     @Override
     public int hashCode() {
         return Objects.hash(
-            principalType, principalRe, hostMatcher, operationRe, resourceRe,
+            principalType, principalRe, hostMatcher, operationRe, operations, resourceRe,
             resourceRePattern, resourceLiteral, resourcePrefix, getPermissionType()
         );
     }

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

@@ -21,6 +21,7 @@
 import java.util.regex.Pattern;
 import java.util.regex.PatternSyntaxException;
 
+import io.aiven.kafka.auth.json.AclOperationType;
 import io.aiven.kafka.auth.json.AclPermissionType;
 
 import com.google.gson.GsonBuilder;
@@ -37,6 +38,7 @@ public abstract class AbstractJsonReader<T> implements JsonReader<T> {
     protected final GsonBuilder gsonBuilder =
         new GsonBuilder()
             .registerTypeAdapter(Pattern.class, new RegexpJsonDeserializer())
+            .registerTypeAdapter(AclOperationType.class, new AclOperationTypeDeserializer())
             .registerTypeAdapter(AclPermissionType.class, new AclPermissionTypeDeserializer());
 
     protected AbstractJsonReader(final Path configFile) {
@@ -72,4 +74,19 @@ public AclPermissionType deserialize(final JsonElement jsonElement,
         }
     }
 
+    protected static class AclOperationTypeDeserializer implements JsonDeserializer<AclOperationType> {
+        @Override
+        public AclOperationType deserialize(final JsonElement jsonElement,
+                                             final Type type,
+                                             final JsonDeserializationContext ctx) throws JsonParseException {
+            try {
+                if (jsonElement.isJsonNull()) {
+                    return AclOperationType.Unknown;
+                }
+                return AclOperationType.valueOf(jsonElement.getAsString());
+            } catch (final IllegalArgumentException e) {
+                throw new JsonParseException("Cannot deserialize operation type", e);
+            }
+        }
+    }
 }

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

@@ -19,9 +19,11 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 import org.apache.kafka.common.acl.AccessControlEntry;
 import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclOperation;
 
 import io.aiven.kafka.auth.json.AivenAcl;
 
@@ -36,7 +38,11 @@ public static List<AclBinding> convert(final AivenAcl aivenAcl) {
             return result;
         }
 
-        for (final var operation : AclOperationsParser.parse(aivenAcl.operationRe.pattern())) {
+        final Iterable<AclOperation> operations = aivenAcl.operations != null
+             ? aivenAcl.operations.stream().map(o -> o.nativeType).collect(Collectors.toList())
+             : AclOperationsParser.parse(aivenAcl.operationRe.pattern());
+
+        for (final var operation : operations) {
             final List<String> principals = AclPrincipalFormatter.parse(
                 aivenAcl.principalType, aivenAcl.principalRe.pattern()
             );