This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 0378a727e5f Fix template-field validation timing in ssh provider
operators (#70315)
0378a727e5f is described below
commit 0378a727e5fcbf63968331944ef7130308039686
Author: Bramhanand Lingala <[email protected]>
AuthorDate: Sat Jul 25 02:20:45 2026 -0400
Fix template-field validation timing in ssh provider operators (#70315)
---
.../ssh/src/airflow/providers/ssh/operators/ssh.py | 7 +--
.../providers/ssh/operators/ssh_remote_job.py | 6 +--
providers/ssh/tests/unit/ssh/operators/test_ssh.py | 32 ++++++++++++
.../unit/ssh/operators/test_ssh_remote_job.py | 60 ++++++++++++++++++++++
.../ci/prek/validate_operators_init_exemptions.txt | 2 -
5 files changed, 99 insertions(+), 8 deletions(-)
diff --git a/providers/ssh/src/airflow/providers/ssh/operators/ssh.py
b/providers/ssh/src/airflow/providers/ssh/operators/ssh.py
index 9ae09d6cbae..1a19591ff9c 100644
--- a/providers/ssh/src/airflow/providers/ssh/operators/ssh.py
+++ b/providers/ssh/src/airflow/providers/ssh/operators/ssh.py
@@ -100,8 +100,6 @@ class SSHOperator(BaseOperator):
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
@@ -123,7 +121,7 @@ class SSHOperator(BaseOperator):
"""Create SSHHook to run commands on remote host."""
if self.ssh_conn_id:
self.log.info("ssh_hook is not provided or invalid. Trying
ssh_conn_id to create SSHHook.")
- if self.remote_host is not None:
+ if self.remote_host:
self.log.info(
"remote_host is provided explicitly. "
"It will replace the remote_host which was defined "
@@ -168,6 +166,9 @@ class SSHOperator(BaseOperator):
if self.command is None:
raise AirflowException("SSH operator error: SSH command not
specified. Aborting.")
+ if self.remote_host:
+ self.ssh_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
diff --git
a/providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py
b/providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py
index dc3414eca68..7c126c7c6e7 100644
--- a/providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py
+++ b/providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py
@@ -135,9 +135,6 @@ class SSHRemoteJobOperator(BaseOperator):
self.command = command
self.remote_host = remote_host
self.environment = environment
-
- if remote_base_dir is not None:
- self._validate_base_dir(remote_base_dir)
self.remote_base_dir = remote_base_dir
self.poll_interval = poll_interval
@@ -271,6 +268,9 @@ class SSHRemoteJobOperator(BaseOperator):
if not self.command:
raise AirflowException("SSH operator error: command not
specified.")
+ if self.remote_base_dir is not None:
+ self._validate_base_dir(self.remote_base_dir)
+
ti = context["ti"]
self._job_id = generate_job_id(
dag_id=ti.dag_id,
diff --git a/providers/ssh/tests/unit/ssh/operators/test_ssh.py
b/providers/ssh/tests/unit/ssh/operators/test_ssh.py
index 8ea788166f8..ff956df1864 100644
--- a/providers/ssh/tests/unit/ssh/operators/test_ssh.py
+++ b/providers/ssh/tests/unit/ssh/operators/test_ssh.py
@@ -186,6 +186,38 @@ class TestSSHOperator:
command=None,
).execute(None)
+ def test_init_does_not_mutate_injected_ssh_hook_remote_host(self):
+ """
+ __init__ must not push remote_host onto an injected ssh_hook: at
construction time
+ remote_host is the raw, un-rendered Jinja expression (e.g. "{{
dag_run.conf['host'] }}"),
+ not the real value, so mutating the hook here would poison it with the
wrong host.
+ """
+ original_remote_host = self.hook.remote_host
+ SSHOperator(
+ task_id="test_init_no_mutate",
+ ssh_hook=self.hook,
+ command=COMMAND,
+ remote_host="{{ dag_run.conf['host'] }}",
+ )
+ assert self.hook.remote_host == original_remote_host
+
+ def test_execute_applies_remote_host_to_injected_ssh_hook(self):
+ """
+ execute() must push remote_host onto the injected ssh_hook using its
final
+ (post-templating) value, before the hook is used to connect.
+ """
+ task = SSHOperator(
+ task_id="test_execute_applies_remote_host",
+ ssh_hook=self.hook,
+ command=COMMAND,
+ remote_host="{{ dag_run.conf['host'] }}",
+ )
+ task.remote_host = "rendered.host.internal"
+
+ task.execute(None)
+
+ assert task.ssh_hook.remote_host == "rendered.host.internal"
+
@pytest.mark.parametrize(
("command", "get_pty_in", "get_pty_out"),
[
diff --git a/providers/ssh/tests/unit/ssh/operators/test_ssh_remote_job.py
b/providers/ssh/tests/unit/ssh/operators/test_ssh_remote_job.py
index 6ecf80baef2..55e144bab27 100644
--- a/providers/ssh/tests/unit/ssh/operators/test_ssh_remote_job.py
+++ b/providers/ssh/tests/unit/ssh/operators/test_ssh_remote_job.py
@@ -202,6 +202,66 @@ class TestSSHRemoteJobOperator:
with pytest.raises(AirflowException, match="command not specified"):
op.execute({})
+ def test_init_does_not_validate_remote_base_dir(self):
+ """
+ __init__ must not validate remote_base_dir: at construction time this
is the raw,
+ un-rendered Jinja expression (e.g. "{{ dag_run.conf['base_dir'] }}"),
not the real path,
+ so any validation performed here operates on the wrong value.
+ """
+
+ op = SSHRemoteJobOperator(
+ task_id="test_task",
+ ssh_conn_id="test_conn",
+ command="/path/to/script.sh",
+ remote_base_dir="../etc",
+ )
+ assert op.remote_base_dir == "../etc"
+
+ def test_execute_validates_remote_base_dir(self):
+ """
+ execute() must validate remote_base_dir using its final
(post-templating) value, and
+ must do so before any SSH connection is opened.
+ """
+ op = SSHRemoteJobOperator(
+ task_id="test_task",
+ ssh_conn_id="test_conn",
+ command="/path/to/script.sh",
+ remote_base_dir="{{ dag_run.conf['base_dir'] }}",
+ )
+ # Simulate what Airflow's templating engine does between __init__ and
execute():
+ # the template field is overwritten in place with its rendered value.
+ op.remote_base_dir = "/tmp/../etc"
+
+ with pytest.raises(ValueError, match="path traversal not allowed"):
+ op.execute({})
+
+ self.mock_hook.get_conn.assert_not_called()
+
+ def test_execute_accepts_valid_custom_remote_base_dir(self):
+ """A valid, non-default remote_base_dir must pass validation and
proceed normally."""
+ self.mock_hook.exec_ssh_client_command.return_value = (
+ 0,
+ b"af_test_dag_test_task_run1_try1_abc123",
+ b"",
+ )
+
+ op = SSHRemoteJobOperator(
+ task_id="test_task",
+ ssh_conn_id="test_conn",
+ command="/path/to/script.sh",
+ remote_os="posix",
+ remote_base_dir="/opt/custom-airflow-jobs",
+ )
+ op.remote_base_dir = "/opt/custom-airflow-jobs" # post-templating
value
+
+ mock_ti = mock.MagicMock()
+ mock_ti.dag_id, mock_ti.task_id, mock_ti.run_id, mock_ti.try_number =
"d", "t", "r", 1
+
+ with pytest.raises(TaskDeferred):
+ op.execute({"ti": mock_ti})
+
+ assert op._paths.job_dir.startswith("/opt/custom-airflow-jobs")
+
@mock.patch.object(SSHRemoteJobOperator, "defer")
def test_execute_complete_re_defers_if_not_done(self, mock_defer):
"""Test that execute_complete re-defers if job is not done."""
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt
b/scripts/ci/prek/validate_operators_init_exemptions.txt
index 6adf724c9c9..29ee3dd2639 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -56,8 +56,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/trigger_dagrun.py::TriggerDagRunOperator
providers/standard/src/airflow/providers/standard/sensors/date_time.py::DateTimeSensor
