diff --git a/shell/src/main/java/org/apache/zeppelin/shell/TerminalInterpreter.java b/shell/src/main/java/org/apache/zeppelin/shell/TerminalInterpreter.java index 173ccba2936..1ce5b7d1452 100644 --- a/shell/src/main/java/org/apache/zeppelin/shell/TerminalInterpreter.java +++ b/shell/src/main/java/org/apache/zeppelin/shell/TerminalInterpreter.java @@ -42,6 +42,7 @@ import java.io.IOException; import java.net.InetAddress; import java.net.URL; +import java.security.SecureRandom; import java.util.Date; import java.util.HashMap; import java.util.List; @@ -65,6 +66,11 @@ public TerminalInterpreter(Properties property) { private int terminalPort = 0; private String terminalHostIp; + // Per-terminal-server secret every websocket client must present; it reaches + // legitimate clients only through the paragraph result, i.e. through + // Zeppelin's own note ACLs. + private String terminalToken; + // Internal and external IP mapping of zeppelin server private HashMap mapIpMapping = new HashMap<>(); private Gson gson = new Gson(); @@ -114,7 +120,8 @@ public InterpreterResult internalInterpret(String cmd, InterpreterContext contex LOGGER.info("Terminal host IP: " + terminalHostIp); LOGGER.info("Terminal port: " + terminalPort); String allowedOrigin = generateOrigin(terminalHostIp, terminalPort); - terminalThread = new TerminalThread(terminalPort, allowedOrigin); + terminalToken = generateAuthToken(); + terminalThread = new TerminalThread(terminalPort, allowedOrigin, terminalToken); terminalThread.start(); } catch (IOException e) { LOGGER.error(e.getMessage(), e); @@ -170,7 +177,8 @@ public void createTerminalDashboard(String noteId, String paragraphId, String ho HashMap jinjaParams = new HashMap(); Date now = new Date(); String terminalServerUrl = generateOrigin(hostIp, port) + - "?noteId=" + noteId + "¶graphId=" + paragraphId + "&t=" + now.getTime(); + "?noteId=" + noteId + "¶graphId=" + paragraphId + "&t=" + now.getTime() + + "&token=" + terminalToken; jinjaParams.put("HOST_NAME", hostName); jinjaParams.put("HOST_IP", hostIp); jinjaParams.put("TERMINAL_SERVER_URL", terminalServerUrl); @@ -191,6 +199,19 @@ private String generateOrigin(String hostIp, int port) { return "http://" + hostIp + ":" + port; } + // The terminal websocket exposes an OS shell, so it must be gated by a + // secret nobody can guess: 256 bits from a CSPRNG, hex encoded. + private static String generateAuthToken() { + SecureRandom random = new SecureRandom(); + byte[] bytes = new byte[32]; + random.nextBytes(bytes); + StringBuilder sb = new StringBuilder(bytes.length * 2); + for (byte b : bytes) { + sb.append(String.format("%02x", b)); + } + return sb.toString(); + } + @Override public void cancel(InterpreterContext context) { } @@ -251,6 +272,11 @@ public String getTerminalHostIp() { return terminalHostIp; } + @VisibleForTesting + public String getTerminalToken() { + return terminalToken; + } + @VisibleForTesting public boolean terminalThreadIsRunning() { return terminalThread.isRunning(); diff --git a/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalManager.java b/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalManager.java index 089ab26dcff..95ac46e6b39 100644 --- a/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalManager.java +++ b/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalManager.java @@ -71,8 +71,7 @@ public void removeTerminalService(TerminalSocket terminalSocket) { if (terminalSocket2Service.containsKey(terminalSocketHashcode)) { terminalSocket2Service.remove(terminalSocketHashcode); } else { - LOGGER.error("Can't find TerminalSocket: " + terminalSocketHashcode); - LOGGER.error(terminalSocket2Service.toString()); + LOGGER.error("Can't find TerminalSocket: {}", terminalSocketHashcode); } } @@ -81,7 +80,7 @@ public void cleanIntpContext(String noteId) { for (Map.Entry entity : noteParagraphId2IntpContext.entrySet()) { String key = entity.getKey(); if (key.contains(keyPrefix)) { - LOGGER.info("cleanIntpContext : " + key); + LOGGER.info("cleanIntpContext : {}", key); noteParagraphId2IntpContext.remove(key); } } @@ -109,8 +108,7 @@ public void onWebSocketConnect(String noteId, String paragraphId) { intpContext.getAngularObjectRegistry().add(TERMINAL_SOCKET_STATUS, TERMINAL_SOCKET_CONNECT, intpContext.getNoteId(), intpContext.getParagraphId()); } else { - LOGGER.error("Can't find InterpreterContext from : " + id); - LOGGER.error(noteParagraphId2IntpContext.toString()); + LOGGER.error("Can't find InterpreterContext from : {}", id); } } @@ -121,8 +119,7 @@ public void onWebSocketClose(TerminalSocket terminalSocket, String noteId, Strin intpContext.getAngularObjectRegistry().add(TERMINAL_SOCKET_STATUS, TERMINAL_SOCKET_CLOSE, intpContext.getNoteId(), intpContext.getParagraphId()); } else { - LOGGER.error("Can't find InterpreterContext from : " + id); - LOGGER.error(noteParagraphId2IntpContext.toString()); + LOGGER.error("Can't find InterpreterContext from : {}", id); } removeTerminalService(terminalSocket); @@ -136,8 +133,7 @@ public void onWebSocketError(TerminalSocket terminalSocket, String noteId, Strin intpContext.getAngularObjectRegistry().add(TERMINAL_SOCKET_STATUS, TERMINAL_SOCKET_ERROR, intpContext.getNoteId(), intpContext.getParagraphId()); } else { - LOGGER.error("Can't find InterpreterContext from : " + id); - LOGGER.error(noteParagraphId2IntpContext.toString()); + LOGGER.error("Can't find InterpreterContext from : {}", id); } } diff --git a/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalThread.java b/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalThread.java index a2a10866773..28877eea50a 100644 --- a/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalThread.java +++ b/shell/src/main/java/org/apache/zeppelin/shell/terminal/TerminalThread.java @@ -41,10 +41,12 @@ public class TerminalThread extends Thread { private int port = 0; private String allowedOrigin; + private String authToken; - public TerminalThread(int port, String allowedOrigin) { + public TerminalThread(int port, String allowedOrigin, String authToken) { this.port = port; this.allowedOrigin = allowedOrigin; + this.authToken = authToken; } @Override @@ -77,11 +79,17 @@ public void run() { try { JakartaWebSocketServletContainerInitializer.configure(context, - (servletContext, container) -> - container.addEndpoint( + (servletContext, container) -> { + ServerEndpointConfig endpointConfig = ServerEndpointConfig.Builder.create(TerminalSocket.class, "/") - .configurator(new TerminalSessionConfigurator(allowedOrigin)) - .build())); + .configurator(new TerminalSessionConfigurator(allowedOrigin)) + .build(); + // TerminalSocket verifies this secret on every connection before + // exposing the shell; the Origin check alone is forgeable. + endpointConfig.getUserProperties() + .put(TerminalSocket.AUTH_TOKEN_PROPERTY, authToken); + container.addEndpoint(endpointConfig); + }); jettyServer.start(); jettyServer.join(); } catch (Exception e) { diff --git a/shell/src/main/java/org/apache/zeppelin/shell/terminal/websocket/TerminalSocket.java b/shell/src/main/java/org/apache/zeppelin/shell/terminal/websocket/TerminalSocket.java index 16af219d7af..9a9a7f497ec 100644 --- a/shell/src/main/java/org/apache/zeppelin/shell/terminal/websocket/TerminalSocket.java +++ b/shell/src/main/java/org/apache/zeppelin/shell/terminal/websocket/TerminalSocket.java @@ -26,36 +26,66 @@ import jakarta.websocket.ClientEndpoint; import jakarta.websocket.CloseReason; +import jakarta.websocket.EndpointConfig; import jakarta.websocket.OnClose; import jakarta.websocket.OnError; import jakarta.websocket.OnMessage; import jakarta.websocket.OnOpen; import jakarta.websocket.Session; import jakarta.websocket.server.ServerEndpoint; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.util.List; import java.util.Map; @ClientEndpoint @ServerEndpoint(value = "/") public class TerminalSocket { private static final Logger LOGGER = LoggerFactory.getLogger(TerminalSocket.class); + + // Key under which TerminalThread publishes the per-server auth token + public static final String AUTH_TOKEN_PROPERTY = "zeppelin.terminal.auth.token"; + private TerminalService terminalService; private TerminalManager terminalManager = TerminalManager.getInstance(); private String noteId; private String paragraphId; + private volatile boolean authorized = false; public TerminalSocket() { terminalService = terminalManager.addTerminalService(this); } @OnOpen - public void onWebSocketConnect(Session sess) { - LOGGER.info("Socket Connected: {}", sess); + public void onWebSocketConnect(Session sess, EndpointConfig config) { + // This endpoint hands out an OS shell: require the per-server secret that + // only reaches clients through the paragraph result (note ACLs). The + // Origin check is not authentication - any non-browser client forges it. + String expectedToken = (String) config.getUserProperties().get(AUTH_TOKEN_PROPERTY); + if (!isTokenValid(expectedToken, sess.getRequestParameterMap().get("token"))) { + LOGGER.warn("Rejecting terminal websocket connection without a valid auth token: {}", + sess.getId()); + try { + sess.close(new CloseReason(CloseReason.CloseCodes.VIOLATED_POLICY, + "Missing or invalid terminal auth token")); + } catch (IOException e) { + LOGGER.error(e.getMessage(), e); + } + return; + } + authorized = true; + LOGGER.info("Socket Connected: {}", sess.getId()); terminalService.onWebSocketConnect(sess); } @OnMessage public void onWebSocketText(String message) { + if (!authorized) { + LOGGER.warn("Ignoring message from unauthorized terminal websocket connection"); + return; + } if (LOGGER.isDebugEnabled()) { LOGGER.debug("Received TEXT message: {}", message); } @@ -86,15 +116,29 @@ public void onWebSocketText(String message) { @OnClose public void onWebSocketClose(CloseReason reason) { LOGGER.info("Socket Closed: {}", reason); - - terminalManager.onWebSocketClose(this, noteId, paragraphId); + if (authorized && noteId != null && paragraphId != null) { + terminalManager.onWebSocketClose(this, noteId, paragraphId); + } else { + terminalManager.removeTerminalService(this); + } } @OnError public void onWebSocketError(Throwable cause) { LOGGER.warn(cause.getMessage(), cause); + if (authorized && noteId != null && paragraphId != null) { + terminalManager.onWebSocketError(this, noteId, paragraphId); + } + } - terminalManager.onWebSocketError(this, noteId, paragraphId); + private static boolean isTokenValid(String expectedToken, List suppliedTokens) { + if (expectedToken == null || expectedToken.isEmpty() + || suppliedTokens == null || suppliedTokens.isEmpty()) { + return false; + } + return MessageDigest.isEqual( + expectedToken.getBytes(StandardCharsets.UTF_8), + suppliedTokens.get(0).getBytes(StandardCharsets.UTF_8)); } private Map getMessageMap(String message) { diff --git a/shell/src/main/resources/html/js/index.js b/shell/src/main/resources/html/js/index.js index a1a96b8ad78..a22bc2b3c2c 100644 --- a/shell/src/main/resources/html/js/index.js +++ b/shell/src/main/resources/html/js/index.js @@ -12,8 +12,9 @@ var noteId = getParams("noteId"); var paragraphId = getParams("paragraphId"); +var token = getParams("token"); -var ws = new WebSocket("ws://" + location.host + "/terminal/"); +var ws = new WebSocket("ws://" + location.host + "/terminal/?token=" + encodeURIComponent(token)); ws.onopen = () => { // alert("ws.onopen"); diff --git a/shell/src/test/java/org/apache/zeppelin/shell/TerminalInterpreterTest.java b/shell/src/test/java/org/apache/zeppelin/shell/TerminalInterpreterTest.java index b4f78b471db..646029a522f 100644 --- a/shell/src/test/java/org/apache/zeppelin/shell/TerminalInterpreterTest.java +++ b/shell/src/test/java/org/apache/zeppelin/shell/TerminalInterpreterTest.java @@ -42,6 +42,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; import java.io.IOException; import java.net.URI; @@ -89,7 +90,7 @@ void testInvalidCommand() { assertTrue(running); URI webSocketConnectionUri = URI.create("ws://" + terminal.getTerminalHostIp() + - ":" + terminal.getTerminalPort() + "/terminal/"); + ":" + terminal.getTerminalPort() + "/terminal/?token=" + terminal.getTerminalToken()); LOGGER.info("webSocketConnectionUri: " + webSocketConnectionUri); String origin = "http://" + terminal.getTerminalHostIp() + ":" + terminal.getTerminalPort(); LOGGER.info("origin: " + origin); @@ -175,7 +176,7 @@ void testValidCommand() { assertTrue(running); URI webSocketConnectionUri = URI.create("ws://" + terminal.getTerminalHostIp() + - ":" + terminal.getTerminalPort() + "/terminal/"); + ":" + terminal.getTerminalPort() + "/terminal/?token=" + terminal.getTerminalToken()); LOGGER.info("webSocketConnectionUri: " + webSocketConnectionUri); String origin = "http://" + terminal.getTerminalHostIp() + ":" + terminal.getTerminalPort(); LOGGER.info("origin: " + origin); @@ -258,7 +259,7 @@ void testValidOrigin() { assertTrue(running); URI webSocketConnectionUri = URI.create("ws://" + terminal.getTerminalHostIp() + - ":" + terminal.getTerminalPort() + "/terminal/"); + ":" + terminal.getTerminalPort() + "/terminal/?token=" + terminal.getTerminalToken()); LOGGER.info("webSocketConnectionUri: " + webSocketConnectionUri); String origin = "http://" + terminal.getTerminalHostIp() + ":" + terminal.getTerminalPort(); LOGGER.info("origin: " + origin); @@ -308,7 +309,7 @@ void testInvalidOrigin() { assertTrue(running); URI webSocketConnectionUri = URI.create("ws://" + terminal.getTerminalHostIp() + - ":" + terminal.getTerminalPort() + "/terminal/"); + ":" + terminal.getTerminalPort() + "/terminal/?token=" + terminal.getTerminalToken()); LOGGER.info("webSocketConnectionUri: " + webSocketConnectionUri); String origin = "http://invalid-origin"; LOGGER.info("origin: " + origin); @@ -350,6 +351,64 @@ void testInvalidOrigin() { assertTrue(exception.getMessage().contains("403 Forbidden")); } + @Test + void testMissingTokenRejected() { + Session session = null; + WebSocketContainer webSocketContainer = null; + + // mock connect terminal + boolean running = terminal.terminalThreadIsRunning(); + assertTrue(running); + + // valid Origin, but no per-server auth token in the query string + URI webSocketConnectionUri = URI.create("ws://" + terminal.getTerminalHostIp() + + ":" + terminal.getTerminalPort() + "/terminal/"); + LOGGER.info("webSocketConnectionUri: " + webSocketConnectionUri); + String origin = "http://" + terminal.getTerminalHostIp() + ":" + terminal.getTerminalPort(); + LOGGER.info("origin: " + origin); + ClientEndpointConfig clientEndpointConfig = getOriginRequestHeaderConfig(origin); + webSocketContainer = ContainerProvider.getWebSocketContainer(); + + try { + // Attempt Connect + session = webSocketContainer.connectToServer( + TerminalSocketTest.class, clientEndpointConfig, webSocketConnectionUri); + // the server must close the connection instead of exposing the shell + boolean closed = false; + for (int i = 0; i < 30; i++) { + if (!session.isOpen()) { + closed = true; + break; + } + Thread.sleep(100); + } + assertTrue(closed, "WebSocket connection without auth token should be closed by the server"); + } catch (DeploymentException | IOException e) { + // a handshake rejected by the server is an acceptable outcome as well + LOGGER.info("Terminal connection without token rejected during handshake: " + e.getMessage()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + fail("Test interrupted while waiting for session closure: " + e.getMessage()); + } finally { + if (session != null) { + try { + session.close(); + } catch (IOException e) { + LOGGER.error(e.getMessage(), e); + } + } + + // Force lifecycle stop when done with container. + if (webSocketContainer instanceof LifeCycle) { + try { + ((LifeCycle) webSocketContainer).stop(); + } catch (Exception e) { + LOGGER.error(e.getMessage(), e); + } + } + } + } + private static ClientEndpointConfig getOriginRequestHeaderConfig(String origin) { Configurator configurator = new Configurator() { @Override