feat(storage): add bucket IP filtering samples and tests

storage/samples/snippets/bucket_ip_filter_test.py

@@ -0,0 +1,89 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import uuid
+
+from google.api_core import exceptions
+from google.cloud import storage
+import pytest
+
+import storage_create_bucket_ip_filtering
+import storage_delete_ip_filtering_rules
+import storage_disable_ip_filtering
+import storage_enable_ip_filtering
+import storage_get_ip_filtering
+import storage_list_buckets_ip_filtering
+
+
+@pytest.fixture
+def test_bucket():
+    storage_client = storage.Client()
+    bucket_name = f"ipfilter-test-{uuid.uuid4().hex[:10]}"
+    yield bucket_name
+    try:
+        bucket = storage_client.get_bucket(bucket_name)
+        bucket.delete(force=True)
+    except Exception:
+        pass
+
+
+def test_ip_filter_lifecycle(test_bucket, capsys):
+    public_range = "0.0.0.0/0"
+    project_id = storage.Client().project
+    vpc_network = f"projects/{project_id}/global/networks/default"
+    vpc_range = "10.0.0.0/24"
+
+    # 1. Create with IP filtering
+    try:
+        created = storage_create_bucket_ip_filtering.create_bucket_ip_filtering(
+            test_bucket, public_range
+        )
+    except (exceptions.Forbidden, exceptions.BadRequest) as e:
+        pytest.skip(f"Skipping test due to insufficient permissions on project: {e}")
+
+    assert created.ip_filter is not None
+    assert created.ip_filter.mode == "Disabled"
+
+    # 2. Enable IP filtering
+    enabled = storage_enable_ip_filtering.enable_ip_filtering(
+        test_bucket, public_range, vpc_network, vpc_range
+    )
+    assert enabled.ip_filter.mode == "Enabled"
+
+    # 3. Get IP filtering
+    fetched = storage_get_ip_filtering.get_ip_filtering(test_bucket)
+    assert fetched.mode == "Enabled"
+
+    # 4. Disable IP filtering
+    disabled = storage_disable_ip_filtering.disable_ip_filtering(test_bucket)
+    assert disabled.ip_filter.mode == "Disabled"
+
+    # 5. Delete IP filtering rules
+    modified = storage_delete_ip_filtering_rules.delete_ip_filtering_rules(
+        test_bucket,
+        public_range_to_delete=public_range,
+        vpc_network_to_delete=vpc_network,
+    )
+    assert (
+        public_range
+        not in modified.ip_filter.public_network_source.allowed_ip_cidr_ranges
+    )
+    assert not any(
+        v.network == vpc_network for v in modified.ip_filter.vpc_network_sources
+    )
+
+    # 6. List buckets with IP filtering
+    storage_list_buckets_ip_filtering.list_buckets_ip_filtering()
+    out, _ = capsys.readouterr()
+    assert test_bucket in out

storage/samples/snippets/storage_create_bucket_ip_filtering.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+# [START storage_create_bucket_ip_filtering]
+from google.cloud import storage
+from google.cloud.storage.ip_filter import IPFilter, PublicNetworkSource
+
+
+def create_bucket_ip_filtering(
+    bucket_name: str, public_cidr_range: str
+) -> storage.Bucket:
+    """Creates a new bucket with initial IP filtering rules pre-configured."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+    # public_cidr_range = "192.0.2.0/24"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+
+    ip_filter = IPFilter()
+    ip_filter.mode = "Disabled"
+    ip_filter.public_network_source = PublicNetworkSource(
+        allowed_ip_cidr_ranges=[public_cidr_range]
+    )
+    ip_filter.allow_all_service_agent_access = True
+
+    bucket.ip_filter = ip_filter
+    new_bucket = storage_client.create_bucket(bucket)
+
+    print(
+        f"Created bucket {new_bucket.name} with IP filtering mode: {new_bucket.ip_filter.mode}"
+    )
+    return new_bucket
+
+
+# [END storage_create_bucket_ip_filtering]
+
+if __name__ == "__main__":
+    if len(sys.argv) < 3:
+        print(
+            "Usage: python storage_create_bucket_ip_filtering.py <bucket_name> <public_cidr_range>"
+        )
+        sys.exit(1)
+    create_bucket_ip_filtering(bucket_name=sys.argv[1], public_cidr_range=sys.argv[2])

storage/samples/snippets/storage_delete_ip_filtering_rules.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+# [START storage_delete_ip_filtering_rules]
+from typing import Optional
+
+from google.cloud import storage
+
+
+def delete_ip_filtering_rules(
+    bucket_name: str,
+    public_range_to_delete: Optional[str] = None,
+    vpc_network_to_delete: Optional[str] = None,
+) -> storage.Bucket:
+    """Selectively removes specific public CIDR ranges or VPC network sources."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+    # public_range_to_delete = "192.0.2.0/24"
+    # vpc_network_to_delete = "projects/my-project/global/networks/my-network"
+
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(bucket_name)
+
+    ip_filter = bucket.ip_filter
+    if not ip_filter:
+        print(f"Bucket {bucket_name} has no IP Filter configuration.")
+        return bucket
+
+    modified = False
+    if public_range_to_delete and ip_filter.public_network_source:
+        ranges = ip_filter.public_network_source.allowed_ip_cidr_ranges
+        if ranges and public_range_to_delete in ranges:
+            ranges.remove(public_range_to_delete)
+            modified = True
+
+    if vpc_network_to_delete and ip_filter.vpc_network_sources:
+        initial_len = len(ip_filter.vpc_network_sources)
+        ip_filter.vpc_network_sources = [
+            v
+            for v in ip_filter.vpc_network_sources
+            if v.network != vpc_network_to_delete
+        ]
+        if len(ip_filter.vpc_network_sources) != initial_len:
+            modified = True
+
+    if modified:
+        # Re-assign to the bucket property to force google-cloud-storage to register
+        # the nested changes for the patch() call.
+        bucket.ip_filter = ip_filter
+        bucket.patch()
+        print(f"Updated IP filtering rules for bucket {bucket_name}.")
+    else:
+        print("No changes were made to the bucket's IP filters.")
+
+    return bucket
+
+
+# [END storage_delete_ip_filtering_rules]
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print(
+            "Usage: python storage_delete_ip_filtering_rules.py <bucket_name> [public_range_to_delete] [vpc_network_to_delete]"
+        )
+        sys.exit(1)
+    delete_ip_filtering_rules(
+        bucket_name=sys.argv[1],
+        public_range_to_delete=sys.argv[2] if len(sys.argv) > 2 else None,
+        vpc_network_to_delete=sys.argv[3] if len(sys.argv) > 3 else None,
+    )

storage/samples/snippets/storage_disable_ip_filtering.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+# [START storage_disable_ip_filtering]
+from google.cloud import storage
+
+
+def disable_ip_filtering(bucket_name: str) -> storage.Bucket:
+    """Disables IP filtering on a bucket without deleting existing rules."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(bucket_name)
+
+    ip_filter = bucket.ip_filter
+    if not ip_filter:
+        print(f"No IP filter configuration found for bucket {bucket_name}.")
+        return bucket
+
+    ip_filter.mode = "Disabled"
+    # Re-assign to the bucket property to force google-cloud-storage to register
+    # the nested changes for the patch() call.
+    bucket.ip_filter = ip_filter
+    bucket.patch()
+    print(f"IP filtering disabled for bucket {bucket_name}.")
+    return bucket
+
+
+# [END storage_disable_ip_filtering]
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: python storage_disable_ip_filtering.py <bucket_name>")
+        sys.exit(1)
+    disable_ip_filtering(bucket_name=sys.argv[1])

storage/samples/snippets/storage_enable_ip_filtering.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python
+
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+
+# [START storage_enable_ip_filtering]
+from google.cloud import storage
+from google.cloud.storage.ip_filter import (
+    IPFilter,
+    PublicNetworkSource,
+    VpcNetworkSource,
+)
+
+
+def enable_ip_filtering(
+    bucket_name: str, public_range: str, vpc_network: str, vpc_range: str
+) -> storage.Bucket:
+    """Enables and configures IP filtering rules on an existing bucket."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+    # public_range = "192.0.2.0/24"
+    # vpc_network = "projects/my-project/global/networks/my-network"
+    # vpc_range = "10.0.0.0/24"
+
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(bucket_name)
+
+    ip_filter = bucket.ip_filter or IPFilter()
+    ip_filter.mode = "Enabled"
+    ip_filter.allow_all_service_agent_access = True
+    ip_filter.allow_cross_org_vpcs = True
+
+    if ip_filter.public_network_source is None:
+        ip_filter.public_network_source = PublicNetworkSource(allowed_ip_cidr_ranges=[])
+    elif ip_filter.public_network_source.allowed_ip_cidr_ranges is None:
+        ip_filter.public_network_source.allowed_ip_cidr_ranges = []
+
+    if (
+        public_range
+        and public_range not in ip_filter.public_network_source.allowed_ip_cidr_ranges
+    ):
+        ip_filter.public_network_source.allowed_ip_cidr_ranges.append(public_range)
+
+    if ip_filter.vpc_network_sources is None:
+        ip_filter.vpc_network_sources = []
+
+    existing_vpc = next(
+        (v for v in ip_filter.vpc_network_sources if v.network == vpc_network),
+        None,
+    )
+    if existing_vpc:
+        if existing_vpc.allowed_ip_cidr_ranges is None:
+            existing_vpc.allowed_ip_cidr_ranges = []
+        if vpc_range and vpc_range not in existing_vpc.allowed_ip_cidr_ranges:
+            existing_vpc.allowed_ip_cidr_ranges.append(vpc_range)
+    elif vpc_network:
+        vpc_sources = [vpc_range] if vpc_range else []
+        ip_filter.vpc_network_sources.append(
+            VpcNetworkSource(network=vpc_network, allowed_ip_cidr_ranges=vpc_sources)
+        )
+
+    bucket.ip_filter = ip_filter
+    bucket.patch()
+
+    print(f"Enabled IP filtering for bucket {bucket.name}.")
+    return bucket
+
+
+# [END storage_enable_ip_filtering]
+
+if __name__ == "__main__":
+    if len(sys.argv) < 5:
+        print(
+            "Usage: python storage_enable_ip_filtering.py <bucket_name> <public_range> <vpc_network> <vpc_range>"
+        )
+        sys.exit(1)
+    enable_ip_filtering(
+        bucket_name=sys.argv[1],
+        public_range=sys.argv[2],
+        vpc_network=sys.argv[3],
+        vpc_range=sys.argv[4],
+    )