This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 82b804fe272 KAFKA-20088: CIDR-based Host Patterns for ACLs (#22501)
82b804fe272 is described below
commit 82b804fe27252816e05cf9360d27f6e743592621
Author: Maros Orsak <[email protected]>
AuthorDate: Thu Jul 9 16:11:24 2026 +0200
KAFKA-20088: CIDR-based Host Patterns for ACLs (#22501)
Implements KIP-1276.
Signed-off-by: see-quick <[email protected]>
Reviewers: Mickael Maison <[email protected]>, Luke Chen
<[email protected]>
---
LICENSE-binary | 1 +
build.gradle | 4 +
checkstyle/import-control-metadata.xml | 1 +
gradle/dependencies.gradle | 2 +
.../apache/kafka/controller/AclControlManager.java | 48 +++++-
.../apache/kafka/controller/QuorumController.java | 2 +-
.../kafka/metadata/authorizer/CidrUtils.java | 63 +++++++
.../authorizer/StandardAuthorizerData.java | 2 +-
.../kafka/controller/AclControlManagerTest.java | 121 ++++++++++++--
.../kafka/metadata/authorizer/CidrUtilsTest.java | 112 +++++++++++++
.../authorizer/StandardAuthorizerTest.java | 182 +++++++++++++++++++++
.../apache/kafka/controller/MockAclMutator.java | 3 +-
.../kafka/server/common/MetadataVersion.java | 8 +-
.../apache/kafka/common/test/api/ClusterTest.java | 2 +-
14 files changed, 530 insertions(+), 21 deletions(-)
diff --git a/LICENSE-binary b/LICENSE-binary
index cb2e37445e8..bec83923222 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -210,6 +210,7 @@ License Version 2.0:
- commons-collections-3.2.2
- commons-digester-2.1
- commons-logging-1.3.5
+- commons-net-3.13.0
- commons-validator-1.10.1
- hash4j-0.22.0
- jackson-annotations-2.21
diff --git a/build.gradle b/build.gradle
index 970f527a524..1bd566ccee4 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1462,6 +1462,10 @@ project(':metadata') {
implementation project(':server-common')
implementation project(':clients')
implementation project(':raft')
+ // commons-io is a transitive dependency not needed by
SubnetUtils/SubnetUtils6
+ implementation(libs.commonsNet) {
+ exclude group: 'commons-io', module: 'commons-io'
+ }
implementation libs.jacksonDatabind
implementation libs.jacksonJDK8Datatypes
implementation libs.metrics
diff --git a/checkstyle/import-control-metadata.xml
b/checkstyle/import-control-metadata.xml
index cdc73dcf65d..1ee3207de44 100644
--- a/checkstyle/import-control-metadata.xml
+++ b/checkstyle/import-control-metadata.xml
@@ -168,6 +168,7 @@
<allow pkg="org.apache.kafka.common.metrics" />
<allow pkg="org.apache.kafka.common.metrics.internals" />
<allow pkg="org.apache.kafka.common.metrics.stats" />
+ <allow pkg="org.apache.commons.net.util" />
</subpackage>
<subpackage name="bootstrap">
<allow pkg="org.apache.kafka.snapshot" />
diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle
index 2e1105702f4..7a4aefd1751 100644
--- a/gradle/dependencies.gradle
+++ b/gradle/dependencies.gradle
@@ -58,6 +58,7 @@ versions += [
bndlib: "7.1.0",
checkstyle: project.hasProperty('checkstyleVersion') ? checkstyleVersion :
"12.3.1",
commonsValidator: "1.10.1",
+ commonsNet: "3.13.0",
classgraph: "4.8.179",
gradle: "9.4.1",
grgit: "5.3.3",
@@ -158,6 +159,7 @@ libs += [
bndlib:"biz.aQute.bnd:biz.aQute.bndlib:$versions.bndlib",
caffeine: "com.github.ben-manes.caffeine:caffeine:$versions.caffeine",
classgraph: "io.github.classgraph:classgraph:$versions.classgraph",
+ commonsNet: "commons-net:commons-net:$versions.commonsNet",
commonsValidator:
"commons-validator:commons-validator:$versions.commonsValidator",
jacksonAnnotations:
"com.fasterxml.jackson.core:jackson-annotations:$versions.jacksonAnnotations",
jacksonDatabind:
"com.fasterxml.jackson.core:jackson-databind:$versions.jackson",
diff --git
a/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java
b/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java
index c3ce52870db..22808ced18d 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java
@@ -23,16 +23,19 @@ import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.UnknownServerException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.metadata.AccessControlEntryRecord;
import org.apache.kafka.common.metadata.RemoveAccessControlEntryRecord;
import org.apache.kafka.common.requests.ApiError;
import org.apache.kafka.common.utils.internals.LogContext;
+import org.apache.kafka.metadata.authorizer.CidrUtils;
import org.apache.kafka.metadata.authorizer.StandardAcl;
import org.apache.kafka.metadata.authorizer.StandardAclWithId;
import org.apache.kafka.server.authorizer.AclCreateResult;
import org.apache.kafka.server.authorizer.AclDeleteResult;
import
org.apache.kafka.server.authorizer.AclDeleteResult.AclBindingDeleteResult;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.mutable.BoundedList;
import org.apache.kafka.server.mutable.BoundedListTooLongException;
import org.apache.kafka.timeline.SnapshotRegistry;
@@ -92,14 +95,14 @@ public class AclControlManager {
this.existingAcls = new TimelineHashSet<>(snapshotRegistry, 0);
}
- ControllerResult<List<AclCreateResult>> createAcls(List<AclBinding> acls) {
+ ControllerResult<List<AclCreateResult>> createAcls(List<AclBinding> acls,
MetadataVersion metadataVersion) {
Set<StandardAcl> aclsToCreate = new HashSet<>(acls.size());
List<AclCreateResult> results = new ArrayList<>(acls.size());
List<ApiMessageAndVersion> records =
BoundedList.newArrayBacked(MAX_RECORDS_PER_USER_OP);
for (AclBinding acl : acls) {
try {
- validateNewAcl(acl);
+ validateNewAcl(acl, metadataVersion.isCidrAclSupported());
} catch (Throwable t) {
ApiException e = (t instanceof ApiException) ? (ApiException)
t :
new UnknownServerException("Unknown error while trying to
create ACL", t);
@@ -132,7 +135,7 @@ public class AclControlManager {
return uuid;
}
- static void validateNewAcl(AclBinding binding) {
+ static void validateNewAcl(AclBinding binding, boolean isCidrAclSupported)
{
switch (binding.pattern().resourceType()) {
case UNKNOWN:
case ANY:
@@ -174,6 +177,45 @@ public class AclControlManager {
binding.entry().principal() + "` " + "(no colon is present
separating the " +
"principal type from the principal name)");
}
+ validateHostPattern(binding.entry().host(), isCidrAclSupported);
+ }
+
+ /**
+ * Validates the host pattern of an ACL entry.
+ *
+ * Accepts:
+ * - Wildcard "*" (matches any host)
+ * - Valid IPv4 address (e.g., "192.168.1.1")
+ * - Valid IPv6 address (e.g., "2001:db8::1")
+ * - Valid IPv4 CIDR notation (e.g., "192.168.0.0/24"), which requires
cidrSupported=true
+ * - Valid IPv6 CIDR notation (e.g., "2001:db8::/32"), which requires
cidrSupported=true
+ *
+ * @param host The host pattern to validate
+ * @param isCidrSupported Whether CIDR notation is supported by the
current metadata version
+ * @throws InvalidRequestException if the host pattern is invalid
+ * @throws UnsupportedVersionException if CIDR notation is used but not
supported
+ */
+ static void validateHostPattern(String host, boolean isCidrSupported) {
+ if (host == null || host.isEmpty()) {
+ throw new InvalidRequestException("Host pattern cannot be null or
empty");
+ }
+
+ if ("*".equals(host)) {
+ return;
+ }
+
+ if (host.contains("/")) {
+ if (!isCidrSupported) {
+ throw new UnsupportedVersionException(
+ "CIDR-based ACL host patterns require metadata version " +
+ MetadataVersion.IBP_4_4_IV1 + " or higher.");
+ }
+ try {
+ CidrUtils.validate(host);
+ } catch (IllegalArgumentException e) {
+ throw new InvalidRequestException("Invalid CIDR notation '" +
host + "': " + e.getMessage());
+ }
+ }
}
ControllerResult<List<AclDeleteResult>> deleteAcls(List<AclBindingFilter>
filters) {
diff --git
a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
index 5da6eb2e3aa..6620c7e6dc9 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -2159,7 +2159,7 @@ public final class QuorumController implements Controller
{
List<AclBinding> aclBindings
) {
return appendWriteEvent("createAcls", context.deadlineNs(),
- () -> aclControlManager.createAcls(aclBindings));
+ () -> aclControlManager.createAcls(aclBindings,
featureControl.metadataVersionOrThrow()));
}
@Override
diff --git
a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/CidrUtils.java
b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/CidrUtils.java
new file mode 100644
index 00000000000..318b2c9bff0
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/CidrUtils.java
@@ -0,0 +1,63 @@
+/*
+ * 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.metadata.authorizer;
+
+import org.apache.commons.net.util.SubnetUtils;
+import org.apache.commons.net.util.SubnetUtils6;
+
+public final class CidrUtils {
+
+ private CidrUtils() {}
+
+ private static boolean isIpv6(String cidrPattern) {
+ return cidrPattern.contains(":");
+ }
+
+ /**
+ * Validates a CIDR pattern by parsing it and constructing the appropriate
SubnetUtils.
+ * Throws {@link IllegalArgumentException} if the pattern is invalid.
+ */
+ public static void validate(String cidrPattern) {
+ if (isIpv6(cidrPattern)) {
+ new SubnetUtils6(cidrPattern);
+ } else {
+ new SubnetUtils(cidrPattern);
+ }
+ }
+
+ /**
+ * Checks whether a host IP address falls within a CIDR range.
+ * Returns false if the pattern is null, not a CIDR notation, or invalid.
+ */
+ public static boolean isInRange(String host, String cidrPattern) {
+ if (cidrPattern == null || !cidrPattern.contains("/")) {
+ return false;
+ }
+ try {
+ if (isIpv6(cidrPattern)) {
+ return new SubnetUtils6(cidrPattern).getInfo().isInRange(host);
+ } else {
+ SubnetUtils subnet = new SubnetUtils(cidrPattern);
+ subnet.setInclusiveHostCount(true);
+ return subnet.getInfo().isInRange(host);
+ }
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
+ }
+}
diff --git
a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java
b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java
index e10b73277c1..f56c39dc379 100644
---
a/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java
+++
b/metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java
@@ -489,7 +489,7 @@ public class StandardAuthorizerData {
return null;
}
// Check if the host matches. If it doesn't, return no result (null).
- if (!acl.host().equals(WILDCARD) && !acl.host().equals(host)) {
+ if (!acl.host().equals(WILDCARD) && !acl.host().equals(host) &&
!CidrUtils.isInRange(host, acl.host())) {
return null;
}
// Check if the operation field matches. Here we hit a slight
complication.
diff --git
a/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
b/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
index 73d910dfa44..5982f865c93 100644
---
a/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
+++
b/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
@@ -27,6 +27,7 @@ import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.NotControllerException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.metadata.AccessControlEntryRecord;
import org.apache.kafka.common.metadata.RemoveAccessControlEntryRecord;
import org.apache.kafka.common.resource.PatternType;
@@ -49,6 +50,7 @@ import
org.apache.kafka.server.authorizer.AuthorizableRequestContext;
import org.apache.kafka.server.authorizer.AuthorizationResult;
import org.apache.kafka.server.authorizer.AuthorizerServerInfo;
import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.mutable.BoundedListTooLongException;
import org.apache.kafka.timeline.SnapshotRegistry;
@@ -88,30 +90,30 @@ public class AclControlManagerTest {
public void testValidateNewAcl() {
AclControlManager.validateNewAcl(new AclBinding(
new ResourcePattern(TOPIC, "*", LITERAL),
- new AccessControlEntry("User:*", "*", ALTER, ALLOW)));
+ new AccessControlEntry("User:*", "*", ALTER, ALLOW)), false);
assertEquals("Invalid patternType UNKNOWN",
assertThrows(InvalidRequestException.class, () ->
AclControlManager.validateNewAcl(new AclBinding(
new ResourcePattern(TOPIC, "*", PatternType.UNKNOWN),
- new AccessControlEntry("User:*", "*", ALTER, ALLOW)))).
+ new AccessControlEntry("User:*", "*", ALTER, ALLOW)),
false)).
getMessage());
assertEquals("Invalid resourceType UNKNOWN",
assertThrows(InvalidRequestException.class, () ->
AclControlManager.validateNewAcl(new AclBinding(
new ResourcePattern(ResourceType.UNKNOWN, "*", LITERAL),
- new AccessControlEntry("User:*", "*", ALTER, ALLOW)))).
+ new AccessControlEntry("User:*", "*", ALTER, ALLOW)),
false)).
getMessage());
assertEquals("Invalid operation UNKNOWN",
assertThrows(InvalidRequestException.class, () ->
AclControlManager.validateNewAcl(new AclBinding(
new ResourcePattern(TOPIC, "*", LITERAL),
- new AccessControlEntry("User:*", "*",
AclOperation.UNKNOWN, ALLOW)))).
+ new AccessControlEntry("User:*", "*",
AclOperation.UNKNOWN, ALLOW)), false)).
getMessage());
assertEquals("Invalid permissionType UNKNOWN",
assertThrows(InvalidRequestException.class, () ->
AclControlManager.validateNewAcl(new AclBinding(
new ResourcePattern(TOPIC, "*", LITERAL),
- new AccessControlEntry("User:*", "*", ALTER,
AclPermissionType.UNKNOWN)))).
+ new AccessControlEntry("User:*", "*", ALTER,
AclPermissionType.UNKNOWN)), false)).
getMessage());
}
@@ -125,7 +127,7 @@ public class AclControlManagerTest {
assertThrows(InvalidRequestException.class, () ->
AclControlManager.validateNewAcl(new AclBinding(
new ResourcePattern(TOPIC, "*", LITERAL),
- new AccessControlEntry("invalid", "*", ALTER, ALLOW)))).
+ new AccessControlEntry("invalid", "*", ALTER, ALLOW)),
false)).
getMessage());
}
@@ -139,7 +141,7 @@ public class AclControlManagerTest {
assertThrows(InvalidRequestException.class, () ->
AclControlManager.validateNewAcl(new AclBinding(
new ResourcePattern(TOPIC, "*", LITERAL),
- new AccessControlEntry("", "*", ALTER, ALLOW)))).
+ new AccessControlEntry("", "*", ALTER, ALLOW)), false)).
getMessage());
}
@@ -291,7 +293,7 @@ public class AclControlManagerTest {
new ResourcePattern(TOPIC, "*", PatternType.UNKNOWN),
new AccessControlEntry("User:*", "*", ALTER, ALLOW)));
- ControllerResult<List<AclCreateResult>> createResult =
manager.createAcls(toCreate);
+ ControllerResult<List<AclCreateResult>> createResult =
manager.createAcls(toCreate, MetadataVersion.IBP_4_0_IV0);
List<AclCreateResult> expectedResults = new ArrayList<>();
for (int i = 0; i < 3; i++) {
@@ -346,12 +348,12 @@ public class AclControlManagerTest {
AclBinding aclBinding = new AclBinding(new ResourcePattern(TOPIC,
"topic-1", LITERAL),
new AccessControlEntry("User:user", "10.0.0.1",
AclOperation.ALL, ALLOW));
- ControllerResult<List<AclCreateResult>> createResult =
manager.createAcls(List.of(aclBinding, aclBinding));
+ ControllerResult<List<AclCreateResult>> createResult =
manager.createAcls(List.of(aclBinding, aclBinding),
MetadataVersion.IBP_4_0_IV0);
RecordTestUtils.replayAll(manager, createResult.records());
assertEquals(1, createResult.records().size());
assertEquals(1, manager.idToAcl().size());
- createResult = manager.createAcls(List.of(aclBinding));
+ createResult = manager.createAcls(List.of(aclBinding),
MetadataVersion.IBP_4_0_IV0);
assertEquals(0, createResult.records().size());
assertEquals(1, manager.idToAcl().size());
}
@@ -363,7 +365,7 @@ public class AclControlManagerTest {
AclBinding aclBinding = new AclBinding(new ResourcePattern(TOPIC,
"topic-1", LITERAL),
new AccessControlEntry("User:user", "10.0.0.1",
AclOperation.ALL, ALLOW));
- ControllerResult<List<AclCreateResult>> createResult =
manager.createAcls(List.of(aclBinding));
+ ControllerResult<List<AclCreateResult>> createResult =
manager.createAcls(List.of(aclBinding), MetadataVersion.IBP_4_0_IV0);
RecordTestUtils.replayAll(manager, createResult.records());
Uuid id = ((AccessControlEntryRecord)
createResult.records().get(0).message()).id();
assertEquals(1, createResult.records().size());
@@ -413,13 +415,13 @@ public class AclControlManagerTest {
secondCreate.add(acl.toBinding());
}
}
- ControllerResult<List<AclCreateResult>> firstCreateResult =
manager.createAcls(firstCreate);
+ ControllerResult<List<AclCreateResult>> firstCreateResult =
manager.createAcls(firstCreate, MetadataVersion.IBP_4_0_IV0);
assertEquals((MAX_RECORDS_PER_USER_OP / 2) + 1,
firstCreateResult.response().size());
for (AclCreateResult result : firstCreateResult.response()) {
assertTrue(result.exception().isEmpty());
}
- ControllerResult<List<AclCreateResult>> secondCreateResult =
manager.createAcls(secondCreate);
+ ControllerResult<List<AclCreateResult>> secondCreateResult =
manager.createAcls(secondCreate, MetadataVersion.IBP_4_0_IV0);
assertEquals((MAX_RECORDS_PER_USER_OP / 2) + 1,
secondCreateResult.response().size());
for (AclCreateResult result : secondCreateResult.response()) {
assertTrue(result.exception().isEmpty());
@@ -440,4 +442,97 @@ public class AclControlManagerTest {
assertEquals(BoundedListTooLongException.class,
exception.getCause().getClass());
assertEquals("Cannot remove more than " + MAX_RECORDS_PER_USER_OP + "
acls in a single delete operation.", exception.getCause().getMessage());
}
+
+ @Test
+ public void testValidateHostPatternValid() {
+ // Wildcard
+ AclControlManager.validateHostPattern("*", false);
+ AclControlManager.validateHostPattern("*", true);
+
+ // Plain IPv4 addresses
+ AclControlManager.validateHostPattern("192.168.1.1", false);
+ AclControlManager.validateHostPattern("127.0.0.1", true);
+
+ // Plain IPv6 addresses
+ AclControlManager.validateHostPattern("2001:db8::1", false);
+ AclControlManager.validateHostPattern("::1", true);
+
+ // Hostnames (no slash, so validateHostPattern accepts them as literal
hosts)
+ AclControlManager.validateHostPattern("example.com", false);
+ AclControlManager.validateHostPattern("example.com", true);
+ }
+
+ @Test
+ public void testValidateHostPatternInvalid() {
+ // Null or empty
+ assertThrows(InvalidRequestException.class, () ->
+ AclControlManager.validateHostPattern(null, true));
+ assertThrows(InvalidRequestException.class, () ->
+ AclControlManager.validateHostPattern("", true));
+
+ // Invalid CIDR wraps CidrUtils.validate() IllegalArgumentException
into InvalidRequestException
+ InvalidRequestException e =
assertThrows(InvalidRequestException.class, () ->
+ AclControlManager.validateHostPattern("192.168.0.0/33", true));
+ assertTrue(e.getMessage().contains("Invalid CIDR notation"));
+
+ // Hostname with path is not a valid CIDR
+ e = assertThrows(InvalidRequestException.class, () ->
+ AclControlManager.validateHostPattern("example.com/test", true));
+ assertTrue(e.getMessage().contains("Invalid CIDR notation"));
+ }
+
+ @Test
+ public void testValidateHostPatternCidrNotSupported() {
+ // CIDR patterns should be rejected when cidrSupported is false
+ UnsupportedVersionException e =
assertThrows(UnsupportedVersionException.class, () ->
+ AclControlManager.validateHostPattern("192.168.0.0/24", false));
+ assertTrue(e.getMessage().contains("CIDR-based ACL host patterns
require metadata version"));
+
+ e = assertThrows(UnsupportedVersionException.class, () ->
+ AclControlManager.validateHostPattern("2001:db8::/32", false));
+ assertTrue(e.getMessage().contains("CIDR-based ACL host patterns
require metadata version"));
+ }
+
+ @Test
+ public void testCreateAclWithCidrHosts() {
+ AclControlManager manager = new AclControlManager.Builder().build();
+
+ // Valid CIDR with supported version
+ assertAclCreateSucceeds(manager, "192.168.0.0/24",
MetadataVersion.IBP_4_4_IV1);
+ assertAclCreateSucceeds(manager, "2001:db8::/32",
MetadataVersion.IBP_4_4_IV1);
+
+ // Regular hosts with older version
+ assertAclCreateSucceeds(manager, "192.168.0.1",
MetadataVersion.IBP_4_0_IV0);
+ assertAclCreateSucceeds(manager, "*", MetadataVersion.IBP_4_0_IV0);
+
+ // Invalid CIDR notation
+ assertAclCreateFails(manager, "192.168.0.0/33",
MetadataVersion.IBP_4_4_IV1,
+ InvalidRequestException.class, "Invalid CIDR notation");
+
+ // Valid CIDR with unsupported version
+ assertAclCreateFails(manager, "192.168.0.0/24",
MetadataVersion.IBP_4_0_IV0,
+ UnsupportedVersionException.class, "CIDR-based ACL host patterns
require metadata version");
+ }
+
+ private static void assertAclCreateSucceeds(AclControlManager manager,
String host, MetadataVersion version) {
+ AclBinding acl = new AclBinding(
+ new ResourcePattern(TOPIC, "topic-" + host, LITERAL),
+ new AccessControlEntry("User:test", host, ALTER, ALLOW));
+ ControllerResult<List<AclCreateResult>> result =
manager.createAcls(List.of(acl), version);
+ assertEquals(1, result.response().size());
+ assertFalse(result.response().get(0).exception().isPresent(),
+ "Expected success for host=" + host + " version=" + version);
+ }
+
+ private static void assertAclCreateFails(AclControlManager manager, String
host, MetadataVersion version,
+ Class<? extends Exception>
exClass, String messageContains) {
+ AclBinding acl = new AclBinding(
+ new ResourcePattern(TOPIC, "topic-" + host, LITERAL),
+ new AccessControlEntry("User:test", host, ALTER, ALLOW));
+ ControllerResult<List<AclCreateResult>> result =
manager.createAcls(List.of(acl), version);
+ assertEquals(1, result.response().size());
+ assertTrue(result.response().get(0).exception().isPresent());
+
assertTrue(exClass.isInstance(result.response().get(0).exception().get()));
+
assertTrue(result.response().get(0).exception().get().getMessage().contains(messageContains));
+ }
}
diff --git
a/metadata/src/test/java/org/apache/kafka/metadata/authorizer/CidrUtilsTest.java
b/metadata/src/test/java/org/apache/kafka/metadata/authorizer/CidrUtilsTest.java
new file mode 100644
index 00000000000..f648944f0ba
--- /dev/null
+++
b/metadata/src/test/java/org/apache/kafka/metadata/authorizer/CidrUtilsTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.metadata.authorizer;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.net.InetAddress;
+
+import static org.apache.kafka.metadata.authorizer.CidrUtils.isInRange;
+import static org.apache.kafka.metadata.authorizer.CidrUtils.validate;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
+@Timeout(value = 40)
+public class CidrUtilsTest {
+
+ @Test
+ public void testValidateValidIpv4Cidr() {
+ assertDoesNotThrow(() -> validate("192.168.0.0/24"));
+ assertDoesNotThrow(() -> validate("10.0.0.0/8"));
+ assertDoesNotThrow(() -> validate("172.16.0.0/16"));
+ assertDoesNotThrow(() -> validate("192.168.1.1/32"));
+ assertDoesNotThrow(() -> validate("0.0.0.0/0"));
+ }
+
+ @Test
+ public void testValidateValidIpv6Cidr() {
+ assertDoesNotThrow(() -> validate("2001:db8::/32"));
+ assertDoesNotThrow(() -> validate("2001:db8:abcd::/48"));
+ assertDoesNotThrow(() -> validate("::1/128"));
+ assertDoesNotThrow(() -> validate("::/0"));
+ }
+
+ @Test
+ public void testValidateInvalidIpv4Cidr() {
+ assertThrows(IllegalArgumentException.class, () ->
validate("192.168.0.0/33"));
+ assertThrows(IllegalArgumentException.class, () ->
validate("192.168.0.0/-1"));
+ assertThrows(IllegalArgumentException.class, () ->
validate("192.168.0.256/24"));
+ assertThrows(IllegalArgumentException.class, () ->
validate("192.168.0.0/abc"));
+ assertThrows(IllegalArgumentException.class, () ->
validate("192.168.0.0/"));
+ }
+
+ @Test
+ public void testValidateInvalidIpv6Cidr() {
+ assertThrows(IllegalArgumentException.class, () ->
validate("2001:db8::/129"));
+ }
+
+ @Test
+ public void testValidateIpv4MappedIpv6Cidr() {
+ // ::ffff:x.x.x.x/N contains ':', so isIpv6() returns true and
SubnetUtils6 handles it.
+ assertThrows(IllegalArgumentException.class, () ->
validate("::ffff:192.168.1.0/24"));
+ }
+
+ @Test
+ public void testValidateHostnameWithPath() {
+ assertThrows(IllegalArgumentException.class, () ->
validate("example.com/test"));
+ }
+
+ @Test
+ public void testIsInRangeNullOrNonCidr() {
+ assertFalse(isInRange("192.168.1.1", null));
+ assertFalse(isInRange("192.168.1.1", "192.168.1.1"));
+ }
+
+ @Test
+ public void testIpv4MappedAddress() throws Exception {
+ // JVM normalizes ::ffff:x.x.x.x to plain IPv4 Inet4Address,
+ // so CidrUtils dispatches to SubnetUtils (IPv4 path), not
SubnetUtils6.
+ String normalizedHost =
InetAddress.getByName("::ffff:192.168.1.5").getHostAddress();
+ assertEquals("192.168.1.5", normalizedHost);
+ assertTrue(isInRange(normalizedHost, "192.168.1.0/24"));
+ assertFalse(isInRange(normalizedHost, "10.0.0.0/8"));
+ }
+
+ @Test
+ public void testIpv4MappedAddressWithSubnet() {
+ // ::ffff:x.x.x.x/N contains ':', so isIpv6() returns true and
SubnetUtils6 handles it.
+ // Plain IPv4 hosts don't match this IPv6 CIDR range.
+ assertFalse(isInRange("192.168.1.5", "::ffff:192.168.1.0/24"));
+ assertFalse(isInRange("::1", "::ffff:192.168.1.0/24"));
+ assertFalse(isInRange("2001:db8::1", "::ffff:192.168.1.0/24"));
+
+ // ::x.x.x.x/N (deprecated IPv4-compatible form): JVM keeps it as
Inet6Address,
+ // so CidrUtils dispatches to SubnetUtils6. Plain IPv4 clients won't
match.
+ assertFalse(isInRange("192.168.1.5", "::192.168.1.0/24"));
+ // IPv6 clients within the /24 range do match.
+ assertTrue(isInRange("::1", "::192.168.1.0/24"));
+ assertTrue(isInRange("0:0:0:0:0:0:c0a8:105", "::192.168.1.0/24"));
+ // IPv6 clients outside the /24 range do not match.
+ assertFalse(isInRange("2001:db8::1", "::192.168.1.0/24"));
+ }
+}
diff --git
a/metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerTest.java
b/metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerTest.java
index 3430259ebc5..35e49bb85f5 100644
---
a/metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerTest.java
+++
b/metadata/src/test/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerTest.java
@@ -693,4 +693,186 @@ public class StandardAuthorizerTest {
// StandardAuthorizer has 4 metrics
assertEquals(5, metrics.metrics().size());
}
+
+ @Test
+ public void testAclWithCidrHost() throws Exception {
+ StandardAuthorizer authorizer =
createAndInitializeStandardAuthorizer();
+
+ StandardAcl cidrAcl = new StandardAcl(
+ TOPIC, "test-topic", LITERAL, "User:bob",
+ "192.168.1.0/24", READ, ALLOW);
+ StandardAclWithId aclWithId = withId(cidrAcl);
+ authorizer.addAcl(aclWithId.id(), aclWithId.acl());
+
+ // Boundary IPs inside the /24 range
+ for (String ip : List.of("192.168.1.0", "192.168.1.1",
"192.168.1.128", "192.168.1.255")) {
+ assertEquals(List.of(ALLOWED),
+ authorizer.authorize(
+ new MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "bob"))
+ .setClientAddress(InetAddress.getByName(ip))
+ .build(),
+ List.of(newAction(READ, TOPIC, "test-topic"))),
+ "IP " + ip + " should be allowed within 192.168.1.0/24");
+ }
+
+ // IPs just outside the range
+ for (String ip : List.of("192.168.0.255", "192.168.2.0")) {
+ assertEquals(List.of(DENIED),
+ authorizer.authorize(
+ new MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "bob"))
+ .setClientAddress(InetAddress.getByName(ip))
+ .build(),
+ List.of(newAction(READ, TOPIC, "test-topic"))),
+ "IP " + ip + " should be denied outside 192.168.1.0/24");
+ }
+ }
+
+ @Test
+ public void testAclWithIpv6CidrHost() throws Exception {
+ StandardAuthorizer authorizer =
createAndInitializeStandardAuthorizer();
+
+ StandardAcl cidrAcl = new StandardAcl(
+ TOPIC, "test-topic", LITERAL, "User:bob",
+ "2001:db8::/120", READ, ALLOW);
+ StandardAclWithId aclWithId = withId(cidrAcl);
+ authorizer.addAcl(aclWithId.id(), aclWithId.acl());
+
+ // Boundary IPs inside the /120 range
+ for (String ip : List.of("2001:db8::0", "2001:db8::1", "2001:db8::80",
"2001:db8::ff")) {
+ assertEquals(List.of(ALLOWED),
+ authorizer.authorize(
+ new MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "bob"))
+ .setClientAddress(InetAddress.getByName(ip))
+ .build(),
+ List.of(newAction(READ, TOPIC, "test-topic"))),
+ "IP " + ip + " should be allowed within 2001:db8::/120");
+ }
+
+ // IPs just outside the range
+ for (String ip : List.of("2001:db8::100", "2001:db7::1")) {
+ assertEquals(List.of(DENIED),
+ authorizer.authorize(
+ new MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "bob"))
+ .setClientAddress(InetAddress.getByName(ip))
+ .build(),
+ List.of(newAction(READ, TOPIC, "test-topic"))),
+ "IP " + ip + " should be denied outside 2001:db8::/120");
+ }
+ }
+
+ @Test
+ public void testOverlappingCidrDenyNarrowAllowWide() throws Exception {
+ StandardAuthorizer authorizer =
createAndInitializeStandardAuthorizer();
+
+ StandardAclWithId allowAcl = withId(new StandardAcl(
+ TOPIC, "foo", LITERAL, "User:alice", "10.0.0.0/8", READ, ALLOW));
+ StandardAclWithId denyAcl = withId(new StandardAcl(
+ TOPIC, "foo", LITERAL, "User:alice", "10.0.1.0/24", READ, DENY));
+ authorizer.addAcl(allowAcl.id(), allowAcl.acl());
+ authorizer.addAcl(denyAcl.id(), denyAcl.acl());
+
+ AuthorizableRequestContext aliceInDenyRange = new
MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "alice"))
+ .setClientAddress(InetAddress.getByName("10.0.1.5"))
+ .build();
+ assertEquals(List.of(DENIED),
+ authorizer.authorize(aliceInDenyRange, List.of(newAction(READ,
TOPIC, "foo"))),
+ "Client in overlapping DENY /24 inside ALLOW /8 must be denied");
+
+ AuthorizableRequestContext aliceOutsideDenyRange = new
MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "alice"))
+ .setClientAddress(InetAddress.getByName("10.0.2.1"))
+ .build();
+ assertEquals(List.of(ALLOWED),
+ authorizer.authorize(aliceOutsideDenyRange,
List.of(newAction(READ, TOPIC, "foo"))),
+ "Client in ALLOW /8 but outside DENY /24 must be allowed");
+ }
+
+ @Test
+ public void testOverlappingCidrAllowNarrowDenyWide() throws Exception {
+ StandardAuthorizer authorizer =
createAndInitializeStandardAuthorizer();
+
+ StandardAclWithId denyAcl = withId(new StandardAcl(
+ TOPIC, "foo", LITERAL, "User:alice", "10.0.0.0/8", READ, DENY));
+ StandardAclWithId allowAcl = withId(new StandardAcl(
+ TOPIC, "foo", LITERAL, "User:alice", "10.0.1.0/24", READ, ALLOW));
+ authorizer.addAcl(denyAcl.id(), denyAcl.acl());
+ authorizer.addAcl(allowAcl.id(), allowAcl.acl());
+
+ AuthorizableRequestContext aliceInBothRanges = new
MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "alice"))
+ .setClientAddress(InetAddress.getByName("10.0.1.5"))
+ .build();
+ assertEquals(List.of(DENIED),
+ authorizer.authorize(aliceInBothRanges, List.of(newAction(READ,
TOPIC, "foo"))),
+ "Narrow ALLOW /24 cannot override broad DENY /8 — DENY always
wins");
+
+ AuthorizableRequestContext aliceOnlyInDenyRange = new
MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "alice"))
+ .setClientAddress(InetAddress.getByName("10.0.2.1"))
+ .build();
+ assertEquals(List.of(DENIED),
+ authorizer.authorize(aliceOnlyInDenyRange, List.of(newAction(READ,
TOPIC, "foo"))),
+ "Client in DENY /8 range must be denied");
+ }
+
+ @Test
+ public void testOverlappingCidrWithWildcardAllow() throws Exception {
+ StandardAuthorizer authorizer =
createAndInitializeStandardAuthorizer();
+
+ StandardAclWithId allowAcl = withId(new StandardAcl(
+ TOPIC, "foo", LITERAL, "User:alice", "*", READ, ALLOW));
+ StandardAclWithId denyAcl = withId(new StandardAcl(
+ TOPIC, "foo", LITERAL, "User:alice", "10.0.1.0/24", READ, DENY));
+ authorizer.addAcl(allowAcl.id(), allowAcl.acl());
+ authorizer.addAcl(denyAcl.id(), denyAcl.acl());
+
+ AuthorizableRequestContext aliceInDenyRange = new
MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "alice"))
+ .setClientAddress(InetAddress.getByName("10.0.1.100"))
+ .build();
+ assertEquals(List.of(DENIED),
+ authorizer.authorize(aliceInDenyRange, List.of(newAction(READ,
TOPIC, "foo"))),
+ "DENY CIDR must override wildcard ALLOW");
+
+ AuthorizableRequestContext aliceOutsideDenyRange = new
MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "alice"))
+ .setClientAddress(InetAddress.getByName("10.0.2.100"))
+ .build();
+ assertEquals(List.of(ALLOWED),
+ authorizer.authorize(aliceOutsideDenyRange,
List.of(newAction(READ, TOPIC, "foo"))),
+ "Client outside DENY CIDR with wildcard ALLOW must be allowed");
+ }
+
+ @Test
+ public void testOverlappingCidrSamePermissionBothAllow() throws Exception {
+ StandardAuthorizer authorizer =
createAndInitializeStandardAuthorizer();
+
+ StandardAclWithId broadAllow = withId(new StandardAcl(
+ TOPIC, "foo", LITERAL, "User:alice", "10.0.0.0/8", READ, ALLOW));
+ StandardAclWithId narrowAllow = withId(new StandardAcl(
+ TOPIC, "foo", LITERAL, "User:alice", "10.0.1.0/24", READ, ALLOW));
+ authorizer.addAcl(broadAllow.id(), broadAllow.acl());
+ authorizer.addAcl(narrowAllow.id(), narrowAllow.acl());
+
+ AuthorizableRequestContext aliceInBothRanges = new
MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "alice"))
+ .setClientAddress(InetAddress.getByName("10.0.1.5"))
+ .build();
+ assertEquals(List.of(ALLOWED),
+ authorizer.authorize(aliceInBothRanges, List.of(newAction(READ,
TOPIC, "foo"))),
+ "Overlapping ALLOW CIDRs should both result in ALLOWED");
+
+ AuthorizableRequestContext aliceInBroadOnly = new
MockAuthorizableRequestContext.Builder()
+ .setPrincipal(new KafkaPrincipal(USER_TYPE, "alice"))
+ .setClientAddress(InetAddress.getByName("10.0.2.1"))
+ .build();
+ assertEquals(List.of(ALLOWED),
+ authorizer.authorize(aliceInBroadOnly, List.of(newAction(READ,
TOPIC, "foo"))),
+ "Client in only the broad ALLOW /8 should be allowed");
+ }
}
diff --git
a/metadata/src/testFixtures/java/org/apache/kafka/controller/MockAclMutator.java
b/metadata/src/testFixtures/java/org/apache/kafka/controller/MockAclMutator.java
index 342d164a232..72747d74e89 100644
---
a/metadata/src/testFixtures/java/org/apache/kafka/controller/MockAclMutator.java
+++
b/metadata/src/testFixtures/java/org/apache/kafka/controller/MockAclMutator.java
@@ -26,6 +26,7 @@ import org.apache.kafka.metadata.authorizer.StandardAcl;
import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
import org.apache.kafka.server.authorizer.AclCreateResult;
import org.apache.kafka.server.authorizer.AclDeleteResult;
+import org.apache.kafka.server.common.MetadataVersion;
import java.util.HashMap;
import java.util.List;
@@ -72,7 +73,7 @@ public class MockAclMutator implements AclMutator {
List<AclBinding> aclBindings
) {
Map<Uuid, StandardAcl> prevIdToAcl = new
HashMap<>(aclControl.idToAcl());
- ControllerResult<List<AclCreateResult>> result =
aclControl.createAcls(aclBindings);
+ ControllerResult<List<AclCreateResult>> result =
aclControl.createAcls(aclBindings, MetadataVersion.latestTesting());
RecordTestUtils.replayAll(aclControl, result.records());
syncIdToAcl(prevIdToAcl, aclControl.idToAcl());
return CompletableFuture.completedFuture(result.response());
diff --git
a/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java
b/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java
index f93503ed905..86a11a429fb 100644
---
a/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java
+++
b/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java
@@ -133,8 +133,10 @@ public enum MetadataVersion {
// IBP_4_4_IV0 enables dead-letter queue support for share groups
(KIP-1191). When this version
// is finalized, so will the DLQ support.
- IBP_4_4_IV0(31, "4.4", "IV0", false);
+ IBP_4_4_IV0(31, "4.4", "IV0", false),
+ // Add support for CIDR-based ACL host patterns (KIP-1276).
+ IBP_4_4_IV1(32, "4.4", "IV1", true);
// NOTES when adding a new version:
// Update the default version in @ClusterTest annotation to point to the
latest version
@@ -213,6 +215,10 @@ public enum MetadataVersion {
return this.isAtLeast(IBP_4_0_IV1);
}
+ public boolean isCidrAclSupported() {
+ return this.isAtLeast(IBP_4_4_IV1);
+ }
+
public boolean isMigrationSupported() {
return this.isAtLeast(MetadataVersion.IBP_3_4_IV0);
}
diff --git
a/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterTest.java
b/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterTest.java
index af11eaaeea2..b884ec5734e 100644
---
a/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterTest.java
+++
b/test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterTest.java
@@ -52,7 +52,7 @@ public @interface ClusterTest {
String brokerListener() default DEFAULT_BROKER_LISTENER_NAME;
SecurityProtocol controllerSecurityProtocol() default
SecurityProtocol.PLAINTEXT;
String controllerListener() default DEFAULT_CONTROLLER_LISTENER_NAME;
- MetadataVersion metadataVersion() default MetadataVersion.IBP_4_4_IV0;
+ MetadataVersion metadataVersion() default MetadataVersion.IBP_4_4_IV1;
ClusterConfigProperty[] serverProperties() default {};
// users can add tags that they want to display in test
String[] tags() default {};
