From 8befcf37e4f2d372f1220c12eebef04d056fd8c3 Mon Sep 17 00:00:00 2001 From: smalaviya-crest Date: Mon, 31 Aug 2026 18:10:22 +0530 Subject: [PATCH 1/3] feat(secretmanager): add Cloud SQL managed-rotation samples --- ...ional_secret_with_cloud_sql_credentials.py | 92 ++++++++++++++++ ...enable_regional_secret_managed_rotation.py | 104 ++++++++++++++++++ .../rotate_regional_secret.py | 72 ++++++++++++ .../regional_samples/snippets_test.py | 16 +++ secretmanager/snippets/requirements.txt | 2 +- 5 files changed, 285 insertions(+), 1 deletion(-) create mode 100644 secretmanager/snippets/regional_samples/create_regional_secret_with_cloud_sql_credentials.py create mode 100644 secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py create mode 100644 secretmanager/snippets/regional_samples/rotate_regional_secret.py 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..461c05755f3 --- /dev/null +++ b/secretmanager/snippets/regional_samples/create_regional_secret_with_cloud_sql_credentials.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 +""" +command line application and sample code for creating a new secret that is +eligible for Cloud SQL 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: + """ + Create a new secret with the Cloud SQL DB credentials secret type. This + type is required to enable Secret Manager's automatic rotation of Cloud + SQL passwords. It can only be set when the secret is created, and the + secret's location must match the region of the target Cloud SQL + instance. + """ + + # 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}") + + # This built-in identity is what you grant Cloud SQL IAM permissions to, + # so that Secret Manager can rotate the database password on its behalf. + print( + "Grant this identity Cloud SQL IAM permissions to enable rotation: " + f"{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..ba76fdd10e4 --- /dev/null +++ b/secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py @@ -0,0 +1,104 @@ +#!/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 enabling managed rotation of +a Cloud SQL DB credentials 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: + """ + Enable managed rotation for a Cloud SQL DB credentials secret. This + links the secret to a Cloud SQL instance and database user, and can + only be called once per secret. It adds the secret's first version and + sets the matching password on the Cloud SQL user, taking the place of + a manually added secret version, which this secret type doesn't + support. Afterwards, use rotate_regional_secret.py to trigger further + rotations. + + instance_id is the bare Cloud SQL instance ID (e.g. "my-instance") -- + not a connection name. Neither the project nor the region should be + included: passing "PROJECT_ID:INSTANCE_ID" (as gcloud's own + `enable-managed-rotation --help` examples misleadingly show) or the + full "PROJECT_ID:LOCATION_ID:INSTANCE_ID" connection name both fail -- + the service already knows the project from the secret's own path, and + prepends it internally, so a qualified value ends up double-prefixed. + """ + + # 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. Leaving password unset lets Secret Manager + # generate a secure password itself. + 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 rotate" + ) + 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/rotate_regional_secret.py b/secretmanager/snippets/regional_samples/rotate_regional_secret.py new file mode 100644 index 00000000000..870161271a6 --- /dev/null +++ b/secretmanager/snippets/regional_samples/rotate_regional_secret.py @@ -0,0 +1,72 @@ +#!/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 a managed +rotation of a Cloud SQL DB credentials 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: + """ + Trigger a managed rotation for a Cloud SQL DB credentials secret. + Managed rotation must already be enabled on the secret (see + enable_regional_secret_managed_rotation.py). Each call generates a new + password, updates the Cloud SQL user, and adds the result as a new + secret version. + """ + + # 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..e789e4699c5 100644 --- a/secretmanager/snippets/regional_samples/snippets_test.py +++ b/secretmanager/snippets/regional_samples/snippets_test.py @@ -28,6 +28,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 @@ -442,6 +443,21 @@ 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_create_regional_secret_with_delayed_destroy( regional_client: secretmanager_v1.SecretManagerServiceClient, project_id: str, 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 From e18be756b98dd338687b67d64a98819cef29cf95 Mon Sep 17 00:00:00 2001 From: smalaviya-crest Date: Tue, 1 Sep 2026 10:37:35 +0530 Subject: [PATCH 2/3] feat(secretmanager): added few test case --- ...enable_regional_secret_managed_rotation.py | 3 +- .../regional_samples/snippets_test.py | 60 +++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py b/secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py index ba76fdd10e4..e9923671fa5 100644 --- a/secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py +++ b/secretmanager/snippets/regional_samples/enable_regional_secret_managed_rotation.py @@ -86,7 +86,8 @@ def enable_regional_secret_managed_rotation( 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" + "secret_id", + help="id of the Cloud SQL DB credentials secret to enable rotation on", ) parser.add_argument( "instance_id", diff --git a/secretmanager/snippets/regional_samples/snippets_test.py b/secretmanager/snippets/regional_samples/snippets_test.py index e789e4699c5..2ea5c5c8596 100644 --- a/secretmanager/snippets/regional_samples/snippets_test.py +++ b/secretmanager/snippets/regional_samples/snippets_test.py @@ -43,6 +43,7 @@ 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 @@ -54,6 +55,7 @@ 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 @@ -104,6 +106,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" @@ -356,6 +368,20 @@ def regional_secret_with_delayed_destroy( yield secret_id +@pytest.fixture() +def regional_secret_with_cloud_sql_credentials( + project_id: str, + location_id: str, + secret_id: str, +) -> Iterator[str]: + print(f"creating cloud sql credentials secret {secret_id}") + create_regional_secret_with_cloud_sql_credentials.create_regional_secret_with_cloud_sql_credentials( + project_id, location_id, secret_id + ) + + yield secret_id + + def test_regional_quickstart(project_id: str, location_id: str, secret_id: str) -> None: regional_quickstart.regional_quickstart(project_id, location_id, secret_id) @@ -458,6 +484,40 @@ def test_create_regional_secret_with_cloud_sql_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_create_regional_secret_with_delayed_destroy( regional_client: secretmanager_v1.SecretManagerServiceClient, project_id: str, From 6f33fd672e42f5dd2845badfca0b097c2b1c64cf Mon Sep 17 00:00:00 2001 From: smalaviya-crest Date: Tue, 1 Sep 2026 19:16:44 +0530 Subject: [PATCH 3/3] feat(secretmanager): update test file --- .../regional_samples/snippets_test.py | 77 ++++++++++++++++++- 1 file changed, 76 insertions(+), 1 deletion(-) diff --git a/secretmanager/snippets/regional_samples/snippets_test.py b/secretmanager/snippets/regional_samples/snippets_test.py index 2ea5c5c8596..bbbc6180f5a 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 @@ -96,6 +97,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"] @@ -220,6 +233,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, @@ -370,17 +432,30 @@ def regional_secret_with_delayed_destroy( @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}") - create_regional_secret_with_cloud_sql_credentials.create_regional_secret_with_cloud_sql_credentials( + 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)