diff --git a/moto/core/access_control.py b/moto/core/access_control.py index 800b7550f..3fb11eebd 100644 --- a/moto/core/access_control.py +++ b/moto/core/access_control.py @@ -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 diff --git a/moto/sts/models.py b/moto/sts/models.py index 4f6d2eb2a..c2ff7a8d3 100644 --- a/moto/sts/models.py +++ b/moto/sts/models.py @@ -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) diff --git a/moto/sts/responses.py b/moto/sts/responses.py index 068bd258d..ebdc4321c 100644 --- a/moto/sts/responses.py +++ b/moto/sts/responses.py @@ -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 = """ @@ -99,8 +117,8 @@ GET_FEDERATION_TOKEN_RESPONSE = """ - arn:aws:sts::123456789012:user/moto - AKIAIOSFODNN7EXAMPLE - 123456789012 + {{ arn }} + {{ user_id }} + {{ account_id }} c6104cbe-af31-11e0-8154-cbc7ccf896c7 diff --git a/moto/sts/utils.py b/moto/sts/utils.py index 8e6129728..50767729f 100644 --- a/moto/sts/utils.py +++ b/moto/sts/utils.py @@ -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) + ) diff --git a/tests/test_core/test_auth.py b/tests/test_core/test_auth.py index 3a1107eaa..00229f808 100644 --- a/tests/test_core/test_auth.py +++ b/tests/test_core/test_auth.py @@ -273,6 +273,27 @@ def test_access_denied_with_denying_policy(): ) +@set_initial_no_auth_action_count(3) +@mock_sts +def test_get_caller_identity_allowed_with_denying_policy(): + user_name = 'test-user' + inline_policy_document = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Deny", + "Action": "sts:GetCallerIdentity", + "Resource": "*" + } + ] + } + access_key = create_user_with_access_key_and_inline_policy(user_name, inline_policy_document) + client = boto3.client('sts', region_name='us-east-1', + aws_access_key_id=access_key['AccessKeyId'], + aws_secret_access_key=access_key['SecretAccessKey']) + client.get_caller_identity().should.be.a(dict) + + @set_initial_no_auth_action_count(3) @mock_ec2 def test_allowed_with_wildcard_action(): diff --git a/tests/test_sts/test_sts.py b/tests/test_sts/test_sts.py index 7d3586f2c..b047a8d13 100644 --- a/tests/test_sts/test_sts.py +++ b/tests/test_sts/test_sts.py @@ -8,7 +8,9 @@ from freezegun import freeze_time from nose.tools import assert_raises import sure # noqa -from moto import mock_sts, mock_sts_deprecated + +from moto import mock_sts, mock_sts_deprecated, mock_iam, settings +from moto.iam.models import ACCOUNT_ID from moto.sts.responses import MAX_FEDERATION_TOKEN_POLICY_LENGTH @@ -29,7 +31,8 @@ def test_get_session_token(): @mock_sts_deprecated def test_get_federation_token(): conn = boto.connect_sts() - token = conn.get_federation_token(duration=123, name="Bob") + token_name = "Bob" + token = conn.get_federation_token(duration=123, name=token_name) token.credentials.expiration.should.equal('2012-01-01T12:02:03.000Z') token.credentials.session_token.should.equal( @@ -38,15 +41,17 @@ def test_get_federation_token(): token.credentials.secret_key.should.equal( "wJalrXUtnFEMI/K7MDENG/bPxRfiCYzEXAMPLEKEY") token.federated_user_arn.should.equal( - "arn:aws:sts::123456789012:federated-user/Bob") - token.federated_user_id.should.equal("123456789012:Bob") + "arn:aws:sts::{account_id}:federated-user/{token_name}".format(account_id=ACCOUNT_ID, token_name=token_name)) + token.federated_user_id.should.equal(str(ACCOUNT_ID) + ":" + token_name) @freeze_time("2012-01-01 12:00:00") -@mock_sts_deprecated +@mock_sts def test_assume_role(): - conn = boto.connect_sts() + client = boto3.client( + "sts", region_name='us-east-1') + session_name = "session-name" policy = json.dumps({ "Statement": [ { @@ -61,20 +66,25 @@ def test_assume_role(): }, ] }) - s3_role = "arn:aws:iam::123456789012:role/test-role" - role = conn.assume_role(s3_role, "session-name", - policy, duration_seconds=123) + role_name = "test-role" + s3_role = "arn:aws:iam::{account_id}:role/{role_name}".format(account_id=ACCOUNT_ID, role_name=role_name) + assume_role_response = client.assume_role(RoleArn=s3_role, RoleSessionName=session_name, + Policy=policy, DurationSeconds=900) - 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) + credentials = assume_role_response['Credentials'] + if not settings.TEST_SERVER_MODE: + credentials['Expiration'].isoformat().should.equal('2012-01-01T12:15:00+00:00') + credentials['SessionToken'].should.have.length_of(356) + assert credentials['SessionToken'].startswith("FQoGZXIvYXdzE") + credentials['AccessKeyId'].should.have.length_of(20) + assert credentials['AccessKeyId'].startswith("ASIA") + credentials['SecretAccessKey'].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") + assume_role_response['AssumedRoleUser']['Arn'].should.equal("arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format( + account_id=ACCOUNT_ID, role_name=role_name, session_name=session_name)) + assert assume_role_response['AssumedRoleUser']['AssumedRoleId'].startswith("AROA") + assert assume_role_response['AssumedRoleUser']['AssumedRoleId'].endswith(":" + session_name) + assume_role_response['AssumedRoleUser']['AssumedRoleId'].should.have.length_of(21 + 1 + len(session_name)) @freeze_time("2012-01-01 12:00:00") @@ -96,9 +106,11 @@ def test_assume_role_with_web_identity(): }, ] }) - s3_role = "arn:aws:iam::123456789012:role/test-role" + role_name = "test-role" + s3_role = "arn:aws:iam::{account_id}:role/{role_name}".format(account_id=ACCOUNT_ID, role_name=role_name) + session_name = "session-name" role = conn.assume_role_with_web_identity( - s3_role, "session-name", policy, duration_seconds=123) + s3_role, session_name, policy, duration_seconds=123) credentials = role.credentials credentials.expiration.should.equal('2012-01-01T12:02:03.000Z') @@ -108,18 +120,68 @@ def test_assume_role_with_web_identity(): 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.arn.should.equal("arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format( + account_id=ACCOUNT_ID, role_name=role_name, session_name=session_name)) role.user.assume_role_id.should.contain("session-name") @mock_sts -def test_get_caller_identity(): +def test_get_caller_identity_with_default_credentials(): identity = boto3.client( "sts", region_name='us-east-1').get_caller_identity() - identity['Arn'].should.equal('arn:aws:sts::123456789012:user/moto') + identity['Arn'].should.equal('arn:aws:sts::{account_id}:user/moto'.format(account_id=ACCOUNT_ID)) identity['UserId'].should.equal('AKIAIOSFODNN7EXAMPLE') - identity['Account'].should.equal('123456789012') + identity['Account'].should.equal(str(ACCOUNT_ID)) + + +@mock_sts +@mock_iam +def test_get_caller_identity_with_iam_user_credentials(): + iam_client = boto3.client("iam", region_name='us-east-1') + iam_user_name = "new-user" + iam_user = iam_client.create_user(UserName=iam_user_name)['User'] + access_key = iam_client.create_access_key(UserName=iam_user_name)['AccessKey'] + + identity = boto3.client( + "sts", region_name='us-east-1', aws_access_key_id=access_key['AccessKeyId'], + aws_secret_access_key=access_key['SecretAccessKey']).get_caller_identity() + + identity['Arn'].should.equal(iam_user['Arn']) + identity['UserId'].should.equal(iam_user['UserId']) + identity['Account'].should.equal(str(ACCOUNT_ID)) + + +@mock_sts +@mock_iam +def test_get_caller_identity_with_assumed_role_credentials(): + iam_client = boto3.client("iam", region_name='us-east-1') + sts_client = boto3.client("sts", region_name='us-east-1') + iam_role_name = "new-user" + trust_policy_document = { + "Version": "2012-10-17", + "Statement": { + "Effect": "Allow", + "Principal": {"AWS": "arn:aws:iam::{account_id}:root".format(account_id=ACCOUNT_ID)}, + "Action": "sts:AssumeRole" + } + } + iam_role_arn = iam_client.role_arn = iam_client.create_role( + RoleName=iam_role_name, + AssumeRolePolicyDocument=json.dumps(trust_policy_document) + )['Role']['Arn'] + session_name = "new-session" + assumed_role = sts_client.assume_role(RoleArn=iam_role_arn, + RoleSessionName=session_name) + access_key = assumed_role['Credentials'] + + identity = boto3.client( + "sts", region_name='us-east-1', aws_access_key_id=access_key['AccessKeyId'], + aws_secret_access_key=access_key['SecretAccessKey']).get_caller_identity() + + identity['Arn'].should.equal(assumed_role['AssumedRoleUser']['Arn']) + identity['UserId'].should.equal(assumed_role['AssumedRoleUser']['AssumedRoleId']) + identity['Account'].should.equal(str(ACCOUNT_ID)) @mock_sts