Repository: getmoto/moto
Pull request #6799 (2023-09-10 07:16:03)

IAM validation: Resource is now processed during authorization
This PR uses the validation logic to validate the resource attribute of the principal-attached policies during `sts:AssumeRole`

It relates to the discussion in https://github.com/getmoto/moto/pull/6793#issuecomment-1712505836

What do you think of this draft @bblommers ? I implemented following your directions. I don't have enough experience to make an alternative proposition, and what you offered makes sense. It feels like it's going to be rather intense to implement each map of (service, method) to the corresponding resource attribute, but I don't see an alternative. 

If it looks good for you, I'll include in this PR the logic in TokenResponse._determine_resource() for every actions already supported for STS before submitting it as PR.

It's breaking some tests and there's one I've looked into I don't get:

> FAILED tests/test_s3/test_s3_bucket_policy.py::TestBucketPolicy::test_block_or_allow_get_object[kwargs2-403] - Failed: DID NOT RAISE <class 'botocore.exceptions.ClientError'>

On the previous commit that works, with a breakpoint at this line, I don't understand how the exception raises correctly ? https://github.com/ajoga/moto/blob/fa9aa95156d1e11a3c5a2e06ab8b085ae98c2c1f/tests/test_s3/test_s3_bucket_policy.py#L68 

When I compare my commit that breaks the test and the previous one that does not, I see no difference.

Also, do you see other tests cases needed?

I'll close the initial PR

bblommers: >  It feels like it's going to be rather intense to implement each map of (service, method)
>

Intense is the right word, I think! But that's the beauty of this approach - it can be done on a service by service basis, only if/when people need it.

> It's breaking some tests:
>

Took me a while to get my head around.. but I think the `TestBucketPolicy` test is just wrong. The Moto approach to bucket policies is: [allow, unless there is an explicit deny](https://github.com/getmoto/moto/blob/fa9aa95156d1e11a3c5a2e06ab8b085ae98c2c1f/moto/s3/responses.py#L1299).
But the failing test cases all expect a failure when we allow access to different resources.

So these test cases can just be removed IMO.

> Also, do you see other tests cases needed?
>
The new test in this PR look good. But we should extend this to also test the other methods, if you do extend the `_determine_resource`-logic to include other actions.

Patch:
diff --git a/moto/core/responses.py b/moto/core/responses.py
index 67c2cb6cd..42b852035 100644
--- a/moto/core/responses.py
+++ b/moto/core/responses.py
@@ -157,10 +157,10 @@ class ActionAuthenticatorMixin(object):
         else:
             ActionAuthenticatorMixin.request_count += 1
 
-    def _authenticate_and_authorize_normal_action(self) -> None:
+    def _authenticate_and_authorize_normal_action(self, resource: str = "*") -> None:
         from moto.iam.access_control import IAMRequest
 
-        self._authenticate_and_authorize_action(IAMRequest)
+        self._authenticate_and_authorize_action(IAMRequest, resource)
 
     def _authenticate_and_authorize_s3_action(
         self, bucket_name: Optional[str] = None, key_name: Optional[str] = None
@@ -491,9 +491,13 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
 
     def call_action(self) -> TYPE_RESPONSE:
         headers = self.response_headers
+        if hasattr(self, "_determine_resource"):
+            resource = self._determine_resource()
+        else:
+            resource = "*"
 
         try:
-            self._authenticate_and_authorize_normal_action()
+            self._authenticate_and_authorize_normal_action(resource)
         except HTTPException as http_error:
             response = http_error.description, dict(status=http_error.code)
             return self._send_response(headers, response)
diff --git a/moto/iam/access_control.py b/moto/iam/access_control.py
index 70b506260..2627a4b7a 100644
--- a/moto/iam/access_control.py
+++ b/moto/iam/access_control.py
@@ -380,7 +380,7 @@ class IAMPolicy:
                     permitted = True
         else:  # dict
             iam_policy_statement = IAMPolicyStatement(self._policy_json["Statement"])
-            return iam_policy_statement.is_action_permitted(action)
+            return iam_policy_statement.is_action_permitted(action, resource)
 
         if permitted:
             return PermissionResult.PERMITTED
@@ -408,6 +408,8 @@ class IAMPolicyStatement:
             if self.is_unknown_principal(self._statement.get("Principal")):
                 return PermissionResult.NEUTRAL
             same_resource = self._check_element_matches("Resource", resource)
+            if not same_resource:
+                return PermissionResult.NEUTRAL
             if self._statement["Effect"] == "Allow" and same_resource:
                 return PermissionResult.PERMITTED
             else:  # Deny
diff --git a/moto/iam/models.py b/moto/iam/models.py
index 5346d652a..023bcb493 100644
--- a/moto/iam/models.py
+++ b/moto/iam/models.py
@@ -84,9 +84,17 @@ def mark_account_as_visited(
 ) -> None:
     account = iam_backends[account_id]
     if access_key in account["global"].access_keys:
-        account["global"].access_keys[access_key].last_used = AccessKeyLastUsed(
+        key = account["global"].access_keys[access_key]
+        key.last_used = AccessKeyLastUsed(
             timestamp=utcnow(), service=service, region=region
         )
+        if key.role_arn:
+            try:
+                role = account["global"].get_role_by_arn(key.role_arn)
+                role.last_used = utcnow()
+            except IAMNotFoundException:
+                # User assumes a non-existing role
+                pass
     else:
         # User provided access credentials unknown to us
         pass
@@ -1082,6 +1090,7 @@ class AccessKey(CloudFormationModel):
         self.status = status
         self.create_date = utcnow()
         self.last_used: Optional[datetime] = None
+        self.role_arn: Optional[str] = None
 
     @property
     def created_iso_8601(self) -> str:
diff --git a/moto/sts/models.py b/moto/sts/models.py
index 5423761c5..b465b0039 100644
--- a/moto/sts/models.py
+++ b/moto/sts/models.py
@@ -108,6 +108,7 @@ class STSBackend(BaseBackend):
             duration,
             external_id,
         )
+        access_key.role_arn = role_arn
         account_backend = sts_backends[account_id]["global"]
         account_backend.assumed_roles.append(role)
         return role
diff --git a/moto/sts/responses.py b/moto/sts/responses.py
index c54481652..7e95e6535 100644
--- a/moto/sts/responses.py
+++ b/moto/sts/responses.py
@@ -13,6 +13,11 @@ class TokenResponse(BaseResponse):
     def backend(self) -> STSBackend:
         return sts_backends[self.current_account]["global"]
 
+    def _determine_resource(self) -> str:
+        if "AssumeRole" in self.querystring.get("Action", []):
+            return self.querystring.get("RoleArn")[0]  # type: ignore[index]
+        return "*"
+
     def get_session_token(self) -> str:
         duration = int(self.querystring.get("DurationSeconds", [43200])[0])
         token = self.backend.get_session_token(duration=duration)


Test patch:
diff --git a/tests/test_core/test_auth.py b/tests/test_core/test_auth.py
index b2a051112..8aceb2cf5 100644
--- a/tests/test_core/test_auth.py
+++ b/tests/test_core/test_auth.py
@@ -291,6 +291,47 @@ def test_access_denied_with_not_allowing_policy():
     )
 
 
+@set_initial_no_auth_action_count(3)
+@mock_sts
+def test_access_denied_explicitly_on_specific_resource():
+    user_name = "test-user"
+    forbidden_role_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/forbidden_explicitly"
+    allowed_role_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/allowed_implictly"
+    role_session_name = "dummy"
+    inline_policy_document = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Deny",
+                "Action": ["sts:AssumeRole"],
+                "Resource": forbidden_role_arn,
+            },
+            {"Effect": "Allow", "Action": ["sts:AssumeRole"], "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"],
+    )
+    with pytest.raises(ClientError) as ex:
+        client.assume_role(
+            RoleArn=forbidden_role_arn, RoleSessionName=role_session_name
+        )
+    assert ex.value.response["Error"]["Code"] == "AccessDenied"
+    assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 403
+    assert (
+        ex.value.response["Error"]["Message"]
+        == f"User: arn:aws:iam::{ACCOUNT_ID}:user/{user_name} is not authorized to perform: sts:AssumeRole"
+    )
+    # Not raising means success
+    client.assume_role(RoleArn=allowed_role_arn, RoleSessionName=role_session_name)
+
+
 @set_initial_no_auth_action_count(3)
 @mock_ec2
 def test_access_denied_for_run_instances():
diff --git a/tests/test_iam/test_iam_access_integration.py b/tests/test_iam/test_iam_access_integration.py
index 70a8c03fe..b977fb8aa 100644
--- a/tests/test_iam/test_iam_access_integration.py
+++ b/tests/test_iam/test_iam_access_integration.py
@@ -1,5 +1,9 @@
+import datetime
+
 import boto3
-from moto import mock_ec2, mock_iam
+from moto import mock_ec2, mock_iam, mock_sts, settings
+from moto.iam.models import iam_backends, IAMBackend
+from tests import DEFAULT_ACCOUNT_ID
 
 
 @mock_ec2
@@ -23,3 +27,36 @@ def test_invoking_ec2_mark_access_key_as_used():
     assert "LastUsedDate" in last_used
     assert last_used["ServiceName"] == "ec2"
     assert last_used["Region"] == "us-east-2"
+
+
+@mock_iam
+@mock_sts
+def test_mark_role_as_last_used():
+    role_name = "role_name_created_jan_1st"
+    iam = boto3.client("iam", "us-east-1")
+    sts = boto3.client("sts", "us-east-1")
+
+    role_arn = iam.create_role(RoleName=role_name, AssumeRolePolicyDocument="example")[
+        "Role"
+    ]["Arn"]
+
+    creds = sts.assume_role(RoleArn=role_arn, RoleSessionName="temp_session")[
+        "Credentials"
+    ]
+
+    iam2 = boto3.client(
+        "iam",
+        "us-east-1",
+        aws_access_key_id=creds["AccessKeyId"],
+        aws_secret_access_key=creds["SecretAccessKey"],
+        aws_session_token=creds["SessionToken"],
+    )
+
+    iam2.create_role(RoleName="name", AssumeRolePolicyDocument="example")
+
+    role = iam.get_role(RoleName=role_name)["Role"]
+    assert isinstance(role["RoleLastUsed"]["LastUsedDate"], datetime.datetime)
+
+    if not settings.TEST_SERVER_MODE:
+        iam: IAMBackend = iam_backends[DEFAULT_ACCOUNT_ID]["global"]
+        assert iam.get_role(role_name).last_used is not None
diff --git a/tests/test_iam/test_iam_resets.py b/tests/test_iam/test_iam_resets.py
index 0422a0ec7..49a2397c4 100644
--- a/tests/test_iam/test_iam_resets.py
+++ b/tests/test_iam/test_iam_resets.py
@@ -6,8 +6,8 @@ from moto import mock_iam
 
 # Test IAM User Inline Policy
 def test_policies_are_not_kept_after_mock_ends():
-    iam_client = boto3.client("iam", "us-east-1")
     with mock_iam():
+        iam_client = boto3.client("iam", "us-east-1")
         role_name = "test"
         assume_role_policy_document = {
             "Version": "2012-10-17",
diff --git a/tests/test_s3/test_s3_bucket_policy.py b/tests/test_s3/test_s3_bucket_policy.py
index 1b7bbea62..8a910c61c 100644
--- a/tests/test_s3/test_s3_bucket_policy.py
+++ b/tests/test_s3/test_s3_bucket_policy.py
@@ -34,15 +34,25 @@ class TestBucketPolicy:
     def teardown_class(cls):
         cls.server.stop()
 
+    xfail_reason = "S3 logic for resource-based policy is not yet correctly implemented, see https://github.com/getmoto/moto/pull/6799#issuecomment-1712799688"
+
     @pytest.mark.parametrize(
         "kwargs,status",
         [
             ({}, 200),
             ({"resource": "arn:aws:s3:::mybucket/test_txt"}, 200),
-            ({"resource": "arn:aws:s3:::notmybucket/*"}, 403),
-            ({"resource": "arn:aws:s3:::mybucket/other*"}, 403),
+            pytest.param(
+                {"resource": "arn:aws:s3:::notmybucket/*"},
+                403,
+                marks=pytest.mark.xfail(reason=xfail_reason),
+            ),
+            pytest.param(
+                {"resource": "arn:aws:s3:::mybucket/other*"},
+                403,
+                marks=pytest.mark.xfail(reason=xfail_reason),
+            ),
             ({"resource": ["arn:aws:s3:::mybucket", "arn:aws:s3:::mybucket/*"]}, 200),
-            (
+            pytest.param(
                 {
                     "resource": [
                         "arn:aws:s3:::notmybucket",
@@ -50,12 +60,16 @@ class TestBucketPolicy:
                     ]
                 },
                 403,
+                marks=pytest.mark.xfail(reason=xfail_reason),
             ),
-            (
+            pytest.param(
                 {"resource": ["arn:aws:s3:::mybucket", "arn:aws:s3:::notmybucket/*"]},
                 403,
+                marks=pytest.mark.xfail(reason=xfail_reason),
+            ),
+            pytest.param(
+                {"effect": "Deny"}, 403, marks=pytest.mark.xfail(reason=xfail_reason)
             ),
-            ({"effect": "Deny"}, 403),
         ],
     )
     def test_block_or_allow_get_object(self, kwargs, status):