intpProperties;
@@ -48,6 +56,9 @@ public class ZSession {
private SessionInfo sessionInfo;
private ZeppelinWebSocketClient webSocketClient;
+ private boolean closed;
+ private boolean remoteSessionStopped;
+ private boolean webSocketStopped;
public ZSession(ClientConfig clientConfig,
String interpreter) throws Exception {
@@ -120,14 +131,7 @@ public void start(MessageHandler messageHandler) throws Exception {
this.sessionInfo = zeppelinClient.getSession(getSessionId());
if (messageHandler != null) {
- this.webSocketClient = new ZeppelinWebSocketClient(messageHandler);
- this.webSocketClient.connect(zeppelinClient.getClientConfig().getZeppelinRestUrl()
- .replace("https", "ws").replace("http", "ws") + "/ws");
-
- // call GET_NOTE to establish websocket connection between this session and zeppelin-server
- Message msg = new Message(Message.OP.GET_NOTE);
- msg.put("id", getNoteId());
- this.webSocketClient.send(msg);
+ connectWebSocket(messageHandler);
}
}
@@ -136,13 +140,49 @@ public void start(MessageHandler messageHandler) throws Exception {
*
* @throws Exception
*/
- public void stop() throws Exception {
- if (getSessionId() != null) {
- zeppelinClient.stopSession(getSessionId());
+ public synchronized void stop() throws Exception {
+ if (closed) {
+ return;
}
- if (webSocketClient != null) {
- webSocketClient.stop();
+ Exception failure = null;
+ if (!remoteSessionStopped) {
+ if (getSessionId() == null) {
+ remoteSessionStopped = true;
+ } else {
+ try {
+ zeppelinClient.stopSession(getSessionId());
+ remoteSessionStopped = true;
+ } catch (Exception e) {
+ failure = e;
+ }
+ }
}
+ if (!webSocketStopped) {
+ if (webSocketClient == null) {
+ webSocketStopped = true;
+ } else {
+ try {
+ webSocketClient.stop();
+ webSocketStopped = true;
+ } catch (Exception e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ zeppelinClient.close();
+ closed = true;
+ }
+
+ @Override
+ public void close() throws Exception {
+ stop();
}
/**
@@ -167,9 +207,46 @@ public static ZSession createFromExistingSession(ClientConfig clientConfig,
String interpreter,
String sessionId,
MessageHandler messageHandler) throws Exception {
+ return createFromExistingSession(
+ clientConfig, interpreter, sessionId, client -> { }, messageHandler);
+ }
+
+ /**
+ * Reconnect an existing session after authenticating its isolated REST/WebSocket client.
+ *
+ * For Shiro, the authenticator can call {@code client.login(user, password)}. It is invoked
+ * before both the protected session lookup and WebSocket upgrade.
+ */
+ public static ZSession createFromExistingSession(
+ ClientConfig clientConfig,
+ String interpreter,
+ String sessionId,
+ ClientAuthenticator authenticator,
+ MessageHandler messageHandler) throws Exception {
ZSession session = new ZSession(clientConfig, interpreter, sessionId);
- session.reconnect(messageHandler);
- return session;
+ try {
+ authenticator.authenticate(session.zeppelinClient);
+ session.reconnect(messageHandler);
+ return session;
+ } catch (Exception e) {
+ try {
+ session.closeLocalResources();
+ } catch (Exception cleanupFailure) {
+ e.addSuppressed(cleanupFailure);
+ }
+ throw e;
+ }
+ }
+
+ private void closeLocalResources() throws Exception {
+ try {
+ if (webSocketClient != null) {
+ webSocketClient.stop();
+ }
+ } finally {
+ zeppelinClient.close();
+ closed = true;
+ }
}
private void reconnect(MessageHandler messageHandler) throws Exception {
@@ -179,15 +256,43 @@ private void reconnect(MessageHandler messageHandler) throws Exception {
}
if (messageHandler != null) {
- this.webSocketClient = new ZeppelinWebSocketClient(messageHandler);
- this.webSocketClient.connect(zeppelinClient.getClientConfig().getZeppelinRestUrl()
- .replace("https", "ws").replace("http", "ws") + "/ws");
-
- // call GET_NOTE to establish websocket connection between this session and zeppelin-server
- Message msg = new Message(Message.OP.GET_NOTE);
- msg.put("id", getNoteId());
- this.webSocketClient.send(msg);
+ connectWebSocket(messageHandler);
+ }
+ }
+
+ private void connectWebSocket(MessageHandler messageHandler) throws Exception {
+ this.webSocketClient = new ZeppelinWebSocketClient(messageHandler);
+ URI webSocketUri = toWebSocketUri(zeppelinClient.getClientConfig().getZeppelinRestUrl());
+ this.webSocketClient.connect(webSocketUri.toString(),
+ zeppelinClient.getSessionCookieHeader(webSocketUri));
+
+ // call GET_NOTE to establish websocket connection between this session and zeppelin-server
+ Message msg = new Message(Message.OP.GET_NOTE);
+ msg.put("id", getNoteId());
+ this.webSocketClient.send(msg);
+ }
+
+ static URI toWebSocketUri(String zeppelinRestUrl) throws URISyntaxException {
+ URI restUri = new URI(zeppelinRestUrl);
+ String webSocketScheme;
+ if ("https".equalsIgnoreCase(restUri.getScheme())) {
+ webSocketScheme = "wss";
+ } else if ("http".equalsIgnoreCase(restUri.getScheme())) {
+ webSocketScheme = "ws";
+ } else {
+ throw new IllegalArgumentException("Unsupported Zeppelin REST URL scheme: "
+ + restUri.getScheme());
+ }
+ if (restUri.getRawAuthority() == null) {
+ throw new IllegalArgumentException("Zeppelin REST URL must contain an authority");
+ }
+ String rawBasePath = restUri.getRawPath();
+ if (rawBasePath != null && rawBasePath.length() > 1 && rawBasePath.endsWith("/")) {
+ rawBasePath = rawBasePath.substring(0, rawBasePath.length() - 1);
}
+ String webSocketPath = rawBasePath == null || rawBasePath.isEmpty() || "/".equals(rawBasePath)
+ ? "/ws" : rawBasePath + "/ws";
+ return new URI(webSocketScheme + "://" + restUri.getRawAuthority() + webSocketPath);
}
/**
diff --git a/zeppelin-client/src/main/java/org/apache/zeppelin/client/ZeppelinClient.java b/zeppelin-client/src/main/java/org/apache/zeppelin/client/ZeppelinClient.java
index f5db6f15ecb..83388a6dd1f 100644
--- a/zeppelin-client/src/main/java/org/apache/zeppelin/client/ZeppelinClient.java
+++ b/zeppelin-client/src/main/java/org/apache/zeppelin/client/ZeppelinClient.java
@@ -22,53 +22,57 @@
import kong.unirest.HttpResponse;
import kong.unirest.JsonNode;
import kong.unirest.Unirest;
+import kong.unirest.UnirestInstance;
import kong.unirest.apache.ApacheClient;
import kong.unirest.json.JSONArray;
import kong.unirest.json.JSONObject;
import org.apache.commons.text.StringEscapeUtils;
-import org.apache.http.conn.ssl.NoopHostnameVerifier;
-import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
-import org.apache.http.ssl.SSLContextBuilder;
import org.apache.zeppelin.common.SessionInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import unirest.shaded.org.apache.http.client.CookieStore;
import unirest.shaded.org.apache.http.client.HttpClient;
+import unirest.shaded.org.apache.http.cookie.ClientCookie;
+import unirest.shaded.org.apache.http.cookie.Cookie;
+import unirest.shaded.org.apache.http.impl.client.BasicCookieStore;
+import unirest.shaded.org.apache.http.impl.client.HttpClientBuilder;
import unirest.shaded.org.apache.http.impl.client.HttpClients;
-import javax.net.ssl.SSLContext;
-import java.security.cert.X509Certificate;
+import java.net.URI;
import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Date;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.stream.Collectors;
/**
* Low level api for interacting with Zeppelin. Underneath, it use the zeppelin rest api.
* You can use this class to operate Zeppelin note/paragraph,
* e.g. get/add/delete/update/execute/cancel
*/
-public class ZeppelinClient {
+public class ZeppelinClient implements AutoCloseable {
private static final Logger LOGGER = LoggerFactory.getLogger(ZeppelinClient.class);
- private ClientConfig clientConfig;
+ private final ClientConfig clientConfig;
+ private final UnirestInstance unirest;
+ private final CookieStore cookieStore;
public ZeppelinClient(ClientConfig clientConfig) throws Exception {
this.clientConfig = clientConfig;
- Unirest.config().defaultBaseUrl(clientConfig.getZeppelinRestUrl() + "/api");
-
- if (clientConfig.isUseKnox()) {
- try {
- SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new TrustSelfSignedStrategy() {
- public boolean isTrusted(X509Certificate[] chain, String authType) {
- return true;
- }
- }).build();
- HttpClient customHttpClient = HttpClients.custom().setSSLContext(sslContext)
- .setSSLHostnameVerifier(new NoopHostnameVerifier()).build();
- Unirest.config().httpClient(ApacheClient.builder(customHttpClient));
- } catch (Exception e) {
- throw new Exception("Fail to setup httpclient of Unirest", e);
- }
+ this.unirest = Unirest.spawnInstance();
+ this.cookieStore = new BasicCookieStore();
+ this.unirest.config().defaultBaseUrl(clientConfig.getZeppelinRestUrl() + "/api");
+
+ try {
+ HttpClientBuilder httpClientBuilder =
+ HttpClients.custom().useSystemProperties().setDefaultCookieStore(cookieStore);
+ HttpClient customHttpClient = httpClientBuilder.build();
+ this.unirest.config().httpClient(ApacheClient.builder(customHttpClient));
+ } catch (Exception e) {
+ throw new Exception("Fail to setup httpclient of Unirest", e);
}
}
@@ -76,6 +80,75 @@ public ClientConfig getClientConfig() {
return clientConfig;
}
+ @Override
+ public void close() {
+ unirest.close();
+ }
+
+ /**
+ * Build the Cookie request header for the supplied WebSocket URI from this client's REST
+ * cookie store. Cookies are scoped to the target scheme, host and path before being exposed.
+ */
+ String getSessionCookieHeader(URI requestUri) {
+ Date now = new Date();
+ String requestHost = requestUri.getHost();
+ String requestPath = requestUri.getRawPath();
+ boolean secureRequest = "wss".equalsIgnoreCase(requestUri.getScheme());
+ if (requestHost == null) {
+ return "";
+ }
+ if (requestPath == null || requestPath.isEmpty()) {
+ requestPath = "/";
+ }
+ final String normalizedRequestPath = requestPath;
+ return cookieStore.getCookies().stream()
+ .filter(cookie -> !cookie.isExpired(now))
+ .filter(cookie -> !cookie.isSecure() || secureRequest)
+ .filter(cookie -> domainMatches(cookie, requestHost))
+ .filter(cookie -> pathMatches(cookie.getPath(), normalizedRequestPath))
+ .sorted(Comparator.comparingInt(
+ (Cookie cookie) -> normalizedCookiePath(cookie.getPath()).length()).reversed())
+ .map(cookie -> cookie.getName() + "=" + cookie.getValue())
+ .collect(Collectors.joining("; "));
+ }
+
+ private boolean domainMatches(Cookie cookie, String requestHost) {
+ String cookieDomain = cookie.getDomain();
+ if (cookieDomain == null || cookieDomain.isEmpty()) {
+ return false;
+ }
+
+ String normalizedHost = requestHost.toLowerCase(Locale.ROOT);
+ String normalizedDomain = cookieDomain.toLowerCase(Locale.ROOT);
+ if (normalizedDomain.startsWith(".")) {
+ normalizedDomain = normalizedDomain.substring(1);
+ }
+
+ boolean domainAttributePresent = cookie instanceof ClientCookie
+ && ((ClientCookie) cookie).containsAttribute(ClientCookie.DOMAIN_ATTR);
+ if (!domainAttributePresent) {
+ return normalizedHost.equals(normalizedDomain);
+ }
+ return normalizedHost.equals(normalizedDomain)
+ || normalizedHost.endsWith("." + normalizedDomain);
+ }
+
+ private boolean pathMatches(String cookiePath, String requestPath) {
+ String normalizedCookiePath = normalizedCookiePath(cookiePath);
+ if (requestPath.equals(normalizedCookiePath)) {
+ return true;
+ }
+ if (!requestPath.startsWith(normalizedCookiePath)) {
+ return false;
+ }
+ return normalizedCookiePath.endsWith("/")
+ || requestPath.charAt(normalizedCookiePath.length()) == '/';
+ }
+
+ private String normalizedCookiePath(String cookiePath) {
+ return cookiePath == null || cookiePath.isEmpty() ? "/" : cookiePath;
+ }
+
/**
* Throw exception if the status code is not 200.
*
@@ -119,7 +192,7 @@ private void checkJsonNodeStatus(JsonNode jsonNode) throws Exception {
* @throws Exception
*/
public String getVersion() throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.get("/version")
.asJson();
checkResponse(response);
@@ -137,7 +210,7 @@ public String getVersion() throws Exception {
* @throws Exception
*/
public SessionInfo newSession(String interpreter) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/session")
.header("Content-Type", "application/json")
.queryString("interpreter", interpreter)
@@ -155,7 +228,7 @@ public SessionInfo newSession(String interpreter) throws Exception {
* @throws Exception
*/
public void stopSession(String sessionId) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.delete("/session/{sessionId}")
.routeParam("sessionId", sessionId)
.asJson();
@@ -172,7 +245,7 @@ public void stopSession(String sessionId) throws Exception {
* @throws Exception
*/
public SessionInfo getSession(String sessionId) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.get("/session/{sessionId}")
.routeParam("sessionId", sessionId)
.asJson();
@@ -211,7 +284,7 @@ public List listSessions() throws Exception {
* @throws Exception
*/
public List listSessions(String interpreter) throws Exception {
- GetRequest getRequest = Unirest.get("/session");
+ GetRequest getRequest = unirest.get("/session");
if (interpreter != null) {
getRequest.queryString("interpreter", interpreter);
}
@@ -260,7 +333,7 @@ private SessionInfo createSessionInfoFromJson(JSONObject sessionJson) {
*/
public void login(String userName, String password) throws Exception {
if (clientConfig.isUseKnox()) {
- HttpResponse response = Unirest.get(clientConfig.getKnoxSSOUrl() +
+ HttpResponse response = unirest.get(clientConfig.getKnoxSSOUrl() +
"?originalUrl=" + clientConfig.getZeppelinRestUrl())
.basicAuth(userName, password)
.asString();
@@ -269,7 +342,7 @@ public void login(String userName, String password) throws Exception {
response.getStatus(),
response.getStatusText()));
}
- response = Unirest.get("/security/ticket")
+ response = unirest.get("/security/ticket")
.asString();
if (response.getStatus() != 200) {
throw new Exception(String.format("Fail to get ticket after Knox SSO, status: %s, statusText: %s",
@@ -277,7 +350,7 @@ public void login(String userName, String password) throws Exception {
response.getStatusText()));
}
} else {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/login")
.field("userName", userName)
.field("password", password)
@@ -306,7 +379,7 @@ public String createNote(String notePath, String defaultInterpreterGroup) throws
JSONObject bodyObject = new JSONObject();
bodyObject.put("notePath", notePath);
bodyObject.put("defaultInterpreterGroup", defaultInterpreterGroup);
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/notebook")
.header("Content-Type", "application/json")
.body(bodyObject.toString())
@@ -325,7 +398,7 @@ public String createNote(String notePath, String defaultInterpreterGroup) throws
* @throws Exception
*/
public void deleteNote(String noteId) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.delete("/notebook/{noteId}")
.routeParam("noteId", noteId)
.asJson();
@@ -345,7 +418,7 @@ public void deleteNote(String noteId) throws Exception {
public String cloneNote(String noteId, String destPath) throws Exception {
JSONObject bodyObject = new JSONObject();
bodyObject.put("name", destPath);
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/notebook/{noteId}")
.routeParam("noteId", noteId)
.header("Content-Type", "application/json")
@@ -362,7 +435,7 @@ public void renameNote(String noteId, String newNotePath) throws Exception {
JSONObject bodyObject = new JSONObject();
bodyObject.put("name", newNotePath);
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.put("/notebook/{noteId}/rename")
.routeParam("noteId", noteId)
.header("Content-Type", "application/json")
@@ -382,7 +455,7 @@ public void renameNote(String noteId, String newNotePath) throws Exception {
* @throws Exception
*/
public NoteResult queryNoteResult(String noteId) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.get("/notebook/{noteId}")
.routeParam("noteId", noteId)
.asJson();
@@ -399,7 +472,7 @@ public NoteResult queryNoteResult(String noteId) throws Exception {
public NoteResult queryNoteResultByPath(String notePath) throws Exception {
JSONObject bodyObject = new JSONObject();
bodyObject.put("notePath", notePath);
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/notebook/getByPath")
.header("Content-Type", "application/json")
.body(bodyObject)
@@ -494,7 +567,7 @@ public NoteResult submitNote(String noteId, Map parameters) thro
JSONObject bodyObject = new JSONObject();
bodyObject.put("params", parameters);
// run note in non-blocking and isolated way.
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/notebook/job/{noteId}")
.routeParam("noteId", noteId)
.queryString("blocking", "false")
@@ -515,7 +588,7 @@ public NoteResult submitNote(String noteId, Map parameters) thro
* @throws Exception
*/
public void cancelNote(String noteId) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.delete("/notebook/job/{noteId}")
.routeParam("noteId", noteId)
.asJson();
@@ -534,7 +607,7 @@ public void cancelNote(String noteId) throws Exception {
*/
public String importNote(String notePath, String noteContent) throws Exception {
JSONObject bodyObject = new JSONObject(noteContent);
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/notebook/import")
.queryString("notePath", notePath)
.header("Content-Type", "application/json")
@@ -597,7 +670,7 @@ public String addParagraph(String noteId, String title, String text) throws Exce
JSONObject bodyObject = new JSONObject();
bodyObject.put("title", title);
bodyObject.put("text", text);
- HttpResponse response = Unirest.post("/notebook/{noteId}/paragraph")
+ HttpResponse response = unirest.post("/notebook/{noteId}/paragraph")
.routeParam("noteId", noteId)
.header("Content-Type", "application/json")
.body(bodyObject.toString())
@@ -622,7 +695,7 @@ public void updateParagraph(String noteId, String paragraphId, String title, Str
JSONObject bodyObject = new JSONObject();
bodyObject.put("title", title);
bodyObject.put("text", text);
- HttpResponse response = Unirest.put("/notebook/{noteId}/paragraph/{paragraphId}")
+ HttpResponse response = unirest.put("/notebook/{noteId}/paragraph/{paragraphId}")
.routeParam("noteId", noteId)
.routeParam("paragraphId", paragraphId)
.header("Content-Type", "application/json")
@@ -712,7 +785,7 @@ public ParagraphResult submitParagraph(String noteId,
Map parameters) throws Exception {
JSONObject bodyObject = new JSONObject();
bodyObject.put("params", parameters);
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/notebook/job/{noteId}/{paragraphId}")
.routeParam("noteId", noteId)
.routeParam("paragraphId", paragraphId)
@@ -780,7 +853,7 @@ public ParagraphResult submitParagraph(String noteId, String paragraphId) throws
* @throws Exception
*/
public String nextSessionParagraph(String noteId, int maxParagraph) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.post("/notebook/{noteId}/paragraph/next")
.routeParam("noteId", noteId)
.header("Content-Type", "application/json")
@@ -801,7 +874,7 @@ public String nextSessionParagraph(String noteId, int maxParagraph) throws Excep
* @throws Exception
*/
public void cancelParagraph(String noteId, String paragraphId) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.delete("/notebook/job/{noteId}/{paragraphId}")
.routeParam("noteId", noteId)
.routeParam("paragraphId", paragraphId)
@@ -820,7 +893,7 @@ public void cancelParagraph(String noteId, String paragraphId) throws Exception
* @throws Exception
*/
public ParagraphResult queryParagraphResult(String noteId, String paragraphId) throws Exception {
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.get("/notebook/{noteId}/paragraph/{paragraphId}")
.routeParam("noteId", noteId)
.routeParam("paragraphId", paragraphId)
@@ -900,7 +973,7 @@ public ParagraphResult waitUtilParagraphRunning(String noteId, String paragraphI
public void stopInterpreter(String noteId, String interpreter) throws Exception {
JSONObject bodyObject = new JSONObject();
bodyObject.put("noteId", noteId);
- HttpResponse response = Unirest
+ HttpResponse response = unirest
.put("/interpreter/setting/restart/{interpreter}")
.routeParam("interpreter", interpreter)
.header("Content-Type", "application/json")
diff --git a/zeppelin-client/src/main/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClient.java b/zeppelin-client/src/main/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClient.java
index c09fb845ad8..19752eb1106 100644
--- a/zeppelin-client/src/main/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClient.java
+++ b/zeppelin-client/src/main/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClient.java
@@ -33,6 +33,7 @@
import java.io.IOException;
import java.net.URI;
+import java.net.URISyntaxException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
@@ -61,11 +62,14 @@ public ZeppelinWebSocketClient(MessageHandler messageHandler) {
}
public void connect(String url) throws Exception {
+ connect(url, null);
+ }
+
+ public void connect(String url, String cookieHeader) throws Exception {
+ URI echoUri = new URI(url);
+ ClientUpgradeRequest request = createUpgradeRequest(echoUri, cookieHeader);
this.wsClient = new WebSocketClient();
wsClient.start();
- URI echoUri = new URI(url);
- ClientUpgradeRequest request = new ClientUpgradeRequest();
- request.setHeader("Origin", "*");
CompletableFuture future = wsClient.connect(this, echoUri, request);
try {
future.get(DEFAULT_CONNECT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
@@ -81,6 +85,37 @@ public void connect(String url) throws Exception {
LOGGER.info("WebSocket connect established");
}
+ ClientUpgradeRequest createUpgradeRequest(URI webSocketUri, String cookieHeader)
+ throws URISyntaxException {
+ ClientUpgradeRequest request = new ClientUpgradeRequest();
+ request.setHeader("Origin", toHttpOrigin(webSocketUri));
+ if (cookieHeader != null && !cookieHeader.trim().isEmpty()) {
+ request.setHeader("Cookie", cookieHeader);
+ }
+ return request;
+ }
+
+ private static String toHttpOrigin(URI webSocketUri) throws URISyntaxException {
+ String scheme;
+ if ("wss".equalsIgnoreCase(webSocketUri.getScheme())) {
+ scheme = "https";
+ } else if ("ws".equalsIgnoreCase(webSocketUri.getScheme())) {
+ scheme = "http";
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported WebSocket URL scheme: " + webSocketUri.getScheme());
+ }
+ if (webSocketUri.getHost() == null) {
+ throw new IllegalArgumentException("WebSocket URL must contain a host");
+ }
+ int port = webSocketUri.getPort();
+ if (("http".equals(scheme) && port == 80) || ("https".equals(scheme) && port == 443)) {
+ port = -1;
+ }
+ return new URI(scheme, null, webSocketUri.getHost(), port,
+ null, null, null).toString();
+ }
+
public void addStatementMessageHandler(String statementId,
StatementMessageHandler statementMessageHandler) throws Exception {
if (messageHandler instanceof CompositeMessageHandler) {
diff --git a/zeppelin-client/src/test/java/org/apache/zeppelin/client/ZSessionTest.java b/zeppelin-client/src/test/java/org/apache/zeppelin/client/ZSessionTest.java
new file mode 100644
index 00000000000..d803a03d61c
--- /dev/null
+++ b/zeppelin-client/src/test/java/org/apache/zeppelin/client/ZSessionTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.client;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.Test;
+
+class ZSessionTest {
+
+ @Test
+ void websocketUriPreservesGatewayPathAndSecureTransport() throws Exception {
+ assertEquals("wss://knox.example/gateway/default/zeppelin/ws",
+ ZSession.toWebSocketUri(
+ "https://knox.example/gateway/default/zeppelin").toString());
+ }
+
+ @Test
+ void websocketUriNormalizesATrailingGatewaySlash() throws Exception {
+ assertEquals("wss://knox.example/gateway/default/zeppelin/ws",
+ ZSession.toWebSocketUri(
+ "https://knox.example/gateway/default/zeppelin/").toString());
+ }
+
+ @Test
+ void websocketUriPreservesPercentEncodedGatewaySegments() throws Exception {
+ assertEquals("wss://knox.example/gateway/a%2Fb/ws",
+ ZSession.toWebSocketUri("https://knox.example/gateway/a%2Fb").toString());
+ }
+
+ @Test
+ void websocketUriUsesPlainWebsocketForHttp() throws Exception {
+ assertEquals("ws://localhost:8080/ws",
+ ZSession.toWebSocketUri("http://localhost:8080").toString());
+ }
+
+ @Test
+ void websocketUriRejectsUnsupportedSchemes() {
+ assertThrows(IllegalArgumentException.class,
+ () -> ZSession.toWebSocketUri("ftp://localhost:8080"));
+ }
+
+ @Test
+ void existingSessionCanAuthenticateBeforeProtectedReconnect() throws Exception {
+ AtomicBoolean authenticatedLookup = new AtomicBoolean();
+ HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/api/login", exchange -> {
+ try {
+ exchange.getRequestBody().readAllBytes();
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.getResponseHeaders().add("Set-Cookie", "JSESSIONID=session-cookie; Path=/");
+ writeResponse(exchange, 200, "{}");
+ } finally {
+ exchange.close();
+ }
+ });
+ server.createContext("/api/session/session-id", exchange -> {
+ try {
+ if ("GET".equals(exchange.getRequestMethod())) {
+ authenticatedLookup.set(
+ "JSESSIONID=session-cookie".equals(
+ exchange.getRequestHeaders().getFirst("Cookie")));
+ writeResponse(exchange, 200,
+ "{\"status\":\"OK\",\"body\":{\"sessionId\":\"session-id\","
+ + "\"state\":\"Running\"}}");
+ } else {
+ writeResponse(exchange, 200, "{\"status\":\"OK\",\"body\":{}}");
+ }
+ } finally {
+ exchange.close();
+ }
+ });
+ server.start();
+
+ try {
+ ClientConfig config = new ClientConfig(
+ "http://127.0.0.1:" + server.getAddress().getPort());
+ try (ZSession session = ZSession.createFromExistingSession(
+ config,
+ "spark",
+ "session-id",
+ client -> client.login("user", "password"),
+ null)) {
+ assertEquals("session-id", session.getSessionId());
+ assertTrue(authenticatedLookup.get());
+ }
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ private static void writeResponse(
+ com.sun.net.httpserver.HttpExchange exchange, int status, String json) throws IOException {
+ byte[] body = json.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.sendResponseHeaders(status, body.length);
+ exchange.getResponseBody().write(body);
+ }
+}
diff --git a/zeppelin-client/src/test/java/org/apache/zeppelin/client/ZeppelinClientTest.java b/zeppelin-client/src/test/java/org/apache/zeppelin/client/ZeppelinClientTest.java
new file mode 100644
index 00000000000..c241360e691
--- /dev/null
+++ b/zeppelin-client/src/test/java/org/apache/zeppelin/client/ZeppelinClientTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.client;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class ZeppelinClientTest {
+
+ private HttpServer server;
+
+ @AfterEach
+ void stopServer() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void restSessionCookiesAreIsolatedAndAvailableForWebSocketHandshake() throws Exception {
+ AtomicInteger loginCount = new AtomicInteger();
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/api/login", exchange -> {
+ try {
+ exchange.getRequestBody().readAllBytes();
+ String sessionId = "session-" + loginCount.incrementAndGet();
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.getResponseHeaders().add("Set-Cookie",
+ "JSESSIONID=" + sessionId + "; Path=/; HttpOnly");
+ byte[] body = "{}".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ } finally {
+ exchange.close();
+ }
+ });
+ server.start();
+
+ String serverUrl = "http://127.0.0.1:" + server.getAddress().getPort();
+ try (ZeppelinClient firstClient = new ZeppelinClient(new ClientConfig(serverUrl));
+ ZeppelinClient secondClient = new ZeppelinClient(new ClientConfig(serverUrl))) {
+ firstClient.login("first", "password");
+ secondClient.login("second", "password");
+
+ URI webSocketUri = URI.create(serverUrl.replace("http", "ws") + "/ws");
+ assertEquals("JSESSIONID=session-1", firstClient.getSessionCookieHeader(webSocketUri));
+ assertEquals("JSESSIONID=session-2", secondClient.getSessionCookieHeader(webSocketUri));
+ }
+ }
+
+ @Test
+ void cookieHeaderIsEmptyForAnotherHost() throws Exception {
+ server = startServerWithSessionCookie("session-id");
+ String serverUrl = "http://127.0.0.1:" + server.getAddress().getPort();
+ try (ZeppelinClient client = new ZeppelinClient(new ClientConfig(serverUrl))) {
+ client.login("user", "password");
+
+ assertEquals("", client.getSessionCookieHeader(URI.create("ws://localhost/ws")));
+ }
+ }
+
+ @Test
+ void cookieHeaderHonorsCookiePathAndSecureAttributes() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/api/login", exchange -> {
+ try {
+ exchange.getRequestBody().readAllBytes();
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.getResponseHeaders().add("Set-Cookie", "root-cookie=root; Path=/");
+ exchange.getResponseHeaders().add("Set-Cookie", "ws-cookie=websocket; Path=/ws");
+ exchange.getResponseHeaders().add("Set-Cookie", "api-cookie=rest; Path=/api");
+ exchange.getResponseHeaders().add("Set-Cookie", "secure-cookie=secure; Path=/; Secure");
+ byte[] body = "{}".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ } finally {
+ exchange.close();
+ }
+ });
+ server.start();
+
+ String authority = "127.0.0.1:" + server.getAddress().getPort();
+ try (ZeppelinClient client = new ZeppelinClient(
+ new ClientConfig("http://" + authority))) {
+ client.login("user", "password");
+
+ assertEquals("ws-cookie=websocket; root-cookie=root",
+ client.getSessionCookieHeader(URI.create("ws://" + authority + "/ws")));
+ assertEquals("ws-cookie=websocket; root-cookie=root; secure-cookie=secure",
+ client.getSessionCookieHeader(URI.create("wss://" + authority + "/ws")));
+ }
+ }
+
+ @Test
+ void cookieHeaderPreservesPercentEncodedGatewayPath() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/api/login", exchange -> {
+ try {
+ exchange.getRequestBody().readAllBytes();
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.getResponseHeaders().add(
+ "Set-Cookie", "JSESSIONID=session-id; Path=/gateway/a%2Fb");
+ byte[] body = "{}".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ } finally {
+ exchange.close();
+ }
+ });
+ server.start();
+
+ String authority = "127.0.0.1:" + server.getAddress().getPort();
+ try (ZeppelinClient client = new ZeppelinClient(
+ new ClientConfig("http://" + authority))) {
+ client.login("user", "password");
+
+ assertEquals("JSESSIONID=session-id", client.getSessionCookieHeader(
+ URI.create("ws://" + authority + "/gateway/a%2Fb/ws")));
+ }
+ }
+
+ private HttpServer startServerWithSessionCookie(String sessionId) throws IOException {
+ HttpServer httpServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ httpServer.createContext("/api/login", exchange -> {
+ try {
+ exchange.getRequestBody().readAllBytes();
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.getResponseHeaders().add("Set-Cookie",
+ "JSESSIONID=" + sessionId + "; Path=/; HttpOnly");
+ byte[] body = "{}".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ } finally {
+ exchange.close();
+ }
+ });
+ httpServer.start();
+ return httpServer;
+ }
+}
diff --git a/zeppelin-client/src/test/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClientTest.java b/zeppelin-client/src/test/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClientTest.java
index 7f6cabd1e2d..28ec9445a26 100644
--- a/zeppelin-client/src/test/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClientTest.java
+++ b/zeppelin-client/src/test/java/org/apache/zeppelin/client/websocket/ZeppelinWebSocketClientTest.java
@@ -17,15 +17,56 @@
package org.apache.zeppelin.client.websocket;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.junit.jupiter.api.Test;
import java.time.Duration;
+import java.net.URI;
class ZeppelinWebSocketClientTest {
+ @Test
+ void upgradeRequestIncludesRestSessionCookies() throws Exception {
+ ZeppelinWebSocketClient client = new ZeppelinWebSocketClient(msg -> { });
+
+ ClientUpgradeRequest request = client.createUpgradeRequest(
+ URI.create("wss://knox.example/gateway/default/zeppelin/ws"),
+ "JSESSIONID=session-id; hadoop-jwt=knox-token");
+
+ assertEquals("JSESSIONID=session-id; hadoop-jwt=knox-token",
+ request.getHeader("Cookie"));
+ assertEquals("https://knox.example", request.getHeader("Origin"));
+ }
+
+ @Test
+ void upgradeRequestOmitsCookieHeaderForAnonymousSession() throws Exception {
+ ZeppelinWebSocketClient client = new ZeppelinWebSocketClient(msg -> { });
+
+ ClientUpgradeRequest request = client.createUpgradeRequest(
+ URI.create("ws://localhost:8080/ws"), " ");
+
+ assertNull(request.getHeader("Cookie"));
+ assertEquals("http://localhost:8080", request.getHeader("Origin"));
+ }
+
+ @Test
+ void upgradeRequestCanonicalizesDefaultOriginPorts() throws Exception {
+ ZeppelinWebSocketClient client = new ZeppelinWebSocketClient(msg -> { });
+
+ ClientUpgradeRequest secure = client.createUpgradeRequest(
+ URI.create("wss://zeppelin.example:443/ws"), null);
+ ClientUpgradeRequest plain = client.createUpgradeRequest(
+ URI.create("ws://zeppelin.example:80/ws"), null);
+
+ assertEquals("https://zeppelin.example", secure.getHeader("Origin"));
+ assertEquals("http://zeppelin.example", plain.getHeader("Origin"));
+ }
+
@Test
void connectFailsFastWhenPortClosed() {
ZeppelinWebSocketClient client = new ZeppelinWebSocketClient(msg -> { });
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/AbstractZeppelinIT.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/AbstractZeppelinIT.java
index bc433bcb6fe..91d750b8b10 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/AbstractZeppelinIT.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/AbstractZeppelinIT.java
@@ -98,6 +98,7 @@ protected void authenticationUser(String userName, String password) {
visibilityWait(
By.xpath("//div[contains(@class, 'navbar-collapse')]//li//button[contains(@class, 'nav-btn dropdown-toggle ng-scope')]"),
MAX_BROWSER_TIMEOUT_SEC);
+ manager.waitForWebSocketConnected();
try {
((JavascriptExecutor) manager.getWebDriver()).executeScript(
"$('.modal-backdrop').remove(); $('#loginModal').modal('hide');");
@@ -141,6 +142,7 @@ protected void authenticationUserViaRest(String userName, String password) {
}
manager.getWebDriver().navigate().refresh();
visibilityWait(loggedInUserMenuLocator(), MAX_BROWSER_TIMEOUT_SEC);
+ manager.waitForWebSocketConnected();
}
// Shared locator for the logged-in navbar user menu button. Uses a class-order-agnostic
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/WebDriverManager.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/WebDriverManager.java
index a6aa0341a2b..d4287a6b7d8 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/WebDriverManager.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/WebDriverManager.java
@@ -29,6 +29,7 @@
import java.util.stream.Stream;
import org.apache.commons.lang3.SystemUtils;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.openqa.selenium.By;
import org.openqa.selenium.Dimension;
import org.openqa.selenium.TimeoutException;
@@ -41,7 +42,6 @@
import org.openqa.selenium.firefox.FirefoxProfile;
import org.openqa.selenium.firefox.GeckoDriverService;
import org.openqa.selenium.safari.SafariDriver;
-import org.openqa.selenium.support.ui.ExpectedCondition;
import org.openqa.selenium.support.ui.WebDriverWait;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,6 +60,9 @@
public class WebDriverManager implements Closeable {
public final static Logger LOG = LoggerFactory.getLogger(WebDriverManager.class);
+ private static final By WEBSOCKET_CONNECTED =
+ By.xpath("//i[@uib-tooltip='WebSocket Connected']");
+ private static final By LOGIN_BUTTON = By.cssSelector(".nav-login-btn");
final boolean deleteTempFiles;
final Path logDir;
@@ -77,6 +80,15 @@ public WebDriverManager(int port) throws IOException {
this(true, port);
}
+ public WebDriverManager(boolean deleteTempFiles, ZeppelinConfiguration zConf)
+ throws IOException {
+ this(deleteTempFiles, configureBrowserOrigin(zConf));
+ }
+
+ public WebDriverManager(ZeppelinConfiguration zConf) throws IOException {
+ this(true, zConf);
+ }
+
public WebDriver getWebDriver() {
return this.driver;
}
@@ -151,20 +163,17 @@ private WebDriver constructWebDriver(int port) {
long start = System.currentTimeMillis();
boolean loaded = false;
- driver.manage().timeouts()
- .implicitlyWait(Duration.ofSeconds(AbstractZeppelinIT.MAX_IMPLICIT_WAIT));
+ // Explicit readiness polling must not inherit the normal implicit wait. Otherwise an
+ // authenticated page spends the full implicit timeout looking for the deliberately absent
+ // WebSocket indicator before it can notice that the login UI is ready.
+ driver.manage().timeouts().implicitlyWait(Duration.ZERO);
driver.get(url);
while (System.currentTimeMillis() - start < 60 * 1000) {
// wait for page load
try {
- (new WebDriverWait(driver, Duration.ofSeconds(60))).until(new ExpectedCondition() {
- @Override
- public Boolean apply(WebDriver d) {
- return d.findElement(By.xpath("//i[@uib-tooltip='WebSocket Connected']"))
- .isDisplayed();
- }
- });
+ (new WebDriverWait(driver, Duration.ofSeconds(60)))
+ .until(d -> isDisplayed(d, WEBSOCKET_CONNECTED) || isDisplayed(d, LOGIN_BUTTON));
loaded = true;
break;
} catch (TimeoutException e) {
@@ -174,6 +183,8 @@ public Boolean apply(WebDriver d) {
}
assertTrue(loaded);
+ driver.manage().timeouts()
+ .implicitlyWait(Duration.ofSeconds(AbstractZeppelinIT.MAX_IMPLICIT_WAIT));
try {
// Manually setting fixed window size since `maximize()` crashes for Chrome/Edge driver on linux with xvfb.
@@ -185,6 +196,23 @@ public Boolean apply(WebDriver d) {
return driver;
}
+ public void waitForWebSocketConnected() {
+ (new WebDriverWait(driver, Duration.ofSeconds(60)))
+ .until(d -> isDisplayed(d, WEBSOCKET_CONNECTED));
+ }
+
+ private static boolean isDisplayed(WebDriver driver, By locator) {
+ return driver.findElements(locator).stream().anyMatch(element -> element.isDisplayed());
+ }
+
+ private static int configureBrowserOrigin(ZeppelinConfiguration zConf) {
+ int port = zConf.getServerPort();
+ zConf.setProperty(
+ ZeppelinConfiguration.ConfVars.ZEPPELIN_ALLOWED_ORIGINS.getVarName(),
+ "http://localhost:" + port);
+ return port;
+ }
+
private WebDriver getFirefoxDriver() {
FirefoxProfile profile = new FirefoxProfile();
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/AuthenticationIT.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/AuthenticationIT.java
index fe77c65da7e..39183bd8de6 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/AuthenticationIT.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/AuthenticationIT.java
@@ -77,7 +77,7 @@ static void init() throws Exception {
@BeforeEach
public void startUpManager() throws IOException {
- manager = new WebDriverManager(zepServer.getZeppelinConfiguration().getServerPort());
+ manager = new WebDriverManager(zepServer.getZeppelinConfiguration());
}
@AfterEach
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/InterpreterIT.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/InterpreterIT.java
index 5fe0ff0291d..0f12ac8b5e2 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/InterpreterIT.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/InterpreterIT.java
@@ -45,7 +45,7 @@ static void init() throws Exception {
@BeforeEach
public void startUp() throws IOException {
- manager = new WebDriverManager(zepServer.getZeppelinConfiguration().getServerPort());
+ manager = new WebDriverManager(zepServer.getZeppelinConfiguration());
}
@AfterEach
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/InterpreterModeActionsIT.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/InterpreterModeActionsIT.java
index da204bf98e2..d5bd6547dbd 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/InterpreterModeActionsIT.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/InterpreterModeActionsIT.java
@@ -84,7 +84,7 @@ static void init() throws Exception {
@BeforeEach
public void startUp() throws IOException {
- manager = new WebDriverManager(zepServer.getZeppelinConfiguration().getServerPort());
+ manager = new WebDriverManager(zepServer.getZeppelinConfiguration());
}
@AfterEach
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/ParagraphActionsIT.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/ParagraphActionsIT.java
index 7027d5591b9..7a8ef7efe3c 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/ParagraphActionsIT.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/ParagraphActionsIT.java
@@ -62,7 +62,7 @@ static void init() throws Exception {
@BeforeEach
public void startUp() throws IOException {
- manager = new WebDriverManager(zepServer.getZeppelinConfiguration().getServerPort());
+ manager = new WebDriverManager(zepServer.getZeppelinConfiguration());
}
@AfterEach
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/PersonalizeActionsIT.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/PersonalizeActionsIT.java
index 8abe1ca9a12..ca9de83b0ff 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/PersonalizeActionsIT.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/PersonalizeActionsIT.java
@@ -76,7 +76,7 @@ static void init() throws Exception {
@BeforeEach
public void startUp() throws IOException {
- manager = new WebDriverManager(zepServer.getZeppelinConfiguration().getServerPort());
+ manager = new WebDriverManager(zepServer.getZeppelinConfiguration());
}
@AfterEach
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/SparkParagraphIT.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/SparkParagraphIT.java
index a81ab893ddc..61942ec3fd7 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/SparkParagraphIT.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/SparkParagraphIT.java
@@ -58,7 +58,7 @@ static void init() throws Exception {
@BeforeEach
public void startUp() throws IOException {
- manager = new WebDriverManager(zepServer.getZeppelinConfiguration().getServerPort());
+ manager = new WebDriverManager(zepServer.getZeppelinConfiguration());
createNewNote();
waitForParagraph(1, "READY");
}
diff --git a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java
index ff45f123540..3469cd9e503 100644
--- a/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java
+++ b/zeppelin-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinIT.java
@@ -75,7 +75,7 @@ static void init() throws Exception {
@BeforeEach
public void startUp() throws IOException {
- manager = new WebDriverManager(zepServer.getZeppelinConfiguration().getServerPort());
+ manager = new WebDriverManager(zepServer.getZeppelinConfiguration());
}
@AfterEach
diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinClientIntegrationTest.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinClientIntegrationTest.java
index e1e3bbf3319..e375b426051 100644
--- a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinClientIntegrationTest.java
+++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinClientIntegrationTest.java
@@ -75,7 +75,13 @@ static void init() throws Exception {
@AfterAll
static void destroy() throws Exception {
- zepServer.destroy();
+ try {
+ if (zeppelinClient != null) {
+ zeppelinClient.close();
+ }
+ } finally {
+ zepServer.destroy();
+ }
}
@BeforeEach
diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinClientWithAuthIntegrationTest.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinClientWithAuthIntegrationTest.java
index bfe170916fe..320da8a303d 100644
--- a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinClientWithAuthIntegrationTest.java
+++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinClientWithAuthIntegrationTest.java
@@ -58,7 +58,13 @@ static void init() throws Exception {
@AfterAll
static void destroy() throws Exception {
- zepServer.destroy();
+ try {
+ if (zeppelinClient != null) {
+ zeppelinClient.close();
+ }
+ } finally {
+ zepServer.destroy();
+ }
}
@BeforeEach
@@ -109,4 +115,3 @@ void testLoginFailed() throws Exception {
}
}
}
-
diff --git a/zeppelin-plugins/notebookrepo/github/src/main/java/org/apache/zeppelin/notebook/repo/GitHubNotebookRepo.java b/zeppelin-plugins/notebookrepo/github/src/main/java/org/apache/zeppelin/notebook/repo/GitHubNotebookRepo.java
index 395c79ee775..4e9e5a639e8 100644
--- a/zeppelin-plugins/notebookrepo/github/src/main/java/org/apache/zeppelin/notebook/repo/GitHubNotebookRepo.java
+++ b/zeppelin-plugins/notebookrepo/github/src/main/java/org/apache/zeppelin/notebook/repo/GitHubNotebookRepo.java
@@ -66,10 +66,10 @@ public void init(ZeppelinConfiguration zConf, NoteParser noteParser) throws IOEx
}
@Override
- public Revision checkpoint(String noteId,
- String notePath,
- String commitMessage,
- AuthenticationInfo subject) throws IOException {
+ public synchronized Revision checkpoint(String noteId,
+ String notePath,
+ String commitMessage,
+ AuthenticationInfo subject) throws IOException {
Revision revision = super.checkpoint(noteId, notePath, commitMessage, subject);
updateRemoteStream();
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
index 179dcce6e85..deb7c854612 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
@@ -67,7 +67,8 @@ public class ZeppelinConfiguration {
private static final String ZEPPELIN_SITE_XML = "zeppelin-site.xml";
private static final Logger LOGGER = LoggerFactory.getLogger(ZeppelinConfiguration.class);
- private Boolean anonymousAllowed;
+ private volatile Boolean anonymousAllowed;
+ private volatile String shiroPath;
private static final EnvironmentConfiguration envConfig = new EnvironmentConfiguration();
private static final SystemConfiguration sysConfig = new SystemConfiguration();
@@ -618,10 +619,32 @@ public String getCredentialsPath(boolean absolute) {
}
public String getShiroPath() {
+ String initializedShiroPath = shiroPath;
+ if (initializedShiroPath != null) {
+ return initializedShiroPath;
+ }
+ return resolveShiroPath();
+ }
+
+ private String resolveShiroPath() {
String shiroPath = getAbsoluteDir(String.format("%s/shiro.ini", getConfDir()));
return new File(shiroPath).exists() ? shiroPath : StringUtils.EMPTY;
}
+ /**
+ * Capture the authentication mode once during server construction.
+ *
+ * Jetty's Shiro filter is also configured only at startup. Keeping this decision immutable
+ * prevents a later filesystem change from making notebook authorization believe the running,
+ * Shiro-protected server has switched to anonymous mode.
+ */
+ public synchronized void initializeAuthenticationMode() {
+ if (shiroPath == null) {
+ shiroPath = resolveShiroPath();
+ anonymousAllowed = StringUtils.isBlank(shiroPath);
+ }
+ }
+
public boolean isAuthenticationEnabled() {
return !StringUtils.isBlank(getShiroPath());
}
@@ -680,7 +703,7 @@ public boolean isPathWithScheme(String path){
public boolean isAnonymousAllowed() {
if (anonymousAllowed == null) {
- anonymousAllowed = this.getShiroPath().equals(StringUtils.EMPTY);
+ initializeAuthenticationMode();
}
return anonymousAllowed;
}
@@ -735,6 +758,10 @@ public String getWebsocketMaxTextMessageSize() {
return getString(ConfVars.ZEPPELIN_WEBSOCKET_MAX_TEXT_MESSAGE_SIZE);
}
+ public long getWebsocketAuthorizationRolesRefreshIntervalMs() {
+ return getLong(ConfVars.ZEPPELIN_WEBSOCKET_AUTHORIZATION_ROLES_REFRESH_INTERVAL_MS);
+ }
+
public String getJettyName() {
return getString(ConfVars.ZEPPELIN_SERVER_JETTY_NAME);
}
@@ -1058,6 +1085,8 @@ public enum ConfVars {
ZEPPELIN_CREDENTIALS_PERSIST("zeppelin.credentials.persist", true),
ZEPPELIN_CREDENTIALS_ENCRYPT_KEY("zeppelin.credentials.encryptKey", null),
ZEPPELIN_WEBSOCKET_MAX_TEXT_MESSAGE_SIZE("zeppelin.websocket.max.text.message.size", "10240000"),
+ ZEPPELIN_WEBSOCKET_AUTHORIZATION_ROLES_REFRESH_INTERVAL_MS(
+ "zeppelin.websocket.authorization.roles.refresh.interval.ms", 1000L),
ZEPPELIN_WEBSOCKET_PARAGRAPH_STATUS_PROGRESS("zeppelin.websocket.paragraph_status_progress.enable", true),
ZEPPELIN_SERVER_DEFAULT_DIR_ALLOWED("zeppelin.server.default.dir.allowed", false),
ZEPPELIN_SERVER_XFRAME_OPTIONS("zeppelin.server.xframe.options", "SAMEORIGIN"),
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/AuthorizationService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/AuthorizationService.java
index 70e7aac1592..1ded5ab0b20 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/AuthorizationService.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/AuthorizationService.java
@@ -26,6 +26,7 @@
import jakarta.inject.Inject;
import java.io.IOException;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@@ -38,7 +39,7 @@
public class AuthorizationService {
private static final Logger LOGGER = LoggerFactory.getLogger(AuthorizationService.class);
- private static final Set EMPTY_SET = new HashSet<>();
+ private static final Set EMPTY_SET = Collections.emptySet();
private final ZeppelinConfiguration zConf;
private final ConfigStorage configStorage;
@@ -49,6 +50,16 @@ public class AuthorizationService {
// cached note permission info. (noteId --> NoteAuth)
private Map notesAuth = new ConcurrentHashMap<>();
+ /**
+ * Monotonic generation for every effective ACL or cached-role change.
+ *
+ * Folder operations authorize a metadata snapshot outside this monitor, then reacquire the
+ * monitor through {@link #runWithAuthorizationVersion(long, AuthorizationOperation)} before
+ * mutating the repository. This prevents an ACL or role change from being interleaved between
+ * descendant authorization and a destructive folder mutation.
+ */
+ private long authorizationVersion;
+
@Inject
public AuthorizationService(NoteManager noteManager, ZeppelinConfiguration zConf,
ConfigStorage storage) {
@@ -84,9 +95,10 @@ public AuthorizationService(NoteManager noteManager, ZeppelinConfiguration zConf
* @param subject
* @throws IOException
*/
- public void createNoteAuth(String noteId, AuthenticationInfo subject) {
+ public synchronized void createNoteAuth(String noteId, AuthenticationInfo subject) {
NoteAuth noteAuth = new NoteAuth(noteId, subject, zConf);
this.notesAuth.put(noteId, noteAuth);
+ authorizationVersion++;
}
/**
@@ -98,8 +110,40 @@ public synchronized void saveNoteAuth() throws IOException {
configStorage.save(new NotebookAuthorizationInfoSaving(this.notesAuth));
}
- public void removeNoteAuth(String noteId) {
- this.notesAuth.remove(noteId);
+ public synchronized void removeNoteAuth(String noteId) {
+ if (this.notesAuth.remove(noteId) != null) {
+ authorizationVersion++;
+ }
+ }
+
+ public synchronized long getAuthorizationVersion() {
+ return authorizationVersion;
+ }
+
+ public synchronized boolean isAuthorizationVersionCurrent(long expectedVersion) {
+ return authorizationVersion == expectedVersion;
+ }
+
+ /**
+ * Run one operation only if its authorization preflight still belongs to the current ACL
+ * generation. ACL and cached-role mutations use the same monitor and therefore cannot be
+ * interleaved with the guarded operation.
+ */
+ public synchronized T runWithAuthorizationVersion(
+ long expectedVersion, AuthorizationOperation operation) throws IOException {
+ if (authorizationVersion != expectedVersion) {
+ throw new IOException("Notebook authorization changed while authorizing the operation");
+ }
+ return operation.run();
+ }
+
+ @FunctionalInterface
+ public interface AuthorizationOperation {
+ T run() throws IOException;
+ }
+
+ public boolean hasNoteAuth(String noteId) {
+ return this.notesAuth.containsKey(noteId);
}
// skip empty user and remove the white space around user name.
@@ -129,6 +173,15 @@ public void setRunners(String noteId, Set entities) throws IOException {
setRunners(noteId, entities, true);
}
+ public void setPermissions(
+ String noteId,
+ Set readers,
+ Set runners,
+ Set writers,
+ Set owners) throws IOException {
+ setPermissions(noteId, readers, runners, writers, owners, true);
+ }
+
public void setRoles(String user, Set roles) {
setRoles(user, roles, true);
}
@@ -137,61 +190,105 @@ public void clearPermission(String noteId) throws IOException {
clearPermission(noteId, true);
}
- public void setOwners(String noteId, Set entities, boolean broadcast) throws IOException {
+ public synchronized void setOwners(
+ String noteId, Set entities, boolean broadcast) throws IOException {
entities = normalizeUsers(entities);
NoteAuth noteAuth = notesAuth.get(noteId);
if (noteAuth == null) {
throw new IOException("No noteAuth found for noteId: " + noteId);
}
- noteAuth.setOwners(entities);
+ if (!noteAuth.getOwners().equals(entities)) {
+ noteAuth.setOwners(entities);
+ authorizationVersion++;
+ }
}
- public void setReaders(String noteId, Set entities, boolean broadcast) throws IOException {
+ public synchronized void setReaders(
+ String noteId, Set entities, boolean broadcast) throws IOException {
entities = normalizeUsers(entities);
NoteAuth noteAuth = notesAuth.get(noteId);
if (noteAuth == null) {
throw new IOException("No noteAuth found for noteId: " + noteId);
}
- noteAuth.setReaders(entities);
+ if (!noteAuth.getReaders().equals(entities)) {
+ noteAuth.setReaders(entities);
+ authorizationVersion++;
+ }
}
- public void setRunners(String noteId, Set entities, boolean broadcast) throws IOException {
+ public synchronized void setRunners(
+ String noteId, Set entities, boolean broadcast) throws IOException {
entities = normalizeUsers(entities);
NoteAuth noteAuth = notesAuth.get(noteId);
if (noteAuth == null) {
throw new IOException("No noteAuth found for noteId: " + noteId);
}
- noteAuth.setRunners(entities);
+ if (!noteAuth.getRunners().equals(entities)) {
+ noteAuth.setRunners(entities);
+ authorizationVersion++;
+ }
}
- public void setWriters(String noteId, Set entities, boolean broadcast) throws IOException {
+ public synchronized void setWriters(
+ String noteId, Set entities, boolean broadcast) throws IOException {
entities = normalizeUsers(entities);
NoteAuth noteAuth = notesAuth.get(noteId);
if (noteAuth == null) {
throw new IOException("No noteAuth found for noteId: " + noteId);
}
- noteAuth.setWriters(entities);
+ if (!noteAuth.getWriters().equals(entities)) {
+ noteAuth.setWriters(entities);
+ authorizationVersion++;
+ }
+ }
+
+ public synchronized void setPermissions(
+ String noteId,
+ Set readers,
+ Set runners,
+ Set writers,
+ Set owners,
+ boolean broadcast) throws IOException {
+ Set normalizedReaders = normalizeUsers(readers);
+ Set normalizedRunners = normalizeUsers(runners);
+ Set normalizedWriters = normalizeUsers(writers);
+ Set normalizedOwners = normalizeUsers(owners);
+ NoteAuth noteAuth = notesAuth.get(noteId);
+ if (noteAuth == null) {
+ throw new IOException("No noteAuth found for noteId: " + noteId);
+ }
+ NoteAuth.Permissions current = noteAuth.getPermissions();
+ if (!current.getReaders().equals(normalizedReaders)
+ || !current.getRunners().equals(normalizedRunners)
+ || !current.getWriters().equals(normalizedWriters)
+ || !current.getOwners().equals(normalizedOwners)) {
+ noteAuth.setPermissions(
+ normalizedReaders, normalizedRunners, normalizedWriters, normalizedOwners);
+ authorizationVersion++;
+ }
}
- public void setRoles(String user, Set roles, boolean broadcast) {
+ public synchronized void setRoles(String user, Set roles, boolean broadcast) {
if (StringUtils.isBlank(user)) {
LOGGER.warn("Setting roles for empty user");
return;
}
roles = normalizeUsers(roles);
- userRoles.put(user, roles);
+ Set immutableRoles = Collections.unmodifiableSet(new HashSet<>(roles));
+ Set previousRoles = userRoles.put(user, immutableRoles);
+ if (!immutableRoles.equals(previousRoles)) {
+ authorizationVersion++;
+ }
}
public void clearPermission(String noteId, boolean broadcast) throws IOException {
- NoteAuth noteAuth = notesAuth.get(noteId);
- if (noteAuth == null) {
- throw new IOException("No noteAuth found for noteId: " + noteId);
- }
- noteAuth.setReaders(new HashSet<>());
- noteAuth.setRunners(new HashSet<>());
- noteAuth.setWriters(new HashSet<>());
- noteAuth.setOwners(new HashSet<>());
-
+ setPermissions(
+ noteId,
+ Set.of(),
+ Set.of(),
+ Set.of(),
+ Set.of(),
+ broadcast);
}
public Set getOwners(String noteId) {
@@ -231,32 +328,52 @@ public Set getWriters(String noteId) {
}
public Set getRoles(String user) {
- return userRoles.getOrDefault(user, new HashSet<>());
+ return new HashSet<>(userRoles.getOrDefault(user, EMPTY_SET));
}
public boolean isOwner(String noteId, Set entities) {
- return isMember(entities, getOwners(noteId)) || isAdmin(entities);
+ NoteAuth noteAuth = notesAuth.get(noteId);
+ if (noteAuth == null) {
+ return false;
+ }
+ NoteAuth.Permissions permissions = noteAuth.getPermissions();
+ return isMember(entities, permissions.getOwners()) || isAdmin(entities);
}
public boolean isWriter(String noteId, Set entities) {
- return isMember(entities, getWriters(noteId)) ||
- isMember(entities, getOwners(noteId)) ||
- isAdmin(entities);
+ NoteAuth noteAuth = notesAuth.get(noteId);
+ if (noteAuth == null) {
+ return false;
+ }
+ NoteAuth.Permissions permissions = noteAuth.getPermissions();
+ return isMember(entities, permissions.getWriters())
+ || isMember(entities, permissions.getOwners())
+ || isAdmin(entities);
}
public boolean isReader(String noteId, Set entities) {
- return isMember(entities, getReaders(noteId)) ||
- isMember(entities, getOwners(noteId)) ||
- isMember(entities, getWriters(noteId)) ||
- isMember(entities, getRunners(noteId)) ||
- isAdmin(entities);
+ NoteAuth noteAuth = notesAuth.get(noteId);
+ if (noteAuth == null) {
+ return false;
+ }
+ NoteAuth.Permissions permissions = noteAuth.getPermissions();
+ return isMember(entities, permissions.getReaders())
+ || isMember(entities, permissions.getOwners())
+ || isMember(entities, permissions.getWriters())
+ || isMember(entities, permissions.getRunners())
+ || isAdmin(entities);
}
public boolean isRunner(String noteId, Set entities) {
- return isMember(entities, getRunners(noteId)) ||
- isMember(entities, getWriters(noteId)) ||
- isMember(entities, getOwners(noteId)) ||
- isAdmin(entities);
+ NoteAuth noteAuth = notesAuth.get(noteId);
+ if (noteAuth == null) {
+ return false;
+ }
+ NoteAuth.Permissions permissions = noteAuth.getPermissions();
+ return isMember(entities, permissions.getRunners())
+ || isMember(entities, permissions.getWriters())
+ || isMember(entities, permissions.getOwners())
+ || isAdmin(entities);
}
private boolean isAdmin(Set entities) {
@@ -275,6 +392,9 @@ private boolean isMember(Set a, Set b) {
}
public boolean isOwner(Set userAndRoles, String noteId) {
+ if (!hasNoteAuth(noteId)) {
+ return false;
+ }
if (zConf.isAnonymousAllowed()) {
LOGGER.debug("Zeppelin runs in anonymous mode, everybody is owner");
return true;
@@ -287,6 +407,9 @@ public boolean isOwner(Set userAndRoles, String noteId) {
//TODO(zjffdu) merge this hasWritePermission with isWriter ?
public boolean hasWritePermission(Set userAndRoles, String noteId) {
+ if (!hasNoteAuth(noteId)) {
+ return false;
+ }
if (zConf.isAnonymousAllowed()) {
LOGGER.debug("Zeppelin runs in anonymous mode, everybody is writer");
return true;
@@ -298,6 +421,9 @@ public boolean hasWritePermission(Set userAndRoles, String noteId) {
}
public boolean hasReadPermission(Set userAndRoles, String noteId) {
+ if (!hasNoteAuth(noteId)) {
+ return false;
+ }
if (zConf.isAnonymousAllowed()) {
LOGGER.debug("Zeppelin runs in anonymous mode, everybody is reader");
return true;
@@ -309,6 +435,9 @@ public boolean hasReadPermission(Set userAndRoles, String noteId) {
}
public boolean hasRunPermission(Set userAndRoles, String noteId) {
+ if (!hasNoteAuth(noteId)) {
+ return false;
+ }
if (zConf.isAnonymousAllowed()) {
LOGGER.debug("Zeppelin runs in anonymous mode, everybody is reader");
return true;
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java
index 8b7622e1ef1..c2fc31f2d3d 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Note.java
@@ -104,7 +104,7 @@ public class Note implements JsonSerializable {
* The fair behavior can therefore create a DeadLock.
*/
private transient final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(false);
- private transient boolean removed = false;
+ private transient volatile boolean removed = false;
private transient InterpreterFactory interpreterFactory;
private transient InterpreterSettingManager interpreterSettingManager;
private transient ParagraphJobListener paragraphJobListener;
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteAuth.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteAuth.java
index 7a715159e28..935fcff53d0 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteAuth.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteAuth.java
@@ -20,6 +20,7 @@
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.user.AuthenticationInfo;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -34,10 +35,7 @@ public class NoteAuth {
private final String noteId;
private final ZeppelinConfiguration zConf;
- private Set readers = new HashSet<>();
- private Set writers = new HashSet<>();
- private Set runners = new HashSet<>();
- private Set owners = new HashSet<>();
+ private volatile Permissions permissions = Permissions.empty();
public NoteAuth(String noteId, ZeppelinConfiguration zConf) {
this(noteId, AuthenticationInfo.ANONYMOUS, zConf);
@@ -60,62 +58,112 @@ public NoteAuth(String noteId, AuthenticationInfo subject, ZeppelinConfiguration
public NoteAuth(String noteId, Map> permissions, ZeppelinConfiguration zConf) {
this.noteId = noteId;
this.zConf = zConf;
- this.readers = permissions.getOrDefault("readers", new HashSet<>());
- this.writers = permissions.getOrDefault("writers", new HashSet<>());
- this.runners = permissions.getOrDefault("runners", new HashSet<>());
- this.owners = permissions.getOrDefault("owners", new HashSet<>());
+ this.permissions =
+ new Permissions(
+ immutableLoadedEntities(
+ permissions.getOrDefault("readers", Collections.emptySet())),
+ immutableLoadedEntities(
+ permissions.getOrDefault("runners", Collections.emptySet())),
+ immutableLoadedEntities(
+ permissions.getOrDefault("writers", Collections.emptySet())),
+ immutableLoadedEntities(
+ permissions.getOrDefault("owners", Collections.emptySet())));
}
// used when creating new note
- public void initPermissions(AuthenticationInfo subject) {
+ public synchronized void initPermissions(AuthenticationInfo subject) {
+ Set readers = Collections.emptySet();
+ Set writers = Collections.emptySet();
+ Set runners = Collections.emptySet();
+ Set owners = Collections.emptySet();
if (!AuthenticationInfo.isAnonymous(subject)) {
+ Set owner = Collections.singleton(checkCaseAndConvert(subject.getUser()));
if (zConf.isNotebookPublic()) {
// add current user to owners - can be public
- this.owners.add(checkCaseAndConvert(subject.getUser()));
+ owners = owner;
} else {
// add current user to owners, readers, runners, writers - private note
- this.owners.add(checkCaseAndConvert(subject.getUser()));
- this.readers.add(checkCaseAndConvert(subject.getUser()));
- this.writers.add(checkCaseAndConvert(subject.getUser()));
- this.runners.add(checkCaseAndConvert(subject.getUser()));
+ owners = owner;
+ readers = owner;
+ writers = owner;
+ runners = owner;
}
}
+ setPermissions(readers, runners, writers, owners);
}
public String getNoteId() {
return noteId;
}
- public void setOwners(Set entities) {
- this.owners = checkCaseAndConvert(entities);
- }
-
- public void setReaders(Set entities) {
- this.readers = checkCaseAndConvert(entities);
- }
-
- public void setWriters(Set entities) {
- this.writers = checkCaseAndConvert(entities);
- }
-
- public void setRunners(Set entities) {
- this.runners = checkCaseAndConvert(entities);
+ public synchronized void setOwners(Set entities) {
+ Permissions current = permissions;
+ permissions =
+ new Permissions(
+ current.getReaders(),
+ current.getRunners(),
+ current.getWriters(),
+ immutableEntities(entities));
+ }
+
+ public synchronized void setReaders(Set entities) {
+ Permissions current = permissions;
+ permissions =
+ new Permissions(
+ immutableEntities(entities),
+ current.getRunners(),
+ current.getWriters(),
+ current.getOwners());
+ }
+
+ public synchronized void setWriters(Set entities) {
+ Permissions current = permissions;
+ permissions =
+ new Permissions(
+ current.getReaders(),
+ current.getRunners(),
+ immutableEntities(entities),
+ current.getOwners());
+ }
+
+ public synchronized void setRunners(Set entities) {
+ Permissions current = permissions;
+ permissions =
+ new Permissions(
+ current.getReaders(),
+ immutableEntities(entities),
+ current.getWriters(),
+ current.getOwners());
+ }
+
+ public synchronized void setPermissions(
+ Set readers, Set runners, Set writers, Set owners) {
+ permissions =
+ new Permissions(
+ immutableEntities(readers),
+ immutableEntities(runners),
+ immutableEntities(writers),
+ immutableEntities(owners));
}
public Set getOwners() {
- return this.owners;
+ return permissions.getOwners();
}
public Set getReaders() {
- return this.readers;
+ return permissions.getReaders();
}
public Set getWriters() {
- return this.writers;
+ return permissions.getWriters();
}
public Set getRunners() {
- return this.runners;
+ return permissions.getRunners();
+ }
+
+ Permissions getPermissions() {
+ return permissions;
}
/*
@@ -129,10 +177,18 @@ private Set checkCaseAndConvert(Set entities) {
}
return set2;
} else {
- return entities;
+ return new HashSet<>(entities);
}
}
+ private Set immutableEntities(Set entities) {
+ return Collections.unmodifiableSet(checkCaseAndConvert(entities));
+ }
+
+ private Set immutableLoadedEntities(Set entities) {
+ return Collections.unmodifiableSet(new HashSet<>(entities));
+ }
+
private String checkCaseAndConvert(String entity) {
if (zConf.isUsernameForceLowerCase()) {
return entity.toLowerCase();
@@ -142,11 +198,54 @@ private String checkCaseAndConvert(String entity) {
}
public Map> toMap() {
- Map> map = new HashMap<>();
- map.put("readers", readers);
- map.put("writers", writers);
- map.put("runners", runners);
- map.put("owners", owners);
- return map;
+ return permissions.toMap();
+ }
+
+ static final class Permissions {
+ private final Set readers;
+ private final Set runners;
+ private final Set writers;
+ private final Set owners;
+
+ private Permissions(
+ Set readers, Set runners, Set writers, Set owners) {
+ this.readers = readers;
+ this.runners = runners;
+ this.writers = writers;
+ this.owners = owners;
+ }
+
+ private static Permissions empty() {
+ return new Permissions(
+ Collections.emptySet(),
+ Collections.emptySet(),
+ Collections.emptySet(),
+ Collections.emptySet());
+ }
+
+ Set getReaders() {
+ return readers;
+ }
+
+ Set getRunners() {
+ return runners;
+ }
+
+ Set getWriters() {
+ return writers;
+ }
+
+ Set getOwners() {
+ return owners;
+ }
+
+ Map> toMap() {
+ Map> map = new HashMap<>();
+ map.put("readers", readers);
+ map.put("writers", writers);
+ map.put("runners", runners);
+ map.put("owners", owners);
+ return map;
+ }
}
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteManager.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
index 0635fde994c..dad67301ac8 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
@@ -21,10 +21,12 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;
@@ -35,6 +37,7 @@
import org.apache.zeppelin.notebook.Notebook.NoteProcessor;
import org.apache.zeppelin.notebook.exception.NotePathAlreadyExistsException;
import org.apache.zeppelin.notebook.repo.NotebookRepo;
+import org.apache.zeppelin.notebook.repo.NotebookRepoWithVersionControl;
import org.apache.zeppelin.user.AuthenticationInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,6 +72,8 @@ public class NoteManager {
* operations never observe a tree and a mapping that belong to different generations.
*/
private volatile NoteTree noteTree;
+ private long metadataVersion;
+ private volatile Throwable metadataUnavailableCause;
@Inject
public NoteManager(NotebookRepo notebookRepo, ZeppelinConfiguration zConf) throws IOException {
@@ -102,9 +107,18 @@ private NoteTree buildNoteTree() throws IOException {
}
public Map getNotesInfo() {
+ assertMetadataAvailableUnchecked();
return this.noteTree.notesInfo;
}
+ /** Capture one immutable generation of the note-id/path index for authorization preflight. */
+ public synchronized NoteMetadataSnapshot getNotesInfoSnapshot() throws IOException {
+ assertMetadataAvailable();
+ return new NoteMetadataSnapshot(
+ metadataVersion,
+ Collections.unmodifiableMap(new LinkedHashMap<>(noteTree.notesInfo)));
+ }
+
/**
* Rebuild the notebook metadata from the NotebookRepo. The new tree is built completely
@@ -113,8 +127,11 @@ public Map getNotesInfo() {
*
* @throws IOException
*/
- public void reloadNotes() throws IOException {
- this.noteTree = buildNoteTree();
+ public synchronized void reloadNotes() throws IOException {
+ NoteTree reloadedTree = buildNoteTree();
+ this.noteTree = reloadedTree;
+ metadataUnavailableCause = null;
+ metadataVersion++;
}
/**
@@ -152,6 +169,7 @@ private void addOrUpdateNoteNode(NoteTree tree, NoteInfo noteInfo, boolean check
* @return
*/
public boolean containsNote(String notePath) {
+ assertMetadataAvailableUnchecked();
try {
getNoteNode(notePath);
return true;
@@ -167,6 +185,7 @@ public boolean containsNote(String notePath) {
* @return
*/
public boolean containsFolder(String folderPath) {
+ assertMetadataAvailableUnchecked();
try {
getFolder(folderPath);
return true;
@@ -183,23 +202,28 @@ public boolean containsFolder(String folderPath) {
* @param subject
* @throws IOException
*/
- public void saveNote(Note note, AuthenticationInfo subject) throws IOException {
+ public synchronized void saveNote(Note note, AuthenticationInfo subject) throws IOException {
+ assertMetadataAvailable();
if (note.isRemoved()) {
LOGGER.warn("Try to save note: {} when it is removed", note.getId());
} else {
- addOrUpdateNoteNode(this.noteTree, new NoteInfo(note), false);
- noteCache.putNote(note);
// Make sure to execute `notebookRepo.save()` successfully in concurrent context
// Otherwise, the NullPointerException will be thrown when invoking notebookRepo.get() in the following operations.
- synchronized (this) {
- this.notebookRepo.save(note, subject);
+ String previousPath = noteTree.notesInfo.get(note.getId());
+ addOrUpdateNoteNode(this.noteTree, new NoteInfo(note), false);
+ noteCache.putNote(note);
+ if (!StringUtils.equals(previousPath, note.getPath())) {
+ metadataVersion++;
}
+ this.notebookRepo.save(note, subject);
}
}
- public void addNote(Note note, AuthenticationInfo subject) throws IOException {
+ public synchronized void addNote(Note note, AuthenticationInfo subject) throws IOException {
+ assertMetadataAvailable();
addOrUpdateNoteNode(this.noteTree, new NoteInfo(note), true);
noteCache.putNote(note);
+ metadataVersion++;
}
/**
@@ -212,6 +236,32 @@ public void saveNote(Note note) throws IOException {
saveNote(note, AuthenticationInfo.ANONYMOUS);
}
+ /**
+ * Restore a note revision and publish it to the note cache and path index atomically with
+ * respect to note and folder moves.
+ */
+ public synchronized Note setNoteRevision(
+ String noteId,
+ String notePath,
+ String revisionId,
+ AuthenticationInfo subject) throws IOException {
+ assertMetadataAvailable();
+ String currentPath = noteTree.notesInfo.get(noteId);
+ if (currentPath == null) {
+ throw new IOException("No metadata found for this note: " + noteId);
+ }
+ if (!StringUtils.equals(currentPath, notePath)) {
+ throw new IOException("Note path changed while setting the revision");
+ }
+
+ Note note = ((NotebookRepoWithVersionControl) notebookRepo)
+ .setNoteRevision(noteId, notePath, revisionId, subject);
+ if (note != null) {
+ saveNote(note, subject);
+ }
+ return note;
+ }
+
/**
* Remove note from NotebookRepo and NoteManager
*
@@ -219,12 +269,14 @@ public void saveNote(Note note) throws IOException {
* @param subject
* @throws IOException
*/
- public void removeNote(String noteId, AuthenticationInfo subject) throws IOException {
+ public synchronized void removeNote(String noteId, AuthenticationInfo subject) throws IOException {
+ assertMetadataAvailable();
NoteTree tree = this.noteTree;
String notePath = tree.notesInfo.remove(noteId);
Folder folder = getOrCreateFolder(tree, getFolderName(notePath));
folder.removeNote(getNoteName(notePath));
noteCache.removeNote(noteId);
+ metadataVersion++;
this.notebookRepo.remove(noteId, notePath, subject);
}
@@ -235,64 +287,313 @@ public void moveNote(String noteId,
throw new IOException("No metadata found for this note: " + noteId);
}
- NoteTree tree = this.noteTree;
- if (!isNotePathAvailable(tree, newNotePath)) {
- throw new NotePathAlreadyExistsException("Note '" + newNotePath + "' existed");
+ String notePath;
+ synchronized (this) {
+ assertMetadataAvailable();
+ NoteTree tree = this.noteTree;
+ if (!isNotePathAvailable(tree, newNotePath)) {
+ throw new NotePathAlreadyExistsException("Note '" + newNotePath + "' existed");
+ }
+
+ notePath = tree.notesInfo.get(noteId);
+ NoteNode noteNode = getNoteNode(tree, notePath);
+
+ // Move durable state first. If the repository rejects the destination, the in-memory
+ // path index and cached note must remain on the source path.
+ this.notebookRepo.move(noteId, notePath, newNotePath, subject);
+
+ // move the old NoteNode from notePath to newNotePath
+ noteNode.getParent().removeNote(getNoteName(notePath));
+ noteNode.setNotePath(newNotePath);
+ String newParent = getFolderName(newNotePath);
+ Folder newFolder = getOrCreateFolder(tree, newParent);
+ newFolder.addNoteNode(noteNode);
+
+ // update noteInfo mapping
+ tree.notesInfo.put(noteId, newNotePath);
+ updateCachedNotePath(noteId, newNotePath);
+ metadataVersion++;
+
+ // The cache may evict the note while many notes are moved concurrently. Reload it through
+ // the new metadata path so the repository-backed object also receives the updated path.
+ if (!StringUtils.equals(notePath, newNotePath)) {
+ processNote(noteId,
+ note -> {
+ note.setPath(newNotePath);
+ return null;
+ });
+ }
+
+ // save note if note name is changed, because we need to update the note field in note json.
+ String oldNoteName = getNoteName(notePath);
+ String newNoteName = getNoteName(newNotePath);
+ if (!StringUtils.equals(oldNoteName, newNoteName)) {
+ processNote(noteId,
+ note -> {
+ this.notebookRepo.save(note, subject);
+ return null;
+ });
+ }
}
+ }
- // move the old NoteNode from notePath to newNotePath
- String notePath = tree.notesInfo.get(noteId);
- NoteNode noteNode = getNoteNode(tree, notePath);
- noteNode.getParent().removeNote(getNoteName(notePath));
- noteNode.setNotePath(newNotePath);
- String newParent = getFolderName(newNotePath);
- Folder newFolder = getOrCreateFolder(tree, newParent);
- newFolder.addNoteNode(noteNode);
+ public synchronized void moveFolder(String folderPath,
+ String newFolderPath,
+ AuthenticationInfo subject) throws IOException {
+ moveFolder(folderPath, newFolderPath, subject, -1);
+ }
- // update noteInfo mapping
- tree.notesInfo.put(noteId, newNotePath);
+ public synchronized void moveFolder(
+ String folderPath,
+ String newFolderPath,
+ AuthenticationInfo subject,
+ long expectedMetadataVersion) throws IOException {
+ moveFolder(folderPath, newFolderPath, subject, expectedMetadataVersion, true);
+ }
- // update notebookrepo
- this.notebookRepo.move(noteId, notePath, newNotePath, subject);
+ public synchronized void moveFolder(
+ String folderPath,
+ String newFolderPath,
+ AuthenticationInfo subject,
+ long expectedMetadataVersion,
+ boolean mergeExistingDestination) throws IOException {
- // Update path of the note
- if (!StringUtils.equals(notePath, newNotePath)) {
- processNote(noteId,
- note -> {
- note.setPath(newNotePath);
- return null;
- });
- }
+ assertMetadataVersion(expectedMetadataVersion);
- // save note if note name is changed, because we need to update the note field in note json.
- String oldNoteName = getNoteName(notePath);
- String newNoteName = getNoteName(newNotePath);
- if (!StringUtils.equals(oldNoteName, newNoteName)) {
- processNote(noteId,
- note -> {
- this.notebookRepo.save(note, subject);
- return null;
- });
+ NoteTree tree = this.noteTree;
+ Folder folder = getFolder(tree, folderPath);
+ String sourceFolderPath = folder.getPath();
+ String destinationFolderPath = normalizeFolderPath(newFolderPath);
+ if (StringUtils.equals(sourceFolderPath, destinationFolderPath)) {
+ return;
+ }
+ if (folder == tree.root) {
+ throw new IOException("Can not move the root folder");
+ }
+ if (destinationFolderPath.startsWith(sourceFolderPath + "/")) {
+ throw new IOException(
+ "Can not move folder '" + sourceFolderPath + "' into its own descendant");
+ }
+ if (containsNote(destinationFolderPath)) {
+ throw new NotePathAlreadyExistsException(
+ "Path '" + destinationFolderPath + "' existed");
}
- }
- public void moveFolder(String folderPath,
- String newFolderPath,
- AuthenticationInfo subject) throws IOException {
+ if (containsFolder(destinationFolderPath)) {
+ if (!mergeExistingDestination) {
+ throw new NotePathAlreadyExistsException(
+ "Path '" + destinationFolderPath + "' existed");
+ }
+ Folder destinationFolder = getFolder(tree, destinationFolderPath);
+ if (isSameOrDescendantFolder(destinationFolder, folder)) {
+ throw new IOException(
+ "Can not move folder '" + sourceFolderPath + "' into its own descendant");
+ }
+ mergeFolder(tree, folder, destinationFolder, subject);
+ return;
+ }
// update notebookrepo
- this.notebookRepo.move(folderPath, newFolderPath, subject);
+ this.notebookRepo.move(sourceFolderPath, destinationFolderPath, subject);
// update filesystem tree
- NoteTree tree = this.noteTree;
- Folder folder = getFolder(tree, folderPath);
folder.getParent().removeFolder(folder.getName(), subject);
- Folder newFolder = getOrCreateFolder(tree, newFolderPath);
+ Folder newFolder = getOrCreateFolder(tree, destinationFolderPath);
newFolder.getParent().addFolder(newFolder.getName(), folder);
// update notesInfo
for (NoteInfo noteInfo : folder.getNoteInfoRecursively()) {
tree.notesInfo.put(noteInfo.getId(), noteInfo.getPath());
+ updateCachedNotePath(noteInfo.getId(), noteInfo.getPath());
+ }
+ metadataVersion++;
+ }
+
+ private static String normalizeFolderPath(String folderPath) {
+ StringBuilder normalized = new StringBuilder();
+ for (String token : folderPath.split("/")) {
+ if (!StringUtils.isBlank(token)) {
+ normalized.append('/').append(token);
+ }
+ }
+ return normalized.length() == 0 ? "/" : normalized.toString();
+ }
+
+ private static boolean isSameOrDescendantFolder(Folder folder, Folder possibleAncestor) {
+ for (Folder current = folder; current != null; current = current.parent) {
+ if (current == possibleAncestor) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void mergeFolder(
+ NoteTree tree,
+ Folder sourceFolder,
+ Folder destinationFolder,
+ AuthenticationInfo subject) throws IOException {
+ String sourceFolderPath = sourceFolder.getPath();
+ String destinationFolderPath = destinationFolder.getPath();
+ List noteMoves = new ArrayList<>();
+ for (NoteInfo noteInfo : sourceFolder.getNoteInfoRecursively()) {
+ noteMoves.add(
+ new FolderNoteMove(
+ noteInfo.getId(),
+ noteInfo.getPath(),
+ rebasePath(noteInfo.getPath(), sourceFolderPath, destinationFolderPath)));
+ }
+ noteMoves.sort((first, second) -> first.sourcePath.compareTo(second.sourcePath));
+
+ assertFolderMergeDoesNotOverwrite(tree, sourceFolder, noteMoves, destinationFolderPath);
+ moveFolderNotesWithRollback(noteMoves, subject);
+
+ // Publish the in-memory change only after every durable note move succeeds.
+ sourceFolder.getParent().getFolders().remove(sourceFolder.getName());
+ mergeFolderTrees(sourceFolder, destinationFolder);
+ for (FolderNoteMove noteMove : noteMoves) {
+ tree.notesInfo.put(noteMove.noteId, noteMove.destinationPath);
+ updateCachedNotePath(noteMove.noteId, noteMove.destinationPath);
+ }
+ metadataVersion++;
+ }
+
+ private void assertFolderMergeDoesNotOverwrite(
+ NoteTree tree,
+ Folder sourceFolder,
+ List noteMoves,
+ String destinationFolderPath) throws IOException {
+ Set sourceNoteIds = new HashSet<>();
+ for (FolderNoteMove noteMove : noteMoves) {
+ sourceNoteIds.add(noteMove.noteId);
+ }
+ Set remainingNotePaths = new HashSet<>();
+ for (Map.Entry entry : tree.notesInfo.entrySet()) {
+ if (!sourceNoteIds.contains(entry.getKey())) {
+ remainingNotePaths.add(entry.getValue());
+ }
+ }
+
+ Set remainingFolderPaths = new HashSet<>();
+ collectFolderPathsExcept(tree.root, sourceFolder, remainingFolderPaths);
+
+ Set destinationFolderPaths = new HashSet<>();
+ collectRebasedFolderPaths(
+ sourceFolder, sourceFolder.getPath(), destinationFolderPath, destinationFolderPaths);
+ for (String folderPath : destinationFolderPaths) {
+ if (remainingNotePaths.contains(folderPath)) {
+ throw new NotePathAlreadyExistsException("Path '" + folderPath + "' existed");
+ }
+ }
+
+ Set destinationNotePaths = new HashSet<>();
+ for (FolderNoteMove noteMove : noteMoves) {
+ if (!destinationNotePaths.add(noteMove.destinationPath)
+ || remainingNotePaths.contains(noteMove.destinationPath)
+ || remainingFolderPaths.contains(noteMove.destinationPath)
+ || destinationFolderPaths.contains(noteMove.destinationPath)) {
+ throw new NotePathAlreadyExistsException(
+ "Path '" + noteMove.destinationPath + "' existed");
+ }
+ }
+ }
+
+ private static void collectFolderPathsExcept(
+ Folder folder, Folder excludedFolder, Set folderPaths) {
+ if (folder == excludedFolder) {
+ return;
+ }
+ folderPaths.add(folder.getPath());
+ for (Folder child : folder.getFolders().values()) {
+ collectFolderPathsExcept(child, excludedFolder, folderPaths);
+ }
+ }
+
+ private static void collectRebasedFolderPaths(
+ Folder folder,
+ String sourceFolderPath,
+ String destinationFolderPath,
+ Set folderPaths) {
+ folderPaths.add(rebasePath(folder.getPath(), sourceFolderPath, destinationFolderPath));
+ for (Folder child : folder.getFolders().values()) {
+ collectRebasedFolderPaths(
+ child, sourceFolderPath, destinationFolderPath, folderPaths);
+ }
+ }
+
+ private static String rebasePath(
+ String path, String sourceFolderPath, String destinationFolderPath) {
+ String relativePath = path.substring(sourceFolderPath.length());
+ if ("/".equals(destinationFolderPath)) {
+ return relativePath.isEmpty() ? "/" : relativePath;
+ }
+ return destinationFolderPath + relativePath;
+ }
+
+ private void moveFolderNotesWithRollback(
+ List noteMoves, AuthenticationInfo subject) throws IOException {
+ List attemptedMoves = new ArrayList<>();
+ try {
+ for (FolderNoteMove noteMove : noteMoves) {
+ // A repository move may copy or update part of its state before reporting failure.
+ // Record the attempt first so compensation also covers that ambiguous current move.
+ attemptedMoves.add(noteMove);
+ notebookRepo.move(
+ noteMove.noteId, noteMove.sourcePath, noteMove.destinationPath, subject);
+ }
+ } catch (IOException | RuntimeException failure) {
+ boolean rollbackFailed = false;
+ for (int i = attemptedMoves.size() - 1; i >= 0; i--) {
+ FolderNoteMove attemptedMove = attemptedMoves.get(i);
+ try {
+ notebookRepo.move(
+ attemptedMove.noteId,
+ attemptedMove.destinationPath,
+ attemptedMove.sourcePath,
+ subject);
+ } catch (IOException | RuntimeException rollbackFailure) {
+ failure.addSuppressed(rollbackFailure);
+ rollbackFailed = true;
+ }
+ }
+
+ if (rollbackFailed) {
+ // A failed compensation means the durable paths are no longer known. Poison metadata
+ // before attempting a reload so lock-free readers cannot use the stale tree meanwhile.
+ metadataUnavailableCause = failure;
+ for (FolderNoteMove noteMove : noteMoves) {
+ noteCache.removeNote(noteMove.noteId);
+ }
+ try {
+ reloadNotes();
+ } catch (IOException | RuntimeException reloadFailure) {
+ failure.addSuppressed(reloadFailure);
+ }
+ }
+ throw failure;
+ }
+ }
+
+ private static void mergeFolderTrees(Folder sourceFolder, Folder destinationFolder) {
+ for (Map.Entry entry : sourceFolder.getNotes().entrySet()) {
+ NoteNode noteNode = entry.getValue();
+ destinationFolder.getNotes().put(entry.getKey(), noteNode);
+ noteNode.setParent(destinationFolder);
+ noteNode.updateNotePath();
+ }
+
+ for (Map.Entry entry : sourceFolder.getFolders().entrySet()) {
+ Folder sourceChild = entry.getValue();
+ Folder destinationChild = destinationFolder.getFolder(entry.getKey());
+ if (destinationChild == null) {
+ destinationFolder.getFolders().put(entry.getKey(), sourceChild);
+ sourceChild.setParent(destinationFolder);
+ for (NoteNode noteNode : sourceChild.getNoteNodeRecursively()) {
+ noteNode.updateNotePath();
+ }
+ } else {
+ mergeFolderTrees(sourceChild, destinationChild);
+ }
}
}
@@ -304,6 +605,7 @@ public void moveFolder(String folderPath,
* @throws IOException
*/
public List getNoteInfoRecursively(String folderPath) throws IOException {
+ assertMetadataAvailable();
return getFolder(folderPath).getNoteInfoRecursively();
}
@@ -315,23 +617,167 @@ public List getNoteInfoRecursively(String folderPath) throws IOExcepti
* @return
* @throws IOException
*/
- public List removeFolder(String folderPath, AuthenticationInfo subject) throws IOException {
+ public synchronized List removeFolder(
+ String folderPath, AuthenticationInfo subject) throws IOException {
+ return removeFolder(folderPath, subject, -1);
+ }
- // update notebookrepo
- this.notebookRepo.remove(folderPath, subject);
+ public synchronized List removeFolder(
+ String folderPath,
+ AuthenticationInfo subject,
+ long expectedMetadataVersion) throws IOException {
+
+ return removeFolder(
+ folderPath, subject, expectedMetadataVersion, Collections.emptyList());
+ }
+
+ synchronized List removeFolder(
+ String folderPath,
+ AuthenticationInfo subject,
+ long expectedMetadataVersion,
+ List loadedNotes) throws IOException {
+
+ assertMetadataVersion(expectedMetadataVersion);
+
+ List newlyRemovedNotes = new ArrayList<>();
+ for (Note note : loadedNotes) {
+ if (!note.isRemoved()) {
+ note.setRemoved(true);
+ newlyRemovedNotes.add(note);
+ }
+ }
+
+ try {
+ // update notebookrepo
+ this.notebookRepo.remove(folderPath, subject);
+
+ // update filesystem tree
+ NoteTree tree = this.noteTree;
+ Folder folder = getFolder(tree, folderPath);
+ List noteInfos = folder.getNoteInfoRecursively();
+ if (folder == tree.trash) {
+ folder.clear();
+ } else {
+ folder.getParent().removeFolder(folder.getName(), subject);
+ }
+
+ // update notesInfo and evict the deleted notes from the cache, mirroring removeNote
+ for (NoteInfo noteInfo : noteInfos) {
+ tree.notesInfo.remove(noteInfo.getId());
+ this.noteCache.removeNote(noteInfo.getId());
+ }
+ metadataVersion++;
+
+ return noteInfos;
+ } catch (IOException | RuntimeException e) {
+ for (Note note : newlyRemovedNotes) {
+ note.setRemoved(false);
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * Restore every direct child of the trash against one authorized metadata generation.
+ * Structural changes are blocked for the full preflight and move sequence so a note cannot
+ * be added to the authorized folder after its ACL was checked.
+ *
+ * @return note-id to restored path for callers that need to report the restored entries
+ */
+ public synchronized Map restoreAllFromTrash(
+ AuthenticationInfo subject, long expectedMetadataVersion) throws IOException {
+ assertMetadataVersion(expectedMetadataVersion);
- // update filesystem tree
NoteTree tree = this.noteTree;
- Folder folder = getFolder(tree, folderPath);
- List noteInfos = folder.getParent().removeFolder(folder.getName(), subject);
+ List notes = new ArrayList<>(tree.trash.getNotes().values());
+ List folders = new ArrayList<>(tree.trash.getFolders().values());
+ Map restoredPaths = new LinkedHashMap<>();
+ Map destinations = new LinkedHashMap<>();
+ String trashPrefix = "/" + TRASH_FOLDER;
+
+ for (NoteNode noteNode : notes) {
+ String destination = noteNode.getNotePath().substring(trashPrefix.length());
+ checkRestoreDestination(destination, destinations);
+ }
+ for (Folder folder : folders) {
+ String destination = folder.getPath().substring(trashPrefix.length());
+ checkRestoreDestination(destination, destinations);
+ }
+
+ boolean mutated = false;
+ try {
+ for (NoteNode noteNode : notes) {
+ String noteId = noteNode.getNoteId();
+ String oldPath = noteNode.getNotePath();
+ String newPath = oldPath.substring(trashPrefix.length());
+ notebookRepo.move(noteId, oldPath, newPath, subject);
+ noteNode.getParent().removeNote(getNoteName(oldPath));
+ noteNode.setNotePath(newPath);
+ getOrCreateFolder(tree, getFolderName(newPath)).addNoteNode(noteNode);
+ tree.notesInfo.put(noteId, newPath);
+ updateCachedNotePath(noteId, newPath);
+ restoredPaths.put(noteId, newPath);
+ mutated = true;
+ }
+ for (Folder folder : folders) {
+ String oldPath = folder.getPath();
+ String newPath = oldPath.substring(trashPrefix.length());
+ notebookRepo.move(oldPath, newPath, subject);
+ folder.getParent().removeFolder(folder.getName(), subject);
+ Folder destination = getOrCreateFolder(tree, newPath);
+ destination.getParent().addFolder(destination.getName(), folder);
+ for (NoteInfo noteInfo : folder.getNoteInfoRecursively()) {
+ tree.notesInfo.put(noteInfo.getId(), noteInfo.getPath());
+ updateCachedNotePath(noteInfo.getId(), noteInfo.getPath());
+ restoredPaths.put(noteInfo.getId(), noteInfo.getPath());
+ }
+ mutated = true;
+ }
+ } finally {
+ if (mutated) {
+ metadataVersion++;
+ }
+ }
+ return restoredPaths;
+ }
+
+ private void checkRestoreDestination(
+ String destination, Map destinations) throws IOException {
+ if (destinations.put(destination, Boolean.TRUE) != null
+ || containsNote(destination)
+ || containsFolder(destination)) {
+ throw new NotePathAlreadyExistsException("Path '" + destination + "' existed");
+ }
+ }
+
+ private void assertMetadataVersion(long expectedMetadataVersion) throws IOException {
+ assertMetadataAvailable();
+ if (expectedMetadataVersion >= 0 && metadataVersion != expectedMetadataVersion) {
+ throw new IOException("Notebook metadata changed while authorizing the folder operation");
+ }
+ }
- // update notesInfo and evict the deleted notes from the cache, mirroring removeNote
- for (NoteInfo noteInfo : noteInfos) {
- tree.notesInfo.remove(noteInfo.getId());
- this.noteCache.removeNote(noteInfo.getId());
+ private void assertMetadataAvailable() throws IOException {
+ Throwable cause = metadataUnavailableCause;
+ if (cause != null) {
+ throw new IOException(
+ "Notebook metadata is unavailable after repository recovery failed", cause);
}
+ }
- return noteInfos;
+ private void assertMetadataAvailableUnchecked() {
+ Throwable cause = metadataUnavailableCause;
+ if (cause != null) {
+ throw new IllegalStateException(
+ "Notebook metadata is unavailable after repository recovery failed", cause);
+ }
+ }
+
+ private void updateCachedNotePath(String noteId, String notePath) {
+ Note note = noteCache.getNote(noteId);
+ if (note != null) {
+ note.setPath(notePath);
+ }
}
/**
@@ -345,6 +791,7 @@ public List removeFolder(String folderPath, AuthenticationInfo subject
*/
public T processNote(String noteId, boolean reload, NoteProcessor noteProcessor)
throws IOException {
+ assertMetadataAvailable();
// Read the tree once, so that the mapping lookup below and the tree traversal that
// follows it are both resolved against the same generation of the metadata.
NoteTree tree = this.noteTree;
@@ -353,6 +800,9 @@ public T processNote(String noteId, boolean reload, NoteProcessor notePro
}
String notePath = tree.notesInfo.get(noteId);
NoteNode noteNode = getNoteNode(tree, notePath);
+ if (!StringUtils.equals(noteId, noteNode.getNoteId())) {
+ throw new IOException("Note metadata changed while resolving note: " + noteId);
+ }
return noteNode.loadAndProcessNote(reload, noteProcessor);
}
@@ -374,6 +824,7 @@ public T processNote(String noteId, NoteProcessor noteProcessor) throws I
* @return
*/
public Folder getOrCreateFolder(String folderName) {
+ assertMetadataAvailableUnchecked();
return getOrCreateFolder(this.noteTree, folderName);
}
@@ -394,6 +845,9 @@ private NoteNode getNoteNode(String notePath) throws IOException {
private static NoteNode getNoteNode(NoteTree tree, String notePath) throws IOException {
String[] tokens = notePath.split("/");
+ if (tokens.length == 0) {
+ throw new IOException("Can not find note: " + notePath);
+ }
Folder curFolder = tree.root;
for (int i = 0; i < tokens.length - 1; ++i) {
if (!StringUtils.isBlank(tokens[i])) {
@@ -429,6 +883,7 @@ private static Folder getFolder(NoteTree tree, String folderPath) throws IOExcep
}
public Folder getTrashFolder() {
+ assertMetadataAvailableUnchecked();
return this.noteTree.trash;
}
@@ -461,10 +916,30 @@ private static boolean isNotePathAvailable(NoteTree tree, String notePath) {
}
public String getNoteIdByPath(String notePath) throws IOException {
+ assertMetadataAvailable();
NoteNode noteNode = getNoteNode(notePath);
return noteNode.getNoteId();
}
+ /** Immutable note metadata generation used to bind authorization to a later mutation. */
+ public static final class NoteMetadataSnapshot {
+ private final long version;
+ private final Map notesInfo;
+
+ NoteMetadataSnapshot(long version, Map notesInfo) {
+ this.version = version;
+ this.notesInfo = notesInfo;
+ }
+
+ public long getVersion() {
+ return version;
+ }
+
+ public Map getNotesInfo() {
+ return notesInfo;
+ }
+ }
+
/**
* The two indexes that together locate a note: the folder tree and the noteId -> notePath
* mapping. A note lookup resolves the id through the mapping and then walks the tree, so
@@ -484,6 +959,18 @@ private static class NoteTree {
}
}
+ private static final class FolderNoteMove {
+ private final String noteId;
+ private final String sourcePath;
+ private final String destinationPath;
+
+ private FolderNoteMove(String noteId, String sourcePath, String destinationPath) {
+ this.noteId = noteId;
+ this.sourcePath = sourcePath;
+ this.destinationPath = destinationPath;
+ }
+ }
+
/**
* Represent one folder that could contains sub folders and note files.
*/
@@ -592,6 +1079,11 @@ public List removeFolder(String folderName,
return folder.getNoteInfoRecursively();
}
+ private void clear() {
+ notes.clear();
+ subFolders.clear();
+ }
+
public List getNoteInfoRecursively() {
List notesInfo = new ArrayList<>();
for (NoteNode noteNode : this.notes.values()) {
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java
index 83f0032822f..b56591bba9c 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java
@@ -51,8 +51,6 @@
import org.apache.zeppelin.interpreter.InterpreterSetting;
import org.apache.zeppelin.interpreter.InterpreterSettingManager;
import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
-import org.apache.zeppelin.notebook.NoteManager.Folder;
-import org.apache.zeppelin.notebook.NoteManager.NoteNode;
import org.apache.zeppelin.notebook.repo.NotebookRepo;
import org.apache.zeppelin.notebook.repo.NotebookRepoSync;
import org.apache.zeppelin.notebook.repo.NotebookRepoWithVersionControl;
@@ -312,9 +310,16 @@ public String createNote(String notePath,
new Note(notePath, defaultInterpreterGroup, replFactory, interpreterSettingManager,
paragraphJobListener, credentials, noteEventListeners, zConf,
notebookRepo.getNoteParser());
- noteManager.addNote(note, subject);
- // init noteMeta
+ // Publish authorization before metadata. A folder operation must never observe a new note
+ // without its ACL and interpret the missing ACL as a public, empty permission set.
authorizationService.createNoteAuth(note.getId(), subject);
+ try {
+ noteManager.addNote(note, subject);
+ } catch (IOException e) {
+ authorizationService.removeNoteAuth(note.getId());
+ throw e;
+ }
+ // init noteMeta
authorizationService.saveNoteAuth();
if (save) {
noteManager.saveNote(note, subject);
@@ -430,8 +435,8 @@ private void removeNote(Note note, AuthenticationInfo subject) throws IOExceptio
// Set Remove to true to cancel saving this note
note.setRemoved(true);
noteManager.removeNote(note.getId(), subject);
- authorizationService.removeNoteAuth(note.getId());
fireNoteRemoveEvent(note, subject);
+ authorizationService.removeNoteAuth(note.getId());
}
public void removeCorruptedNote(String noteId, AuthenticationInfo subject) throws IOException {
@@ -546,49 +551,87 @@ public void moveNote(String noteId, String newNotePath, AuthenticationInfo subje
}
public void moveFolder(String folderPath, String newFolderPath, AuthenticationInfo subject) throws IOException {
+ moveFolder(folderPath, newFolderPath, subject, -1);
+ }
+
+ public void moveFolder(
+ String folderPath,
+ String newFolderPath,
+ AuthenticationInfo subject,
+ long expectedMetadataVersion) throws IOException {
+ moveFolder(folderPath, newFolderPath, subject, expectedMetadataVersion, true);
+ }
+
+ public void moveFolder(
+ String folderPath,
+ String newFolderPath,
+ AuthenticationInfo subject,
+ long expectedMetadataVersion,
+ boolean mergeExistingDestination) throws IOException {
LOGGER.info("Move folder from {} to {}", folderPath, newFolderPath);
- noteManager.moveFolder(folderPath, newFolderPath, subject);
+ noteManager.moveFolder(
+ folderPath,
+ newFolderPath,
+ subject,
+ expectedMetadataVersion,
+ mergeExistingDestination);
}
public void removeFolder(String folderPath, AuthenticationInfo subject) throws IOException {
+ removeFolder(folderPath, subject, -1);
+ }
+
+ public void removeFolder(
+ String folderPath,
+ AuthenticationInfo subject,
+ long expectedMetadataVersion) throws IOException {
LOGGER.info("Remove folder {}", folderPath);
- // Notes must be loaded and their remove listeners fired before the folder (and its
- // underlying repo storage) is deleted, otherwise the note content is no longer
- // available to run the same per-note cleanup as removeNote(String, AuthenticationInfo).
+ // Notes must be loaded before the folder (and its underlying repo storage) is deleted,
+ // otherwise the note content is no longer available to run the same per-note cleanup as
+ // removeNote(String, AuthenticationInfo). NoteManager marks these objects removed atomically
+ // with the deletion; listeners run only after the deletion succeeds.
List noteInfos = noteManager.getNoteInfoRecursively(folderPath);
+ Map loadedNotes = new HashMap<>();
for (NoteInfo noteInfo : noteInfos) {
processNote(noteInfo.getId(),
note -> {
if (note != null) {
- note.setRemoved(true);
- authorizationService.removeNoteAuth(note.getId());
- fireNoteRemoveEvent(note, subject);
+ loadedNotes.put(note.getId(), note);
}
return null;
});
}
- noteManager.removeFolder(folderPath, subject);
+ noteManager.removeFolder(
+ folderPath,
+ subject,
+ expectedMetadataVersion,
+ new ArrayList<>(loadedNotes.values()));
+ for (NoteInfo noteInfo : noteInfos) {
+ Note note = loadedNotes.get(noteInfo.getId());
+ if (note != null) {
+ fireNoteRemoveEvent(note, subject);
+ }
+ authorizationService.removeNoteAuth(noteInfo.getId());
+ }
}
public void emptyTrash(AuthenticationInfo subject) throws IOException {
+ emptyTrash(subject, -1);
+ }
+
+ public void emptyTrash(AuthenticationInfo subject, long expectedMetadataVersion)
+ throws IOException {
LOGGER.info("Empty Trash");
- removeFolder("/" + NoteManager.TRASH_FOLDER, subject);
+ removeFolder("/" + NoteManager.TRASH_FOLDER, subject, expectedMetadataVersion);
}
public void restoreAll(AuthenticationInfo subject) throws IOException {
- NoteManager.Folder trash = noteManager.getTrashFolder();
- // restore notes under trash folder
- // If the value changes in the loop, a concurrent modification exception is thrown.
- // Collector implementation of collect methods to maintain immutability.
- List notes = trash.getNotes().values().stream().collect(Collectors.toList());
- for (NoteManager.NoteNode noteNode : notes) {
- moveNote(noteNode.getNoteId(), noteNode.getNotePath().replace("/~Trash", ""), subject);
- }
- // restore folders under trash folder
- List folders = trash.getFolders().values().stream().collect(Collectors.toList());
- for (NoteManager.Folder folder : folders) {
- moveFolder(folder.getPath(), folder.getPath().replace("/~Trash", ""), subject);
- }
+ restoreAll(subject, -1);
+ }
+
+ public void restoreAll(AuthenticationInfo subject, long expectedMetadataVersion)
+ throws IOException {
+ noteManager.restoreAllFromTrash(subject, expectedMetadataVersion);
}
public Revision checkpointNote(String noteId, String notePath, String checkpointMessage,
@@ -615,10 +658,7 @@ public List listRevisionHistory(String noteId,
public Note setNoteRevision(String noteId, String notePath, String revisionId, AuthenticationInfo subject)
throws IOException {
if (((NotebookRepoSync) notebookRepo).isRevisionSupportedInDefaultRepo()) {
- Note note = ((NotebookRepoWithVersionControl) notebookRepo)
- .setNoteRevision(noteId, notePath, revisionId, subject);
- noteManager.saveNote(note);
- return note;
+ return noteManager.setNoteRevision(noteId, notePath, revisionId, subject);
} else {
return null;
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/repo/GitNotebookRepo.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/repo/GitNotebookRepo.java
index 80d336934b9..8f87d8aa778 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/repo/GitNotebookRepo.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/repo/GitNotebookRepo.java
@@ -22,6 +22,7 @@
import org.apache.zeppelin.notebook.NoteParser;
import org.apache.zeppelin.user.AuthenticationInfo;
import org.eclipse.jgit.api.Git;
+import org.eclipse.jgit.api.ResetCommand;
import org.eclipse.jgit.api.errors.GitAPIException;
import org.eclipse.jgit.api.errors.NoHeadException;
import org.eclipse.jgit.diff.DiffEntry;
@@ -31,6 +32,8 @@
import org.eclipse.jgit.lib.ObjectId;
import org.eclipse.jgit.lib.Repository;
import org.eclipse.jgit.revwalk.RevCommit;
+import org.eclipse.jgit.revwalk.RevWalk;
+import org.eclipse.jgit.treewalk.TreeWalk;
import org.eclipse.jgit.treewalk.filter.PathFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -75,38 +78,127 @@ public void init(ZeppelinConfiguration zConf, NoteParser noteParser) throws IOEx
}
@Override
- public void move(String noteId,
- String notePath,
- String newNotePath,
- AuthenticationInfo subject) throws IOException {
+ public synchronized void move(String noteId,
+ String notePath,
+ String newNotePath,
+ AuthenticationInfo subject) throws IOException {
+ ObjectId headBeforeMove = git.getRepository().resolve(Constants.HEAD);
super.move(noteId, notePath, newNotePath, subject);
String noteFileName = buildNoteFileName(noteId, notePath);
String newNoteFileName = buildNoteFileName(noteId, newNotePath);
try {
- git.rm().addFilepattern(noteFileName).call();
+ git.rm().setCached(true).addFilepattern(noteFileName).call();
git.add().addFilepattern(newNoteFileName).call();
git.commit().setMessage("Move note " + noteId + " from " + noteFileName + " to " +
newNoteFileName).call();
- } catch (GitAPIException e) {
- throw new IOException(e);
+ } catch (GitAPIException | RuntimeException e) {
+ if (headContainsMove(headBeforeMove, noteFileName, newNoteFileName, e)) {
+ LOGGER.warn(
+ "Git committed note move from {} to {} before reporting a hook failure; "
+ + "keeping the committed move",
+ notePath,
+ newNotePath,
+ e);
+ return;
+ }
+ throw rollbackFailedMove(
+ "Failed to commit note move from " + notePath + " to " + newNotePath,
+ e,
+ () -> super.move(noteId, newNotePath, notePath, subject),
+ noteFileName,
+ newNoteFileName);
}
}
@Override
- public void move(String folderPath, String newFolderPath,
- AuthenticationInfo subject) throws IOException {
+ public synchronized void move(String folderPath, String newFolderPath,
+ AuthenticationInfo subject) throws IOException {
+ ObjectId headBeforeMove = git.getRepository().resolve(Constants.HEAD);
super.move(folderPath, newFolderPath, subject);
+ String folderName = folderPath.substring(1);
+ String newFolderName = newFolderPath.substring(1);
try {
- git.rm().addFilepattern(folderPath.substring(1)).call();
- git.add().addFilepattern(newFolderPath.substring(1)).call();
+ git.rm().setCached(true).addFilepattern(folderName).call();
+ git.add().addFilepattern(newFolderName).call();
git.commit().setMessage("Move folder " + folderPath + " to " + newFolderPath).call();
- } catch (GitAPIException e) {
- throw new IOException(e);
+ } catch (GitAPIException | RuntimeException e) {
+ if (headContainsMove(headBeforeMove, folderName, newFolderName, e)) {
+ LOGGER.warn(
+ "Git committed folder move from {} to {} before reporting a hook failure; "
+ + "keeping the committed move",
+ folderPath,
+ newFolderPath,
+ e);
+ return;
+ }
+ throw rollbackFailedMove(
+ "Failed to commit folder move from " + folderPath + " to " + newFolderPath,
+ e,
+ () -> super.move(newFolderPath, folderPath, subject),
+ folderName,
+ newFolderName);
+ }
+ }
+
+ private IOException rollbackFailedMove(
+ String message,
+ Throwable cause,
+ IoAction rollback,
+ String... affectedPaths) {
+ IOException failure = new IOException(message, cause);
+ try {
+ rollback.run();
+ } catch (IOException rollbackFailure) {
+ failure.addSuppressed(rollbackFailure);
}
+
+ try {
+ ResetCommand reset = git.reset();
+ for (String affectedPath : affectedPaths) {
+ reset.addPath(affectedPath);
+ }
+ reset.call();
+ } catch (GitAPIException | RuntimeException resetFailure) {
+ failure.addSuppressed(resetFailure);
+ }
+ return failure;
+ }
+
+ private boolean headContainsMove(
+ ObjectId previousHead,
+ String sourcePath,
+ String destinationPath,
+ Throwable failure) {
+ try {
+ ObjectId currentHead = git.getRepository().resolve(Constants.HEAD);
+ boolean headChanged = previousHead == null
+ ? currentHead != null
+ : !previousHead.equals(currentHead);
+ if (!headChanged || currentHead == null) {
+ return false;
+ }
+ try (RevWalk revWalk = new RevWalk(git.getRepository())) {
+ RevCommit currentCommit = revWalk.parseCommit(currentHead);
+ try (TreeWalk source = TreeWalk.forPath(
+ git.getRepository(), sourcePath, currentCommit.getTree());
+ TreeWalk destination = TreeWalk.forPath(
+ git.getRepository(), destinationPath, currentCommit.getTree())) {
+ return source == null && destination != null;
+ }
+ }
+ } catch (IOException | RuntimeException headInspectionFailure) {
+ failure.addSuppressed(headInspectionFailure);
+ return false;
+ }
+ }
+
+ @FunctionalInterface
+ private interface IoAction {
+ void run() throws IOException;
}
@Override
- public void remove(String noteId, String notePath, AuthenticationInfo subject)
+ public synchronized void remove(String noteId, String notePath, AuthenticationInfo subject)
throws IOException {
super.remove(noteId, notePath, subject);
String noteFileName = buildNoteFileName(noteId, notePath);
@@ -119,7 +211,7 @@ public void remove(String noteId, String notePath, AuthenticationInfo subject)
}
@Override
- public void remove(String folderPath, AuthenticationInfo subject) throws IOException {
+ public synchronized void remove(String folderPath, AuthenticationInfo subject) throws IOException {
super.remove(folderPath, subject);
try {
git.rm().addFilepattern(folderPath.substring(1)).call();
@@ -137,10 +229,10 @@ public void remove(String folderPath, AuthenticationInfo subject) throws IOExcep
* @see org.apache.zeppelin.notebook.repo.VFSNotebookRepo#checkpoint(String, String)
*/
@Override
- public Revision checkpoint(String noteId,
- String notePath,
- String commitMessage,
- AuthenticationInfo subject) throws IOException {
+ public synchronized Revision checkpoint(String noteId,
+ String notePath,
+ String commitMessage,
+ AuthenticationInfo subject) throws IOException {
String noteFileName = buildNoteFileName(noteId, notePath);
Revision revision = Revision.EMPTY;
try {
@@ -229,8 +321,8 @@ public List revisionHistory(String noteId,
}
@Override
- public Note setNoteRevision(String noteId, String notePath, String revId,
- AuthenticationInfo subject)
+ public synchronized Note setNoteRevision(String noteId, String notePath, String revId,
+ AuthenticationInfo subject)
throws IOException {
Note revisionNote = get(noteId, notePath, revId, subject);
if (revisionNote != null) {
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java
index 32e433fff8f..db37c381104 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java
@@ -24,6 +24,7 @@
import java.net.URISyntaxException;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -189,11 +190,21 @@ public void move(String noteId,
buildNoteFileName(noteId, notePath), NameScope.DESCENDENT);
FileObject destFileObject = rootNotebookFileObject.resolveFile(
buildNoteFileName(noteId, newNotePath), NameScope.DESCENDENT);
+ if (destFileObject.exists() && !isSameLocalFile(fileObject, destFileObject)) {
+ throw new IOException("Destination note already exists: " + newNotePath);
+ }
// create parent folder first, otherwise move operation will fail
destFileObject.getParent().createFolder();
fileObject.moveTo(destFileObject);
}
+ private static boolean isSameLocalFile(FileObject source, FileObject destination)
+ throws IOException {
+ return "file".equalsIgnoreCase(source.getName().getScheme())
+ && "file".equalsIgnoreCase(destination.getName().getScheme())
+ && Files.isSameFile(source.getPath(), destination.getPath());
+ }
+
@Override
public void move(String folderPath, String newFolderPath,
AuthenticationInfo subject) throws IOException{
@@ -202,6 +213,9 @@ public void move(String folderPath, String newFolderPath,
folderPath.substring(1), NameScope.DESCENDENT);
FileObject destFileObject = rootNotebookFileObject.resolveFile(
newFolderPath.substring(1), NameScope.DESCENDENT);
+ if (destFileObject.exists() && !isSameLocalFile(fileObject, destFileObject)) {
+ throw new IOException("Destination folder already exists: " + newFolderPath);
+ }
// create parent folder first, otherwise move operation will fail
destFileObject.getParent().createFolder();
fileObject.moveTo(destFileObject);
@@ -266,4 +280,3 @@ public void updateSettings(Map settings, AuthenticationInfo subj
}
}
}
-
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/realm/ActiveDirectoryGroupRealm.java b/zeppelin-server/src/main/java/org/apache/zeppelin/realm/ActiveDirectoryGroupRealm.java
index 296b687797f..a4ebbbce854 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/realm/ActiveDirectoryGroupRealm.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/realm/ActiveDirectoryGroupRealm.java
@@ -27,7 +27,7 @@
import org.apache.shiro.authz.AuthorizationInfo;
import org.apache.shiro.authz.SimpleAuthorizationInfo;
import org.apache.shiro.realm.ldap.AbstractLdapRealm;
-import org.apache.shiro.realm.ldap.DefaultLdapContextFactory;
+import org.apache.shiro.realm.ldap.JndiLdapContextFactory;
import org.apache.shiro.realm.ldap.LdapContextFactory;
import org.apache.shiro.realm.ldap.LdapUtils;
import org.apache.shiro.subject.PrincipalCollection;
@@ -105,9 +105,7 @@ protected void onInit() {
public LdapContextFactory getLdapContextFactory() {
if (this.ldapContextFactory == null) {
LOGGER.debug("No LdapContextFactory specified - creating a default instance.");
- DefaultLdapContextFactory defaultFactory = new DefaultLdapContextFactory();
- defaultFactory.setPrincipalSuffix(this.principalSuffix);
- defaultFactory.setSearchBase(this.searchBase);
+ JndiLdapContextFactory defaultFactory = new JndiLdapContextFactory();
defaultFactory.setUrl(this.url);
defaultFactory.setSystemUsername(this.systemUsername);
defaultFactory.setSystemPassword(getSystemPassword());
@@ -292,6 +290,18 @@ public Map getListRoles() {
return roles;
}
+ /**
+ * Resolve every role for one principal with a single Active Directory query.
+ *
+ * This is used when Zeppelin captures an authenticated identity for REST or WebSocket.
+ * Calling {@code Subject.hasRole} once per configured role can otherwise repeat the same LDAP
+ * lookup when Shiro authorization caching is disabled.
+ */
+ public AuthorizationInfo queryForAuthorizationInfo(PrincipalCollection principals)
+ throws NamingException {
+ return queryForAuthorizationInfo(principals, getLdapContextFactory());
+ }
+
private Set getRoleNamesForUser(String username, LdapContext ldapContext)
throws NamingException {
Set roleNames = new LinkedHashSet<>();
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/realm/LdapRealm.java b/zeppelin-server/src/main/java/org/apache/zeppelin/realm/LdapRealm.java
index be8a0f0c68d..9a50f40e40d 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/realm/LdapRealm.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/realm/LdapRealm.java
@@ -48,7 +48,7 @@
import org.apache.hadoop.security.alias.CredentialProvider;
import org.apache.hadoop.security.alias.CredentialProviderFactory;
import org.apache.shiro.SecurityUtils;
-import org.apache.shiro.ShiroException;
+import org.apache.shiro.lang.ShiroException;
import org.apache.shiro.authc.AuthenticationInfo;
import org.apache.shiro.authc.AuthenticationToken;
import org.apache.shiro.authc.SimpleAuthenticationInfo;
@@ -59,6 +59,7 @@
import org.apache.shiro.crypto.hash.Hash;
import org.apache.shiro.crypto.hash.HashRequest;
import org.apache.shiro.crypto.hash.HashService;
+import org.apache.shiro.crypto.hash.SimpleHashProvider;
import org.apache.shiro.realm.ldap.DefaultLdapRealm;
import org.apache.shiro.realm.ldap.JndiLdapContextFactory;
import org.apache.shiro.realm.ldap.LdapContextFactory;
@@ -66,7 +67,7 @@
import org.apache.shiro.session.Session;
import org.apache.shiro.subject.MutablePrincipalCollection;
import org.apache.shiro.subject.PrincipalCollection;
-import org.apache.shiro.util.StringUtils;
+import org.apache.shiro.lang.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -143,7 +144,9 @@ public class LdapRealm extends DefaultLdapRealm {
private static final String DEFAULT_PRINCIPAL_REGEX = "(.*)";
private static final String MEMBER_SUBSTITUTION_TOKEN = "{0}";
- private static final String HASHING_ALGORITHM = "SHA-1";
+ private static final String HASHING_ALGORITHM = "SHA-256";
+ private static final int HASHING_ITERATIONS =
+ SimpleHashProvider.Parameters.DEFAULT_ITERATIONS;
private static final Logger LOGGER = LoggerFactory.getLogger(LdapRealm.class);
static {
@@ -200,6 +203,7 @@ public void setHadoopSecurityCredentialPath(String hadoopSecurityCredentialPath)
public LdapRealm() {
HashedCredentialsMatcher credentialsMatcher = new HashedCredentialsMatcher(HASHING_ALGORITHM);
+ credentialsMatcher.setHashIterations(HASHING_ITERATIONS);
setCredentialsMatcher(credentialsMatcher);
}
@@ -1029,7 +1033,10 @@ protected AuthenticationInfo createAuthenticationInfo(AuthenticationToken token,
HashRequest.Builder builder = new HashRequest.Builder();
Hash credentialsHash = hashService
.computeHash(builder.setSource(token.getCredentials())
- .setAlgorithmName(HASHING_ALGORITHM).build());
+ .setAlgorithmName(HASHING_ALGORITHM)
+ .addParameter(
+ SimpleHashProvider.Parameters.PARAMETER_ITERATIONS, HASHING_ITERATIONS)
+ .build());
return new SimpleAuthenticationInfo(token.getPrincipal(),
credentialsHash.toHex(), credentialsHash.getSalt(),
getName());
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/realm/kerberos/KerberosRealm.java b/zeppelin-server/src/main/java/org/apache/zeppelin/realm/kerberos/KerberosRealm.java
index 0a09853aa7f..beeee990aae 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/realm/kerberos/KerberosRealm.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/realm/kerberos/KerberosRealm.java
@@ -509,7 +509,7 @@ && getTokenMaxInactiveInterval() > 0) {
isCookiePersistent(), isHttps);
}
KerberosToken kerberosToken = new KerberosToken(token.getUserName(), token.toString());
- SecurityUtils.getSubject().login(kerberosToken);
+ loginIfNecessary(SecurityUtils.getSubject(), kerberosToken);
doFilter(filterChain, httpRequest, httpResponse);
}
} else {
@@ -548,6 +548,20 @@ && getTokenMaxInactiveInterval() > 0) {
}
}
+ /**
+ * Preserve the Shiro session when Hadoop's signed Kerberos token identifies the subject that
+ * is already authenticated. Shiro 2 rotates an existing session on every successful login;
+ * re-login on every REST request would therefore invalidate a WebSocket using that session.
+ */
+ void loginIfNecessary(
+ org.apache.shiro.subject.Subject shiroSubject, KerberosToken kerberosToken) {
+ if (shiroSubject.isAuthenticated()
+ && Objects.equals(shiroSubject.getPrincipal(), kerberosToken.getPrincipal())) {
+ return;
+ }
+ shiroSubject.login(kerberosToken);
+ }
+
/**
* It enforces the the Kerberos SPNEGO authentication sequence returning an
* {@link AuthenticationToken} only after the Kerberos SPNEGO sequence has
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/AbstractRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/AbstractRestApi.java
index 67b2ab2adcf..54ba594078d 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/AbstractRestApi.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/AbstractRestApi.java
@@ -18,14 +18,12 @@
package org.apache.zeppelin.rest;
import java.io.IOException;
-import java.util.HashSet;
-import java.util.Set;
import jakarta.ws.rs.WebApplicationException;
import org.apache.zeppelin.service.AuthenticationService;
import org.apache.zeppelin.service.ServiceContext;
+import org.apache.zeppelin.service.ServiceContextFactory;
import org.apache.zeppelin.service.SimpleServiceCallback;
-import org.apache.zeppelin.user.AuthenticationInfo;
import com.google.gson.Gson;
@@ -40,12 +38,7 @@ protected AbstractRestApi(AuthenticationService authenticationService) {
}
protected ServiceContext getServiceContext() {
- AuthenticationInfo authInfo = new AuthenticationInfo(authenticationService.getPrincipal());
- authInfo.setRoles(authenticationService.getAssociatedRoles());
- Set userAndRoles = new HashSet<>();
- userAndRoles.add(authenticationService.getPrincipal());
- userAndRoles.addAll(authenticationService.getAssociatedRoles());
- return new ServiceContext(authInfo, userAndRoles);
+ return ServiceContextFactory.create(authenticationService.getAuthenticatedIdentity());
}
public static class RestServiceCallback extends SimpleServiceCallback {
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/LoginRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/LoginRestApi.java
index d8b8c93b93e..48c72f620e4 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/LoginRestApi.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/LoginRestApi.java
@@ -16,6 +16,7 @@
*/
package org.apache.zeppelin.rest;
+import java.io.Serializable;
import java.text.ParseException;
import java.util.Collection;
import java.util.HashMap;
@@ -39,7 +40,9 @@
import org.apache.shiro.authc.AuthenticationToken;
import org.apache.shiro.authc.UsernamePasswordToken;
import org.apache.shiro.realm.Realm;
+import org.apache.shiro.session.Session;
import org.apache.shiro.subject.Subject;
+import org.apache.shiro.util.ThreadContext;
import org.apache.zeppelin.annotation.ZeppelinApi;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.notebook.AuthorizationService;
@@ -49,6 +52,7 @@
import org.apache.zeppelin.realm.kerberos.KerberosToken;
import org.apache.zeppelin.server.JsonResponse;
import org.apache.zeppelin.service.AuthenticationService;
+import org.apache.zeppelin.socket.ConnectionManager;
import org.apache.zeppelin.ticket.TicketContainer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,14 +68,17 @@ public class LoginRestApi extends AbstractRestApi {
private final ZeppelinConfiguration zConf;
private final AuthorizationService authorizationService;
+ private final ConnectionManager connectionManager;
@Inject
public LoginRestApi(ZeppelinConfiguration zConf,
AuthenticationService authenticationService,
- AuthorizationService authorizationService) {
+ AuthorizationService authorizationService,
+ ConnectionManager connectionManager) {
super(authenticationService);
this.zConf = zConf;
this.authorizationService = authorizationService;
+ this.connectionManager = connectionManager;
}
@GET
@@ -183,8 +190,11 @@ private JsonResponse> proceedToLogin(Subject currentUser, Au
JsonResponse> response = null;
try {
logoutCurrentUser();
- currentUser.getSession(true);
currentUser.login(token);
+ // Shiro rotates any pre-authentication session ID on successful login. Only ensure the
+ // session exists after login so the REST response and subsequent WebSocket upgrade use
+ // the final authenticated session rather than a fixation-prone pre-login session.
+ currentUser.getSession(true);
Set roles = authenticationService.getAssociatedRoles();
String principal = authenticationService.getPrincipal();
@@ -212,10 +222,9 @@ private JsonResponse> proceedToLogin(Subject currentUser, Au
}
/**
- * Post Login
- * Returns userName & password
- * for anonymous access, username is always anonymous.
- * After getting this ticket, access through websockets become safe
+ * Authenticate the Shiro session and return legacy UI identity metadata.
+ * For anonymous access, username is always anonymous. The response ticket is not a REST or
+ * WebSocket authentication credential; the resulting Shiro session cookie authenticates both.
*
* @return 200 response
*/
@@ -226,17 +235,9 @@ public Response postLogin(@FormParam("userName") String userName,
LOGGER.debug("userName: {}", userName);
// ticket set to anonymous for anonymous user. Simplify testing.
Subject currentUser = SecurityUtils.getSubject();
- if (currentUser.isAuthenticated()) {
- currentUser.logout();
- }
LOGGER.debug("currentUser: {}", currentUser);
- JsonResponse> response = null;
- if (!currentUser.isAuthenticated()) {
-
- UsernamePasswordToken token = new UsernamePasswordToken(userName, password);
-
- response = proceedToLogin(currentUser, token);
- }
+ UsernamePasswordToken token = new UsernamePasswordToken(userName, password);
+ JsonResponse> response = proceedToLogin(currentUser, token);
if (response == null) {
response = new JsonResponse<>(Response.Status.FORBIDDEN, "", null);
@@ -291,7 +292,13 @@ private String constructUrl(String providerURL, String redirectParam,
private void logoutCurrentUser() {
Subject currentUser = SecurityUtils.getSubject();
TicketContainer.instance.removeTicket(authenticationService.getPrincipal());
- currentUser.getSession().stop();
- currentUser.logout();
+ Session session = currentUser.getSession(false);
+ Serializable sessionId = session == null ? null : session.getId();
+ org.apache.shiro.mgt.SecurityManager securityManager = ThreadContext.getSecurityManager();
+ try {
+ currentUser.logout();
+ } finally {
+ connectionManager.closeConnectionsForSession(securityManager, sessionId);
+ }
}
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
index 192cd5056c5..109348623ec 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
@@ -301,10 +301,7 @@ public Response putNotePermissions(@PathParam("noteId") String noteId, String re
}
}
- authorizationService.setReaders(noteId, readers);
- authorizationService.setRunners(noteId, runners);
- authorizationService.setWriters(noteId, writers);
- authorizationService.setOwners(noteId, owners);
+ authorizationService.setPermissions(noteId, readers, runners, writers, owners);
LOGGER.debug("After set permissions {} {} {} {}", authorizationService.getOwners(noteId),
authorizationService.getReaders(noteId), authorizationService.getRunners(noteId),
authorizationService.getWriters(noteId));
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/SecurityRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/SecurityRestApi.java
index 89bc317734d..29b048702b4 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/SecurityRestApi.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/SecurityRestApi.java
@@ -52,10 +52,9 @@ public SecurityRestApi(AuthenticationService authenticationService) {
}
/**
- * Get ticket
- * Returns username & ticket
- * for anonymous access, username is always anonymous.
- * After getting this ticket, access through websockets become safe
+ * Return legacy UI identity metadata.
+ * For anonymous access, username is always anonymous. The returned ticket is retained for
+ * response compatibility and is not a REST or WebSocket authentication credential.
*
* @return 200 response
*/
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/server/JettyWebSocketUpgradeFilterInstaller.java b/zeppelin-server/src/main/java/org/apache/zeppelin/server/JettyWebSocketUpgradeFilterInstaller.java
new file mode 100644
index 00000000000..168b49df491
--- /dev/null
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/server/JettyWebSocketUpgradeFilterInstaller.java
@@ -0,0 +1,42 @@
+/*
+ * 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.server;
+
+import org.eclipse.jetty.servlet.FilterHolder;
+import org.eclipse.jetty.webapp.WebAppContext;
+import org.eclipse.jetty.websocket.servlet.WebSocketUpgradeFilter;
+import java.util.EnumSet;
+import jakarta.servlet.DispatcherType;
+
+/** Installs Jetty's default WebSocket upgrade filter after Zeppelin's authentication filter. */
+final class JettyWebSocketUpgradeFilterInstaller {
+
+ private JettyWebSocketUpgradeFilterInstaller() {
+ }
+
+ static FilterHolder installAfterAuthenticationFilter(
+ WebAppContext webApp, FilterHolder authenticationFilter) {
+ webApp.addFilter(
+ authenticationFilter, "/ws", EnumSet.allOf(DispatcherType.class));
+
+ FilterHolder filterHolder = new FilterHolder(WebSocketUpgradeFilter.class);
+ filterHolder.setName(WebSocketUpgradeFilter.class.getName());
+ filterHolder.setAsyncSupported(true);
+ webApp.addFilter(filterHolder, "/*", EnumSet.of(DispatcherType.REQUEST));
+ return filterHolder;
+ }
+}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
index b3f78816aec..725b98b0b99 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
@@ -92,6 +92,7 @@
import org.apache.zeppelin.search.NoSearchService;
import org.apache.zeppelin.search.SearchService;
import org.apache.zeppelin.service.*;
+import org.apache.zeppelin.service.AuthenticatedSessionService;
import org.apache.zeppelin.service.AuthenticationService;
import org.apache.zeppelin.service.auth.AuthenticationServiceFactory;
import org.apache.zeppelin.socket.ConnectionManager;
@@ -147,6 +148,7 @@ public ZeppelinServer(ZeppelinConfiguration zConf) throws IOException {
public ZeppelinServer(ZeppelinConfiguration zConf, String serviceLocatorName) throws IOException {
LOGGER.info("Instantiated ZeppelinServer");
this.zConf = zConf;
+ this.zConf.initializeAuthenticationMode();
if (zConf.isPrometheusMetricEnabled()) {
promMetricRegistry = Optional.of(new PrometheusMeterRegistry(PrometheusConfig.DEFAULT));
} else {
@@ -193,6 +195,7 @@ protected void configure() {
bind(AuthenticationServiceFactory.getAuthServiceClass(zConf))
.to(AuthenticationService.class)
.in(Singleton.class);
+ bindAsContract(AuthenticatedSessionService.class).in(Singleton.class);
bindAsContract(HeliumBundleFactory.class).in(Singleton.class);
bindAsContract(HeliumApplicationFactory.class).in(Singleton.class);
bindAsContract(ConfigurationService.class).in(Singleton.class);
@@ -563,9 +566,16 @@ private void setupRestApiContextHandler(WebAppContext webapp) {
String shiroIniPath = zConf.getShiroPath();
if (!StringUtils.isBlank(shiroIniPath)) {
webapp.setInitParameter("shiroConfigLocations", new File(shiroIniPath).toURI().toString());
- webapp
- .addFilter(ShiroFilter.class, "/api/*", EnumSet.allOf(DispatcherType.class))
- .setInitParameter("staticSecurityManagerEnabled", "true");
+ FilterHolder shiroFilter =
+ webapp.addFilter(
+ ShiroFilter.class, "/api/*", EnumSet.allOf(DispatcherType.class));
+ shiroFilter.setInitParameter("staticSecurityManagerEnabled", "true");
+
+ // Jetty's WebSocket initializer otherwise prepends its upgrade filter and bypasses Shiro
+ // after a successful upgrade. Pre-register the default-named filter after Shiro so the
+ // initializer reuses it in this order.
+ JettyWebSocketUpgradeFilterInstaller.installAfterAuthenticationFilter(
+ webapp, shiroFilter);
webapp.addEventListener(new EnvironmentLoaderListener());
}
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticatedIdentity.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticatedIdentity.java
new file mode 100644
index 00000000000..3fafa3141cf
--- /dev/null
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticatedIdentity.java
@@ -0,0 +1,72 @@
+/*
+ * 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.service;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Immutable authentication identity captured at a transport boundary.
+ *
+ * The session identifier is an opaque handle. It must not be exposed to clients or used as an
+ * authentication credential by itself.
+ */
+public final class AuthenticatedIdentity {
+
+ public static final String ANONYMOUS_PRINCIPAL = "anonymous";
+
+ private static final AuthenticatedIdentity ANONYMOUS =
+ new AuthenticatedIdentity(ANONYMOUS_PRINCIPAL, Collections.emptySet(), false, null);
+
+ private final String principal;
+ private final Set roles;
+ private final boolean authenticated;
+ private final Serializable sessionId;
+
+ public AuthenticatedIdentity(
+ String principal, Set roles, boolean authenticated, Serializable sessionId) {
+ this.principal = Objects.requireNonNull(principal, "principal");
+ this.roles = Collections.unmodifiableSet(new HashSet<>(Objects.requireNonNull(roles, "roles")));
+ this.authenticated = authenticated;
+ this.sessionId = sessionId;
+ }
+
+ public static AuthenticatedIdentity anonymous() {
+ return ANONYMOUS;
+ }
+
+ public String getPrincipal() {
+ return principal;
+ }
+
+ public Set getRoles() {
+ return roles;
+ }
+
+ public boolean isAuthenticated() {
+ return authenticated;
+ }
+
+ public Optional getSessionId() {
+ return Optional.ofNullable(sessionId);
+ }
+}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticatedSessionService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticatedSessionService.java
new file mode 100644
index 00000000000..70249e43ea9
--- /dev/null
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticatedSessionService.java
@@ -0,0 +1,229 @@
+/*
+ * 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.service;
+
+import java.io.Serializable;
+import java.time.Clock;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Objects;
+import java.util.Set;
+import jakarta.inject.Inject;
+import jakarta.inject.Provider;
+import org.apache.shiro.mgt.SecurityManager;
+import org.apache.shiro.session.Session;
+import org.apache.shiro.subject.Subject;
+
+/**
+ * Revalidates the Shiro session captured when a long-lived transport was established.
+ *
+ * The transport retains only an opaque session handle, never a thread-bound {@link Subject}.
+ * This lets REST and WebSocket requests use the same Shiro session lifecycle while ensuring that
+ * logout and expiry are observed before a WebSocket operation is dispatched.
+ */
+public class AuthenticatedSessionService {
+
+ private static final String ROLE_SNAPSHOT_SESSION_ATTRIBUTE =
+ AuthenticatedSessionService.class.getName() + ".roleSnapshot";
+
+ private final Provider authenticationServiceProvider;
+ private final Clock clock;
+
+ @Inject
+ public AuthenticatedSessionService(
+ Provider authenticationServiceProvider) {
+ this(authenticationServiceProvider, Clock.systemUTC());
+ }
+
+ AuthenticatedSessionService(AuthenticationService authenticationService) {
+ this(() -> authenticationService, Clock.systemUTC());
+ }
+
+ AuthenticatedSessionService(AuthenticationService authenticationService, Clock clock) {
+ this(() -> authenticationService, clock);
+ }
+
+ private AuthenticatedSessionService(
+ Provider authenticationServiceProvider, Clock clock) {
+ this.authenticationServiceProvider = authenticationServiceProvider;
+ this.clock = clock;
+ }
+
+ /** Validate that a transport's captured session is still authenticated without touching it. */
+ public void validate(
+ AuthenticatedIdentity connectionIdentity, SecurityManager securityManager) {
+ try {
+ restoreValidatedSubject(connectionIdentity, securityManager);
+ } catch (SessionAuthenticationException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ throw new SessionAuthenticationException("Authenticated session is no longer valid", e);
+ }
+ }
+
+ /**
+ * Revalidate a transport identity and return a fresh principal/role snapshot.
+ *
+ * @param connectionIdentity identity captured at the transport handshake
+ * @param securityManager exact Shiro security manager that authenticated the handshake
+ * @param touchSession whether this operation should extend the Shiro session idle timeout
+ * @return a fresh server-authenticated identity
+ * @throws SessionAuthenticationException when the session is missing, expired or changed
+ */
+ public AuthenticatedIdentity refresh(
+ AuthenticatedIdentity connectionIdentity,
+ SecurityManager securityManager,
+ boolean touchSession) {
+ return refresh(connectionIdentity, securityManager, touchSession, 0);
+ }
+
+ /**
+ * Revalidate a transport identity while allowing a recent role snapshot to be reused.
+ *
+ * The Shiro session and principal are validated on every call. Only the role set may be
+ * reused, which bounds role-revocation latency without delaying logout or session-expiry
+ * detection. A non-positive {@code maxRoleAgeMillis} disables reuse and has the same strict
+ * semantics as the three-argument overload.
+ *
+ * @param connectionIdentity identity captured at the transport handshake
+ * @param securityManager exact Shiro security manager that authenticated the handshake
+ * @param touchSession whether this operation should extend the Shiro session idle timeout
+ * @param maxRoleAgeMillis maximum age of a reusable role snapshot, in milliseconds
+ * @return a server-authenticated identity with current or recently refreshed roles
+ * @throws SessionAuthenticationException when the session is missing, expired or changed
+ */
+ public AuthenticatedIdentity refresh(
+ AuthenticatedIdentity connectionIdentity,
+ SecurityManager securityManager,
+ boolean touchSession,
+ long maxRoleAgeMillis) {
+ Objects.requireNonNull(connectionIdentity, "connectionIdentity");
+
+ try {
+ Subject subject = restoreValidatedSubject(connectionIdentity, securityManager);
+ if (subject == null) {
+ return AuthenticatedIdentity.anonymous();
+ }
+ Session session = subject.getSession(false);
+ if (touchSession) {
+ session.touch();
+ }
+
+ long roleLookupStartedAtMillis = clock.millis();
+ if (maxRoleAgeMillis > 0) {
+ Object cached = session.getAttribute(ROLE_SNAPSHOT_SESSION_ATTRIBUTE);
+ if (cached instanceof RoleSnapshot) {
+ RoleSnapshot roleSnapshot = (RoleSnapshot) cached;
+ if (roleSnapshot.isReusableFor(
+ connectionIdentity, roleLookupStartedAtMillis, maxRoleAgeMillis)) {
+ return roleSnapshot.toIdentity();
+ }
+ }
+ }
+
+ AuthenticatedIdentity refreshed =
+ subject.execute(
+ () -> authenticationServiceProvider.get().getAuthenticatedIdentity());
+ Serializable sessionId = connectionIdentity.getSessionId().orElseThrow(
+ () -> new SessionAuthenticationException("Authenticated session is unavailable"));
+ if (!refreshed.isAuthenticated()
+ || !connectionIdentity.getPrincipal().equals(refreshed.getPrincipal())
+ || !refreshed.getSessionId().filter(sessionId::equals).isPresent()) {
+ throw new SessionAuthenticationException("Authenticated session identity changed");
+ }
+ session.setAttribute(
+ ROLE_SNAPSHOT_SESSION_ATTRIBUTE,
+ new RoleSnapshot(refreshed, roleLookupStartedAtMillis));
+ return refreshed;
+ } catch (SessionAuthenticationException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ throw new SessionAuthenticationException("Authenticated session is no longer valid", e);
+ }
+ }
+
+ private Subject restoreValidatedSubject(
+ AuthenticatedIdentity connectionIdentity, SecurityManager securityManager) {
+ Objects.requireNonNull(connectionIdentity, "connectionIdentity");
+ if (!connectionIdentity.isAuthenticated()
+ && AuthenticatedIdentity.ANONYMOUS_PRINCIPAL.equals(
+ connectionIdentity.getPrincipal())
+ && connectionIdentity.getSessionId().isEmpty()) {
+ // Reaching the endpoint means Shiro's configured `/ws` chain admitted this anonymous
+ // handshake, or authentication is disabled entirely.
+ return null;
+ }
+
+ Serializable sessionId = connectionIdentity.getSessionId().orElseThrow(
+ () -> new SessionAuthenticationException("Authenticated session is unavailable"));
+ if (!connectionIdentity.isAuthenticated()) {
+ throw new SessionAuthenticationException("Authenticated session is unavailable");
+ }
+ if (securityManager == null) {
+ throw new SessionAuthenticationException("Authentication manager is unavailable");
+ }
+
+ Subject subject = restoreSubject(sessionId, securityManager);
+ Session session = subject.getSession(false);
+ if (session == null || !subject.isAuthenticated() || !sessionId.equals(session.getId())) {
+ throw new SessionAuthenticationException("Authenticated session is no longer valid");
+ }
+ String principal =
+ subject.execute(() -> authenticationServiceProvider.get().getPrincipal());
+ if (!connectionIdentity.getPrincipal().equals(principal)) {
+ throw new SessionAuthenticationException("Authenticated session identity changed");
+ }
+ return subject;
+ }
+
+ Subject restoreSubject(Serializable sessionId, SecurityManager securityManager) {
+ return new Subject.Builder(securityManager)
+ .sessionId(sessionId)
+ .sessionCreationEnabled(false)
+ .buildSubject();
+ }
+
+ private static final class RoleSnapshot implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final String principal;
+ private final Set roles;
+ private final Serializable sessionId;
+ private final long capturedAtMillis;
+
+ private RoleSnapshot(AuthenticatedIdentity identity, long capturedAtMillis) {
+ this.principal = identity.getPrincipal();
+ this.roles = Collections.unmodifiableSet(new HashSet<>(identity.getRoles()));
+ this.sessionId = identity.getSessionId().orElse(null);
+ this.capturedAtMillis = capturedAtMillis;
+ }
+
+ private boolean isReusableFor(
+ AuthenticatedIdentity connectionIdentity, long nowMillis, long maxRoleAgeMillis) {
+ long ageMillis = nowMillis - capturedAtMillis;
+ return ageMillis >= 0
+ && ageMillis < maxRoleAgeMillis
+ && principal.equals(connectionIdentity.getPrincipal())
+ && connectionIdentity.getSessionId().filter(sessionId::equals).isPresent();
+ }
+
+ private AuthenticatedIdentity toIdentity() {
+ return new AuthenticatedIdentity(principal, roles, true, sessionId);
+ }
+ }
+}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticationService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticationService.java
index eaab4d4e425..1822ed6f77d 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticationService.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/AuthenticationService.java
@@ -29,6 +29,13 @@
*/
public interface AuthenticationService {
+ /**
+ * Capture the current principal, roles and session handle as one identity snapshot.
+ *
+ * @return the identity bound to the current request
+ */
+ AuthenticatedIdentity getAuthenticatedIdentity();
+
/**
* Get current principal/username.
* @return
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/ConfigurationService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/ConfigurationService.java
index a3b226770d6..3938e23c3c3 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/ConfigurationService.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/ConfigurationService.java
@@ -47,6 +47,13 @@ public int getWsMaxMessageSize() {
return Integer.parseInt(zConf.getWebsocketMaxTextMessageSize());
}
+ /** Properties safe to expose to every authenticated notebook client. */
+ public Map getClientProperties() {
+ return Map.of(
+ ZeppelinConfiguration.ConfVars.ZEPPELIN_WEBSOCKET_MAX_TEXT_MESSAGE_SIZE.getVarName(),
+ zConf.getWebsocketMaxTextMessageSize());
+ }
+
public Map getPropertiesWithPrefix(String prefix,
ServiceContext context,
ServiceCallback> callback)
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NoAuthenticationService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NoAuthenticationService.java
index d41273b3aa3..174b469fe3f 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NoAuthenticationService.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NoAuthenticationService.java
@@ -38,6 +38,11 @@ public NoAuthenticationService() {
LOGGER.info("NoAuthenticationService is initialized");
}
+ @Override
+ public AuthenticatedIdentity getAuthenticatedIdentity() {
+ return AuthenticatedIdentity.anonymous();
+ }
+
@Override
public String getPrincipal() {
return ANONYMOUS;
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java
index 554b85f4de2..f99fed621de 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java
@@ -82,8 +82,9 @@
public class NotebookService {
private static final Logger LOGGER = LoggerFactory.getLogger(NotebookService.class);
+ private static final String TRASH_PATH = "/" + NoteManager.TRASH_FOLDER;
private static final DateTimeFormatter TRASH_CONFLICT_TIMESTAMP_FORMATTER =
- DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneId.systemDefault());
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH-mm-ss").withZone(ZoneId.systemDefault());
private final ZeppelinConfiguration zConf;
private final Notebook notebook;
@@ -108,6 +109,10 @@ public String getHomeNote(ServiceContext context,
if (StringUtils.isBlank(noteId)) {
callback.onSuccess(null, context);
} else {
+ if (!checkPermission(
+ noteId, Permission.READER, Message.OP.GET_HOME_NOTE, context, callback)) {
+ return noteId;
+ }
notebook.processNote(noteId,
note -> {
if (note != null && !checkPermission(noteId, Permission.READER, Message.OP.GET_HOME_NOTE, context,
@@ -133,6 +138,11 @@ public T getNote(String noteId,
ServiceContext context,
ServiceCallback callback,
NoteProcessor noteProcessor) throws IOException {
+ // Authorize before processNote: a reload reads from the repository and replaces the
+ // shared cache entry, so it must never be triggered by a user who cannot read the note.
+ if (!checkPermission(noteId, Permission.READER, Message.OP.GET_NOTE, context, callback)) {
+ return null;
+ }
return notebook.processNote(noteId, reload,
note -> {
if (note == null) {
@@ -239,11 +249,17 @@ String normalizeNotePath(String notePath) throws IOException {
notePath = notePath.replace("\r", " ").replace("\n", " ");
- notePath = NotebookPathValidator.decodeRepeatedly(notePath);
+ notePath = NotebookPathValidator.decodeRepeatedly(notePath)
+ .replace("\r", " ").replace("\n", " ");
if (notePath.endsWith("/")) {
throw new IOException("Note name shouldn't end with '/'");
}
+ NotebookPathValidator.rejectTraversalSegments(notePath);
+ if (notePath.contains("//")) {
+ throw new IOException("Empty path segments are not allowed");
+ }
+
int pos = notePath.lastIndexOf("/");
if ((notePath.length() - pos) > 255) {
throw new IOException("Note name must be less than 255");
@@ -687,13 +703,13 @@ public void restoreNote(String noteId,
return null;
}
- if (!note.getPath().startsWith("/" + NoteManager.TRASH_FOLDER)) {
+ if (!isTrashDescendant(note.getPath())) {
callback.onFailure(new IOException("Can not restore this note " + note.getPath() +
" as it is not in trash folder"), context);
return null;
}
try {
- String destNotePath = note.getPath().replace("/" + NoteManager.TRASH_FOLDER, "");
+ String destNotePath = restoreDestination(note.getPath());
notebook.moveNote(noteId, destNotePath, context.getAutheInfo());
callback.onSuccess(note, context);
} catch (IOException e) {
@@ -709,17 +725,34 @@ public void restoreFolder(String folderPath,
ServiceContext context,
ServiceCallback callback) throws IOException {
- if (!folderPath.startsWith("/" + NoteManager.TRASH_FOLDER)) {
- callback.onFailure(new IOException("Can not restore this folder: " + folderPath +
+ String normalizedFolderPath = normalizeNotePath(folderPath);
+ if (!isTrashDescendant(normalizedFolderPath)) {
+ callback.onFailure(new IOException("Can not restore this folder: " + normalizedFolderPath +
" as it is not in trash folder"), context);
return;
}
+ FolderPermissionSnapshot authorization = checkFolderPermission(
+ normalizedFolderPath, Permission.WRITER, Message.OP.RESTORE_FOLDER, context, callback);
+ if (authorization == null) {
+ return;
+ }
try {
- String destFolderPath = folderPath.replace("/" + NoteManager.TRASH_FOLDER, "");
- notebook.moveFolder(folderPath, destFolderPath, context.getAutheInfo());
+ String destFolderPath = restoreDestination(normalizedFolderPath);
+ authorizationService.runWithAuthorizationVersion(
+ authorization.getAuthorizationVersion(),
+ () -> {
+ notebook.moveFolder(
+ normalizedFolderPath,
+ destFolderPath,
+ context.getAutheInfo(),
+ authorization.getMetadata().getVersion(),
+ false);
+ return null;
+ });
callback.onSuccess(null, context);
} catch (IOException e) {
- callback.onFailure(new IOException("Fail to restore folder: " + folderPath, e), context);
+ callback.onFailure(
+ new IOException("Fail to restore folder: " + normalizedFolderPath, e), context);
}
}
@@ -728,8 +761,23 @@ public void restoreFolder(String folderPath,
public void restoreAll(ServiceContext context,
ServiceCallback callback) throws IOException {
+ FolderPermissionSnapshot authorization = checkFolderPermission(
+ "/" + NoteManager.TRASH_FOLDER,
+ Permission.WRITER,
+ Message.OP.RESTORE_ALL,
+ context,
+ callback);
+ if (authorization == null) {
+ return;
+ }
try {
- notebook.restoreAll(context.getAutheInfo());
+ authorizationService.runWithAuthorizationVersion(
+ authorization.getAuthorizationVersion(),
+ () -> {
+ notebook.restoreAll(
+ context.getAutheInfo(), authorization.getMetadata().getVersion());
+ return null;
+ });
callback.onSuccess(null, context);
} catch (IOException e) {
callback.onFailure(new IOException("Fail to restore all", e), context);
@@ -998,6 +1046,10 @@ public void removeNoteForms(String noteId,
String formName,
ServiceContext context,
ServiceCallback callback) throws IOException {
+ if (!checkPermission(noteId, Permission.WRITER, Message.OP.REMOVE_NOTE_FORMS, context,
+ callback)) {
+ return;
+ }
notebook.processNote(noteId,
note -> {
if (note == null) {
@@ -1024,13 +1076,18 @@ public NotebookRepoWithVersionControl.Revision checkpointNote(
ServiceContext context,
ServiceCallback callback) throws IOException {
+ if (!checkPermission(noteId, Permission.WRITER, Message.OP.CHECKPOINT_NOTE, context,
+ callback)) {
+ return null;
+ }
+
NotebookRepoWithVersionControl.Revision revision = notebook.processNote(noteId,
note -> {
if (note == null) {
callback.onFailure(new NoteNotFoundException(noteId), context);
return null;
}
- if (!checkPermission(noteId, Permission.WRITER, Message.OP.REMOVE_NOTE_FORMS, context,
+ if (!checkPermission(noteId, Permission.WRITER, Message.OP.CHECKPOINT_NOTE, context,
callback)) {
return null;
}
@@ -1046,6 +1103,10 @@ public List listRevisionHistory(
ServiceContext context,
ServiceCallback> callback) throws IOException {
+ if (!checkPermission(
+ noteId, Permission.READER, Message.OP.LIST_REVISION_HISTORY, context, callback)) {
+ return null;
+ }
List revisions = notebook.processNote(noteId,
note -> {
@@ -1053,16 +1114,12 @@ public List listRevisionHistory(
callback.onFailure(new NoteNotFoundException(noteId), context);
return null;
}
- // TODO(zjffdu) Disable checking permission for now, otherwise zeppelin will send 2 AUTH_INFO
- // message to frontend when frontend try to get note without proper privilege.
- // if (!checkPermission(noteId, Permission.READER, Message.OP.LIST_REVISION_HISTORY, context,
- // callback)) {
- // return null;
- // }
return notebook.listRevisionHistory(noteId, note.getPath(), context.getAutheInfo());
});
- callback.onSuccess(revisions, context);
+ if (revisions != null) {
+ callback.onSuccess(revisions, context);
+ }
return revisions;
}
@@ -1072,6 +1129,11 @@ public void setNoteRevision(String noteId,
ServiceContext context,
ServiceCallback callback) throws IOException {
+ if (!checkPermission(noteId, Permission.WRITER, Message.OP.SET_NOTE_REVISION, context,
+ callback)) {
+ return;
+ }
+
notebook.processNote(noteId,
note -> {
if (note == null) {
@@ -1103,6 +1165,11 @@ public Note getNotebyRevision(String noteId,
ServiceContext context,
ServiceCallback callback) throws IOException {
+ if (!checkPermission(noteId, Permission.READER, Message.OP.NOTE_REVISION, context,
+ callback)) {
+ return null;
+ }
+
return notebook.processNote(noteId,
note -> {
if (note == null) {
@@ -1126,6 +1193,11 @@ public void getNoteByRevisionForCompare(String noteId,
ServiceContext context,
ServiceCallback callback) throws IOException {
+ if (!checkPermission(noteId, Permission.READER, Message.OP.NOTE_REVISION_FOR_COMPARE, context,
+ callback)) {
+ return;
+ }
+
notebook.processNote(noteId ,
note -> {
if (note == null) {
@@ -1157,6 +1229,10 @@ public List completion(
ServiceContext context,
ServiceCallback