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

fix(efs): check all public conditions (#3872)

Browse Source
This commit is contained in:
Sergio Garcia
2024-04-30 13:08:05 +02:00
committed by GitHub
parent f16857fdf1
commit b54ecb50bf
5 changed files with 282 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible.py
+10 -18
View File
@@ -1,5 +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
class efs_not_publicly_accessible(Check):
@@ -19,23 +20,14 @@ class efs_not_publicly_accessible(Check):
report.status = "FAIL"
report.status_extended = f"EFS {fs.id} doesn't have any policy which means it grants full access to any client."
else:
for statement in fs.policy["Statement"]:
if statement["Effect"] == "Allow":
if (
("Principal" in statement and statement["Principal"] == "*")
or (
"Principal" in statement
and "AWS" in statement["Principal"]
and statement["Principal"]["AWS"] == "*"
)
or (
"CanonicalUser" in statement["Principal"]
and statement["Principal"]["CanonicalUser"] == "*"
)
):
report.status = "FAIL"
report.status_extended = f"EFS {fs.id} has a policy which allows access to everyone."
break
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 everyone."
)
break
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
@@ -0,0 +1,45 @@
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
tests/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible_test.py
+125 -10
View File
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from prowler.providers.aws.services.efs.efs_service import FileSystem
@@ -32,6 +31,48 @@ filesystem_invalid_policy = {
],
}
# https://docs.aws.amazon.com/efs/latest/ug/access-control-block-public-access.html#what-is-a-public-policy
filesystem_policy_with_source_arn_condition = {
"Version": "2012-10-17",
"Id": "efs-policy-wizard-15ad9567-2546-4bbb-8168-5541b6fc0e55",
"Statement": [
{
"Sid": "efs-statement-14a7191c-9401-40e7-a388-6af6cfb7dd9c",
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": [
"elasticfilesystem:ClientMount",
"elasticfilesystem:ClientWrite",
"elasticfilesystem:ClientRootAccess",
],
"Condition": {
"ArnEquals": {
"aws:SourceArn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
}
},
}
],
}
# https://docs.aws.amazon.com/efs/latest/ug/access-control-block-public-access.html#what-is-a-public-policy
filesystem_policy_with_mount_target_condition = {
"Version": "2012-10-17",
"Id": "efs-policy-wizard-15ad9567-2546-4bbb-8168-5541b6fc0e55",
"Statement": [
{
"Sid": "efs-statement-14a7191c-9401-40e7-a388-6af6cfb7dd9c",
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": [
"elasticfilesystem:ClientMount",
"elasticfilesystem:ClientWrite",
"elasticfilesystem:ClientRootAccess",
],
"Condition": {"Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"}},
}
],
}
class Test_efs_not_publicly_accessible:
def test_efs_valid_policy(self):
@@ -59,12 +100,82 @@ class Test_efs_not_publicly_accessible:
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"has a policy which does not allow access to everyone",
result[0].status_extended,
assert (
result[0].status_extended
== f"EFS {file_system_id} has a policy which does not allow access to everyone."
)
assert result[0].resource_id == file_system_id
assert result[0].resource_arn == efs_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
def test_efs_valid_policy_with_mount_target_condition(self):
efs_client = mock.MagicMock
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}"
efs_client.filesystems = [
FileSystem(
id=file_system_id,
arn=efs_arn,
region=AWS_REGION,
policy=filesystem_policy_with_mount_target_condition,
backup_policy=None,
encrypted=True,
)
]
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
):
from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import (
efs_not_publicly_accessible,
)
check = efs_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"EFS {file_system_id} has a policy which does not allow access to everyone."
)
assert result[0].resource_id == file_system_id
assert result[0].resource_arn == efs_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
def test_efs_valid_policy_with_source_arn_condition(self):
efs_client = mock.MagicMock
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}"
efs_client.filesystems = [
FileSystem(
id=file_system_id,
arn=efs_arn,
region=AWS_REGION,
policy=filesystem_policy_with_source_arn_condition,
backup_policy=None,
encrypted=True,
)
]
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
):
from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import (
efs_not_publicly_accessible,
)
check = efs_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"EFS {file_system_id} has a policy which does not allow access to everyone."
)
assert result[0].resource_id == file_system_id
assert result[0].resource_arn == efs_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
def test_efs_invalid_policy(self):
efs_client = mock.MagicMock
@@ -92,12 +203,14 @@ class Test_efs_not_publicly_accessible:
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"has a policy which allows access to everyone",
result[0].status_extended,
assert (
result[0].status_extended
== f"EFS {file_system_id} has a policy which allows access to everyone."
)
assert result[0].resource_id == file_system_id
assert result[0].resource_arn == efs_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
def test_efs_no_policy(self):
efs_client = mock.MagicMock
@@ -124,9 +237,11 @@ class Test_efs_not_publicly_accessible:
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"doesn't have any policy which means it grants full access to any client",
result[0].status_extended,
assert (
result[0].status_extended
== f"EFS {file_system_id} doesn't have any policy which means it grants full access to any client."
)
assert result[0].resource_id == file_system_id
assert result[0].resource_arn == efs_arn
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
tests/providers/aws/services/efs/lib/lib_test.py
+102
View File
@@ -0,0 +1,102 @@
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)