Merge branch 'main' into nexus-metric-meter

Author: VegetarianOrc

temporalio/client.py

@@ -125,7 +125,7 @@ async def connect(
         default_workflow_query_reject_condition: Optional[
             temporalio.common.QueryRejectCondition
         ] = None,
-        tls: Union[bool, TLSConfig] = False,
+        tls: Union[bool, TLSConfig, None] = None,
         retry_config: Optional[RetryConfig] = None,
         keep_alive_config: Optional[KeepAliveConfig] = KeepAliveConfig.default,
         rpc_metadata: Mapping[str, Union[str, bytes]] = {},
@@ -166,9 +166,11 @@ async def connect(
                 condition for workflow queries if not set during query. See
                 :py:meth:`WorkflowHandle.query` for details on the rejection
                 condition.
-            tls: If false, the default, do not use TLS. If true, use system
-                default TLS configuration. If TLS configuration present, that
-                TLS configuration will be used.
+            tls: If ``None``, the default, TLS will be enabled automatically
+                when ``api_key`` is provided, otherwise TLS is disabled. If
+                ``False``, do not use TLS. If ``True``, use system default TLS
+                configuration. If TLS configuration present, that TLS
+                configuration will be used.
             retry_config: Retry configuration for direct service calls (when
                 opted in) or all high-level calls made by this client (which all
                 opt-in to retries by default). If unset, a default retry
@@ -247,6 +249,7 @@ def __init__(
             default_workflow_query_reject_condition=default_workflow_query_reject_condition,
             header_codec_behavior=header_codec_behavior,
         )
+        self._initial_config = config.copy()
 
         for plugin in plugins:
             config = plugin.configure_client(config)
@@ -261,12 +264,16 @@ def _init_from_config(self, config: ClientConfig):
         for interceptor in reversed(list(self._config["interceptors"])):
             self._impl = interceptor.intercept_client(self._impl)
 
-    def config(self) -> ClientConfig:
+    def config(self, *, active_config: bool = False) -> ClientConfig:
         """Config, as a dictionary, used to create this client.
 
+        Args:
+            active_config: If true, return the modified configuration in use rather than the initial one
+                provided to the client.
+
         This makes a shallow copy of the config each call.
         """
-        config = self._config.copy()
+        config = self._config.copy() if active_config else self._initial_config.copy()
         config["interceptors"] = list(config["interceptors"])
         return config
 
@@ -4290,7 +4297,8 @@ async def _to_proto(
             await _apply_headers(
                 self.headers,
                 action.start_workflow.header.fields,
-                client.config()["header_codec_behavior"] == HeaderCodecBehavior.CODEC
+                client.config(active_config=True)["header_codec_behavior"]
+                == HeaderCodecBehavior.CODEC
                 and not self._from_raw,
                 client.data_converter.payload_codec,
             )
@@ -6917,7 +6925,8 @@ async def _apply_headers(
         await _apply_headers(
             source,
             dest,
-            self._client.config()["header_codec_behavior"] == HeaderCodecBehavior.CODEC,
+            self._client.config(active_config=True)["header_codec_behavior"]
+            == HeaderCodecBehavior.CODEC,
             self._client.data_converter.payload_codec,
         )
 

temporalio/service.py

@@ -136,7 +136,7 @@ class ConnectConfig:
 
     target_host: str
     api_key: Optional[str] = None
-    tls: Union[bool, TLSConfig] = False
+    tls: Union[bool, TLSConfig, None] = None
     retry_config: Optional[RetryConfig] = None
     keep_alive_config: Optional[KeepAliveConfig] = KeepAliveConfig.default
     rpc_metadata: Mapping[str, Union[str, bytes]] = field(default_factory=dict)
@@ -172,6 +172,10 @@ def _to_bridge_config(self) -> temporalio.bridge.client.ClientConfig:
         elif self.tls:
             target_url = f"https://{self.target_host}"
             tls_config = TLSConfig()._to_bridge_config()
+        # Enable TLS by default when API key is provided and tls not explicitly set
+        elif self.tls is None and self.api_key is not None:
+            target_url = f"https://{self.target_host}"
+            tls_config = TLSConfig()._to_bridge_config()
         else:
             target_url = f"http://{self.target_host}"
             tls_config = None

temporalio/worker/_replayer.py

@@ -81,6 +81,7 @@ def __init__(
             disable_safe_workflow_eviction=disable_safe_workflow_eviction,
             header_codec_behavior=header_codec_behavior,
         )
+        self._initial_config = self._config.copy()
 
         # Apply plugin configuration
         self.plugins = plugins
@@ -91,13 +92,17 @@ def __init__(
         if not self._config["workflows"]:
             raise ValueError("At least one workflow must be specified")
 
-    def config(self) -> ReplayerConfig:
+    def config(self, *, active_config: bool = False) -> ReplayerConfig:
         """Config, as a dictionary, used to create this replayer.
 
+        Args:
+            active_config: If true, return the modified configuration in use rather than the initial one
+                provided to the client.
+
         Returns:
             Configuration, shallow-copied.
         """
-        config = self._config.copy()
+        config = self._config.copy() if active_config else self._initial_config.copy()
         config["workflows"] = list(config["workflows"])
         return config
 

temporalio/worker/_worker.py

@@ -375,8 +375,9 @@ def __init__(
                     f"The same plugin type {type(client_plugin)} is present from both client and worker. It may run twice and may not be the intended behavior."
                 )
         plugins = plugins_from_client + list(plugins)
+        self._initial_config = config.copy()
 
-        self.plugins = plugins
+        self._plugins = plugins
         for plugin in plugins:
             config = plugin.configure_worker(config)
 
@@ -387,7 +388,6 @@ def _init_from_config(self, client: temporalio.client.Client, config: WorkerConf
         Client is safe to take separately since it can't be modified by worker plugins.
         """
         self._config = config
-
         if not (
             config["activities"]
             or config["nexus_service_handlers"]
@@ -408,7 +408,7 @@ def _init_from_config(self, client: temporalio.client.Client, config: WorkerConf
             )
 
         # Prepend applicable client interceptors to the given ones
-        client_config = config["client"].config()
+        client_config = config["client"].config(active_config=True)
         interceptors_from_client = cast(
             List[Interceptor],
             [i for i in client_config["interceptors"] if isinstance(i, Interceptor)],
@@ -554,7 +554,7 @@ def check_activity(activity):
                 maximum=config["max_concurrent_activity_task_polls"]
             )
 
-        deduped_plugin_names = list(set([plugin.name() for plugin in self.plugins]))
+        deduped_plugin_names = list(set([plugin.name() for plugin in self._plugins]))
 
         # Create bridge worker last. We have empirically observed that if it is
         # created before an error is raised from the activity worker
@@ -622,13 +622,17 @@ def check_activity(activity):
             ),
         )
 
-    def config(self) -> WorkerConfig:
+    def config(self, *, active_config: bool = False) -> WorkerConfig:
         """Config, as a dictionary, used to create this worker.
 
+        Args:
+            active_config: If true, return the modified configuration in use rather than the initial one
+                provided to the worker.
+
         Returns:
             Configuration, shallow-copied.
         """
-        config = self._config.copy()
+        config = self._config.copy() if active_config else self._initial_config.copy()
         config["activities"] = list(config.get("activities", []))
         config["workflows"] = list(config.get("workflows", []))
         return config
@@ -702,7 +706,7 @@ def make_lambda(plugin, next):
             return lambda w: plugin.run_worker(w, next)
 
         next_function = lambda w: w._run()
-        for plugin in reversed(self.plugins):
+        for plugin in reversed(self._plugins):
             next_function = make_lambda(plugin, next_function)
 
         await next_function(self)

tests/api/test_grpc_stub.py

@@ -129,6 +129,7 @@ async def test_grpc_metadata():
         f"localhost:{port}",
         api_key="my-api-key",
         rpc_metadata={"my-meta-key": "my-meta-val"},
+        tls=False,
     )
     workflow_server.assert_last_metadata(
         {

tests/test_envconfig.py

@@ -1024,7 +1024,7 @@ async def test_e2e_multi_profile_different_client_connections(client: Client):
     assert dev_client.service_client.config.target_host == target_host
     assert dev_client.namespace == "dev"
     assert dev_client.service_client.config.api_key is None
-    assert dev_client.service_client.config.tls is False
+    assert dev_client.service_client.config.tls is None
 
     assert prod_client.service_client.config.target_host == target_host
     assert prod_client.namespace == "prod"

tests/test_plugins.py

@@ -56,6 +56,7 @@ async def connect_service_client(
         next: Callable[[ConnectConfig], Awaitable[ServiceClient]],
     ) -> ServiceClient:
         config.api_key = "replaced key"
+        config.tls = False
         return await next(config)
 
 
@@ -150,7 +151,7 @@ async def test_worker_plugin_basic_config(client: Client) -> None:
         activities=[never_run_activity],
         plugins=[MyWorkerPlugin()],
     )
-    task_queue = worker.config().get("task_queue")
+    task_queue = worker.config(active_config=True).get("task_queue")
     assert task_queue is not None and task_queue.startswith("replaced_queue")
 
     # Test client plugin propagation to worker plugins
@@ -160,7 +161,7 @@ async def test_worker_plugin_basic_config(client: Client) -> None:
     worker = Worker(
         client, task_queue="queue" + str(uuid.uuid4()), activities=[never_run_activity]
     )
-    task_queue = worker.config().get("task_queue")
+    task_queue = worker.config(active_config=True).get("task_queue")
     assert task_queue is not None and task_queue.startswith("combined")
 
     # Test both. Client propagated plugins are called first, so the worker plugin overrides in this case
@@ -170,7 +171,7 @@ async def test_worker_plugin_basic_config(client: Client) -> None:
         activities=[never_run_activity],
         plugins=[MyWorkerPlugin()],
     )
-    task_queue = worker.config().get("task_queue")
+    task_queue = worker.config(active_config=True).get("task_queue")
     assert task_queue is not None and task_queue.startswith("replaced_queue")
 
 
@@ -202,7 +203,8 @@ async def test_worker_sandbox_restrictions(client: Client) -> None:
     assert (
         "my_module"
         in cast(
-            SandboxedWorkflowRunner, worker.config().get("workflow_runner")
+            SandboxedWorkflowRunner,
+            worker.config(active_config=True).get("workflow_runner"),
         ).restrictions.passthrough_modules
     )
 
@@ -276,8 +278,11 @@ async def test_replay(client: Client) -> None:
         )
         await handle.result()
     replayer = Replayer(workflows=[], plugins=[plugin])
-    assert len(replayer.config().get("workflows") or []) == 1
-    assert replayer.config().get("data_converter") == pydantic_data_converter
+    assert len(replayer.config(active_config=True).get("workflows") or []) == 1
+    assert (
+        replayer.config(active_config=True).get("data_converter")
+        == pydantic_data_converter
+    )
 
     await replayer.replay_workflow(await handle.fetch_history())
 
@@ -303,19 +308,28 @@ async def test_simple_plugins(client: Client) -> None:
         plugins=[plugin],
     )
     # On a sequence, a value is appended
-    assert worker.config().get("workflows") == [HelloWorkflow, HelloWorkflow2]
+    assert worker.config(active_config=True).get("workflows") == [
+        HelloWorkflow,
+        HelloWorkflow2,
+    ]
 
     # Test with plugin registered in client
     worker = Worker(
         new_client,
         task_queue="queue" + str(uuid.uuid4()),
         activities=[never_run_activity],
     )
-    assert worker.config().get("workflows") == [HelloWorkflow2]
+    assert worker.config(active_config=True).get("workflows") == [HelloWorkflow2]
 
     replayer = Replayer(workflows=[HelloWorkflow], plugins=[plugin])
-    assert replayer.config().get("data_converter") == pydantic_data_converter
-    assert replayer.config().get("workflows") == [HelloWorkflow, HelloWorkflow2]
+    assert (
+        replayer.config(active_config=True).get("data_converter")
+        == pydantic_data_converter
+    )
+    assert replayer.config(active_config=True).get("workflows") == [
+        HelloWorkflow,
+        HelloWorkflow2,
+    ]
 
 
 async def test_simple_plugins_callables(client: Client) -> None:
@@ -350,7 +364,7 @@ def converter(old: Optional[DataConverter]):
         activities=[never_run_activity],
         plugins=[plugin],
     )
-    assert worker.config().get("workflows") == []
+    assert worker.config(active_config=True).get("workflows") == []
 
 
 class MediumPlugin(SimplePlugin):
@@ -371,5 +385,5 @@ async def test_medium_plugin(client: Client) -> None:
         plugins=[plugin],
         workflows=[HelloWorkflow],
     )
-    task_queue = worker.config().get("task_queue")
+    task_queue = worker.config(active_config=True).get("task_queue")
     assert task_queue is not None and task_queue.startswith("override")

tests/test_service.py

@@ -171,6 +171,55 @@ async def test_grpc_status(client: Client, env: WorkflowEnvironment):
     )
 
 
+def test_connect_config_tls_enabled_by_default_when_api_key_provided():
+    """Test that TLS is enabled by default when API key is provided and tls is not configured."""
+    config = temporalio.service.ConnectConfig(
+        target_host="localhost:7233",
+        api_key="test-api-key",
+    )
+    # TLS should be auto-enabled when api_key is provided and tls not explicitly set
+    bridge_config = config._to_bridge_config()
+    assert bridge_config.target_url == "https://localhost:7233"
+    assert bridge_config.tls_config is not None
+
+
+def test_connect_config_tls_can_be_explicitly_disabled_even_when_api_key_provided():
+    """Test that TLS can be explicitly disabled even when API key is provided."""
+    config = temporalio.service.ConnectConfig(
+        target_host="localhost:7233",
+        api_key="test-api-key",
+        tls=False,
+    )
+    # TLS should remain disabled when explicitly set to False
+    assert config.tls is False
+
+
+def test_connect_config_tls_disabled_by_default_when_no_api_key():
+    """Test that TLS is disabled by default when no API key is provided."""
+    config = temporalio.service.ConnectConfig(
+        target_host="localhost:7233",
+    )
+    # TLS should remain disabled when no api_key is provided
+    bridge_config = config._to_bridge_config()
+    assert bridge_config.target_url == "http://localhost:7233"
+    assert bridge_config.tls_config is None
+
+
+def test_connect_config_tls_explicit_config_preserved():
+    """Test that explicit TLS configuration is preserved regardless of API key."""
+    tls_config = temporalio.service.TLSConfig(
+        server_root_ca_cert=b"test-cert",
+        domain="test-domain",
+    )
+    config = temporalio.service.ConnectConfig(
+        target_host="localhost:7233",
+        api_key="test-api-key",
+        tls=tls_config,
+    )
+    # Explicit TLS config should be preserved
+    assert config.tls == tls_config
+
+
 async def test_rpc_execution_not_unknown(client: Client):
     """
     Execute each rpc method and expect a failure, but ensure the failure is not that the rpc method is unknown