diff --git a/fixlib/fixlib/baseresources.py b/fixlib/fixlib/baseresources.py index 9c63a85c52..d18d0797a3 100644 --- a/fixlib/fixlib/baseresources.py +++ b/fixlib/fixlib/baseresources.py @@ -1677,7 +1677,8 @@ class PermissionLevel(StrEnum): read = "read" tagging = "tagging" write = "write" - permission_management = "permission" + permission = "permission" + can_become = "can_become" # aka assume role unknown = "unknown" # in case a resource is not in the levels database diff --git a/plugins/aws/fix_plugin_aws/access_edges.py b/plugins/aws/fix_plugin_aws/access_edges.py index 3974be9466..bfe341a5b5 100644 --- a/plugins/aws/fix_plugin_aws/access_edges.py +++ b/plugins/aws/fix_plugin_aws/access_edges.py @@ -2,7 +2,7 @@ from attr import frozen, define from fix_plugin_aws.resource.base import AwsAccount, AwsResource, GraphBuilder -from typing import List, Literal, Set, Optional, Tuple, Union, Pattern +from typing import Dict, List, Literal, Set, Optional, Tuple, Union, Pattern from fixlib.baseresources import ( PermissionCondition, @@ -471,7 +471,15 @@ def is_service_linked_role(principal: AwsResource) -> bool: return False +action_level_overrides = { + "sts:AssumeRole": PermissionLevel.can_become, +} + + def get_action_level(action: str) -> PermissionLevel: + if override := action_level_overrides.get(action): + return override + service, action_name = action.split(":") level = "" action_data = get_action_data(service, action_name) @@ -491,7 +499,7 @@ def get_action_level(action: str) -> PermissionLevel: elif level == "Write": return PermissionLevel.write elif level == "Permissions management": - return PermissionLevel.permission_management + return PermissionLevel.permission else: return PermissionLevel.unknown @@ -805,6 +813,11 @@ def add_access_edges(self) -> None: if not permissions: continue - reported = to_json({"permissions": permissions}, strip_nulls=True) + access: Dict[PermissionLevel, bool] = {} + + for permission in permissions: + access[permission.level] = True + + reported = to_json({"permissions": permissions, "access": access}, strip_nulls=True) self.builder.add_edge(from_node=context.principal, edge_type=EdgeType.iam, reported=reported, node=node)