chore(aws): handle NotAction cases in IAM policies (#5035)

This commit is contained in:
Sergio Garcia
2024-09-24 08:36:11 -04:00
committed by GitHub
parent 3951295c0c
commit 25327d618d
10 changed files with 406 additions and 125 deletions
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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,
@@ -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,
)
@@ -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)
@@ -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)
@@ -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