Optimize hidden checks (#1226)

Co-authored-by: Steven Silvester <steven.silvester@ieee.org>

Author: vidartf

jupyter_server/base/handlers.py

@@ -893,7 +893,7 @@ def validate_absolute_path(self, root, absolute_path):
         abs_path = super().validate_absolute_path(root, absolute_path)
         abs_root = os.path.abspath(root)
         assert abs_path is not None
-        if is_hidden(abs_path, abs_root) and not self.contents_manager.allow_hidden:
+        if not self.contents_manager.allow_hidden and is_hidden(abs_path, abs_root):
             self.log.info(
                 "Refusing to serve hidden file, via 404 Error, use flag 'ContentsManager.allow_hidden' to enable"
             )

jupyter_server/files/handlers.py

@@ -48,7 +48,7 @@ async def get(self, path, include_body=True):
         self.check_xsrf_cookie()
         cm = self.contents_manager
 
-        if await ensure_async(cm.is_hidden(path)) and not cm.allow_hidden:
+        if not cm.allow_hidden and await ensure_async(cm.is_hidden(path)):
             self.log.info("Refusing to serve hidden file, via 404 Error")
             raise web.HTTPError(404)
 

jupyter_server/services/contents/filemanager.py

@@ -226,7 +226,7 @@ def _base_model(self, path):
 
         four_o_four = "file or directory does not exist: %r" % path
 
-        if is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        if not self.allow_hidden and is_hidden(os_path, self.root_dir):
             self.log.info("Refusing to serve hidden file or directory %r, via 404 Error", os_path)
             raise web.HTTPError(404, four_o_four)
 
@@ -278,7 +278,7 @@ def _dir_model(self, path, content=True):
 
         if not os.path.isdir(os_path):
             raise web.HTTPError(404, four_o_four)
-        elif is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        elif not self.allow_hidden and is_hidden(os_path, self.root_dir):
             self.log.info("Refusing to serve hidden directory %r, via 404 Error", os_path)
             raise web.HTTPError(404, four_o_four)
 
@@ -414,7 +414,7 @@ def get(self, path, content=True, type=None, format=None):
         if not self.exists(path):
             raise web.HTTPError(404, four_o_four)
 
-        if is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        if not self.allow_hidden and is_hidden(os_path, self.root_dir):
             self.log.info("Refusing to serve hidden file or directory %r, via 404 Error", os_path)
             raise web.HTTPError(404, four_o_four)
 
@@ -437,7 +437,7 @@ def get(self, path, content=True, type=None, format=None):
 
     def _save_directory(self, os_path, model, path=""):
         """create a directory"""
-        if is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        if not self.allow_hidden and is_hidden(os_path, self.root_dir):
             raise web.HTTPError(400, "Cannot create directory %r" % os_path)
         if not os.path.exists(os_path):
             with self.perm_to_403():
@@ -460,7 +460,7 @@ def save(self, model, path=""):
 
         os_path = self._get_os_path(path)
 
-        if is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        if not self.allow_hidden and is_hidden(os_path, self.root_dir):
             raise web.HTTPError(400, f"Cannot create file or directory {os_path!r}")
 
         self.log.debug("Saving %s", os_path)
@@ -506,7 +506,7 @@ def delete_file(self, path):
         os_path = self._get_os_path(path)
         rm = os.unlink
 
-        if is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        if not self.allow_hidden and is_hidden(os_path, self.root_dir):
             raise web.HTTPError(400, f"Cannot delete file or directory {os_path!r}")
 
         four_o_four = "file or directory does not exist: %r" % path
@@ -576,9 +576,9 @@ def rename_file(self, old_path, new_path):
         new_os_path = self._get_os_path(new_path)
         old_os_path = self._get_os_path(old_path)
 
-        if (
+        if not self.allow_hidden and (
             is_hidden(old_os_path, self.root_dir) or is_hidden(new_os_path, self.root_dir)
-        ) and not self.allow_hidden:
+        ):
             raise web.HTTPError(400, f"Cannot rename file or directory {old_os_path!r}")
 
         # Should we proceed with the move?
@@ -741,7 +741,7 @@ async def _dir_model(self, path, content=True):
 
         if not os.path.isdir(os_path):
             raise web.HTTPError(404, four_o_four)
-        elif is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        elif not self.allow_hidden and is_hidden(os_path, self.root_dir):
             self.log.info("Refusing to serve hidden directory %r, via 404 Error", os_path)
             raise web.HTTPError(404, four_o_four)
 
@@ -896,7 +896,7 @@ async def get(self, path, content=True, type=None, format=None):
 
     async def _save_directory(self, os_path, model, path=""):
         """create a directory"""
-        if is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        if not self.allow_hidden and is_hidden(os_path, self.root_dir):
             raise web.HTTPError(400, "Cannot create hidden directory %r" % os_path)
         if not os.path.exists(os_path):
             with self.perm_to_403():
@@ -961,7 +961,7 @@ async def delete_file(self, path):
         os_path = self._get_os_path(path)
         rm = os.unlink
 
-        if is_hidden(os_path, self.root_dir) and not self.allow_hidden:
+        if not self.allow_hidden and is_hidden(os_path, self.root_dir):
             raise web.HTTPError(400, f"Cannot delete file or directory {os_path!r}")
 
         if not os.path.exists(os_path):
@@ -1035,9 +1035,9 @@ async def rename_file(self, old_path, new_path):
         new_os_path = self._get_os_path(new_path)
         old_os_path = self._get_os_path(old_path)
 
-        if (
+        if not self.allow_hidden and (
             is_hidden(old_os_path, self.root_dir) or is_hidden(new_os_path, self.root_dir)
-        ) and not self.allow_hidden:
+        ):
             raise web.HTTPError(400, f"Cannot rename file or directory {old_os_path!r}")
 
         # Should we proceed with the move?

jupyter_server/services/contents/handlers.py

@@ -115,7 +115,7 @@ async def get(self, path=""):
             raise web.HTTPError(400, "Content %r is invalid" % content_str)
         content = int(content_str or "")
 
-        if await ensure_async(cm.is_hidden(path)) and not cm.allow_hidden:
+        if not cm.allow_hidden and await ensure_async(cm.is_hidden(path)):
             raise web.HTTPError(404, f"file or directory {path!r} does not exist")
 
         model = await ensure_async(
@@ -141,10 +141,10 @@ async def patch(self, path=""):
         old_path = model.get("path")
         if (
             old_path
+            and not cm.allow_hidden
             and (
                 await ensure_async(cm.is_hidden(path)) or await ensure_async(cm.is_hidden(old_path))
             )
-            and not cm.allow_hidden
         ):
             raise web.HTTPError(400, f"Cannot rename file or directory {path!r}")
 
@@ -252,10 +252,10 @@ async def put(self, path=""):
         if model:
             if model.get("copy_from"):
                 raise web.HTTPError(400, "Cannot copy with PUT, only POST")
-            if (
+            if not cm.allow_hidden and (
                 (model.get("path") and await ensure_async(cm.is_hidden(model.get("path"))))
                 or await ensure_async(cm.is_hidden(path))
-            ) and not cm.allow_hidden:
+            ):
                 raise web.HTTPError(400, f"Cannot create file or directory {path!r}")
 
             exists = await ensure_async(self.contents_manager.file_exists(path))
@@ -275,7 +275,7 @@ async def delete(self, path=""):
         """delete a file in the given path"""
         cm = self.contents_manager
 
-        if await ensure_async(cm.is_hidden(path)) and not cm.allow_hidden:
+        if not cm.allow_hidden and await ensure_async(cm.is_hidden(path)):
             raise web.HTTPError(400, f"Cannot delete file or directory {path!r}")
 
         self.log.warning("delete %s", path)

tests/base/test_handlers.py

@@ -82,7 +82,8 @@ async def test_authenticated_file_handler(jp_serverapp, tmpdir):
 
     handler = AuthenticatedFileHandler(app.web_app, request, path=str(tmpdir))
     for key in list(handler.settings):
-        del handler.settings[key]
+        if key != "contents_manager":
+            del handler.settings[key]
     handler.check_xsrf_cookie = MagicMock()  # type:ignore
     handler._jupyter_current_user = "foo"  # type:ignore
     with warnings.catch_warnings():

tests/services/contents/test_api.py

@@ -4,6 +4,7 @@
 import warnings
 from base64 import decodebytes, encodebytes
 from unicodedata import normalize
+from unittest.mock import patch
 
 import pytest
 import tornado
@@ -185,7 +186,6 @@ async def test_get_text_file_contents(jp_fetch, contents, path, name):
     assert expected_http_error(e, 400)
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Disabled retrieving hidden files on Windows")
 async def test_get_404_hidden(jp_fetch, contents, contents_dir):
     # Create text files
     hidden_dir = contents_dir / ".hidden"
@@ -398,7 +398,6 @@ async def test_upload_txt(jp_fetch, contents, contents_dir, _check_created):
     assert model["content"] == body
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Disabled uploading hidden files on Windows")
 async def test_upload_txt_hidden(jp_fetch, contents, contents_dir):
     with pytest.raises(tornado.httpclient.HTTPClientError) as e:
         body = "ünicode téxt"
@@ -551,7 +550,6 @@ async def test_copy_put_400(jp_fetch, contents, contents_dir, _check_created):
     assert expected_http_error(e, 400)
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Disabled copying hidden files on Windows")
 async def test_copy_put_400_hidden(
     jp_fetch,
     contents,
@@ -598,7 +596,6 @@ async def test_copy_put_400_hidden(
     assert expected_http_error(e, 400)
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Disabled copying hidden files on Windows")
 async def test_copy_400_hidden(
     jp_fetch,
     contents,
@@ -686,7 +683,7 @@ async def test_delete_dirs(jp_fetch, contents, folders):
     assert model["content"] == []
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Disabled deleting non-empty dirs on Windows")
+@pytest.mark.xfail(sys.platform == "win32", reason="Deleting non-empty dirs on Windows")
 async def test_delete_non_empty_dir(jp_fetch, contents):
     # Delete a folder
     await jp_fetch("api", "contents", "å b", method="DELETE")
@@ -696,14 +693,12 @@ async def test_delete_non_empty_dir(jp_fetch, contents):
     assert expected_http_error(e, 404)
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Disabled deleting hidden dirs on Windows")
 async def test_delete_hidden_dir(jp_fetch, contents):
     with pytest.raises(tornado.httpclient.HTTPClientError) as e:
         await jp_fetch("api", "contents", ".hidden", method="DELETE")
     assert expected_http_error(e, 400)
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Disabled deleting hidden dirs on Windows")
 async def test_delete_hidden_file(jp_fetch, contents):
     # Test deleting file in a hidden directory
     with pytest.raises(tornado.httpclient.HTTPClientError) as e:
@@ -747,7 +742,6 @@ async def test_rename(jp_fetch, jp_base_url, contents, contents_dir):
     assert "a.ipynb" not in nbnames
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Disabled copying hidden files on Windows")
 async def test_rename_400_hidden(jp_fetch, jp_base_url, contents, contents_dir):
     with pytest.raises(tornado.httpclient.HTTPClientError) as e:
         old_path = ".hidden/old.txt"
@@ -1018,3 +1012,73 @@ async def test_trust(jp_fetch, contents):
             allow_nonstandard_methods=True,
         )
         assert r.code == 201
+
+
+@patch(
+    "jupyter_core.paths.is_hidden",
+    side_effect=AssertionError("Should not call is_hidden if not important"),
+)
+@patch(
+    "jupyter_server.services.contents.filemanager.is_hidden",
+    side_effect=AssertionError("Should not call is_hidden if not important"),
+)
+async def test_regression_is_hidden(m1, m2, jp_fetch, jp_serverapp, contents, _check_created):
+    # check that no is_hidden check runs if configured to allow hidden files
+    contents_dir = contents["contents_dir"]
+
+    hidden_dir = contents_dir / ".hidden"
+    hidden_dir.mkdir(parents=True, exist_ok=True)
+    txt = "visible text file in hidden dir"
+    txtname = hidden_dir.joinpath("visible.txt")
+    txtname.write_text(txt, encoding="utf-8")
+
+    # Our role here is to check that the side-effect never triggers
+    jp_serverapp.contents_manager.allow_hidden = True
+    r = await jp_fetch(
+        "api",
+        "contents",
+        ".hidden",
+    )
+    assert r.code == 200
+
+    r = await jp_fetch(
+        "api",
+        "contents",
+        ".hidden",
+        method="POST",
+        body=json.dumps(
+            {
+                "copy_from": ".hidden/visible.txt",
+            }
+        ),
+    )
+    _check_created(r, str(contents_dir), ".hidden", "visible-Copy1.txt", type="file")
+
+    r = await jp_fetch(
+        "api",
+        "contents",
+        ".hidden",
+        "visible-Copy1.txt",
+        method="DELETE",
+    )
+    assert r.code == 204
+
+    model = {
+        "content": "foo",
+        "format": "text",
+        "type": "file",
+    }
+    r = await jp_fetch(
+        "api", "contents", ".hidden", "new.txt", method="PUT", body=json.dumps(model)
+    )
+    _check_created(r, str(contents_dir), ".hidden", "new.txt", type="file")
+
+    # sanity check that is actually triggers when flag set to false
+    jp_serverapp.contents_manager.allow_hidden = False
+    with pytest.raises(tornado.httpclient.HTTPClientError) as e:
+        await jp_fetch(
+            "api",
+            "contents",
+            ".hidden",
+        )
+    assert expected_http_error(e, 500)

tests/services/contents/test_manager.py

@@ -296,7 +296,6 @@ async def test_403(jp_file_contents_manager_class, tmp_path):
         assert e.status_code == 403
 
 
-@pytest.mark.skipif(sys.platform.startswith("win"), reason="Can't test hidden files on Windows")
 async def test_400(jp_file_contents_manager_class, tmp_path):  # noqa
     # Test Delete behavior
     # Test delete of file in hidden directory
@@ -406,7 +405,6 @@ async def test_400(jp_file_contents_manager_class, tmp_path):  # noqa
     assert excinfo.value.status_code == 400
 
 
-@pytest.mark.skipif(sys.platform.startswith("win"), reason="Can't test hidden files on Windows")
 async def test_404(jp_file_contents_manager_class, tmp_path):
     # Test visible file in hidden folder
     with pytest.raises(HTTPError) as excinfo:
@@ -1010,3 +1008,43 @@ async def test_validate_notebook_model(jp_contents_manager):
         cm.validate_notebook_model(model)
         assert mock_validate_nb.call_count == 1
         mock_validate_nb.reset_mock()
+
+
+@patch(
+    "jupyter_core.paths.is_hidden",
+    side_effect=AssertionError("Should not call is_hidden if not important"),
+)
+@patch(
+    "jupyter_server.services.contents.filemanager.is_hidden",
+    side_effect=AssertionError("Should not call is_hidden if not important"),
+)
+async def test_regression_is_hidden(m1, m2, jp_contents_manager):
+    cm = jp_contents_manager
+    cm.allow_hidden = True
+    # Our role here is to check that the side-effect never triggers
+    dirname = "foo/.hidden_dir"
+    await make_populated_dir(cm, dirname)
+    await ensure_async(cm.get(dirname))
+    await check_populated_dir_files(cm, dirname)
+    await ensure_async(cm.get(path="/".join([dirname, "nb.ipynb"])))
+    await ensure_async(cm.get(path="/".join([dirname, "file.txt"])))
+    await ensure_async(cm.new(path="/".join([dirname, "nb2.ipynb"])))
+    await ensure_async(cm.new(path="/".join([dirname, "file2.txt"])))
+    await ensure_async(cm.new(path="/".join([dirname, "subdir"]), model={"type": "directory"}))
+    await ensure_async(
+        cm.copy(
+            from_path="/".join([dirname, "file.txt"]), to_path="/".join([dirname, "file-copy.txt"])
+        )
+    )
+    await ensure_async(
+        cm.rename_file(
+            old_path="/".join([dirname, "file-copy.txt"]),
+            new_path="/".join([dirname, "file-renamed.txt"]),
+        )
+    )
+    await ensure_async(cm.delete_file(path="/".join([dirname, "file-renamed.txt"])))
+
+    # sanity check that is actually triggers when flag set to false
+    cm.allow_hidden = False
+    with pytest.raises(AssertionError):
+        await ensure_async(cm.get(dirname))