Merge pull request #368 from spulec/kms

Add basic KMS key endpoints

moto/__init__.py

@@ -16,6 +16,7 @@ from .emr import mock_emr  # flake8: noqa
 from .glacier import mock_glacier  # flake8: noqa
 from .iam import mock_iam  # flake8: noqa
 from .kinesis import mock_kinesis  # flake8: noqa
+from .kms import mock_kms  # flake8: noqa
 from .rds import mock_rds  # flake8: noqa
 from .rds2 import mock_rds2  # flake8: noqa
 from .redshift import mock_redshift  # flake8: noqa

moto/backends.py

@@ -8,6 +8,7 @@ from moto.elb import elb_backend
 from moto.emr import emr_backend
 from moto.glacier import glacier_backend
 from moto.kinesis import kinesis_backend
+from moto.kms import kms_backend
 from moto.rds import rds_backend
 from moto.redshift import redshift_backend
 from moto.s3 import s3_backend
@@ -27,6 +28,7 @@ BACKENDS = {
     'emr': emr_backend,
     'glacier': glacier_backend,
     'kinesis': kinesis_backend,
+    'kms': kms_backend,
     'redshift': redshift_backend,
     'rds': rds_backend,
     's3': s3_backend,

moto/kms/__init__.py

@@ -0,0 +1,12 @@
+from __future__ import unicode_literals
+from .models import kms_backends
+from ..core.models import MockAWS
+
+kms_backend = kms_backends['us-east-1']
+
+
+def mock_kms(func=None):
+    if func:
+        return MockAWS(kms_backends)(func)
+    else:
+        return MockAWS(kms_backends)

moto/kms/models.py

@@ -0,0 +1,54 @@
+from __future__ import unicode_literals
+
+import boto.kms
+from moto.core import BaseBackend
+from .utils import generate_key_id
+
+
+class Key(object):
+    def __init__(self, policy, key_usage, description, region):
+        self.id = generate_key_id()
+        self.policy = policy
+        self.key_usage = key_usage
+        self.description = description
+        self.enabled = True
+        self.region = region
+        self.account_id = "0123456789012"
+
+    @property
+    def arn(self):
+        return "arn:aws:kms:{0}:{1}:key/{2}".format(self.region, self.account_id, self.id)
+
+    def to_dict(self):
+        return {
+            "KeyMetadata": {
+                "AWSAccountId": self.account_id,
+                "Arn": self.arn,
+                "CreationDate": "2015-01-01 00:00:00",
+                "Description": self.description,
+                "Enabled": self.enabled,
+                "KeyId": self.id,
+                "KeyUsage": self.key_usage,
+            }
+        }
+
+
+class KmsBackend(BaseBackend):
+
+    def __init__(self):
+        self.keys = {}
+
+    def create_key(self, policy, key_usage, description, region):
+        key = Key(policy, key_usage, description, region)
+        self.keys[key.id] = key
+        return key
+
+    def describe_key(self, key_id):
+        return self.keys[key_id]
+
+    def list_keys(self):
+        return self.keys.values()
+
+kms_backends = {}
+for region in boto.kms.regions():
+    kms_backends[region.name] = KmsBackend()

moto/kms/responses.py

@@ -0,0 +1,48 @@
+from __future__ import unicode_literals
+
+import json
+
+from moto.core.responses import BaseResponse
+from .models import kms_backends
+
+
+class KmsResponse(BaseResponse):
+
+    @property
+    def parameters(self):
+        return json.loads(self.body.decode("utf-8"))
+
+    @property
+    def kms_backend(self):
+        return kms_backends[self.region]
+
+    def create_key(self):
+        policy = self.parameters.get('Policy')
+        key_usage = self.parameters.get('KeyUsage')
+        description = self.parameters.get('Description')
+
+        key = self.kms_backend.create_key(policy, key_usage, description, self.region)
+        return json.dumps(key.to_dict())
+
+    def describe_key(self):
+        key_id = self.parameters.get('KeyId')
+        try:
+            key = self.kms_backend.describe_key(key_id)
+        except KeyError:
+            self.headers['status'] = 404
+            return "{}", self.headers
+        return json.dumps(key.to_dict())
+
+    def list_keys(self):
+        keys = self.kms_backend.list_keys()
+
+        return json.dumps({
+            "Keys": [
+                {
+                    "KeyArn": key.arn,
+                    "KeyId": key.id,
+                } for key in keys
+            ],
+            "NextMarker": None,
+            "Truncated": False,
+        })

moto/kms/urls.py

@@ -0,0 +1,10 @@
+from __future__ import unicode_literals
+from .responses import KmsResponse
+
+url_bases = [
+    "https?://kms.(.+).amazonaws.com",
+]
+
+url_paths = {
+    '{0}/$': KmsResponse.dispatch,
+}

moto/kms/utils.py

@@ -0,0 +1,7 @@
+from __future__ import unicode_literals
+
+import uuid
+
+
+def generate_key_id():
+    return str(uuid.uuid4())

tests/test_kms/test_kms.py

@@ -0,0 +1,46 @@
+from __future__ import unicode_literals
+
+import boto.kms
+from boto.exception import JSONResponseError
+import sure  # noqa
+
+from moto import mock_kms
+
+
+@mock_kms
+def test_create_key():
+    conn = boto.kms.connect_to_region("us-west-2")
+
+    key = conn.create_key(policy="my policy", description="my key", key_usage='ENCRYPT_DECRYPT')
+
+    key['KeyMetadata']['Description'].should.equal("my key")
+    key['KeyMetadata']['KeyUsage'].should.equal("ENCRYPT_DECRYPT")
+    key['KeyMetadata']['Enabled'].should.equal(True)
+
+
+@mock_kms
+def test_describe_key():
+    conn = boto.kms.connect_to_region("us-west-2")
+    key = conn.create_key(policy="my policy", description="my key", key_usage='ENCRYPT_DECRYPT')
+    key_id = key['KeyMetadata']['KeyId']
+
+    key = conn.describe_key(key_id)
+    key['KeyMetadata']['Description'].should.equal("my key")
+    key['KeyMetadata']['KeyUsage'].should.equal("ENCRYPT_DECRYPT")
+
+
+@mock_kms
+def test_describe_missing_key():
+    conn = boto.kms.connect_to_region("us-west-2")
+    conn.describe_key.when.called_with("not-a-key").should.throw(JSONResponseError)
+
+
+@mock_kms
+def test_list_keys():
+    conn = boto.kms.connect_to_region("us-west-2")
+
+    conn.create_key(policy="my policy", description="my key1", key_usage='ENCRYPT_DECRYPT')
+    conn.create_key(policy="my policy", description="my key2", key_usage='ENCRYPT_DECRYPT')
+
+    keys = conn.list_keys()
+    keys['Keys'].should.have.length_of(2)

tests/test_kms/test_server.py

@@ -0,0 +1,25 @@
+from __future__ import unicode_literals
+
+import json
+import sure  # noqa
+
+import moto.server as server
+from moto import mock_kms
+
+'''
+Test the different server responses
+'''
+
+
+@mock_kms
+def test_list_keys():
+    backend = server.create_backend_app("kms")
+    test_client = backend.test_client()
+
+    res = test_client.get('/?Action=ListKeys')
+
+    json.loads(res.data.decode("utf-8")).should.equal({
+        "Keys": [],
+        "NextMarker": None,
+        "Truncated": False,
+    })