diff --git a/prowler/providers/aws/services/iam/iam_aws_attached_policy_no_administrative_privileges/iam_aws_attached_policy_no_administrative_privileges.py b/prowler/providers/aws/services/iam/iam_aws_attached_policy_no_administrative_privileges/iam_aws_attached_policy_no_administrative_privileges.py index 2d14c0d6f5..03577c7e25 100644 --- a/prowler/providers/aws/services/iam/iam_aws_attached_policy_no_administrative_privileges/iam_aws_attached_policy_no_administrative_privileges.py +++ b/prowler/providers/aws/services/iam/iam_aws_attached_policy_no_administrative_privileges/iam_aws_attached_policy_no_administrative_privileges.py @@ -1,5 +1,6 @@ from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.iam.iam_client import iam_client +from prowler.providers.aws.services.iam.lib.policy import check_admin_access class iam_aws_attached_policy_no_administrative_privileges(Check): @@ -16,27 +17,9 @@ class iam_aws_attached_policy_no_administrative_privileges(Check): report.status = "PASS" report.status_extended = f"{policy.type} policy {policy.name} is attached but does not allow '*:*' administrative privileges." if policy.document: - # Check the statements, if one includes *:* stop iterating over the rest - if not isinstance(policy.document["Statement"], list): - policy_statements = [policy.document["Statement"]] - else: - policy_statements = policy.document["Statement"] - for statement in policy_statements: - # Check policies with "Effect": "Allow" with "Action": "*" over "Resource": "*". - if ( - statement["Effect"] == "Allow" - and "Action" in statement - and ( - statement["Action"] == "*" - or statement["Action"] == ["*"] - ) - and ( - statement["Resource"] == "*" - or statement["Resource"] == ["*"] - ) - ): - report.status = "FAIL" - report.status_extended = f"{policy.type} policy {policy.name} is attached and allows '*:*' administrative privileges." - break + if check_admin_access(policy.document): + report.status = "FAIL" + report.status_extended = f"{policy.type} policy {policy.name} is attached and allows '*:*' administrative privileges." + break findings.append(report) return findings diff --git a/prowler/providers/aws/services/iam/iam_customer_attached_policy_no_administrative_privileges/iam_customer_attached_policy_no_administrative_privileges.py b/prowler/providers/aws/services/iam/iam_customer_attached_policy_no_administrative_privileges/iam_customer_attached_policy_no_administrative_privileges.py index dc6a1285d5..44dfd0f3c5 100644 --- a/prowler/providers/aws/services/iam/iam_customer_attached_policy_no_administrative_privileges/iam_customer_attached_policy_no_administrative_privileges.py +++ b/prowler/providers/aws/services/iam/iam_customer_attached_policy_no_administrative_privileges/iam_customer_attached_policy_no_administrative_privileges.py @@ -1,5 +1,6 @@ from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.iam.iam_client import iam_client +from prowler.providers.aws.services.iam.lib.policy import check_admin_access class iam_customer_attached_policy_no_administrative_privileges(Check): @@ -16,27 +17,9 @@ class iam_customer_attached_policy_no_administrative_privileges(Check): report.status = "PASS" report.status_extended = f"{policy.type} policy {policy.name} is attached but does not allow '*:*' administrative privileges." if policy.document: - # Check the statements, if one includes *:* stop iterating over the rest - if not isinstance(policy.document["Statement"], list): - policy_statements = [policy.document["Statement"]] - else: - policy_statements = policy.document["Statement"] - for statement in policy_statements: - # Check policies with "Effect": "Allow" with "Action": "*" over "Resource": "*". - if ( - statement["Effect"] == "Allow" - and "Action" in statement - and ( - statement["Action"] == "*" - or statement["Action"] == ["*"] - ) - and ( - statement["Resource"] == "*" - or statement["Resource"] == ["*"] - ) - ): - report.status = "FAIL" - report.status_extended = f"{policy.type} policy {policy.name} is attached and allows '*:*' administrative privileges." - break + if check_admin_access(policy.document): + report.status = "FAIL" + report.status_extended = f"{policy.type} policy {policy.name} is attached and allows '*:*' administrative privileges." + break findings.append(report) return findings diff --git a/prowler/providers/aws/services/iam/iam_customer_unattached_policy_no_administrative_privileges/iam_customer_unattached_policy_no_administrative_privileges.py b/prowler/providers/aws/services/iam/iam_customer_unattached_policy_no_administrative_privileges/iam_customer_unattached_policy_no_administrative_privileges.py index f37c7538dd..f01b564d02 100644 --- a/prowler/providers/aws/services/iam/iam_customer_unattached_policy_no_administrative_privileges/iam_customer_unattached_policy_no_administrative_privileges.py +++ b/prowler/providers/aws/services/iam/iam_customer_unattached_policy_no_administrative_privileges/iam_customer_unattached_policy_no_administrative_privileges.py @@ -1,5 +1,6 @@ from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.iam.iam_client import iam_client +from prowler.providers.aws.services.iam.lib.policy import check_admin_access class iam_customer_unattached_policy_no_administrative_privileges(Check): @@ -16,27 +17,9 @@ class iam_customer_unattached_policy_no_administrative_privileges(Check): report.status = "PASS" report.status_extended = f"{policy.type} policy {policy.name} is unattached and does not allow '*:*' administrative privileges." if policy.document: - # Check the statements, if one includes *:* stop iterating over the rest - if not isinstance(policy.document["Statement"], list): - policy_statements = [policy.document["Statement"]] - else: - policy_statements = policy.document["Statement"] - for statement in policy_statements: - # Check policies with "Effect": "Allow" with "Action": "*" over "Resource": "*". - if ( - statement["Effect"] == "Allow" - and "Action" in statement - and ( - statement["Action"] == "*" - or statement["Action"] == ["*"] - ) - and ( - statement["Resource"] == "*" - or statement["Resource"] == ["*"] - ) - ): - report.status = "FAIL" - report.status_extended = f"{policy.type} policy {policy.name} is unattached and allows '*:*' administrative privileges." - break + if check_admin_access(policy.document): + report.status = "FAIL" + report.status_extended = f"{policy.type} policy {policy.name} is unattached and allows '*:*' administrative privileges." + break findings.append(report) return findings diff --git a/prowler/providers/aws/services/iam/iam_inline_policy_no_administrative_privileges/iam_inline_policy_no_administrative_privileges.py b/prowler/providers/aws/services/iam/iam_inline_policy_no_administrative_privileges/iam_inline_policy_no_administrative_privileges.py index 1a0abf1388..daab80b03d 100644 --- a/prowler/providers/aws/services/iam/iam_inline_policy_no_administrative_privileges/iam_inline_policy_no_administrative_privileges.py +++ b/prowler/providers/aws/services/iam/iam_inline_policy_no_administrative_privileges/iam_inline_policy_no_administrative_privileges.py @@ -1,5 +1,6 @@ from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.iam.iam_client import iam_client +from prowler.providers.aws.services.iam.lib.policy import check_admin_access class iam_inline_policy_no_administrative_privileges(Check): @@ -18,28 +19,8 @@ class iam_inline_policy_no_administrative_privileges(Check): resource_attached = report.resource_arn.split("/")[-1] report.status_extended = f"{policy.type} policy {policy.name} attached to {resource_type_str} {resource_attached} does not allow '*:*' administrative privileges." - if policy.document: - # Check the statements, if one includes *:* stop iterating over the rest - if not isinstance(policy.document["Statement"], list): - policy_statements = [policy.document["Statement"]] - else: - policy_statements = policy.document["Statement"] - for statement in policy_statements: - # Check policies with "Effect": "Allow" with "Action": "*" over "Resource": "*". - if ( - statement["Effect"] == "Allow" - and "Action" in statement - and ( - statement["Action"] == "*" - or statement["Action"] == ["*"] - ) - and ( - statement["Resource"] == "*" - or statement["Resource"] == ["*"] - ) - ): - report.status = "FAIL" - report.status_extended = f"{policy.type} policy {policy.name} attached to {resource_type_str} {resource_attached} allows '*:*' administrative privileges." - break + if policy.document and check_admin_access(policy.document): + report.status = "FAIL" + report.status_extended = f"{policy.type} policy {policy.name} attached to {resource_type_str} {resource_attached} allows '*:*' administrative privileges." findings.append(report) return findings diff --git a/prowler/providers/aws/services/iam/lib/policy.py b/prowler/providers/aws/services/iam/lib/policy.py index ab5b716d57..27ef05c30f 100644 --- a/prowler/providers/aws/services/iam/lib/policy.py +++ b/prowler/providers/aws/services/iam/lib/policy.py @@ -1,6 +1,7 @@ from ipaddress import ip_address, ip_network from prowler.lib.logger import logger +from prowler.providers.aws.aws_provider import read_aws_regions_file def is_policy_cross_account(policy: dict, audited_account: str) -> bool: @@ -174,3 +175,110 @@ def is_condition_restricting_from_private_ip(condition_statement: dict) -> bool: ) return is_from_private_ip + + +def process_actions(effect, actions, target_set): + """ + process_actions processes the actions in the policy. + Args: + effect (str): The effect of the policy. + actions (str or list): The actions to process. + target_set (set): The set to store the actions. + """ + if effect in ["Allow", "Deny"] and actions: + if isinstance(actions, str): + target_set.add(actions) + elif isinstance(actions, list): + target_set.update(actions) + + +def check_admin_access(policy: dict) -> bool: + """ + check_admin_access checks if the policy allows admin access. + Args: + policy (dict): The policy to check. + Returns: + bool: True if the policy allows admin access, False otherwise. + """ + + if policy: + allowed_actions = set() + allowed_not_actions = set() + denied_actions = set() + denied_not_actions = set() + + statements = policy.get("Statement", []) + if not isinstance(statements, list): + statements = [statements] + + for statement in statements: + if statement.get("Resource") in [ + "*", + ["*"], + ["*/*"], + "*/*", + ["*:*"], + "*:*", + ] or ( + statement.get("NotResource") + and statement.get("NotResource") not in ["*", ["*"]] + ): + effect = statement.get("Effect") + actions = statement.get("Action") + not_actions = statement.get("NotAction") + if effect == "Allow": + process_actions(effect, actions, allowed_actions) + process_actions(effect, not_actions, allowed_not_actions) + elif effect == "Deny": + process_actions(effect, actions, denied_actions) + process_actions(effect, not_actions, denied_not_actions) + + # If there is only NotAction, it allows the rest of the actions + if not allowed_actions and allowed_not_actions: + allowed_actions.add("*") + # Check for invalid services in allowed NotAction + if allowed_not_actions: + invalid_not_actions = check_invalid_not_actions(allowed_not_actions) + if invalid_not_actions: + # Since it is an invalid NotAction, it allows all AWS actions + allowed_actions.add("*") + + if "*" in allowed_actions: + return True + return False + + +def check_invalid_not_actions(not_actions): + """ + Checks if the actions in NotAction have services that are not part of AWS. + Args: + not_actions (str or list): The NotAction to check. + Returns: + dict: A dictionary with invalid services and their actions. + """ + invalid_services = {} + + if isinstance(not_actions, str): + not_actions = [not_actions] + + for action in not_actions: + service = action.split(":")[0] + if not is_valid_aws_service(service): + if service not in invalid_services: + invalid_services[service] = [] + invalid_services[service].append(action) + + return invalid_services + + +def is_valid_aws_service(service): + """ + Checks if a service is a valid AWS service using aws_regions_by_service.json. + Args: + service (str): The service to check. + Returns: + bool: True if the service is valid, False otherwise. + """ + if service in read_aws_regions_file()["services"]: + return True + return False diff --git a/prowler/providers/aws/services/iam/lib/privilege_escalation.py b/prowler/providers/aws/services/iam/lib/privilege_escalation.py index 35b1b8dc35..3ed2b67093 100644 --- a/prowler/providers/aws/services/iam/lib/privilege_escalation.py +++ b/prowler/providers/aws/services/iam/lib/privilege_escalation.py @@ -1,4 +1,8 @@ from prowler.lib.logger import logger +from prowler.providers.aws.services.iam.lib.policy import ( + check_invalid_not_actions, + process_actions, +) # Does the tool analyze both users and roles, or just one or the other? --> Everything using AttachementCount. # Does the tool take a principal-centric or policy-centric approach? --> Policy-centric approach. @@ -7,6 +11,7 @@ from prowler.lib.logger import logger # Does the tool handle transitive privesc paths (i.e., attack chains)? --> Not yet. # Does the tool handle the DENY effect as expected? --> Yes, it checks DENY's statements with Action and NotAction. # Does the tool handle NotAction as expected? --> Yes +# Does the tool handle NotAction with invalid actions as expected? --> Yes # Does the tool handle Condition constraints? --> Not yet. # Does the tool handle service control policy (SCP) restrictions? --> No, SCP are within Organizations AWS API. @@ -89,13 +94,17 @@ privilege_escalation_policies_combination = { def find_privilege_escalation_combinations( - allowed_actions: set, denied_actions: set, denied_not_actions: set + allowed_actions: set, + denied_actions: set, + allowed_not_actions: set, + denied_not_actions: set, ) -> set: """ find_privilege_escalation_combinations finds the privilege escalation combinations. Args: allowed_actions (set): The allowed actions. denied_actions (set): The denied actions. + allowed_not_actions (set): The allowed not actions. denied_not_actions (set): The denied not actions. Returns: set: The privilege escalation combinations. @@ -103,27 +112,33 @@ def find_privilege_escalation_combinations( # Store all the action's combinations policies_combination = set() + hard_allowed_not_actions = set() try: - # First, we need to perform a left join with ALLOWED_ACTIONS and DENIED_ACTIONS - left_actions = allowed_actions.difference(denied_actions) - # Then, we need to find the DENIED_NOT_ACTIONS in LEFT_ACTIONS + # First, we need to perform a difference with allowed_actions and denied_actions + allowed_actions = allowed_actions.difference(denied_actions) + # Then, we need to do perform a difference with allowed_not_actions and denied_not_actions + allowed_not_actions = allowed_not_actions.difference(denied_not_actions) + # If there are allowed_not_actions, we have to check if there are allowed_actions that are not allowed by allowed_not_actions + if allowed_not_actions: + # If allowed_actions is *, we need to save allowed_not_actions since we cannot subtract them + if "*" in allowed_actions: + hard_allowed_not_actions = allowed_not_actions + else: + allowed_actions = allowed_actions - allowed_not_actions + # If there are denied_not_actions, means that every other action is denied if denied_not_actions: - privileged_actions = left_actions.intersection(denied_not_actions) - # If there is no Denied Not Actions - else: - privileged_actions = left_actions - + allowed_actions = allowed_actions.intersection(denied_not_actions) for values in privilege_escalation_policies_combination.values(): for val in values: val_set = set() val_set.add(val) # Look for specific api:action - if privileged_actions.intersection(val_set) == val_set: + if allowed_actions.intersection(val_set) == val_set: policies_combination.add(val) # Look for api:* else: - for permission in privileged_actions: + for permission in allowed_actions: # Here we have to handle if the api-action is admin, so "*" api_action = permission.split(":") # len() == 2, so api:action @@ -138,10 +153,15 @@ def find_privilege_escalation_combinations( # len() == 1, so * elif len(api_action) == 1: - api = api_action[0] - # Add permissions if the API is present - if api == "*": - policies_combination.add(val) + # Unless the action is *, we have to check if the action to evaluate is in the hard_allowed_not_actions + if ( + not hard_allowed_not_actions + or val not in hard_allowed_not_actions + ): + api = api_action[0] + # Add permissions if the API is present + if api == "*": + policies_combination.add(val) except Exception as error: logger.error( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" @@ -150,21 +170,6 @@ def find_privilege_escalation_combinations( return policies_combination -def process_actions(effect, actions, target_set): - """ - process_actions processes the actions in the policy. - Args: - effect (str): The effect of the policy. - actions (str or list): The actions to process. - target_set (set): The set to store the actions. - """ - if effect in ["Allow", "Deny"] and actions: - if isinstance(actions, str): - target_set.add(actions) - elif isinstance(actions, list): - target_set.update(actions) - - def check_privilege_escalation(policy: dict) -> str: """ check_privilege_escalation checks if the policy allows privilege escalation. @@ -178,6 +183,7 @@ def check_privilege_escalation(policy: dict) -> str: if policy: allowed_actions = set() + allowed_not_actions = set() denied_actions = set() denied_not_actions = set() @@ -192,16 +198,26 @@ def check_privilege_escalation(policy: dict) -> str: if effect == "Allow": process_actions(effect, actions, allowed_actions) - process_actions(effect, not_actions, denied_not_actions) + process_actions(effect, not_actions, allowed_not_actions) elif effect == "Deny": process_actions(effect, actions, denied_actions) process_actions(effect, not_actions, denied_not_actions) + # If there is only NotAction, it allows the rest of the actions + if not allowed_actions and allowed_not_actions: + allowed_actions.add("*") + # Check for invalid services in allowed NotAction + if allowed_not_actions: + invalid_not_actions = check_invalid_not_actions(allowed_not_actions) + if invalid_not_actions: + # Since it is an invalid NotAction, it allows all AWS actions + allowed_actions.add("*") + policies_combination = find_privilege_escalation_combinations( - allowed_actions, denied_actions, denied_not_actions + allowed_actions, denied_actions, allowed_not_actions, denied_not_actions ) - # Check all policies combinations and see if matchs with some combo key + # Check all policies combinations and see if matches with some combo key combos = set() for ( key, diff --git a/tests/providers/aws/services/iam/iam_inline_policy_allows_privilege_escalation/iam_inline_policy_allows_privilege_escalation_test.py b/tests/providers/aws/services/iam/iam_inline_policy_allows_privilege_escalation/iam_inline_policy_allows_privilege_escalation_test.py index 93c016ef4f..fe32dc63c8 100644 --- a/tests/providers/aws/services/iam/iam_inline_policy_allows_privilege_escalation/iam_inline_policy_allows_privilege_escalation_test.py +++ b/tests/providers/aws/services/iam/iam_inline_policy_allows_privilege_escalation/iam_inline_policy_allows_privilege_escalation_test.py @@ -1115,3 +1115,60 @@ class Test_iam_inline_policy_allows_privilege_escalation: finding.status_extended == f"Inline Policy '{policy_name_1}' attached to role {role_arn} does not allow privilege escalation." ) + + @mock_aws + def test_iam_policy_random_not_action(self): + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + iam_client = client("iam", region_name=AWS_REGION_US_EAST_1) + role_name = "test_role" + role_arn = iam_client.create_role( + RoleName=role_name, + AssumeRolePolicyDocument=dumps(ADMINISTRATOR_ROLE_ASSUME_ROLE_POLICY), + )["Role"]["Arn"] + + policy_name = "privileged_policy_1" + policy_document = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "NotAction": "prowler:action", + "Resource": "*", + }, + ], + } + + _ = iam_client.put_role_policy( + RoleName=role_name, + PolicyName=policy_name, + PolicyDocument=dumps(policy_document), + ) + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + from prowler.providers.aws.services.iam.iam_service import IAM + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.iam.iam_inline_policy_allows_privilege_escalation.iam_inline_policy_allows_privilege_escalation.iam_client", + new=IAM(aws_provider), + ): + from prowler.providers.aws.services.iam.iam_inline_policy_allows_privilege_escalation.iam_inline_policy_allows_privilege_escalation import ( + iam_inline_policy_allows_privilege_escalation, + ) + + check = iam_inline_policy_allows_privilege_escalation() + result = check.execute() + assert len(result) == 1 + for finding in result: + if finding.resource_id == policy_name: + assert finding.status == "FAIL" + assert finding.resource_id == policy_name + assert finding.resource_arn == role_arn + assert finding.region == AWS_REGION_US_EAST_1 + assert finding.resource_tags == [] + assert search( + f"Inline Policy '{policy_name}' attached to role {role_arn} allows privilege escalation using the following actions:", + finding.status_extended, + ) diff --git a/tests/providers/aws/services/iam/iam_policy_allows_privilege_escalation/iam_policy_allows_privilege_escalation_test.py b/tests/providers/aws/services/iam/iam_policy_allows_privilege_escalation/iam_policy_allows_privilege_escalation_test.py index 41fe22aaca..5b0e747f41 100644 --- a/tests/providers/aws/services/iam/iam_policy_allows_privilege_escalation/iam_policy_allows_privilege_escalation_test.py +++ b/tests/providers/aws/services/iam/iam_policy_allows_privilege_escalation/iam_policy_allows_privilege_escalation_test.py @@ -989,3 +989,56 @@ class Test_iam_policy_allows_privilege_escalation: finding.status_extended == f"Custom Policy {policy_arn_1} does not allow privilege escalation." ) + + @mock_aws + def test_iam_policy_random_not_action(self): + iam_client = client("iam", region_name=AWS_REGION_US_EAST_1) + policy_name = "policy1" + policy_document = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "NotAction": "prowler:action", + "Resource": "*", + }, + ], + } + policy_arn = iam_client.create_policy( + PolicyName=policy_name, PolicyDocument=dumps(policy_document) + )["Policy"]["Arn"] + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + from prowler.providers.aws.services.iam.iam_service import IAM + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.iam.iam_policy_allows_privilege_escalation.iam_policy_allows_privilege_escalation.iam_client", + new=IAM(aws_provider), + ): + from prowler.providers.aws.services.iam.iam_policy_allows_privilege_escalation.iam_policy_allows_privilege_escalation import ( + iam_policy_allows_privilege_escalation, + ) + + check = iam_policy_allows_privilege_escalation() + result = check.execute() + assert len(result) == 1 + for finding in result: + if finding.resource_id == policy_name: + assert finding.status == "FAIL" + assert finding.resource_id == policy_name + assert finding.resource_arn == policy_arn + assert finding.region == AWS_REGION_US_EAST_1 + assert finding.resource_tags == [] + assert search( + f"Custom Policy {policy_arn} allows privilege escalation using the following actions:", + finding.status_extended, + ) + # Since the policy is admin all the possible privilege escalation paths should be present + for permissions in privilege_escalation_policies_combination: + for permission in privilege_escalation_policies_combination[ + permissions + ]: + assert search(permission, finding.status_extended) diff --git a/tests/providers/aws/services/iam/lib/policy_test.py b/tests/providers/aws/services/iam/lib/policy_test.py index a9ce8312dd..c6be021691 100644 --- a/tests/providers/aws/services/iam/lib/policy_test.py +++ b/tests/providers/aws/services/iam/lib/policy_test.py @@ -1,4 +1,5 @@ from prowler.providers.aws.services.iam.lib.policy import ( + check_admin_access, check_full_service_access, is_condition_restricting_from_private_ip, is_policy_cross_account, @@ -277,3 +278,49 @@ class Test_Policy: "IpAddress": {"aws:SourceIp": "256.256.256.256"}, } assert not is_condition_restricting_from_private_ip(condition_from_invalid_ip) + + def test_check_admin_access(self): + policy = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": ["*"], "Resource": "*"}], + } + assert check_admin_access(policy) + + def test_check_admin_access_false(self): + policy = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": ["s3:*"], "Resource": "*"}], + } + assert not check_admin_access(policy) + + def test_check_admin_access_not_action(self): + policy = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "NotAction": "s3:*", "Resource": "*"}], + } + assert check_admin_access(policy) + + def test_check_admin_access_not_action_with_random_action(self): + policy = { + "Version": "2012-10-17", + "Statement": [ + {"Effect": "Allow", "NotAction": "prowler:action", "Resource": "*"} + ], + } + assert check_admin_access(policy) + + def test_check_admin_access_not_resource(self): + policy = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": "*", "NotResource": "*"}], + } + assert not check_admin_access(policy) + + def test_check_admin_access_not_resource_with_random_resource(self): + policy = { + "Version": "2012-10-17", + "Statement": [ + {"Effect": "Allow", "Action": "*", "NotResource": "prowler:resource"} + ], + } + assert check_admin_access(policy) diff --git a/tests/providers/aws/services/iam/lib/privilege_escalation_test.py b/tests/providers/aws/services/iam/lib/privilege_escalation_test.py index 7349b5dda1..fbd55742d5 100644 --- a/tests/providers/aws/services/iam/lib/privilege_escalation_test.py +++ b/tests/providers/aws/services/iam/lib/privilege_escalation_test.py @@ -8,6 +8,7 @@ class Test_PrivilegeEscalation: def test_find_privilege_escalation_combinations_no_priv_escalation(self): allowed_actions = set() denied_actions = set() + allowed_not_actions = set() denied_not_actions = set() allowed_actions.add("s3:GetObject") @@ -16,7 +17,7 @@ class Test_PrivilegeEscalation: assert ( find_privilege_escalation_combinations( - allowed_actions, denied_actions, denied_not_actions + allowed_actions, denied_actions, allowed_not_actions, denied_not_actions ) == set() ) @@ -26,13 +27,14 @@ class Test_PrivilegeEscalation: ): allowed_actions = set() denied_actions = set() + allowed_not_actions = set() denied_not_actions = set() allowed_actions.add("iam:*") denied_actions.add("ec2:RunInstances") assert find_privilege_escalation_combinations( - allowed_actions, denied_actions, denied_not_actions + allowed_actions, denied_actions, allowed_not_actions, denied_not_actions ) == { "iam:Put*", "iam:AddUserToGroup", @@ -54,13 +56,14 @@ class Test_PrivilegeEscalation: def test_find_privilege_escalation_combinations_priv_escalation_iam_PassRole(self): allowed_actions = set() + allowed_not_actions = set() denied_actions = set() denied_not_actions = set() allowed_actions.add("iam:PassRole") assert find_privilege_escalation_combinations( - allowed_actions, denied_actions, denied_not_actions + allowed_actions, denied_actions, allowed_not_actions, denied_not_actions ) == {"iam:PassRole"} def test_check_privilege_escalation_no_priv_escalation(self): @@ -127,3 +130,70 @@ class Test_PrivilegeEscalation: result = check_privilege_escalation(policy) assert "iam:PassRole" in result + + def test_check_privilege_escalation_priv_escalation_not_action( + self, + ): + policy = { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "Statement1", + "Effect": "Allow", + "NotAction": "iam:Put*", + "Resource": "*", + } + ], + } + + result = check_privilege_escalation(policy) + assert "iam:Put*" not in result + assert "iam:AddUserToGroup" in result + assert "iam:AttachRolePolicy" in result + assert "iam:PassRole" in result + assert "iam:CreateLoginProfile" in result + assert "iam:CreateAccessKey" in result + assert "iam:AttachGroupPolicy" in result + assert "iam:SetDefaultPolicyVersion" in result + assert "iam:PutRolePolicy" in result + assert "iam:UpdateAssumeRolePolicy" in result + assert "iam:*" in result + assert "iam:PutGroupPolicy" in result + assert "iam:PutUserPolicy" in result + assert "iam:CreatePolicyVersion" in result + assert "iam:AttachUserPolicy" in result + assert "iam:UpdateLoginProfile" in result + + def test_check_privilege_escalation_priv_escalation_with_invalid_not_action( + self, + ): + policy = { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "Statement1", + "Effect": "Allow", + "NotAction": "prowler:action", + "Resource": "*", + } + ], + } + + result = check_privilege_escalation(policy) + assert "proler:action" not in result + assert "iam:Put*" in result + assert "iam:AddUserToGroup" in result + assert "iam:AttachRolePolicy" in result + assert "iam:PassRole" in result + assert "iam:CreateLoginProfile" in result + assert "iam:CreateAccessKey" in result + assert "iam:AttachGroupPolicy" in result + assert "iam:SetDefaultPolicyVersion" in result + assert "iam:PutRolePolicy" in result + assert "iam:UpdateAssumeRolePolicy" in result + assert "iam:*" in result + assert "iam:PutGroupPolicy" in result + assert "iam:PutUserPolicy" in result + assert "iam:CreatePolicyVersion" in result + assert "iam:AttachUserPolicy" in result + assert "iam:UpdateLoginProfile" in result