Merge pull request #2377 from acsbendi/get-caller-identity

GetCallerIdentity returns real data based on the access key used
This commit is contained in:
Mike Grima 2019-08-22 09:50:59 -07:00 committed by GitHub
commit 5fd2c30f97
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 175 additions and 42 deletions

View file

@ -106,7 +106,7 @@ class AssumedRoleAccessKey(object):
self._access_key_id = access_key_id
self._secret_access_key = assumed_role.secret_access_key
self._session_token = assumed_role.session_token
self._owner_role_name = assumed_role.arn.split("/")[-1]
self._owner_role_name = assumed_role.role_arn.split("/")[-1]
self._session_name = assumed_role.session_name
if headers["X-Amz-Security-Token"] != self._session_token:
raise CreateAccessKeyFailure(reason="InvalidToken")
@ -172,6 +172,8 @@ class IAMRequestBase(object):
self._raise_signature_does_not_match()
def check_action_permitted(self):
if self._action == 'sts:GetCallerIdentity': # always allowed, even if there's an explicit Deny for it
return True
policies = self._access_key.collect_policies()
permitted = False

View file

@ -2,7 +2,8 @@ from __future__ import unicode_literals
import datetime
from moto.core import BaseBackend, BaseModel
from moto.core.utils import iso_8601_datetime_with_milliseconds
from moto.sts.utils import random_access_key_id, random_secret_access_key, random_session_token
from moto.iam.models import ACCOUNT_ID
from moto.sts.utils import random_access_key_id, random_secret_access_key, random_session_token, random_assumed_role_id
class Token(BaseModel):
@ -22,7 +23,7 @@ class AssumedRole(BaseModel):
def __init__(self, role_session_name, role_arn, policy, duration, external_id):
self.session_name = role_session_name
self.arn = role_arn
self.role_arn = role_arn
self.policy = policy
now = datetime.datetime.utcnow()
self.expiration = now + datetime.timedelta(seconds=duration)
@ -30,11 +31,24 @@ class AssumedRole(BaseModel):
self.access_key_id = "ASIA" + random_access_key_id()
self.secret_access_key = random_secret_access_key()
self.session_token = random_session_token()
self.assumed_role_id = "AROA" + random_assumed_role_id()
@property
def expiration_ISO8601(self):
return iso_8601_datetime_with_milliseconds(self.expiration)
@property
def user_id(self):
return self.assumed_role_id + ":" + self.session_name
@property
def arn(self):
return "arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format(
account_id=ACCOUNT_ID,
role_name=self.role_arn.split("/")[-1],
session_name=self.session_name
)
class STSBackend(BaseBackend):
@ -54,6 +68,12 @@ class STSBackend(BaseBackend):
self.assumed_roles.append(role)
return role
def get_assumed_role_from_access_key(self, access_key_id):
for assumed_role in self.assumed_roles:
if assumed_role.access_key_id == access_key_id:
return assumed_role
return None
def assume_role_with_web_identity(self, **kwargs):
return self.assume_role(**kwargs)

View file

@ -1,6 +1,8 @@
from __future__ import unicode_literals
from moto.core.responses import BaseResponse
from moto.iam.models import ACCOUNT_ID
from moto.iam import iam_backend
from .exceptions import STSValidationError
from .models import sts_backend
@ -31,7 +33,7 @@ class TokenResponse(BaseResponse):
token = sts_backend.get_federation_token(
duration=duration, name=name, policy=policy)
template = self.response_template(GET_FEDERATION_TOKEN_RESPONSE)
return template.render(token=token)
return template.render(token=token, account_id=ACCOUNT_ID)
def assume_role(self):
role_session_name = self.querystring.get('RoleSessionName')[0]
@ -71,7 +73,23 @@ class TokenResponse(BaseResponse):
def get_caller_identity(self):
template = self.response_template(GET_CALLER_IDENTITY_RESPONSE)
return template.render()
# Default values in case the request does not use valid credentials generated by moto
user_id = "AKIAIOSFODNN7EXAMPLE"
arn = "arn:aws:sts::{account_id}:user/moto".format(account_id=ACCOUNT_ID)
access_key_id = self.get_current_user()
assumed_role = sts_backend.get_assumed_role_from_access_key(access_key_id)
if assumed_role:
user_id = assumed_role.user_id
arn = assumed_role.arn
user = iam_backend.get_user_from_access_key_id(access_key_id)
if user:
user_id = user.id
arn = user.arn
return template.render(account_id=ACCOUNT_ID, user_id=user_id, arn=arn)
GET_SESSION_TOKEN_RESPONSE = """<GetSessionTokenResponse xmlns="https://sts.amazonaws.com/doc/2011-06-15/">
@ -99,8 +117,8 @@ GET_FEDERATION_TOKEN_RESPONSE = """<GetFederationTokenResponse xmlns="https://st
<AccessKeyId>AKIAIOSFODNN7EXAMPLE</AccessKeyId>
</Credentials>
<FederatedUser>
<Arn>arn:aws:sts::123456789012:federated-user/{{ token.name }}</Arn>
<FederatedUserId>123456789012:{{ token.name }}</FederatedUserId>
<Arn>arn:aws:sts::{{ account_id }}:federated-user/{{ token.name }}</Arn>
<FederatedUserId>{{ account_id }}:{{ token.name }}</FederatedUserId>
</FederatedUser>
<PackedPolicySize>6</PackedPolicySize>
</GetFederationTokenResult>
@ -121,7 +139,7 @@ ASSUME_ROLE_RESPONSE = """<AssumeRoleResponse xmlns="https://sts.amazonaws.com/d
</Credentials>
<AssumedRoleUser>
<Arn>{{ role.arn }}</Arn>
<AssumedRoleId>ARO123EXAMPLE123:{{ role.session_name }}</AssumedRoleId>
<AssumedRoleId>{{ role.user_id }}</AssumedRoleId>
</AssumedRoleUser>
<PackedPolicySize>6</PackedPolicySize>
</AssumeRoleResult>
@ -153,9 +171,9 @@ ASSUME_ROLE_WITH_WEB_IDENTITY_RESPONSE = """<AssumeRoleWithWebIdentityResponse x
GET_CALLER_IDENTITY_RESPONSE = """<GetCallerIdentityResponse xmlns="https://sts.amazonaws.com/doc/2011-06-15/">
<GetCallerIdentityResult>
<Arn>arn:aws:sts::123456789012:user/moto</Arn>
<UserId>AKIAIOSFODNN7EXAMPLE</UserId>
<Account>123456789012</Account>
<Arn>{{ arn }}</Arn>
<UserId>{{ user_id }}</UserId>
<Account>{{ account_id }}</Account>
</GetCallerIdentityResult>
<ResponseMetadata>
<RequestId>c6104cbe-af31-11e0-8154-cbc7ccf896c7</RequestId>

View file

@ -6,15 +6,12 @@ import string
import six
ACCOUNT_SPECIFIC_ACCESS_KEY_PREFIX = "8NWMTLYQ"
ACCOUNT_SPECIFIC_ASSUMED_ROLE_ID_PREFIX = "3X42LBCD"
SESSION_TOKEN_PREFIX = "FQoGZXIvYXdzEBYaD"
def random_access_key_id():
return ACCOUNT_SPECIFIC_ACCESS_KEY_PREFIX + ''.join(six.text_type(
random.choice(
string.ascii_uppercase + string.digits
)) for _ in range(8)
)
return ACCOUNT_SPECIFIC_ACCESS_KEY_PREFIX + _random_uppercase_or_digit_sequence(8)
def random_secret_access_key():
@ -23,3 +20,16 @@ def random_secret_access_key():
def random_session_token():
return SESSION_TOKEN_PREFIX + base64.b64encode(os.urandom(266))[len(SESSION_TOKEN_PREFIX):].decode()
def random_assumed_role_id():
return ACCOUNT_SPECIFIC_ASSUMED_ROLE_ID_PREFIX + _random_uppercase_or_digit_sequence(9)
def _random_uppercase_or_digit_sequence(length):
return ''.join(
six.text_type(
random.choice(
string.ascii_uppercase + string.digits
)) for _ in range(length)
)