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 it
    if (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为tkgup@qq.com，验证符合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=tkgup666@qq.com => tkgup666

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

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