diff --git a/secretmanager/snippets/create_secret_with_cmek.py b/secretmanager/snippets/create_secret_with_cmek.py new file mode 100644 index 00000000000..56bf73e4523 --- /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_v1 + + +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_v1.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..e80d1bf7b83 --- /dev/null +++ b/secretmanager/snippets/create_secret_with_expiration.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_expiration] +from datetime import datetime, timedelta, timezone + +from google.cloud import secretmanager_v1 +from google.protobuf import timestamp_pb2 + + +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 1 hour + create_secret_with_expiration("my-project", "my-secret-with-expiry") + """ + expire_time = datetime.now(timezone.utc) + timedelta(hours=1) + # Create the Secret Manager client. + client = secretmanager_v1.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..a189a2c8443 --- /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, timezone + +from google.cloud import secretmanager_v1 +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 24 hours + create_secret_with_rotation( + "my-project", + "my-rotating-secret", + "projects/my-project/topics/my-rotation-topic" + ) + """ + rotation_period_hours = 24 + next_rotation_time = datetime.now(timezone.utc) + timedelta(hours=24) + # Create the Secret Manager client + client = secretmanager_v1.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_timestamp = timestamp_pb2.Timestamp() + next_rotation_timestamp.FromDatetime(next_rotation_time) + + # 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_timestamp, + "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..2db1eb47851 --- /dev/null +++ b/secretmanager/snippets/create_secret_with_topic.py @@ -0,0 +1,76 @@ +#!/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_v1 + + +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_v1.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..c3f3f8bbf60 --- /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_v1 +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_v1.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..6bc7106f007 --- /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_v1 +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_v1.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..78eee7b5524 --- /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. + + +# [START secretmanager_detach_tag_binding] +from google.cloud import resourcemanager_v3 +from google.cloud import secretmanager_v1 + + +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. + rm_client = resourcemanager_v3.TagBindingsClient() + + # Build the resource name of the parent secret. + client = secretmanager_v1.SecretManagerServiceClient() + secret_name = client.secret_path(project_id, 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 rm_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 = rm_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__": + 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 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..240d626a830 --- /dev/null +++ b/secretmanager/snippets/list_tag_bindings.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 +""" +command line application and sample code for listing tag bindings attached to a secret. +""" + + +# [START secretmanager_list_tag_bindings] +# Import the Resource Manager client library. +from google.cloud import resourcemanager_v3 +from google.cloud import secretmanager_v1 + + +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_v1.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. + request = resourcemanager_v3.ListTagBindingsRequest(parent=parent) + + # Retrieve and process tag bindings + bindings = client.list_tag_bindings(request=request) + found_bindings = False + + print(f"Tag bindings for {secret_name}:") + for binding in bindings: + print(f"- Tag Value: {binding.tag_value}") + found_bindings = True + + if not found_bindings: + print(f"No tag bindings found for {secret_name}.") + + +# [END secretmanager_list_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("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..9093f546beb --- /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. + + +# [START secretmanager_create_regional_secret_with_cmek] +from google.api_core import client_options +from google.cloud import secretmanager_v1 + + +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_v1.SecretManagerServiceClient( + client_options=client_option + ) + + # Build the resource name of the parent project with location + parent = client.common_location_path(project_id, 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__": + 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( + "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..e00eef48b12 --- /dev/null +++ b/secretmanager/snippets/regional_samples/create_regional_secret_with_expire_time.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_create_regional_secret_with_expire_time] +from datetime import datetime, timedelta, timezone + +from google.api_core import client_options +from google.cloud import secretmanager_v1 +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(timezone.utc) + 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_v1.SecretManagerServiceClient( + client_options=client_option + ) + + # Build the resource name of the parent project with location + parent = client.common_location_path(project_id, 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..70c181762ba --- /dev/null +++ b/secretmanager/snippets/regional_samples/create_regional_secret_with_rotation.py @@ -0,0 +1,117 @@ +#!/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, timezone + +from google.api_core import client_options +from google.cloud import secretmanager_v1 +from google.protobuf import duration_pb2, timestamp_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(timezone.utc) + 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_v1.SecretManagerServiceClient( + client_options=client_option + ) + + # Build the resource name of the parent project with location + parent = client.common_location_path(project_id, 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..bc802f854c7 --- /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.api_core import client_options +from google.cloud import secretmanager_v1 + + +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_v1.SecretManagerServiceClient( + client_options=client_option + ) + + # Build the resource name of the parent project with location + parent = client.common_location_path(project_id, 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..eff41569104 --- /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. + + +# [START secretmanager_delete_regional_secret_expiration] +from google.api_core import client_options +from google.cloud import secretmanager_v1 +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_v1.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__": + 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() + + 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..d3833b09f96 --- /dev/null +++ b/secretmanager/snippets/regional_samples/delete_regional_secret_rotation.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. + + +# [START secretmanager_delete_regional_secret_rotation] +from google.api_core import client_options +from google.cloud import secretmanager_v1 +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_v1.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__": + 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() + + 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..e2b0b344a2b --- /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.api_core import client_options +from google.cloud import resourcemanager_v3 + + +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..ee50eec70cd --- /dev/null +++ b/secretmanager/snippets/regional_samples/list_regional_secret_tag_bindings.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_list_regional_secret_tag_bindings] +from google.api_core import client_options +from google.cloud import resourcemanager_v3 + + +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 is stored (e.g., "us-central1") + secret_id (str): The ID of the secret to list tag bindings for. + + Example: + list_regional_secret_tag_bindings( + "my-project", + "us-central1", + "my-regional-secret-with-cmek", + ) + """ + # 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}:") + found_bindings = False + + # 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}") + found_bindings = True + + if not found_bindings: + 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..89d740560d3 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 datetime, timedelta, timezone import os import time from typing import Iterator, Tuple, Union @@ -28,15 +28,22 @@ from regional_samples import bind_tags_to_regional_secret from regional_samples import create_regional_secret from regional_samples import create_regional_secret_with_annotations +from regional_samples import create_regional_secret_with_cmek from regional_samples import create_regional_secret_with_delayed_destroy +from regional_samples import create_regional_secret_with_expire_time from regional_samples import create_regional_secret_with_labels +from regional_samples import create_regional_secret_with_rotation from regional_samples import create_regional_secret_with_tags +from regional_samples import create_regional_secret_with_topic from regional_samples import delete_regional_secret from regional_samples import delete_regional_secret_annotation +from regional_samples import delete_regional_secret_expiration from regional_samples import delete_regional_secret_label +from regional_samples import delete_regional_secret_rotation from regional_samples import delete_regional_secret_with_etag from regional_samples import destroy_regional_secret_version from regional_samples import destroy_regional_secret_version_with_etag +from regional_samples import detach_regional_tag from regional_samples import disable_regional_secret_delayed_destroy from regional_samples import disable_regional_secret_version from regional_samples import disable_regional_secret_version_with_etag @@ -48,12 +55,16 @@ from regional_samples import get_regional_secret_version from regional_samples import iam_grant_access_with_regional_secret from regional_samples import iam_revoke_access_with_regional_secret +from regional_samples import list_regional_secret_tag_bindings from regional_samples import list_regional_secret_versions from regional_samples import list_regional_secret_versions_with_filter from regional_samples import list_regional_secrets from regional_samples import list_regional_secrets_with_filter from regional_samples import regional_quickstart from regional_samples import update_regional_secret +from regional_samples import update_regional_secret_expiration +from regional_samples import update_regional_secret_rotation +from regional_samples import update_regional_secret_with_alias from regional_samples import update_regional_secret_with_delayed_destroy from regional_samples import update_regional_secret_with_etag from regional_samples import view_regional_secret_annotations @@ -103,6 +114,21 @@ 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 kms_key_name() -> str: + return os.environ["GOOGLE_CLOUD_REGIONAL_KMS_KEY_NAME"] + + +@pytest.fixture() +def rotation_period_hours() -> int: + return 24 + + @pytest.fixture() def ttl() -> str: return "300s" @@ -858,3 +884,336 @@ def test_view_regional_secret_labels( out, _ = capsys.readouterr() assert label_key in out + + +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: + 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_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(timezone.utc) + 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.astimezone(timezone.utc) + retrieved_timestamp = int(retrieved_expire_time.timestamp()) + + # Convert expected datetime to seconds + + expire_time = int(expire_time.timestamp()) + + time_diff = abs(retrieved_timestamp - expire_time) + assert time_diff <= 1, f"ExpireTime difference too large: {time_diff} seconds. " + + +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(timezone.utc) + 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.astimezone(timezone.utc) + retrieved_timestamp = int(retrieved_expire_time.timestamp()) + + new_expire = int(new_expire.timestamp()) + time_diff = abs(retrieved_timestamp - 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_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_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 ( + retrieved_secret.rotation == secretmanager_v1.types.Rotation() + ), f"Rotation is {repr(retrieved_secret.rotation)}, expected None or empty" + + +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 + + retrieved_secret = get_regional_secret.get_regional_secret( + project_id, location_id, secret_id + ) + + assert ( + len(retrieved_secret.topics) == 1 + ), f"Expected 1 topic, got {len(retrieved_secret.topics)}" + assert ( + retrieved_secret.topics[0].name == topic_name + ), f"Topic mismatch: got {retrieved_secret.topics[0].name}, want {topic_name}" + + +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_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 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..6f137064a8b --- /dev/null +++ b/secretmanager/snippets/regional_samples/update_regional_secret_expiration.py @@ -0,0 +1,99 @@ +#!/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, timezone + +from google.api_core import client_options +from google.cloud import secretmanager_v1 +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(timezone.utc) + 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_v1.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..662fa2ac21c --- /dev/null +++ b/secretmanager/snippets/regional_samples/update_regional_secret_rotation.py @@ -0,0 +1,108 @@ +#!/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_rotation_period] +from google.api_core import client_options +from google.cloud import secretmanager_v1 +from google.protobuf import duration_pb2, field_mask_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_v1.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__": + 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_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..09eb07124fa --- /dev/null +++ b/secretmanager/snippets/regional_samples/update_regional_secret_with_alias.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_update_regional_secret_with_alias] +from google.api_core import client_options +from google.cloud import secretmanager_v1 +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_v1.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__": + 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_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..fcdcaf02f05 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 datetime, timedelta, timezone import os import time from typing import Iterator, Optional, Tuple, Union @@ -20,7 +20,7 @@ from google.api_core import exceptions, retry from google.cloud import resourcemanager_v3 -from google.cloud import secretmanager +from google.cloud import secretmanager, secretmanager_v1 from google.protobuf.duration_pb2 import Duration import pytest @@ -31,17 +31,24 @@ from consume_event_notification import consume_event_notification from create_secret import create_secret from create_secret_with_annotations import create_secret_with_annotations +from create_secret_with_cmek import create_secret_with_cmek from create_secret_with_delayed_destroy import create_secret_with_delayed_destroy +from create_secret_with_expiration import create_secret_with_expiration from create_secret_with_labels import create_secret_with_labels +from create_secret_with_rotation import create_secret_with_rotation from create_secret_with_tags import create_secret_with_tags +from create_secret_with_topic import create_secret_with_topic from create_secret_with_user_managed_replication import create_ummr_secret from create_update_secret_label import create_update_secret_label from delete_secret import delete_secret from delete_secret_annotation import delete_secret_annotation +from delete_secret_expiration import delete_secret_expiration from delete_secret_label import delete_secret_label +from delete_secret_rotation import delete_secret_rotation from delete_secret_with_etag import delete_secret_with_etag from destroy_secret_version import destroy_secret_version from destroy_secret_version_with_etag import destroy_secret_version_with_etag +from detach_tag_binding import detach_tag from disable_secret_version import disable_secret_version from disable_secret_version_with_etag import disable_secret_version_with_etag from disable_secret_with_delayed_destroy import disable_secret_with_delayed_destroy @@ -56,8 +63,11 @@ from list_secret_versions_with_filter import list_secret_versions_with_filter from list_secrets import list_secrets from list_secrets_with_filter import list_secrets_with_filter +from list_tag_bindings import list_tag_bindings from quickstart import quickstart from update_secret import update_secret +from update_secret_expiration import update_secret_expiration +from update_secret_rotation import update_secret_rotation from update_secret_with_alias import update_secret_with_alias from update_secret_with_delayed_destroy import update_secret_with_delayed_destroy from update_secret_with_etag import update_secret_with_etag @@ -90,6 +100,21 @@ 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_KMS_KEY_NAME"] + + @pytest.fixture() def ttl() -> Optional[str]: return "300s" @@ -745,3 +770,265 @@ 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_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( + capsys: pytest.LogCaptureFixture, + 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) + + # Now detach the tag + + detach_tag(project_id, secret_id, tag_value) + + out, _ = capsys.readouterr() + assert "Detached tag value" in out + + list_tag_bindings(project_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_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(timezone.utc) + 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.astimezone(timezone.utc) + + retrieved_timestamp = int(retrieved_expire_time.timestamp()) + expected_timestamp = int(expire_time.timestamp()) + + time_diff = abs(retrieved_timestamp - expected_timestamp) + 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(timezone.utc) + 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.astimezone(timezone.utc) + retrieved_timestamp = int(retrieved_expire_time.timestamp()) + + new_expire = int(new_expire.timestamp()) + time_diff = abs(retrieved_timestamp - 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 "Removed rotation from secret" in out + assert secret_id in out + + retrieved_secret = get_secret(project_id, secret_id) + assert ( + retrieved_secret.rotation == secretmanager_v1.types.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 + + retrieved_secret = get_secret(project_id, secret_id) + + assert ( + len(retrieved_secret.topics) == 1 + ), f"Expected 1 topic, got {len(retrieved_secret.topics)}" + assert ( + retrieved_secret.topics[0].name == topic_name + ), f"Topic mismatch: got {retrieved_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..730c8e06987 --- /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. + + +# [START secretmanager_update_secret_expiration] +from datetime import datetime, timedelta, timezone + +from google.cloud import secretmanager_v1 +from google.protobuf import timestamp_pb2 + + +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 2 hours from now + update_secret_expiration( + "my-project", + "my-secret-with-expiration" + ) + """ + new_expire_time = datetime.now(timezone.utc) + timedelta(hours=2) + + # Create the Secret Manager client. + client = secretmanager_v1.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__": + 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_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..ae2b3cc3208 --- /dev/null +++ b/secretmanager/snippets/update_secret_rotation.py @@ -0,0 +1,86 @@ +#!/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_v1 +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 + Example: + # Update the rotation period of a secret to 60 days + update_secret_rotation( + "my-project", + "my-secret-with-rotation", + ) + """ + new_rotation_period_hours = 48 + # Create the Secret Manager client + client = secretmanager_v1.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, + ) diff --git a/testing/test-env.tmpl.sh b/testing/test-env.tmpl.sh index 1bfacfb2833..938e12352ac 100644 --- a/testing/test-env.tmpl.sh +++ b/testing/test-env.tmpl.sh @@ -80,6 +80,9 @@ export PUBSUB_VERIFICATION_TOKEN=1234abc # Secret Manager Test Vars export GCLOUD_SECRETS_SERVICE_ACCOUNT= +export GOOGLE_CLOUD_TOPIC_NAME= +export GOOGLE_CLOUD_KMS_KEY_NAME= +export GOOGLE_CLOUD_REGIONAL_KMS_KEY_NAME= # Automl # A centralized project is used to remove duplicate work across all 7 languages