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..beb9551d510 --- /dev/null +++ b/storage/samples/snippets/bucket_ip_filter_test.py @@ -0,0 +1,279 @@ +# 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 unittest.mock import MagicMock, patch +import uuid + +from google.api_core import exceptions +from google.cloud import storage +from google.cloud.storage.ip_filter import ( + IPFilter, + PublicNetworkSource, + VpcNetworkSource, +) +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 = "192.0.2.0/24" + 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. Delete IP filtering rules + modified = storage_delete_ip_filtering_rules.delete_ip_filtering_rules( + test_bucket, public_range_to_delete=public_range + ) + assert ( + public_range + not in modified.ip_filter.public_network_source.allowed_ip_cidr_ranges + ) + + # 5. Disable IP filtering + disabled = storage_disable_ip_filtering.disable_ip_filtering(test_bucket) + assert disabled.ip_filter.mode == "Disabled" + + # 6. List buckets with IP filtering + storage_list_buckets_ip_filtering.list_buckets_ip_filtering() + out, _ = capsys.readouterr() + assert test_bucket in out + + +def test_create_bucket_ip_filtering_unit(): + with patch("google.cloud.storage.Client") as mock_client_cls: + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_bucket = MagicMock() + mock_client.bucket.return_value = mock_bucket + mock_created = MagicMock() + mock_created.name = "test-bucket" + mock_ip_filter = IPFilter() + mock_ip_filter.mode = "Disabled" + mock_created.ip_filter = mock_ip_filter + mock_client.create_bucket.return_value = mock_created + + result = storage_create_bucket_ip_filtering.create_bucket_ip_filtering( + "test-bucket", "192.0.2.0/24" + ) + assert result == mock_created + mock_client.create_bucket.assert_called_once_with(mock_bucket) + assert mock_bucket.ip_filter.mode == "Disabled" + assert ( + "192.0.2.0/24" + in mock_bucket.ip_filter.public_network_source.allowed_ip_cidr_ranges + ) + + +def test_enable_ip_filtering_unit(): + with patch("google.cloud.storage.Client") as mock_client_cls: + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_bucket = MagicMock() + mock_bucket.name = "test-bucket" + mock_bucket.ip_filter = None + mock_client.get_bucket.return_value = mock_bucket + + result = storage_enable_ip_filtering.enable_ip_filtering( + "test-bucket", + "192.0.2.0/24", + "projects/p/global/networks/n", + "10.0.0.0/24", + ) + assert result == mock_bucket + assert mock_bucket.ip_filter.mode == "Enabled" + assert ( + "192.0.2.0/24" + in mock_bucket.ip_filter.public_network_source.allowed_ip_cidr_ranges + ) + assert len(mock_bucket.ip_filter.vpc_network_sources) == 1 + assert ( + mock_bucket.ip_filter.vpc_network_sources[0].network + == "projects/p/global/networks/n" + ) + mock_bucket.patch.assert_called_once() + + # Enable again with same network and additional range + storage_enable_ip_filtering.enable_ip_filtering( + "test-bucket", + "192.0.2.0/24", + "projects/p/global/networks/n", + "10.0.1.0/24", + ) + assert ( + "10.0.1.0/24" + in mock_bucket.ip_filter.vpc_network_sources[0].allowed_ip_cidr_ranges + ) + + +def test_get_ip_filtering_unit(capsys): + with patch("google.cloud.storage.Client") as mock_client_cls: + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_bucket = MagicMock() + mock_bucket.ip_filter = None + mock_client.get_bucket.return_value = mock_bucket + + assert storage_get_ip_filtering.get_ip_filtering("test-bucket") is None + out, _ = capsys.readouterr() + assert "Bucket test-bucket has no IP Filter configured." in out + + ip_filter = IPFilter() + ip_filter.mode = "Enabled" + ip_filter.public_network_source = PublicNetworkSource( + allowed_ip_cidr_ranges=["192.0.2.0/24"] + ) + ip_filter.vpc_network_sources = [ + VpcNetworkSource( + network="projects/p/global/networks/n", + allowed_ip_cidr_ranges=["10.0.0.0/24"], + ) + ] + mock_bucket.ip_filter = ip_filter + + res = storage_get_ip_filtering.get_ip_filtering("test-bucket") + assert res == ip_filter + out, _ = capsys.readouterr() + assert "Mode: Enabled" in out + assert "Public CIDR Ranges: ['192.0.2.0/24']" in out + assert "VPC Network: projects/p/global/networks/n" in out + + +def test_delete_ip_filtering_rules_unit(capsys): + with patch("google.cloud.storage.Client") as mock_client_cls: + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_bucket = MagicMock() + mock_bucket.name = "test-bucket" + mock_bucket.ip_filter = None + mock_client.get_bucket.return_value = mock_bucket + + # No filter configured + storage_delete_ip_filtering_rules.delete_ip_filtering_rules("test-bucket") + out, _ = capsys.readouterr() + assert "Bucket test-bucket has no IP Filter configuration." in out + + ip_filter = IPFilter() + ip_filter.public_network_source = PublicNetworkSource( + allowed_ip_cidr_ranges=["192.0.2.0/24"] + ) + ip_filter.vpc_network_sources = [ + VpcNetworkSource( + network="projects/p/global/networks/n", + allowed_ip_cidr_ranges=["10.0.0.0/24"], + ) + ] + mock_bucket.ip_filter = ip_filter + + # Delete existing range and VPC + storage_delete_ip_filtering_rules.delete_ip_filtering_rules( + "test-bucket", + public_range_to_delete="192.0.2.0/24", + vpc_network_to_delete="projects/p/global/networks/n", + ) + assert ( + "192.0.2.0/24" + not in mock_bucket.ip_filter.public_network_source.allowed_ip_cidr_ranges + ) + assert len(mock_bucket.ip_filter.vpc_network_sources) == 0 + mock_bucket.patch.assert_called_once() + + # Delete non-existent + mock_bucket.patch.reset_mock() + storage_delete_ip_filtering_rules.delete_ip_filtering_rules( + "test-bucket", public_range_to_delete="non-existent" + ) + mock_bucket.patch.assert_not_called() + + +def test_disable_ip_filtering_unit(capsys): + with patch("google.cloud.storage.Client") as mock_client_cls: + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_bucket = MagicMock() + mock_bucket.name = "test-bucket" + mock_bucket.ip_filter = None + mock_client.get_bucket.return_value = mock_bucket + + # No IP filter + storage_disable_ip_filtering.disable_ip_filtering("test-bucket") + out, _ = capsys.readouterr() + assert "No IP filter configuration found" in out + + ip_filter = IPFilter() + ip_filter.mode = "Enabled" + mock_bucket.ip_filter = ip_filter + storage_disable_ip_filtering.disable_ip_filtering("test-bucket") + assert mock_bucket.ip_filter.mode == "Disabled" + mock_bucket.patch.assert_called_once() + + +def test_list_buckets_ip_filtering_unit(capsys): + with patch("google.cloud.storage.Client") as mock_client_cls: + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + b1 = MagicMock() + b1.name = "b1" + b1.ip_filter = None + b2 = MagicMock() + b2.name = "b2" + f2 = IPFilter() + f2.mode = "Enabled" + b2.ip_filter = f2 + mock_client.list_buckets.return_value = [b1, b2] + + storage_list_buckets_ip_filtering.list_buckets_ip_filtering() + mock_client.list_buckets.assert_called_once_with(projection="full") + out, _ = capsys.readouterr() + assert "Bucket: b1, IP Filter Mode: Not Configured" in out + assert "Bucket: b2, IP Filter Mode: Enabled" in out 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..82f51a13b42 --- /dev/null +++ b/storage/samples/snippets/storage_create_bucket_ip_filtering.py @@ -0,0 +1,62 @@ +#!/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, public_cidr_range="192.0.2.0/24"): + """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) < 2: + print( + "Usage: python storage_create_bucket_ip_filtering.py [public_cidr_range]" + ) + sys.exit(1) + if len(sys.argv) > 2: + create_bucket_ip_filtering( + bucket_name=sys.argv[1], public_cidr_range=sys.argv[2] + ) + else: + create_bucket_ip_filtering(bucket_name=sys.argv[1]) 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..7d86b711322 --- /dev/null +++ b/storage/samples/snippets/storage_delete_ip_filtering_rules.py @@ -0,0 +1,81 @@ +#!/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 google.cloud import storage + + +def delete_ip_filtering_rules( + bucket_name, public_range_to_delete=None, vpc_network_to_delete=None +): + """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) + + if not bucket.ip_filter: + print(f"Bucket {bucket_name} has no IP Filter configuration.") + return bucket + + modified = False + if public_range_to_delete and bucket.ip_filter.public_network_source: + ranges = bucket.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 bucket.ip_filter.vpc_network_sources: + initial_len = len(bucket.ip_filter.vpc_network_sources) + bucket.ip_filter.vpc_network_sources = [ + v + for v in bucket.ip_filter.vpc_network_sources + if v.network != vpc_network_to_delete + ] + if len(bucket.ip_filter.vpc_network_sources) != initial_len: + modified = True + + if modified: + # Re-assign to a local variable and back to the bucket property to force + # google-cloud-storage to register the nested changes for the patch() call. + ip_filter = bucket.ip_filter + 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 [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..8e26338eab8 --- /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): + """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) + + if not bucket.ip_filter: + print(f"No IP filter configuration found for bucket {bucket_name}.") + return bucket + + ip_filter = bucket.ip_filter + 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..b4b5453a7af --- /dev/null +++ b/storage/samples/snippets/storage_enable_ip_filtering.py @@ -0,0 +1,93 @@ +#!/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, public_range, vpc_network, vpc_range): + """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 " + ) + 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], + ) 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..19c151197bd --- /dev/null +++ b/storage/samples/snippets/storage_get_ip_filtering.py @@ -0,0 +1,61 @@ +#!/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 google.cloud import storage + + +def get_ip_filtering(bucket_name): + """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 Configuration for {bucket_name}:") + print(f"Mode: {ip_filter.mode}") + print(f"Allow All Service Agent Access: {ip_filter.allow_all_service_agent_access}") + print(f"Allow Cross Org VPCs: {ip_filter.allow_cross_org_vpcs}") + + if ip_filter.public_network_source: + print( + f"Public CIDR Ranges: {ip_filter.public_network_source.allowed_ip_cidr_ranges}" + ) + + if ip_filter.vpc_network_sources: + for vpc in ip_filter.vpc_network_sources: + print( + f"VPC Network: {vpc.network}, CIDR Ranges: {vpc.allowed_ip_cidr_ranges}" + ) + + 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..2dc9dd5dd18 --- /dev/null +++ b/storage/samples/snippets/storage_list_buckets_ip_filtering.py @@ -0,0 +1,34 @@ +#!/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 google.cloud import storage + + +def list_buckets_ip_filtering(): + """Lists all buckets in the project with their IP filtering status.""" + storage_client = storage.Client() + buckets = storage_client.list_buckets(projection="full") + + for bucket in buckets: + status = bucket.ip_filter.mode if bucket.ip_filter else "Not Configured" + print(f"Bucket: {bucket.name}, IP Filter Mode: {status}") + + +# [END storage_list_buckets_ip_filtering] + +if __name__ == "__main__": + list_buckets_ip_filtering()