kafka的ACL配置的sasl.kerberos.principal.to.local.rules配置解释

1、Kerberos中的介绍

Kerberos 关于此配置项的解释
https://web.mit.edu/Kerberos/krb5-latest/doc/admin/conf_files/krb5_conf.html 中的 auth_to_local

2、自定义sasl user name

https://kafka.apache.org/documentation/#security_authz

规则模板：

RULE:[n:string](regexp)s/pattern/replacement/g/L

官网规则示例：

sasl.kerberos.principal.to.local.rules=RULE:[1:$1@$0](.*@MYDOMAIN.COM)s/@.*//,DEFAULT

详见kafka的源码 org.apache.kafka.common.security.kerberos.KerberosRule#apply

String apply(String[] params) throws IOException {String result = null;if (isDefault) {if (defaultRealm.equals(params[0])) {result = params[1];}} else if (params.length - 1 == numOfComponents) {String base = replaceParameters(format, params);if (match == null || match.matcher(base).matches()) {if (fromPattern == null) {result = base;} else {result = replaceSubstitution(base, fromPattern, toPattern,  repeat);}}}if (result != null && NON_SIMPLE_PATTERN.matcher(result).find()) {throw new NoMatchingRule("Non-simple name " + result + " after auth_to_local rule " + this);}if (toLowerCase && result != null) {result = result.toLowerCase(Locale.ENGLISH);} else if (toUpperCase && result != null) {result = result.toUpperCase(Locale.ENGLISH);}return result;
}

传入参数为：

if (kerberosName.hostName() == null) {// if it is already simple, just return itif (kerberosName.realm() == null)return kerberosName.serviceName();params = new String[]{kerberosName.realm(), kerberosName.serviceName()};
} else {params = new String[]{kerberosName.realm(), kerberosName.serviceName(), kerberosName.hostName()};
}

其中realm为域名，serviceName为用户名，hostName为主机名
通过源码可以看到其中的$0 $1 $2 并不是后面regexp正则匹配后的反引用，而是配置的 Kerberos 主体（principal）
整个规则理解为将 Kerberos 主体（principal）映射到本地用户名（local principal）的转换规则
其中解释顺序为：

1. 组装string
2. 验证组装的string是否匹配regexp
3. 将组装的string进行正则替换，将pattern替换为replacement，根据是否配置/g确定是替换1次还是替换所有
4. 将最终结果转换为大写或者小写

综上，regexp仅用于了开始组装结果的校验，并没有进行值的提取（建立引用）

示例结果

servicename=tkgup
realm=qq.com

经过RULE:[1:$1@$0](.*@qq.com)s/@.*//后，得到string为[e-mail]，验证符合regexp，再经过后面正则查找替换得到：tkgup

3、自定义ssl 的用户名

规则模板：

RULE:pattern/replacement/[LU]

规则示例：

RULE:^.*[Cc][Nn]=([a-zA-Z0-9.]*).*$/$1/L,DEFAULT

具体解析见：
org.apache.kafka.common.security.ssl.SslPrincipalMapper.Rule#apply
解释顺序为：

1. 将distinguishedName：principal.getName()正则pattern替换为replacement，会解析反引用，如$1、$2
2. 将最终结果转换为大写或者小写

在此规则下：
tkgup.cn=[e-mail] => tkgup666

4、关于kafka配置kerberos以及开启acl的实践

参考：Kafka配置Kerberos安全认证及与Java程序集成
参考：kafka 配置kerberos校验以及开启acl实践 https://cloud.tencent.com/developer/article/1593946
参考：Kafka配置kerberos安全认证