This is an automated email from the ASF dual-hosted git repository.
dabla 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 a67d43f2baf Fix SSHHookAsync deadlock on the extra and compress like
SSHHook (#76528)
a67d43f2baf is described below
commit a67d43f2baff85888d7323ce5d90e618f63a3701
Author: David Blain <[email protected]>
AuthorDate: Thu Nov 12 19:40:16 2026 +0100
Fix SSHHookAsync deadlock on the extra and compress like SSHHook (#76528)
* Fix SSHHookAsync deadlock on the extra and compress like SSHHook
Parsing the connection extra through extra_dejson masks its secrets with
a synchronous supervisor call, which raises DeadlockImminentError when
the hook opens a connection in an async task that has other calls in
flight on the same event loop. The connection itself was already fetched
asynchronously; the extra now is too.
The async hook also left compression to asyncssh, which prefers none,
while SSHHook compresses unless the compress extra is false. Bulk
transfers over the async hook were therefore slower than over the sync
one for no reason.
* Tighten SSHHookAsync extra tests and document compress side effects
- The asynchronous-extra test builds the connection with an empty extra,
so it only passes when the extra read through common.compat reaches
_parse_extras; patches are decorators with autospec on the compat
helpers (asyncssh.connect stays an AsyncMock, as autospec on an
@async_context_manager function is not awaitable).
- The _parse_extras tests pass the extra through one helper on a real
Connection instead of bare MagicMocks; the helper is renamed
_parse_extra_dict so it no longer shares the method's name.
- The compress check is spelled as in SSHHook.
- The compress docs note that the async hook ignores Compression from
~/.ssh/config and that each compressed connection holds zlib state.
* Correct SSHHookAsync compression comment and compress docs
asyncssh offered zlib first by default up to 2.17 and moved "none"
first in 2.18, so the explicit list keeps the same order on every
supported version. Neither hook reads Compression from ~/.ssh/config,
and compress is a connection extra, so the docs point at the connection
SSHRemoteJobOperator uses rather than at the triggerer.
* Raise the asyncssh floor to 2.18 and move the triggerer note to the
operator docs
With asyncssh 2.18 as the floor, SSHHookAsync no longer has to explain what
older versions offered by default. The SSH connection docs keep only what holds
for the connection; the note about zlib state on a triggerer moves to the
SSHRemoteJobOperator page.
* Test that asyncssh offers no compression first by default
SSHHookAsync passes its own compression list because asyncssh prefers none
from 2.18, the floor. The test fails on an older asyncssh, where zlib came
first, so lowering the floor means handling both defaults in the hook.
---
providers/ssh/README.rst | 2 +-
providers/ssh/docs/connections/ssh.rst | 1 +
providers/ssh/docs/index.rst | 2 +-
providers/ssh/docs/operators/ssh_remote_job.rst | 5 +
providers/ssh/pyproject.toml | 4 +-
.../ssh/src/airflow/providers/ssh/hooks/ssh.py | 22 +++-
.../ssh/tests/unit/ssh/hooks/test_ssh_async.py | 117 +++++++++++++--------
uv.lock | 2 +-
8 files changed, 104 insertions(+), 51 deletions(-)
diff --git a/providers/ssh/README.rst b/providers/ssh/README.rst
index 40753ae63cb..bd337b3e7e8 100644
--- a/providers/ssh/README.rst
+++ b/providers/ssh/README.rst
@@ -55,7 +55,7 @@ PIP package Version required
========================================== ==================
``apache-airflow`` ``>=2.11.0``
``apache-airflow-providers-common-compat`` ``>=1.12.0``
-``asyncssh`` ``>=2.12.0``
+``asyncssh`` ``>=2.18.0``
``paramiko`` ``>=4.0.0``
========================================== ==================
diff --git a/providers/ssh/docs/connections/ssh.rst
b/providers/ssh/docs/connections/ssh.rst
index 9cfa3f88cdd..9477b9400da 100644
--- a/providers/ssh/docs/connections/ssh.rst
+++ b/providers/ssh/docs/connections/ssh.rst
@@ -51,6 +51,7 @@ Extra (optional)
* ``timeout`` - Deprecated - use conn_timeout instead.
* ``cmd_timeout`` - Timeout (in seconds) for executing the command. The
default is 10 seconds. `null` value means no timeout.
* ``compress`` - ``true`` to ask the remote client/server to compress
traffic; ``false`` to refuse compression. Default is ``true``.
+ Neither hook reads ``Compression`` from ``~/.ssh/config``. Each
compressed connection holds about 0.3 MB of zlib state while open.
* ``no_host_key_check`` - Set to ``true`` to connect to hosts that have no
entry in ``~/.ssh/known_hosts`` (Hosts file), accepting unknown host keys
without requiring a known-hosts entry. Default is ``false``, which restricts
connecting to hosts already present in the known hosts file. The default
provides maximum protection against trojan horse attacks, but can be
troublesome when the ``/etc/ssh/ssh_known_hosts`` file is poorly maintained or
connections to new hosts are frequently m [...]
* ``allow_host_key_change`` - Set to ``true`` if you want to allow
connecting to hosts that has host key changed or when you get 'REMOTE HOST
IDENTIFICATION HAS CHANGED' error. This won't protect against
Man-In-The-Middle attacks. Other possible solution is to remove the host entry
from ``~/.ssh/known_hosts`` file. Default is ``false``.
* ``look_for_keys`` - Set to ``false`` if you want to disable searching
for discoverable private key files in ``~/.ssh/``
diff --git a/providers/ssh/docs/index.rst b/providers/ssh/docs/index.rst
index 44894c6de33..6050e6b4659 100644
--- a/providers/ssh/docs/index.rst
+++ b/providers/ssh/docs/index.rst
@@ -94,7 +94,7 @@ PIP package Version required
========================================== ==================
``apache-airflow`` ``>=2.11.0``
``apache-airflow-providers-common-compat`` ``>=1.12.0``
-``asyncssh`` ``>=2.12.0``
+``asyncssh`` ``>=2.18.0``
``paramiko`` ``>=4.0.0``
========================================== ==================
diff --git a/providers/ssh/docs/operators/ssh_remote_job.rst
b/providers/ssh/docs/operators/ssh_remote_job.rst
index a21106fac2a..52c39dcfa6c 100644
--- a/providers/ssh/docs/operators/ssh_remote_job.rst
+++ b/providers/ssh/docs/operators/ssh_remote_job.rst
@@ -294,6 +294,11 @@ poll loop instead of reconnecting on every status check.
To push a high fan-out
releasing the entire fan-out simultaneously. See the "Placing Limits on
Mapped Tasks" section of
:doc:`apache-airflow:authoring-and-scheduling/dynamic-task-mapping` for the
available limits.
+The trigger's connection compresses traffic by default and holds about 0.3 MB
of zlib state for
+as long as the job runs, which adds up on a triggerer monitoring many jobs.
Setting
+``"compress": "false"`` in the extra of the SSH connection avoids it, and also
turns compression
+off for every other task using that connection.
+
Comparison with SSHOperator
----------------------------
diff --git a/providers/ssh/pyproject.toml b/providers/ssh/pyproject.toml
index 8cb9696ee74..867b9095c64 100644
--- a/providers/ssh/pyproject.toml
+++ b/providers/ssh/pyproject.toml
@@ -59,8 +59,8 @@ requires-python = ">=3.11"
# After you modify the dependencies, and rebuild your Breeze CI image with
``breeze ci-image build``
dependencies = [
"apache-airflow>=2.11.0",
- "apache-airflow-providers-common-compat>=1.12.0",
- "asyncssh>=2.12.0",
+ "apache-airflow-providers-common-compat>=1.12.0", # use next version
+ "asyncssh>=2.18.0",
"paramiko>=4.0.0",
]
diff --git a/providers/ssh/src/airflow/providers/ssh/hooks/ssh.py
b/providers/ssh/src/airflow/providers/ssh/hooks/ssh.py
index cbfad4a1ecc..9481c943a25 100644
--- a/providers/ssh/src/airflow/providers/ssh/hooks/ssh.py
+++ b/providers/ssh/src/airflow/providers/ssh/hooks/ssh.py
@@ -33,7 +33,7 @@ from paramiko.config import SSH_PORT
from tenacity import Retrying, stop_after_attempt, wait_fixed, wait_random
from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.common.compat.connection import get_async_connection,
get_async_extra_dejson
from airflow.providers.common.compat.sdk import AirflowException, BaseHook
from airflow.providers.ssh.tunnel import AsyncSSHTunnel, SSHTunnel
from airflow.utils.platform import getuser
@@ -642,10 +642,16 @@ class SSHHookAsync(BaseHook):
self.passphrase = passphrase
self.private_key = private_key
self.keepalive_interval = keepalive_interval
+ # Default value, overridable from Connection, as on SSHHook
+ self.compress = True
+
+ def _parse_extras(self, conn: Any, extra_options: dict[str, Any]) -> None:
+ """
+ Parse the connection's extra fields into instance fields.
- def _parse_extras(self, conn: Any) -> None:
- """Parse extra fields from the connection into instance fields."""
- extra_options = conn.extra_dejson
+ :param conn: the connection, for its host
+ :param extra_options: the connection's extra, already read without a
synchronous SDK call
+ """
if "key_file" in extra_options and self.key_file == "":
self.key_file = extra_options["key_file"]
if "known_hosts" in extra_options:
@@ -658,6 +664,8 @@ class SSHHookAsync(BaseHook):
)
if "private_key" in extra_options:
self.private_key = extra_options["private_key"]
+ if "compress" in extra_options and
str(extra_options["compress"]).lower() == "false":
+ self.compress = False
host_key = extra_options.get("host_key")
nhkc_raw = extra_options.get("no_host_key_check")
@@ -708,7 +716,9 @@ class SSHHookAsync(BaseHook):
conn = await get_async_connection(self.ssh_conn_id)
if conn.extra is not None:
- self._parse_extras(conn)
+ # extra_dejson masks the secrets with a synchronous SDK call,
which raises
+ # DeadlockImminentError in an async task; read the extra
asynchronously.
+ self._parse_extras(conn, await get_async_extra_dejson(conn))
def _get_value(self_val, conn_val, default=None):
if self_val is not None:
@@ -722,6 +732,8 @@ class SSHHookAsync(BaseHook):
"port": _get_value(self.port, conn.port, SSH_PORT),
"username": _get_value(self.username, conn.login),
"password": _get_value(self.password, conn.password),
+ # asyncssh offers "none" first by default; offer zlib first, as
paramiko does for SSHHook.
+ "compression_algs": ["[email protected]", "zlib", "none"] if
self.compress else ["none"],
}
if self.key_file:
conn_config["client_keys"] = self.key_file
diff --git a/providers/ssh/tests/unit/ssh/hooks/test_ssh_async.py
b/providers/ssh/tests/unit/ssh/hooks/test_ssh_async.py
index 5598f4ee288..df7f05e8977 100644
--- a/providers/ssh/tests/unit/ssh/hooks/test_ssh_async.py
+++ b/providers/ssh/tests/unit/ssh/hooks/test_ssh_async.py
@@ -21,7 +21,9 @@ import json
import warnings
from unittest import mock
+import asyncssh
import pytest
+from asyncssh.compression import get_default_compression_algs
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.models import Connection
@@ -32,6 +34,11 @@ def _connection(extra: dict, host: str = "test.host") ->
Connection:
return Connection(conn_id="test_conn", conn_type="ssh", host=host,
extra=json.dumps(extra))
+def _parse_extra_dict(hook: SSHHookAsync, extra: dict, host: str =
"test.host") -> None:
+ """Parse *extra* as ``_get_conn`` does: the connection plus its extra,
read beforehand."""
+ hook._parse_extras(_connection(extra, host=host), extra)
+
+
class TestSSHHookAsync:
def test_init_with_conn_id(self):
"""Test initialization with connection ID."""
@@ -62,28 +69,22 @@ class TestSSHHookAsync:
def test_parse_extras_key_file(self):
"""Test parsing key_file from connection extras."""
hook = SSHHookAsync(ssh_conn_id="test_conn")
- mock_conn = mock.MagicMock()
- mock_conn.extra_dejson = {"key_file": "/path/to/key"}
- mock_conn.host = "test.host"
- hook._parse_extras(mock_conn)
+ _parse_extra_dict(hook, {"key_file": "/path/to/key"})
assert hook.key_file == "/path/to/key"
def test_parse_extras_no_host_key_check(self):
"""Test parsing no_host_key_check from connection extras."""
hook = SSHHookAsync(ssh_conn_id="test_conn")
- mock_conn = mock.MagicMock()
- mock_conn.extra_dejson = {"no_host_key_check": "true"}
- mock_conn.host = "test.host"
- hook._parse_extras(mock_conn)
+ _parse_extra_dict(hook, {"no_host_key_check": "true"})
assert hook.known_hosts == "none"
def test_parse_extras_no_host_key_check_defaults_to_false(self):
"""A connection that omits ``no_host_key_check`` keeps host key
verification enabled."""
hook = SSHHookAsync(ssh_conn_id="test_conn")
- hook._parse_extras(_connection({}))
+ _parse_extra_dict(hook, {})
assert hook.known_hosts != "none"
def test_parse_extras_honours_deprecated_alias(self):
@@ -91,7 +92,7 @@ class TestSSHHookAsync:
hook = SSHHookAsync(ssh_conn_id="test_conn")
with pytest.warns(AirflowProviderDeprecationWarning,
match="ignore_hostkey_verification"):
- hook._parse_extras(_connection({"ignore_hostkey_verification":
True}))
+ _parse_extra_dict(hook, {"ignore_hostkey_verification": True})
assert hook.known_hosts == "none"
def test_parse_extras_canonical_key_wins_over_alias(self):
@@ -100,65 +101,50 @@ class TestSSHHookAsync:
with warnings.catch_warnings():
warnings.simplefilter("error", AirflowProviderDeprecationWarning)
- hook._parse_extras(_connection({"no_host_key_check": False,
"ignore_hostkey_verification": True}))
+ _parse_extra_dict(hook, {"no_host_key_check": False,
"ignore_hostkey_verification": True})
assert hook.known_hosts != "none"
def test_parse_extras_host_key_uses_host_override(self):
"""The known_hosts entry names the host actually connected to, not the
connection's host."""
hook = SSHHookAsync(ssh_conn_id="test_conn", host="override.example")
- hook._parse_extras(_connection({"host_key": "ssh-ed25519 AAAAC3..."},
host="connection.example"))
+ _parse_extra_dict(hook, {"host_key": "ssh-ed25519 AAAAC3..."},
host="connection.example")
assert hook.known_hosts == b"override.example ssh-ed25519 AAAAC3..."
def test_parse_extras_bare_host_key_is_rsa(self):
"""A bare base64 host key is RSA, as on the sync hook, and asyncssh
needs the type."""
hook = SSHHookAsync(ssh_conn_id="test_conn")
- hook._parse_extras(_connection({"host_key": "AAAAB3NzaC1yc2E"}))
+ _parse_extra_dict(hook, {"host_key": "AAAAB3NzaC1yc2E"})
assert hook.known_hosts == b"test.host ssh-rsa AAAAB3NzaC1yc2E"
def test_parse_extras_host_key(self):
"""Test parsing host_key from connection extras."""
hook = SSHHookAsync(ssh_conn_id="test_conn")
- mock_conn = mock.MagicMock()
- mock_conn.extra_dejson = {"host_key": " ssh-rsa AAAAB3... user@host ",
"no_host_key_check": "false"}
- mock_conn.host = "test.host"
- hook._parse_extras(mock_conn)
+ _parse_extra_dict(hook, {"host_key": " ssh-rsa AAAAB3... user@host ",
"no_host_key_check": "false"})
assert hook.known_hosts == b"test.host ssh-rsa AAAAB3..."
def test_parse_extras_dss_host_key_raises(self):
"""Test that ssh-dss host_key is rejected."""
hook = SSHHookAsync(ssh_conn_id="test_conn")
- mock_conn = mock.MagicMock()
- mock_conn.extra_dejson = {"host_key": "ssh-dss\tAAAAB3...",
"no_host_key_check": "false"}
- mock_conn.host = "test.host"
with pytest.raises(ValueError, match="DSA/DSS host keys"):
- hook._parse_extras(mock_conn)
+ _parse_extra_dict(hook, {"host_key": "ssh-dss\tAAAAB3...",
"no_host_key_check": "false"})
def test_parse_extras_host_key_with_no_check_raises(self):
"""Test that host_key with no_host_key_check raises error."""
hook = SSHHookAsync(ssh_conn_id="test_conn")
- mock_conn = mock.MagicMock()
- mock_conn.extra_dejson = {
- "host_key": "ssh-rsa AAAAB3...",
- "no_host_key_check": "true",
- }
- mock_conn.host = "test.host"
with pytest.raises(ValueError, match="Host key check was skipped"):
- hook._parse_extras(mock_conn)
+ _parse_extra_dict(hook, {"host_key": "ssh-rsa AAAAB3...",
"no_host_key_check": "true"})
def test_parse_extras_private_key(self):
"""Test parsing private_key from connection extras."""
hook = SSHHookAsync(ssh_conn_id="test_conn")
- mock_conn = mock.MagicMock()
test_key = "test-private-key-content"
- mock_conn.extra_dejson = {"private_key": test_key}
- mock_conn.host = "test.host"
- hook._parse_extras(mock_conn)
+ _parse_extra_dict(hook, {"private_key": test_key})
assert hook.private_key == test_key
@pytest.mark.asyncio
@@ -171,19 +157,15 @@ class TestSSHHookAsync:
username="testuser",
)
- mock_conn_obj = mock.MagicMock()
- mock_conn_obj.extra_dejson = {"no_host_key_check": "true"}
- mock_conn_obj.host = None
- mock_conn_obj.port = None
- mock_conn_obj.login = None
- mock_conn_obj.password = None
- mock_conn_obj.extra = "{}"
+ mock_conn_obj = Connection(
+ conn_id="test_conn", conn_type="ssh",
extra=json.dumps({"no_host_key_check": "true"})
+ )
- mock_ssh_client = mock.AsyncMock()
+ mock_ssh_client = mock.AsyncMock(spec=asyncssh.SSHClientConnection)
with mock.patch(
"airflow.providers.ssh.hooks.ssh.get_async_connection",
- new_callable=mock.AsyncMock,
+ autospec=True,
return_value=mock_conn_obj,
):
with mock.patch("asyncssh.connect", new_callable=mock.AsyncMock)
as mock_connect:
@@ -197,6 +179,59 @@ class TestSSHHookAsync:
assert call_kwargs["username"] == "testuser"
assert result == mock_ssh_client
+ @mock.patch.object(Connection, "extra_dejson",
new_callable=mock.PropertyMock, side_effect=AssertionError)
+ @mock.patch(
+ "airflow.providers.ssh.hooks.ssh.get_async_extra_dejson",
+ autospec=True,
+ return_value={"no_host_key_check": "true"},
+ )
+ # asyncssh.connect is wrapped by @async_context_manager, so autospec would
not be awaitable.
+ @mock.patch("asyncssh.connect", new_callable=mock.AsyncMock)
+ @mock.patch("airflow.providers.ssh.hooks.ssh.get_async_connection",
autospec=True)
+ @pytest.mark.asyncio
+ async def test_get_conn_reads_extra_asynchronously(
+ self, mock_get_connection, mock_connect, mock_get_extra,
mock_extra_dejson
+ ):
+ """``extra_dejson`` masks synchronously, so the async hook reads the
extra through common.compat."""
+ # The connection's own extra is empty: only the asynchronously read
extra can disable the check.
+ mock_get_connection.return_value = _connection({})
+
+ await SSHHookAsync(ssh_conn_id="test_conn")._get_conn()
+
+
mock_get_extra.assert_awaited_once_with(mock_get_connection.return_value)
+ mock_extra_dejson.assert_not_called()
+ assert mock_connect.call_args.kwargs["known_hosts"] is None
+
+ @pytest.mark.parametrize(
+ ("extra", "compression_algs"),
+ [
+ pytest.param({}, ["[email protected]", "zlib", "none"],
id="default"),
+ pytest.param({"compress": "true"}, ["[email protected]", "zlib",
"none"], id="true"),
+ pytest.param({"compress": "false"}, ["none"], id="false-string"),
+ pytest.param({"compress": False}, ["none"], id="false-bool"),
+ ],
+ )
+ @mock.patch("asyncssh.connect", new_callable=mock.AsyncMock)
+ @mock.patch("airflow.providers.ssh.hooks.ssh.get_async_connection",
autospec=True)
+ @pytest.mark.asyncio
+ async def test_get_conn_compresses_like_sync_hook(
+ self, mock_get_connection, mock_connect, extra, compression_algs
+ ):
+ """zlib is offered unless the ``compress`` extra is false, as
``SSHHook`` does."""
+ mock_get_connection.return_value = _connection({"no_host_key_check":
"true", **extra})
+
+ await SSHHookAsync(ssh_conn_id="test_conn")._get_conn()
+
+ assert mock_connect.call_args.kwargs["compression_algs"] ==
compression_algs
+
+ def test_asyncssh_offers_no_compression_first_by_default(self):
+ """
+ ``_get_conn`` passes its own list because asyncssh prefers ``none``
from 2.18, the floor.
+
+ Older versions preferred zlib, so supporting them means handling both
defaults in the hook.
+ """
+ assert get_default_compression_algs()[0] == b"none"
+
@mock.patch("airflow.providers.ssh.hooks.ssh.os.path.isfile",
return_value=False)
@mock.patch("asyncssh.connect", new_callable=mock.AsyncMock)
@mock.patch(
diff --git a/uv.lock b/uv.lock
index c6d4f6b2c83..a1af60266fe 100644
--- a/uv.lock
+++ b/uv.lock
@@ -8654,7 +8654,7 @@ docs = [
requires-dist = [
{ name = "apache-airflow", editable = "." },
{ name = "apache-airflow-providers-common-compat", editable =
"providers/common/compat" },
- { name = "asyncssh", specifier = ">=2.12.0" },
+ { name = "asyncssh", specifier = ">=2.18.0" },
{ name = "paramiko", specifier = ">=4.0.0" },
]
