Merge pull request #2313 from caguado/fix/2312

Implement assume_role_with_web_identity

IMPLEMENTATION_COVERAGE.md

@@ -4128,7 +4128,7 @@
 ## sts - 42% implemented
 - [X] assume_role
 - [ ] assume_role_with_saml
-- [ ] assume_role_with_web_identity
+- [X] assume_role_with_web_identity
 - [ ] decode_authorization_message
 - [ ] get_caller_identity
 - [X] get_federation_token

moto/sts/models.py

@@ -50,5 +50,8 @@ class STSBackend(BaseBackend):
         role = AssumedRole(**kwargs)
         return role
 
+    def assume_role_with_web_identity(self, **kwargs):
+        return self.assume_role(**kwargs)
+
 
 sts_backend = STSBackend()

moto/sts/responses.py

@@ -39,6 +39,24 @@ class TokenResponse(BaseResponse):
         template = self.response_template(ASSUME_ROLE_RESPONSE)
         return template.render(role=role)
 
+    def assume_role_with_web_identity(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_with_web_identity(
+            role_session_name=role_session_name,
+            role_arn=role_arn,
+            policy=policy,
+            duration=duration,
+            external_id=external_id,
+        )
+        template = self.response_template(ASSUME_ROLE_WITH_WEB_IDENTITY_RESPONSE)
+        return template.render(role=role)
+
     def get_caller_identity(self):
         template = self.response_template(GET_CALLER_IDENTITY_RESPONSE)
         return template.render()
@@ -100,6 +118,27 @@ ASSUME_ROLE_RESPONSE = """<AssumeRoleResponse xmlns="https://sts.amazonaws.com/d
   </ResponseMetadata>
 </AssumeRoleResponse>"""
 
+
+ASSUME_ROLE_WITH_WEB_IDENTITY_RESPONSE = """<AssumeRoleWithWebIdentityResponse xmlns="https://sts.amazonaws.com/doc/2011-06-15/">
+  <AssumeRoleWithWebIdentityResult>
+    <Credentials>
+      <SessionToken>{{ role.session_token }}</SessionToken>
+      <SecretAccessKey>{{ role.secret_access_key }}</SecretAccessKey>
+      <Expiration>{{ role.expiration_ISO8601 }}</Expiration>
+      <AccessKeyId>{{ role.access_key_id }}</AccessKeyId>
+    </Credentials>
+    <AssumedRoleUser>
+      <Arn>{{ role.arn }}</Arn>
+      <AssumedRoleId>ARO123EXAMPLE123:{{ role.session_name }}</AssumedRoleId>
+    </AssumedRoleUser>
+    <PackedPolicySize>6</PackedPolicySize>
+  </AssumeRoleWithWebIdentityResult>
+  <ResponseMetadata>
+    <RequestId>c6104cbe-af31-11e0-8154-cbc7ccf896c7</RequestId>
+  </ResponseMetadata>
+</AssumeRoleWithWebIdentityResponse>"""
+
+
 GET_CALLER_IDENTITY_RESPONSE = """<GetCallerIdentityResponse xmlns="https://sts.amazonaws.com/doc/2011-06-15/">
   <GetCallerIdentityResult>
     <Arn>arn:aws:sts::123456789012:user/moto</Arn>

tests/test_sts/test_sts.py

@@ -74,6 +74,41 @@ def test_assume_role():
     role.user.assume_role_id.should.contain("session-name")
 
 
+@freeze_time("2012-01-01 12:00:00")
+@mock_sts_deprecated
+def test_assume_role_with_web_identity():
+    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_with_web_identity(
+            s3_role, "session-name", policy, duration_seconds=123)
+
+    credentials = role.credentials
+    credentials.expiration.should.equal('2012-01-01T12:02:03.000Z')
+    credentials.session_token.should.have.length_of(356)
+    assert credentials.session_token.startswith("FQoGZXIvYXdzE")
+    credentials.access_key.should.have.length_of(20)
+    assert credentials.access_key.startswith("ASIA")
+    credentials.secret_key.should.have.length_of(40)
+
+    role.user.arn.should.equal("arn:aws:iam::123456789012:role/test-role")
+    role.user.assume_role_id.should.contain("session-name")
+
+
 @mock_sts
 def test_get_caller_identity():
     identity = boto3.client(