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 d675437ccdf Add an allowlist config option for Kafka connection string
callbacks (#72208)
d675437ccdf is described below
commit d675437ccdf87a3e36d8c8559bdcff6e62a5f618
Author: Christos Bisias <[email protected]>
AuthorDate: Tue Sep 8 14:10:06 2026 +0300
Add an allowlist config option for Kafka connection string callbacks
(#72208)
---
providers/apache/kafka/docs/configurations-ref.rst | 7 ++-
providers/apache/kafka/docs/connections/kafka.rst | 12 +++++
providers/apache/kafka/provider.yaml | 22 +++++++++
.../providers/apache/kafka/get_provider_info.py | 14 +++++-
.../airflow/providers/apache/kafka/hooks/base.py | 55 ++++++++++++++++++++--
.../tests/unit/apache/kafka/hooks/test_base.py | 55 ++++++++++++++++++++++
.../tests/unit/apache/kafka/hooks/test_consume.py | 3 ++
.../tests/unit/apache/kafka/hooks/test_produce.py | 3 ++
8 files changed, 164 insertions(+), 7 deletions(-)
diff --git a/providers/apache/kafka/docs/configurations-ref.rst
b/providers/apache/kafka/docs/configurations-ref.rst
index be5f3ad011d..2a5b108d497 100644
--- a/providers/apache/kafka/docs/configurations-ref.rst
+++ b/providers/apache/kafka/docs/configurations-ref.rst
@@ -76,8 +76,11 @@ To enable event publishing you need to
The connection's ``extra`` JSON accepts the full confluent-kafka client
configuration — including SASL/TLS options and callbacks (e.g. ``error_cb``,
-``oauth_cb``) given as dotted-path strings, which are resolved to callables
-before the producer is built.
+``oauth_cb``) given as dotted-path strings. A string-valued callback is only
+resolved when its full importable path is listed in the
+:ref:`config:apache_kafka__callback_allowlist` option; the example below
requires
+``callback_allowlist = my_company.auth.oauth_cb``. This is enforced for
security
+reasons, to prevent malicious callbacks from being executed.
.. code-block:: json
diff --git a/providers/apache/kafka/docs/connections/kafka.rst
b/providers/apache/kafka/docs/connections/kafka.rst
index 80954116615..486a5186d3f 100644
--- a/providers/apache/kafka/docs/connections/kafka.rst
+++ b/providers/apache/kafka/docs/connections/kafka.rst
@@ -40,6 +40,18 @@ used to specify a callback function by providing a path to
the function. e.g ``"
of parameters are described in the
`Confluent Kafka python library
<https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md>`_.
+.. warning::
+
+ Callback options supplied as dotted-path strings (``error_cb``,
``throttle_cb``, ``stats_cb``,
+ ``log_cb``, ``oauth_cb``, ``on_commit``) are only imported when listed in
the
+ :ref:`config:apache_kafka__callback_allowlist` configuration. Each
allowlist entry is the full
+ importable path of the callback itself — module plus attribute, e.g.
+ ``my_company.kafka.auth.oauth_cb`` — and must match the connection value
exactly; a bare module
+ such as ``my_company.kafka.auth`` does not lead to authorization of the
callables inside it.
+ This is enforced for security reasons, to prevent malicious callbacks from
being executed.
+ The allowlist is empty by default, which disables string-valued callbacks
entirely.
+ Managed authentication (Amazon MSK IAM, Google Managed Kafka) does not
rely on this and is unaffected.
+
If you are defining the Airflow connection from the Airflow UI, the ``extra``
field will be renamed to ``Config Dict``.
Most operators and hooks will check that at the minimum the
``bootstrap.servers`` key exists and has a value set to be valid.
diff --git a/providers/apache/kafka/provider.yaml
b/providers/apache/kafka/provider.yaml
index d530a1c5ff2..09f9eb64d92 100644
--- a/providers/apache/kafka/provider.yaml
+++ b/providers/apache/kafka/provider.yaml
@@ -138,6 +138,28 @@ plugins:
plugin-class:
airflow.providers.apache.kafka.plugins.event_producer.KafkaEventProducerPlugin
config:
+ apache_kafka:
+ description: |
+ Common settings for the Apache Kafka Provider.
+ options:
+ callback_allowlist:
+ description: |
+ Comma-separated list of callbacks that may be resolved from a Kafka
+ connection extra into confluent-kafka callback options (``error_cb``,
+ ``throttle_cb``, ``stats_cb``, ``log_cb``, ``oauth_cb``,
``on_commit``).
+ Each entry is the full importable path of the callback itself —
module
+ plus attribute, e.g. ``my_company.kafka.auth.oauth_cb`` — matched
exactly
+ against the dotted-path string on the connection extra (a bare
module such
+ as ``my_company.kafka.auth`` won't lead to authorization of the
callables
+ inside it). This is enforced for security reasons, to prevent
malicious
+ callbacks from being executed: only paths listed here are resolved,
and any
+ other string-valued callback is refused. Empty (default) disables
+ string-valued callbacks entirely. Managed authentication (Amazon MSK
IAM,
+ Google Managed Kafka) does not rely on this and is unaffected.
+ version_added: 1.16.0
+ type: string
+ example: "my_company.kafka.auth.oauth_cb"
+ default: ""
kafka_event_producer:
description: |
Settings for the Kafka event producer plugin that publishes Airflow
diff --git
a/providers/apache/kafka/src/airflow/providers/apache/kafka/get_provider_info.py
b/providers/apache/kafka/src/airflow/providers/apache/kafka/get_provider_info.py
index 31162c55425..80c1744717a 100644
---
a/providers/apache/kafka/src/airflow/providers/apache/kafka/get_provider_info.py
+++
b/providers/apache/kafka/src/airflow/providers/apache/kafka/get_provider_info.py
@@ -108,6 +108,18 @@ def get_provider_info():
}
],
"config": {
+ "apache_kafka": {
+ "description": "Common settings for the Apache Kafka
Provider.\n",
+ "options": {
+ "callback_allowlist": {
+ "description": "Comma-separated list of callbacks that
may be resolved from a Kafka\nconnection extra into confluent-kafka callback
options (``error_cb``,\n``throttle_cb``, ``stats_cb``, ``log_cb``,
``oauth_cb``, ``on_commit``).\nEach entry is the full importable path of the
callback itself — module\nplus attribute, e.g.
``my_company.kafka.auth.oauth_cb`` — matched exactly\nagainst the dotted-path
string on the connection extra (a bare module such\nas ``my_company [...]
+ "version_added": "1.16.0",
+ "type": "string",
+ "example": "my_company.kafka.auth.oauth_cb",
+ "default": "",
+ }
+ },
+ },
"kafka_event_producer": {
"description": "Settings for the Kafka event producer plugin
that publishes Airflow\nDagRun and TaskInstance state-change events to a Kafka
topic.\n",
"options": {
@@ -203,6 +215,6 @@ def get_provider_info():
"default": "60",
},
},
- }
+ },
},
}
diff --git
a/providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py
b/providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py
index 4fb4cc71707..5132fa8a1a9 100644
--- a/providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py
+++ b/providers/apache/kafka/src/airflow/providers/apache/kafka/hooks/base.py
@@ -23,12 +23,18 @@ from typing import Any
from confluent_kafka.admin import AdminClient
from airflow.providers.common.compat.module_loading import import_string
-from airflow.providers.common.compat.sdk import BaseHook
+from airflow.providers.common.compat.sdk import BaseHook, conf
# librdkafka config options whose values are callables. They can be provided
as dotted-path
# strings on the connection extra and are resolved to callables before the
client is built.
CALLBACK_CONFIG_KEYS = ("error_cb", "throttle_cb", "stats_cb", "log_cb",
"oauth_cb", "on_commit")
+
+KAFKA_COMMON_CONFIG_SECTION = "apache_kafka"
+# Configuration allowlist for callbacks that are allowed to be resolved from a
connection extra.
+CALLBACK_ALLOWLIST_CONFIG_OPTION = "callback_allowlist"
+
+
# Amazon MSK bootstrap servers follow a predictable naming scheme, e.g.
# b-1.demo.abcde1.c2.kafka.us-east-1.amazonaws.com:9098
(provisioned)
# boot-abcde1.c2.kafka-serverless.us-east-1.amazonaws.com:9098
(serverless)
@@ -90,12 +96,53 @@ class KafkaBaseHook(BaseHook):
def _get_client(self, config) -> Any:
return AdminClient(config)
+ @staticmethod
+ def _get_callback_allowlist() -> frozenset[str]:
+ conf_value = (
+ conf.get(KAFKA_COMMON_CONFIG_SECTION,
CALLBACK_ALLOWLIST_CONFIG_OPTION, fallback="") or ""
+ )
+ return frozenset(cb_path.strip() for cb_path in conf_value.split(",")
if cb_path.strip())
+
def _resolve_callbacks(self, config: dict[str, Any]) -> None:
- """Resolve callback options provided as dotted-path strings into
callables."""
+ """
+ Resolve callback options provided as dotted-path strings into
callables.
+
+ A callback is resolved only when its full importable path is listed in
+ the ``[apache_kafka] callback_allowlist`` configuration. This is
enforced
+ for security reasons, to prevent malicious callbacks from being
executed.
+ """
+ allowlist: frozenset[str] | None = None
for key in CALLBACK_CONFIG_KEYS:
value = config.get(key)
- if isinstance(value, str):
- config[key] = import_string(value)
+ if not isinstance(value, str):
+ continue
+ if allowlist is None:
+ # Get the allowlist from the config only once, if it hasn't
+ # already been initialized by a previous iteration.
+ allowlist = self._get_callback_allowlist()
+ if not allowlist:
+ # If the allowlist is empty, break the iteration immediately
by raising an error.
+ self.log.warning(
+ "Kafka connection %r requests callback %s=%r, but [%s] %s
is empty. Add the "
+ "full importable path of each callback you trust (e.g.
'my_pkg.auth.oauth_cb') "
+ "to allow string-valued callbacks.",
+ self.kafka_config_id,
+ key,
+ value,
+ KAFKA_COMMON_CONFIG_SECTION,
+ CALLBACK_ALLOWLIST_CONFIG_OPTION,
+ )
+ raise ValueError(
+ f"Refusing to resolve Kafka callback {key}={value!r}: the "
+ f"[{KAFKA_COMMON_CONFIG_SECTION}]
{CALLBACK_ALLOWLIST_CONFIG_OPTION} is empty."
+ )
+ if value not in allowlist:
+ raise ValueError(
+ f"Refusing to resolve Kafka callback {key}={value!r}: it
is not in the "
+ f"[{KAFKA_COMMON_CONFIG_SECTION}]
{CALLBACK_ALLOWLIST_CONFIG_OPTION} "
+ f"({', '.join(sorted(allowlist))})."
+ )
+ config[key] = import_string(value)
def _build_config(self) -> dict[str, Any]:
"""
diff --git a/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_base.py
b/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_base.py
index 4d472615344..bedc253942c 100644
--- a/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_base.py
+++ b/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_base.py
@@ -29,6 +29,10 @@ from airflow.providers.apache.kafka.hooks.base import (
)
from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException
+from tests_common.test_utils.config import conf_vars
+
+CALLBACK_ALLOWLIST = {("apache_kafka", "callback_allowlist"):
"json.loads,json.dumps"}
+
try:
import importlib.util
@@ -66,6 +70,7 @@ class TestKafkaBaseHook:
with pytest.raises(ValueError, match="must be provided"):
hook.get_conn()
+ @conf_vars(CALLBACK_ALLOWLIST)
@mock.patch(f"{BASEHOOK_PATCH_PATH}.get_connection")
def test_callbacks_resolved_from_connection_dotted_path(self,
mock_get_connection):
stats_cb = MagicMock()
@@ -82,6 +87,41 @@ class TestKafkaBaseHook:
# Already-callable values on the connection are passed through
unchanged.
assert config["stats_cb"] is stats_cb
+ @conf_vars(CALLBACK_ALLOWLIST)
+ @mock.patch(f"{BASEHOOK_PATCH_PATH}.get_connection")
+ def test_callback_resolution_tolerates_whitespace_in_allowlist(self,
mock_get_connection):
+ mock_get_connection.return_value.extra_dejson = {
+ "bootstrap.servers": "localhost:9092",
+ "error_cb": "json.loads",
+ }
+ with conf_vars({("apache_kafka", "callback_allowlist"): " json.loads
,, json.dumps "}):
+ config = SomeKafkaHook().get_conn
+ assert config["error_cb"] is json.loads
+
+ @mock.patch.object(KafkaBaseHook, "log")
+ @mock.patch(f"{BASEHOOK_PATCH_PATH}.get_connection")
+ def test_string_callback_refused_when_allowlist_empty(self,
mock_get_connection, mock_log, hook):
+ mock_get_connection.return_value.extra_dejson = {
+ "bootstrap.servers": "localhost:9092",
+ "oauth_cb": "os.system",
+ }
+ with pytest.raises(ValueError, match="callback_allowlist is empty"):
+ _ = hook.get_conn
+ mock_log.warning.assert_called_once()
+
+ @conf_vars(CALLBACK_ALLOWLIST)
+ @mock.patch.object(KafkaBaseHook, "log")
+ @mock.patch(f"{BASEHOOK_PATCH_PATH}.get_connection")
+ def test_string_callback_refused_when_not_in_allowlist(self,
mock_get_connection, mock_log, hook):
+ mock_get_connection.return_value.extra_dejson = {
+ "bootstrap.servers": "localhost:9092",
+ "oauth_cb": "os.system",
+ }
+ with pytest.raises(ValueError, match="oauth_cb='os.system': it is not
in the"):
+ _ = hook.get_conn
+ # A populated allowlist is a configuration choice, not a missing one,
so no "configure" hint.
+ mock_log.warning.assert_not_called()
+
@mock.patch("airflow.providers.apache.kafka.hooks.base.AdminClient")
@mock.patch(f"{BASEHOOK_PATCH_PATH}.get_connection")
def test_test_connection(self, mock_get_connection, admin_client, hook):
@@ -107,6 +147,7 @@ class TestKafkaBaseHook:
mock_admin_instance.list_topics.assert_called_once_with(timeout=TIMEOUT)
assert connection == (False, "Failed to establish connection.")
+ @conf_vars(CALLBACK_ALLOWLIST)
@mock.patch("airflow.providers.apache.kafka.hooks.base.AdminClient")
@mock.patch(f"{BASEHOOK_PATCH_PATH}.get_connection")
def test_test_connection_resolves_callbacks(self, mock_get_connection,
admin_client, hook):
@@ -119,6 +160,20 @@ class TestKafkaBaseHook:
assert hook.test_connection() == (True, "Connection successful.")
admin_client.assert_called_once_with({"bootstrap.servers":
"localhost:9092", "error_cb": json.loads})
+ @mock.patch("airflow.providers.apache.kafka.hooks.base.AdminClient")
+ @mock.patch(f"{BASEHOOK_PATCH_PATH}.get_connection")
+ def test_test_connection_reports_refused_callback(self,
mock_get_connection, admin_client, hook):
+ # A callback outside the allowlist must surface as a failed connection
test, not build a client.
+ mock_get_connection.return_value.extra_dejson = {
+ "bootstrap.servers": "localhost:9092",
+ "oauth_cb": "os.system",
+ }
+ is_success, message = hook.test_connection()
+ assert is_success is False
+ assert "Refusing to resolve Kafka callback" in message
+ assert "callback_allowlist" in message
+ admin_client.assert_not_called()
+
@mock.patch("airflow.providers.apache.kafka.hooks.base.AdminClient")
@mock.patch(f"{BASEHOOK_PATCH_PATH}.get_connection")
def test_test_connection_exception(self, mock_get_connection,
admin_client, hook):
diff --git
a/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_consume.py
b/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_consume.py
index 4036980c626..bc48fa4dfae 100644
--- a/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_consume.py
+++ b/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_consume.py
@@ -27,6 +27,8 @@ from airflow.models import Connection
# Import Hook
from airflow.providers.apache.kafka.hooks.consume import KafkaConsumerHook,
error_callback
+from tests_common.test_utils.config import conf_vars
+
class TestConsumerHook:
"""
@@ -68,6 +70,7 @@ class TestConsumerHook:
assert consumer == mock_consumer.return_value
mock_consumer.return_value.subscribe.assert_called_once_with(["test_1"])
+ @conf_vars({("apache_kafka", "callback_allowlist"): "json.loads"})
@patch("airflow.providers.apache.kafka.hooks.consume.Consumer")
def
test_user_error_cb_from_connection_resolved_and_not_overridden_by_default(
self, mock_consumer, create_connection_without_db
diff --git
a/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_produce.py
b/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_produce.py
index e8f8cd8f667..6c35eeb606f 100644
--- a/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_produce.py
+++ b/providers/apache/kafka/tests/unit/apache/kafka/hooks/test_produce.py
@@ -26,6 +26,8 @@ from confluent_kafka.admin import AdminClient
from airflow.models import Connection
from airflow.providers.apache.kafka.hooks.produce import KafkaProducerHook
+from tests_common.test_utils.config import conf_vars
+
log = logging.getLogger(__name__)
@@ -61,6 +63,7 @@ class TestProducerHook:
mock_client.return_value = mock_client_spec
assert self.hook.get_producer() == self.hook.get_conn
+ @conf_vars({("apache_kafka", "callback_allowlist"): "json.dumps"})
@patch("airflow.providers.apache.kafka.hooks.produce.Producer")
def test_connection_callback_resolved_from_dotted_path(self,
mock_producer, create_connection_without_db):
# A dotted-path ``oauth_cb`` on the connection extras is resolved to
the callable
