diff --git a/dapr/conf/helpers.py b/dapr/conf/helpers.py index d2d187628..a743b8dbd 100644 --- a/dapr/conf/helpers.py +++ b/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}'" ) diff --git a/tests/conf/helpers_test.py b/tests/conf/helpers_test.py index d09eb3d9d..a80d76360 100644 --- a/tests/conf/helpers_test.py +++ b/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', diff --git a/tests/ext/workflow/test_workflow_runtime.py b/tests/ext/workflow/test_workflow_runtime.py index 35bb9e872..e14a3b1f2 100644 --- a/tests/ext/workflow/test_workflow_runtime.py +++ b/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."""