Make encryption default if Security is given arguments

distributed/distributed.yaml

@@ -130,7 +130,7 @@ distributed:
       connect: 10s          # time before connecting fails
       tcp: 30s              # time before calling an unresponsive connection dead
 
-    require-encryption: False   # Whether to require encryption on non-local comms
+    require-encryption: None  # Whether to require encryption on non-local comms
 
     tls:
       ciphers: null   # Allowed ciphers, specified as an OpenSSL cipher string.

distributed/security.py

@@ -60,13 +60,15 @@ class Security:
         "tls_worker_cert",
     )
 
-    def __init__(self, **kwargs):
+    def __init__(self, require_encryption=None, **kwargs):
         extra = set(kwargs).difference(self.__slots__)
         if extra:
             raise TypeError("Unknown parameters: %r" % sorted(extra))
-        self._set_field(
-            kwargs, "require_encryption", "distributed.comm.require-encryption"
-        )
+        if require_encryption is None:
+            require_encryption = dask.config.get("distributed.comm.require-encryption")
+        if require_encryption is None:
+            require_encryption = not not kwargs
+        self.require_encryption = require_encryption
         self._set_field(kwargs, "tls_ciphers", "distributed.comm.tls.ciphers")
         self._set_field(kwargs, "tls_ca_file", "distributed.comm.tls.ca-file")
         self._set_field(kwargs, "tls_client_key", "distributed.comm.tls.client.key")

distributed/tests/test_tls_functional.py

@@ -201,11 +201,14 @@ async def test_security_dict_input(cleanup):
     scheduler = conf["distributed"]["comm"]["tls"]["scheduler"]["cert"]
 
     async with Scheduler(
-        security={"tls_ca_file": ca_file, "tls_scheduler_cert": scheduler}
+        host="localhost",
+        security={"tls_ca_file": ca_file, "tls_scheduler_cert": scheduler},
     ) as s:
+        assert s.address.startswith("tls://")
         async with Worker(
             s.address, security={"tls_ca_file": ca_file, "tls_worker_cert": worker}
         ) as w:
+            assert w.address.startswith("tls://")
             async with Client(
                 s.address,
                 security={"tls_ca_file": ca_file, "tls_client_cert": client},