diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py index 3414efa2a..819fcea2f 100644 --- a/moto/secretsmanager/models.py +++ b/moto/secretsmanager/models.py @@ -1011,6 +1011,14 @@ class SecretsManagerBackend(BaseBackend): ) stages.remove(version_stage) + elif version_stage == "AWSCURRENT": + current_version = [ + v + for v in secret.versions + if "AWSCURRENT" in secret.versions[v]["version_stages"] + ][0] + err = f"The parameter RemoveFromVersionId can't be empty. Staging label AWSCURRENT is currently attached to version {current_version}, so you must explicitly reference that version in RemoveFromVersionId." + raise InvalidParameterException(err) if move_to_version_id: if move_to_version_id not in secret.versions: @@ -1026,6 +1034,9 @@ class SecretsManagerBackend(BaseBackend): # Whenever you move AWSCURRENT, Secrets Manager automatically # moves the label AWSPREVIOUS to the version that AWSCURRENT # was removed from. + for version in secret.versions: + if "AWSPREVIOUS" in secret.versions[version]["version_stages"]: + secret.versions[version]["version_stages"].remove("AWSPREVIOUS") secret.versions[remove_from_version_id]["version_stages"].append( "AWSPREVIOUS" ) diff --git a/tests/test_secretsmanager/__init__.py b/tests/test_secretsmanager/__init__.py index 08a1c1568..cc6729aa9 100644 --- a/tests/test_secretsmanager/__init__.py +++ b/tests/test_secretsmanager/__init__.py @@ -1 +1,43 @@ -# This file is intentionally left blank. +import os +from functools import wraps +from uuid import uuid4 + +import boto3 + +from moto import mock_aws + + +def secretsmanager_aws_verified(func): + """ + Function that is verified to work against AWS. + Can be run against AWS at any time by setting: + MOTO_TEST_ALLOW_AWS_REQUEST=true + + If this environment variable is not set, the function runs in a `mock_aws` context. + """ + + @wraps(func) + def pagination_wrapper(): + allow_aws_request = ( + os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true" + ) + + if allow_aws_request: + return create_secret_and_execute(func) + else: + with mock_aws(): + return create_secret_and_execute(func) + + def create_secret_and_execute(func): + sm_client = boto3.client("secretsmanager", "us-east-1") + + secret_arn = sm_client.create_secret( + Name=f"moto_secret_{str(uuid4())[0:6]}", + SecretString="old_secret", + )["ARN"] + try: + return func(secret_arn) + finally: + sm_client.delete_secret(SecretId=secret_arn) + + return pagination_wrapper diff --git a/tests/test_secretsmanager/test_secretsmanager.py b/tests/test_secretsmanager/test_secretsmanager.py index bec994d1f..6da6e034c 100644 --- a/tests/test_secretsmanager/test_secretsmanager.py +++ b/tests/test_secretsmanager/test_secretsmanager.py @@ -14,6 +14,8 @@ from freezegun import freeze_time from moto import mock_aws, settings from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID +from . import secretsmanager_aws_verified + DEFAULT_SECRET_NAME = "test-secret7" @@ -1733,3 +1735,84 @@ def test_update_secret_with_client_request_token(): assert pve.value.response["Error"]["Message"] == ( "ClientRequestToken must be 32-64 characters long." ) + + +@secretsmanager_aws_verified +@pytest.mark.aws_verified +def test_update_secret_version_stage_manually(secret_arn=None): + sm_client = boto3.client("secretsmanager", "us-east-1") + current_version = sm_client.put_secret_value( + SecretId=secret_arn, + SecretString="previous_secret", + VersionStages=["AWSCURRENT"], + )["VersionId"] + + initial_secret = sm_client.get_secret_value( + SecretId=secret_arn, VersionStage="AWSCURRENT" + ) + assert initial_secret["VersionStages"] == ["AWSCURRENT"] + assert initial_secret["SecretString"] == "previous_secret" + + token = str(uuid4()) + sm_client.put_secret_value( + SecretId=secret_arn, + ClientRequestToken=token, + SecretString="new_secret", + VersionStages=["AWSPENDING"], + ) + + pending_secret = sm_client.get_secret_value( + SecretId=secret_arn, VersionStage="AWSPENDING" + ) + assert pending_secret["VersionStages"] == ["AWSPENDING"] + assert pending_secret["SecretString"] == "new_secret" + + sm_client.update_secret_version_stage( + SecretId=secret_arn, + VersionStage="AWSCURRENT", + MoveToVersionId=token, + RemoveFromVersionId=current_version, + ) + + current_secret = sm_client.get_secret_value( + SecretId=secret_arn, VersionStage="AWSCURRENT" + ) + assert list(sorted(current_secret["VersionStages"])) == ["AWSCURRENT", "AWSPENDING"] + assert current_secret["SecretString"] == "new_secret" + + previous_secret = sm_client.get_secret_value( + SecretId=secret_arn, VersionStage="AWSPREVIOUS" + ) + assert previous_secret["VersionStages"] == ["AWSPREVIOUS"] + assert previous_secret["SecretString"] == "previous_secret" + + +@secretsmanager_aws_verified +@pytest.mark.aws_verified +def test_update_secret_version_stage_dont_specify_current_stage(secret_arn=None): + sm_client = boto3.client("secretsmanager", "us-east-1") + current_version = sm_client.put_secret_value( + SecretId=secret_arn, + SecretString="previous_secret", + VersionStages=["AWSCURRENT"], + )["VersionId"] + + token = str(uuid4()) + sm_client.put_secret_value( + SecretId=secret_arn, + ClientRequestToken=token, + SecretString="new_secret", + VersionStages=["AWSPENDING"], + ) + + # Without specifying version that currently has stage AWSCURRENT + with pytest.raises(ClientError) as exc: + sm_client.update_secret_version_stage( + SecretId=secret_arn, VersionStage="AWSCURRENT", MoveToVersionId=token + ) + err = exc.value.response["Error"] + assert err["Code"] == "InvalidParameterException" + assert ( + err["Message"] + == f"The parameter RemoveFromVersionId can't be empty. Staging label AWSCURRENT is currently attached to version {current_version}, so you must explicitly reference that version in RemoveFromVersionId." + )