diff --git a/secretmanager/snippets/create_secret_with_type.py b/secretmanager/snippets/create_secret_with_type.py new file mode 100644 index 00000000000..5cd19ecd290 --- /dev/null +++ b/secretmanager/snippets/create_secret_with_type.py @@ -0,0 +1,80 @@ +#!/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 creating a new secret with a +secret type restriction. +""" + +# [START secretmanager_create_secret_with_type] +import argparse + +# Import the Secret Manager client library. +from google.cloud import secretmanager + + +def create_secret_with_type( + project_id: str, + secret_id: str, + secret_type: secretmanager.Secret.SecretType, +) -> secretmanager.Secret: + """ + Creates a new secret with the given secret type. + Note: CLOUD_SQL_DB_CREDENTIALS is only supported in the regional secret. + """ + + # 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 the given secret type restriction. + response = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": { + "replication": {"automatic": {}}, + "secret_type": secret_type, + }, + } + ) + + # Print the new secret name. + print(f"Created secret with secret type: {response.name}") + + return response + + +# [END secretmanager_create_secret_with_type] + +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( + "secret_type", + choices=[t.name for t in secretmanager.Secret.SecretType if t.value != 0], + help="secret type restriction to apply", + ) + args = parser.parse_args() + + create_secret_with_type( + args.project_id, + args.secret_id, + secretmanager.Secret.SecretType[args.secret_type], + ) diff --git a/secretmanager/snippets/get_secret_type.py b/secretmanager/snippets/get_secret_type.py new file mode 100644 index 00000000000..4c3b9614c06 --- /dev/null +++ b/secretmanager/snippets/get_secret_type.py @@ -0,0 +1,56 @@ +#!/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 getting the secret type of a +secret. +""" + +# [START secretmanager_get_secret_type] +import argparse + +# Import the Secret Manager client library. +from google.cloud import secretmanager + + +def get_secret_type(project_id: str, secret_id: str) -> secretmanager.Secret: + """ + Gets the secret type of the given secret. + """ + + # Create the Secret Manager client. + client = secretmanager.SecretManagerServiceClient() + + # Build the resource name of the secret. + name = client.secret_path(project_id, secret_id) + + # Get the secret. + response = client.get_secret(request={"name": name}) + + print(f"Found secret {response.name} with secret type {response.secret_type.name}") + + return response + + +# [END secretmanager_get_secret_type] + +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 get the type of") + args = parser.parse_args() + + get_secret_type(args.project_id, args.secret_id) diff --git a/secretmanager/snippets/regional_samples/create_regional_secret_with_cloud_sql_credentials.py b/secretmanager/snippets/regional_samples/create_regional_secret_with_cloud_sql_credentials.py new file mode 100644 index 00000000000..a547a6f9072 --- /dev/null +++ b/secretmanager/snippets/regional_samples/create_regional_secret_with_cloud_sql_credentials.py @@ -0,0 +1,88 @@ +#!/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 creating a new secret with type +CLOUD_SQL_DB_CREDENTIALS, eligible for managed rotation. +""" + +# [START secretmanager_create_regional_secret_with_cloud_sql_credentials] +import argparse + +# Import the Secret Manager client library. +from google.cloud import secretmanager_v1 + + +def create_regional_secret_with_cloud_sql_credentials( + project_id: str, + location_id: str, + secret_id: str, +) -> secretmanager_v1.Secret: + """ + Creates a new regional secret with type CLOUD_SQL_DB_CREDENTIALS. + """ + + # Endpoint to call the regional Secret Manager API. + api_endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + + # Create the Secret Manager client. + client = secretmanager_v1.SecretManagerServiceClient( + client_options={"api_endpoint": api_endpoint}, + ) + + # Build the resource name of the parent project. + parent = f"projects/{project_id}/locations/{location_id}" + + # Create the secret. + response = client.create_secret( + request={ + "parent": parent, + "secret_id": secret_id, + "secret": { + "secret_type": secretmanager_v1.Secret.SecretType.CLOUD_SQL_DB_CREDENTIALS, + }, + } + ) + + # Print the new secret name. + print(f"Created secret: {response.name}") + + # Grant this identity the Cloud SQL User rotate IAM permissions to enable + # managed rotation. + print( + "Grant the Cloud SQL User rotate IAM permissions to enable managed " + f"rotation to: {response.policy_member.iam_policy_uid_principal}" + ) + + return response + + +# [END secretmanager_create_regional_secret_with_cloud_sql_credentials] + +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( + "location_id", + help="id of the location where secret is to be created; must match " + "the Cloud SQL instance's region", + ) + parser.add_argument("secret_id", help="id of the secret to create") + args = parser.parse_args() + + create_regional_secret_with_cloud_sql_credentials( + args.project_id, args.location_id, args.secret_id + ) diff --git a/secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py b/secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py new file mode 100644 index 00000000000..4369cd1259d --- /dev/null +++ b/secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py @@ -0,0 +1,95 @@ +#!/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 to enable managed rotation of a +CLOUD_SQL_DB_CREDENTIALS typed secret. +""" + +# [START secretmanager_enable_regional_secret_managed_rotation] +import argparse + +# Import the Secret Manager client library. +from google.cloud import secretmanager_v1 + + +def enable_regional_secret_managed_rotation( + project_id: str, + location_id: str, + secret_id: str, + instance_id: str, + username: str, +) -> secretmanager_v1.SecretVersion: + """ + Enables managed rotation of a CLOUD_SQL_DB_CREDENTIALS typed secret. + It validates and enables the rotation, adding a version and sets the + passed password (optional). + Note: AddSecretVersion is disabled on the CLOUD_SQL_DB_CREDENTIALS + currently and for any necessary manual rotations please trigger + rotate_secret. + """ + + # Endpoint to call the regional Secret Manager API. + api_endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + + # Create the Secret Manager client. + client = secretmanager_v1.SecretManagerServiceClient( + client_options={"api_endpoint": api_endpoint}, + ) + + # Build the resource name of the secret. + parent = f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + + # Enable managed rotation. + response = client.enable_managed_rotation( + request={ + "parent": parent, + "cloud_sql_single_user_credentials": { + "instance_id": instance_id, + "username": username, + }, + } + ) + + print(f"Enabled managed rotation, created secret version: {response.name}") + + return response + + +# [END secretmanager_enable_regional_secret_managed_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("location_id", help="id of location where secret is stored") + parser.add_argument( + "secret_id", + help="id of the Cloud SQL DB credentials secret to enable rotation on", + ) + parser.add_argument( + "instance_id", + help="bare id of the Cloud SQL instance (no project or region prefix)", + ) + parser.add_argument("username", help="username of the Cloud SQL database user") + args = parser.parse_args() + + enable_regional_secret_managed_rotation( + args.project_id, + args.location_id, + args.secret_id, + args.instance_id, + args.username, + ) diff --git a/secretmanager/snippets/regional_samples/get_regional_secret_type.py b/secretmanager/snippets/regional_samples/get_regional_secret_type.py new file mode 100644 index 00000000000..9554d615655 --- /dev/null +++ b/secretmanager/snippets/regional_samples/get_regional_secret_type.py @@ -0,0 +1,67 @@ +#!/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 getting the secret type of a +regional secret. +""" + +# [START secretmanager_get_regional_secret_type] +import argparse + +# Import the Secret Manager client library. +from google.cloud import secretmanager_v1 + + +def get_regional_secret_type( + project_id: str, location_id: str, secret_id: str +) -> secretmanager_v1.Secret: + """ + Gets the secret type of the given regional secret. + """ + + # Endpoint to call the regional Secret Manager API. + api_endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + + # Create the Secret Manager client. + client = secretmanager_v1.SecretManagerServiceClient( + client_options={"api_endpoint": api_endpoint}, + ) + + # Build the resource name of the secret. + name = f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + + # Get the secret. + response = client.get_secret(request={"name": name}) + + print( + f"Found regional secret {response.name} with secret type " + f"{response.secret_type.name}" + ) + + return response + + +# [END secretmanager_get_regional_secret_type] + +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("location_id", help="id of location where secret is stored") + parser.add_argument("secret_id", help="id of the secret to get the type of") + args = parser.parse_args() + + get_regional_secret_type(args.project_id, args.location_id, args.secret_id) diff --git a/secretmanager/snippets/regional_samples/rotate_regional_secret.py b/secretmanager/snippets/regional_samples/rotate_regional_secret.py new file mode 100644 index 00000000000..cfdd4074b2e --- /dev/null +++ b/secretmanager/snippets/regional_samples/rotate_regional_secret.py @@ -0,0 +1,69 @@ +#!/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 triggering an adhoc rotation for +the managed CLOUD_SQL_DB_CREDENTIALS typed secret. +""" + +# [START secretmanager_rotate_regional_secret] +import argparse + +# Import the Secret Manager client library. +from google.cloud import secretmanager_v1 + + +def rotate_regional_secret( + project_id: str, + location_id: str, + secret_id: str, +) -> secretmanager_v1.SecretVersion: + """ + Triggers an adhoc rotation for the managed CLOUD_SQL_DB_CREDENTIALS + typed secret. + """ + + # Endpoint to call the regional Secret Manager API. + api_endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + + # Create the Secret Manager client. + client = secretmanager_v1.SecretManagerServiceClient( + client_options={"api_endpoint": api_endpoint}, + ) + + # Build the resource name of the secret. + parent = f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + + # Rotate the secret. + response = client.rotate_secret(request={"parent": parent}) + + print(f"Rotated secret, created secret version: {response.name}") + + return response + + +# [END secretmanager_rotate_regional_secret] + +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("location_id", help="id of location where secret is stored") + parser.add_argument( + "secret_id", help="id of the Cloud SQL DB credentials secret to rotate" + ) + args = parser.parse_args() + + rotate_regional_secret(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..f3b13a8bbda 100644 --- a/secretmanager/snippets/regional_samples/snippets_test.py +++ b/secretmanager/snippets/regional_samples/snippets_test.py @@ -20,6 +20,7 @@ from google.api_core import exceptions, retry from google.cloud import resourcemanager_v3 from google.cloud import secretmanager_v1 +from google.iam.v1 import policy_pb2 from google.protobuf.duration_pb2 import Duration import pytest @@ -28,6 +29,7 @@ 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_cloud_sql_credentials from regional_samples import create_regional_secret_with_delayed_destroy from regional_samples import create_regional_secret_with_labels from regional_samples import create_regional_secret_with_tags @@ -42,9 +44,11 @@ from regional_samples import disable_regional_secret_version_with_etag from regional_samples import edit_regional_secret_annotations from regional_samples import edit_regional_secret_label +from regional_samples import enable_regional_secret_managed_rotation from regional_samples import enable_regional_secret_version from regional_samples import enable_regional_secret_version_with_etag from regional_samples import get_regional_secret +from regional_samples import get_regional_secret_type 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 @@ -53,9 +57,11 @@ 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 rotate_regional_secret from regional_samples import update_regional_secret from regional_samples import update_regional_secret_with_delayed_destroy from regional_samples import update_regional_secret_with_etag +from regional_samples import update_regional_secret_with_managed_rotation_schedule from regional_samples import view_regional_secret_annotations from regional_samples import view_regional_secret_labels @@ -93,6 +99,18 @@ def tag_values_client() -> resourcemanager_v3.TagValuesClient: return resourcemanager_v3.TagValuesClient() +@pytest.fixture() +def projects_client() -> resourcemanager_v3.ProjectsClient: + return resourcemanager_v3.ProjectsClient() + + +# Role granted to a Cloud SQL DB credentials secret's built-in identity so +# that managed rotation can update the Cloud SQL user's password. This grant +# is per-secret (the member is the secret's own generated principal), so it +# has to be made fresh for every secret managed_rotation tests create. +CLOUD_SQL_ROLE = "roles/cloudsql.admin" + + @pytest.fixture() def project_id() -> str: return os.environ["GOOGLE_CLOUD_PROJECT"] @@ -103,6 +121,16 @@ def iam_user() -> str: return "serviceAccount:" + os.environ["GCLOUD_SECRETS_SERVICE_ACCOUNT"] +@pytest.fixture() +def cloud_sql_instance_id() -> str: + return os.environ["CLOUD_SQL_INSTANCE"] + + +@pytest.fixture() +def cloud_sql_username() -> str: + return os.environ["CLOUD_SQL_USER"] + + @pytest.fixture() def ttl() -> str: return "300s" @@ -207,6 +235,55 @@ def retry_client_delete_tag_key( return response.name +@retry.Retry(predicate=retry.if_exception_type(exceptions.Aborted)) +def grant_cloud_sql_role( + projects_client: resourcemanager_v3.ProjectsClient, + project_id: str, + member: str, +) -> None: + """ + Grants CLOUD_SQL_ROLE to member on the project. SetIamPolicy replaces + the whole policy, so this reads the current policy, adds the member to + the existing (or a new) binding for the role, and writes it back with + the same etag -- retrying the whole read-modify-write if another writer + raced us (Aborted, from an etag mismatch). + """ + resource = f"projects/{project_id}" + policy = projects_client.get_iam_policy(request={"resource": resource}) + + for binding in policy.bindings: + if binding.role == CLOUD_SQL_ROLE: + if member not in binding.members: + binding.members.append(member) + break + else: + policy.bindings.append( + policy_pb2.Binding(role=CLOUD_SQL_ROLE, members=[member]) + ) + + projects_client.set_iam_policy(request={"resource": resource, "policy": policy}) + + +@retry.Retry(predicate=retry.if_exception_type(exceptions.Aborted)) +def revoke_cloud_sql_role( + projects_client: resourcemanager_v3.ProjectsClient, + project_id: str, + member: str, +) -> None: + """Removes member from CLOUD_SQL_ROLE on the project, added by grant_cloud_sql_role.""" + resource = f"projects/{project_id}" + policy = projects_client.get_iam_policy(request={"resource": resource}) + + changed = False + for binding in policy.bindings: + if binding.role == CLOUD_SQL_ROLE and member in binding.members: + binding.members.remove(member) + changed = True + + if changed: + projects_client.set_iam_policy(request={"resource": resource, "policy": policy}) + + @pytest.fixture() def secret_id( regional_client: secretmanager_v1.SecretManagerServiceClient, @@ -355,6 +432,33 @@ def regional_secret_with_delayed_destroy( yield secret_id +@pytest.fixture() +def regional_secret_with_cloud_sql_credentials( + projects_client: resourcemanager_v3.ProjectsClient, + project_id: str, + location_id: str, + secret_id: str, +) -> Iterator[str]: + print(f"creating cloud sql credentials secret {secret_id}") + secret = create_regional_secret_with_cloud_sql_credentials.create_regional_secret_with_cloud_sql_credentials( + project_id, location_id, secret_id + ) + + # enable_managed_rotation needs this secret's own built-in identity + # granted Cloud SQL IAM permissions first -- there's no broader grant + # that covers a secret before it exists, so every secret created here + # needs its own grant/revoke around the test that uses it. + member = secret.policy_member.iam_policy_uid_principal + grant_cloud_sql_role(projects_client, project_id, member) + # IAM grants are eventually consistent; give it a moment before a caller + # tries to use it for managed rotation. + time.sleep(10) + + yield secret_id + + revoke_cloud_sql_role(projects_client, project_id, member) + + def test_regional_quickstart(project_id: str, location_id: str, secret_id: str) -> None: regional_quickstart.regional_quickstart(project_id, location_id, secret_id) @@ -442,6 +546,110 @@ def test_create_regional_secret_with_annotations( assert secret_id in secret.name +def test_create_regional_secret_with_cloud_sql_credentials( + project_id: str, + location_id: str, + secret_id: str, +) -> None: + secret = create_regional_secret_with_cloud_sql_credentials.create_regional_secret_with_cloud_sql_credentials( + project_id, location_id, secret_id + ) + assert secret_id in secret.name + assert ( + secret.secret_type + == secretmanager_v1.Secret.SecretType.CLOUD_SQL_DB_CREDENTIALS + ) + + +def test_enable_regional_secret_managed_rotation( + regional_secret_with_cloud_sql_credentials: str, + project_id: str, + location_id: str, + cloud_sql_instance_id: str, + cloud_sql_username: str, +) -> None: + secret_id = regional_secret_with_cloud_sql_credentials + version = ( + enable_regional_secret_managed_rotation.enable_regional_secret_managed_rotation( + project_id, + location_id, + secret_id, + cloud_sql_instance_id, + cloud_sql_username, + ) + ) + assert secret_id in version.name + assert version.state == secretmanager_v1.SecretVersion.State.ENABLED + + +def test_rotate_regional_secret( + regional_secret_with_cloud_sql_credentials: str, + project_id: str, + location_id: str, + cloud_sql_instance_id: str, + cloud_sql_username: str, +) -> None: + secret_id = regional_secret_with_cloud_sql_credentials + first_version = ( + enable_regional_secret_managed_rotation.enable_regional_secret_managed_rotation( + project_id, + location_id, + secret_id, + cloud_sql_instance_id, + cloud_sql_username, + ) + ) + rotated_version = rotate_regional_secret.rotate_regional_secret( + project_id, location_id, secret_id + ) + assert secret_id in rotated_version.name + assert rotated_version.name != first_version.name + assert rotated_version.state == secretmanager_v1.SecretVersion.State.ENABLED + + +def test_update_regional_secret_with_managed_rotation_schedule( + regional_secret_with_cloud_sql_credentials: str, + project_id: str, + location_id: str, + cloud_sql_instance_id: str, + cloud_sql_username: str, +) -> None: + secret_id = regional_secret_with_cloud_sql_credentials + enable_regional_secret_managed_rotation.enable_regional_secret_managed_rotation( + project_id, location_id, secret_id, cloud_sql_instance_id, cloud_sql_username + ) + before = int(time.time()) + rotation_period_seconds = 3600 + secret = update_regional_secret_with_managed_rotation_schedule.update_regional_secret_with_managed_rotation_schedule( + project_id, + location_id, + secret_id, + rotation_period_seconds, + ) + assert secret_id in secret.name + assert ( + secret.rotation.next_rotation_time.timestamp() + >= before + rotation_period_seconds + ) + assert secret.rotation.rotation_period.seconds == rotation_period_seconds + + +def test_get_regional_secret_type( + project_id: str, + location_id: str, + regional_secret_with_cloud_sql_credentials: str, +) -> None: + secret_id = regional_secret_with_cloud_sql_credentials + secret = get_regional_secret_type.get_regional_secret_type( + project_id, location_id, secret_id + ) + assert secret_id in secret.name + assert ( + secret.secret_type + == secretmanager_v1.Secret.SecretType.CLOUD_SQL_DB_CREDENTIALS + ) + + def test_create_regional_secret_with_delayed_destroy( regional_client: secretmanager_v1.SecretManagerServiceClient, project_id: str, @@ -449,7 +657,9 @@ def test_create_regional_secret_with_delayed_destroy( secret_id: str, version_destroy_ttl: int, ) -> None: - secret = create_regional_secret_with_delayed_destroy.create_regional_secret_with_delayed_destroy(project_id, location_id, secret_id, version_destroy_ttl) + secret = create_regional_secret_with_delayed_destroy.create_regional_secret_with_delayed_destroy( + project_id, location_id, secret_id, version_destroy_ttl + ) assert secret_id in secret.name assert timedelta(seconds=version_destroy_ttl) == secret.version_destroy_ttl @@ -552,7 +762,11 @@ def test_disable_regional_secret_delayed_destroy( location_id: str, ) -> None: secret_id = regional_secret_with_delayed_destroy - updated_secret = disable_regional_secret_delayed_destroy.disable_regional_secret_delayed_destroy(project_id, location_id, secret_id) + updated_secret = ( + disable_regional_secret_delayed_destroy.disable_regional_secret_delayed_destroy( + project_id, location_id, secret_id + ) + ) assert updated_secret.version_destroy_ttl == timedelta(0) @@ -836,12 +1050,16 @@ def test_update_regional_secret_with_delayed_destroy( regional_secret_with_delayed_destroy: str, project_id: str, location_id: str, - version_destroy_ttl: int + version_destroy_ttl: int, ) -> None: secret_id = regional_secret_with_delayed_destroy updated_version_delayed_destroy = 118400 - updated_secret = update_regional_secret_with_delayed_destroy.update_regional_secret_with_delayed_destroy(project_id, location_id, secret_id, updated_version_delayed_destroy) - assert updated_secret.version_destroy_ttl == timedelta(seconds=updated_version_delayed_destroy) + updated_secret = update_regional_secret_with_delayed_destroy.update_regional_secret_with_delayed_destroy( + project_id, location_id, secret_id, updated_version_delayed_destroy + ) + assert updated_secret.version_destroy_ttl == timedelta( + seconds=updated_version_delayed_destroy + ) def test_view_regional_secret_labels( diff --git a/secretmanager/snippets/regional_samples/update_regional_secret_with_managed_rotation_schedule.py b/secretmanager/snippets/regional_samples/update_regional_secret_with_managed_rotation_schedule.py new file mode 100644 index 00000000000..28c171394eb --- /dev/null +++ b/secretmanager/snippets/regional_samples/update_regional_secret_with_managed_rotation_schedule.py @@ -0,0 +1,106 @@ +#!/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 updating the rotation schedule +of a CLOUD_SQL_DB_CREDENTIALS typed secret. +""" + +# [START secretmanager_update_regional_secret_with_managed_rotation_schedule] +import argparse +import time + +# Import the Secret Manager client library. +from google.cloud import secretmanager_v1 +from google.protobuf.duration_pb2 import Duration +from google.protobuf.timestamp_pb2 import Timestamp + + +def update_regional_secret_with_managed_rotation_schedule( + project_id: str, + location_id: str, + secret_id: str, + rotation_period_seconds: int, +) -> secretmanager_v1.Secret: + """ + Updates the rotation schedule of a CLOUD_SQL_DB_CREDENTIALS typed secret. + """ + + # Endpoint to call the regional Secret Manager API. + api_endpoint = f"secretmanager.{location_id}.rep.googleapis.com" + + # Create the Secret Manager client. + client = secretmanager_v1.SecretManagerServiceClient( + client_options={"api_endpoint": api_endpoint}, + ) + + # Build the resource name of the secret. + name = f"projects/{project_id}/locations/{location_id}/secrets/{secret_id}" + + # The rotation schedule of a CLOUD_SQL_DB_CREDENTIALS secret can be set + # before or after enabling managed rotation; EnableManagedRotation does not + # need to be called first. Other secret types also support a rotation + # schedule, but only when Pub/Sub topics are configured. Pub/Sub topics are + # not required for CLOUD_SQL_DB_CREDENTIALS. + # next_rotation_time and rotation_period must be set together. + next_rotation_timestamp = int(time.time()) + rotation_period_seconds + + # Build the updated secret. + secret = { + "name": name, + "rotation": { + "next_rotation_time": Timestamp(seconds=next_rotation_timestamp), + "rotation_period": Duration(seconds=rotation_period_seconds), + }, + } + + # Mask only the rotation subfields being set, not the whole "rotation" + # submessage. + update_mask = {"paths": ["rotation.next_rotation_time", "rotation.rotation_period"]} + + # Update the secret. + response = client.update_secret( + request={"secret": secret, "update_mask": update_mask} + ) + + print(f"Updated regional secret rotation schedule: {response.name}") + + return response + + +# [END secretmanager_update_regional_secret_with_managed_rotation_schedule] + +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("location_id", help="id of location where secret is stored") + parser.add_argument( + "secret_id", + help="id of the Cloud SQL DB credentials secret to reconfigure", + ) + parser.add_argument( + "rotation_period_seconds", + type=int, + help="seconds between rotations; must be at least 3600 (1 hour)", + ) + args = parser.parse_args() + + update_regional_secret_with_managed_rotation_schedule( + args.project_id, + args.location_id, + args.secret_id, + args.rotation_period_seconds, + ) diff --git a/secretmanager/snippets/requirements.txt b/secretmanager/snippets/requirements.txt index 2e6bd673f37..da4cda938eb 100644 --- a/secretmanager/snippets/requirements.txt +++ b/secretmanager/snippets/requirements.txt @@ -1,4 +1,4 @@ protobuf==6.33.6 google-cloud-resource-manager==1.18.0 -google-cloud-secret-manager==2.29.0 +google-cloud-secret-manager==2.30.0 google-crc32c==1.8.0 diff --git a/secretmanager/snippets/snippets_test.py b/secretmanager/snippets/snippets_test.py index 554e61b240a..02d490d70cb 100644 --- a/secretmanager/snippets/snippets_test.py +++ b/secretmanager/snippets/snippets_test.py @@ -12,18 +12,19 @@ # See the License for the specific language governing permissions and import base64 -import uuid +from datetime import timedelta import os import time -from datetime import timedelta from typing import Iterator, Optional, Tuple, Union +import uuid from google.api_core import exceptions, retry from google.cloud import resourcemanager_v3 from google.cloud import secretmanager from google.protobuf.duration_pb2 import Duration +import pytest -from access_secret_version import access_secret_version +from access_secret_version import access_secret_version # noqa: I100 from add_secret_version import add_secret_version from bind_tags_to_secret import bind_tags_to_secret from consume_event_notification import consume_event_notification @@ -32,6 +33,7 @@ from create_secret_with_delayed_destroy import create_secret_with_delayed_destroy from create_secret_with_labels import create_secret_with_labels from create_secret_with_tags import create_secret_with_tags +from create_secret_with_type import create_secret_with_type 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 @@ -47,6 +49,7 @@ from enable_secret_version import enable_secret_version from enable_secret_version_with_etag import enable_secret_version_with_etag from get_secret import get_secret +from get_secret_type import get_secret_type from get_secret_version import get_secret_version from iam_grant_access import iam_grant_access from iam_revoke_access import iam_revoke_access @@ -62,8 +65,6 @@ from view_secret_annotations import view_secret_annotations from view_secret_labels import view_secret_labels -import pytest - @pytest.fixture() def client() -> secretmanager.SecretManagerServiceClient: @@ -420,6 +421,17 @@ def test_bind_tags_to_secret( assert tag_value in tag_resp.tag_value +def test_create_secret_with_type( + project_id: str, + secret_id: str, +) -> None: + secret = create_secret_with_type( + project_id, secret_id, secretmanager.Secret.SecretType.ACCESS_KEY + ) + assert secret_id in secret.name + assert secret.secret_type == secretmanager.Secret.SecretType.ACCESS_KEY + + def test_create_secret_without_ttl( project_id: str, secret_id: str, @@ -595,6 +607,19 @@ def test_get_secret( assert secret_id in snippet_secret.name +def test_get_secret_type( + client: secretmanager.SecretManagerServiceClient, + project_id: str, + secret_id: str, +) -> None: + create_secret_with_type( + project_id, secret_id, secretmanager.Secret.SecretType.ACCESS_KEY + ) + secret = get_secret_type(project_id, secret_id) + assert secret_id in secret.name + assert secret.secret_type == secretmanager.Secret.SecretType.ACCESS_KEY + + def test_iam_grant_access( client: secretmanager.SecretManagerServiceClient, secret: Tuple[str, str, str], @@ -745,7 +770,7 @@ def test_update_secret_with_alias(secret_version: Tuple[str, str, str, str]) -> def test_update_secret_with_delayed_destroy( - secret_with_delayed_destroy: Tuple[str, str] + secret_with_delayed_destroy: Tuple[str, str], ) -> None: project_id, secret_id = secret_with_delayed_destroy updated_version_destroy_ttl_value = 118400