[HOTFIX] Authenticate RemoteInterpreterEventServer callbacks

bin/interpreter.sh

@@ -146,7 +146,7 @@ if [[ -z "$ZEPPELIN_IMPERSONATE_CMD" ]]; then
         ZEPPELIN_IMPERSONATE_RUN_CMD=("ssh" "${ZEPPELIN_IMPERSONATE_USER}@localhost")
     fi
 else
-    ZEPPELIN_IMPERSONATE_RUN_CMD=$(eval "echo ${ZEPPELIN_IMPERSONATE_CMD} ")
+    ZEPPELIN_IMPERSONATE_RUN_CMD=("${ZEPPELIN_IMPERSONATE_CMD[@]}")
 fi
 
 
@@ -249,14 +249,19 @@ if [[ -n "$ZEPPELIN_IMPERSONATE_USER" ]]; then
     if [[ -f "${ZEPPELIN_CONF_DIR}/zeppelin-env.sh" ]]; then
         COMMAND_STRING+="source ${ZEPPELIN_CONF_DIR}/zeppelin-env.sh; "
     fi
+
+    # Pass the callback credential over stdin so it is neither exposed in the
+    # impersonation command line nor lost when ssh/sudo changes the environment.
+    COMMAND_STRING+="IFS= read -r ZEPPELIN_INTERPRETER_EVENT_TOKEN; export ZEPPELIN_INTERPRETER_EVENT_TOKEN; "
 
     # Add interpreter command to the command string
     IFS=' ' read -r -a JAVA_INTP_OPTS_ARRAY <<< "${JAVA_INTP_OPTS}"
     IFS=' ' read -r -a ZEPPELIN_INTP_MEM_ARRAY <<< "${ZEPPELIN_INTP_MEM}"
     COMMAND_STRING+="${ZEPPELIN_RUNNER} ${JAVA_INTP_OPTS_ARRAY[@]} ${ZEPPELIN_INTP_MEM_ARRAY[@]} -cp '${ZEPPELIN_INTP_CLASSPATH_OVERRIDES}:${ZEPPELIN_INTP_CLASSPATH}' ${ZEPPELIN_SERVER} ${CALLBACK_HOST} ${PORT} ${INTP_GROUP_ID} ${INTP_PORT}"
 
     # Set INTERPRETER_RUN_COMMAND with the impersonation command and command string
-    INTERPRETER_RUN_COMMAND=("${ZEPPELIN_IMPERSONATE_CMD[@]}" "${COMMAND_STRING}")
+    INTERPRETER_RUN_COMMAND=("${ZEPPELIN_IMPERSONATE_RUN_CMD[@]}" "${COMMAND_STRING}")
+    ZEPPELIN_INTERPRETER_EVENT_TOKEN_VIA_STDIN=true
   fi
 fi
 
@@ -280,5 +285,13 @@ fi
 # Don't remove this echo, it is for diagnose, this line of output will be redirected to java log4j output.
 # Output that starts with `[INFO]` will be redirected to log4j INFO output. Other outputs from interpreter.sh
 # will be redirected to log4j DEBUG output.
-echo "[INFO] Interpreter launch command: ${INTERPRETER_RUN_COMMAND[@]}"
-exec "${INTERPRETER_RUN_COMMAND[@]}"
+INTERPRETER_LOG_COMMAND="${INTERPRETER_RUN_COMMAND[*]}"
+if [[ -n "${ZEPPELIN_INTERPRETER_EVENT_TOKEN}" ]]; then
+  INTERPRETER_LOG_COMMAND="${INTERPRETER_LOG_COMMAND//${ZEPPELIN_INTERPRETER_EVENT_TOKEN}/[REDACTED]}"
+fi
+echo "[INFO] Interpreter launch command: ${INTERPRETER_LOG_COMMAND}"
+if [[ "${ZEPPELIN_INTERPRETER_EVENT_TOKEN_VIA_STDIN}" == "true" ]]; then
+  exec "${INTERPRETER_RUN_COMMAND[@]}" <<< "${ZEPPELIN_INTERPRETER_EVENT_TOKEN}"
+else
+  exec "${INTERPRETER_RUN_COMMAND[@]}"
+fi

helium-dev/src/test/java/org/apache/zeppelin/helium/ZeppelinDevServerTest.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *    http://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 org.apache.zeppelin.helium;
+
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ZeppelinDevServerTest {
+
+  @Test
+  void startsWithoutInterpreterCallbackCredentials() throws Exception {
+    int port = RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces();
+    ZeppelinDevServer server = new ZeppelinDevServer(port);
+    try {
+      server.start();
+      long deadline = System.currentTimeMillis() + 10_000;
+      while (!server.isRunning() && System.currentTimeMillis() < deadline) {
+        Thread.sleep(50);
+      }
+
+      assertTrue(server.isRunning());
+      assertTrue(RemoteInterpreterUtils.checkIfRemoteEndpointAccessible("localhost", port));
+      server.init(new HashMap<>());
+    } finally {
+      if (server.isRunning()) {
+        server.shutdown();
+      }
+      server.join(10_000);
+    }
+    assertFalse(server.isRunning());
+  }
+}

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java

@@ -27,6 +27,7 @@
 
 
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,8 +56,17 @@ public static void main(String[] args) {
       int port = Integer.parseInt(args[1]);
       String interpreter = args[2];
       String localRepoPath = args[3];
+      String interpreterGroupId = System.getenv("INTERPRETER_GROUP_ID");
+      String callbackToken = System.getenv(RemoteInterpreterEventClient.CALLBACK_TOKEN_ENV);
+      if (StringUtils.isAnyBlank(interpreterGroupId, callbackToken)) {
+        // Externally managed interpreters may be started before Zeppelin can provision their
+        // callback credential through init(). They must use a pre-populated local repository.
+        LOGGER.warn("Skip remote interpreter library synchronization because callback "
+            + "credentials are not available");
+        return;
+      }
       RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
-          zeppelinServerHost, port, 3);
+          zeppelinServerHost, port, 3, interpreterGroupId, callbackToken);
 
       RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
           intpEventClient, new File(localRepoPath));

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java

@@ -19,7 +19,9 @@
 import com.google.gson.Gson;
 import org.apache.thrift.protocol.TBinaryProtocol;
 import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.transport.TSaslClientTransport;
 import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
 import org.apache.thrift.transport.TTransportException;
 import org.apache.zeppelin.display.AngularObject;
 import org.apache.zeppelin.display.AngularObjectRegistryListener;
@@ -45,9 +47,22 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.RealmCallback;
+import javax.security.sasl.RealmChoiceCallback;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.List;
 import java.util.Map;
 
@@ -60,28 +75,89 @@ public class RemoteInterpreterEventClient implements ResourcePoolConnector,
   private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterEventClient.class);
   private static final Gson GSON = new Gson();
 
+  public static final String CALLBACK_TOKEN_ENV = "ZEPPELIN_INTERPRETER_EVENT_TOKEN";
+  public static final String CALLBACK_TOKEN_PROPERTY = "zeppelin.interpreter.event.token";
+  public static final String INTERPRETER_GROUP_PROPERTY = "zeppelin.interpreter.group.id";
+
+  public static final String SASL_MECHANISM = "DIGEST-MD5";
+  public static final String SASL_PROTOCOL = "zeppelin";
+  public static final String SASL_SERVER_NAME = "interpreter-event";
+  public static final Map<String, String> SASL_PROPERTIES = Map.of(
+      Sasl.QOP, "auth-int",
+      Sasl.SERVER_AUTH, "true");
+
   private PooledRemoteClient<RemoteInterpreterEventService.Client> remoteClient;
   private String intpGroupId;
 
-  public RemoteInterpreterEventClient(String intpEventHost, int intpEventPort, int connectionPoolSize) {
+  public RemoteInterpreterEventClient(String intpEventHost,
+                                      int intpEventPort,
+                                      int connectionPoolSize,
+                                      String intpGroupId,
+                                      String callbackToken) {
+    if (intpGroupId == null || intpGroupId.isEmpty()) {
+      throw new IllegalArgumentException("Interpreter group id must not be empty");
+    }
+    if (callbackToken == null || callbackToken.isEmpty()) {
+      throw new IllegalArgumentException("Interpreter event callback token must not be empty");
+    }
+    this.intpGroupId = intpGroupId;
+    String authenticationId = callbackAuthenticationId(intpGroupId, callbackToken);
     this.remoteClient = new PooledRemoteClient<>(() -> {
-      TSocket transport = new TSocket(intpEventHost, intpEventPort);
+      TSocket socket = new TSocket(intpEventHost, intpEventPort);
+      TTransport transport;
       try {
+        transport = new TSaslClientTransport(
+            SASL_MECHANISM,
+            authenticationId,
+            SASL_PROTOCOL,
+            SASL_SERVER_NAME,
+            SASL_PROPERTIES,
+            clientCallbackHandler(authenticationId, callbackToken),
+            socket);
         transport.open();
-      } catch (TTransportException e) {
+      } catch (SaslException | TTransportException e) {
         throw new IOException(e);
       }
       TProtocol protocol = new TBinaryProtocol(transport);
       return new RemoteInterpreterEventService.Client(protocol);
     }, connectionPoolSize);
   }
 
-  public <R> R callRemoteFunction(PooledRemoteClient.RemoteFunction<R, RemoteInterpreterEventService.Client> func) {
-    return remoteClient.callRemoteFunction(func);
+  public static String callbackAuthenticationId(String intpGroupId, String callbackToken) {
+    try {
+      MessageDigest digest = MessageDigest.getInstance("SHA-256");
+      digest.update(intpGroupId.getBytes(StandardCharsets.UTF_8));
+      digest.update((byte) 0);
+      byte[] authenticationId = digest.digest(callbackToken.getBytes(StandardCharsets.UTF_8));
+      return Base64.getUrlEncoder().withoutPadding().encodeToString(authenticationId);
+    } catch (NoSuchAlgorithmException e) {
+      throw new IllegalStateException("SHA-256 is unavailable", e);
+    }
   }
 
-  public void setIntpGroupId(String intpGroupId) {
-    this.intpGroupId = intpGroupId;
+  private static CallbackHandler clientCallbackHandler(String authenticationId,
+                                                       String callbackToken) {
+    return callbacks -> {
+      for (Callback callback : callbacks) {
+        if (callback instanceof NameCallback) {
+          ((NameCallback) callback).setName(authenticationId);
+        } else if (callback instanceof PasswordCallback) {
+          ((PasswordCallback) callback).setPassword(callbackToken.toCharArray());
+        } else if (callback instanceof RealmCallback) {
+          RealmCallback realmCallback = (RealmCallback) callback;
+          realmCallback.setText(realmCallback.getDefaultText());
+        } else if (callback instanceof RealmChoiceCallback) {
+          ((RealmChoiceCallback) callback).setSelectedIndex(0);
+        } else {
+          throw new UnsupportedCallbackException(callback);
+        }
+      }
+    };
+  }
+
+  public <R> R callRemoteFunction(
+      PooledRemoteClient.RemoteFunction<R, RemoteInterpreterEventService.Client> func) {
+    return remoteClient.callRemoteFunction(func);
   }
 
   public void registerInterpreterProcess(RegisterInfo registerInfo) {