Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 50 additions & 0 deletions samples/snippets/encryption_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand Down Expand Up @@ -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
Original file line number Diff line number Diff line change
@@ -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")
Original file line number Diff line number Diff line change
@@ -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")
Original file line number Diff line number Diff line change
@@ -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")
Loading