@@ -62,6 +62,8 @@
 
 _TOP_GOOGLE_CONSTRUCTOR_METHOD_TAG = "top_google_constructor_method"
 
+_ENFORCE_BUCKET_OWNERSHIP_ENV_VAR = "GOOGLE_CLOUD_AIPLATFORM_ENFORCE_BUCKET_OWNERSHIP"
+
 
 class _Product(enum.Enum):
     """Notebook product types."""
@@ -133,6 +135,7 @@ def __init__(self):
         self._request_metadata = None
         self._resource_type = None
         self._async_rest_credentials = None
+        self._enforce_bucket_ownership = None
 
     def init(
         self,
@@ -153,6 +156,7 @@ def init(
         api_key: Optional[str] = None,
         api_transport: Optional[str] = None,
         request_metadata: Optional[Sequence[Tuple[str, str]]] = None,
+        enforce_bucket_ownership: Optional[bool] = None,
     ):
         """Updates common initialization parameters with provided options.
 
@@ -220,6 +224,13 @@ def init(
                 beta state (preview).
             request_metadata:
                 Optional. Additional gRPC metadata to send with every client request.
+            enforce_bucket_ownership (bool):
+                Optional. If True, raise an error when an existing GCS bucket
+                that the SDK writes pipeline artifacts to is not owned by the
+                project, or when its owner cannot be verified. If False or not
+                set, only a warning is shown. Can also be enabled with the
+                GOOGLE_CLOUD_AIPLATFORM_ENFORCE_BUCKET_OWNERSHIP=true
+                environment variable.
         Raises:
             ValueError:
                 If experiment_description is provided but experiment is not.
@@ -287,6 +298,8 @@ def init(
             self._request_metadata = request_metadata
         if api_key is not None:
             self._api_key = api_key
+        if enforce_bucket_ownership is not None:
+            self._enforce_bucket_ownership = enforce_bucket_ownership
         self._resource_type = None
 
         # Finally, perform secondary state updates
@@ -388,6 +401,15 @@ def staging_bucket(self) -> Optional[str]:
         """Default staging bucket, if provided."""
         return self._staging_bucket
 
+    @property
+    def enforce_bucket_ownership(self) -> bool:
+        """Whether to raise an error for a GCS bucket owned by another project."""
+        if self._enforce_bucket_ownership is not None:
+            return self._enforce_bucket_ownership
+        return os.environ.get(
+            _ENFORCE_BUCKET_OWNERSHIP_ENV_VAR, ""
+        ).strip().lower() in ("1", "true", "yes")
+
     @property
     def credentials(self) -> Optional[auth_credentials.Credentials]:
         """Default credentials."""
 
@@ -468,6 +468,9 @@ def submit(
                 location=self.location,
                 credentials=self.credentials,
             )
+        except gcs_utils.BucketOwnershipError:
+            # Raised only in strict mode: aiplatform.init(enforce_bucket_ownership=True).
+            raise
         except:  # noqa: E722
             _LOGGER.exception(
                 "Error when trying to get or create a GCS bucket for the pipeline output artifacts"
 
@@ -25,7 +25,7 @@
 import os
 import pathlib
 import tempfile
-from typing import Optional, TYPE_CHECKING
+from typing import Optional, Tuple, TYPE_CHECKING
 import warnings
 
 from google.auth import credentials as auth_credentials
@@ -63,6 +63,45 @@
 _DEFAULT_STAGING_BUCKET_SALT = str(uuid.uuid4())
 
 
+class BucketOwnershipError(ValueError):
+    """Raised when a GCS bucket does not belong to the expected project."""
+
+
+def _get_bucket_project_numbers(
+    bucket: storage.Bucket,
+    expected_project: str,
+    client: storage.Client,
+) -> Tuple[Optional[str], Optional[str]]:
+    """Gets the project number of a GCS bucket and of the expected project.
+
+    Args:
+        bucket: The GCS bucket to check.
+        expected_project: The project ID that should own the bucket.
+        client: Storage client instance.
+
+    Returns:
+        A tuple of the bucket's project number and the expected project's
+        number. Each value is None if it could not be retrieved.
+    """
+    bucket_project_number = None
+    expected_project_number = None
+    try:
+        bucket.reload(client=client)
+        if bucket.project_number is not None:
+            bucket_project_number = str(bucket.project_number)
+    except Exception:
+        pass
+    try:
+        expected_project_number = str(
+            resource_manager_utils.get_project_number(
+                expected_project, credentials=getattr(client, "_credentials", None)
+            )
+        )
+    except Exception:
+        pass
+    return bucket_project_number, expected_project_number
+
+
 def _verify_bucket_ownership(
     bucket: storage.Bucket,
     expected_project: str,
@@ -81,15 +120,70 @@ def _verify_bucket_ownership(
     Returns:
         True if the bucket belongs to the expected project, False otherwise.
     """
-    try:
-        bucket.reload(client=client)
-        bucket_project_number = str(bucket.project_number)
-        expected_project_number = str(
-            resource_manager_utils.get_project_number(expected_project)
+    bucket_project_number, expected_project_number = _get_bucket_project_numbers(
+        bucket, expected_project, client
+    )
+    return (
+        bucket_project_number is not None
+        and bucket_project_number == expected_project_number
+    )
+
+
+def _check_pipeline_bucket_ownership(
+    bucket: storage.Bucket,
+    expected_project: str,
+    client: storage.Client,
+    is_sdk_generated: bool = False,
+) -> None:
+    """Warns, or raises, if a bucket is not owned by the project.
+
+    Args:
+        bucket: The existing GCS bucket for pipeline artifacts.
+        expected_project: The project ID that should own the bucket.
+        client: Storage client instance.
+        is_sdk_generated: Whether the SDK generated the bucket name. A
+            generated name can be predictable, so a mismatch always raises.
+
+    Raises:
+        BucketOwnershipError: If the bucket is not owned by `expected_project`,
+            or its owner cannot be verified, and either `is_sdk_generated` is
+            True or `aiplatform.init(enforce_bucket_ownership=True)` is set.
+    """
+    bucket_project_number, expected_project_number = _get_bucket_project_numbers(
+        bucket, expected_project, client
+    )
+    if bucket_project_number is not None and (
+        bucket_project_number == expected_project_number
+    ):
+        return
+
+    if bucket_project_number is None or expected_project_number is None:
+        message = (
+            f'Could not verify that GCS bucket "{bucket.name}" is owned by '
+            f'project "{expected_project}".'
         )
-        return bucket_project_number == expected_project_number
-    except Exception:
-        return False
+    else:
+        message = (
+            f'GCS bucket "{bucket.name}" is owned by project number '
+            f'"{bucket_project_number}", not by project "{expected_project}" '
+            f'(project number "{expected_project_number}").'
+        )
+    message += (
+        " Pipeline artifacts will be written to this bucket. If you do not "
+        "own this bucket, other users may be able to read or modify the "
+        "artifacts. Use a bucket owned by your project, for example by "
+        "passing pipeline_root or by setting "
+        "aiplatform.init(staging_bucket='gs://your-bucket')."
+    )
+
+    if is_sdk_generated or initializer.global_config.enforce_bucket_ownership:
+        raise BucketOwnershipError(message)
+    warnings.warn(
+        f"{message} To raise an error instead, set "
+        "aiplatform.init(enforce_bucket_ownership=True).",
+        UserWarning,
+        stacklevel=2,
+    )
 
 
 def blob_from_uri(uri: str, client: storage.Client) -> storage.Blob:
@@ -335,12 +429,12 @@ def create_gcs_bucket_for_pipeline_artifacts_if_it_does_not_exist(
     service_account = service_account or initializer.global_config.service_account
     credentials = credentials or initializer.global_config.credentials
 
+    default_output_artifacts_gcs_dir = generate_gcs_directory_for_pipeline_artifacts(
+        project=project,
+        location=location,
+    )
     output_artifacts_gcs_dir = (
-        output_artifacts_gcs_dir
-        or generate_gcs_directory_for_pipeline_artifacts(
-            project=project,
-            location=location,
-        )
+        output_artifacts_gcs_dir or default_output_artifacts_gcs_dir
     )
 
     # Creating the bucket if needed
@@ -377,13 +471,21 @@ def create_gcs_bucket_for_pipeline_artifacts_if_it_does_not_exist(
             f"serviceAccount:{service_account}"
         )
         pipelines_bucket.set_iam_policy(bucket_iam_policy)
-    elif not _verify_bucket_ownership(pipelines_bucket, project, storage_client):
-        raise ValueError(
-            f'Output artifacts bucket "{pipelines_bucket.name}" exists but does '
-            f'not belong to project "{project}". This may indicate a '
-            f"bucket squatting attack. Please provide an explicit "
-            f"output_artifacts_gcs_dir parameter or configure one via "
-            f"aiplatform.init(output_artifacts_gcs_dir='gs://your-bucket')."
+    else:
+        # PipelineJob generates its default pipeline_root before calling this
+        # function, so compare bucket names instead of checking for None.
+        is_sdk_generated = (
+            pipelines_bucket.name
+            == bucket_from_uri(
+                uri=default_output_artifacts_gcs_dir,
+                client=storage_client,
+            ).name
+        )
+        _check_pipeline_bucket_ownership(
+            pipelines_bucket,
+            project,
+            storage_client,
+            is_sdk_generated=is_sdk_generated,
         )
     return output_artifacts_gcs_dir
 
 
@@ -138,6 +138,24 @@ def test_init_service_account_sets_service_account(self):
         initializer.global_config.init(service_account=_TEST_SERVICE_ACCOUNT)
         assert initializer.global_config.service_account == _TEST_SERVICE_ACCOUNT
 
+    def test_enforce_bucket_ownership_defaults_to_false(self):
+        with patch.dict(
+            os.environ, {initializer._ENFORCE_BUCKET_OWNERSHIP_ENV_VAR: ""}
+        ):
+            assert initializer.global_config.enforce_bucket_ownership is False
+
+    def test_init_enforce_bucket_ownership_sets_enforce_bucket_ownership(self):
+        initializer.global_config.init(enforce_bucket_ownership=True)
+        assert initializer.global_config.enforce_bucket_ownership is True
+
+    def test_enforce_bucket_ownership_from_env_var(self):
+        with patch.dict(
+            os.environ, {initializer._ENFORCE_BUCKET_OWNERSHIP_ENV_VAR: "true"}
+        ):
+            assert initializer.global_config.enforce_bucket_ownership is True
+            initializer.global_config.init(enforce_bucket_ownership=False)
+            assert initializer.global_config.enforce_bucket_ownership is False
+
     @patch.object(_experiment_tracker, "set_experiment")
     def test_init_experiment_sets_experiment(self, set_experiment_mock):
         initializer.global_config.init(experiment=_TEST_EXPERIMENT)
 
@@ -1627,6 +1627,42 @@ def test_submit_call_gcs_utils_get_or_create_with_correct_arguments(
             credentials=_TEST_CREDENTIALS,
         )
 
+    @pytest.mark.parametrize(
+        "job_spec",
+        [_TEST_PIPELINE_SPEC_JSON, _TEST_PIPELINE_SPEC_YAML, _TEST_PIPELINE_JOB],
+    )
+    def test_submit_raises_when_pipeline_bucket_belongs_to_another_project(
+        self,
+        mock_pipeline_service_create,
+        mock_pipeline_service_get,
+        job_spec,
+        mock_load_yaml_and_json,
+    ):
+        job = pipeline_jobs.PipelineJob(
+            display_name=_TEST_PIPELINE_JOB_DISPLAY_NAME,
+            template_path=_TEST_TEMPLATE_PATH,
+            job_id=_TEST_PIPELINE_JOB_ID,
+            parameter_values=_TEST_PIPELINE_PARAMETER_VALUES,
+            enable_caching=True,
+            project=_TEST_PROJECT,
+            pipeline_root=_TEST_GCS_OUTPUT_DIRECTORY,
+            location=_TEST_LOCATION,
+            credentials=_TEST_CREDENTIALS,
+        )
+
+        with mock.patch.object(
+            gcs_utils,
+            "create_gcs_bucket_for_pipeline_artifacts_if_it_does_not_exist",
+            side_effect=gcs_utils.BucketOwnershipError("not owned"),
+        ):
+            with pytest.raises(gcs_utils.BucketOwnershipError):
+                job.submit(
+                    service_account=_TEST_SERVICE_ACCOUNT,
+                    network=_TEST_NETWORK,
+                )
+
+        mock_pipeline_service_create.assert_not_called()
+
     @pytest.mark.parametrize(
         "job_spec",
         [_TEST_PIPELINE_SPEC_JSON, _TEST_PIPELINE_SPEC_YAML, _TEST_PIPELINE_JOB],