Check 'distributed.scheduler.pickle' in Scheduler.run_function (#4838)

distributed/scheduler.py

@@ -6278,6 +6278,13 @@ def run_function(self, stream, function, args=(), kwargs={}, wait=True):
         """
         from .worker import run
 
+        if not dask.config.get("distributed.scheduler.pickle"):
+            raise ValueError(
+                "Cannot run function as the scheduler has been explicitly disallowed from "
+                "deserializing arbitrary bytestrings using pickle via the "
+                "'distributed.scheduler.pickle' configuration setting."
+            )
+
         self.log_event("all", {"action": "run-function", "function": function})
         return run(self, stream, function=function, args=args, kwargs=kwargs, wait=wait)
 

distributed/tests/test_scheduler.py

@@ -1100,6 +1100,15 @@ def f(dask_scheduler=None):
     assert response == s.address
 
 
+@gen_cluster(client=True, config={"distributed.scheduler.pickle": False})
+async def test_run_on_scheduler_disabled(c, s, a, b):
+    def f(dask_scheduler=None):
+        return dask_scheduler.address
+
+    with pytest.raises(ValueError, match="disallowed from deserializing"):
+        await c._run_on_scheduler(f)
+
+
 @gen_cluster(client=True)
 async def test_close_worker(c, s, a, b):
     assert len(s.workers) == 2