diff --git a/secretmanager/snippets/create_secret_with_cmek.py b/secretmanager/snippets/create_secret_with_cmek.py new file mode 100644 index 00000000000..7ed7892b873 --- /dev/null +++ b/secretmanager/snippets/create_secret_with_cmek.py @@ -0,0 +1,84 @@ +#!/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 secretmanager_create_secret_with_cmek] +from google.cloud import secretmanager + + +def create_secret_with_cmek( + project_id: str, secret_id: str, kms_key_name: str +) -> None: + """ + Creates a new secret with a customer-managed encryption key (CMEK). + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to create + kms_key_name (str): Full resource name of the KMS key + (e.g., "projects/my-project/locations/global/keyRings/{keyringname}/cryptoKeys/{keyname}") + + Example: + # Create a secret with a customer-managed encryption key + create_secret_with_cmek( + "my-project", + "my-secret-with-cmek", + "projects/my-project/locations/global/keyRings/my-keyring/cryptoKeys/my-key" + ) + """ + # Create the Secret Manager client. + client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the parent project. + parent = f"projects/{project_id}" + + # Create the secret with automatic replication and CMEK. + secret = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": { + "replication": { + "automatic": { + "customer_managed_encryption": { + "kms_key_name": kms_key_name + } + } + } + }, + } + ) + + print(f"Created secret {secret.name} with CMEK key {kms_key_name}") + + +# [END secretmanager_create_secret_with_cmek] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to create") + parser.add_argument( + "kms_key_name", help="full resource name of the KMS key" + ) + args = parser.parse_args() + + create_secret_with_cmek(args.project_id, args.secret_id, args.kms_key_name) diff --git a/secretmanager/snippets/create_secret_with_expiration.py b/secretmanager/snippets/create_secret_with_expiration.py new file mode 100644 index 00000000000..352d77dd051 --- /dev/null +++ b/secretmanager/snippets/create_secret_with_expiration.py @@ -0,0 +1,77 @@ +#!/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 secretmanager_create_secret_with_expiration] +from datetime import datetime, timedelta +from google.protobuf import timestamp_pb2 +from google.cloud import secretmanager + + +def create_secret_with_expiration(project_id: str, secret_id: str) -> None: + """ + Create a new secret with an expiration time. + + Args: + project_id (str): The ID of the project where the secret will be created. + secret_id (str): The ID for the secret to create. + + Example: + # Create a secret that expires in 24 hours + create_secret_with_expiration("my-project", "my-secret-with-expiry") + """ + expire_time = datetime.now() + timedelta(hours=1) + # Create the Secret Manager client. + client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the parent project. + parent = f"projects/{project_id}" + + # Convert the Python datetime to a Protobuf Timestamp + timestamp = timestamp_pb2.Timestamp() + timestamp.FromDatetime(expire_time) + + # Create the secret with automatic replication and expiration time. + secret = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": { + "replication": { + "automatic": {}, + }, + "expire_time": timestamp, + }, + } + ) + + print(f"Created secret {secret.name} with expiration time {expire_time}") + + +# [END secretmanager_create_secret_with_expiration] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to create") + args = parser.parse_args() + + create_secret_with_expiration(args.project_id, args.secret_id) diff --git a/secretmanager/snippets/create_secret_with_rotation.py b/secretmanager/snippets/create_secret_with_rotation.py new file mode 100644 index 00000000000..4d22ced2bc0 --- /dev/null +++ b/secretmanager/snippets/create_secret_with_rotation.py @@ -0,0 +1,102 @@ +#!/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 secretmanager_create_secret_with_rotation] +from datetime import datetime, timedelta +from google.cloud import secretmanager +from google.protobuf import duration_pb2 +from google.protobuf import timestamp_pb2 + + +def create_secret_with_rotation( + project_id: str, secret_id: str, topic_name: str +) -> None: + """ + Creates a new secret with rotation configured. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to create + topic_name (str): Resource name of the Pub/Sub topic for rotation notifications + Example: + # Create a secret with automatic rotation every 30 days + create_secret_with_rotation( + "my-project", + "my-rotating-secret", + "projects/my-project/topics/my-rotation-topic" + ) + """ + rotation_period_hours = 24 + # Create the Secret Manager client + client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the parent project + parent = f"projects/{project_id}" + + # Convert rotation period to protobuf Duration + rotation_period = duration_pb2.Duration() + rotation_period.seconds = ( + rotation_period_hours * 3600 + ) # Convert hours to seconds + + # Set next rotation time to 24 hours from now + next_rotation_time = timestamp_pb2.Timestamp() + next_rotation_time.FromDatetime(datetime.now() + timedelta(hours=24)) + + # Create the secret with rotation configuration + secret = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": { + "replication": {"automatic": {}}, + "topics": [{"name": topic_name}], + "rotation": { + "next_rotation_time": next_rotation_time, + "rotation_period": rotation_period, + }, + }, + } + ) + + print( + f"Created secret {secret.name} with rotation period {rotation_period_hours} hours and topic {topic_name}" + ) + + +# [END secretmanager_create_secret_with_rotation] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="ID of the GCP project") + parser.add_argument("secret_id", help="ID of the secret to create") + parser.add_argument( + "topic_name", + help="Resource name of the Pub/Sub topic for rotation notifications", + ) + args = parser.parse_args() + + create_secret_with_rotation( + args.project_id, + args.secret_id, + args.topic_name, + ) diff --git a/secretmanager/snippets/create_secret_with_topic.py b/secretmanager/snippets/create_secret_with_topic.py new file mode 100644 index 00000000000..05789f8ddbd --- /dev/null +++ b/secretmanager/snippets/create_secret_with_topic.py @@ -0,0 +1,78 @@ +#!/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 secretmanager_create_secret_with_topic] +from google.cloud import secretmanager + + +def create_secret_with_topic( + project_id: str, secret_id: str, topic_name: str +) -> None: + """ + Creates a new secret with a notification topic configured. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to create + topic_name (str): Full name of the topic in the format "projects/my-project/topics/my-topic" + Example: + # Create a secret with a Pub/Sub notification configuration + create_secret_with_topic( + "my-project", + "my-secret-with-notifications", + "projects/my-project/topics/my-secret-topic" + ) + """ + # Create the Secret Manager client. + client = secretmanager.SecretManagerServiceClient() + + # Build the parent name. + parent = f"projects/{project_id}" + + # Create the secret with topic configuration. + secret = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": { + "replication": {"automatic": {}}, + "topics": [{"name": topic_name}], + }, + } + ) + + print(f"Created secret {secret.name} with topic {topic_name}") + + +# [END secretmanager_create_secret_with_topic] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to create") + parser.add_argument( + "topic_name", + help="name of the topic in the format 'projects/my-project/topics/my-topic'", + ) + args = parser.parse_args() + + create_secret_with_topic(args.project_id, args.secret_id, args.topic_name) diff --git a/secretmanager/snippets/delete_secret_expiration.py b/secretmanager/snippets/delete_secret_expiration.py new file mode 100644 index 00000000000..066d2f6fcba --- /dev/null +++ b/secretmanager/snippets/delete_secret_expiration.py @@ -0,0 +1,68 @@ +#!/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 secretmanager_delete_secret_expiration] +from google.cloud import secretmanager +from google.protobuf.field_mask_pb2 import FieldMask + + +def delete_secret_expiration(project_id: str, secret_id: str) -> None: + """ + Removes the expiration time from a secret. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to update + Example: + # Remove the expiration time from a secret that was previously scheduled for deletion + delete_secret_expiration( + "my-project", + "my-secret-with-expiration" + ) + """ + # Create the Secret Manager client. + client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the secret + name = client.secret_path(project_id, secret_id) + + # Create the update mask. + update_mask = FieldMask(paths=["expire_time"]) + + # Build the request. + request = {"secret": {"name": name}, "update_mask": update_mask} + + # Update the secret. + secret = client.update_secret(request=request) + + print(f"Removed expiration from secret {secret.name}") + + +# [END secretmanager_delete_secret_expiration] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to act on") + args = parser.parse_args() + + delete_secret_expiration(args.project_id, args.secret_id) diff --git a/secretmanager/snippets/delete_secret_rotation.py b/secretmanager/snippets/delete_secret_rotation.py new file mode 100644 index 00000000000..d94770e833b --- /dev/null +++ b/secretmanager/snippets/delete_secret_rotation.py @@ -0,0 +1,68 @@ +#!/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 secretmanager_delete_secret_rotation] +from google.cloud import secretmanager +from google.protobuf.field_mask_pb2 import FieldMask + + +def delete_secret_rotation(project_id: str, secret_id: str) -> None: + """ + Removes the rotation configuration from a secret. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret with rotation to remove + Example: + # Remove the rotation configuration from a secret + delete_secret_rotation( + "my-project", + "my-secret-with-rotation" + ) + """ + # Create the Secret Manager client. + client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the secret + name = client.secret_path(project_id, secret_id) + + # Create the update mask. + update_mask = FieldMask(paths=["rotation"]) + + # Build the request. + request = {"secret": {"name": name}, "update_mask": update_mask} + + # Update the secret. + secret = client.update_secret(request=request) + + print(f"Removed rotation from secret {secret.name}") + + +# [END secretmanager_delete_secret_rotation] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to act on") + args = parser.parse_args() + + delete_secret_rotation(args.project_id, args.secret_id) diff --git a/secretmanager/snippets/detach_tag_binding.py b/secretmanager/snippets/detach_tag_binding.py new file mode 100644 index 00000000000..4afa9153566 --- /dev/null +++ b/secretmanager/snippets/detach_tag_binding.py @@ -0,0 +1,85 @@ +#!/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 argparse + +# [START secretmanager_detach_tag_binding] +from google.cloud import resourcemanager_v3 + + +def detach_tag(project_id: str, secret_id: str, tag_value: str) -> None: + """ + Detaches a tag value from a secret. + + Args: + project_id (str): The project ID where the secret exists. + secret_id (str): The ID of the secret from which to detach the tag. + tag_value (str): The tag value to detach (e.g., "tagValues/123456789012"). + + Example: + # Detach a tag value from a secret + detach_tag("my-project", "my-secret", "tagValues/123456789012") + """ + # Create the Resource Manager client. + client = resourcemanager_v3.TagBindingsClient() + + # Build the resource name of the parent secret. + secret_name = f"projects/{project_id}/secrets/{secret_id}" + parent = f"//secretmanager.googleapis.com/{secret_name}" + + # Find the binding name for the given tag value + binding_name = None + request = resourcemanager_v3.ListTagBindingsRequest(parent=parent) + + for binding in client.list_tag_bindings(request=request): + if binding.tag_value == tag_value: + binding_name = binding.name + break + + if binding_name is None: + print(f"Tag binding for value {tag_value} not found on {secret_name}.") + return + + # Delete the tag binding + request = resourcemanager_v3.DeleteTagBindingRequest(name=binding_name) + operation = client.delete_tag_binding(request=request) + + # Wait for the operation to complete + operation.result() + + print(f"Detached tag value {tag_value} from {secret_name}") + + +# [END secretmanager_detach_tag_binding] + + +if __name__ == "__main__": + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument( + "secret_id", help="id of the secret to detach tag from" + ) + parser.add_argument( + "tag_value", + help="tag value to detach (e.g., 'tagValues/123456789012')", + ) + args = parser.parse_args() + + detach_tag(args.project_id, args.secret_id, args.tag_value) diff --git a/secretmanager/snippets/list_tag_bindings.py b/secretmanager/snippets/list_tag_bindings.py new file mode 100644 index 00000000000..ea04fed9c16 --- /dev/null +++ b/secretmanager/snippets/list_tag_bindings.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 +""" +command line application and sample code for listing tag bindings attached to a secret. +""" + +import argparse + +# [START secretmanager_list_tag_bindings] +# Import the Resource Manager client library. +from google.cloud import resourcemanager_v3 +from google.cloud import secretmanager + + +def list_tag_bindings(project_id: str, secret_id: str) -> None: + """ + Lists all tag bindings attached to a secret. + + Args: + project_id (str): The project ID where the secret exists. + secret_id (str): The ID of the secret to list tag bindings for. + + Example: + # List tag bindings for a secret + list_tag_bindings("my-project", "my-secret") + """ + + # Create the Resource Manager client. + client = resourcemanager_v3.TagBindingsClient() + sm_client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the parent secret. + secret_name = sm_client.secret_path(project_id, secret_id) + + parent = f"//secretmanager.googleapis.com/{secret_name}" + + # List all tag bindings. + tag_bindings = [] + request = resourcemanager_v3.ListTagBindingsRequest(parent=parent) + + # Retrieve and process tag bindings + bindings = client.list_tag_bindings(request=request) + count = 0 + + print(f"Tag bindings for {secret_name}:") + for binding in bindings: + print(f"- Tag Value: {binding.tag_value}") + tag_bindings.append(binding) + count += 1 + + if count == 0: + print(f"No tag bindings found for {secret_name}.") + + +# [END secretmanager_list_tag_bindings] + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument( + "secret_id", help="id of the secret to list tag bindings for" + ) + args = parser.parse_args() + + list_tag_bindings(args.project_id, args.secret_id) diff --git a/secretmanager/snippets/regional_samples/create_regional_secret_with_cmek.py b/secretmanager/snippets/regional_samples/create_regional_secret_with_cmek.py new file mode 100644 index 00000000000..b2cfd7b64ae --- /dev/null +++ b/secretmanager/snippets/regional_samples/create_regional_secret_with_cmek.py @@ -0,0 +1,94 @@ +#!/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 argparse + +# [START secretmanager_create_regional_secret_with_cmek] +from google.cloud import secretmanager +from google.api_core import client_options + + +def create_regional_secret_with_cmek( + project_id: str, secret_id: str, location_id: str, kms_key_name: str +) -> None: + """ + Creates a new regional secret encrypted with a customer-managed encryption key (CMEK). + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to create + location_id (str): Region where the secret should be stored (e.g., "us-central1") + kms_key_name (str): Full resource name of the KMS key + (e.g., "projects/my-project/locations/{location_id}/keyRings/{keyringname}/cryptoKeys/{keyname}") + + Example: + # Create a regional secret with a customer-managed encryption key + create_regional_secret_with_cmek( + "my-project", + "my-regional-secret-with-cmek", + "us-central1", + "projects/my-project/locations/us-central1/keyRings/my-keyring/cryptoKeys/my-key" + ) + """ + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Build the resource name of the parent project with location + parent = f"projects/{project_id}/locations/{location_id}" + + # Create the secret with CMEK + secret = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": { + "customer_managed_encryption": {"kms_key_name": kms_key_name} + }, + } + ) + + print(f"Created secret {secret.name} with CMEK key {kms_key_name}") + + +# [END secretmanager_create_regional_secret_with_cmek] + + +if __name__ == "__main__": + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to create") + parser.add_argument( + "location_id", + help="region where the secret should be stored (e.g., us-central1)", + ) + parser.add_argument( + "kms_key_name", help="full resource name of the KMS key" + ) + args = parser.parse_args() + + create_regional_secret_with_cmek( + args.project_id, args.secret_id, args.location_id, args.kms_key_name + ) diff --git a/secretmanager/snippets/regional_samples/create_regional_secret_with_expire_time.py b/secretmanager/snippets/regional_samples/create_regional_secret_with_expire_time.py new file mode 100644 index 00000000000..46d29879bb6 --- /dev/null +++ b/secretmanager/snippets/regional_samples/create_regional_secret_with_expire_time.py @@ -0,0 +1,90 @@ +#!/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 secretmanager_create_regional_secret_with_expire_time] +from datetime import datetime, timedelta +from google.cloud import secretmanager +from google.api_core import client_options +from google.protobuf import timestamp_pb2 + + +def create_regional_secret_with_expire_time( + project_id: str, secret_id: str, location_id: str +) -> None: + """ + Creates a new regional secret with an expiration time. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to create + location_id (str): Region where the secret should be stored (e.g., "us-central1") + + Example: + # Create a regional secret that expires in 1 hour + create_regional_secret_with_expire_time("my-project", "my-secret-with-expiry", "us-central1") + """ + # Set expiration time to 1 hour from now + expire_time = datetime.now() + timedelta(hours=1) + + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Build the resource name of the parent project with location + parent = f"projects/{project_id}/locations/{location_id}" + + # Convert the Python datetime to a Protobuf Timestamp + timestamp = timestamp_pb2.Timestamp() + timestamp.FromDatetime(expire_time) + + # Create the secret with expiration time + secret = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": {"expire_time": timestamp}, + } + ) + + print(f"Created secret {secret.name} with expiration time {expire_time}") + + +# [END secretmanager_create_regional_secret_with_expire_time] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to create") + parser.add_argument( + "location_id", + help="region where the secret should be stored (e.g., us-central1)", + ) + args = parser.parse_args() + + create_regional_secret_with_expire_time( + args.project_id, args.secret_id, args.location_id + ) diff --git a/secretmanager/snippets/regional_samples/create_regional_secret_with_rotation.py b/secretmanager/snippets/regional_samples/create_regional_secret_with_rotation.py new file mode 100644 index 00000000000..6487497791d --- /dev/null +++ b/secretmanager/snippets/regional_samples/create_regional_secret_with_rotation.py @@ -0,0 +1,116 @@ +#!/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 secretmanager_create_regional_secret_with_rotation] +from datetime import datetime, timedelta +from google.cloud import secretmanager +from google.api_core import client_options +from google.protobuf import timestamp_pb2, duration_pb2 + + +def create_regional_secret_with_rotation( + project_id: str, secret_id: str, location_id: str, topic_name: str +) -> None: + """ + Creates a new regional secret with rotation configured. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to create + location_id (str): Region where the secret should be stored (e.g., "us-central1") + topic_name (str): Full resource name of the Pub/Sub topic for notifications + (e.g., "projects/my-project/topics/my-topic") + + Example: + # Create a regional secret with rotation + create_regional_secret_with_rotation( + "my-project", + "my-secret-with-rotation", + "us-central1", + "projects/my-project/topics/my-topic" + ) + """ + # Set rotation period to 24 hours + rotation_period = timedelta(hours=24) + + # Set next rotation time to 24 hours from now + next_rotation_time = datetime.now() + timedelta(hours=24) + + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Build the resource name of the parent project with location + parent = f"projects/{project_id}/locations/{location_id}" + + # Convert the Python datetime to a Protobuf Timestamp + next_rotation_timestamp = timestamp_pb2.Timestamp() + next_rotation_timestamp.FromDatetime(next_rotation_time) + + # Convert the Python timedelta to a Protobuf Duration + rotation_period_proto = duration_pb2.Duration() + rotation_period_proto.FromTimedelta(rotation_period) + + # Create the secret with rotation configuration and topic + secret = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": { + "topics": [{"name": topic_name}], + "rotation": { + "next_rotation_time": next_rotation_timestamp, + "rotation_period": rotation_period_proto, + }, + }, + } + ) + + print( + f"Created secret {secret.name} with rotation period {rotation_period} and topic {topic_name}" + ) + + +# [END secretmanager_create_regional_secret_with_rotation] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to create") + parser.add_argument( + "location_id", + help="region where the secret should be stored (e.g., us-central1)", + ) + parser.add_argument( + "topic_name", + help="full resource name of the Pub/Sub topic for notifications", + ) + args = parser.parse_args() + + create_regional_secret_with_rotation( + args.project_id, args.secret_id, args.location_id, args.topic_name + ) diff --git a/secretmanager/snippets/regional_samples/create_regional_secret_with_topic.py b/secretmanager/snippets/regional_samples/create_regional_secret_with_topic.py new file mode 100644 index 00000000000..edbc5d11b56 --- /dev/null +++ b/secretmanager/snippets/regional_samples/create_regional_secret_with_topic.py @@ -0,0 +1,92 @@ +#!/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 secretmanager_create_regional_secret_with_topic] +from google.cloud import secretmanager +from google.api_core import client_options + + +def create_regional_secret_with_topic( + project_id: str, secret_id: str, location_id: str, topic_name: str +) -> None: + """ + Creates a new regional secret with a Pub/Sub topic configured for notifications. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to create + location_id (str): Region where the secret should be stored (e.g., "us-central1") + topic_name (str): Full resource name of the Pub/Sub topic for notifications + (e.g., "projects/my-project/topics/my-topic") + + Example: + # Create a regional secret with a Pub/Sub topic + create_regional_secret_with_topic( + "my-project", + "my-secret-with-topic", + "us-central1", + "projects/my-project/topics/my-topic" + ) + """ + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Build the resource name of the parent project with location + parent = f"projects/{project_id}/locations/{location_id}" + + # Create the secret with a topic for notifications + secret = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": {"topics": [{"name": topic_name}]}, + } + ) + + print(f"Created secret {secret.name} with topic {topic_name}") + + +# [END secretmanager_create_regional_secret_with_topic] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret to create") + parser.add_argument( + "location_id", + help="region where the secret should be stored (e.g., us-central1)", + ) + parser.add_argument( + "topic_name", + help="full resource name of the Pub/Sub topic for notifications", + ) + args = parser.parse_args() + + create_regional_secret_with_topic( + args.project_id, args.secret_id, args.location_id, args.topic_name + ) diff --git a/secretmanager/snippets/regional_samples/delete_regional_secret_expiration.py b/secretmanager/snippets/regional_samples/delete_regional_secret_expiration.py new file mode 100644 index 00000000000..c48e15e5486 --- /dev/null +++ b/secretmanager/snippets/regional_samples/delete_regional_secret_expiration.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 argparse + +# [START secretmanager_delete_regional_secret_expiration] +from google.cloud import secretmanager +from google.api_core import client_options +from google.protobuf import field_mask_pb2 + + +def delete_regional_secret_expiration( + project_id: str, secret_id: str, location_id: str +) -> None: + """ + Removes the expiration time from a regional secret. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret + location_id (str): Region where the secret is stored (e.g., "us-central1") + + Example: + # Remove expiration from a regional secret + delete_regional_secret_expiration( + "my-project", + "my-secret", + "us-central1" + ) + """ + # Construct the secret name from the component parts + secret_name = ( + f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + ) + + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Create a field mask to update only the expire_time field + update_mask = field_mask_pb2.FieldMask(paths=["expire_time"]) + + # Update the secret to remove the expiration time + secret = client.update_secret( + request={ + "secret": { + "name": secret_name, + }, + "update_mask": update_mask, + } + ) + + print(f"Removed expiration from secret {secret.name}") + + +# [END secretmanager_delete_regional_secret_expiration] + + +if __name__ == "__main__": + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret") + parser.add_argument( + "location_id", + help="region where the secret is stored (e.g., us-central1)", + ) + args = parser.parse_args() + + delete_regional_secret_expiration( + args.project_id, args.secret_id, args.location_id + ) diff --git a/secretmanager/snippets/regional_samples/delete_regional_secret_rotation.py b/secretmanager/snippets/regional_samples/delete_regional_secret_rotation.py new file mode 100644 index 00000000000..37b10524bd8 --- /dev/null +++ b/secretmanager/snippets/regional_samples/delete_regional_secret_rotation.py @@ -0,0 +1,92 @@ +#!/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 argparse + +# [START secretmanager_delete_regional_secret_rotation] +from google.cloud import secretmanager +from google.api_core import client_options +from google.protobuf import field_mask_pb2 + + +def delete_regional_secret_rotation( + project_id: str, secret_id: str, location_id: str +) -> None: + """ + Removes the rotation configuration from a regional secret. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret + location_id (str): Region where the secret is stored (e.g., "us-central1") + + Example: + # Delete rotation configuration from a regional secret + delete_regional_secret_rotation( + "my-project", + "my-secret-with-rotation", + "us-central1" + ) + """ + # Construct the secret name from the component parts + secret_name = ( + f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + ) + + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Create a field mask to update only the rotation field + update_mask = field_mask_pb2.FieldMask(paths=["rotation"]) + + # Update the secret to remove the rotation configuration + result = client.update_secret( + request={ + "secret": { + "name": secret_name, + }, + "update_mask": update_mask, + } + ) + + print(f"Removed rotation from secret {result.name}") + + +# [END secretmanager_delete_regional_secret_rotation] + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret") + parser.add_argument( + "location_id", + help="region where the secret is stored (e.g., us-central1)", + ) + args = parser.parse_args() + + delete_regional_secret_rotation( + args.project_id, args.secret_id, args.location_id + ) diff --git a/secretmanager/snippets/regional_samples/detach_regional_tag.py b/secretmanager/snippets/regional_samples/detach_regional_tag.py new file mode 100644 index 00000000000..e6b47c47ca9 --- /dev/null +++ b/secretmanager/snippets/regional_samples/detach_regional_tag.py @@ -0,0 +1,105 @@ +#!/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 secretmanager_detach_regional_tag_binding] +from google.cloud import resourcemanager_v3 +from google.api_core import client_options + + +def detach_regional_tag( + project_id: str, location_id: str, secret_id: str, tag_value: str +) -> None: + """ + Detaches a tag value from a regional secret. + + Args: + project_id (str): ID of the Google Cloud project + location_id (str): Region where the secret is stored (e.g., "us-central1") + secret_id (str): ID of the secret + tag_value (str): Tag value to detach (e.g., "tagValues/123456789012") + + Example: + # Detach a tag value from a regional secret + detach_regional_tag( + "my-project", + "us-central1", + "my-secret", + "tagValues/123456789012" + ) + """ + # Set up the endpoint for the regional resource manager + rm_endpoint = f"{location_id}-cloudresourcemanager.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=rm_endpoint) + + # Create the Tag Bindings client with the regional endpoint + tag_bindings_client = resourcemanager_v3.TagBindingsClient( + client_options=client_option + ) + + secret_name = ( + f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + ) + + # Format the parent resource for the tag bindings request + parent = f"//secretmanager.googleapis.com/{secret_name}" + + # Find the binding with the specified tag value + binding_name = None + request = resourcemanager_v3.ListTagBindingsRequest(parent=parent) + tag_bindings = tag_bindings_client.list_tag_bindings(request=request) + + for binding in tag_bindings: + if binding.tag_value == tag_value: + binding_name = binding.name + break + + if binding_name is None: + print(f"Tag binding for value {tag_value} not found on {secret_name}.") + return + + # Delete the tag binding + request = resourcemanager_v3.DeleteTagBindingRequest(name=binding_name) + operation = tag_bindings_client.delete_tag_binding(request=request) + + # Wait for the operation to complete + operation.result() + + print(f"Detached tag value {tag_value} from {secret_name}") + + +# [END secretmanager_detach_regional_tag_binding] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument( + "location_id", help="id of location where secret is stored" + ) + parser.add_argument("secret_id", help="id of the secret") + parser.add_argument( + "tag_value", help="tag value to detach (e.g., tagValues/123456789012)" + ) + args = parser.parse_args() + + detach_regional_tag( + args.project_id, args.location_id, args.secret_id, args.tag_value + ) diff --git a/secretmanager/snippets/regional_samples/list_regional_secret_tag_bindings.py b/secretmanager/snippets/regional_samples/list_regional_secret_tag_bindings.py new file mode 100644 index 00000000000..351d39c94f0 --- /dev/null +++ b/secretmanager/snippets/regional_samples/list_regional_secret_tag_bindings.py @@ -0,0 +1,91 @@ +#!/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 secretmanager_list_regional_secret_tag_bindings] +from google.cloud import resourcemanager_v3 +from google.api_core import client_options + + +def list_regional_secret_tag_bindings( + project_id: str, location_id: str, secret_id: str +) -> None: + """ + Lists tag bindings for a regional secret. + + Args: + project_id (str): ID of the Google Cloud project + location_id (str): Region where the secret should be stored (e.g., "us-central1") + secret_id (str): ID of the secret to create + + Example: + # Create a regional secret with a customer-managed encryption key + list_regional_secret_tag_bindings( + "my-project", + "my-regional-secret-with-cmek", + "us-central1", + ) + """ + # Set up the endpoint for the regional resource manager + rm_endpoint = f"{location_id}-cloudresourcemanager.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=rm_endpoint) + + # Create the Tag Bindings client with the regional endpoint + tag_bindings_client = resourcemanager_v3.TagBindingsClient( + client_options=client_option + ) + + name = f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + + # Format the parent resource for the tag bindings request + parent = f"//secretmanager.googleapis.com/{name}" + + # List the tag bindings + print(f"Tag bindings for {name}:") + count = 0 + + # Use the list_tag_bindings method to get all tag bindings + request = resourcemanager_v3.ListTagBindingsRequest(parent=parent) + tag_bindings = tag_bindings_client.list_tag_bindings(request=request) + + # Iterate through the results + for binding in tag_bindings: + print(f"- Tag Value: {binding.tag_value}") + count += 1 + + if count == 0: + print(f"No tag bindings found for {name}.") + + +# [END secretmanager_list_regional_secret_tag_bindings] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument( + "location_id", help="id of location where secret is stored" + ) + parser.add_argument("secret_id", help="id of the secret in which to list") + args = parser.parse_args() + + list_regional_secret_tag_bindings( + args.project_id, args.location_id, args.secret_id + ) diff --git a/secretmanager/snippets/regional_samples/snippets_test.py b/secretmanager/snippets/regional_samples/snippets_test.py index 436b8d0d11b..e7886d85239 100644 --- a/secretmanager/snippets/regional_samples/snippets_test.py +++ b/secretmanager/snippets/regional_samples/snippets_test.py @@ -11,7 +11,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and -from datetime import timedelta +from datetime import timedelta, datetime import os import time from typing import Iterator, Tuple, Union @@ -58,7 +58,17 @@ from regional_samples import update_regional_secret_with_etag from regional_samples import view_regional_secret_annotations from regional_samples import view_regional_secret_labels - +from regional_samples import create_regional_secret_with_cmek +from regional_samples import create_regional_secret_with_expire_time +from regional_samples import create_regional_secret_with_rotation +from regional_samples import create_regional_secret_with_topic +from regional_samples import delete_regional_secret_expiration +from regional_samples import update_regional_secret_expiration +from regional_samples import list_regional_secret_tag_bindings +from regional_samples import detach_regional_tag +from regional_samples import update_regional_secret_rotation +from regional_samples import delete_regional_secret_rotation +from regional_samples import update_regional_secret_with_alias @pytest.fixture() def location_id() -> str: @@ -102,6 +112,17 @@ def project_id() -> str: def iam_user() -> str: return "serviceAccount:" + os.environ["GCLOUD_SECRETS_SERVICE_ACCOUNT"] +@pytest.fixture() +def topic_name() -> str: + return os.environ["GOOGLE_CLOUD_TOPIC_NAME"] + +@pytest.fixture() +def rotation_period_hours() -> int: + return 24 + +@pytest.fixture() +def kms_key_name() -> str: + return os.environ["GOOGLE_CLOUD_REGIONAL_KMS_KEY_NAME"] @pytest.fixture() def ttl() -> str: @@ -858,3 +879,219 @@ def test_view_regional_secret_labels( out, _ = capsys.readouterr() assert label_key in out + +def test_create_regional_secret_with_cmek(capsys, project_id: str, secret_id: str, location_id: str, kms_key_name: str): + + create_regional_secret_with_cmek.create_regional_secret_with_cmek(project_id, secret_id, location_id, kms_key_name) + + # Check the output contains expected text + out, _ = capsys.readouterr() + assert "Created secret" in out + assert secret_id in out + assert kms_key_name in out + + retrieved_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + + # Check that the CMEK key name matches what we specified + actual_key_name = retrieved_secret.customer_managed_encryption.kms_key_name + assert actual_key_name == kms_key_name, f"CMEK key name mismatch: got {actual_key_name}, want {kms_key_name}" + +def test_create_regional_secret_with_expire_time(project_id: str, secret_id: str, location_id: str) -> None: + # Set expire time to 1 hour from now + expire_time = datetime.now() + timedelta(hours=1) + create_regional_secret_with_expire_time.create_regional_secret_with_expire_time(project_id, secret_id, location_id) + + retrieved_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + # Verify the secret has an expiration time + assert retrieved_secret.expire_time is not None, "ExpireTime is None, expected non-None" + retrieved_expire_time = retrieved_secret.expire_time.replace(tzinfo=None) + retrieved_expire_time = int(retrieved_expire_time.timestamp()) + + # Convert expected datetime to seconds + expire_time = int(expire_time.timestamp()) + + time_diff = abs(retrieved_expire_time - expire_time) + assert time_diff <= 1, ( + f"ExpireTime difference too large: {time_diff} seconds. " + ) + +def test_create_regional_secret_with_rotation(capsys: pytest.LogCaptureFixture, project_id: str, secret_id: str, location_id: str, topic_name: str, rotation_period_hours: int) -> None: + + # Create the secret with rotation + create_regional_secret_with_rotation.create_regional_secret_with_rotation(project_id, secret_id, location_id, topic_name) + + # Verify output contains expected message + out, _ = capsys.readouterr() + assert "Created secret" in out, f"Expected 'Created secret' in output, got: {out}" + + retrieved_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + + # Verify rotation is configured + assert retrieved_secret.rotation is not None, "Rotation is None, expected non-None" + + # Verify rotation period is set correctly (24 hours = 86400 seconds) + expected_seconds = rotation_period_hours * 3600 + actual_seconds = retrieved_secret.rotation.rotation_period.total_seconds() + assert actual_seconds == expected_seconds, f"RotationPeriod mismatch: got {actual_seconds}, want {expected_seconds}" + + # Verify next rotation time is set + assert retrieved_secret.rotation.next_rotation_time is not None, "NextRotationTime is None, expected non-None" + +def test_update_regional_secret_rotation_period( + capsys: pytest.LogCaptureFixture, + project_id: str, + secret_id: str, + location_id: str, + topic_name: str +) -> None: + + create_regional_secret_with_rotation.create_regional_secret_with_rotation(project_id, secret_id, location_id, topic_name) + capsys.readouterr() + + updated_rotation_hours = 48 + update_regional_secret_rotation.update_regional_secret_rotation_period(project_id, secret_id, location_id) + + # Verify output contains the secret ID + out, _ = capsys.readouterr() + assert secret_id in out, f"Expected '{secret_id}' in output, got: {out}" + + retrieved_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + assert retrieved_secret.rotation is not None, "GetSecret: Rotation is nil, expected non-nil" + expected_seconds = updated_rotation_hours * 3600 + actual_seconds = retrieved_secret.rotation.rotation_period.total_seconds() + assert actual_seconds == expected_seconds, f"RotationPeriod mismatch: got {actual_seconds}, want {expected_seconds}" + + +def test_create_regional_secret_with_topic(capsys, project_id: str, secret_id: str, location_id: str, topic_name: str): + + # Call the function being tested + create_regional_secret_with_topic.create_regional_secret_with_topic(project_id, secret_id, location_id, topic_name) + + # Check the output contains expected text + out, _ = capsys.readouterr() + assert "Created secret" in out + + retrived_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + + assert len(retrived_secret.topics) == 1, f"Expected 1 topic, got {len(retrived_secret.topics)}" + assert retrived_secret.topics[0].name == topic_name, f"Topic mismatch: got {retrived_secret.topics[0].name}, want {topic_name}" + +def test_update_regional_secret_expiration( + capsys: pytest.LogCaptureFixture, + project_id: str, + secret_id: str, + location_id: str +) -> None: + create_regional_secret_with_expire_time.create_regional_secret_with_expire_time(project_id, secret_id, location_id) + + # Update expire time to 2 hours + new_expire = datetime.now() + timedelta(hours=2) + update_regional_secret_expiration.update_regional_secret_expiration(project_id, secret_id, location_id) + + # Verify output contains expected message + out, _ = capsys.readouterr() + assert "Updated secret" in out + + retrieved_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + assert retrieved_secret.expire_time is not None, "ExpireTime is None, expected non-None" + retrieved_expire_time = retrieved_secret.expire_time.replace(tzinfo=None) + retrieved_expire_time = int(retrieved_expire_time.timestamp()) + + new_expire = int(new_expire.timestamp()) + time_diff = abs(retrieved_expire_time - new_expire) + assert time_diff <= 1, ( + f"ExpireTime difference too large: {time_diff} seconds. " + ) + +def test_delete_regional_secret_expiration(capsys: pytest.LogCaptureFixture, project_id: str, secret_id: str, location_id: str) -> None: + + create_regional_secret_with_expire_time.create_regional_secret_with_expire_time(project_id, secret_id, location_id) + + delete_regional_secret_expiration.delete_regional_secret_expiration(project_id, secret_id, location_id) + out, _ = capsys.readouterr() + assert "Removed expiration" in out + + # Verify expire time is removed with GetSecret + retrieved_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + assert retrieved_secret.expire_time is None, f"ExpireTime is {retrieved_secret.expire_time}, expected None" + +def test_list_regional_secret_tag_bindings(capsys: pytest.LogCaptureFixture, project_id: str, location_id: str, tag_key_and_tag_value: Tuple[str, str], secret_id: str) -> None: + tag_key, tag_value = tag_key_and_tag_value + create_regional_secret_with_tags.create_regional_secret_with_tags( + project_id, location_id, secret_id, tag_key, tag_value + ) + + # Call the function being tested + list_regional_secret_tag_bindings.list_regional_secret_tag_bindings(project_id, location_id, secret_id) + + # Verify the tag value is in the returned bindings + out, _ = capsys.readouterr() + assert secret_id in out + assert tag_value in out + +def test_detach_regional_tag( + capsys: pytest.LogCaptureFixture, + project_id: str, + location_id: str, + tag_key_and_tag_value: Tuple[str, str], + secret_id: str +) -> None: + """Test detaching a tag from a regional secret.""" + tag_key, tag_value = tag_key_and_tag_value + + # Create a secret and bind the tag to it for testing detach + create_regional_secret_with_tags.create_regional_secret_with_tags( + project_id, location_id, secret_id, tag_key, tag_value + ) + + # Call the function being tested - detach the tag + detach_regional_tag.detach_regional_tag( + project_id, location_id, secret_id, tag_value + ) + + # Verify the output contains the expected message + out, _ = capsys.readouterr() + assert "Detached tag value" in out + + # List the tags to verify the tag was detached + list_regional_secret_tag_bindings.list_regional_secret_tag_bindings( + project_id, location_id, secret_id + ) + + # Verify the tag value is no longer in the returned bindings + out, _ = capsys.readouterr() + assert tag_value not in out + +def test_delete_regional_secret_rotation( + capsys: pytest.LogCaptureFixture, + project_id: str, + secret_id: str, + location_id: str, + topic_name: str +) -> None: + # First create a secret with rotation configuration + create_regional_secret_with_rotation.create_regional_secret_with_rotation( + project_id, secret_id, location_id, topic_name + ) + + # Call the function to delete the rotation configuration + delete_regional_secret_rotation.delete_regional_secret_rotation( + project_id, secret_id, location_id + ) + + # Check the output contains the expected message + out, _ = capsys.readouterr() + assert "Removed rotation" in out + assert secret_id in out + + # Verify rotation is removed with GetSecret + retrieved_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + + # Check that rotation configuration is removed + assert not retrieved_secret.rotation, f"Rotation is {repr(retrieved_secret.rotation)}, expected None or empty" + +def test_update_regional_secret_with_alias(project_id: str, location_id: str, regional_secret_version: Tuple[str, str, str]) -> None: + secret_id, _, _ = regional_secret_version + update_regional_secret_with_alias.update_regional_secret_with_alias(project_id, secret_id, location_id) + retrieved_secret = get_regional_secret.get_regional_secret(project_id, location_id, secret_id) + assert retrieved_secret.version_aliases["test"] == 1 \ No newline at end of file diff --git a/secretmanager/snippets/regional_samples/update_regional_secret_expiration.py b/secretmanager/snippets/regional_samples/update_regional_secret_expiration.py new file mode 100644 index 00000000000..071058ab980 --- /dev/null +++ b/secretmanager/snippets/regional_samples/update_regional_secret_expiration.py @@ -0,0 +1,98 @@ +#!/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 secretmanager_update_regional_secret_expiration] +from datetime import datetime, timedelta +from google.cloud import secretmanager +from google.api_core import client_options +from google.protobuf import field_mask_pb2, timestamp_pb2 + + +def update_regional_secret_expiration( + project_id: str, secret_id: str, location_id: str +) -> None: + """ + Updates the expiration time of a regional secret. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret + location_id (str): Region where the secret is stored (e.g., "us-central1") + + Example: + # Update expiration time of a regional secret + update_regional_secret_expiration( + "my-project", + "my-secret", + "us-central1" + ) + """ + # Construct the secret name from the component parts + secret_name = ( + f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + ) + + # Set new expiration time to 2 hours from now + new_expire = datetime.now() + timedelta(hours=2) + + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Convert the Python datetime to a Protobuf Timestamp + timestamp = timestamp_pb2.Timestamp() + timestamp.FromDatetime(new_expire) + + # Create a field mask to update only the expire_time field + update_mask = field_mask_pb2.FieldMask(paths=["expire_time"]) + + # Update the secret with the new expiration time + secret = client.update_secret( + request={ + "secret": {"name": secret_name, "expire_time": timestamp}, + "update_mask": update_mask, + } + ) + + print(f"Updated secret {secret.name} expiration time to {new_expire}") + + +# [END secretmanager_update_regional_secret_expiration] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret") + parser.add_argument( + "location_id", + help="region where the secret is stored (e.g., us-central1)", + ) + args = parser.parse_args() + + update_regional_secret_expiration( + args.project_id, args.secret_id, args.location_id + ) diff --git a/secretmanager/snippets/regional_samples/update_regional_secret_rotation.py b/secretmanager/snippets/regional_samples/update_regional_secret_rotation.py new file mode 100644 index 00000000000..b5136fee7e7 --- /dev/null +++ b/secretmanager/snippets/regional_samples/update_regional_secret_rotation.py @@ -0,0 +1,109 @@ +#!/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 argparse + +# [START secretmanager_update_regional_secret_rotation_period] +from datetime import timedelta +from google.cloud import secretmanager +from google.api_core import client_options +from google.protobuf import field_mask_pb2, duration_pb2 + + +def update_regional_secret_rotation_period( + project_id: str, secret_id: str, location_id: str +) -> None: + """ + Updates the rotation period of a regional secret. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret + location_id (str): Region where the secret is stored (e.g., "us-central1") + + Example: + # Update rotation period of a regional secret + update_regional_secret_rotation_period( + "my-project", + "my-secret", + "us-central1" + ) + """ + # Construct the secret name from the component parts + secret_name = ( + f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + ) + + # Set updated rotation period to 48 hours + new_rotation_period_hours = 48 + + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Convert the Python timedelta to a Protobuf Duration + rotation_period = duration_pb2.Duration() + rotation_period.seconds = ( + new_rotation_period_hours * 3600 + ) # Convert hours to seconds + + # Create a field mask to update only the rotation_period field + update_mask = field_mask_pb2.FieldMask(paths=["rotation.rotation_period"]) + + # Update the secret with the new rotation period + result = client.update_secret( + request={ + "secret": { + "name": secret_name, + "rotation": {"rotation_period": rotation_period}, + }, + "update_mask": update_mask, + } + ) + + # Get the rotation period in hours for display + rotation_hours = result.rotation.rotation_period.seconds / 3600 + + print( + f"Updated secret {result.name} rotation period to {rotation_hours} hours" + ) + + +# [END secretmanager_update_regional_secret_rotation_period] + + +if __name__ == "__main__": + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret") + parser.add_argument( + "location_id", + help="region where the secret is stored (e.g., us-central1)", + ) + args = parser.parse_args() + + update_regional_secret_rotation_period( + args.project_id, args.secret_id, args.location_id + ) diff --git a/secretmanager/snippets/regional_samples/update_regional_secret_with_alias.py b/secretmanager/snippets/regional_samples/update_regional_secret_with_alias.py new file mode 100644 index 00000000000..282c7264d07 --- /dev/null +++ b/secretmanager/snippets/regional_samples/update_regional_secret_with_alias.py @@ -0,0 +1,90 @@ +#!/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 argparse + +# [START secretmanager_update_regional_secret_with_alias] +from google.cloud import secretmanager +from google.api_core import client_options +from google.protobuf import field_mask_pb2 + + +def update_regional_secret_with_alias( + project_id: str, secret_id: str, location_id: str +) -> None: + """ + Updates the alias map on an existing regional secret. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret + location_id (str): Region where the secret is stored (e.g., "us-central1") + + Example: + # Update a regional secret with an alias + update_regional_secret_with_alias( + "my-project", + "my-secret", + "us-central1" + ) + """ + # Construct the secret name from the component parts + secret_name = ( + f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + ) + + # Set up the endpoint for the specific region + endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + client_option = client_options.ClientOptions(api_endpoint=endpoint) + + # Create the Secret Manager client with the regional endpoint + client = secretmanager.SecretManagerServiceClient( + client_options=client_option + ) + + # Create a field mask to update only the version_aliases field + update_mask = field_mask_pb2.FieldMask(paths=["version_aliases"]) + + # Update the secret with the new alias map + result = client.update_secret( + request={ + "secret": {"name": secret_name, "version_aliases": {"test": 1}}, + "update_mask": update_mask, + } + ) + + print(f"Updated regional secret: {result.name}") + + +# [END secretmanager_update_regional_secret_with_alias] + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="id of the GCP project") + parser.add_argument("secret_id", help="id of the secret") + parser.add_argument( + "location_id", + help="region where the secret is stored (e.g., us-central1)", + ) + args = parser.parse_args() + + update_regional_secret_with_alias( + args.project_id, args.secret_id, args.location_id + ) diff --git a/secretmanager/snippets/snippets_test.py b/secretmanager/snippets/snippets_test.py index dbcdde921a2..ff53970482b 100644 --- a/secretmanager/snippets/snippets_test.py +++ b/secretmanager/snippets/snippets_test.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and import base64 -from datetime import timedelta +from datetime import timedelta, datetime import os import time from typing import Iterator, Optional, Tuple, Union @@ -63,7 +63,16 @@ from update_secret_with_etag import update_secret_with_etag from view_secret_annotations import view_secret_annotations from view_secret_labels import view_secret_labels - +from list_tag_bindings import list_tag_bindings +from detach_tag_binding import detach_tag +from create_secret_with_expiration import create_secret_with_expiration +from update_secret_expiration import update_secret_expiration +from delete_secret_expiration import delete_secret_expiration +from create_secret_with_rotation import create_secret_with_rotation +from update_secret_rotation import update_secret_rotation +from delete_secret_rotation import delete_secret_rotation +from create_secret_with_topic import create_secret_with_topic +from create_secret_with_cmek import create_secret_with_cmek @pytest.fixture() def client() -> secretmanager.SecretManagerServiceClient: @@ -84,6 +93,17 @@ def tag_values_client() -> resourcemanager_v3.TagValuesClient: def project_id() -> str: return os.environ["GOOGLE_CLOUD_PROJECT"] +@pytest.fixture() +def topic_name() -> str: + return os.environ["GOOGLE_CLOUD_TOPIC_NAME"] + +@pytest.fixture() +def rotation_period_hours() -> int: + return 24 + +@pytest.fixture() +def kms_key_name() -> str: + return os.environ["GOOGLE_CLOUD_KMS_KEY_NAME"] @pytest.fixture() def iam_user() -> str: @@ -419,6 +439,54 @@ def test_bind_tags_to_secret( assert secret_id in tag_resp.parent assert tag_value in tag_resp.tag_value +def test_list_tag_bindings( + capsys: pytest.LogCaptureFixture, + project_id: str, + tag_key_and_tag_value: Tuple[str, str], + secret_id: str, +) -> None: + # Get the tag value from the fixture + _, tag_value = tag_key_and_tag_value + + # Create the secret and bind tag (using existing fixtures) + bind_tags_to_secret(project_id, secret_id, tag_value) + + # Call the function being tested + list_tag_bindings(project_id, secret_id) + + # Verify the tag value is in the returned bindings + out, _ = capsys.readouterr() + assert secret_id in out + assert tag_value in out + +def test_detach_tag( + project_id: str, + tag_key_and_tag_value: Tuple[str, str], + secret_id: str, +) -> None: + """Test detaching a tag from a secret.""" + # Get the tag value from the fixture + _, tag_value = tag_key_and_tag_value + + # First bind the tag to the secret + bind_tags_to_secret(project_id, secret_id, tag_value) + secret_name = f"projects/{project_id}/secrets/{secret_id}" + + # Now detach the tag + detach_tag(project_id, secret_id, tag_value) + + client = resourcemanager_v3.TagBindingsClient() + parent = f"//secretmanager.googleapis.com/{secret_name}" + request = resourcemanager_v3.ListTagBindingsRequest(parent=parent) + + # Check that none of the bindings contain our tag value + tag_found = False + for binding in client.list_tag_bindings(request=request): + if binding.tag_value == tag_value: + tag_found = True + break + + assert not tag_found, f"Tag value {tag_value} should have been detached but was found" def test_create_secret_without_ttl( project_id: str, @@ -745,3 +813,153 @@ def test_update_secret_with_delayed_destroy(secret_with_delayed_destroy: Tuple[s updated_version_destroy_ttl_value = 118400 updated_secret = update_secret_with_delayed_destroy(project_id, secret_id, updated_version_destroy_ttl_value) assert updated_secret.version_destroy_ttl == timedelta(seconds=updated_version_destroy_ttl_value) + +def test_create_secret_with_expiration(project_id: str, secret_id: str) -> None: + """Test creating a secret with an expiration time.""" + + # Set expire time to 1 hour from now + expire_time = datetime.now() + timedelta(hours=1) + create_secret_with_expiration(project_id, secret_id) + + retrieved_secret = get_secret(project_id, secret_id) + # Verify the secret has an expiration time + assert retrieved_secret.expire_time is not None, "ExpireTime is None, expected non-None" + retrieved_expire_time = retrieved_secret.expire_time.replace(tzinfo=None) + retrieved_expire_time = int(retrieved_expire_time.timestamp()) + + # Convert expected datetime to seconds + expire_time = int(expire_time.timestamp()) + + time_diff = abs(retrieved_expire_time - expire_time) + assert time_diff <= 1, ( + f"ExpireTime difference too large: {time_diff} seconds. " + ) + +def test_update_secret_expiration( + capsys: pytest.LogCaptureFixture, + project_id: str, + secret_id: str, +) -> None: + create_secret_with_expiration(project_id, secret_id) + + # Update expire time to 2 hours + new_expire = datetime.now() + timedelta(hours=2) # 2 hours from now in seconds + update_secret_expiration(project_id, secret_id) + + # Verify output contains expected message + out, _ = capsys.readouterr() + assert "Updated secret" in out + + retrieved_secret = get_secret(project_id, secret_id) + assert retrieved_secret.expire_time is not None, "ExpireTime is None, expected non-None" + retrieved_expire_time = retrieved_secret.expire_time.replace(tzinfo=None) + retrieved_expire_time = int(retrieved_expire_time.timestamp()) + + new_expire = int(new_expire.timestamp()) + time_diff = abs(retrieved_expire_time - new_expire) + assert time_diff <= 1, ( + f"ExpireTime difference too large: {time_diff} seconds. " + ) + +def test_delete_expiration(capsys: pytest.LogCaptureFixture, project_id: str, secret_id: str) -> None: + + create_secret_with_expiration(project_id, secret_id) + + delete_secret_expiration(project_id, secret_id) + out, _ = capsys.readouterr() + assert "Removed expiration" in out + + # Verify expire time is removed with GetSecret + retrieved_secret = get_secret(project_id, secret_id) + assert retrieved_secret.expire_time is None, f"ExpireTime is {retrieved_secret.expire_time}, expected None" + +def test_create_secret_with_rotation(capsys: pytest.LogCaptureFixture, project_id: str, secret_id: str, topic_name: str, rotation_period_hours: int) -> None: + """Test creating a secret with rotation configuration.""" + + # Create the secret with rotation + create_secret_with_rotation(project_id, secret_id, topic_name) + + # Verify output contains expected message + out, _ = capsys.readouterr() + assert "Created secret" in out, f"Expected 'Created secret' in output, got: {out}" + + retrieved_secret = get_secret(project_id, secret_id) + + # Verify rotation is configured + assert retrieved_secret.rotation is not None, "Rotation is None, expected non-None" + + # Verify rotation period is set correctly (24 hours = 86400 seconds) + expected_seconds = rotation_period_hours * 3600 + actual_seconds = retrieved_secret.rotation.rotation_period.total_seconds() + assert actual_seconds == expected_seconds, f"RotationPeriod mismatch: got {actual_seconds}, want {expected_seconds}" + + # Verify next rotation time is set + assert retrieved_secret.rotation.next_rotation_time is not None, "NextRotationTime is None, expected non-None" + + +def test_update_secret_rotation_period( + capsys: pytest.LogCaptureFixture, + project_id: str, + secret_id: str, + topic_name: str +) -> None: + + create_secret_with_rotation(project_id, secret_id, topic_name) + capsys.readouterr() + + updated_rotation_hours = 48 + update_secret_rotation(project_id, secret_id) + + # Verify output contains the secret ID + out, _ = capsys.readouterr() + assert secret_id in out, f"Expected '{secret_id}' in output, got: {out}" + + retrieved_secret = get_secret(project_id, secret_id) + assert retrieved_secret.rotation is not None, "GetSecret: Rotation is nil, expected non-nil" + expected_seconds = updated_rotation_hours * 3600 + actual_seconds = retrieved_secret.rotation.rotation_period.total_seconds() + assert actual_seconds == expected_seconds, f"RotationPeriod mismatch: got {actual_seconds}, want {expected_seconds}" + +def test_delete_secret_rotation(capsys: pytest.LogCaptureFixture, project_id: str, secret_id: str, topic_name: str) -> None: + + create_secret_with_rotation(project_id, secret_id, topic_name) + + # Delete the rotation + delete_secret_rotation(project_id, secret_id) + out, _ = capsys.readouterr() + assert f"Removed rotation from secret" in out + assert secret_id in out + + retrieved_secret = get_secret(project_id, secret_id) + assert not retrieved_secret.rotation, f"Rotation is {repr(retrieved_secret.rotation)}, expected None or empty" + +def test_create_secret_with_topic(capsys, project_id: str, secret_id: str, topic_name: str): + + # Call the function being tested + create_secret_with_topic(project_id, secret_id, topic_name) + + # Check the output contains expected text + out, _ = capsys.readouterr() + assert "Created secret" in out + + retrived_secret = get_secret(project_id, secret_id) + + assert len(retrived_secret.topics) == 1, f"Expected 1 topic, got {len(retrived_secret.topics)}" + assert retrived_secret.topics[0].name == topic_name, f"Topic mismatch: got {retrived_secret.topics[0].name}, want {topic_name}" + +def test_create_secret_with_cmek(capsys, project_id: str, secret_id: str, kms_key_name: str): + + create_secret_with_cmek(project_id, secret_id, kms_key_name) + + # Check the output contains expected text + out, _ = capsys.readouterr() + assert "Created secret" in out + assert secret_id in out + assert kms_key_name in out + + # Verify CMEK key with GetSecret + retrieved_secret = get_secret(project_id, secret_id) + + # Check that the CMEK key name matches what we specified + actual_key_name = retrieved_secret.replication.automatic.customer_managed_encryption.kms_key_name + assert actual_key_name == kms_key_name, f"CMEK key name mismatch: got {actual_key_name}, want {kms_key_name}" diff --git a/secretmanager/snippets/update_secret_expiration.py b/secretmanager/snippets/update_secret_expiration.py new file mode 100644 index 00000000000..166674bc922 --- /dev/null +++ b/secretmanager/snippets/update_secret_expiration.py @@ -0,0 +1,75 @@ +#!/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 argparse + +# [START secretmanager_update_secret_expiration] +from datetime import datetime, timedelta +from google.protobuf import timestamp_pb2 +from google.cloud import secretmanager + + +def update_secret_expiration(project_id: str, secret_id: str) -> None: + """ + Update the expiration time of an existing secret. + + Args: + project_id: ID of the Google Cloud project. + secret_id: ID of the secret to update. + + Example: + # Update the expiration time of a secret to 48 hours from now + update_secret_expiration( + "my-project", + "my-secret-with-expiration" + ) + """ + new_expire_time = datetime.now() + timedelta(hours=2) + + # Create the Secret Manager client. + client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the secret. + name = client.secret_path(project_id, secret_id) + + # Update the expire_time. + timestamp = timestamp_pb2.Timestamp() + timestamp.FromDatetime(new_expire_time) + secret = {"name": name, "expire_time": timestamp} + update_mask = {"paths": ["expire_time"]} + response = client.update_secret( + request={"secret": secret, "update_mask": update_mask} + ) + + # Print the updated secret name. + print( + f"Updated secret {response.name} expiration time to {new_expire_time}" + ) + + +# [END secretmanager_update_secret_expiration] + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="ID of the GCP project") + parser.add_argument("secret_id", help="ID of the secret to update") + args = parser.parse_args() + + update_secret_expiration(args.project_id, args.secret_id) diff --git a/secretmanager/snippets/update_secret_rotation.py b/secretmanager/snippets/update_secret_rotation.py new file mode 100644 index 00000000000..bd546920769 --- /dev/null +++ b/secretmanager/snippets/update_secret_rotation.py @@ -0,0 +1,90 @@ +#!/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 secretmanager_update_secret_rotation] +from google.cloud import secretmanager +from google.protobuf import duration_pb2 +from google.protobuf.field_mask_pb2 import FieldMask + + +def update_secret_rotation(project_id: str, secret_id: str) -> None: + """ + Updates the rotation period of a secret. + + Args: + project_id (str): ID of the Google Cloud project + secret_id (str): ID of the secret to update + rotation_period_hours (int): New rotation period in hours + Example: + # Update the rotation period of a secret to 60 days + update_secret_rotation( + "my-project", + "my-secret-with-rotation", + "projects/my-project/topics/my-rotation-topic" + ) + """ + new_rotation_period_hours = 48 + # Create the Secret Manager client + client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the secret + name = client.secret_path(project_id, secret_id) + + # Convert rotation period to protobuf Duration + rotation_period = duration_pb2.Duration() + rotation_period.seconds = ( + new_rotation_period_hours * 3600 + ) # Convert hours to seconds + + # Create the update mask + update_mask = FieldMask(paths=["rotation.rotation_period"]) + + # Build the request + request = { + "secret": { + "name": name, + "rotation": {"rotation_period": rotation_period}, + }, + "update_mask": update_mask, + } + + # Update the secret + result = client.update_secret(request=request) + + rotation_hours = result.rotation.rotation_period.seconds / 3600 + print( + f"Updated secret {result.name} rotation period to {rotation_hours} hours" + ) + + +# [END secretmanager_update_secret_rotation] + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("project_id", help="ID of the GCP project") + parser.add_argument("secret_id", help="ID of the secret to update") + args = parser.parse_args() + + update_secret_rotation( + args.project_id, + args.secret_id, + )