[ZEPPELIN-5934] Check folder permissions before rename, trash and remove

zeppelin-server/src/main/java/org/apache/zeppelin/notebook/Notebook.java

@@ -540,6 +540,10 @@ public boolean containsFolder(String folderPath) {
     return noteManager.containsFolder(folderPath);
   }
 
+  public List<NoteInfo> getNoteInfoRecursively(String folderPath) throws IOException {
+    return noteManager.getNoteInfoRecursively(folderPath);
+  }
+
   public void moveNote(String noteId, String newNotePath, AuthenticationInfo subject) throws IOException {
     LOGGER.info("Move note {} to {}", noteId, newNotePath);
     noteManager.moveNote(noteId, newNotePath, subject);

zeppelin-server/src/main/java/org/apache/zeppelin/service/NotebookService.java

@@ -715,6 +715,11 @@ public void restoreFolder(String folderPath,
       return;
     }
     try {
+      // RESTORE_NOTE only asks for WRITER, so restoring a whole folder stays on the same level.
+      if (!checkFolderPermission(folderPath, Permission.WRITER, Message.OP.RESTORE_FOLDER,
+          context, callback)) {
+        return;
+      }
       String destFolderPath = folderPath.replace("/" + NoteManager.TRASH_FOLDER, "");
       notebook.moveFolder(folderPath, destFolderPath, context.getAutheInfo());
       callback.onSuccess(null, context);
@@ -1261,17 +1266,22 @@ public void moveFolderToTrash(String folderPath,
                               ServiceContext context,
                               ServiceCallback<Void> callback) throws IOException {
 
-    //TODO(zjffdu) folder permission check
     //TODO(zjffdu) folderPath is relative path, need to fix it in frontend
     LOGGER.info("Move folder {} to trash", folderPath);
 
+    String srcFolderPath = "/" + folderPath;
+    if (!checkFolderPermission(srcFolderPath, Permission.OWNER,
+        Message.OP.MOVE_FOLDER_TO_TRASH, context, callback)) {
+      return;
+    }
+
     String destFolderPath = "/" + NoteManager.TRASH_FOLDER + "/" + folderPath;
     if (notebook.containsNote(destFolderPath)) {
       destFolderPath = destFolderPath + " " +
           TRASH_CONFLICT_TIMESTAMP_FORMATTER.format(Instant.now());
     }
 
-    notebook.moveFolder("/" + folderPath, destFolderPath, context.getAutheInfo());
+    notebook.moveFolder(srcFolderPath, destFolderPath, context.getAutheInfo());
     callback.onSuccess(null, context);
   }
 
@@ -1291,6 +1301,10 @@ public List<NoteInfo> removeFolder(String folderPath,
                            ServiceContext context,
                            ServiceCallback<List<NoteInfo>> callback) throws IOException {
     try {
+      if (!checkFolderPermission(folderPath, Permission.OWNER, Message.OP.REMOVE_FOLDER,
+          context, callback)) {
+        return null;
+      }
       notebook.removeFolder(folderPath, context.getAutheInfo());
       List<NoteInfo> notesInfo = notebook.getNotesInfo(
               noteId -> authorizationService.isReader(noteId, context.getUserAndRoles()));
@@ -1306,10 +1320,13 @@ public List<NoteInfo> renameFolder(String folderPath,
                            String newFolderPath,
                            ServiceContext context,
                            ServiceCallback<List<NoteInfo>> callback) throws IOException {
-    //TODO(zjffdu) folder permission check
-
     try {
-      notebook.moveFolder(normalizeNotePath(folderPath),
+      String normalizedFolderPath = normalizeNotePath(folderPath);
+      if (!checkFolderPermission(normalizedFolderPath, Permission.OWNER,
+          Message.OP.FOLDER_RENAME, context, callback)) {
+        return null;
+      }
+      notebook.moveFolder(normalizedFolderPath,
               normalizeNotePath(newFolderPath), context.getAutheInfo());
       List<NoteInfo> notesInfo = notebook.getNotesInfo(
               noteId -> authorizationService.isReader(noteId, context.getUserAndRoles()));
@@ -1536,34 +1553,78 @@ private <T> boolean checkPermission(String noteId,
                                       Message.OP op,
                                       ServiceContext context,
                                       ServiceCallback<T> callback) throws IOException {
-    boolean isAllowed = false;
-    Set<String> allowed = null;
+    if (hasPermission(noteId, permission, context.getUserAndRoles())) {
+      return true;
+    } else {
+      String errorMsg = "Insufficient privileges to " + permission + " note.\n" +
+          "Allowed users or roles: " + getAllowedEntities(noteId, permission) + "\n" +
+          "But the user " + context.getAutheInfo().getUser() +
+          " belongs to: " + context.getUserAndRoles();
+      callback.onFailure(new ForbiddenException(errorMsg), context);
+      return false;
+    }
+  }
+
+  /**
+   * Zeppelin has no folder level ACL, so the permission of a folder is derived from the notes
+   * under it: the operation is allowed only when the caller holds the required permission on
+   * every one of them, and a folder holding no note is allowed.
+   *
+   * @return true when the operation may proceed, false after the callback has been failed
+   */
+  private <T> boolean checkFolderPermission(String folderPath,
+                                            Permission permission,
+                                            Message.OP op,
+                                            ServiceContext context,
+                                            ServiceCallback<T> callback) throws IOException {
+    List<String> deniedNotePaths = new ArrayList<>();
+    for (NoteInfo noteInfo : notebook.getNoteInfoRecursively(folderPath)) {
+      if (!hasPermission(noteInfo.getId(), permission, context.getUserAndRoles())) {
+        deniedNotePaths.add(noteInfo.getPath());
+      }
+    }
+    if (deniedNotePaths.isEmpty()) {
+      return true;
+    } else {
+      // Denied paths go to the log only: the caller may not be allowed to read those notes.
+      LOGGER.info("Permission check failed for {} on folder {}, user {} lacks {} on {}",
+          op, folderPath, context.getAutheInfo().getUser(), permission, deniedNotePaths);
+      String errorMsg = "Insufficient privileges to " + permission + " folder " + folderPath +
+          ".\n" + deniedNotePaths.size() + " of the notes it holds require " + permission +
+          " privileges.\n" + "But the user " + context.getAutheInfo().getUser() +
+          " belongs to: " + context.getUserAndRoles();
+      callback.onFailure(new ForbiddenException(errorMsg), context);
+      return false;
+    }
+  }
+
+  private boolean hasPermission(String noteId, Permission permission, Set<String> userAndRoles) {
     switch (permission) {
       case READER:
-        isAllowed = authorizationService.isReader(noteId, context.getUserAndRoles());
-        allowed = authorizationService.getReaders(noteId);
-        break;
+        return authorizationService.isReader(noteId, userAndRoles);
       case WRITER:
-        isAllowed = authorizationService.isWriter(noteId, context.getUserAndRoles());
-        allowed = authorizationService.getWriters(noteId);
-        break;
+        return authorizationService.isWriter(noteId, userAndRoles);
       case RUNNER:
-        isAllowed = authorizationService.isRunner(noteId, context.getUserAndRoles());
-        allowed = authorizationService.getRunners(noteId);
-        break;
+        return authorizationService.isRunner(noteId, userAndRoles);
       case OWNER:
-        isAllowed = authorizationService.isOwner(noteId, context.getUserAndRoles());
-        allowed = authorizationService.getOwners(noteId);
-        break;
+        return authorizationService.isOwner(noteId, userAndRoles);
+      default:
+        return false;
     }
-    if (isAllowed) {
-      return true;
-    } else {
-      String errorMsg = "Insufficient privileges to " + permission + " note.\n" +
-          "Allowed users or roles: " + allowed + "\n" + "But the user " +
-          context.getAutheInfo().getUser() + " belongs to: " + context.getUserAndRoles();
-      callback.onFailure(new ForbiddenException(errorMsg), context);
-      return false;
+  }
+
+  private Set<String> getAllowedEntities(String noteId, Permission permission) {
+    switch (permission) {
+      case READER:
+        return authorizationService.getReaders(noteId);
+      case WRITER:
+        return authorizationService.getWriters(noteId);
+      case RUNNER:
+        return authorizationService.getRunners(noteId);
+      case OWNER:
+        return authorizationService.getOwners(noteId);
+      default:
+        return Collections.emptySet();
     }
   }
 

zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java

@@ -19,6 +19,7 @@
 package org.apache.zeppelin.service;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -36,6 +37,7 @@
 import java.io.FileWriter;
 import java.io.IOException;
 import java.nio.file.Files;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -66,6 +68,7 @@
 import org.apache.zeppelin.notebook.repo.NotebookRepo;
 import org.apache.zeppelin.notebook.repo.VFSNotebookRepo;
 import org.apache.zeppelin.notebook.scheduler.QuartzSchedulerService;
+import org.apache.zeppelin.rest.exception.ForbiddenException;
 import org.apache.zeppelin.search.LuceneSearch;
 import org.apache.zeppelin.search.SearchService;
 import org.apache.zeppelin.storage.ConfigStorage;
@@ -88,6 +91,7 @@ class NotebookServiceTest {
   private File confDir;
   private SearchService searchService;
   private Notebook notebook;
+  private AuthorizationService authorizationService;
   private ServiceContext context =
       new ServiceContext(AuthenticationInfo.ANONYMOUS, new HashSet<>());
 
@@ -136,8 +140,7 @@ void setUp(TestInfo testInfo) throws Exception {
     when(mockInterpreterSetting.getStatus()).thenReturn(InterpreterSetting.Status.READY);
     Credentials credentials = new Credentials();
     NoteManager noteManager = new NoteManager(notebookRepo, zConf);
-    AuthorizationService authorizationService =
-        new AuthorizationService(noteManager, zConf, storage);
+    authorizationService = new AuthorizationService(noteManager, zConf, storage);
     notebook =
         new Notebook(
             zConf,
@@ -410,6 +413,115 @@ void testNoteOperations() throws IOException {
     assertEquals(0, notesInfo.size());
   }
 
+  @Test
+  void testFolderOperationsRequirePermissionOnEveryNote() throws IOException {
+    ServiceContext user1 = userContext("user1");
+    ServiceContext user2 = userContext("user2");
+
+    // note2 sits in a sub folder, so it is only reached by walking the folder recursively
+    String note1Id = notebookService.createNote("/folder_1/note1", "test", true, user1, callback);
+    String note2Id =
+        notebookService.createNote("/folder_1/nested/note2", "test", true, user1, callback);
+    // user2 owns one of the two notes
+    authorizationService.setOwners(note1Id, new HashSet<>(Arrays.asList("user1", "user2")));
+    assertTrue(authorizationService.isOwner(note1Id, user2.getUserAndRoles()));
+    assertFalse(authorizationService.isOwner(note2Id, user2.getUserAndRoles()));
+
+    reset(callback);
+    assertNull(notebookService.renameFolder("/folder_1", "/folder_2", user2, callback));
+    assertForbidden();
+
+    reset(callback);
+    notebookService.moveFolderToTrash("folder_1", user2, callback);
+    assertForbidden();
+
+    reset(callback);
+    assertNull(notebookService.removeFolder("/folder_1", user2, callback));
+    String errorMsg = assertForbidden();
+    // the message names the folder but not the note that blocked the call
+    assertTrue(errorMsg.contains("/folder_1"), errorMsg);
+    assertFalse(errorMsg.contains("note2"), errorMsg);
+
+    // nothing was applied
+    reset(callback);
+    List<NoteInfo> notesInfo = notebookService.listNotesInfo(false, user1, callback);
+    assertEquals(new HashSet<>(Arrays.asList("/folder_1/note1", "/folder_1/nested/note2")),
+        notesInfo.stream().map(NoteInfo::getPath).collect(Collectors.toSet()));
+
+    // the owner of every note succeeds
+    reset(callback);
+    notesInfo = notebookService.renameFolder("/folder_1", "/folder_2", user1, callback);
+    verify(callback).onSuccess(notesInfo, user1);
+    assertEquals(new HashSet<>(Arrays.asList("/folder_2/note1", "/folder_2/nested/note2")),
+        notesInfo.stream().map(NoteInfo::getPath).collect(Collectors.toSet()));
+
+    reset(callback);
+    notesInfo = notebookService.removeFolder("/folder_2", user1, callback);
+    verify(callback).onSuccess(notesInfo, user1);
+    assertEquals(0, notesInfo.size());
+  }
+
+  @Test
+  void testRestoreFolderRequiresWriterPermission() throws IOException {
+    ServiceContext user1 = userContext("user1");
+    ServiceContext user2 = userContext("user2");
+    ServiceContext user3 = userContext("user3");
+
+    String noteId = notebookService.createNote("/Backup/note1", "test", true, user1, callback);
+    // user2 may write the note, user3 may only read it
+    authorizationService.setWriters(noteId, new HashSet<>(Arrays.asList("user1", "user2")));
+    authorizationService.setReaders(noteId,
+        new HashSet<>(Arrays.asList("user1", "user2", "user3")));
+    assertTrue(authorizationService.isReader(noteId, user3.getUserAndRoles()));
+    assertFalse(authorizationService.isWriter(noteId, user3.getUserAndRoles()));
+    assertTrue(authorizationService.isWriter(noteId, user2.getUserAndRoles()));
+    notebookService.moveFolderToTrash("Backup", user1, callback);
+
+    // a reader is not enough
+    reset(callback);
+    notebookService.restoreFolder("/~Trash/Backup", user3, callback);
+    assertForbidden();
+
+    // a writer is, like RESTORE_NOTE
+    reset(callback);
+    notebookService.restoreFolder("/~Trash/Backup", user2, callback);
+    verify(callback).onSuccess(null, user2);
+    notebookService.getNote(noteId, user2, callback,
+        restoredNote -> {
+          assertEquals("/Backup/note1", restoredNote.getPath());
+          return null;
+        });
+  }
+
+  @Test
+  void testRemoveFolderHoldingNoNoteIsAllowed() throws IOException {
+    ServiceContext user1 = userContext("user1");
+    ServiceContext user2 = userContext("user2");
+
+    String noteId = notebookService.createNote("/folder_1/note1", "test", true, user1, callback);
+    // removing the last note leaves the folder itself in the tree
+    notebookService.removeNote(noteId, user1, callback);
+
+    reset(callback);
+    List<NoteInfo> notesInfo = notebookService.removeFolder("/folder_1", user2, callback);
+    verify(callback).onSuccess(notesInfo, user2);
+    assertEquals(0, notesInfo.size());
+  }
+
+  private ServiceContext userContext(String user) {
+    return new ServiceContext(new AuthenticationInfo(user),
+        new HashSet<>(Arrays.asList(user)));
+  }
+
+  // returns the message the caller would receive, to check what it does not reveal
+  private String assertForbidden() throws IOException {
+    ArgumentCaptor<Exception> exception = ArgumentCaptor.forClass(Exception.class);
+    verify(callback).onFailure(exception.capture(), any(ServiceContext.class));
+    assertTrue(exception.getValue() instanceof ForbiddenException,
+        "Expected a ForbiddenException but got: " + exception.getValue());
+    return ((ForbiddenException) exception.getValue()).getResponse().getEntity().toString();
+  }
+
   @Test
   void testNoteUpdate() throws IOException {
     // create note