Repository: iterative/dvc

Issue:
Allow passing fsspec options to WebHDFS to enable Kerberos and https
After #6662 it is no longer possible to use custom hdfscli clients with WebHDFS, as a result we are no longer able to connect using Kerberos or HTTPS.

fsspec allows passing in options to enable kerberos and https, we simply need to pass them through and expose them in the DVC config file:
https://filesystem-spec.readthedocs.io/en/latest/_modules/fsspec/implementations/webhdfs.html

This would also be an opportunity to remove the no longer used hdfscli options.

Discussion:
gudmundur-heimisson: I'm not sure if I linked the issue correctly but it is #6935 
efiop: Oh, looks like we also didn't update the docs before 🙁 We'll need to https://dvc.org/doc/command-reference/remote/modify accordingly.
gudmundur-heimisson: I will create a pull request in the dvc.org repo and link here once it's up for the documentation.

Patch:
diff --git a/dvc/config_schema.py b/dvc/config_schema.py
index 42c5ee37f..7427825ca 100644
--- a/dvc/config_schema.py
+++ b/dvc/config_schema.py
@@ -179,10 +179,12 @@ SCHEMA = {
                 },
                 "hdfs": {"user": str, "kerb_ticket": str, **REMOTE_COMMON},
                 "webhdfs": {
-                    "hdfscli_config": str,
-                    "webhdfs_token": str,
-                    "user": str,
-                    "webhdfs_alias": str,
+                    "kerberos": Bool,
+                    "kerberos_principal": str,
+                    "proxy_to": str,
+                    "ssl_verify": Any(Bool, str),
+                    "token": str,
+                    "use_https": Bool,
                     **REMOTE_COMMON,
                 },
                 "azure": {
diff --git a/dvc/fs/webhdfs.py b/dvc/fs/webhdfs.py
index 964b5b3fb..9a51c9cbd 100644
--- a/dvc/fs/webhdfs.py
+++ b/dvc/fs/webhdfs.py
@@ -31,9 +31,10 @@ class WebHDFSFileSystem(CallbackMixin, FSSpecWrapper):
         )
 
     def _prepare_credentials(self, **config):
-        if "webhdfs_token" in config:
-            config["token"] = config.pop("webhdfs_token")
-
+        self._ssl_verify = config.pop("ssl_verify", True)
+        principal = config.pop("kerberos_principal", None)
+        if principal:
+            config["kerb_kwargs"] = {"principal": principal}
         return config
 
     @wrap_prop(threading.Lock())
@@ -41,7 +42,9 @@ class WebHDFSFileSystem(CallbackMixin, FSSpecWrapper):
     def fs(self):
         from fsspec.implementations.webhdfs import WebHDFS
 
-        return WebHDFS(**self.fs_args)
+        fs = WebHDFS(**self.fs_args)
+        fs.session.verify = self._ssl_verify
+        return fs
 
     def checksum(self, path_info):
         path = self._with_bucket(path_info)
diff --git a/dvc/ui/__init__.py b/dvc/ui/__init__.py
index 5ba189f7a..bed4664b4 100644
--- a/dvc/ui/__init__.py
+++ b/dvc/ui/__init__.py
@@ -84,21 +84,35 @@ class Console:
         )
 
     def write_json(
-        self, data: Any, indent: int = 2, default: Callable = None
+        self,
+        data: Any,
+        indent: int = None,
+        highlight: bool = None,
+        stderr: bool = False,
+        skip_keys: bool = False,
+        ensure_ascii: bool = True,
+        check_circular: bool = True,
+        allow_nan: bool = True,
+        default: Optional[Callable[[Any], Any]] = None,
+        sort_keys: bool = False,
     ) -> None:
-        import json
-
-        if self.isatty():
-            from rich.highlighter import JSONHighlighter
-
-            j = json.dumps(data, indent=indent, default=default)
-            highlighter = JSONHighlighter()
-            text = highlighter(j)
-            text.no_wrap = True
-            text.overflow = None
-            return self.write(text, styled=True)
-
-        return self.write(json.dumps(data, default=default))
+        if highlight is None:
+            highlight = self.isatty()
+        if indent is None and self.isatty():
+            indent = 2
+
+        console = self.error_console if stderr else self.rich_console
+        return console.print_json(
+            data=data,
+            indent=indent,
+            highlight=bool(highlight),
+            skip_keys=skip_keys,
+            ensure_ascii=ensure_ascii,
+            check_circular=check_circular,
+            allow_nan=allow_nan,
+            default=default,
+            sort_keys=sort_keys,
+        )
 
     def write(
         self,
diff --git a/setup.cfg b/setup.cfg
index c731000f4..f492b259b 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -67,7 +67,7 @@ install_requires =
     pygtrie>=2.3.2
     dpath>=2.0.2,<3
     shtab>=1.3.4,<2
-    rich>=10.9.0
+    rich>=10.13.0
     dictdiffer>=0.8.1
     python-benedict>=0.24.2
     pyparsing>=2.4.7
@@ -112,6 +112,7 @@ ssh_gssapi = sshfs[gssapi]>=2021.8.1
 webdav = webdav4>=0.9.3
 # not to break `dvc[webhdfs]`
 webhdfs =
+    requests-kerberos==0.13.0
 terraform = tpi[ssh]>=2.1.0
 tests =
     %(terraform)s


Test patch:
diff --git a/tests/unit/remote/test_webhdfs.py b/tests/unit/remote/test_webhdfs.py
index b267c32fa..341aa4928 100644
--- a/tests/unit/remote/test_webhdfs.py
+++ b/tests/unit/remote/test_webhdfs.py
@@ -1,21 +1,54 @@
+from unittest.mock import Mock, create_autospec
+
+import pytest
+import requests
+
 from dvc.fs.webhdfs import WebHDFSFileSystem
 
+host = "host"
+kerberos = False
+kerberos_principal = "principal"
+port = 12345
+proxy_to = "proxy"
+ssl_verify = False
+token = "token"
+use_https = True
 user = "test"
-webhdfs_token = "token"
-webhdfs_alias = "alias-name"
-hdfscli_config = "path/to/cli/config"
-
-
-def test_init(dvc):
-    url = "webhdfs://test@127.0.0.1:50070"
-    config = {
-        "host": url,
-        "webhdfs_token": webhdfs_token,
-        "webhdfs_alias": webhdfs_alias,
-        "hdfscli_config": hdfscli_config,
-        "user": user,
+
+
+@pytest.fixture()
+def webhdfs_config():
+    url = f"webhdfs://{user}@{host}:{port}"
+    url_config = WebHDFSFileSystem._get_kwargs_from_urls(url)
+    return {
+        "kerberos": kerberos,
+        "kerberos_principal": kerberos_principal,
+        "proxy_to": proxy_to,
+        "ssl_verify": ssl_verify,
+        "token": token,
+        "use_https": use_https,
+        **url_config,
     }
 
-    fs = WebHDFSFileSystem(**config)
-    assert fs.fs_args["token"] == webhdfs_token
+
+def test_init(dvc, webhdfs_config):
+    fs = WebHDFSFileSystem(**webhdfs_config)
+    assert fs.fs_args["host"] == host
+    assert fs.fs_args["token"] == token
     assert fs.fs_args["user"] == user
+    assert fs.fs_args["port"] == port
+    assert fs.fs_args["kerberos"] == kerberos
+    assert fs.fs_args["kerb_kwargs"] == {"principal": kerberos_principal}
+    assert fs.fs_args["proxy_to"] == proxy_to
+    assert fs.fs_args["use_https"] == use_https
+
+
+def test_verify_ssl(dvc, webhdfs_config, monkeypatch):
+    mock_session = create_autospec(requests.Session)
+    monkeypatch.setattr(requests, "Session", Mock(return_value=mock_session))
+    # can't have token at the same time as user or proxy_to
+    del webhdfs_config["token"]
+    fs = WebHDFSFileSystem(**webhdfs_config)
+    # ssl verify can't be set until after the file system is instantiated
+    fs.fs  # pylint: disable=pointless-statement
+    assert mock_session.verify == ssl_verify