samples: add samples for bucket encryption enforcement config

samples/snippets/encryption_test.py

@@ -27,6 +27,10 @@
 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_bucket_encryption_enforcement_config
+from google.cloud.storage.bucket import EncryptionEnforcementConfig
 
 BUCKET = os.environ["CLOUD_STORAGE_BUCKET"]
 KMS_KEY = os.environ["MAIN_CLOUD_KMS_KEY"]
@@ -85,11 +89,7 @@ def test_blob():
     except NotFound as e:
         # For the case that the rotation succeeded.
         print(f"Ignoring 404, detail: {e}")
-        blob = Blob(
-            blob_name,
-            bucket,
-            encryption_key=TEST_ENCRYPTION_KEY_2_DECODED
-        )
+        blob = Blob(blob_name, bucket, encryption_key=TEST_ENCRYPTION_KEY_2_DECODED)
         blob.delete()
 
 
@@ -126,3 +126,106 @@ def test_object_csek_to_cmek(test_blob):
     )
 
     assert cmek_blob.download_as_bytes(), test_blob_content
+
+
+@pytest.fixture
+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 create_enforcement_bucket(bucket_name):
+    """Sets up a bucket with GMEK AND CSEK Restricted"""
+    client = storage.Client()
+    bucket = client.bucket(bucket_name)
+
+    bucket.encryption.google_managed_encryption_enforcement_config = (
+        EncryptionEnforcementConfig(restriction_mode="FullyRestricted")
+    )
+    bucket.encryption.customer_managed_encryption_enforcement_config = (
+        EncryptionEnforcementConfig(restriction_mode="NotRestricted")
+    )
+    bucket.encryption.customer_supplied_encryption_enforcement_config = (
+        EncryptionEnforcementConfig(restriction_mode="FullyRestricted")
+    )
+
+    bucket.create()
+    return bucket
+
+
+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.encryption.google_managed_encryption_enforcement_config.restriction_mode
+        == "FullyRestricted"
+    )
+    assert (
+        bucket.encryption.customer_managed_encryption_enforcement_config.restriction_mode
+        == "NotRestricted"
+    )
+    assert (
+        bucket.encryption.customer_supplied_encryption_enforcement_config.restriction_mode
+        == "FullyRestricted"
+    )
+
+
+def test_get_bucket_encryption_enforcement_config(enforcement_bucket, capsys):
+    # Pre-setup: Creating a bucket
+    create_enforcement_bucket(enforcement_bucket)
+
+    storage_get_bucket_encryption_enforcement_config.get_bucket_encryption_enforcement_config(
+        enforcement_bucket
+    )
+
+    out, _ = capsys.readouterr()
+    assert f"Encryption Enforcement Config for bucket {enforcement_bucket}" in out
+    assert (
+        "Customer-managed encryption enforcement config restriction mode: NotRestricted"
+        in out
+    )
+    assert (
+        "Customer-supplied encryption enforcement config restriction mode: FullyRestricted"
+        in out
+    )
+    assert (
+        "Google-managed encryption enforcement config restriction mode: FullyRestricted"
+        in out
+    )
+
+
+def test_update_encryption_enforcement_config(enforcement_bucket):
+    # Pre-setup: Create a bucket in a different state before update
+    create_enforcement_bucket(enforcement_bucket)
+
+    storage_update_bucket_encryption_enforcement_config.update_bucket_encryption_enforcement_config(
+        enforcement_bucket
+    )
+
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(enforcement_bucket)
+
+    assert (
+        bucket.encryption.google_managed_encryption_enforcement_config.restriction_mode
+        == "NotRestricted"
+    )
+    assert (
+        bucket.encryption.customer_managed_encryption_enforcement_config.restriction_mode
+        == "FullyRestricted"
+    )
+    assert (
+        bucket.encryption.customer_supplied_encryption_enforcement_config.restriction_mode
+        == "FullyRestricted"
+    )

samples/snippets/storage_get_bucket_encryption_enforcement_config.py

@@ -0,0 +1,48 @@
+# 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.encryption.customer_managed_encryption_enforcement_config
+    csek_config = bucket.encryption.customer_supplied_encryption_enforcement_config
+    gmek_config = bucket.encryption.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")

samples/snippets/storage_set_bucket_encryption_enforcement_config.py

@@ -0,0 +1,55 @@
+# 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.encryption.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.encryption.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.encryption.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")

samples/snippets/storage_update_bucket_encryption_enforcement_config.py

@@ -0,0 +1,60 @@
+# 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_bucket_encryption_enforcement_config]
+from google.cloud import storage
+from google.cloud.storage.bucket import EncryptionEnforcementConfig
+
+
+def update_bucket_encryption_enforcement_config(bucket_name):
+    """Updates the encryption enforcement policy for a bucket."""
+    # The ID of your GCS bucket with GMEK and CSEK restricted
+    # bucket_name = "your-unique-bucket-name"
+
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(bucket_name)
+
+    # Update a specific type (e.g., change GMEK to NotRestricted)
+    bucket.encryption.google_managed_encryption_enforcement_config = (
+        EncryptionEnforcementConfig(restriction_mode="NotRestricted")
+    )
+
+    # Update another type (e.g., change CMEK to FullyRestricted)
+    bucket.encryption.customer_managed_encryption_enforcement_config = (
+        EncryptionEnforcementConfig(restriction_mode="FullyRestricted")
+    )
+
+    # Keeping CSEK unchanged
+    bucket.encryption.customer_supplied_encryption_enforcement_config = (
+        EncryptionEnforcementConfig(restriction_mode="FullyRestricted")
+    )
+
+    bucket.patch()
+
+    print(f"Encryption enforcement policy updated for bucket {bucket.name}.")
+
+    gmek = bucket.encryption.google_managed_encryption_enforcement_config
+    cmek = bucket.encryption.customer_managed_encryption_enforcement_config
+    csek = bucket.encryption.customer_supplied_encryption_enforcement_config
+
+    print(f"GMEK restriction mode: {gmek.restriction_mode if gmek else 'None'}")
+    print(f"CMEK restriction mode: {cmek.restriction_mode if cmek else 'None'}")
+    print(f"CSEK restriction mode: {csek.restriction_mode if csek else 'None'}")
+
+
+# [END storage_update_bucket_encryption_enforcement_config]
+
+
+if __name__ == "__main__":
+    update_bucket_encryption_enforcement_config(bucket_name="your-unique-bucket-name")