(kafka) branch 3.8 updated: KAFKA-18051: Disallow creating ACLs with principals that do not contain a colon (#17883)

[cmccabe]

This is an automated email from the ASF dual-hosted git repository.

cmccabe pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/3.8 by this push:
     new e749b571953 KAFKA-18051: Disallow creating ACLs with principals that 
do not contain a colon (#17883)
e749b571953 is described below

commit e749b57195328862b779056f07124083c383e45d
Author: Colin Patrick McCabe <cmcc...@apache.org>
AuthorDate: Fri Nov 22 16:50:33 2024 -0800

    KAFKA-18051: Disallow creating ACLs with principals that do not contain a 
colon (#17883)
    
    Kafka Principals must contain a colon. We should enforce this in createAcls.
    
    Reviewers: David Arthur <mum...@gmail.com>
---
 .../apache/kafka/controller/AclControlManager.java |  6 +++++
 .../kafka/controller/AclControlManagerTest.java    | 28 ++++++++++++++++++++++
 2 files changed, 34 insertions(+)

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 3e1804cf43c..f849b540a57 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java
@@ -159,6 +159,12 @@ public class AclControlManager {
         if (binding.pattern().name() == null || 
binding.pattern().name().isEmpty()) {
             throw new InvalidRequestException("Resource name should not be 
empty");
         }
+        int colonIndex = binding.entry().principal().indexOf(":");
+        if (colonIndex == -1) {
+            throw new InvalidRequestException("Could not parse principal from 
`" +
+                binding.entry().principal() + "` " + "(no colon is present 
separating the " +
+                "principal type from the principal name)");
+        }
     }
 
     ControllerResult<List<AclDeleteResult>> deleteAcls(List<AclBindingFilter> 
filters) {
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 dd6c2d15185..a62b9f00d6d 100644
--- 
a/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
+++ 
b/metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
@@ -113,6 +113,34 @@ public class AclControlManagerTest {
                 getMessage());
     }
 
+    /**
+     * Verify that validateNewAcl catches invalid ACLs with principals that do 
not contain a colon.
+     */
+    @Test
+    public void testValidateAclWithBadPrincipal() {
+        assertEquals("Could not parse principal from `invalid` (no colon is 
present " +
+                "separating the principal type from the principal name)",
+            assertThrows(InvalidRequestException.class, () ->
+                AclControlManager.validateNewAcl(new AclBinding(
+                    new ResourcePattern(TOPIC, "*", LITERAL),
+                    new AccessControlEntry("invalid", "*", ALTER, ALLOW)))).
+                getMessage());
+    }
+
+    /**
+     * Verify that validateNewAcl catches invalid ACLs with principals that do 
not contain a colon.
+     */
+    @Test
+    public void testValidateAclWithEmptyPrincipal() {
+        assertEquals("Could not parse principal from `` (no colon is present " 
+
+                "separating the principal type from the principal name)",
+            assertThrows(InvalidRequestException.class, () ->
+                AclControlManager.validateNewAcl(new AclBinding(
+                    new ResourcePattern(TOPIC, "*", LITERAL),
+                    new AccessControlEntry("", "*", ALTER, ALLOW)))).
+                        getMessage());
+    }
+
     /**
      * Verify that validateFilter catches invalid filters.
      */