From eb566801213dd0c06a541baf707db72d487e18cb Mon Sep 17 00:00:00 2001
From: Dr Alex Mitre <30060514+mitre88@users.noreply.github.com>
Date: Mon, 27 Jul 2026 09:31:49 -0600
Subject: [PATCH] Validate storage transfer job body after template rendering
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The operator's body is a template field, but it was deep-copied and
validated in __init__, so a fully templated body skipped validation
entirely — the checks ran against the Jinja expression instead of the
rendered dict, silently bypassing the AWS-credential restriction. Part
of the burn-down tracked in #70296.
---
 .../operators/cloud_storage_transfer_service.py      |  6 +++---
 .../operators/test_cloud_storage_transfer_service.py | 12 ++++++++++++
 .../ci/prek/validate_operators_init_exemptions.txt   |  1 -
 3 files changed, 15 insertions(+), 4 deletions(-)

diff --git a/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py b/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py
index 60682bfd0add9..1de3adb776452 100644
--- a/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py
+++ b/providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py
@@ -257,19 +257,19 @@ def __init__(
     ) -> None:
         super().__init__(**kwargs)
         self.body = body
-        if isinstance(self.body, dict):
-            self.body = deepcopy(body)
         self.aws_conn_id = aws_conn_id
         self.gcp_conn_id = gcp_conn_id
         self.api_version = api_version
         self.project_id = project_id
         self.google_impersonation_chain = google_impersonation_chain
-        self._validate_inputs()
 
     def _validate_inputs(self) -> None:
         TransferJobValidator(body=self.body).validate_body()
 
     def execute(self, context: Context) -> dict:
+        if isinstance(self.body, dict):
+            self.body = deepcopy(self.body)
+        self._validate_inputs()
         TransferJobPreprocessor(body=self.body, aws_conn_id=self.aws_conn_id).process_body()
         hook = CloudDataTransferServiceHook(
             api_version=self.api_version,
diff --git a/providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py b/providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py
index 248a5e3a2d443..50c2987ebae09 100644
--- a/providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py
+++ b/providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py
@@ -283,6 +283,18 @@ def test_verify_success(self, body):
 
 
 class TestGcpStorageTransferJobCreateOperator:
+    @mock.patch(
+        "airflow.providers.google.cloud.operators.cloud_storage_transfer_service.CloudDataTransferServiceHook"
+    )
+    def test_templated_body_validated_at_execute_time(self, mock_hook):
+        op = CloudDataTransferServiceCreateJobOperator(body="{{ var.value.body }}", task_id=TASK_ID)
+        # Template rendering replaces the Jinja expression with the resolved value before execute.
+        op.body = {"transferSpec": {"awsS3DataSource": {"awsAccessKey": TEST_AWS_ACCESS_KEY}}}
+
+        with pytest.raises(AirflowException, match="AWS credentials detected inside the body parameter"):
+            op.execute(context=mock.MagicMock())
+        mock_hook.return_value.create_transfer_job.assert_not_called()
+
     @mock.patch(
         "airflow.providers.google.cloud.operators.cloud_storage_transfer_service.CloudDataTransferServiceHook"
     )
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt
index 86ae9cd475947..e4e6af93ad8ec 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -19,7 +19,6 @@ providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
 providers/google/src/airflow/providers/google/cloud/operators/bigquery.py::BigQueryInsertJobOperator
 providers/google/src/airflow/providers/google/cloud/operators/cloud_batch.py::CloudBatchSubmitJobOperator
 providers/google/src/airflow/providers/google/cloud/operators/cloud_build.py::CloudBuildCreateBuildOperator
-providers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py::CloudDataTransferServiceCreateJobOperator
 providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocCreateClusterOperator
 providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocSubmitJobOperator
 providers/google/src/airflow/providers/google/cloud/operators/functions.py::CloudFunctionDeployFunctionOperator