CVE-2026-86792

Airflow Kafka provider RCE via unfiltered import_string in connections

• 470
• Input Validation and Sanitization
• Local

Apache Airflow Apache Kafka provider versions 1.15.0 before 2.0.0 resolve dotted-path strings found in a Kafka connection's `extra` field into Python callables via `import_string`, with no allowlist, and hand them to the confluent-kafka client which invokes them. Deployments that have enabled the Kafka event producer — `dag_run_events_enabled` or `task_instance_events_enabled`, both disabled by default — build that client inside the scheduler process, so a user whose only privilege is editing Airflow connections gains arbitrary code execution in the control plane; the Airflow security model limits connection-configuration users to code execution on workers, not the scheduler. Deployments using Google Managed Kafka are not affected, because that code path overwrites any user-supplied `oauth_cb`; plain brokers and Amazon MSK are exposed. Users are recommended to upgrade to apache-airflow-providers-apache-kafka 2.0.0 or later, which adds an allowlist configuration option for connection-string callbacks.

CVE

CVE-2026-86792

CWE

470

Published

2026-09-16

OWASP

A08 Software and Data Integrity Failures

Orthogonal defect classification

Checking

Code defect classification

Missing Check

Category

Input Validation and Sanitization

Subcategory

Insecure Parsing or Deserialization

Accessibility scope

Local

Impact

Arbitrary Code Execution

Affected component

apache-airflow-providers-apache-kafka

Fixed by upgrading

Yes

Solution

Upgrade to apache‑airflow‑providers‑apache‑kafka 2.0.0 or later.

from airflow.utils.helpers import import_string
import confluent_kafka

def build_consumer(conn_extra):
    config = {
        "bootstrap.servers": conn_extra.get("bootstrap_servers", ""),
        "group.id": conn_extra.get("group_id", "default"),
    }
    # VULNERABLE: arbitrary import_string without allowlist
    if "oauth_cb" in conn_extra:
        config["oauth_cb"] = import_string(conn_extra["oauth_cb"])
    return confluent_kafka.Consumer(config)

from airflow.utils.helpers import import_string
from airflow.providers.apache.kafka.hooks.kafka import KAFKA_OAUTH_CALLBACK_ALLOWLIST
import confluent_kafka

def build_consumer(conn_extra):
    config = {
        "bootstrap.servers": conn_extra.get("bootstrap_servers", ""),
        "group.id": conn_extra.get("group_id", "default"),
    }
    # FIX: allowlist check before import_string
    cb_path = conn_extra.get("oauth_cb")
    if cb_path:
        if cb_path in KAFKA_OAUTH_CALLBACK_ALLOWLIST:
            config["oauth_cb"] = import_string(cb_path)
        else:
            raise ValueError("OAuth callback not allowed")
    return confluent_kafka.Consumer(config)

__VAITP_MODEL_REFUSED__

@misc{vaitp:cve202686792,
  title        = {{Airflow Kafka provider RCE via unfiltered import_string in connections}},
  author       = {Bogaerts, Fr\'ed\'eric and Ivaki, Naghmeh and Fonseca, Jos\'e},
  year         = {2026},
  note         = {VAITP Python Vulnerability Dataset, entry CVE-2026-86792},
  howpublished = {\url{https://netpack.pt/vaitp/vulnerability/CVE-2026-86792/}}
}