core sts endpoints completed

README.md

@@ -67,6 +67,8 @@ It gets even better! Moto isn't just S3. Here's the status of the other AWS serv
 |---------------------------------------------------------------------------|
 | SQS                   | @mock_sqs      | core endpoints done              |
 |---------------------------------------------------------------------------|
+| STS                   | @mock_sts      | core endpoints done              |
+|---------------------------------------------------------------------------|
 ```
 
 ### Another Example

moto/__init__.py

@@ -6,3 +6,4 @@ from .ec2 import mock_ec2
 from .s3 import mock_s3
 from .ses import mock_ses
 from .sqs import mock_sqs
+from .sts import mock_sts

moto/core/utils.py

@@ -69,3 +69,12 @@ class convert_flask_to_httpretty_response(object):
         # result is a status, headers, response tuple
         status, headers, response = result
         return response, status, headers
+
+
+def iso_8601_datetime(datetime):
+    return datetime.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def rfc_1123_datetime(datetime):
+    RFC1123 = '%a, %d %b %Y %H:%M:%S GMT'
+    return datetime.strftime(RFC1123)

moto/s3/models.py

@@ -2,6 +2,7 @@ import datetime
 import md5
 
 from moto.core import BaseBackend
+from moto.core.utils import iso_8601_datetime, rfc_1123_datetime
 from .utils import clean_key_name
 
 
@@ -27,14 +28,13 @@ class FakeKey(object):
 
     @property
     def last_modified_ISO8601(self):
-        return self.last_modified.strftime("%Y-%m-%dT%H:%M:%SZ")
+        return iso_8601_datetime(self.last_modified)
 
     @property
     def last_modified_RFC1123(self):
         # Different datetime formats depending on how the key is obtained
         # https://github.com/boto/boto/issues/466
-        RFC1123 = '%a, %d %b %Y %H:%M:%S GMT'
-        return self.last_modified.strftime(RFC1123)
+        return rfc_1123_datetime(self.last_modified)
 
     @property
     def metadata(self):

moto/server.py

@@ -8,6 +8,7 @@ from moto.ec2 import ec2_backend  # flake8: noqa
 from moto.s3 import s3_backend  # flake8: noqa
 from moto.ses import ses_backend  # flake8: noqa
 from moto.sqs import sqs_backend  # flake8: noqa
+from moto.sts import sts_backend  # flake8: noqa
 
 from moto.core.utils import convert_flask_to_httpretty_response
 

moto/sts/__init__.py

@@ -0,0 +1,2 @@
+from .models import sts_backend
+mock_sts = sts_backend.decorator

moto/sts/models.py

@@ -0,0 +1,39 @@
+import datetime
+from moto.core import BaseBackend
+from moto.core.utils import iso_8601_datetime
+
+
+class Token(object):
+    def __init__(self, duration):
+        now = datetime.datetime.now()
+        self.expiration = now + datetime.timedelta(seconds=duration)
+
+    @property
+    def expiration_ISO8601(self):
+        return iso_8601_datetime(self.expiration)
+
+
+class AssumedRole(object):
+    def __init__(self, role_session_name, role_arn, policy, duration, external_id):
+        self.session_name = role_session_name
+        self.arn = role_arn
+        self.policy = policy
+        now = datetime.datetime.now()
+        self.expiration = now + datetime.timedelta(seconds=duration)
+        self.external_id = external_id
+
+    @property
+    def expiration_ISO8601(self):
+        return iso_8601_datetime(self.expiration)
+
+
+class STSBackend(BaseBackend):
+    def get_session_token(self, duration):
+        token = Token(duration=duration)
+        return token
+
+    def assume_role(self, **kwargs):
+        role = AssumedRole(**kwargs)
+        return role
+
+sts_backend = STSBackend()

moto/sts/responses.py

@@ -0,0 +1,67 @@
+from jinja2 import Template
+
+from moto.core.responses import BaseResponse
+from .models import sts_backend
+
+
+class TokenResponse(BaseResponse):
+
+    def get_session_token(self):
+        duration = int(self.querystring.get('DurationSeconds', [43200])[0])
+        token = sts_backend.get_session_token(duration=duration)
+        template = Template(GET_SESSION_TOKEN_RESPONSE)
+        return template.render(token=token)
+
+    def assume_role(self):
+        role_session_name = self.querystring.get('RoleSessionName')[0]
+        role_arn = self.querystring.get('RoleArn')[0]
+
+        policy = self.querystring.get('Policy', [None])[0]
+        duration = int(self.querystring.get('DurationSeconds', [3600])[0])
+        external_id = self.querystring.get('ExternalId', [None])[0]
+
+        role = sts_backend.assume_role(
+            role_session_name=role_session_name,
+            role_arn=role_arn,
+            policy=policy,
+            duration=duration,
+            external_id=external_id,
+        )
+        template = Template(ASSUME_ROLE_RESPONSE)
+        return template.render(role=role)
+
+
+GET_SESSION_TOKEN_RESPONSE = """<GetSessionTokenResponse xmlns="https://sts.amazonaws.com/doc/2011-06-15/">
+  <GetSessionTokenResult>
+    <Credentials>
+      <SessionToken>AQoEXAMPLEH4aoAH0gNCAPyJxz4BlCFFxWNE1OPTgk5TthT+FvwqnKwRcOIfrRh3c/LTo6UDdyJwOOvEVPvLXCrrrUtdnniCEXAMPLE/IvU1dYUg2RVAJBanLiHb4IgRmpRV3zrkuWJOgQs8IZZaIv2BXIa2R4OlgkBN9bkUDNCJiBeb/AXlzBBko7b15fjrBs2+cTQtpZ3CYWFXG8C5zqx37wnOE49mRl/+OtkIKGO7fAE</SessionToken>
+      <SecretAccessKey>wJalrXUtnFEMI/K7MDENG/bPxRfiCYzEXAMPLEKEY</SecretAccessKey>
+      <Expiration>{{ token.expiration_ISO8601 }}</Expiration>
+      <AccessKeyId>AKIAIOSFODNN7EXAMPLE</AccessKeyId>
+    </Credentials>
+  </GetSessionTokenResult>
+  <ResponseMetadata>
+    <RequestId>58c5dbae-abef-11e0-8cfe-09039844ac7d</RequestId>
+  </ResponseMetadata>
+</GetSessionTokenResponse>"""
+
+
+ASSUME_ROLE_RESPONSE = """<AssumeRoleResponse xmlns="https://sts.amazonaws.com/doc/
+2011-06-15/">
+  <AssumeRoleResult>
+    <Credentials>
+      <SessionToken>BQoEXAMPLEH4aoAH0gNCAPyJxz4BlCFFxWNE1OPTgk5TthT+FvwqnKwRcOIfrRh3c/LTo6UDdyJwOOvEVPvLXCrrrUtdnniCEXAMPLE/IvU1dYUg2RVAJBanLiHb4IgRmpRV3zrkuWJOgQs8IZZaIv2BXIa2R4OlgkBN9bkUDNCJiBeb/AXlzBBko7b15fjrBs2+cTQtpZ3CYWFXG8C5zqx37wnOE49mRl/+OtkIKGO7fAE</SessionToken>
+      <SecretAccessKey>aJalrXUtnFEMI/K7MDENG/bPxRfiCYzEXAMPLEKEY</SecretAccessKey>
+      <Expiration>{{ role.expiration_ISO8601 }}</Expiration>
+      <AccessKeyId>AKIAIOSFODNN7EXAMPLE</AccessKeyId>
+    </Credentials>
+    <AssumedRoleUser>
+      <Arn>{{ role.arn }}</Arn>
+      <AssumedRoleId>ARO123EXAMPLE123:{{ role.session_name }}</AssumedRoleId>
+    </AssumedRoleUser>
+    <PackedPolicySize>6</PackedPolicySize>
+  </AssumeRoleResult>
+  <ResponseMetadata>
+    <RequestId>c6104cbe-af31-11e0-8154-cbc7ccf896c7</RequestId>
+  </ResponseMetadata>
+</AssumeRoleResponse>"""

moto/sts/urls.py

@@ -0,0 +1,9 @@
+from .responses import TokenResponse
+
+url_bases = [
+    "https?://sts.amazonaws.com"
+]
+
+url_paths = {
+    '{0}/$': TokenResponse().dispatch,
+}

tests/test_sts/test_server.py

@@ -0,0 +1,16 @@
+import sure  # flake8: noqa
+
+import moto.server as server
+
+'''
+Test the different server responses
+'''
+server.configure_urls("sts")
+
+
+def test_sts_get_session_token():
+    test_client = server.app.test_client()
+    res = test_client.get('/?Action=GetSessionToken')
+    res.status_code.should.equal(200)
+    res.data.should.contain("SessionToken")
+    res.data.should.contain("AccessKeyId")

tests/test_sts/test_sts.py

@@ -0,0 +1,52 @@
+import json
+
+import boto
+from boto.exception import BotoServerError
+from freezegun import freeze_time
+import sure  # flake8: noqa
+
+from moto import mock_sts
+
+
+@freeze_time("2012-01-01 12:00:00")
+@mock_sts
+def test_get_session_token():
+    conn = boto.connect_sts()
+    token = conn.get_session_token(duration=123)
+
+    token.expiration.should.equal('2012-01-01T12:02:03Z')
+    token.session_token.should.equal("AQoEXAMPLEH4aoAH0gNCAPyJxz4BlCFFxWNE1OPTgk5TthT+FvwqnKwRcOIfrRh3c/LTo6UDdyJwOOvEVPvLXCrrrUtdnniCEXAMPLE/IvU1dYUg2RVAJBanLiHb4IgRmpRV3zrkuWJOgQs8IZZaIv2BXIa2R4OlgkBN9bkUDNCJiBeb/AXlzBBko7b15fjrBs2+cTQtpZ3CYWFXG8C5zqx37wnOE49mRl/+OtkIKGO7fAE")
+    token.access_key.should.equal("AKIAIOSFODNN7EXAMPLE")
+    token.secret_key.should.equal("wJalrXUtnFEMI/K7MDENG/bPxRfiCYzEXAMPLEKEY")
+
+
+@freeze_time("2012-01-01 12:00:00")
+@mock_sts
+def test_assume_role():
+    conn = boto.connect_sts()
+
+    policy = json.dumps({
+        "Statement": [
+            {
+                "Sid": "Stmt13690092345534",
+                "Action": [
+                    "S3:ListBucket"
+                ],
+                "Effect": "Allow",
+                "Resource": [
+                    "arn:aws:s3:::foobar-tester"
+                ]
+            },
+        ]
+    })
+    s3_role = "arn:aws:iam::123456789012:role/test-role"
+    role = conn.assume_role(s3_role, "session-name", policy, duration_seconds=123)
+
+    credentials = role.credentials
+    credentials.expiration.should.equal('2012-01-01T12:02:03Z')
+    credentials.session_token.should.equal("BQoEXAMPLEH4aoAH0gNCAPyJxz4BlCFFxWNE1OPTgk5TthT+FvwqnKwRcOIfrRh3c/LTo6UDdyJwOOvEVPvLXCrrrUtdnniCEXAMPLE/IvU1dYUg2RVAJBanLiHb4IgRmpRV3zrkuWJOgQs8IZZaIv2BXIa2R4OlgkBN9bkUDNCJiBeb/AXlzBBko7b15fjrBs2+cTQtpZ3CYWFXG8C5zqx37wnOE49mRl/+OtkIKGO7fAE")
+    credentials.access_key.should.equal("AKIAIOSFODNN7EXAMPLE")
+    credentials.secret_key.should.equal("aJalrXUtnFEMI/K7MDENG/bPxRfiCYzEXAMPLEKEY")
+
+    role.user.arn.should.equal("arn:aws:iam::123456789012:role/test-role")
+    role.user.assume_role_id.should.contain("session-name")