Fix SSHOperator templated remote host handling

providers/ssh/src/airflow/providers/ssh/operators/ssh.py

@@ -100,8 +100,6 @@ def __init__(
         super().__init__(**kwargs)
         if ssh_hook and isinstance(ssh_hook, SSHHook):
             self.ssh_hook = ssh_hook
-            if remote_host is not None:
-                self.ssh_hook.remote_host = remote_host
         self.ssh_conn_id = ssh_conn_id
         self.remote_host = remote_host
         self.command = command
@@ -168,6 +166,9 @@ def execute(self, context=None) -> bytes | str:
         if self.command is None:
             raise AirflowException("SSH operator error: SSH command not specified. Aborting.")
 
+        if self.remote_host is not None:
+            self.hook.remote_host = self.remote_host
+
         # Forcing get_pty to True if the command begins with "sudo".
         self.get_pty = self.command.startswith("sudo") or self.get_pty
 

providers/ssh/tests/unit/ssh/operators/test_ssh.py

@@ -186,6 +186,22 @@ def test_arg_checking(self, get_conn, run_ssh_client_command):
                 command=None,
             ).execute(None)
 
+    def test_templated_remote_host_updates_hook_after_rendering(self):
+        self.hook.remote_host = "connection_host"
+        task = SSHOperator(
+            task_id="test",
+            ssh_hook=self.hook,
+            command=COMMAND,
+            remote_host="{{ params.remote_host }}",
+        )
+
+        assert self.hook.remote_host == "connection_host"
+
+        task.render_template_fields(context={"params": {"remote_host": "operator_remote_host"}})
+        task.execute(None)
+
+        assert self.hook.remote_host == "operator_remote_host"
+
     @pytest.mark.parametrize(
         ("command", "get_pty_in", "get_pty_out"),
         [

scripts/ci/prek/validate_operators_init_exemptions.txt

@@ -65,7 +65,6 @@ providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/oracle
 providers/microsoft/psrp/src/airflow/providers/microsoft/psrp/operators/psrp.py::PsrpOperator
 providers/oracle/src/airflow/providers/oracle/transfers/oracle_to_oracle.py::OracleToOracleOperator
 providers/papermill/src/airflow/providers/papermill/operators/papermill.py::PapermillOperator
-providers/ssh/src/airflow/providers/ssh/operators/ssh.py::SSHOperator
 providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py::SSHRemoteJobOperator
 providers/standard/src/airflow/providers/standard/operators/bash.py::BashOperator
 providers/standard/src/airflow/providers/standard/operators/hitl.py::HITLOperator