Add configurable automountServiceAccountToken for the KubernetesPodOperator

kubernetes-tests/tests/kubernetes_tests/test_kubernetes_pod_operator.py

@@ -134,6 +134,7 @@ def setup_tests(self, test_label):
             },
             "spec": {
                 "affinity": {},
+                "automountServiceAccountToken": True,
                 "containers": [
                     {
                         "image": "ubuntu:16.04",
@@ -998,6 +999,7 @@ def test_pod_template_file(
             },
             "spec": {
                 "affinity": {},
+                "automountServiceAccountToken": True,
                 "containers": [
                     {
                         "args": ["--vm", "1", "--vm-bytes", "150M", "--vm-hang", "1"],

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

@@ -180,6 +180,7 @@ class KubernetesPodOperator(BaseOperator):
         If more than one secret is required, provide a
         comma separated list: secret_a,secret_b
     :param service_account_name: Name of the service account
+    :param automount_service_account_token: indicates whether pods running as this service account should have an API token automatically mounted
     :param hostnetwork: If True enable host networking on the pod.
     :param host_aliases: A list of host aliases to apply to the containers in the pod.
     :param tolerations: A list of kubernetes tolerations.
@@ -302,6 +303,7 @@ def __init__(
         node_selector: dict | None = None,
         image_pull_secrets: list[k8s.V1LocalObjectReference] | None = None,
         service_account_name: str | None = None,
+        automount_service_account_token: bool = True,
         hostnetwork: bool = False,
         host_aliases: list[k8s.V1HostAlias] | None = None,
         tolerations: list[k8s.V1Toleration] | None = None,
@@ -380,6 +382,7 @@ def __init__(
         self.config_file = config_file
         self.image_pull_secrets = convert_image_pull_secrets(image_pull_secrets) if image_pull_secrets else []
         self.service_account_name = service_account_name
+        self.automount_service_account_token = automount_service_account_token
         self.hostnetwork = hostnetwork
         self.host_aliases = host_aliases
         self.tolerations = (
@@ -1175,6 +1178,7 @@ def build_pod_request_obj(self, context: Context | None = None) -> k8s.V1Pod:
                 ],
                 image_pull_secrets=self.image_pull_secrets,
                 service_account_name=self.service_account_name,
+                automount_service_account_token=self.automount_service_account_token,
                 host_network=self.hostnetwork,
                 hostname=self.hostname,
                 subdomain=self.subdomain,

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

@@ -1296,6 +1296,19 @@ def test_node_selector(self):
         assert isinstance(pod.spec.node_selector, dict)
         assert sanitized_pod["spec"]["nodeSelector"] == node_selector
 
+    def test_automount_service_account_token(self):
+        automount_service_account_token = False
+
+        k = KubernetesPodOperator(
+            task_id="task",
+            automount_service_account_token=automount_service_account_token,
+        )
+
+        pod = k.build_pod_request_obj(create_context(k))
+        sanitized_pod = self.sanitize_for_serialization(pod)
+        assert isinstance(pod.spec.automount_service_account_token, bool)
+        assert sanitized_pod["spec"]["automountServiceAccountToken"] == automount_service_account_token
+
     @pytest.mark.parametrize("do_xcom_push", [True, False])
     @patch(f"{POD_MANAGER_CLASS}.extract_xcom")
     @patch(f"{POD_MANAGER_CLASS}.await_xcom_sidecar_container_start")