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"}):