diff --git a/setup.py b/setup.py index b777bb8b2..844abd7de 100644 --- a/setup.py +++ b/setup.py @@ -18,7 +18,7 @@ tests_require = [ "nose~=1.0", "mock~=2.0.0", - "moto~=0.4.25", + "moto~=0.4.30", "testfixtures~=4.10.0", ] diff --git a/stacker/lookups/handlers/kms.py b/stacker/lookups/handlers/kms.py index b0b5044fd..c0ca10184 100644 --- a/stacker/lookups/handlers/kms.py +++ b/stacker/lookups/handlers/kms.py @@ -1,6 +1,4 @@ -import base64 - -import botocore.session +import boto3 from ...util import read_value_from_path @@ -48,8 +46,6 @@ def handler(value, **kwargs): if "@" in value: region, value = value.split("@", 1) - s = botocore.session.get_session() - kms = s.create_client("kms", region_name=region) - decoded = base64.b64decode(value) - response = kms.decrypt(CiphertextBlob=decoded) - return response["Plaintext"] + kms = boto3.client("kms", region_name=region) + decoded = value.decode("base64") + return kms.decrypt(CiphertextBlob=decoded)["Plaintext"] diff --git a/stacker/tests/lookups/handlers/test_kms.py b/stacker/tests/lookups/handlers/test_kms.py index 7f9762781..32053aa7b 100644 --- a/stacker/tests/lookups/handlers/test_kms.py +++ b/stacker/tests/lookups/handlers/test_kms.py @@ -1,28 +1,32 @@ -import base64 -from mock import patch import unittest +from moto import mock_kms + +import boto3 + from stacker.lookups.handlers.kms import handler class TestKMSHandler(unittest.TestCase): - def setUp(self): - patcher = patch("botocore.session") - self.addCleanup(patcher.stop) - self.session = patcher.start() - self.kms = self.session.get_session().create_client() - self.input = base64.b64encode("encrypted test value") - self.value = {"Plaintext": "test value"} + self.plain = "my secret" + with mock_kms(): + kms = boto3.client("kms", region_name="us-east-1") + self.secret = kms.encrypt( + KeyId="alias/stacker", + Plaintext=self.plain.encode("base64") + )["CiphertextBlob"] def test_kms_handler(self): - self.kms.decrypt.return_value = self.value - decrypted = handler(self.input) - self.assertEqual(decrypted, self.value["Plaintext"]) + with mock_kms(): + decrypted = handler(self.secret) + print "DECRYPTED: %s" % decrypted + self.assertEqual(decrypted, self.plain) def test_kms_handler_with_region(self): - handler("us-west-2@{}".format(self.input)) - self.assertEqual(self.kms.decrypt.call_args[1]["CiphertextBlob"], - "encrypted test value") - kwargs = self.session.get_session().create_client.call_args[1] - self.assertEqual(kwargs["region_name"], "us-west-2") + region = "us-east-1" + value = "%s@%s" % (region, self.secret) + print value + with mock_kms(): + decrypted = handler(value) + self.assertEqual(decrypted, self.plain)