diff --git a/samples/snippets/encryption_test.py b/samples/snippets/encryption_test.py index 9039b1fad..9229ea607 100644 --- a/samples/snippets/encryption_test.py +++ b/samples/snippets/encryption_test.py @@ -27,6 +27,9 @@ import storage_object_csek_to_cmek import storage_rotate_encryption_key import storage_upload_encrypted_file +import storage_get_bucket_encryption_enforcement_config +import storage_set_bucket_encryption_enforcement_config +import storage_update_encryption_enforcement_config BUCKET = os.environ["CLOUD_STORAGE_BUCKET"] KMS_KEY = os.environ["MAIN_CLOUD_KMS_KEY"] @@ -126,3 +129,50 @@ def test_object_csek_to_cmek(test_blob): ) assert cmek_blob.download_as_bytes(), test_blob_content + + +@pytest.fixture(scope="module") +def enforcement_bucket(): + bucket_name = f"test_encryption_enforcement_{uuid.uuid4().hex}" + yield bucket_name + + storage_client = storage.Client() + try: + bucket = storage_client.get_bucket(bucket_name) + bucket.delete(force=True) + except Exception: + pass + + +def test_set_bucket_encryption_enforcement_config(enforcement_bucket): + storage_set_bucket_encryption_enforcement_config.set_bucket_encryption_enforcement_config( + enforcement_bucket + ) + + storage_client = storage.Client() + bucket = storage_client.get_bucket(enforcement_bucket) + + assert bucket.google_managed_encryption_enforcement_config.restriction_mode == "FullyRestricted" + assert bucket.customer_managed_encryption_enforcement_config.restriction_mode == "NotRestricted" + assert bucket.customer_supplied_encryption_enforcement_config.restriction_mode == "FullyRestricted" + + +def test_get_bucket_encryption_enforcement_config(enforcement_bucket): + # This just exercises the get snippet. If it crashes, the test fails. + # The assertions on the state were done in the set test. + storage_get_bucket_encryption_enforcement_config.get_bucket_encryption_enforcement_config( + enforcement_bucket + ) + + +def test_update_encryption_enforcement_config(enforcement_bucket): + storage_update_encryption_enforcement_config.update_encryption_enforcement_config( + enforcement_bucket + ) + + storage_client = storage.Client() + bucket = storage_client.get_bucket(enforcement_bucket) + + assert bucket.google_managed_encryption_enforcement_config.restriction_mode == "FullyRestricted" + assert bucket.customer_managed_encryption_enforcement_config.restriction_mode == "NotRestricted" + assert bucket.customer_supplied_encryption_enforcement_config is None diff --git a/samples/snippets/storage_get_bucket_encryption_enforcement_config.py b/samples/snippets/storage_get_bucket_encryption_enforcement_config.py new file mode 100644 index 000000000..269a41376 --- /dev/null +++ b/samples/snippets/storage_get_bucket_encryption_enforcement_config.py @@ -0,0 +1,40 @@ +# 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_get_bucket_encryption_enforcement_config] +from google.cloud import storage + + +def get_bucket_encryption_enforcement_config(bucket_name): + """Gets the bucket encryption enforcement configuration.""" + # The ID of your GCS bucket + # bucket_name = "your-unique-bucket-name" + + storage_client = storage.Client() + bucket = storage_client.get_bucket(bucket_name) + + print(f"Encryption Enforcement Config for bucket {bucket.name}:") + + cmek_config = bucket.customer_managed_encryption_enforcement_config + csek_config = bucket.customer_supplied_encryption_enforcement_config + gmek_config = bucket.google_managed_encryption_enforcement_config + + print(f"Customer-managed encryption enforcement config restriction mode: {cmek_config.restriction_mode if cmek_config else None}") + print(f"Customer-supplied encryption enforcement config restriction mode: {csek_config.restriction_mode if csek_config else None}") + print(f"Google-managed encryption enforcement config restriction mode: {gmek_config.restriction_mode if gmek_config else None}") +# [END storage_get_bucket_encryption_enforcement_config] + + +if __name__ == "__main__": + get_bucket_encryption_enforcement_config(bucket_name="your-unique-bucket-name") diff --git a/samples/snippets/storage_set_bucket_encryption_enforcement_config.py b/samples/snippets/storage_set_bucket_encryption_enforcement_config.py new file mode 100644 index 000000000..ac10eb44d --- /dev/null +++ b/samples/snippets/storage_set_bucket_encryption_enforcement_config.py @@ -0,0 +1,53 @@ +# 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_set_bucket_encryption_enforcement_config] +from google.cloud import storage +from google.cloud.storage.bucket import EncryptionEnforcementConfig + + +def set_bucket_encryption_enforcement_config(bucket_name): + """Creates a bucket with encryption enforcement configuration.""" + # The ID of your GCS bucket + # bucket_name = "your-unique-bucket-name" + + storage_client = storage.Client() + bucket = storage_client.bucket(bucket_name) + + # Setting restriction_mode to "FullyRestricted" for Google-managed encryption (GMEK) + # means objects cannot be created using the default Google-managed keys. + bucket.google_managed_encryption_enforcement_config = EncryptionEnforcementConfig( + restriction_mode="FullyRestricted" + ) + + # Setting restriction_mode to "NotRestricted" for Customer-managed encryption (CMEK) + # ensures that objects ARE permitted to be created using Cloud KMS keys. + bucket.customer_managed_encryption_enforcement_config = EncryptionEnforcementConfig( + restriction_mode="NotRestricted" + ) + + # Setting restriction_mode to "FullyRestricted" for Customer-supplied encryption (CSEK) + # prevents objects from being created using raw, client-side provided keys. + bucket.customer_supplied_encryption_enforcement_config = EncryptionEnforcementConfig( + restriction_mode="FullyRestricted" + ) + + bucket.create() + + print(f"Created bucket {bucket.name} with Encryption Enforcement Config.") +# [END storage_set_bucket_encryption_enforcement_config] + + +if __name__ == "__main__": + set_bucket_encryption_enforcement_config(bucket_name="your-unique-bucket-name") diff --git a/samples/snippets/storage_update_encryption_enforcement_config.py b/samples/snippets/storage_update_encryption_enforcement_config.py new file mode 100644 index 000000000..0fa38ee01 --- /dev/null +++ b/samples/snippets/storage_update_encryption_enforcement_config.py @@ -0,0 +1,43 @@ +# 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_update_encryption_enforcement_config] +from google.cloud import storage +from google.cloud.storage.bucket import EncryptionEnforcementConfig + + +def update_encryption_enforcement_config(bucket_name): + """Updates the encryption enforcement policy for a bucket.""" + # The ID of your GCS bucket + # bucket_name = "your-unique-bucket-name" + + storage_client = storage.Client() + bucket = storage_client.get_bucket(bucket_name) + + # 1. Update a specific type (e.g., change GMEK to FullyRestricted) + bucket.google_managed_encryption_enforcement_config = EncryptionEnforcementConfig(restriction_mode="FullyRestricted") + bucket.customer_managed_encryption_enforcement_config = EncryptionEnforcementConfig(restriction_mode="NotRestricted") + + # 2. Remove a specific type (e.g., remove CSEK enforcement) + bucket.customer_supplied_encryption_enforcement_config = None + + bucket.patch() + + print(f"Encryption enforcement policy updated for bucket {bucket.name}.") + print("GMEK is now fully restricted, CMEK is now not restricted, and CSEK enforcement has been removed.") +# [END storage_update_encryption_enforcement_config] + + +if __name__ == "__main__": + update_encryption_enforcement_config(bucket_name="your-unique-bucket-name")