diff --git a/storage/samples/snippets/bucket_ip_filter_test.py b/storage/samples/snippets/bucket_ip_filter_test.py new file mode 100644 index 00000000000..61ebc5c4ecd --- /dev/null +++ b/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}" + ) diff --git a/storage/samples/snippets/requirements.txt b/storage/samples/snippets/requirements.txt index 751f8cfbe53..51243745ba0 100644 --- a/storage/samples/snippets/requirements.txt +++ b/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' diff --git a/storage/samples/snippets/storage_create_bucket_ip_filtering.py b/storage/samples/snippets/storage_create_bucket_ip_filtering.py new file mode 100644 index 00000000000..447597458b3 --- /dev/null +++ b/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 " + ) + sys.exit(1) + create_bucket_ip_filtering(bucket_name=sys.argv[1], public_cidr_range=sys.argv[2]) diff --git a/storage/samples/snippets/storage_delete_ip_filtering_rules.py b/storage/samples/snippets/storage_delete_ip_filtering_rules.py new file mode 100644 index 00000000000..7514bf3be06 --- /dev/null +++ b/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 " + "[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, + ) diff --git a/storage/samples/snippets/storage_disable_ip_filtering.py b/storage/samples/snippets/storage_disable_ip_filtering.py new file mode 100644 index 00000000000..351d92fa474 --- /dev/null +++ b/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 ") + sys.exit(1) + disable_ip_filtering(bucket_name=sys.argv[1]) diff --git a/storage/samples/snippets/storage_enable_ip_filtering.py b/storage/samples/snippets/storage_enable_ip_filtering.py new file mode 100644 index 00000000000..11087607a86 --- /dev/null +++ b/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 ") + sys.exit(1) + enable_ip_filtering(bucket_name=sys.argv[1]) diff --git a/storage/samples/snippets/storage_get_ip_filtering.py b/storage/samples/snippets/storage_get_ip_filtering.py new file mode 100644 index 00000000000..beb2463b09a --- /dev/null +++ b/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 ") + sys.exit(1) + get_ip_filtering(bucket_name=sys.argv[1]) diff --git a/storage/samples/snippets/storage_list_buckets_ip_filtering.py b/storage/samples/snippets/storage_list_buckets_ip_filtering.py new file mode 100644 index 00000000000..16279afb83c --- /dev/null +++ b/storage/samples/snippets/storage_list_buckets_ip_filtering.py @@ -0,0 +1,42 @@ +#!/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. + +# [START storage_list_buckets_ip_filtering] +from typing import List + +from google.cloud import storage + + +def list_buckets_ip_filtering() -> List[storage.Bucket]: + """Lists all buckets in the project with their IP filtering status.""" + storage_client = storage.Client() + buckets = list(storage_client.list_buckets()) + + for bucket in buckets: + status = ( + bucket.ip_filter.mode + if bucket.ip_filter and bucket.ip_filter.mode + else "Not Configured" + ) + print(f"Bucket: {bucket.name}, IP Filter Mode: {status}") + + return buckets + + +# [END storage_list_buckets_ip_filtering] + +if __name__ == "__main__": + list_buckets_ip_filtering()