Support Sasl for GraalVM (#375)

build.gradle

@@ -10,6 +10,11 @@ buildscript {
 
 
 subprojects { Project subproject ->
+
+    if (subproject.path.contains("tests")) {
+        return
+    }
+
     group "io.micronaut.kafka"
 
     apply plugin: "io.micronaut.build.internal.common"

gradle.properties

@@ -2,10 +2,11 @@ projectVersion=3.3.3-SNAPSHOT
 micronautDocsVersion=2.0.0.RC1
 micronautBuildVersion=1.1.5
 micronautVersion=2.4.2
+micronautGradlePluginVersion=1.5.0
 groovyVersion=3.0.5
 spockVersion=2.0-M3-groovy-3.0
-
 kafkaVersion=2.8.0
+testContainersVersion=1.15.3
 
 title=Micronaut Kafka
 projectDesc=Integration between Micronaut and Kafka Messaging

kafka-streams/build.gradle

@@ -20,8 +20,8 @@ dependencies {
     testImplementation "org.apache.kafka:kafka-clients:${kafkaVersion}:test"
     testImplementation "org.apache.kafka:kafka_2.12:${kafkaVersion}"
     testImplementation "org.apache.kafka:kafka_2.12:${kafkaVersion}:test"
-    testImplementation "org.testcontainers:kafka:1.15.3" //Breaking change in java container after 1.12 see https://github.com/docker-java/docker-java/pull/1392
-    testImplementation "org.testcontainers:spock:1.15.3"
+    testImplementation "org.testcontainers:kafka:$testContainersVersion"
+    testImplementation "org.testcontainers:spock:$testContainersVersion"
 
     testRuntimeOnly "io.micronaut:micronaut-runtime"
     testRuntimeOnly "io.micronaut:micronaut-http-server-netty"

kafka/build.gradle

@@ -27,7 +27,7 @@ dependencies {
     testImplementation "org.apache.kafka:kafka_2.12:${kafkaVersion}"
     testImplementation "org.apache.kafka:kafka_2.12:${kafkaVersion}:test"
 
-    testImplementation "org.testcontainers:kafka:1.15.3"
+    testImplementation "org.testcontainers:kafka:$testContainersVersion"
     testImplementation 'io.opentracing.contrib:opentracing-kafka-client:0.1.15'
     testImplementation 'io.opentracing:opentracing-mock:0.33.0'
 

kafka/src/main/java/io/micronaut/configuration/kafka/graal/ByteBufferUnmapperSubstitute.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2017-2021 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.configuration.kafka.graal;
+
+import com.oracle.svm.core.annotate.Substitute;
+import com.oracle.svm.core.annotate.TargetClass;
+import org.apache.kafka.common.utils.ByteBufferUnmapper;
+import org.apache.kafka.common.utils.Java;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+
+@TargetClass(value = ByteBufferUnmapper.class)
+@SuppressWarnings("MissingJavadocType")
+public final class ByteBufferUnmapperSubstitute {
+
+    @Substitute
+    public static void unmap(String resourceDescription, ByteBuffer buffer) throws IOException {
+        if (!buffer.isDirect()) {
+            throw new IllegalArgumentException("Unmapping only works with direct buffers");
+        }
+
+        try {
+            if (Java.IS_JAVA9_COMPATIBLE) {
+                Class<?> unsafeClass = Class.forName("sun.misc.Unsafe");
+                Method m = unsafeClass.getMethod("cleaner", void.class, ByteBuffer.class);
+                Field f = unsafeClass.getDeclaredField("theUnsafe");
+                f.setAccessible(true);
+                Object theUnsafe = f.get(null);
+                m.invoke(theUnsafe, buffer);
+            } else {
+                Class<?> directBufferClass = Class.forName("java.nio.DirectByteBuffer");
+                Method cleanerMethod = directBufferClass.getMethod("cleaner");
+                cleanerMethod.setAccessible(true);
+                Object cleaner = cleanerMethod.invoke(buffer);
+                if (cleaner != null) {
+                    Class<?> cleanerClass = Class.forName("sun.misc.Cleaner");
+                    Method cleanMethod = cleanerClass.getMethod("clean");
+                    cleanMethod.setAccessible(true);
+                    cleanMethod.invoke(cleaner);
+                }
+            }
+        } catch (Throwable throwable) {
+            throw new IOException("Unable to unmap the mapped buffer: " + resourceDescription, throwable);
+        }
+    }
+
+}

kafka/src/main/java/io/micronaut/configuration/kafka/graal/SaslClientCallbackHandlerSubstitute.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright 2017-2020 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.configuration.kafka.graal;
+
+import com.oracle.svm.core.annotate.Alias;
+import com.oracle.svm.core.annotate.Inject;
+import com.oracle.svm.core.annotate.Substitute;
+import com.oracle.svm.core.annotate.TargetClass;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
+import org.apache.kafka.common.security.auth.SaslExtensions;
+import org.apache.kafka.common.security.auth.SaslExtensionsCallback;
+import org.apache.kafka.common.security.authenticator.LoginManager;
+import org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler;
+import org.apache.kafka.common.security.scram.ScramExtensionsCallback;
+import org.apache.kafka.common.security.scram.internals.ScramMechanism;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.auth.Subject;
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.spi.LoginModule;
+import javax.security.sasl.AuthorizeCallback;
+import javax.security.sasl.RealmCallback;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+// Copy from: https://github.com/oracle/helidon/blob/785ce38ca06b268d16e387fb6498aaaa890695cc/messaging/kafka/src/main/java/io/helidon/messaging/connectors/kafka/SaslClientCallbackHandlerSubstitution.java
+// Workaround for https://github.com/oracle/graal/issues/2745
+@TargetClass(value = SaslClientCallbackHandler.class)
+@SuppressWarnings("MissingJavadocType")
+public final class SaslClientCallbackHandlerSubstitute implements AuthenticateCallbackHandler {
+
+    @Alias
+    private String mechanism;
+
+    @Inject
+    private Logger logger;
+
+    @Inject
+    private Subject subject;
+
+    @Substitute
+    public SaslClientCallbackHandlerSubstitute() {
+        logger = LoggerFactory.getLogger(LoginManager.class);
+    }
+
+    @Override
+    @Substitute
+    public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
+        this.mechanism = saslMechanism;
+        this.subject = null;
+
+        int entrySize = jaasConfigEntries.size();
+        if (entrySize == 0) {
+            logger.warn("Missing JAAS config entry, missing or malformed sasl.jaas.config property.");
+            return;
+        } else if (entrySize > 1) {
+            logger.warn("Multiple JAAS config entries, Kafka client's sasl.jaas.config can have only one JAAS config entry.");
+            return;
+        }
+
+        AppConfigurationEntry jaasConfigEntry = jaasConfigEntries.get(0);
+        String jaasLoginModuleName = jaasConfigEntry.getLoginModuleName();
+        subject = new Subject();
+
+        try {
+            Class.forName(jaasLoginModuleName)
+                    .asSubclass(LoginModule.class)
+                    .getDeclaredConstructor()
+                    .newInstance()
+                    .initialize(subject, this, new HashMap<>(), jaasConfigEntry.getOptions());
+        } catch (ReflectiveOperationException e) {
+            throw new KafkaException("Can't instantiate JAAS login module" + jaasLoginModuleName, e);
+        }
+    }
+
+    @Override
+    @Substitute
+    public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
+        // Subject.getSubject doesn't return proper subject in native image
+        // Remove substitution when https://github.com/oracle/graal/issues/2745 is fixed
+        // Subject subject = Subject.getSubject(AccessController.getContext());
+
+        for (Callback callback : callbacks) {
+            if (callback instanceof NameCallback) {
+                NameCallback nc = (NameCallback) callback;
+                if (subject != null && !subject.getPublicCredentials(String.class).isEmpty()) {
+                    nc.setName(subject.getPublicCredentials(String.class).iterator().next());
+                } else {
+                    nc.setName(nc.getDefaultName());
+                }
+            } else if (callback instanceof PasswordCallback) {
+                if (subject != null && !subject.getPrivateCredentials(String.class).isEmpty()) {
+                    char[] password = subject.getPrivateCredentials(String.class).iterator().next().toCharArray();
+                    ((PasswordCallback) callback).setPassword(password);
+                } else {
+                    String errorMessage = "Could not login: the client is being asked for a password, but the Kafka"
+                            + " client code does not currently support obtaining a password from the user.";
+                    throw new UnsupportedCallbackException(callback, errorMessage);
+                }
+            } else if (callback instanceof RealmCallback) {
+                RealmCallback rc = (RealmCallback) callback;
+                rc.setText(rc.getDefaultText());
+            } else if (callback instanceof AuthorizeCallback) {
+                AuthorizeCallback ac = (AuthorizeCallback) callback;
+                String authId = ac.getAuthenticationID();
+                String authzId = ac.getAuthorizationID();
+                ac.setAuthorized(authId.equals(authzId));
+                if (ac.isAuthorized()) {
+                    ac.setAuthorizedID(authzId);
+                }
+            } else if (callback instanceof ScramExtensionsCallback) {
+                if (ScramMechanism.isScram(mechanism) && subject != null && !subject.getPublicCredentials(Map.class).isEmpty()) {
+                    @SuppressWarnings("unchecked")
+                    Map<String, String> extensions =
+                            (Map<String, String>) subject.getPublicCredentials(Map.class).iterator().next();
+                    ((ScramExtensionsCallback) callback).extensions(extensions);
+                }
+            } else if (callback instanceof SaslExtensionsCallback) {
+                if (!SaslConfigs.GSSAPI_MECHANISM.equals(mechanism)
+                        && subject != null && !subject.getPublicCredentials(SaslExtensions.class).isEmpty()) {
+                    SaslExtensions extensions = subject.getPublicCredentials(SaslExtensions.class).iterator().next();
+                    ((SaslExtensionsCallback) callback).extensions(extensions);
+                }
+            } else {
+                throw new UnsupportedCallbackException(callback, "Unrecognized SASL ClientCallback");
+            }
+        }
+    }
+
+    @Override
+    @Substitute
+    public void close() {
+    }
+}