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

storage/samples/snippets/bucket_ip_filter_test.py

@@ -0,0 +1,90 @@
+# 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.
+
+from typing import Generator
+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() -> Generator[str, None, None]:
+    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: str, capsys: pytest.CaptureFixture) -> None:
+    public_range = "0.0.0.0/0"
+    project_id = storage.Client().project
+    vpc_network = f"projects/{project_id}/global/networks/default"
+
+    # 1. Create with IP filtering
+    try:
+        created = storage_create_bucket_ip_filtering.create_bucket_ip_filtering(
+            test_bucket, public_range
+        )
+        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)
+        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 (
+            modified.ip_filter.public_network_source is None
+            or 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
+    except (exceptions.Forbidden, exceptions.BadRequest) as e:
+        pytest.skip(
+            "Skipping test due to insufficient permissions or IP filter"
+            f" network restriction: {e}"
+        )

storage/samples/snippets/requirements.txt

@@ -1,5 +1,5 @@
 google-cloud-pubsub==2.29.0
-google-cloud-storage==3.1.0
+google-cloud-storage==3.14.1
 pandas===1.3.5; python_version == '3.7'
 pandas===2.0.3; python_version == '3.8'
 pandas==2.2.3; python_version >= '3.9'

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,76 @@
+#!/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"No IP filter configuration found for bucket {bucket_name}.")
+        return bucket
+
+    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)
+
+    if vpc_network_to_delete and 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
+        ]
+
+    bucket.ip_filter = ip_filter
+    bucket.patch()
+
+    print(f"Updated IP filtering rules for bucket {bucket_name}.")
+    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,52 @@
+#!/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
+
+
+def enable_ip_filtering(bucket_name: str) -> storage.Bucket:
+    """Enables IP filtering on an existing bucket."""
+    # 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 = "Enabled"
+    # 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"Enabled IP filtering for bucket {bucket.name}.")
+    return bucket
+
+
+# [END storage_enable_ip_filtering]
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: python storage_enable_ip_filtering.py <bucket_name>")
+        sys.exit(1)
+    enable_ip_filtering(bucket_name=sys.argv[1])

storage/samples/snippets/storage_get_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_get_ip_filtering]
+from typing import Optional
+
+from google.cloud import storage
+from google.cloud.storage.ip_filter import IPFilter
+
+
+def get_ip_filtering(bucket_name: str) -> Optional[IPFilter]:
+    """Retrieves and prints the IP filtering configuration of a bucket."""
+    # 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"Bucket {bucket_name} has no IP Filter configured.")
+        return None
+
+    print(f"IP Filter mode: {ip_filter.mode}")
+    print(f"IP Filter configuration: {ip_filter._to_api_resource()}")
+
+    return ip_filter
+
+
+# [END storage_get_ip_filtering]
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: python storage_get_ip_filtering.py <bucket_name>")
+        sys.exit(1)
+    get_ip_filtering(bucket_name=sys.argv[1])