From e7d46eb371677c3bfec3a606f0daa8d88baf5d0e Mon Sep 17 00:00:00 2001 From: Denis Samoilov Date: Wed, 21 May 2025 10:04:51 +0300 Subject: [PATCH 1/6] CIEM-45:fix split --- iam-ape/iam_ape/expand_policy.py | 39 +++++++- iam-ape/iam_ape/main.py | 166 +++++++++++++++++-------------- 2 files changed, 132 insertions(+), 73 deletions(-) diff --git a/iam-ape/iam_ape/expand_policy.py b/iam-ape/iam_ape/expand_policy.py index 31789a4..8e4e1eb 100644 --- a/iam-ape/iam_ape/expand_policy.py +++ b/iam-ape/iam_ape/expand_policy.py @@ -333,11 +333,47 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: for allow_action_set in allow_actions.values(): squashed_policies.update(allow_action_set) + # Arrange by source, action, resource, and not_resource to detect split conditions + actions_by_source_action_resource: Dict[ + Tuple[str, str, Optional[str], Optional[str]], List[Action] + ] = defaultdict(list) + + for action_tuple in squashed_policies: + key = (action_tuple.source, action_tuple.action, action_tuple.resource, action_tuple.not_resource) + actions_by_source_action_resource[key].append(action_tuple) + + # Merge conditions for actions from the same source + merged_actions: Set[Action] = set() + for actions_list in actions_by_source_action_resource.values(): + if len(actions_list) == 1: + merged_actions.add(actions_list[0]) + else: + # Multiple actions with same source/action/resource - merge their conditions + merged_condition = {} + for action in actions_list: + if action.condition: + for operator, operator_conditions in action.condition.items(): + if operator in merged_condition: + if isinstance(merged_condition[operator], dict) and isinstance(operator_conditions, dict): + merged_condition[operator].update(operator_conditions) + else: + merged_condition[operator] = operator_conditions + + # Create a new merged action + merged_action = Action( + action=actions_list[0].action, + resource=actions_list[0].resource, + not_resource=actions_list[0].not_resource, + condition=merged_condition if merged_condition else None, + source=actions_list[0].source + ) + merged_actions.add(merged_action) + # Arrange by Resource/NotResource by_resource_notresource: Dict[ Tuple[Optional[str], Optional[str]], Dict[Optional[HashableDict], Set[str]] ] = defaultdict(lambda: defaultdict(set)) - for action_tuple in squashed_policies: + for action_tuple in merged_actions: by_resource_notresource[(action_tuple.resource, action_tuple.not_resource)][ HashableDict.recursively(action_tuple.condition) ].add(action_tuple.action) @@ -412,5 +448,6 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: policy_res["Statement"] = [admin_statement] else: policy_res["Statement"] = list(final_statements.values()) + print("Before normalize_policy:", policy_res) return normalize_policy(policy_res) diff --git a/iam-ape/iam_ape/main.py b/iam-ape/iam_ape/main.py index 97a1881..3222ecb 100644 --- a/iam-ape/iam_ape/main.py +++ b/iam-ape/iam_ape/main.py @@ -4,7 +4,7 @@ import os import re import sys -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, Dict, List, Optional, Tuple, Union, Set import boto3 from botocore.exceptions import ClientError, ProfileNotFound @@ -16,7 +16,8 @@ IamApeException, InvalidArnException, ) -from iam_ape.helper_classes import PolicyWithSource +from iam_ape.expand_policy import PolicyExpander +from iam_ape.helper_classes import PolicyWithSource, Action from iam_ape.helper_functions import deep_update from iam_ape.helper_types import AwsPolicyType, EntityType, FinalReportT @@ -258,76 +259,97 @@ def build_arg_parser() -> argparse.ArgumentParser: return arg_parser -def main() -> int: - initialize_logger() - logger = logging.getLogger("IAM-APE") - - arg_parser = build_arg_parser() - arguments = arg_parser.parse_args() - - if not arguments.quiet: - print_banner() - - if arguments.verbose: - logging.root.setLevel(logging.DEBUG) - elif arguments.quiet: - logging.root.setLevel(logging.WARNING) - - if not (arguments.arn or arguments.update): - arg_parser.print_help() - return 0 - - if arguments.update: - scrape_iam_actions() - if not arguments.arn: - return 0 - - try: - entity_type, entity_account = validate_arn(arguments.arn) - except InvalidArnException as e: - logger.error(e) - return -1 - - try: - auth_details = get_auth_details(arguments.input, arguments.profile) - except (AwsAuthorizationException, ProfileNotFound) as e: - logger.error(e) - return -1 - - scp_policies = None - if arguments.input is None or arguments.scp_policy is not None: - scp_policies = get_scp_policies( - arguments.scp_policy, arguments.profile, entity_account - ) - - logger.info("Evaluating effective permissions") - calculator = EffectivePolicyEvaluator( - authorization_details=auth_details, scp_policies=scp_policies - ) - - try: - res = calculator.evaluate(arn=arguments.arn, entity_type=entity_type) - except IamApeException as e: - logger.error(e) - return -1 - - out: Union[AwsPolicyType, FinalReportT] - if arguments.format == "clean": - out = calculator.policy_expander.shrink_policy(res.allowed_permissions) - else: - out = calculator.create_json_report(res) - - if arguments.output == "stdout": - logger.info(f"Effective permissions policy for {arguments.arn}\n") - print(json.dumps(out, indent=2)) - else: - with open(arguments.output, "w") as f: - json.dump(out, f, indent=2) - logger.info( - f"Wrote effective permissions for {arguments.arn} to {arguments.output}" - ) - - return 0 +# def main() -> int: +# initialize_logger() +# logger = logging.getLogger("IAM-APE") +# +# arg_parser = build_arg_parser() +# arguments = arg_parser.parse_args() +# +# if not arguments.quiet: +# print_banner() +# +# if arguments.verbose: +# logging.root.setLevel(logging.DEBUG) +# elif arguments.quiet: +# logging.root.setLevel(logging.WARNING) +# +# if not (arguments.arn or arguments.update): +# arg_parser.print_help() +# return 0 +# +# if arguments.update: +# scrape_iam_actions() +# if not arguments.arn: +# return 0 +# +# try: +# entity_type, entity_account = validate_arn(arguments.arn) +# except InvalidArnException as e: +# logger.error(e) +# return -1 +# +# try: +# auth_details = get_auth_details(arguments.input, arguments.profile) +# except (AwsAuthorizationException, ProfileNotFound) as e: +# logger.error(e) +# return -1 +# +# scp_policies = None +# if arguments.input is None or arguments.scp_policy is not None: +# scp_policies = get_scp_policies( +# arguments.scp_policy, arguments.profile, entity_account +# ) +# +# logger.info("Evaluating effective permissions") +# calculator = EffectivePolicyEvaluator( +# authorization_details=auth_details, scp_policies=scp_policies +# ) +# +# try: +# res = calculator.evaluate(arn=arguments.arn, entity_type=entity_type) +# except IamApeException as e: +# logger.error(e) +# return -1 +# out: Union[AwsPolicyType, FinalReportT] +# if arguments.format == "clean": +# out = calculator.policy_expander.shrink_policy(res.allowed_permissions) +# else: +# out = calculator.create_json_report(res) +# +# if arguments.output == "stdout": +# logger.info(f"Effective permissions policy for {arguments.arn}\n") +# print(json.dumps(out, indent=2)) +# else: +# with open(arguments.output, "w") as f: +# json.dump(out, f, indent=2) +# logger.info( +# f"Wrote effective permissions for {arguments.arn} to {arguments.output}" +# ) +# +# return 0 +def main(): + with open("allowed_permissions_debug.json") as f: + raw_data = json.load(f) + + # Reconstruct allowed_permissions with Action objects + allowed_permissions: Dict[str, Set[Action]] = { + key: { + Action( + action=a["action"], + resource=a.get("resource"), + not_resource=a.get("not_resource"), + condition=a.get("condition"), + source=a["source"], + ) + for a in actions + } + for key, actions in raw_data.items() + } + policy_expander = PolicyExpander() + shrunk_policy=policy_expander.shrink_policy(allowed_permissions) + print("~~" * 20) + print(json.dumps(shrunk_policy, indent=2)) if __name__ == "__main__": From d23e9ffef11c6c2be73e96df84433bf0d97e2700 Mon Sep 17 00:00:00 2001 From: Denis Samoilov Date: Wed, 21 May 2025 10:08:59 +0300 Subject: [PATCH 2/6] CIEM-45:remove log --- iam-ape/iam_ape/expand_policy.py | 1 - 1 file changed, 1 deletion(-) diff --git a/iam-ape/iam_ape/expand_policy.py b/iam-ape/iam_ape/expand_policy.py index 8e4e1eb..5caf640 100644 --- a/iam-ape/iam_ape/expand_policy.py +++ b/iam-ape/iam_ape/expand_policy.py @@ -448,6 +448,5 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: policy_res["Statement"] = [admin_statement] else: policy_res["Statement"] = list(final_statements.values()) - print("Before normalize_policy:", policy_res) return normalize_policy(policy_res) From 2604e66df79020f98db5e962bf0ea64f0a4b2482 Mon Sep 17 00:00:00 2001 From: Denis Samoilov Date: Wed, 21 May 2025 10:09:46 +0300 Subject: [PATCH 3/6] CIEM-45:remove log --- iam-ape/iam_ape/main.py | 166 +++++++++++++++++----------------------- 1 file changed, 72 insertions(+), 94 deletions(-) diff --git a/iam-ape/iam_ape/main.py b/iam-ape/iam_ape/main.py index 3222ecb..97a1881 100644 --- a/iam-ape/iam_ape/main.py +++ b/iam-ape/iam_ape/main.py @@ -4,7 +4,7 @@ import os import re import sys -from typing import Any, Dict, List, Optional, Tuple, Union, Set +from typing import Any, Dict, List, Optional, Tuple, Union import boto3 from botocore.exceptions import ClientError, ProfileNotFound @@ -16,8 +16,7 @@ IamApeException, InvalidArnException, ) -from iam_ape.expand_policy import PolicyExpander -from iam_ape.helper_classes import PolicyWithSource, Action +from iam_ape.helper_classes import PolicyWithSource from iam_ape.helper_functions import deep_update from iam_ape.helper_types import AwsPolicyType, EntityType, FinalReportT @@ -259,97 +258,76 @@ def build_arg_parser() -> argparse.ArgumentParser: return arg_parser -# def main() -> int: -# initialize_logger() -# logger = logging.getLogger("IAM-APE") -# -# arg_parser = build_arg_parser() -# arguments = arg_parser.parse_args() -# -# if not arguments.quiet: -# print_banner() -# -# if arguments.verbose: -# logging.root.setLevel(logging.DEBUG) -# elif arguments.quiet: -# logging.root.setLevel(logging.WARNING) -# -# if not (arguments.arn or arguments.update): -# arg_parser.print_help() -# return 0 -# -# if arguments.update: -# scrape_iam_actions() -# if not arguments.arn: -# return 0 -# -# try: -# entity_type, entity_account = validate_arn(arguments.arn) -# except InvalidArnException as e: -# logger.error(e) -# return -1 -# -# try: -# auth_details = get_auth_details(arguments.input, arguments.profile) -# except (AwsAuthorizationException, ProfileNotFound) as e: -# logger.error(e) -# return -1 -# -# scp_policies = None -# if arguments.input is None or arguments.scp_policy is not None: -# scp_policies = get_scp_policies( -# arguments.scp_policy, arguments.profile, entity_account -# ) -# -# logger.info("Evaluating effective permissions") -# calculator = EffectivePolicyEvaluator( -# authorization_details=auth_details, scp_policies=scp_policies -# ) -# -# try: -# res = calculator.evaluate(arn=arguments.arn, entity_type=entity_type) -# except IamApeException as e: -# logger.error(e) -# return -1 -# out: Union[AwsPolicyType, FinalReportT] -# if arguments.format == "clean": -# out = calculator.policy_expander.shrink_policy(res.allowed_permissions) -# else: -# out = calculator.create_json_report(res) -# -# if arguments.output == "stdout": -# logger.info(f"Effective permissions policy for {arguments.arn}\n") -# print(json.dumps(out, indent=2)) -# else: -# with open(arguments.output, "w") as f: -# json.dump(out, f, indent=2) -# logger.info( -# f"Wrote effective permissions for {arguments.arn} to {arguments.output}" -# ) -# -# return 0 -def main(): - with open("allowed_permissions_debug.json") as f: - raw_data = json.load(f) - - # Reconstruct allowed_permissions with Action objects - allowed_permissions: Dict[str, Set[Action]] = { - key: { - Action( - action=a["action"], - resource=a.get("resource"), - not_resource=a.get("not_resource"), - condition=a.get("condition"), - source=a["source"], - ) - for a in actions - } - for key, actions in raw_data.items() - } - policy_expander = PolicyExpander() - shrunk_policy=policy_expander.shrink_policy(allowed_permissions) - print("~~" * 20) - print(json.dumps(shrunk_policy, indent=2)) +def main() -> int: + initialize_logger() + logger = logging.getLogger("IAM-APE") + + arg_parser = build_arg_parser() + arguments = arg_parser.parse_args() + + if not arguments.quiet: + print_banner() + + if arguments.verbose: + logging.root.setLevel(logging.DEBUG) + elif arguments.quiet: + logging.root.setLevel(logging.WARNING) + + if not (arguments.arn or arguments.update): + arg_parser.print_help() + return 0 + + if arguments.update: + scrape_iam_actions() + if not arguments.arn: + return 0 + + try: + entity_type, entity_account = validate_arn(arguments.arn) + except InvalidArnException as e: + logger.error(e) + return -1 + + try: + auth_details = get_auth_details(arguments.input, arguments.profile) + except (AwsAuthorizationException, ProfileNotFound) as e: + logger.error(e) + return -1 + + scp_policies = None + if arguments.input is None or arguments.scp_policy is not None: + scp_policies = get_scp_policies( + arguments.scp_policy, arguments.profile, entity_account + ) + + logger.info("Evaluating effective permissions") + calculator = EffectivePolicyEvaluator( + authorization_details=auth_details, scp_policies=scp_policies + ) + + try: + res = calculator.evaluate(arn=arguments.arn, entity_type=entity_type) + except IamApeException as e: + logger.error(e) + return -1 + + out: Union[AwsPolicyType, FinalReportT] + if arguments.format == "clean": + out = calculator.policy_expander.shrink_policy(res.allowed_permissions) + else: + out = calculator.create_json_report(res) + + if arguments.output == "stdout": + logger.info(f"Effective permissions policy for {arguments.arn}\n") + print(json.dumps(out, indent=2)) + else: + with open(arguments.output, "w") as f: + json.dump(out, f, indent=2) + logger.info( + f"Wrote effective permissions for {arguments.arn} to {arguments.output}" + ) + + return 0 if __name__ == "__main__": From 5046a97786ada2c1eeca65f35a9ca1d57ed4a8d4 Mon Sep 17 00:00:00 2001 From: Denis Samoilov Date: Wed, 21 May 2025 10:27:15 +0300 Subject: [PATCH 4/6] mypy --- iam-ape/iam_ape/expand_policy.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/iam-ape/iam_ape/expand_policy.py b/iam-ape/iam_ape/expand_policy.py index 5caf640..4cc0599 100644 --- a/iam-ape/iam_ape/expand_policy.py +++ b/iam-ape/iam_ape/expand_policy.py @@ -349,7 +349,7 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: merged_actions.add(actions_list[0]) else: # Multiple actions with same source/action/resource - merge their conditions - merged_condition = {} + merged_condition: Dict[str, Any] = {} for action in actions_list: if action.condition: for operator, operator_conditions in action.condition.items(): @@ -385,7 +385,8 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: for resource_notresource, condition_actions in by_resource_notresource.items(): resource, notresource = resource_notresource for condition, actions_set in condition_actions.items(): - key = frozenset(actions_set), condition, notresource + key: Tuple[FrozenSet[str], Optional[HashableDict], Optional[str]] = ( + frozenset(actions_set), condition, notresource) if resource: by_action_condition_notresource[key].add(resource) elif not by_action_condition_notresource.get(key): From 6dcf99fa8d542c0381375601796712e5b31af29f Mon Sep 17 00:00:00 2001 From: Denis Samoilov Date: Wed, 21 May 2025 10:33:20 +0300 Subject: [PATCH 5/6] mypy --- iam-ape/iam_ape/expand_policy.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/iam-ape/iam_ape/expand_policy.py b/iam-ape/iam_ape/expand_policy.py index 4cc0599..ca3a4d9 100644 --- a/iam-ape/iam_ape/expand_policy.py +++ b/iam-ape/iam_ape/expand_policy.py @@ -339,8 +339,8 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: ] = defaultdict(list) for action_tuple in squashed_policies: - key = (action_tuple.source, action_tuple.action, action_tuple.resource, action_tuple.not_resource) - actions_by_source_action_resource[key].append(action_tuple) + source_key = (action_tuple.source, action_tuple.action, action_tuple.resource, action_tuple.not_resource) + actions_by_source_action_resource[source_key].append(action_tuple) # Merge conditions for actions from the same source merged_actions: Set[Action] = set() From 996b876999958a25ef9880933fb669d564b980b5 Mon Sep 17 00:00:00 2001 From: Denis Samoilov Date: Wed, 21 May 2025 10:37:12 +0300 Subject: [PATCH 6/6] black --- iam-ape/iam_ape/expand_policy.py | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/iam-ape/iam_ape/expand_policy.py b/iam-ape/iam_ape/expand_policy.py index ca3a4d9..320c513 100644 --- a/iam-ape/iam_ape/expand_policy.py +++ b/iam-ape/iam_ape/expand_policy.py @@ -339,7 +339,12 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: ] = defaultdict(list) for action_tuple in squashed_policies: - source_key = (action_tuple.source, action_tuple.action, action_tuple.resource, action_tuple.not_resource) + source_key = ( + action_tuple.source, + action_tuple.action, + action_tuple.resource, + action_tuple.not_resource, + ) actions_by_source_action_resource[source_key].append(action_tuple) # Merge conditions for actions from the same source @@ -354,8 +359,12 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: if action.condition: for operator, operator_conditions in action.condition.items(): if operator in merged_condition: - if isinstance(merged_condition[operator], dict) and isinstance(operator_conditions, dict): - merged_condition[operator].update(operator_conditions) + if isinstance( + merged_condition[operator], dict + ) and isinstance(operator_conditions, dict): + merged_condition[operator].update( + operator_conditions + ) else: merged_condition[operator] = operator_conditions @@ -365,7 +374,7 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: resource=actions_list[0].resource, not_resource=actions_list[0].not_resource, condition=merged_condition if merged_condition else None, - source=actions_list[0].source + source=actions_list[0].source, ) merged_actions.add(merged_action) @@ -386,7 +395,10 @@ def shrink_policy(self, allow_actions: Dict[str, Set[Action]]) -> AwsPolicyType: resource, notresource = resource_notresource for condition, actions_set in condition_actions.items(): key: Tuple[FrozenSet[str], Optional[HashableDict], Optional[str]] = ( - frozenset(actions_set), condition, notresource) + frozenset(actions_set), + condition, + notresource, + ) if resource: by_action_condition_notresource[key].add(resource) elif not by_action_condition_notresource.get(key):