Allow dictionaries in security= keywords (#3874)

Previously Dask Server and Client objects had to be instantiated with a
Security object for security.  Now we also allow for dictionary inputs
that will be splatted out to the Security object.  This helps when using
systems like dask-spec with Security.

distributed/cli/dask_scheduler.py

@@ -11,7 +11,7 @@
 
 from tornado.ioloop import IOLoop
 
-from distributed import Scheduler, Security
+from distributed import Scheduler
 from distributed.preloading import validate_preload_argv
 from distributed.cli.utils import check_python_3, install_signal_handlers
 from distributed.utils import deserialize_for_cli
@@ -157,17 +157,15 @@ def main(
     if port is None and (not host or not re.search(r":\d", host)):
         port = 8786
 
-    sec = Security(
-        **{
-            k: v
-            for k, v in [
-                ("tls_ca_file", tls_ca_file),
-                ("tls_scheduler_cert", tls_cert),
-                ("tls_scheduler_key", tls_key),
-            ]
-            if v is not None
-        }
-    )
+    sec = {
+        k: v
+        for k, v in [
+            ("tls_ca_file", tls_ca_file),
+            ("tls_scheduler_cert", tls_cert),
+            ("tls_scheduler_key", tls_key),
+        ]
+        if v is not None
+    }
 
     if "DASK_INTERNAL_INHERIT_CONFIG" in os.environ:
         config = deserialize_for_cli(os.environ["DASK_INTERNAL_INHERIT_CONFIG"])

distributed/cli/dask_worker.py

@@ -11,7 +11,7 @@
 import click
 import dask
 from dask.system import CPU_COUNT
-from distributed import Nanny, Security
+from distributed import Nanny
 from distributed.cli.utils import check_python_3, install_signal_handlers
 from distributed.comm import get_address_host_port
 from distributed.preloading import validate_preload_argv
@@ -278,17 +278,15 @@ def main(
         )
         dashboard = bokeh
 
-    sec = Security(
-        **{
-            k: v
-            for k, v in [
-                ("tls_ca_file", tls_ca_file),
-                ("tls_worker_cert", tls_cert),
-                ("tls_worker_key", tls_key),
-            ]
-            if v is not None
-        }
-    )
+    sec = {
+        k: v
+        for k, v in [
+            ("tls_ca_file", tls_ca_file),
+            ("tls_worker_cert", tls_cert),
+            ("tls_worker_key", tls_key),
+        ]
+        if v is not None
+    }
 
     if nprocs > 1 and not nanny:
         logger.error(

distributed/client.py

@@ -666,6 +666,8 @@ def __init__(
 
         if security is None:
             security = Security()
+        elif isinstance(security, dict):
+            security = Security(**security)
         elif security is True:
             security = Security.temporary()
             self._startup_kwargs["security"] = security

distributed/nanny.py

@@ -97,6 +97,9 @@ def __init__(
     ):
         self._setup_logging(logger)
         self.loop = loop or IOLoop.current()
+
+        if isinstance(security, dict):
+            security = Security(**security)
         self.security = security or Security()
         assert isinstance(self.security, Security)
         self.connection_args = self.security.get_connection_args("worker")

distributed/scheduler.py

@@ -1128,6 +1128,8 @@ def __init__(
             preload_argv = dask.config.get("distributed.scheduler.preload-argv")
         self.preloads = preloading.process_preloads(self, preload, preload_argv)
 
+        if isinstance(security, dict):
+            security = Security(**security)
         self.security = security or Security()
         assert isinstance(self.security, Security)
         self.connection_args = self.security.get_connection_args("scheduler")

distributed/tests/test_tls_functional.py

@@ -3,12 +3,21 @@
 Most are taken from other test files and adapted.
 """
 import asyncio
+import pytest
 
-from distributed import Nanny, worker_client, Queue
+from distributed import Scheduler, Worker, Client, Nanny, worker_client, Queue
 from distributed.client import wait
 from distributed.metrics import time
 from distributed.nanny import Nanny
-from distributed.utils_test import gen_tls_cluster, inc, double, slowinc, slowadd
+from distributed.utils_test import (  # noqa: F401
+    gen_tls_cluster,
+    inc,
+    double,
+    slowinc,
+    slowadd,
+    tls_config,
+    cleanup,
+)
 
 
 @gen_tls_cluster(client=True)
@@ -172,3 +181,35 @@ async def test_retire_workers(c, s, a, b):
     while a.status != "closed":
         await asyncio.sleep(0.01)
         assert time() < start + 5
+
+
+@pytest.mark.asyncio
+async def test_security_dict_input_no_security(cleanup):
+    async with Scheduler(security={}) as s:
+        async with Worker(s.address, security={}) as w:
+            async with Client(s.address, security={}, asynchronous=True) as c:
+                result = await c.submit(inc, 1)
+                assert result == 2
+
+
+@pytest.mark.asyncio
+async def test_security_dict_input(cleanup):
+    conf = tls_config()
+    ca_file = conf["distributed"]["comm"]["tls"]["ca-file"]
+    client = conf["distributed"]["comm"]["tls"]["client"]["cert"]
+    worker = conf["distributed"]["comm"]["tls"]["worker"]["cert"]
+    scheduler = conf["distributed"]["comm"]["tls"]["scheduler"]["cert"]
+
+    async with Scheduler(
+        security={"tls_ca_file": ca_file, "tls_scheduler_cert": scheduler}
+    ) as s:
+        async with Worker(
+            s.address, security={"tls_ca_file": ca_file, "tls_worker_cert": worker}
+        ) as w:
+            async with Client(
+                s.address,
+                security={"tls_ca_file": ca_file, "tls_client_cert": client},
+                asynchronous=True,
+            ) as c:
+                result = await c.submit(inc, 1)
+                assert result == 2

distributed/worker.py

@@ -507,6 +507,8 @@ def __init__(
             self, preload, preload_argv, file_dir=self.local_directory
         )
 
+        if isinstance(security, dict):
+            security = Security(**security)
         self.security = security or Security()
         assert isinstance(self.security, Security)
         self.connection_args = self.security.get_connection_args("worker")