[Backport release-1.19] fix: accept absolute unix socket paths in GrpcEndpoint

dapr/conf/helpers.py

@@ -55,6 +55,11 @@ def scheme(self) -> str:
         return self._scheme
 
     def _set_hostname(self):
+        if self._parsed_url.scheme == 'unix':
+            # netloc + path keeps the filesystem path's case; parsed_url.hostname lowercases it.
+            self._hostname = f'{self._parsed_url.netloc}{self._parsed_url.path}'
+            return
+
         if self._parsed_url.hostname is None:
             self._hostname = URIParseConfig.DEFAULT_HOSTNAME
             return
@@ -171,8 +176,17 @@ def _set_tls(self):
     def tls(self) -> bool:
         return self._tls
 
+    def _is_unix_socket_path(self) -> bool:
+        """Whether this is a unix socket whose file path urlparse put in ``path``."""
+        return self._parsed_url.scheme == 'unix' and bool(self._parsed_url.path)
+
     def _validate_path_and_query(self) -> None:
-        if self._parsed_url.path:
+        if self._url.startswith('unix://') and self._parsed_url.netloc:
+            raise ValueError(
+                f"unix:// endpoints require an absolute path: '{self._url}'; "
+                'use unix:relative or unix:///absolute'
+            )
+        if self._parsed_url.path and not self._is_unix_socket_path():
             raise ValueError(
                 f"paths are not supported for gRPC endpoints: '{self._parsed_url.path}'"
             )

tests/conf/helpers_test.py

@@ -217,25 +217,67 @@ def test_parse_grpc_endpoint(self):
                 'port': '',
                 'endpoint': 'unix:my.sock',
             },
-            # Unix sockets with absolute path
+            # gRPC needs an absolute path after unix://, so an authority can never connect
+            {'url': 'unix://my.sock', 'error': True},
+            {'url': 'unix://my.sock?tls=true', 'error': True},
+            {'url': 'unix://relative/dir/x.sock', 'error': True},
+            # Relative socket names keep their case
             {
-                'url': 'unix://my.sock',
+                'url': 'unix:My.sock',
                 'error': False,
                 'secure': False,
                 'scheme': 'unix',
-                'host': 'my.sock',
+                'host': 'My.sock',
+                'port': '',
+                'endpoint': 'unix:My.sock',
+            },
+            {
+                'url': 'unix:run/My.sock',
+                'error': False,
+                'secure': False,
+                'scheme': 'unix',
+                'host': 'run/My.sock',
+                'port': '',
+                'endpoint': 'unix:run/My.sock',
+            },
+            # Unix sockets with a filesystem path, which urlparse puts in the path component
+            {
+                'url': 'unix:///tmp/dapr.sock',
+                'error': False,
+                'secure': False,
+                'scheme': 'unix',
+                'host': '/tmp/dapr.sock',
                 'port': '',
-                'endpoint': 'unix://my.sock',
+                'endpoint': 'unix:///tmp/dapr.sock',
             },
             {
-                'url': 'unix://my.sock?tls=true',
+                'url': 'unix:///tmp/dapr.sock?tls=true',
                 'error': False,
                 'secure': True,
                 'scheme': 'unix',
-                'host': 'my.sock',
+                'host': '/tmp/dapr.sock',
+                'port': '',
+                'endpoint': 'unix:///tmp/dapr.sock',
+            },
+            {
+                'url': 'unix:/tmp/dapr.sock',
+                'error': False,
+                'secure': False,
+                'scheme': 'unix',
+                'host': '/tmp/dapr.sock',
+                'port': '',
+                'endpoint': 'unix:/tmp/dapr.sock',
+            },
+            {
+                'url': 'unix:///private/tmp/dapr-sockets/Dapr-MyApp-grpc.socket',
+                'error': False,
+                'secure': False,
+                'scheme': 'unix',
+                'host': '/private/tmp/dapr-sockets/Dapr-MyApp-grpc.socket',
                 'port': '',
-                'endpoint': 'unix://my.sock',
+                'endpoint': 'unix:///private/tmp/dapr-sockets/Dapr-MyApp-grpc.socket',
             },
+            {'url': 'unix:///tmp/dapr.sock?abc=1', 'error': True},
             # Unix abstract sockets
             {
                 'url': 'unix-abstract:my.sock',

tests/ext/workflow/test_workflow_runtime.py

@@ -106,6 +106,34 @@ def test_timeout_interceptor_preserves_custom_interceptor_order(self):
             self.assertIsInstance(interceptors[2], DaprClientTimeoutInterceptor)
 
 
+class WorkflowRuntimeUnixSocketTest(unittest.TestCase):
+    """An absolute unix socket in DAPR_GRPC_ENDPOINT must reach the worker as a gRPC target."""
+
+    def setUp(self):
+        self._registry_patch = mock.patch(
+            'dapr.ext.workflow._durabletask.worker._Registry',
+            return_value=FakeTaskHubGrpcWorker(),
+        )
+        self._registry_patch.start()
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def test_absolute_unix_socket_endpoint_is_forwarded(self):
+        socket_endpoint = 'unix:///private/tmp/dapr-sockets/dapr-myapp-grpc.socket'
+        with (
+            mock.patch.object(settings, 'DAPR_GRPC_ENDPOINT', socket_endpoint),
+            mock.patch(
+                'dapr.ext.workflow._durabletask.worker.TaskHubGrpcWorker'
+            ) as mock_worker_cls,
+        ):
+            WorkflowRuntime()
+            call_kwargs = mock_worker_cls.call_args[1]
+
+            self.assertEqual(call_kwargs['host_address'], socket_endpoint)
+            self.assertFalse(call_kwargs['secure_channel'])
+
+
 class WorkflowRuntimeStatefulHistoryTest(unittest.TestCase):
     """The stateful-history options must reach the worker, not stop at the public API."""