From e75e4732c9ad67a5bf7a267346f0774c2a41b4e9 Mon Sep 17 00:00:00 2001
From: ryannatesmith <natesmith@yandex.ru>
Date: Sun, 20 Jan 2019 22:31:36 -0800
Subject: [PATCH] MINOR: Rejoin split ssl principal mapping rules (#6099)

* Join ssl principal mapping rules correctly before evaluating.

Java properties splits the configuration array on commas, and that leads to rules containing commas being split before being evaluated. This commit adds a code change to re-join those strings into full rules before evaluating them. The function assumes every rule is either DEFAULT or begins with the prefix RULE:
---
 .../security/ssl/SslPrincipalMapper.java      | 29 +++++++++++++++++--
 .../security/ssl/SslPrincipalMapperTest.java  | 25 ++++++++++++++++
 2 files changed, 52 insertions(+), 2 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslPrincipalMapper.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslPrincipalMapper.java
index 7ec4a79b2eb..3b95e1a5f4d 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslPrincipalMapper.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslPrincipalMapper.java
@@ -17,9 +17,9 @@
 package org.apache.kafka.common.security.ssl;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
+import java.util.Collections;
+import java.util.ArrayList;
 import java.util.Locale;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -39,7 +39,32 @@ public class SslPrincipalMapper {
         return new SslPrincipalMapper(parseRules(rules));
     }
 
+    private static List<String> joinSplitRules(List<String> rules) {
+        String rule = "RULE:";
+        String defaultRule = "DEFAULT";
+        List<String> retVal = new ArrayList<>();
+        StringBuilder currentRule = new StringBuilder();
+        for (String r : rules) {
+            if (currentRule.length() > 0) {
+                if (r.startsWith(rule) || r.equals(defaultRule)) {
+                    retVal.add(currentRule.toString());
+                    currentRule.setLength(0);
+                    currentRule.append(r);
+                } else {
+                    currentRule.append(String.format(",%s", r));
+                }
+            } else {
+                currentRule.append(r);
+            }
+        }
+        if (currentRule.length() > 0) {
+            retVal.add(currentRule.toString());
+        }
+        return retVal;
+    }
+
     private static List<Rule> parseRules(List<String> rules) {
+        rules = joinSplitRules(rules);
         List<Rule> result = new ArrayList<>();
         for (String rule : rules) {
             Matcher matcher = RULE_PARSER.matcher(rule);
diff --git a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslPrincipalMapperTest.java b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslPrincipalMapperTest.java
index c647fd00a3d..56ef977a112 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslPrincipalMapperTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslPrincipalMapperTest.java
@@ -36,6 +36,16 @@ public class SslPrincipalMapperTest {
         testValidRule(Arrays.asList("RULE:^cn=(.?),ou=(.?),dc=(.?),dc=(.?)$/$1@$2/U"));
     }
 
+    @Test
+    public void testValidSplitRules() {
+        testValidRule(Arrays.asList("DEFAULT"));
+        testValidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/$1/"));
+        testValidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/$1/L", "DEFAULT"));
+        testValidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=(.*?),O=(.*?),L=(.*?)", "ST=(.*?)", "C=(.*?)$/$1@$2/"));
+        testValidRule(Arrays.asList("RULE:^.*[Cc][Nn]=([a-zA-Z0-9.]*).*$/$1/L"));
+        testValidRule(Arrays.asList("RULE:^cn=(.?)", "ou=(.?)", "dc=(.?)", "dc=(.?)$/$1@$2/U"));
+    }
+
     private void testValidRule(List<String> rules) {
         SslPrincipalMapper.fromRules(rules);
     }
@@ -55,6 +65,21 @@ public class SslPrincipalMapperTest {
         testInvalidRule(Arrays.asList("RULE:^CN=(.*?),OU=ServiceUsers.*$/LU"));
     }
 
+    @Test
+    public void testInvalidSplitRules() {
+        testInvalidRule(Arrays.asList("default"));
+        testInvalidRule(Arrays.asList("DEFAUL"));
+        testInvalidRule(Arrays.asList("DEFAULT/L"));
+        testInvalidRule(Arrays.asList("DEFAULT/U"));
+
+        testInvalidRule(Arrays.asList("RULE:CN=(.*?)", "OU=ServiceUsers.*/$1"));
+        testInvalidRule(Arrays.asList("rule:^CN=(.*?)", "OU=ServiceUsers.*$/$1/"));
+        testInvalidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/$1/L/U"));
+        testInvalidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/L"));
+        testInvalidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/U"));
+        testInvalidRule(Arrays.asList("RULE:^CN=(.*?)", "OU=ServiceUsers.*$/LU"));
+    }
+
     private void testInvalidRule(List<String> rules) {
         try {
             System.out.println(SslPrincipalMapper.fromRules(rules));