Allow configuring XCom sidecar container security context

airflow-core/tests/unit/always/test_project_structure.py

@@ -97,7 +97,6 @@ def test_providers_modules_should_have_tests(self):
             "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_kubernetes_pod.py",
             "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_delete_from.py",
             "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_k8s_hashlib_wrapper.py",
-            "providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_xcom_sidecar.py",
             "providers/common/sql/tests/unit/common/sql/datafusion/test_base.py",
             "providers/common/sql/tests/unit/common/sql/datafusion/test_exceptions.py",
             "providers/common/ai/tests/unit/common/ai/test_exceptions.py",

docs/spelling_wordlist.txt

@@ -1175,6 +1175,7 @@ onboarded
 onboarding
 OnFailure
 Oozie
+OPA
 OpenAI
 openai
 openapi

kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py

@@ -964,6 +964,7 @@ def test_pod_template_file(
         hook_mock.return_value.is_in_cluster = False
         hook_mock.return_value.get_xcom_sidecar_container_image.return_value = None
         hook_mock.return_value.get_xcom_sidecar_container_resources.return_value = None
+        hook_mock.return_value.get_xcom_sidecar_container_security_context.return_value = None
         hook_mock.return_value.get_connection.return_value = Connection(conn_id="kubernetes_default")
         extract_xcom_mock.return_value = "{}"
         k = KubernetesPodOperator(

providers/cncf/kubernetes/docs/connections/kubernetes.rst

@@ -71,6 +71,17 @@ Xcom sidecar image
   Define the ``image`` used by the ``PodDefaults.SIDECAR_CONTAINER`` (defaults to ``"alpine"``) to allow private
   repositories, as well as custom image overrides.
 
+Xcom sidecar resources (JSON format)
+  Define the resource ``requests``/``limits`` for the XCom sidecar container as a JSON object, e.g.
+  ``{"requests": {"cpu": "1m", "memory": "10Mi"}}``.
+
+Xcom sidecar security context (JSON format)
+  Define the ``securityContext`` for the XCom sidecar container as a JSON object, e.g.
+  ``{"allowPrivilegeEscalation": false, "readOnlyRootFilesystem": true, "seccompProfile": {"type": "RuntimeDefault"}}``.
+  Useful when clusters enforce Pod Security Standards or admission policies (e.g. OPA/Gatekeeper) on the
+  injected sidecar. ``KubernetesPodOperator`` Dag authors can override this per task via the
+  ``xcom_sidecar_container_security_context`` argument.
+
 Example storing connection in env var using URI format:
 
 .. code-block:: bash

providers/cncf/kubernetes/docs/operators.rst

@@ -197,6 +197,12 @@ alongside the Pod. The Pod must write the XCom value into this location at the `
 .. note::
   An invalid json content will fail, example ``echo 'hello' > /airflow/xcom/return.json`` fail and  ``echo '\"hello\"' > /airflow/xcom/return.json`` work
 
+.. note::
+  In clusters that enforce Pod Security Standards or admission policies (e.g. OPA/Gatekeeper), the injected
+  XCom sidecar container may be rejected unless it declares a security context. Set a cluster-wide default via
+  the ``xcom_sidecar_container_security_context`` field on the Kubernetes connection, or override it per task
+  with the ``xcom_sidecar_container_security_context`` argument of ``KubernetesPodOperator``.
+
 
 See the following example on how this occurs:
 

providers/cncf/kubernetes/provider.yaml

@@ -233,6 +233,12 @@ connection-types:
           type:
             - string
             - 'null'
+      xcom_sidecar_container_security_context:
+        label: XCom sidecar security context (JSON format)
+        schema:
+          type:
+            - string
+            - 'null'
     ui-field-behaviour:
       hidden-fields:
         - host

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/get_provider_info.py

@@ -108,6 +108,10 @@ def get_provider_info():
                         "label": "XCom sidecar resources (JSON format)",
                         "schema": {"type": ["string", "null"]},
                     },
+                    "xcom_sidecar_container_security_context": {
+                        "label": "XCom sidecar security context (JSON format)",
+                        "schema": {"type": ["string", "null"]},
+                    },
                 },
                 "ui-field-behaviour": {
                     "hidden-fields": ["host", "schema", "login", "password", "port", "extra"]

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py

@@ -152,6 +152,9 @@ def get_xcom_sidecar_container_image(self) -> str | None:
     def get_xcom_sidecar_container_resources(self) -> str | None:
         """Return the xcom sidecar resources that defined in the connection."""
 
+    def get_xcom_sidecar_container_security_context(self) -> str | None:
+        """Return the xcom sidecar security context that defined in the connection."""
+
 
 class KubernetesHook(BaseHook, PodOperatorHookProtocol):
     """
@@ -213,6 +216,9 @@ def get_connection_form_widgets(cls) -> dict[str, Any]:
             "xcom_sidecar_container_resources": StringField(
                 lazy_gettext("XCom sidecar resources (JSON format)"), widget=BS3TextFieldWidget()
             ),
+            "xcom_sidecar_container_security_context": StringField(
+                lazy_gettext("XCom sidecar security context (JSON format)"), widget=BS3TextFieldWidget()
+            ),
         }
 
     @classmethod
@@ -526,6 +532,13 @@ def get_xcom_sidecar_container_resources(self):
             return None
         return json.loads(field)
 
+    def get_xcom_sidecar_container_security_context(self):
+        """Return the xcom sidecar security context that defined in the connection."""
+        field = self._get_field("xcom_sidecar_container_security_context")
+        if not field:
+            return None
+        return json.loads(field)
+
     def get_pod_log_stream(
         self,
         pod_name: str,

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py

@@ -204,6 +204,8 @@ class KubernetesPodOperator(BaseOperator):
     :param tolerations: A list of kubernetes tolerations.
     :param security_context: security options the pod should run with (PodSecurityContext).
     :param container_security_context: security options the container should run with.
+    :param xcom_sidecar_container_security_context: security options the xcom sidecar container should
+        run with. Overrides the value configured on the Kubernetes connection.
     :param dnspolicy: dnspolicy for the pod.
     :param dns_config: dns configuration (ip addresses, searches, options) for the pod.
     :param hostname: hostname for the pod. (templated)
@@ -343,6 +345,7 @@ def __init__(
         tolerations: list[k8s.V1Toleration] | None = None,
         security_context: k8s.V1PodSecurityContext | dict | None = None,
         container_security_context: k8s.V1SecurityContext | dict | None = None,
+        xcom_sidecar_container_security_context: k8s.V1SecurityContext | dict | None = None,
         dnspolicy: str | None = None,
         dns_config: k8s.V1PodDNSConfig | None = None,
         hostname: str | None = None,
@@ -431,6 +434,7 @@ def __init__(
         )
         self.security_context = security_context or {}
         self.container_security_context = container_security_context
+        self.xcom_sidecar_container_security_context = xcom_sidecar_container_security_context
         self.dnspolicy = dnspolicy
         self.dns_config = dns_config
         self.hostname = hostname
@@ -1555,6 +1559,11 @@ def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:
                 pod,
                 sidecar_container_image=self.hook.get_xcom_sidecar_container_image(),
                 sidecar_container_resources=self.hook.get_xcom_sidecar_container_resources(),
+                sidecar_container_security_context=(
+                    self.xcom_sidecar_container_security_context
+                    if self.xcom_sidecar_container_security_context is not None
+                    else self.hook.get_xcom_sidecar_container_security_context()
+                ),
             )
 
         labels = self._get_ti_pod_labels(context)

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/xcom_sidecar.py

@@ -58,6 +58,7 @@ def add_xcom_sidecar(
     *,
     sidecar_container_image: str | None = None,
     sidecar_container_resources: k8s.V1ResourceRequirements | dict | None = None,
+    sidecar_container_security_context: k8s.V1SecurityContext | dict | None = None,
 ) -> k8s.V1Pod:
     """Add sidecar."""
     pod_cp = copy.deepcopy(pod)
@@ -69,6 +70,8 @@ def add_xcom_sidecar(
     sidecar.image = sidecar_container_image or PodDefaults.SIDECAR_CONTAINER.image
     if sidecar_container_resources:
         sidecar.resources = sidecar_container_resources
+    if sidecar_container_security_context is not None:
+        sidecar.security_context = sidecar_container_security_context
     pod_cp.spec.containers.append(sidecar)
 
     return pod_cp

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/decorators/test_kubernetes_commons.py

@@ -111,6 +111,7 @@ def setup(self, dag_maker):
             }
         )
         self.mock_hook = mock.patch(HOOK_CLASS).start()
+        self.mock_hook.return_value.get_xcom_sidecar_container_security_context.return_value = None
 
         # Without this patch each time pod manager would try to extract logs from the pod
         # and log an error about it's inability to get containers for the log

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py

@@ -227,6 +227,19 @@ def setup_connections(self, create_connection_without_db):
                 },
             ),
             ("sidecar_container_resources_empty", {"xcom_sidecar_container_resources": ""}),
+            (
+                "sidecar_container_security_context",
+                {
+                    "xcom_sidecar_container_security_context": json.dumps(
+                        {
+                            "allowPrivilegeEscalation": False,
+                            "readOnlyRootFilesystem": True,
+                            "seccompProfile": {"type": "RuntimeDefault"},
+                        }
+                    ),
+                },
+            ),
+            ("sidecar_container_security_context_empty", {"xcom_sidecar_container_security_context": ""}),
         ]
         for conn_id, extra in connections:
             create_connection_without_db(
@@ -560,6 +573,27 @@ def test_get_xcom_sidecar_container_resources(self, conn_id, expected):
         hook = KubernetesHook(conn_id=conn_id)
         assert hook.get_xcom_sidecar_container_resources() == expected
 
+    @pytest.mark.parametrize(
+        ("conn_id", "expected"),
+        (
+            pytest.param(
+                "sidecar_container_security_context",
+                {
+                    "allowPrivilegeEscalation": False,
+                    "readOnlyRootFilesystem": True,
+                    "seccompProfile": {"type": "RuntimeDefault"},
+                },
+                id="sidecar-with-security-context",
+            ),
+            pytest.param(
+                "sidecar_container_security_context_empty", None, id="sidecar-without-security-context"
+            ),
+        ),
+    )
+    def test_get_xcom_sidecar_container_security_context(self, conn_id, expected):
+        hook = KubernetesHook(conn_id=conn_id)
+        assert hook.get_xcom_sidecar_container_security_context() == expected
+
     @patch("kubernetes.config.kube_config.KubeConfigLoader")
     @patch("kubernetes.config.kube_config.KubeConfigMerger")
     def test_client_types(self, mock_kube_config_merger, mock_kube_config_loader):

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py

@@ -930,6 +930,60 @@ def test_xcom_sidecar_container_resources_custom(self):
             pod = k.build_pod_request_obj(create_context(k))
             assert pod.spec.containers[1].resources == resources
 
+    def test_xcom_sidecar_container_security_context_default(self):
+        k = KubernetesPodOperator(
+            name="test",
+            task_id="task",
+            do_xcom_push=True,
+        )
+        pod = k.build_pod_request_obj(create_context(k))
+        assert pod.spec.containers[1].security_context is None
+
+    @patch(f"{HOOK_CLASS}.get_xcom_sidecar_container_security_context")
+    def test_xcom_sidecar_container_security_context_from_connection(self, mock_get_security_context):
+        security_context = {
+            "allowPrivilegeEscalation": False,
+            "readOnlyRootFilesystem": True,
+            "seccompProfile": {"type": "RuntimeDefault"},
+        }
+        mock_get_security_context.return_value = security_context
+        k = KubernetesPodOperator(
+            name="test",
+            task_id="task",
+            do_xcom_push=True,
+        )
+        pod = k.build_pod_request_obj(create_context(k))
+        assert pod.spec.containers[1].security_context == security_context
+
+    @patch(f"{HOOK_CLASS}.get_xcom_sidecar_container_security_context")
+    def test_xcom_sidecar_container_security_context_operator_overrides_connection(
+        self, mock_get_security_context
+    ):
+        mock_get_security_context.return_value = {"readOnlyRootFilesystem": False}
+        operator_security_context = {"readOnlyRootFilesystem": True}
+        k = KubernetesPodOperator(
+            name="test",
+            task_id="task",
+            do_xcom_push=True,
+            xcom_sidecar_container_security_context=operator_security_context,
+        )
+        pod = k.build_pod_request_obj(create_context(k))
+        assert pod.spec.containers[1].security_context == operator_security_context
+
+    @patch(f"{HOOK_CLASS}.get_xcom_sidecar_container_security_context")
+    def test_xcom_sidecar_container_security_context_empty_operator_overrides_connection(
+        self, mock_get_security_context
+    ):
+        mock_get_security_context.return_value = {"readOnlyRootFilesystem": True}
+        k = KubernetesPodOperator(
+            name="test",
+            task_id="task",
+            do_xcom_push=True,
+            xcom_sidecar_container_security_context={},
+        )
+        pod = k.build_pod_request_obj(create_context(k))
+        assert pod.spec.containers[1].security_context == {}
+
     def test_image_pull_policy_correctly_set(self):
         k = KubernetesPodOperator(
             task_id="task",
@@ -1993,6 +2047,7 @@ def test_get_logs_but_not_for_base_container(
     ):
         hook_mock.return_value.get_xcom_sidecar_container_image.return_value = None
         hook_mock.return_value.get_xcom_sidecar_container_resources.return_value = None
+        hook_mock.return_value.get_xcom_sidecar_container_security_context.return_value = None
         k = KubernetesPodOperator(
             namespace="default",
             image="ubuntu:16.04",

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/utils/test_xcom_sidecar.py

@@ -0,0 +1,71 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import pytest
+from kubernetes.client import models as k8s
+
+from airflow.providers.cncf.kubernetes.utils.xcom_sidecar import PodDefaults, add_xcom_sidecar
+
+
+def _base_pod() -> k8s.V1Pod:
+    return k8s.V1Pod(spec=k8s.V1PodSpec(containers=[k8s.V1Container(name="base")]))
+
+
+def _sidecar_of(pod: k8s.V1Pod) -> k8s.V1Container:
+    return next(c for c in pod.spec.containers if c.name == PodDefaults.SIDECAR_CONTAINER_NAME)
+
+
+@pytest.mark.parametrize(
+    "security_context",
+    [
+        pytest.param(
+            {
+                "allowPrivilegeEscalation": False,
+                "readOnlyRootFilesystem": True,
+                "seccompProfile": {"type": "RuntimeDefault"},
+            },
+            id="dict",
+        ),
+        pytest.param(
+            k8s.V1SecurityContext(
+                allow_privilege_escalation=False,
+                read_only_root_filesystem=True,
+                seccomp_profile=k8s.V1SeccompProfile(type="RuntimeDefault"),
+            ),
+            id="model",
+        ),
+    ],
+)
+def test_add_xcom_sidecar_sets_security_context(security_context):
+    pod = add_xcom_sidecar(_base_pod(), sidecar_container_security_context=security_context)
+    assert _sidecar_of(pod).security_context == security_context
+
+
+def test_add_xcom_sidecar_without_security_context():
+    pod = add_xcom_sidecar(_base_pod())
+    assert _sidecar_of(pod).security_context is None
+
+
+def test_add_xcom_sidecar_empty_security_context():
+    pod = add_xcom_sidecar(_base_pod(), sidecar_container_security_context={})
+    assert _sidecar_of(pod).security_context == {}
+
+
+def test_add_xcom_sidecar_does_not_mutate_shared_default():
+    add_xcom_sidecar(_base_pod(), sidecar_container_security_context={"readOnlyRootFilesystem": True})
+    assert PodDefaults.SIDECAR_CONTAINER.security_context is None

task-sdk/src/airflow/sdk/definitions/decorators/__init__.pyi

@@ -535,6 +535,7 @@ class TaskDecoratorCollection:
         tolerations: list[k8s.V1Toleration] | None = None,
         security_context: k8s.V1PodSecurityContext | dict | None = None,
         container_security_context: k8s.V1SecurityContext | dict | None = None,
+        xcom_sidecar_container_security_context: k8s.V1SecurityContext | dict | None = None,
         dnspolicy: str | None = None,
         dns_config: k8s.V1PodDNSConfig | None = None,
         hostname: str | None = None,
@@ -624,6 +625,8 @@ class TaskDecoratorCollection:
         :param security_context: Security options the pod should run with
             (PodSecurityContext).
         :param container_security_context: security options the container should run with.
+        :param xcom_sidecar_container_security_context: security options the xcom sidecar container
+            should run with. Overrides the value configured on the Kubernetes connection.
         :param dnspolicy: DNS policy for the pod.
         :param dns_config: dns configuration (ip addresses, searches, options) for the pod.
         :param hostname: hostname for the pod.
@@ -708,6 +711,7 @@ class TaskDecoratorCollection:
         tolerations: list[k8s.V1Toleration] | None = None,
         security_context: k8s.V1PodSecurityContext | dict | None = None,
         container_security_context: k8s.V1SecurityContext | dict | None = None,
+        xcom_sidecar_container_security_context: k8s.V1SecurityContext | dict | None = None,
         dnspolicy: str | None = None,
         dns_config: k8s.V1PodDNSConfig | None = None,
         hostname: str | None = None,
@@ -794,6 +798,8 @@ class TaskDecoratorCollection:
         :param security_context: Security options the pod should run with
             (PodSecurityContext).
         :param container_security_context: security options the container should run with.
+        :param xcom_sidecar_container_security_context: security options the xcom sidecar container
+            should run with. Overrides the value configured on the Kubernetes connection.
         :param dnspolicy: DNS policy for the pod.
         :param dns_config: dns configuration (ip addresses, searches, options) for the pod.
         :param hostname: hostname for the pod.