"""Test AssumeRoleSecretsManager.""" from datetime import ( UTC, datetime, timedelta, ) from typing import Any from unittest.mock import MagicMock, patch import pytest from botocore.exceptions import ClientError from m2mconfig.connectors.assume_role_secrets_manager import ( AssumeRoleCredentials, AssumeRoleSecretsManager, SecretScheduledForDeletionError, is_expired, ) @pytest.mark.parametrize( "payload, expected", [ ( {}, { "aws_access_key_id": None, "aws_secret_access_key": None, "aws_session_token": None, "expiration": None, }, ), ( {"AccessKeyId": "key"}, { "aws_access_key_id": "key", "aws_secret_access_key": None, "aws_session_token": None, "expiration": None, }, ), ( {"AccessKeyId": "key", "SecretAccessKey": "secret"}, { "aws_access_key_id": "key", "aws_secret_access_key": "secret", "aws_session_token": None, "expiration": None, }, ), ( {"AccessKeyId": "key", "SecretAccessKey": "secret", "SessionToken": "sesh"}, { "aws_access_key_id": "key", "aws_secret_access_key": "secret", "aws_session_token": "sesh", "expiration": None, }, ), ( { "AccessKeyId": "123", "SecretAccessKey": "secret", "SessionToken": "tok", "Expiration": datetime(2024, 3, 1, 0, 0), }, { "aws_access_key_id": "123", "aws_secret_access_key": "secret", "aws_session_token": "tok", "expiration": datetime(2024, 3, 1, 0, 0), }, ), ], ) def test_assume_role_credentials_schema( payload: dict[str, Any], expected: dict[str, Any], ) -> None: """Test AssumeRoleCredentials schema.""" assert AssumeRoleCredentials.model_validate(payload).model_dump() == expected def test_is_expired() -> None: """Test is_expired""" future_datetime = datetime.now(UTC) + timedelta(3600) assert not is_expired(future_datetime) past_datetime = datetime.now(UTC) + timedelta(-3600) assert is_expired(past_datetime) @pytest.fixture() def mock_sts_client() -> MagicMock: """Fixture for mock_sts_client""" mock_sts_client = MagicMock() mock_sts_client.assume_role.return_value = { "Credentials": {}, } return mock_sts_client @pytest.fixture() @patch.object( AssumeRoleSecretsManager, "_check_expiration_and_setup_secrets_manager_client" ) def mock_assume_role_secrets_manager(mock_setup: MagicMock) -> AssumeRoleSecretsManager: """Returns a AssumeRoleSecretsManager that has mocked out its AssumeRole/SecretsManager client setup.""" mock_setup.return_value = True return AssumeRoleSecretsManager( environment="fun", sts_client=MagicMock(), aws_account_id="123", aws_assume_role_name="dogs-engineer-dogs", ) @pytest.mark.parametrize( "credentials,expected_cred_kwargs,cred_description", [ ( {}, { "aws_access_key_id": None, "aws_secret_access_key": None, "aws_session_token": None, }, "empty access credentials can be passed to set up secrets manager client", ), ( { "AccessKeyId": "123", "SecretAccessKey": "secret", "SessionToken": "tok", }, { "aws_access_key_id": "123", "aws_secret_access_key": "secret", "aws_session_token": "tok", }, "AssumeRole credentials are transformed to kwargs before passed to set up secrets manager client", ), ( { "AccessKeyId": "123", "SecretAccessKey": "secret", "SessionToken": "tok", "expiration": "now", }, { "aws_access_key_id": "123", "aws_secret_access_key": "secret", "aws_session_token": "tok", }, "expiration should not not be passed to set up secrets manager client", ), ], ) @patch("m2mconfig.connectors.assume_role_secrets_manager.boto3") @patch("m2mconfig.connectors.assume_role_secrets_manager.uuid4") def test_arsm_constructor( mock_uuid4: MagicMock, mock_boto3: MagicMock, credentials: dict[str, Any], expected_cred_kwargs: dict[str, Any], cred_description: str, mock_sts_client: MagicMock, ) -> None: """Test AssumeRoleSecretsManager constructor sets up secrets manager.""" mock_boto3.client.return_value = MagicMock() mock_uuid4.return_value = "hey-uuid" mock_sts_client.assume_role.return_value = {"Credentials": credentials} AssumeRoleSecretsManager( environment="fun", sts_client=mock_sts_client, aws_account_id="123", aws_assume_role_name="dogs-engineer-dogs", ) mock_sts_client.assume_role.assert_called_once_with( RoleArn="arn:aws:iam::123:role/fun-dogs-engineer-dogs", RoleSessionName="hey-uuid", DurationSeconds=3600, ) ( mock_boto3.client.assert_called_once_with( "secretsmanager", **expected_cred_kwargs ), cred_description, ) @pytest.mark.parametrize( "describe_secret_value,expected_result,expected_log_message", [ ( { "ARN": "arn:aws:secretsmanager:us-east-1:591204808501:secret:qa/lambda-test-m2m-client/M2M_AUTH0_CLIENT_CREDENTIALS-H3WvVs", "Name": "qa/lambda-test-m2m-client/M2M_AUTH0_CLIENT_CREDENTIALS", }, True, None, ), ( ClientError( error_response={"Error": {"Code": "ResourceNotFoundException"}}, operation_name="describe_secret", ), False, "Secret does not exist", ), ( ClientError( error_response={ "Error": { "Code": "AccessDeniedException", "Message": "User: arn:aws:sts::123:assumed-role/qa-assume-auth0-m2m-config/xxxx is not authorized to perform", } }, operation_name="describe_secret", ), False, "User: arn:aws:sts::123:assumed-role/qa-assume-auth0-m2m-config/xxxx is not authorized to perform", ), ( Exception("Bad juju"), False, "Could not get secret", ), ], ) @patch.object( AssumeRoleSecretsManager, "_check_expiration_and_setup_secrets_manager_client" ) def test_arsm_does_secret_exist( mock_setup_fn: MagicMock, describe_secret_value: Any, expected_result: bool, expected_log_message: Any, mock_assume_role_secrets_manager: AssumeRoleSecretsManager, caplog: pytest.LogCaptureFixture, ) -> None: """Test does_secret_exist.""" secret_name = "qa/this/M2M_AUTH0_CLIENT_CREDENTIALS" mock_secrets_manager_client = MagicMock() mock_secrets_manager_client.describe_secret.side_effect = describe_secret_value mock_assume_role_secrets_manager._secrets_manager_client = ( mock_secrets_manager_client ) result = mock_assume_role_secrets_manager.does_secret_exist(secret_name) assert result == expected_result mock_setup_fn.assert_called_once() mock_secrets_manager_client.describe_secret.assert_called_once_with( SecretId=secret_name ) if expected_log_message: assert expected_log_message in caplog.text @pytest.mark.parametrize( "get_secret_value_response,expect_exception,expected_secret,description", [ ({"SecretString": "dog"}, False, "dog", "SecretString is returned"), ( {"SecretString": "dog", "SecretBinary": b"some binary"}, False, "dog", "SecretString is returned even if SecretBinary is present", ), ( {"SecretBinary": b"some binary"}, True, None, "Only SecretString is supported", ), ({}, True, None, "Exception when no SecretString and no SecretBinary"), ( ClientError( error_response={"Error": {"Code": "Does Not Matter"}}, operation_name="describe_secret", ), True, None, "Exception on any ClientError", ), ], ) @patch.object( AssumeRoleSecretsManager, "_check_expiration_and_setup_secrets_manager_client" ) def test_arsm_get_secret_string( mock_setup_fn: MagicMock, mock_assume_role_secrets_manager: AssumeRoleSecretsManager, get_secret_value_response: Any, expect_exception: bool, expected_secret: Any, description: str, ) -> None: """Test get_secret.""" secret_name = "qa/this/M2M_AUTH0_CLIENT_CREDENTIALS" mock_secrets_manager_client = MagicMock() mock_secrets_manager_client.get_secret_value.side_effect = ( get_secret_value_response, ) mock_assume_role_secrets_manager._secrets_manager_client = ( mock_secrets_manager_client ) if expect_exception: with pytest.raises(Exception): # noqa: B017 mock_assume_role_secrets_manager.get_secret_string(secret_name) else: result = mock_assume_role_secrets_manager.get_secret_string(secret_name) assert result == expected_secret, description mock_setup_fn.assert_called_once() mock_secrets_manager_client.get_secret_value.assert_called_once_with( SecretId=secret_name ) @pytest.mark.parametrize( "exception_message, raise_scheduled_for_deletion_err, description", [ ( "An error occurred (InvalidRequestException) when calling the GetSecretValue operation: Some other message", False, "ClientError gets raised as is when not related to secret deletion", ), ( "An error occurred (InvalidRequestException) when calling the GetSecretValue operation: You can't perform this operation on the secret because it was marked for deletion.", True, "Custom SecretScheduledForDeletionError gets raised when secret is marked for deletion", ), ], ) @patch.object( AssumeRoleSecretsManager, "_check_expiration_and_setup_secrets_manager_client" ) def test_arsm_get_secret_string_invalid_request_exception( mock_setup_fn: MagicMock, mock_assume_role_secrets_manager: AssumeRoleSecretsManager, exception_message: str, raise_scheduled_for_deletion_err: bool, description: str, ) -> None: """Test get_secret_string handles InvalidRequestException when secret is marked for deletion.""" secret_name = "qa/this/M2M_AUTH0_CLIENT_CREDENTIALS" mock_secrets_manager_client = MagicMock() mock_secrets_manager_client.get_secret_value.side_effect = ClientError( error_response={ "Error": {"Code": "InvalidRequestException", "Message": exception_message} }, operation_name="describe_secret", ) mock_assume_role_secrets_manager._secrets_manager_client = ( mock_secrets_manager_client ) if raise_scheduled_for_deletion_err: with pytest.raises( SecretScheduledForDeletionError ) as scheduled_for_deletion_err: mock_assume_role_secrets_manager.get_secret_string(secret_name) assert ( scheduled_for_deletion_err.value.args[0] == f"Secret name: {secret_name} is scheduled for deletion" ), description mock_setup_fn.assert_called_once() mock_secrets_manager_client.get_secret_value.assert_called_once_with( SecretId=secret_name ) else: with pytest.raises(ClientError) as client_error: mock_assume_role_secrets_manager.get_secret_string(secret_name) assert ( client_error.value.response["Error"]["Message"] == exception_message ), description mock_setup_fn.assert_called_once() mock_secrets_manager_client.get_secret_value.assert_called_once_with( SecretId=secret_name ) @pytest.mark.parametrize( "put_secret_value_response,expected_result,description", [ ( {"VersionId": "123"}, True, "VersionId indicates a new version of the secret was created", ), ( {"Arn": "123"}, False, "Missing VersionId indicates a new version of the secret was not created", ), ( ClientError( error_response={"Error": {"Code": "Does Not Matter"}}, operation_name="put_secret_value", ), False, "Exception on any ClientError", ), ], ) @patch.object( AssumeRoleSecretsManager, "_check_expiration_and_setup_secrets_manager_client" ) def test_arsm_save_secret_string( mock_setup_fn: MagicMock, mock_assume_role_secrets_manager: AssumeRoleSecretsManager, put_secret_value_response: Any, expected_result: bool, description: str, ) -> None: """Test save_secret as a string.""" secret_name = "qa/this/M2M_AUTH0_CLIENT_CREDENTIALS" secret_value = '{"grant_type": "client_credentials", "audience": "an_api_audience", "client_id": "some-client-id", "client_secret": "some-secret-string"}' mock_secrets_manager_client = MagicMock() mock_secrets_manager_client.put_secret_value.side_effect = ( put_secret_value_response, ) mock_assume_role_secrets_manager._secrets_manager_client = ( mock_secrets_manager_client ) result = mock_assume_role_secrets_manager.save_secret_string( secret_name, secret_value ) assert result == expected_result, description mock_setup_fn.assert_called_once() mock_secrets_manager_client.put_secret_value.assert_called_once_with( SecretId=secret_name, SecretString=secret_value, )