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