From 672b5f1cd7578055973fcf6fc09161fe71a9a58e Mon Sep 17 00:00:00 2001 From: hypnguyen1209 Date: Mon, 13 Jul 2026 01:39:31 +0000 Subject: [PATCH] Validate deferred trigger classpath resolves to a BaseTrigger subclass 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 ++++++++++++++- .../tests/unit/jobs/test_triggerer_job.py | 20 +++++++++++++++++++ 3 files changed, 36 insertions(+), 1 deletion(-) create mode 100644 airflow-core/newsfragments/69792.bugfix.rst diff --git a/airflow-core/newsfragments/69792.bugfix.rst b/airflow-core/newsfragments/69792.bugfix.rst new file mode 100644 index 0000000000000..5b9dfc15fc90d --- /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 355763aed5f13..78bf1d21f93d1 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py @@ -1696,8 +1696,22 @@ def get_trigger_by_classpath(self, classpath: str) -> type[BaseTrigger]: """ 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 0f307fc150520..f40f281c313aa 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py @@ -1135,6 +1135,26 @@ def test_shared_stream_cohort_grace_period_config_wiring(self) -> None: 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"}):