This is an automated email from the ASF dual-hosted git repository.
potiuk 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 166a1ba1eec Validate deferred trigger classpath resolves to a
BaseTrigger subclass (#69792)
166a1ba1eec is described below
commit 166a1ba1eecf969ed33ad17dd1232b20d7ce5aea
Author: Nguyen Van Hiep <[email protected]>
AuthorDate: Sun Aug 30 02:34:20 2026 +0700
Validate deferred trigger classpath resolves to a BaseTrigger subclass
(#69792)
The triggerer resolves a trigger from the ``classpath`` carried in the
deferred-task Execution API payload via ``import_string(classpath)`` and
then
instantiates it with ``trigger_class(**kwargs)``.
``get_trigger_by_classpath``
did not check that the imported object is actually a ``BaseTrigger``
subclass,
so any importable callable (e.g. ``subprocess.check_output``) could be
loaded
and invoked in the long-running triggerer process.
Harden this by rejecting, before caching and before instantiation, any
``classpath`` that does not resolve to a ``type`` that is a ``BaseTrigger``
subclass. Legitimate triggers are unaffected (they are always
``BaseTrigger``
subclasses); an invalid classpath now fails the trigger cleanly instead of
being instantiated. Adds a unit test covering both the accepted and rejected
cases.
---
airflow-core/newsfragments/69792.bugfix.rst | 1 +
.../src/airflow/jobs/triggerer_job_runner.py | 16 +++++++++++++++-
airflow-core/tests/unit/jobs/test_triggerer_job.py | 20 ++++++++++++++++++++
3 files changed, 36 insertions(+), 1 deletion(-)
diff --git a/airflow-core/newsfragments/69792.bugfix.rst
b/airflow-core/newsfragments/69792.bugfix.rst
new file mode 100644
index 00000000000..5b9dfc15fc9
--- /dev/null
+++ b/airflow-core/newsfragments/69792.bugfix.rst
@@ -0,0 +1 @@
+Reject deferred-task trigger classpaths that do not resolve to a
``BaseTrigger`` subclass before the class is instantiated in the triggerer, so
a deferred task cannot cause an arbitrary importable callable to be invoked in
the triggerer process.
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index 6af0d1d5dfc..982702256f0 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -1733,8 +1733,22 @@ class TriggerRunner:
"""
Get a trigger class by its classpath ("path.to.module.classname").
+ The resolved object must be a
:class:`~airflow.triggers.base.BaseTrigger`
+ subclass. This is validated before the class is cached and, crucially,
+ before it is ever instantiated in ``create_triggers`` -- ``classpath``
+ originates from the (attacker-influenceable) deferred-task payload, so
+ without this check an arbitrary importable callable could be invoked in
+ the triggerer process.
+
Uses a cache dictionary to speed up lookups after the first time.
"""
if classpath not in self.trigger_cache:
- self.trigger_cache[classpath] = import_string(classpath)
+ trigger_class = import_string(classpath)
+ if not (isinstance(trigger_class, type) and
issubclass(trigger_class, BaseTrigger)):
+ raise TypeError(
+ f"The trigger classpath {classpath!r} does not resolve to
a "
+ f"{BaseTrigger.__module__}.{BaseTrigger.__qualname__}
subclass; "
+ f"refusing to load it."
+ )
+ self.trigger_cache[classpath] = trigger_class
return self.trigger_cache[classpath]
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 5c4c247d2a0..6f5de7eef8e 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -1196,6 +1196,26 @@ class TestTriggerRunner:
trigger_runner = TriggerRunner()
assert trigger_runner._shared_streams._cohort_grace_period == 3.0
+ def test_get_trigger_by_classpath_requires_basetrigger_subclass(self) ->
None:
+ """
+ ``classpath`` comes from the (attacker-influenceable) deferred-task
payload, so
+ ``get_trigger_by_classpath`` must refuse anything that is not a
``BaseTrigger``
+ subclass before it is cached and instantiated -- otherwise an
arbitrary importable
+ callable (e.g. ``subprocess.check_output``) could be invoked in the
triggerer.
+ """
+ trigger_runner = TriggerRunner()
+
+ # A real BaseTrigger subclass resolves and is cached.
+ assert (
+
trigger_runner.get_trigger_by_classpath("airflow.triggers.testing.SuccessTrigger")
+ is SuccessTrigger
+ )
+
+ # An arbitrary importable callable is rejected and never cached.
+ with pytest.raises(TypeError, match="does not resolve to a"):
+ trigger_runner.get_trigger_by_classpath("subprocess.check_output")
+ assert "subprocess.check_output" not in trigger_runner.trigger_cache
+
@pytest.mark.asyncio
async def
test_block_watchdog_does_not_log_when_threshold_is_not_exceeded(self) -> None:
with conf_vars({("triggerer",
"blocked_main_thread_warning_threshold"): "0.5"}):
