Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion dapr/conf/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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}'"
)
Expand Down
56 changes: 49 additions & 7 deletions tests/conf/helpers_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand Down
28 changes: 28 additions & 0 deletions tests/ext/workflow/test_workflow_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down
Loading