Skip to content

Commit cefa2e8

Browse files
nidhiii-27nidhiii-27
authored
feat(storage): add bucket IP filtering samples and tests (#14594)
* feat(storage): add bucket IP filtering samples and tests Add Python code samples and tests demonstrating Cloud Storage Bucket IP filtering. Fixes: b/544985518 [Generated-by: AI] * fix(storage): resolve review comments for bucket IP filtering Address review feedback regarding self-assignment, defensive None checks, and CLI arg handling. [Generated-by: AI] * fix(storage): address reviewer comments on bucket IP filter samples [Generated-by: AI] * refactor(storage): align bucket IP filter samples with simplified Java pattern - Simplify enable_ip_filtering to take bucket_name only, verify ip_filter exists, set mode to Enabled, reassign, patch, print confirmation, and return bucket. - Simplify get_ip_filtering to display mode and print serialized ip_filter dictionary. - Streamline delete_ip_filtering_rules logic by eliminating unnecessary modification flags. - Remove projection="full" in list_buckets_ip_filtering and handle potential None mode. - Pin google-cloud-storage==3.14.1 in requirements.txt. - Update bucket_ip_filter_test.py to match simplified signatures, add None check on public_network_source, and handle IP filter network constraints. [Generated-by: AI] * fix(storage): address reviewer comments on bucket IP filter samples and tests [Generated-by: AI] * fix(storage): register IP filter mode change in test teardown [Generated-by: AI]
1 parent 383e0d3 commit cefa2e8

8 files changed

Lines changed: 502 additions & 1 deletion

‎storage/samples/snippets/bucket_ip_filter_test.py‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
# Copyright 2026 Google LLC
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
from typing import Generator
16+
import uuid
17+
18+
from google.api_core import exceptions
19+
from google.cloud import storage
20+
import pytest
21+
22+
import storage_create_bucket_ip_filtering
23+
import storage_delete_ip_filtering_rules
24+
import storage_disable_ip_filtering
25+
import storage_enable_ip_filtering
26+
import storage_get_ip_filtering
27+
import storage_list_buckets_ip_filtering
28+
29+
30+
@pytest.fixture
31+
def test_bucket() -> Generator[str, None, None]:
32+
storage_client = storage.Client()
33+
bucket_name = f"ipfilter-test-{uuid.uuid4().hex[:10]}"
34+
yield bucket_name
35+
try:
36+
bucket = storage_client.get_bucket(bucket_name)
37+
ip_filter = bucket.ip_filter
38+
if ip_filter and ip_filter.mode != "Disabled":
39+
ip_filter.mode = "Disabled"
40+
bucket.ip_filter = ip_filter
41+
bucket.patch()
42+
bucket.delete(force=True)
43+
except Exception:
44+
pass
45+
46+
47+
def test_ip_filter_lifecycle(test_bucket: str, capsys: pytest.CaptureFixture) -> None:
48+
public_range = "0.0.0.0/0"
49+
project_id = storage.Client().project
50+
vpc_network = f"projects/{project_id}/global/networks/default"
51+
52+
# 1. Create with IP filtering
53+
try:
54+
created = storage_create_bucket_ip_filtering.create_bucket_ip_filtering(
55+
test_bucket, public_range
56+
)
57+
assert created.ip_filter is not None
58+
assert created.ip_filter.mode == "Disabled"
59+
60+
# 2. Enable IP filtering
61+
enabled = storage_enable_ip_filtering.enable_ip_filtering(
62+
test_bucket,
63+
public_range=public_range,
64+
vpc_network=vpc_network,
65+
vpc_range="10.0.0.0/24",
66+
)
67+
assert enabled.ip_filter.mode == "Enabled"
68+
69+
# 3. Get IP filtering
70+
fetched = storage_get_ip_filtering.get_ip_filtering(test_bucket)
71+
assert fetched.mode == "Enabled"
72+
73+
# 4. Disable IP filtering
74+
disabled = storage_disable_ip_filtering.disable_ip_filtering(test_bucket)
75+
assert disabled.ip_filter.mode == "Disabled"
76+
77+
# 5. Delete IP filtering rules
78+
modified = storage_delete_ip_filtering_rules.delete_ip_filtering_rules(
79+
test_bucket,
80+
public_range_to_delete=public_range,
81+
vpc_network_to_delete=vpc_network,
82+
)
83+
assert (
84+
modified.ip_filter.public_network_source is None
85+
or public_range
86+
not in modified.ip_filter.public_network_source.allowed_ip_cidr_ranges
87+
)
88+
assert not any(
89+
v.network == vpc_network for v in modified.ip_filter.vpc_network_sources
90+
)
91+
92+
# 6. List buckets with IP filtering
93+
storage_list_buckets_ip_filtering.list_buckets_ip_filtering()
94+
out, _ = capsys.readouterr()
95+
assert test_bucket in out
96+
except exceptions.Forbidden as e:
97+
pytest.skip(
98+
"Skipping test due to insufficient permissions or IP filter"
99+
f" network restriction: {e}"
100+
)

‎storage/samples/snippets/requirements.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
google-cloud-pubsub==2.29.0
2-
google-cloud-storage==3.1.0
2+
google-cloud-storage==3.14.1
33
pandas===1.3.5; python_version == '3.7'
44
pandas===2.0.3; python_version == '3.8'
55
pandas==2.2.3; python_version >= '3.9'

‎storage/samples/snippets/storage_create_bucket_ip_filtering.py‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
#!/usr/bin/env python
2+
3+
# Copyright 2026 Google LLC
4+
#
5+
# Licensed under the Apache License, Version 2.0 (the "License");
6+
# you may not use this file except in compliance with the License.
7+
# You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
17+
import sys
18+
19+
# [START storage_create_bucket_ip_filtering]
20+
from google.cloud import storage
21+
from google.cloud.storage.ip_filter import IPFilter, PublicNetworkSource
22+
23+
24+
def create_bucket_ip_filtering(
25+
bucket_name: str, public_cidr_range: str
26+
) -> storage.Bucket:
27+
"""Creates a new bucket with initial IP filtering rules pre-configured."""
28+
# The ID of your GCS bucket
29+
# bucket_name = "your-bucket-name"
30+
# public_cidr_range = "192.0.2.0/24"
31+
32+
storage_client = storage.Client()
33+
bucket = storage_client.bucket(bucket_name)
34+
35+
ip_filter = IPFilter()
36+
ip_filter.mode = "Disabled"
37+
ip_filter.public_network_source = PublicNetworkSource(
38+
allowed_ip_cidr_ranges=[public_cidr_range]
39+
)
40+
ip_filter.allow_all_service_agent_access = True
41+
42+
bucket.ip_filter = ip_filter
43+
new_bucket = storage_client.create_bucket(bucket)
44+
45+
print(
46+
f"Created bucket {new_bucket.name} with IP filtering mode: {new_bucket.ip_filter.mode}"
47+
)
48+
return new_bucket
49+
50+
51+
# [END storage_create_bucket_ip_filtering]
52+
53+
if __name__ == "__main__":
54+
if len(sys.argv) < 3:
55+
print(
56+
"Usage: python storage_create_bucket_ip_filtering.py <bucket_name> <public_cidr_range>"
57+
)
58+
sys.exit(1)
59+
create_bucket_ip_filtering(bucket_name=sys.argv[1], public_cidr_range=sys.argv[2])

‎storage/samples/snippets/storage_delete_ip_filtering_rules.py‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
#!/usr/bin/env python
2+
3+
# Copyright 2026 Google LLC
4+
#
5+
# Licensed under the Apache License, Version 2.0 (the "License");
6+
# you may not use this file except in compliance with the License.
7+
# You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
17+
import sys
18+
19+
# [START storage_delete_ip_filtering_rules]
20+
from typing import Optional
21+
22+
from google.cloud import storage
23+
24+
25+
def delete_ip_filtering_rules(
26+
bucket_name: str,
27+
public_range_to_delete: Optional[str] = None,
28+
vpc_network_to_delete: Optional[str] = None,
29+
) -> storage.Bucket:
30+
"""Selectively removes specific public CIDR ranges or VPC network sources."""
31+
# The ID of your GCS bucket
32+
# bucket_name = "your-bucket-name"
33+
# public_range_to_delete = "192.0.2.0/24"
34+
# vpc_network_to_delete = "projects/my-project/global/networks/my-network"
35+
36+
storage_client = storage.Client()
37+
bucket = storage_client.get_bucket(bucket_name)
38+
39+
ip_filter = bucket.ip_filter
40+
if not ip_filter:
41+
print(f"No IP filter configuration found for bucket {bucket_name}.")
42+
return bucket
43+
44+
if public_range_to_delete and ip_filter.public_network_source:
45+
ranges = ip_filter.public_network_source.allowed_ip_cidr_ranges
46+
if ranges and public_range_to_delete in ranges:
47+
ranges.remove(public_range_to_delete)
48+
49+
if vpc_network_to_delete and ip_filter.vpc_network_sources:
50+
ip_filter.vpc_network_sources = [
51+
v
52+
for v in ip_filter.vpc_network_sources
53+
if v.network != vpc_network_to_delete
54+
]
55+
56+
# Re-assign to the bucket property to force google-cloud-storage to register
57+
# the nested changes for the patch() call.
58+
bucket.ip_filter = ip_filter
59+
bucket.patch()
60+
61+
print(f"Updated IP filtering rules for bucket {bucket_name}.")
62+
return bucket
63+
64+
65+
# [END storage_delete_ip_filtering_rules]
66+
67+
if __name__ == "__main__":
68+
if len(sys.argv) < 2:
69+
print(
70+
"Usage: python storage_delete_ip_filtering_rules.py <bucket_name> "
71+
"[public_range_to_delete] [vpc_network_to_delete]"
72+
)
73+
sys.exit(1)
74+
delete_ip_filtering_rules(
75+
bucket_name=sys.argv[1],
76+
public_range_to_delete=sys.argv[2] if len(sys.argv) > 2 else None,
77+
vpc_network_to_delete=sys.argv[3] if len(sys.argv) > 3 else None,
78+
)

‎storage/samples/snippets/storage_disable_ip_filtering.py‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
#!/usr/bin/env python
2+
3+
# Copyright 2026 Google LLC
4+
#
5+
# Licensed under the Apache License, Version 2.0 (the "License");
6+
# you may not use this file except in compliance with the License.
7+
# You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
17+
import sys
18+
19+
# [START storage_disable_ip_filtering]
20+
from google.cloud import storage
21+
22+
23+
def disable_ip_filtering(bucket_name: str) -> storage.Bucket:
24+
"""Disables IP filtering on a bucket without deleting existing rules."""
25+
# The ID of your GCS bucket
26+
# bucket_name = "your-bucket-name"
27+
28+
storage_client = storage.Client()
29+
bucket = storage_client.get_bucket(bucket_name)
30+
31+
ip_filter = bucket.ip_filter
32+
if not ip_filter:
33+
print(f"No IP filter configuration found for bucket {bucket_name}.")
34+
return bucket
35+
36+
ip_filter.mode = "Disabled"
37+
# Re-assign to the bucket property to force google-cloud-storage to register
38+
# the nested changes for the patch() call.
39+
bucket.ip_filter = ip_filter
40+
bucket.patch()
41+
print(f"IP filtering disabled for bucket {bucket_name}.")
42+
return bucket
43+
44+
45+
# [END storage_disable_ip_filtering]
46+
47+
if __name__ == "__main__":
48+
if len(sys.argv) < 2:
49+
print("Usage: python storage_disable_ip_filtering.py <bucket_name>")
50+
sys.exit(1)
51+
disable_ip_filtering(bucket_name=sys.argv[1])

0 commit comments

Comments
 (0)