Fix nullpointer that occurs when OAuthKafkaPrincipalBuilder is used with Kerberos listener

.travis.yml

@@ -13,6 +13,7 @@ jobs:
         - hydra
         - hydra-jwt
         - mockoauth
+        - kerberos
       apt:
         packages:
         - maven
@@ -26,6 +27,7 @@ jobs:
         - hydra
         - hydra-jwt
         - mockoauth
+        - kerberos
       apt:
         packages:
         - maven
@@ -43,6 +45,7 @@ addons:
   - hydra
   - hydra-jwt
   - mockoauth
+  - kerberos
 env:
   global:
   - PULL_REQUEST=${TRAVIS_PULL_REQUEST}

.travis/build.sh

@@ -2,7 +2,7 @@
 set -e
 
 clearDockerEnv() {
-  docker rm -f kafka zookeeper keycloak keycloak-import hydra hydra-import hydra-jwt hydra-jwt-import || true
+  docker rm -f kafka zookeeper keycloak keycloak-import hydra hydra-import hydra-jwt hydra-jwt-import kerberos || true
   DOCKER_TEST_NETWORKS=$(docker network ls | grep test | awk '{print $1}')
   [ "$DOCKER_TEST_NETWORKS" != "" ] && docker network rm $DOCKER_TEST_NETWORKS
 }

oauth-server/src/main/java/io/strimzi/kafka/oauth/server/OAuthKafkaPrincipalBuilder.java

@@ -18,10 +18,12 @@
 import org.apache.kafka.common.security.auth.KafkaPrincipal;
 import org.apache.kafka.common.security.auth.SaslAuthenticationContext;
 import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder;
+import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;
 import org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslServer;
 import org.apache.kafka.common.security.plain.internals.PlainSaslServer;
 
+import javax.security.auth.kerberos.KerberosPrincipal;
 import javax.security.sasl.SaslServer;
 import java.io.IOException;
 import java.lang.reflect.Field;
@@ -51,7 +53,8 @@
  */
 public class OAuthKafkaPrincipalBuilder extends DefaultKafkaPrincipalBuilder implements Configurable {
 
-    private static final SetAccessibleAction SET_PRINCIPAL_MAPPER = SetAccessibleAction.newInstance();
+    private static final SetAccessibleAction SET_PRINCIPAL_MAPPER = SetAccessibleAction.newInstance("sslPrincipalMapper");
+    private static final SetAccessibleAction SET_KERBEROS_SHORT_NAMER = SetAccessibleAction.newInstance("kerberosShortNamer");
 
     private static final int OAUTH_DATA_TAG = 575;
 
@@ -74,9 +77,9 @@ void invoke(DefaultKafkaPrincipalBuilder target, Object value) throws IllegalAcc
             field.set(target, value);
         }
 
-        static SetAccessibleAction newInstance() {
+        static SetAccessibleAction newInstance(String fieldName) {
             try {
-                return new SetAccessibleAction(DefaultKafkaPrincipalBuilder.class.getDeclaredField("sslPrincipalMapper"));
+                return new SetAccessibleAction(DefaultKafkaPrincipalBuilder.class.getDeclaredField(fieldName));
             } catch (NoSuchFieldException e) {
                 throw new IllegalStateException("Failed to install OAuthKafkaPrincipalBuilder. This Kafka version does not seem to be supported", e);
             }
@@ -91,12 +94,21 @@ public OAuthKafkaPrincipalBuilder() {
         super(null, null);
     }
 
+
+
     @Override
     public void configure(Map<String, ?> configs) {
 
         Object sslPrincipalMappingRules = configs.get(BrokerSecurityConfigs.SSL_PRINCIPAL_MAPPING_RULES_CONFIG);
         Object sslPrincipalMapper;
 
+
+        @SuppressWarnings("unchecked")
+        List<String> principalToLocalRules = (List<String>) configs.get(BrokerSecurityConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_CONFIG);
+        String defaultRealm;
+        Object kerberosShortNamer;
+
+
         try {
             Class<?> clazz = Class.forName("org.apache.kafka.common.security.ssl.SslPrincipalMapper");
             try {
@@ -120,6 +132,18 @@ public void configure(Map<String, ?> configs) {
 
             SET_PRINCIPAL_MAPPER.invoke(this, sslPrincipalMapper);
 
+            try {
+                defaultRealm = new KerberosPrincipal("tmp", 1).getRealm();
+            } catch (Exception ex) {
+                defaultRealm = "";
+            }
+
+            if (principalToLocalRules != null) {
+                kerberosShortNamer = KerberosShortNamer.fromUnparsedRules(defaultRealm, principalToLocalRules);
+                SET_KERBEROS_SHORT_NAMER.invoke(this, kerberosShortNamer);
+            }
+
+
         } catch (RuntimeException
                 | ClassNotFoundException
                 | NoSuchMethodException

testsuite/README.md

@@ -20,6 +20,7 @@ Then, you have to add some entries to your `/etc/hosts` file:
     127.0.0.1            hydra-jwt
     127.0.0.1            kafka
     127.0.0.1            mockoauth
+    127.0.0.1			 kerberos
 
 That's needed for host resolution, because Kafka brokers and Kafka clients connecting to Keycloak / Hydra have to use the 
 same hostname to ensure compatibility of generated access tokens.

testsuite/docker/kerberos/Dockerfile

@@ -0,0 +1,10 @@
+FROM ubuntu:22.04
+
+RUN DEBIAN_FRONTEND=noninteractive apt-get update -y && apt-get install -y krb5-kdc krb5-admin-server
+
+EXPOSE 88 749
+
+ADD ./config.sh /config.sh
+
+ENTRYPOINT ["/config.sh"]
+

testsuite/docker/kerberos/config.sh

@@ -0,0 +1,97 @@
+#!/bin/bash
+
+[[ "TRACE" ]] && set -x
+
+: ${REALM:=KERBEROS}
+: ${DOMAIN_REALM:=kerberos}
+: ${KERB_MASTER_KEY:=masterkey}
+: ${KERB_ADMIN_USER:=admin}
+: ${KERB_ADMIN_PASS:=admin}
+: ${KAFKA_USER:=kafka}
+: ${KAFKA_HOST:=kafka}
+: ${KAFKA_CLIENT_USER:=client}
+
+fix_nameserver() {
+  cat>/etc/resolv.conf<<EOF
+nameserver $NAMESERVER_IP
+search $SEARCH_DOMAINS
+EOF
+}
+
+fix_hostname() {
+  sed -i "/^hosts:/ s/ *files dns/ dns files/" /etc/nsswitch.conf
+}
+
+create_config() {
+  : ${KDC_ADDRESS:=$(hostname -f)}
+
+  cat>/etc/krb5.conf<<EOF
+[logging]
+ default = FILE:/var/log/kerberos/krb5libs.log
+ kdc = FILE:/var/log/kerberos/krb5kdc.log
+ admin_server = FILE:/var/log/kerberos/kadmind.log
+
+[libdefaults]
+ default_realm = $REALM
+ dns_lookup_realm = false
+ dns_lookup_kdc = false
+ ticket_lifetime = 24h
+ renew_lifetime = 7d
+ forwardable = true
+
+[realms]
+ $REALM = {
+  kdc = $KDC_ADDRESS
+  admin_server = $KDC_ADDRESS
+ }
+
+[domain_realm]
+ .$DOMAIN_REALM = $REALM
+ $DOMAIN_REALM = $REALM
+EOF
+}
+
+create_db() {
+  kdb5_util -P $KERB_MASTER_KEY -r $REALM create -s
+}
+
+start_kdc() {
+  service krb5-kdc start
+  service krb5-admin-server start
+}
+
+restart_kdc() {
+  service krb5-kdc restart
+  service krb5-admin-server restart
+}
+
+create_admin_user() {
+  kadmin.local -q "addprinc -pw $KERB_ADMIN_PASS $KERB_ADMIN_USER/admin"
+  echo "*/admin@$REALM *" > /etc/krb5kdc/kadm5.acl
+}
+
+create_kafka_user() {
+  kadmin.local -q "addprinc -randkey $KAFKA_HOST/$KAFKA_USER@$REALM"
+  kadmin.local -q "ktadd -k /keytabs/kafka_broker.keytab $KAFKA_HOST/$KAFKA_USER@$REALM"
+  kadmin.local -q "addprinc -randkey $KAFKA_HOST/$KAFKA_CLIENT_USER@$REALM"
+  kadmin.local -q "ktadd -k /keytabs/kafka_client.keytab $KAFKA_HOST/$KAFKA_CLIENT_USER@$REALM"
+  chmod 666 /keytabs/kafka_broker.keytab
+  chmod 666 /keytabs/kafka_client.keytab
+}
+
+
+
+if [ ! -f /kerberos_initialized ]; then
+  mkdir -p /var/log/kerberos
+  create_config
+  create_db
+  create_admin_user
+  create_kafka_user
+  start_kdc
+
+  touch /kerberos_initialized
+else
+  start_kdc
+fi
+
+tail -F /var/log/kerberos/krb5kdc.log

testsuite/docker/kerberos/kafka_server_jaas.conf

@@ -0,0 +1,7 @@
+KafkaServer {
+    com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true
+    storeKey=true
+    keyTab="/opt/kafka/keytabs/kafka_broker.keytab"
+    principal="kafka/kafka@KERBEROS";
+};

testsuite/docker/kerberos/krb5.conf

@@ -0,0 +1,21 @@
+[logging]
+ default = FILE:/var/log/kerberos/krb5libs.log
+ kdc = FILE:/var/log/kerberos/krb5kdc.log
+ admin_server = FILE:/var/log/kerberos/kadmind.log
+[libdefaults]
+ default_realm = KERBEROS
+ dns_lookup_realm = false
+ dns_lookup_kdc = false
+ ticket_lifetime = 24h
+ renew_lifetime = 7d
+ forwardable = true
+ rdns = false
+ ignore_acceptor_hostname = true
+[realms]
+ KERBEROS = {
+  kdc = kerberos
+  admin_server = kerberos
+ }
+[domain_realm]
+ .kerberos = KERBEROS
+ kerberos = KERBEROS

testsuite/keycloak-authz-zk-tests/src/test/java/io/strimzi/testsuite/oauth/authz/kraft/KeycloakZKAuthorizationTests.java

@@ -116,6 +116,7 @@ public void doTest() throws Exception {
 
         } catch (Throwable e) {
             log.error("Keycloak ZK Authorization Test failed: ", e);
+            e.printStackTrace();
             throw e;
         }
     }