Skip to content

Validate storage transfer job body after template rendering #70529

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mitre88 wants to merge 1 commit into
base: main
Choose a base branch
from
+15 −4
Open

Validate storage transfer job body after template rendering #70529

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions ...ers/google/src/airflow/providers/google/cloud/operators/cloud_storage_transfer_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

@potiuk potiuk Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The deepcopy is load-bearing and its purpose isn't obvious from here — TransferJobPreprocessor.process_body() calls _inject_aws_credentials() and _reformat_schedule(), which mutate the body in place. Without the copy, the operator would write AWS credentials into the caller's own dict.

Two things worth doing:

  1. A one-line comment saying why the copy exists. The original had the copy sitting next to the assignment in __init__ where the reader could at least see it was defensive; now it's a bare reassignment in execute that looks removable. This is precisely the kind of line someone "simplifies" away later.
  2. A test for it. The new test covers the validation-raises path, but nothing asserts the behaviour the deepcopy is there for. Something like: build the operator with a dict containing an awsAccessKey, run a successful execute, and assert the dict you passed in is unchanged. Without that, the copy could be dropped in a future refactor and the whole suite would stay green while the operator started mutating user data.

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

self._validate_inputs()
TransferJobPreprocessor(body=self.body, aws_conn_id=self.aws_conn_id).process_body()
hook = CloudDataTransferServiceHook(
api_version=self.api_version,
Expand Down
12 changes: 12 additions & 0 deletions providers/google/tests/unit/google/cloud/operators/test_cloud_storage_transfer_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down
1 change: 0 additions & 1 deletion scripts/ci/prek/validate_operators_init_exemptions.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down