ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-20 11:01:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(aws): improve IAM Resource Policy public logic (#5067)

Browse Source 
Co-authored-by: Rubén De la Torre Vico <rubendltv22@gmail.com>
This commit is contained in:
Sergio Garcia
2024-09-25 08:33:41 -04:00
committed by GitHub
parent ebbccd04f1
commit 0a2b7cf152
20 changed files with 2162 additions and 2041 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/lib/policy_condition_parser/__init__.py
View File
prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py
-157
View File
@@ -1,157 +0,0 @@
def is_condition_block_restrictive(
condition_statement: dict,
source_account: str,
is_cross_account_allowed=False,
):
"""
is_condition_block_restrictive parses the IAM Condition policy block and, by default, returns True if the source_account passed as argument is within, False if not.
If argument is_cross_account_allowed is True it tests if the Condition block includes any of the operators allowlisted returning True if does, False if not.
@param condition_statement: dict with an IAM Condition block, e.g.:
{
"StringLike": {
"AWS:SourceAccount": 111122223333
}
}
@param source_account: str with a 12-digit AWS Account number, e.g.: 111122223333
@param is_cross_account_allowed: bool to allow cross-account access, e.g.: True
"""
is_condition_valid = False
# The conditions must be defined in lowercase since the context key names are not case-sensitive.
# For example, including the aws:SourceAccount context key is equivalent to testing for AWS:SourceAccount
# https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition.html
valid_condition_options = {
"StringEquals": [
"aws:sourceaccount",
"aws:sourceowner",
"s3:resourceaccount",
"aws:principalaccount",
"aws:resourceaccount",
"aws:sourcearn",
"aws:sourcevpc",
"aws:sourcevpce",
],
"StringLike": [
"aws:sourceaccount",
"aws:sourceowner",
"aws:sourcearn",
"aws:principalarn",
"aws:resourceaccount",
"aws:principalaccount",
"aws:sourcevpc",
"aws:sourcevpce",
],
"ArnLike": ["aws:sourcearn", "aws:principalarn"],
"ArnEquals": ["aws:sourcearn", "aws:principalarn"],
}
for condition_operator, condition_operator_key in valid_condition_options.items():
if condition_operator in condition_statement:
for value in condition_operator_key:
# We need to transform the condition_statement into lowercase
condition_statement[condition_operator] = {
k.lower(): v
for k, v in condition_statement[condition_operator].items()
}
if value in condition_statement[condition_operator]:
# values are a list
if isinstance(
condition_statement[condition_operator][value],
list,
):
is_condition_key_restrictive = True
# if cross account is not allowed check for each condition block looking for accounts
# different than default
if not is_cross_account_allowed:
# if there is an arn/account without the source account -> we do not consider it safe
# here by default we assume is true and look for false entries
for item in condition_statement[condition_operator][value]:
if source_account not in item:
is_condition_key_restrictive = False
break
if is_condition_key_restrictive:
is_condition_valid = True
# value is a string
elif isinstance(
condition_statement[condition_operator][value],
str,
):
if is_cross_account_allowed:
is_condition_valid = True
else:
if (
source_account
in condition_statement[condition_operator][value]
):
is_condition_valid = True
return is_condition_valid
def is_condition_block_restrictive_organization(
condition_statement: dict,
):
"""
is_condition_block_restrictive_organization parses the IAM Condition policy block and returns True if the condition_statement is restrictive for the organization, False if not.
@param condition_statement: dict with an IAM Condition block, e.g.:
{
"StringLike": {
"AWS:PrincipalOrgID": "o-111122223333"
}
}
"""
is_condition_valid = False
# The conditions must be defined in lowercase since the context key names are not case-sensitive.
# For example, including the aws:PrincipalOrgID context key is equivalent to testing for AWS:PrincipalOrgID
# https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition.html
valid_condition_options = {
"StringEquals": [
"aws:principalorgid",
],
"StringLike": [
"aws:principalorgid",
],
}
for condition_operator, condition_operator_key in valid_condition_options.items():
if condition_operator in condition_statement:
for value in condition_operator_key:
# We need to transform the condition_statement into lowercase
condition_statement[condition_operator] = {
k.lower(): v
for k, v in condition_statement[condition_operator].items()
}
if value in condition_statement[condition_operator]:
# values are a list
if isinstance(
condition_statement[condition_operator][value],
list,
):
is_condition_valid = True
for item in condition_statement[condition_operator][value]:
if item == "*":
is_condition_valid = False
break
# value is a string
elif isinstance(
condition_statement[condition_operator][value],
str,
):
if "*" not in condition_statement[condition_operator][value]:
is_condition_valid = True
return is_condition_valid
prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible.py
+6 -31
View File
@@ -1,5 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.awslambda.awslambda_client import awslambda_client
from prowler.providers.aws.services.iam.lib.policy import is_policy_public
class awslambda_function_not_publicly_accessible(Check):
@@ -14,37 +15,11 @@ class awslambda_function_not_publicly_accessible(Check):
report.status = "PASS"
report.status_extended = f"Lambda function {function.name} has a policy resource-based policy not public."
public_access = False
if function.policy:
for statement in function.policy["Statement"]:
# Only check allow statements
if statement["Effect"] == "Allow" and (
"*" in statement["Principal"]
or (
isinstance(statement["Principal"], dict)
and (
"*" in statement["Principal"].get("AWS", "")
or "*"
in statement["Principal"].get("CanonicalUser", "")
or ( # Check if function can be invoked by other AWS services
(
".amazonaws.com"
in statement["Principal"].get("Service", "")
)
and (
"*" in statement.get("Action", "")
or "InvokeFunction"
in statement.get("Action", "")
)
)
)
)
):
public_access = True
break
if public_access:
if is_policy_public(
function.policy,
awslambda_client.audited_account,
is_cross_account_allowed=True,
):
report.status = "FAIL"
report.status_extended = f"Lambda function {function.name} has a policy resource-based policy with public access."
prowler/providers/aws/services/dynamodb/dynamodb_table_cross_account_access/dynamodb_table_cross_account_access.py
+5 -33
View File
@@ -1,8 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
is_condition_block_restrictive,
)
from prowler.providers.aws.services.dynamodb.dynamodb_client import dynamodb_client
from prowler.providers.aws.services.iam.lib.policy import is_policy_public
class dynamodb_table_cross_account_access(Check):
@@ -20,36 +18,10 @@ class dynamodb_table_cross_account_access(Check):
)
if table.policy:
report.status_extended = f"DynamoDB table {table.name} has a resource-based policy but is not cross account."
cross_account_access = False
policy_statements = table.policy["Statement"]
if isinstance(
policy_statements, dict
): # Normalize single statement to list
policy_statements = [policy_statements]
for statement in policy_statements:
if not cross_account_access:
if statement["Effect"] == "Allow":
if "AWS" in statement["Principal"]:
principals = statement["Principal"]["AWS"]
if not isinstance(principals, list):
principals = [principals]
else:
principals = [statement["Principal"]]
for aws_account in principals:
if (
dynamodb_client.audited_account not in aws_account
or "*" == aws_account
):
cross_account_access = True
# Check if the condition block is restrictive
conditions = statement.get("Condition", {})
if is_condition_block_restrictive(
conditions, dynamodb_client.audited_account
):
cross_account_access = False
else:
break
if cross_account_access:
if is_policy_public(
table.policy,
source_account=dynamodb_client.audited_account,
):
report.status = "FAIL"
report.status_extended = f"DynamoDB table {table.name} has a resource-based policy allowing cross account access."
findings.append(report)
prowler/providers/aws/services/ecr/ecr_repositories_not_publicly_accessible/ecr_repositories_not_publicly_accessible.py
+4 -10
View File
@@ -1,5 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ecr.ecr_client import ecr_client
from prowler.providers.aws.services.iam.lib.policy import is_policy_public
class ecr_repositories_not_publicly_accessible(Check):
@@ -16,16 +17,9 @@ class ecr_repositories_not_publicly_accessible(Check):
report.status_extended = (
f"Repository {repository.name} is not publicly accesible."
)
if repository.policy:
for statement in repository.policy["Statement"]:
if statement["Effect"] == "Allow":
if "*" in statement["Principal"] or (
"AWS" in statement["Principal"]
and "*" in statement["Principal"]["AWS"]
):
report.status = "FAIL"
report.status_extended = f"Repository {repository.name} policy may allow anonymous users to perform actions (Principal: '*')."
break
if is_policy_public(repository.policy):
report.status = "FAIL"
report.status_extended = f"Repository {repository.name} policy may allow anonymous users to perform actions (Principal: '*')."
findings.append(report)
prowler/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible.py
+10 -9
View File
@@ -1,6 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.efs.efs_client import efs_client
from prowler.providers.aws.services.efs.lib.lib import is_public_access_allowed
from prowler.providers.aws.services.iam.lib.policy import is_policy_public
class efs_not_publicly_accessible(Check):
@@ -17,13 +17,14 @@ class efs_not_publicly_accessible(Check):
if not fs.policy:
report.status = "FAIL"
report.status_extended = f"EFS {fs.id} doesn't have any policy which means it grants full access to any client within the VPC."
else:
for statement in fs.policy.get("Statement", []):
if statement.get("Effect") == "Allow" and is_public_access_allowed(
statement
):
report.status = "FAIL"
report.status_extended = f"EFS {fs.id} has a policy which allows access to any client within the VPC."
break
elif is_policy_public(fs.policy) and any(
statement.get("Condition", {})
.get("Bool", {})
.get("elasticfilesystem:AccessedViaMountTarget", "false")
!= "true"
for statement in fs.policy.get("Statement", [])
):
report.status = "FAIL"
report.status_extended = f"EFS {fs.id} has a policy which allows access to any client within the VPC."
findings.append(report)
return findings
prowler/providers/aws/services/efs/lib/__init__.py
View File
prowler/providers/aws/services/efs/lib/lib.py
-45
View File
@@ -1,45 +0,0 @@
def is_public_access_allowed(statement):
"""
Check if the statement allows public access
Args:
statement: dict: Statement from the policy
Returns:
bool: True if the statement allows public access, False otherwise
"""
principal = statement.get("Principal")
if principal == "*" or (isinstance(principal, dict) and "*" in principal.values()):
return not has_secure_conditions(statement)
return False
def has_secure_conditions(statement):
"""
Check if the statement has secure conditions
Args:
statement: dict: Statement from the policy
Returns:
bool: True if the statement has secure conditions, False otherwise
"""
conditions = statement.get("Condition", {})
allowed_conditions = {
"aws:SourceArn",
"aws:SourceVpc",
"aws:SourceVpce",
"aws:SourceOwner",
"aws:SourceAccount",
}
if (
"Bool" in conditions
and conditions["Bool"].get("elasticfilesystem:AccessedViaMountTarget") == "true"
):
return True
# Check for conditions with nested keys
for _, conditions_dict in conditions.items():
for key, value in conditions_dict.items():
if isinstance(value, dict):
if set(value.keys()).intersection(allowed_conditions):
return True
elif key in allowed_conditions:
return True
return False
prowler/providers/aws/services/iam/iam_role_cross_service_confused_deputy_prevention/iam_role_cross_service_confused_deputy_prevention.py
+7 -22
View File
@@ -1,8 +1,6 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
is_condition_block_restrictive,
)
from prowler.providers.aws.services.iam.iam_client import iam_client
from prowler.providers.aws.services.iam.lib.policy import is_policy_public
class iam_role_cross_service_confused_deputy_prevention(Check):
@@ -19,25 +17,12 @@ class iam_role_cross_service_confused_deputy_prevention(Check):
report.resource_tags = role.tags
report.status = "FAIL"
report.status_extended = f"IAM Service Role {role.name} does not prevent against a cross-service confused deputy attack."
for statement in role.assume_role_policy["Statement"]:
if (
statement["Effect"] == "Allow"
and (
"sts:AssumeRole" in statement["Action"]
or "sts:*" in statement["Action"]
or "*" in statement["Action"]
)
# Need to make sure we are checking the part of the assume role policy document that provides a service access
and "Service" in statement["Principal"]
# Check to see if the appropriate condition statements have been implemented
and "Condition" in statement
and is_condition_block_restrictive(
statement["Condition"], iam_client.audited_account
)
):
report.status = "PASS"
report.status_extended = f"IAM Service Role {role.name} prevents against a cross-service confused deputy attack."
break
if not is_policy_public(
role.assume_role_policy,
not_allowed_actions=["sts:AssumeRole", "sts:*"],
):
report.status = "PASS"
report.status_extended = f"IAM Service Role {role.name} prevents against a cross-service confused deputy attack."
findings.append(report)
prowler/providers/aws/services/iam/lib/policy.py
+253 -33
View File
@@ -43,39 +43,6 @@ def is_policy_cross_account(policy: dict, audited_account: str) -> bool:
return False
def is_policy_public(policy: dict) -> bool:
"""
is_policy_public checks if the policy is publicly accessible.
If the "Principal" element value is set to { "AWS": "*" } and the policy statement is not using any Condition clauses to filter the access, the selected policy is publicly accessible.
Args:
policy (dict): The policy to check.
Returns:
bool: True if the policy is publicly accessible, False otherwise.
"""
if policy and "Statement" in policy:
for statement in policy["Statement"]:
if (
"Principal" in statement
and (
"*" == statement["Principal"]
or "arn:aws:iam::*:root" in statement["Principal"]
)
and "Condition" not in statement
):
return True
elif "Principal" in statement and "AWS" in statement["Principal"]:
if isinstance(statement["Principal"]["AWS"], str):
principals = [statement["Principal"]["AWS"]]
else:
principals = statement["Principal"]["AWS"]
for principal_arn in principals:
if (
principal_arn == "*" or principal_arn == "arn:aws:iam::*:root"
) and "Condition" not in statement:
return True
return False
def check_full_service_access(service: str, policy: dict) -> bool:
"""
check_full_service_access checks if the policy allows full access to a service.
@@ -177,6 +144,259 @@ def is_condition_restricting_from_private_ip(condition_statement: dict) -> bool:
return is_from_private_ip
# TODO: Add logic for deny statements
def is_policy_public(
policy: dict,
source_account: str = "",
is_cross_account_allowed=False,
not_allowed_actions: list = [],
) -> bool:
"""
Check if the policy allows public access to the resource.
If the policy gives access to an AWS service principal is considered public if the policy is not pair with conditions since it can be invoked by AWS services in other accounts.
Args:
policy (dict): The AWS policy to check
source_account (str): The account to check if the access is restricted to it, default: ""
is_cross_account_allowed (bool): If the policy can allow cross-account access, default: False
not_allowed_actions (list): List of actions that are not allowed, default: []. If not_allowed_actions is empty, the function will not consider the actions in the policy.
Returns:
bool: True if the policy allows public access, False otherwise
"""
is_public = False
for statement in policy.get("Statement", []):
# Only check allow statements
if statement["Effect"] == "Allow":
principal = statement.get("Principal", "")
if (
"*" in principal
or "arn:aws:iam::*:root" in principal
or (
isinstance(principal, dict)
and (
"*" in principal.get("AWS", "")
or "arn:aws:iam::*:root" in principal.get("AWS", "")
or (
isinstance(principal.get("AWS"), list)
and (
"*" in principal["AWS"]
or "arn:aws:iam::*:root" in principal["AWS"]
)
)
or "*" in principal.get("CanonicalUser", "")
or "arn:aws:iam::*:root" in principal.get("CanonicalUser", "")
or ( # Check if function can be invoked by other AWS services
(
".amazonaws.com" in principal.get("Service", "")
or "*" in principal.get("Service", "")
)
)
)
)
) and (
not not_allowed_actions # If not_allowed_actions is empty, the function will not consider the actions in the policy
or (
statement.get(
"Action"
) # If the statement has no action, it is not public
and (
(
(
isinstance(statement.get("Action", ""), list)
and "*" in statement["Action"]
)
or (
isinstance(statement.get("Action", ""), str)
and statement.get("Action", "") == "*"
)
)
or (
isinstance(statement.get("Action", ""), list)
and any(
action in not_allowed_actions
for action in statement["Action"]
)
)
or (statement.get("Action", "") in not_allowed_actions)
)
)
):
is_public = (
not is_condition_block_restrictive(
statement.get("Condition", {}),
source_account,
is_cross_account_allowed,
)
and not is_condition_block_restrictive_organization(
statement.get("Condition", {})
)
and not is_condition_restricting_from_private_ip(
statement.get("Condition", {})
)
)
if is_public:
break
return is_public
def is_condition_block_restrictive(
condition_statement: dict,
source_account: str = "",
is_cross_account_allowed=False,
):
"""
is_condition_block_restrictive parses the IAM Condition policy block and, by default, returns True if the source_account passed as argument is within, False if not.
If argument is_cross_account_allowed is True it tests if the Condition block includes any of the operators allowlisted returning True if does, False if not.
Args:
condition_statement: dict with an IAM Condition block, e.g.:
{
"StringLike": {
"AWS:SourceAccount": 111122223333
}
}
source_account: str with a 12-digit AWS Account number, e.g.: 111122223333, default: ""
is_cross_account_allowed: bool to allow cross-account access, e.g.: True, default: False
"""
is_condition_valid = False
# The conditions must be defined in lowercase since the context key names are not case-sensitive.
# For example, including the aws:SourceAccount context key is equivalent to testing for AWS:SourceAccount
# https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition.html
valid_condition_options = {
"StringEquals": [
"aws:sourceaccount",
"aws:sourceowner",
"s3:resourceaccount",
"aws:principalaccount",
"aws:resourceaccount",
"aws:sourcearn",
"aws:sourcevpc",
"aws:sourcevpce",
],
"StringLike": [
"aws:sourceaccount",
"aws:sourceowner",
"aws:sourcearn",
"aws:principalarn",
"aws:resourceaccount",
"aws:principalaccount",
"aws:sourcevpc",
"aws:sourcevpce",
],
"ArnLike": ["aws:sourcearn", "aws:principalarn"],
"ArnEquals": ["aws:sourcearn", "aws:principalarn"],
}
for condition_operator, condition_operator_key in valid_condition_options.items():
if condition_operator in condition_statement:
for value in condition_operator_key:
# We need to transform the condition_statement into lowercase
condition_statement[condition_operator] = {
k.lower(): v
for k, v in condition_statement[condition_operator].items()
}
if value in condition_statement[condition_operator]:
# values are a list
if isinstance(
condition_statement[condition_operator][value],
list,
):
is_condition_key_restrictive = True
# if cross account is not allowed check for each condition block looking for accounts
# different than default
if not is_cross_account_allowed:
# if there is an arn/account without the source account -> we do not consider it safe
# here by default we assume is true and look for false entries
for item in condition_statement[condition_operator][value]:
if source_account not in item:
is_condition_key_restrictive = False
break
if is_condition_key_restrictive:
is_condition_valid = True
# value is a string
elif isinstance(
condition_statement[condition_operator][value],
str,
):
if is_cross_account_allowed:
is_condition_valid = True
else:
if (
source_account
in condition_statement[condition_operator][value]
):
is_condition_valid = True
return is_condition_valid
def is_condition_block_restrictive_organization(
condition_statement: dict,
):
"""
is_condition_block_restrictive_organization parses the IAM Condition policy block and returns True if the condition_statement is restrictive for the organization, False if not.
@param condition_statement: dict with an IAM Condition block, e.g.:
{
"StringLike": {
"AWS:PrincipalOrgID": "o-111122223333"
}
}
"""
is_condition_valid = False
# The conditions must be defined in lowercase since the context key names are not case-sensitive.
# For example, including the aws:PrincipalOrgID context key is equivalent to testing for AWS:PrincipalOrgID
# https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition.html
valid_condition_options = {
"StringEquals": [
"aws:principalorgid",
],
"StringLike": [
"aws:principalorgid",
],
}
for condition_operator, condition_operator_key in valid_condition_options.items():
if condition_operator in condition_statement:
for value in condition_operator_key:
# We need to transform the condition_statement into lowercase
condition_statement[condition_operator] = {
k.lower(): v
for k, v in condition_statement[condition_operator].items()
}
if value in condition_statement[condition_operator]:
# values are a list
if isinstance(
condition_statement[condition_operator][value],
list,
):
is_condition_valid = True
for item in condition_statement[condition_operator][value]:
if item == "*":
is_condition_valid = False
break
# value is a string
elif isinstance(
condition_statement[condition_operator][value],
str,
):
if "*" not in condition_statement[condition_operator][value]:
is_condition_valid = True
return is_condition_valid
def process_actions(effect, actions, target_set):
"""
process_actions processes the actions in the policy.
prowler/providers/aws/services/kms/kms_key_not_publicly_accessible/kms_key_not_publicly_accessible.py
+6 -27
View File
@@ -1,4 +1,5 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.iam.lib.policy import is_policy_public
from prowler.providers.aws.services.kms.kms_client import kms_client
@@ -17,32 +18,10 @@ class kms_key_not_publicly_accessible(Check):
report.resource_tags = key.tags
report.region = key.region
# If the "Principal" element value is set to { "AWS": "*" } and the policy statement is not using any Condition clauses to filter the access, the selected AWS KMS master key is publicly accessible.
if key.policy and "Statement" in key.policy:
for statement in key.policy["Statement"]:
if (
"Principal" in statement
and "*" == statement["Principal"]
and "Condition" not in statement
):
report.status = "FAIL"
report.status_extended = (
f"KMS key {key.id} may be publicly accessible."
)
elif (
"Principal" in statement and "AWS" in statement["Principal"]
):
if isinstance(statement["Principal"]["AWS"], str):
principals = [statement["Principal"]["AWS"]]
else:
principals = statement["Principal"]["AWS"]
for principal_arn in principals:
if (
principal_arn == "*"
and "Condition" not in statement
):
report.status = "FAIL"
report.status_extended = (
f"KMS key {key.id} may be publicly accessible."
)
if is_policy_public(key.policy, not_allowed_actions=["kms:*"]):
report.status = "FAIL"
report.status_extended = (
f"KMS key {key.id} may be publicly accessible."
)
findings.append(report)
return findings
prowler/providers/aws/services/s3/s3_bucket_policy_public_write_access/s3_bucket_policy_public_write_access.py
+13 -31
View File
@@ -1,4 +1,5 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.iam.lib.policy import is_policy_public
from prowler.providers.aws.services.s3.s3_client import s3_client
from prowler.providers.aws.services.s3.s3control_client import s3control_client
@@ -37,37 +38,18 @@ class s3_bucket_policy_public_write_access(Check):
else:
report.status = "PASS"
report.status_extended = f"S3 Bucket {bucket.name} does not allow public write access in the bucket policy."
for statement in bucket.policy["Statement"]:
if (
statement["Effect"] == "Allow"
and "Condition" not in statement
and (
"Principal" in statement
and "*" in str(statement["Principal"])
)
and (
(
isinstance(statement["Action"], list)
and (
"s3:PutObject" in statement["Action"]
or "*" in statement["Action"]
or "s3:*" in statement["Action"]
or "s3:Put*" in statement["Action"]
)
)
or (
isinstance(statement["Action"], str)
and (
"s3:PutObject" == statement["Action"]
or "*" == statement["Action"]
or "s3:*" == statement["Action"]
or "s3:Put*" == statement["Action"]
)
)
)
):
report.status = "FAIL"
report.status_extended = f"S3 Bucket {bucket.name} allows public write access in the bucket policy."
if is_policy_public(
bucket.policy,
not_allowed_actions=[
"s3:PutObject",
"s3:DeleteObject",
"s3:*",
"s3:Put*",
"s3:Delete*",
],
):
report.status = "FAIL"
report.status_extended = f"S3 Bucket {bucket.name} allows public write access in the bucket policy."
findings.append(report)
return findings
prowler/providers/aws/services/s3/s3_bucket_public_access/s3_bucket_public_access.py
+4 -40
View File
@@ -1,10 +1,5 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
is_condition_block_restrictive,
)
from prowler.providers.aws.services.iam.lib.policy import (
is_condition_restricting_from_private_ip,
)
from prowler.providers.aws.services.iam.lib.policy import is_policy_public
from prowler.providers.aws.services.s3.s3_client import s3_client
from prowler.providers.aws.services.s3.s3control_client import s3control_client
@@ -51,39 +46,8 @@ class s3_bucket_public_access(Check):
report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket ACL."
# 4. Check bucket policy
if bucket.policy:
for statement in bucket.policy.get("Statement", []):
if (
"Principal" in statement
and statement["Effect"] == "Allow"
and not is_condition_block_restrictive(
statement.get("Condition", {}), "", True
)
and (
not is_condition_restricting_from_private_ip(
statement.get("Condition", {})
)
if statement.get("Condition", {}).get(
"IpAddress", {}
)
else True
)
):
if "*" == statement["Principal"]:
report.status = "FAIL"
report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket policy."
elif "AWS" in statement["Principal"]:
principals = (
statement["Principal"]["AWS"]
if isinstance(
statement["Principal"]["AWS"], list
)
else [statement["Principal"]["AWS"]]
)
for principal_arn in principals:
if principal_arn == "*":
report.status = "FAIL"
report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket policy."
if is_policy_public(bucket.policy):
report.status = "FAIL"
report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket policy."
findings.append(report)
return findings
prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.py
+1 -1
View File
@@ -1,5 +1,5 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
from prowler.providers.aws.services.iam.lib.policy import (
is_condition_block_restrictive,
is_condition_block_restrictive_organization,
)
prowler/providers/aws/services/sqs/sqs_queues_not_publicly_accessible/sqs_queues_not_publicly_accessible.py
+1 -3
View File
@@ -1,7 +1,5 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
is_condition_block_restrictive,
)
from prowler.providers.aws.services.iam.lib.policy import is_condition_block_restrictive
from prowler.providers.aws.services.sqs.sqs_client import sqs_client
prowler/providers/aws/services/vpc/vpc_endpoint_connections_trust_boundaries/vpc_endpoint_connections_trust_boundaries.py
+1 -3
View File
@@ -1,9 +1,7 @@
from re import compile
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
is_condition_block_restrictive,
)
from prowler.providers.aws.services.iam.lib.policy import is_condition_block_restrictive
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
tests/providers/aws/lib/policy_condition_parser/policy_condition_parser_test.py
-1439
View File
File diff suppressed because it is too large Load Diff
tests/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_test.py
+261 -53
View File
@@ -79,7 +79,6 @@ class Test_awslambda_function_not_publicly_accessible:
StatementId="public-access",
Action="lambda:InvokeFunction",
Principal="*",
SourceArn=function_arn,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
@@ -112,6 +111,83 @@ class Test_awslambda_function_not_publicly_accessible:
)
assert result[0].resource_tags == [{"tag1": "value1", "tag2": "value2"}]
@mock_aws
def test_function_public_with_source_account(self):
# Create the mock IAM role
iam_client = client("iam", region_name=AWS_REGION_EU_WEST_1)
role_name = "test-role"
assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
role_arn = iam_client.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=dumps(assume_role_policy_document),
)["Role"]["Arn"]
function_name = "test-lambda"
# Create the lambda function using boto3 client
lambda_client = client("lambda", region_name=AWS_REGION_EU_WEST_1)
function_arn = lambda_client.create_function(
FunctionName=function_name,
Runtime="nodejs4.3",
Role=role_arn,
Handler="index.handler",
Code={"ZipFile": b"fileb://file-path/to/your-deployment-package.zip"},
Description="Test Lambda function",
Timeout=3,
MemorySize=128,
Publish=True,
Tags={"tag1": "value1", "tag2": "value2"},
)["FunctionArn"]
# Attach the policy to the lambda function with a wildcard principal
lambda_client.add_permission(
FunctionName=function_name,
StatementId="non-public-access",
Action="lambda:InvokeFunction",
Principal="*",
SourceArn=function_arn,
SourceAccount=AWS_ACCOUNT_NUMBER,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
from prowler.providers.aws.services.awslambda.awslambda_service import Lambda
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
new=Lambda(aws_provider),
):
# Test Check
from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible import (
awslambda_function_not_publicly_accessible,
)
check = awslambda_function_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == function_name
assert result[0].resource_arn == function_arn
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Lambda function {function_name} has a policy resource-based policy not public."
)
assert result[0].resource_tags == [{"tag1": "value1", "tag2": "value2"}]
@mock_aws
def test_function_not_public(self):
# Create the mock IAM role
@@ -190,6 +266,7 @@ class Test_awslambda_function_not_publicly_accessible:
def test_function_public_with_canonical(self):
lambda_client = mock.MagicMock
lambda_client.audited_account = AWS_ACCOUNT_NUMBER
function_name = "test-lambda"
function_runtime = "nodejs4.3"
function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"
@@ -418,59 +495,190 @@ class Test_awslambda_function_not_publicly_accessible:
)
assert result[0].resource_tags == [{"tag1": "value1", "tag2": "value2"}]
# def test_function_could_be_invoked_by_specific_aws_account(self):
# lambda_client = mock.MagicMock
# function_name = "test-lambda"
# function_runtime = "nodejs4.3"
# function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"
# lambda_policy = { # If there is an ALB or API Gateway in specified AWS Account, the lambda function could be invoked and exposed by them
# "Version": "2012-10-17",
# "Statement": [
# {
# "Sid": "public-access",
# "Principal": {"AWS": AWS_ACCOUNT_NUMBER},
# "Effect": "Allow",
# "Action": [
# "lambda:InvokeFunction",
# ],
# "Resource": [function_arn],
# }
# ],
# }
def test_function_could_be_invoked_by_specific_aws_account(self):
lambda_client = mock.MagicMock
lambda_client.audited_account = AWS_ACCOUNT_NUMBER
function_name = "test-lambda"
function_runtime = "nodejs4.3"
function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"
lambda_policy = {
"Version": "2012-10-17",
"Id": "default",
"Statement": [
{
"Sid": "awslambda-myLambdaScript-LambdaInvokePermission",
"Effect": "Allow",
"Principal": {"Service": "ses.amazonaws.com"},
"Action": "lambda:InvokeFunction",
"Resource": f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function:{function_name}",
"Condition": {
"StringEquals": {"AWS:SourceAccount": AWS_ACCOUNT_NUMBER}
},
}
],
}
# lambda_client.functions = {
# "function_name": Function(
# name=function_name,
# security_groups=[],
# arn=function_arn,
# region=AWS_REGION_EU_WEST_1,
# runtime=function_runtime,
# policy=lambda_policy,
# )
# }
lambda_client.functions = {
"function_name": Function(
name=function_name,
security_groups=[],
arn=function_arn,
region=AWS_REGION_EU_WEST_1,
runtime=function_runtime,
policy=lambda_policy,
)
}
# with mock.patch(
# "prowler.providers.common.provider.Provider.get_global_provider",
# return_value=set_mocked_aws_provider(),
# ), mock.patch(
# "prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
# new=lambda_client,
# ):
# # Test Check
# from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible import (
# awslambda_function_not_publicly_accessible,
# )
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
new=lambda_client,
):
from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible import (
awslambda_function_not_publicly_accessible,
)
# check = awslambda_function_not_publicly_accessible()
# result = check.execute()
check = awslambda_function_not_publicly_accessible()
result = check.execute()
# assert len(result) == 1
# assert result[0].region == AWS_REGION_EU_WEST_1
# assert result[0].resource_id == function_name
# assert result[0].resource_arn == function_arn
# assert result[0].status == "FAIL"
# assert (
# result[0].status_extended
# == f"Lambda function {function_name} has a policy resource-based policy with public access."
# )
# assert result[0].resource_tags == []
assert len(result) == 1
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == function_name
assert result[0].resource_arn == function_arn
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Lambda function {function_name} has a policy resource-based policy not public."
)
assert result[0].resource_tags == []
def test_function_could_be_invoked_by_specific_other_aws_account(self):
lambda_client = mock.MagicMock
lambda_client.audited_account = AWS_ACCOUNT_NUMBER
function_name = "test-lambda"
function_runtime = "nodejs4.3"
function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"
lambda_policy = {
"Version": "2012-10-17",
"Id": "default",
"Statement": [
{
"Sid": "awslambda-myLambdaScript-LambdaInvokePermission",
"Effect": "Allow",
"Principal": {"Service": "ses.amazonaws.com"},
"Action": "lambda:InvokeFunction",
"Resource": f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function:{function_name}",
"Condition": {
"StringEquals": {"AWS:SourceAccount": "000000000000"}
},
}
],
}
lambda_client.functions = {
"function_name": Function(
name=function_name,
security_groups=[],
arn=function_arn,
region=AWS_REGION_EU_WEST_1,
runtime=function_runtime,
policy=lambda_policy,
)
}
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
new=lambda_client,
):
from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible import (
awslambda_function_not_publicly_accessible,
)
check = awslambda_function_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].region == AWS_REGION_EU_WEST_1
assert result[0].resource_id == function_name
assert result[0].resource_arn == function_arn
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Lambda function {function_name} has a policy resource-based policy not public."
)
assert result[0].resource_tags == []
def test_function_public_policy_with_several_statements(self):
lambda_client = mock.MagicMock
lambda_client.audited_account = AWS_ACCOUNT_NUMBER
function_name = "test-lambda"
function_runtime = "nodejs4.3"
function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"
lambda_policy = {
"Version": "2012-10-17",
"Id": "default",
"Statement": [
{
"Sid": "AllowExecutionFromAPIGateway",
"Effect": "Allow",
"Principal": {"Service": "apigateway.amazonaws.com"},
"Action": "lambda:InvokeFunction",
"Resource": f"arn:aws:lambda:eu-central-1:{AWS_ACCOUNT_NUMBER}:function:foo",
"Condition": {
"ArnLike": {
"AWS:SourceArn": f"arn:aws:execute-api:eu-central-1:{AWS_ACCOUNT_NUMBER}:bar/*/GET/proxy+"
}
},
},
{
"Sid": "FunctionURLAllowPublicAccess",
"Effect": "Allow",
"Principal": "*",
"Action": "lambda:InvokeFunctionUrl",
"Resource": f"arn:aws:lambda:eu-central-1:{AWS_ACCOUNT_NUMBER}:function:foo",
"Condition": {
"StringEquals": {"lambda:FunctionUrlAuthType": "NONE"}
},
},
],
}
lambda_client.functions = {
"function_name": Function(
name=function_name,
security_groups=[],
arn=function_arn,
region=AWS_REGION_EU_WEST_1,
runtime=function_runtime,
policy=lambda_policy,
)
}
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_aws_provider(),
), mock.patch(
"prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible.awslambda_client",
new=lambda_client,
):
from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible import (
awslambda_function_not_publicly_accessible,
)
check = awslambda_function_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Lambda function {function_name} has a policy resource-based policy with public access."
)
assert result[0].resource_id == function_name
assert result[0].resource_arn == function_arn
assert result[0].region == AWS_REGION_EU_WEST_1
tests/providers/aws/services/efs/lib/lib_test.py
-102
View File
@@ -1,102 +0,0 @@
from prowler.providers.aws.services.efs.lib.lib import is_public_access_allowed
class Test_EFS_lib:
def test__is_public_access_allowed__(self):
statement = {
"Effect": "Allow",
"Principal": "*",
"Action": "elasticfilesystem:ClientMount",
"Resource": "*",
}
assert is_public_access_allowed(statement)
def test__is_public_access_allowed__with_principal_dict(self):
statement = {
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": "elasticfilesystem:ClientMount",
"Resource": "*",
}
assert is_public_access_allowed(statement)
def test__is_public_access_allowed__with_secure_conditions(self):
statement = {
"Effect": "Allow",
"Principal": "*",
"Action": "elasticfilesystem:ClientMount",
"Resource": "*",
"Condition": {"Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"}},
}
assert not is_public_access_allowed(statement)
def test__is_public_access_allowed__with_secure_conditions_and_allowed_conditions(
self,
):
statement = {
"Effect": "Allow",
"Principal": "*",
"Action": "elasticfilesystem:ClientMount",
"Resource": "*",
"Condition": {
"Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"},
"StringEquals": {"aws:SourceOwner": "123456789012"},
},
}
assert not is_public_access_allowed(statement)
def test__is_public_access_allowed__with_secure_conditions_and_allowed_conditions_nested(
self,
):
statement = {
"Effect": "Allow",
"Principal": "*",
"Action": "elasticfilesystem:ClientMount",
"Resource": "*",
"Condition": {
"Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"},
"StringEquals": {"aws:SourceOwner": "123456789012"},
"StringEqualsIfExists": {"aws:SourceVpce": "vpce-1234567890abcdef0"},
},
}
assert not is_public_access_allowed(statement)
def test__is_public_access_allowed__with_secure_conditions_and_allowed_conditions_nested_dict(
self,
):
statement = {
"Effect": "Allow",
"Principal": "*",
"Action": "elasticfilesystem:ClientMount",
"Resource": "*",
"Condition": {
"Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"},
"StringEquals": {"aws:SourceOwner": "123456789012"},
"StringEqualsIfExists": {
"aws:SourceVpce": {
"vpce-1234567890abcdef0": "vpce-1234567890abcdef0"
}
},
},
}
assert not is_public_access_allowed(statement)
def test__is_public_access_allowed__with_secure_conditions_and_allowed_conditions_nested_dict_key(
self,
):
statement = {
"Effect": "Allow",
"Principal": "*",
"Action": "elasticfilesystem:ClientMount",
"Resource": "*",
"Condition": {
"Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"},
"StringEquals": {"aws:SourceOwner": "123456789012"},
"StringEqualsIfExists": {
"aws:SourceVpce": {
"vpce-1234567890abcdef0": "vpce-1234567890abcdef0"
}
},
},
}
assert not is_public_access_allowed(statement)
tests/providers/aws/services/iam/lib/policy_test.py
+1590 -2
View File
File diff suppressed because it is too large Load Diff