feat(efs): add new check efs_mount_target_not_publicly_accesible (#5275)

Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Mario Rodriguez Lopez
2024-10-04 17:41:51 +02:00
committed by GitHub
parent f2301d5ed6
commit a31b15c26c
12 changed files with 720 additions and 321 deletions
@@ -5,7 +5,7 @@ from prowler.providers.aws.services.efs.efs_client import efs_client
class efs_encryption_at_rest_enabled(Check):
def execute(self):
findings = []
for fs in efs_client.filesystems:
for fs in efs_client.filesystems.values():
report = Check_Report_AWS(self.metadata())
report.region = fs.region
report.resource_id = fs.id
@@ -5,7 +5,7 @@ from prowler.providers.aws.services.efs.efs_client import efs_client
class efs_have_backup_enabled(Check):
def execute(self):
findings = []
for fs in efs_client.filesystems:
for fs in efs_client.filesystems.values():
report = Check_Report_AWS(self.metadata())
report.region = fs.region
report.resource_id = fs.id
@@ -0,0 +1,34 @@
{
"Provider": "aws",
"CheckID": "efs_mount_target_not_publicly_accessible",
"CheckTitle": "EFS mount targets should not be publicly accessible",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "efs",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:elasticfilesystem:{region}:{account-id}:file-system/{filesystem-id}",
"Severity": "medium",
"ResourceType": "AwsEfsFileSystem",
"Description": "This control checks whether an Amazon EFS mount target is associated with a public subnet since it can be accessed from the internet.",
"Risk": "Mount targets in public subnets may expose your EFS to unauthorized access or attacks.",
"RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/efs-mount-target-public-accessible.html",
"Remediation": {
"Code": {
"CLI": "aws efs create-mount-target --file-system-id <filesystem-id> --subnet-id <private-subnet-id> --security-groups <sg-ids>",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/efs-controls.html#efs-6",
"Terraform": ""
},
"Recommendation": {
"Text": "Recreate the EFS mount target in a private subnet to ensure it is not publicly accessible.",
"Url": "https://docs.aws.amazon.com/efs/latest/ug/accessing-fs.html"
}
},
"Categories": [
"internet-exposed"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,28 @@
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.vpc.vpc_client import vpc_client
class efs_mount_target_not_publicly_accessible(Check):
def execute(self):
findings = []
for fs in efs_client.filesystems.values():
report = Check_Report_AWS(self.metadata())
report.region = fs.region
report.resource_id = fs.id
report.resource_arn = fs.arn
report.resource_tags = fs.tags
report.status = "PASS"
report.status_extended = (
f"EFS {fs.id} does not have any public mount targets."
)
mount_targets = []
for mt in fs.mount_targets:
if vpc_client.vpc_subnets[mt.subnet_id].public:
mount_targets.append(mt)
if mount_targets:
report.status = "FAIL"
report.status_extended = f"EFS {fs.id} has public mount targets: {', '.join([mt.id for mt in mount_targets])}"
findings.append(report)
return findings
@@ -6,7 +6,7 @@ from prowler.providers.aws.services.iam.lib.policy import is_policy_public
class efs_not_publicly_accessible(Check):
def execute(self):
findings = []
for fs in efs_client.filesystems:
for fs in efs_client.filesystems.values():
report = Check_Report_AWS(self.metadata())
report.region = fs.region
report.resource_id = fs.id
@@ -14,9 +14,12 @@ class EFS(AWSService):
def __init__(self, provider):
# Call AWSService's __init__
super().__init__(__class__.__name__, provider)
self.filesystems = []
self.filesystems = {}
self.__threading_call__(self._describe_file_systems)
self._describe_file_system_policies()
self.__threading_call__(
self._describe_file_system_policies, self.filesystems.values()
)
self.__threading_call__(self._describe_mount_targets, self.filesystems.values())
def _describe_file_systems(self, regional_client):
logger.info("EFS - Describing file systems...")
@@ -31,55 +34,98 @@ class EFS(AWSService):
if not self.audit_resources or (
is_resource_filtered(efs_arn, self.audit_resources)
):
self.filesystems.append(
FileSystem(
id=efs_id,
arn=efs_arn,
region=regional_client.region,
policy=None,
backup_policy=None,
encrypted=efs["Encrypted"],
tags=efs.get("Tags"),
)
self.filesystems[efs_arn] = FileSystem(
id=efs_id,
arn=efs_arn,
region=regional_client.region,
encrypted=efs["Encrypted"],
tags=efs.get("Tags"),
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _describe_file_system_policies(self):
def _describe_file_system_policies(self, filesystem):
logger.info("EFS - Describing file system policies...")
try:
for filesystem in self.filesystems:
for region, client in self.regional_clients.items():
if filesystem.region == region:
try:
filesystem.backup_policy = client.describe_backup_policy(
FileSystemId=filesystem.id
)["BackupPolicy"]["Status"]
except ClientError as e:
if e.response["Error"]["Code"] == "PolicyNotFound":
filesystem.backup_policy = "DISABLED"
try:
fs_policy = client.describe_file_system_policy(
FileSystemId=filesystem.id
client = self.regional_clients[filesystem.region]
try:
filesystem.backup_policy = client.describe_backup_policy(
FileSystemId=filesystem.id
)["BackupPolicy"]["Status"]
except ClientError as error:
if error.response["Error"]["Code"] == "PolicyNotFound":
filesystem.backup_policy = "DISABLED"
logger.warning(
f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
logger.error(
f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
try:
fs_policy = client.describe_file_system_policy(
FileSystemId=filesystem.id
)
if "Policy" in fs_policy:
filesystem.policy = json.loads(fs_policy["Policy"])
except ClientError as error:
if error.response["Error"]["Code"] == "PolicyNotFound":
filesystem.policy = {}
logger.warning(
f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else:
logger.error(
f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def _describe_mount_targets(self, filesystem):
logger.info("EFS - Describing mount targets...")
try:
client = self.regional_clients[filesystem.region]
describe_mount_target_paginator = client.get_paginator(
"describe_mount_targets"
)
for page in describe_mount_target_paginator.paginate(
FileSystemId=filesystem.id
):
for mount_target in page["MountTargets"]:
mount_target_id = mount_target["MountTargetId"]
mount_target_arn = f"arn:{self.audited_partition}:elasticfilesystem:{client.region}:{self.audited_account}:mount-target/{mount_target_id}"
if not self.audit_resources or (
is_resource_filtered(mount_target_arn, self.audit_resources)
):
self.filesystems[filesystem.arn].mount_targets.append(
MountTarget(
id=mount_target_id,
file_system_id=mount_target["FileSystemId"],
subnet_id=mount_target["SubnetId"],
)
if "Policy" in fs_policy:
filesystem.policy = json.loads(fs_policy["Policy"])
except ClientError as e:
if e.response["Error"]["Code"] == "PolicyNotFound":
filesystem.policy = {}
)
except Exception as error:
logger.error(
f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class MountTarget(BaseModel):
id: str
file_system_id: str
subnet_id: str
class FileSystem(BaseModel):
id: str
arn: str
region: str
policy: Optional[dict]
backup_policy: Optional[str]
policy: Optional[dict] = {}
backup_policy: Optional[str] = "DISABLED"
encrypted: bool
mount_targets: list[MountTarget] = []
tags: Optional[list] = []
@@ -1,34 +1,37 @@
from re import search
from unittest import mock
from prowler.providers.aws.services.efs.efs_service import FileSystem
from boto3 import client
from moto import mock_aws
# Mock Test Region
AWS_REGION = "eu-west-1"
AWS_ACCOUNT_NUMBER = "123456789012"
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
file_system_id = "fs-c7a0456e"
backup_valid_policy_status = "ENABLED"
CREATION_TOKEN = "fs-123"
class Test_efs_encryption_at_rest_enabled:
@mock_aws
def test_efs_encryption_enabled(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=None,
backup_policy=backup_valid_policy_status,
encrypted=True,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
filesystem = efs_client.create_file_system(
CreationToken=CREATION_TOKEN, Encrypted=True
)
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{filesystem['FileSystemId']}"
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_encryption_at_rest_enabled.efs_encryption_at_rest_enabled.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_encryption_at_rest_enabled.efs_encryption_at_rest_enabled import (
efs_encryption_at_rest_enabled,
@@ -38,26 +41,33 @@ class Test_efs_encryption_at_rest_enabled:
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search("has encryption at rest enabled", result[0].status_extended)
assert result[0].resource_id == file_system_id
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].status_extended
== f"EFS {filesystem['FileSystemId']} has encryption at rest enabled."
)
assert result[0].resource_id == filesystem["FileSystemId"]
assert result[0].resource_arn == efs_arn
@mock_aws
def test_efs_encryption_disabled(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=None,
backup_policy=backup_valid_policy_status,
encrypted=False,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
filesystem = efs_client.create_file_system(
CreationToken=CREATION_TOKEN, Encrypted=False
)
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{filesystem['FileSystemId']}"
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_encryption_at_rest_enabled.efs_encryption_at_rest_enabled.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_encryption_at_rest_enabled.efs_encryption_at_rest_enabled import (
efs_encryption_at_rest_enabled,
@@ -67,8 +77,10 @@ class Test_efs_encryption_at_rest_enabled:
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"does not have encryption at rest enabled", result[0].status_extended
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].status_extended
== f"EFS {filesystem['FileSystemId']} does not have encryption at rest enabled."
)
assert result[0].resource_id == file_system_id
assert result[0].resource_id == filesystem["FileSystemId"]
assert result[0].resource_arn == efs_arn
@@ -1,36 +1,36 @@
from re import search
from unittest import mock
from prowler.providers.aws.services.efs.efs_service import FileSystem
from boto3 import client
from moto import mock_aws
# Mock Test Region
AWS_REGION = "eu-west-1"
AWS_ACCOUNT_NUMBER = "123456789012"
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
file_system_id = "fs-c7a0456e"
backup_valid_policy_status = "ENABLED"
backup_valid_invalid_policy_status_1 = "DISABLING"
backup_valid_invalid_policy_status_2 = "DISABLED"
CREATION_TOKEN = "fs-123"
class Test_efs_have_backup_enabled:
@mock_aws
def test_efs_valid_backup_policy(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=None,
backup_policy=backup_valid_policy_status,
encrypted=True,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(
CreationToken=CREATION_TOKEN, Backup=True
)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled import (
efs_have_backup_enabled,
@@ -41,25 +41,29 @@ class Test_efs_have_backup_enabled:
assert len(result) == 1
assert result[0].status == "PASS"
assert search("has backup enabled", result[0].status_extended)
assert result[0].resource_id == file_system_id
assert result[0].resource_arn == efs_arn
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
@mock_aws
def test_efs_invalid_policy_backup_1(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=None,
backup_policy=backup_valid_invalid_policy_status_1,
encrypted=True,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(
CreationToken=CREATION_TOKEN, Backup=False
)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled import (
efs_have_backup_enabled,
@@ -70,25 +74,29 @@ class Test_efs_have_backup_enabled:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search("does not have backup enabled", result[0].status_extended)
assert result[0].resource_id == file_system_id
assert result[0].resource_arn == efs_arn
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
@mock_aws
def test_efs_invalid_policy_backup_2(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=None,
backup_policy=backup_valid_invalid_policy_status_2,
encrypted=True,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(
CreationToken=CREATION_TOKEN, Backup=False
)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled import (
efs_have_backup_enabled,
@@ -99,5 +107,8 @@ class Test_efs_have_backup_enabled:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search("does not have backup enabled", result[0].status_extended)
assert result[0].resource_id == file_system_id
assert result[0].resource_arn == efs_arn
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
@@ -0,0 +1,188 @@
from unittest import mock
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
CREATION_TOKEN = "fs-123"
class Test_efs_mount_target_not_publicly_accessible:
@mock_aws
def test_efs_no_file_system(self):
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible import (
efs_mount_target_not_publicly_accessible,
)
check = efs_mount_target_not_publicly_accessible()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_efs_no_mount_target(self):
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible import (
efs_mount_target_not_publicly_accessible,
)
check = efs_mount_target_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"EFS {file_system['FileSystemId']} does not have any public mount targets."
)
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
@mock_aws
def test_efs_mount_target_public_subnet(self):
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN)
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
vpc = ec2_client.create_vpc(
CidrBlock="172.28.7.0/24", InstanceTenancy="default"
)
igw = ec2_client.create_internet_gateway()
ec2_client.attach_internet_gateway(
InternetGatewayId=igw["InternetGateway"]["InternetGatewayId"],
VpcId=vpc["Vpc"]["VpcId"],
)
subnet_public = ec2_client.create_subnet(
VpcId=vpc["Vpc"]["VpcId"],
CidrBlock="172.28.7.192/26",
AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
)
route_table = ec2_client.create_route_table(VpcId=vpc["Vpc"]["VpcId"])
ec2_client.create_route(
RouteTableId=route_table["RouteTable"]["RouteTableId"],
DestinationCidrBlock="0.0.0.0/0", # Ruta predeterminada para todo el tráfico
GatewayId=igw["InternetGateway"]["InternetGatewayId"],
)
ec2_client.associate_route_table(
RouteTableId=route_table["RouteTable"]["RouteTableId"],
SubnetId=subnet_public["Subnet"]["SubnetId"],
)
mount_target = efs_client.create_mount_target(
FileSystemId=file_system["FileSystemId"],
SubnetId=subnet_public["Subnet"]["SubnetId"],
)
from prowler.providers.aws.services.efs.efs_service import EFS
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
), mock.patch(
"prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.vpc_client",
new=VPC(aws_provider),
):
from prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible import (
efs_mount_target_not_publicly_accessible,
)
check = efs_mount_target_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"EFS {file_system['FileSystemId']} has public mount targets: {mount_target['MountTargetId']}"
)
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
@mock_aws
def test_efs_mount_target_private_subnet(self):
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN)
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
vpc = ec2_client.create_vpc(
CidrBlock="172.28.7.0/24", InstanceTenancy="default"
)
subnet_private = ec2_client.create_subnet(
VpcId=vpc["Vpc"]["VpcId"],
CidrBlock="172.28.7.192/26",
AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
)
efs_client.create_mount_target(
FileSystemId=file_system["FileSystemId"],
SubnetId=subnet_private["Subnet"]["SubnetId"],
)
from prowler.providers.aws.services.efs.efs_service import EFS
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
), mock.patch(
"prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.vpc_client",
new=VPC(aws_provider),
):
from prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible import (
efs_mount_target_not_publicly_accessible,
)
check = efs_mount_target_not_publicly_accessible()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"EFS {file_system['FileSystemId']} does not have any public mount targets."
)
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
@@ -1,96 +1,137 @@
import json
from unittest import mock
from prowler.providers.aws.services.efs.efs_service import FileSystem
import botocore
from boto3 import client
from moto import mock_aws
# Mock Test Region
AWS_REGION = "eu-west-1"
AWS_ACCOUNT_NUMBER = "123456789012"
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
file_system_id = "fs-c7a0456e"
CREATION_TOKEN = "fs-123"
filesystem_policy = {
"Id": "1",
"Statement": [
{
"Effect": "Allow",
"Action": ["elasticfilesystem:ClientMount"],
"Principal": {"AWS": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"},
}
],
}
FILE_SYSTEM_POLICY = json.dumps(
{
"Id": "1",
"Statement": [
{
"Effect": "Allow",
"Action": ["elasticfilesystem:ClientMount"],
"Principal": {"AWS": "arn:aws:iam::123456789012:root"},
}
],
}
)
filesystem_invalid_policy = {
"Id": "1",
"Statement": [
{
"Effect": "Allow",
"Action": ["elasticfilesystem:ClientMount"],
"Principal": {"AWS": "*"},
}
],
}
FILE_SYSTEM_INVALID_POLICY = json.dumps(
{
"Id": "1",
"Statement": [
{
"Effect": "Allow",
"Action": ["elasticfilesystem:ClientMount"],
"Principal": {"AWS": "*"},
}
],
}
)
# 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"
}
},
}
],
}
FILE_SYSTEM_POLICY_WITH_SOURCE_ARN_CONDITION = json.dumps(
{
"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"}},
}
],
}
FILE_SYSTEM_POLICY_WITH_MOUNT_TARGET_CONDITION = json.dumps(
{
"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"}
},
}
],
}
)
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeFileSystemPolicy":
return {"Policy": FILE_SYSTEM_POLICY}
return make_api_call(self, operation_name, kwarg)
def mock_make_api_call_v2(self, operation_name, kwarg):
if operation_name == "DescribeFileSystemPolicy":
return {"Policy": FILE_SYSTEM_POLICY_WITH_MOUNT_TARGET_CONDITION}
return make_api_call(self, operation_name, kwarg)
def mock_make_api_call_v3(self, operation_name, kwarg):
if operation_name == "DescribeFileSystemPolicy":
return {"Policy": FILE_SYSTEM_POLICY_WITH_SOURCE_ARN_CONDITION}
return make_api_call(self, operation_name, kwarg)
def mock_make_api_call_v4(self, operation_name, kwarg):
if operation_name == "DescribeFileSystemPolicy":
return {"Policy": FILE_SYSTEM_INVALID_POLICY}
return make_api_call(self, operation_name, kwarg)
class Test_efs_not_publicly_accessible:
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_efs_valid_policy(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,
backup_policy=None,
encrypted=True,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import (
efs_not_publicly_accessible,
@@ -102,29 +143,32 @@ class Test_efs_not_publicly_accessible:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"EFS {file_system_id} has a policy which does not allow access to any client within the VPC."
== f"EFS {file_system['FileSystemId']} has a policy which does not allow access to any client within the VPC."
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
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 == []
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call_v2)
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,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import (
efs_not_publicly_accessible,
@@ -136,29 +180,32 @@ class Test_efs_not_publicly_accessible:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"EFS {file_system_id} has a policy which does not allow access to any client within the VPC."
== f"EFS {file_system['FileSystemId']} has a policy which does not allow access to any client within the VPC."
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
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 == []
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call_v3)
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,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import (
efs_not_publicly_accessible,
@@ -170,30 +217,32 @@ class Test_efs_not_publicly_accessible:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"EFS {file_system_id} has a policy which does not allow access to any client within the VPC."
== f"EFS {file_system['FileSystemId']} has a policy which does not allow access to any client within the VPC."
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
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 == []
@mock_aws
@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call_v4)
def test_efs_invalid_policy(self):
efs_client = mock.MagicMock
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}"
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
efs_client.filesystems = [
FileSystem(
id=file_system_id,
arn=efs_arn,
region=AWS_REGION,
policy=filesystem_invalid_policy,
backup_policy=None,
encrypted=True,
)
]
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import (
efs_not_publicly_accessible,
@@ -205,29 +254,31 @@ class Test_efs_not_publicly_accessible:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"EFS {file_system_id} has a policy which allows access to any client within the VPC."
== f"EFS {file_system['FileSystemId']} has a policy which allows access to any client within the VPC."
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
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 == []
@mock_aws
def test_efs_no_policy(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=None,
backup_policy=None,
encrypted=True,
)
]
efs_client = client("efs", region_name=AWS_REGION_US_EAST_1)
file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN)
from prowler.providers.aws.services.efs.efs_service import EFS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.services.efs.efs_service.EFS",
efs_client,
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client",
new=EFS(aws_provider),
):
from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import (
efs_not_publicly_accessible,
@@ -239,9 +290,12 @@ class Test_efs_not_publicly_accessible:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"EFS {file_system_id} doesn't have any policy which means it grants full access to any client within the VPC."
== f"EFS {file_system['FileSystemId']} doesn't have any policy which means it grants full access to any client within the VPC."
)
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == file_system["FileSystemId"]
assert (
result[0].resource_arn
== f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
)
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 == []
@@ -2,8 +2,6 @@ import json
from unittest.mock import patch
import botocore
from boto3 import client
from moto import mock_aws
from prowler.providers.aws.services.efs.efs_service import EFS
from tests.providers.aws.utils import (
@@ -15,13 +13,11 @@ from tests.providers.aws.utils import (
# Mocking Access Analyzer Calls
make_api_call = botocore.client.BaseClient._make_api_call
file_system_id = "fs-c7a0456e"
FILE_SYSTEM_ID = "fs-c7a0456e"
creation_token = "console-d215fa78-1f83-4651-b026-facafd8a7da7"
CREATION_TOKEN = "console-d215fa78-1f83-4651-b026-facafd8a7da7"
backup_policy_status = "ENABLED"
filesystem_policy = {
FILESYSTEM_POLICY = {
"Id": "1",
"Statement": [
{
@@ -34,10 +30,35 @@ filesystem_policy = {
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "DescribeFileSystems":
return {
"FileSystems": [
{
"FileSystemId": FILE_SYSTEM_ID,
"Encrypted": True,
"Tags": [{"Key": "test", "Value": "test"}],
"BackupPolicy": {"Status": "ENABLED"},
"Policy": json.dumps(FILESYSTEM_POLICY),
}
]
}
if operation_name == "DescribeMountTargets":
return {
"MountTargets": [
{
"MountTargetId": "fsmt-123",
"FileSystemId": FILE_SYSTEM_ID,
"SubnetId": "subnet-123",
"LifeCycleState": "available",
"OwnerId": AWS_ACCOUNT_NUMBER,
"VpcId": "vpc-123",
}
]
}
if operation_name == "DescribeFileSystemPolicy":
return {"FileSystemId": file_system_id, "Policy": json.dumps(filesystem_policy)}
return {"FileSystemId": FILE_SYSTEM_ID, "Policy": json.dumps(FILESYSTEM_POLICY)}
if operation_name == "DescribeBackupPolicy":
return {"BackupPolicy": {"Status": backup_policy_status}}
return {"BackupPolicy": {"Status": "ENABLED"}}
return make_api_call(self, operation_name, kwarg)
@@ -66,35 +87,40 @@ class Test_EFS:
access_analyzer = EFS(set_mocked_aws_provider())
assert access_analyzer.service == "efs"
@mock_aws
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test EFS describe file systems
def test_describe_file_systems(self):
efs_client = client("efs", AWS_REGION_EU_WEST_1)
efs = efs_client.create_file_system(
CreationToken=creation_token,
Encrypted=True,
Tags=[
{"Key": "test", "Value": "test"},
],
)
filesystem = EFS(set_mocked_aws_provider())
assert len(filesystem.filesystems) == 1
assert filesystem.filesystems[0].id == efs["FileSystemId"]
assert filesystem.filesystems[0].encrypted == efs["Encrypted"]
assert filesystem.filesystems[0].tags == [
aws_provider = set_mocked_aws_provider()
efs = EFS(aws_provider)
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}"
assert len(efs.filesystems) == 1
assert efs.filesystems[efs_arn].id == FILE_SYSTEM_ID
assert efs.filesystems[efs_arn].encrypted
assert efs.filesystems[efs_arn].tags == [
{"Key": "test", "Value": "test"},
]
@mock_aws
# Test EFS describe file systems
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test EFS describe file systems policies
def test_describe_file_system_policies(self):
efs_client = client("efs", AWS_REGION_EU_WEST_1)
efs = efs_client.create_file_system(
CreationToken=creation_token, Encrypted=True
aws_provider = set_mocked_aws_provider()
efs = EFS(aws_provider)
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}"
assert len(efs.filesystems) == 1
assert efs.filesystems[efs_arn].id == FILE_SYSTEM_ID
assert efs.filesystems[efs_arn].encrypted
assert efs.filesystems[efs_arn].backup_policy == "ENABLED"
assert efs.filesystems[efs_arn].policy == FILESYSTEM_POLICY
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
# Test EFS describe mount targets
def test_describe_mount_targets(self):
aws_provider = set_mocked_aws_provider()
efs = EFS(aws_provider)
assert len(efs.filesystems) == 1
efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}"
assert (
efs.filesystems[efs_arn].mount_targets[0].file_system_id == FILE_SYSTEM_ID
)
filesystem = EFS(set_mocked_aws_provider())
assert len(filesystem.filesystems) == 1
assert filesystem.filesystems[0].id == efs["FileSystemId"]
assert filesystem.filesystems[0].encrypted == efs["Encrypted"]
assert filesystem.filesystems[0].backup_policy == backup_policy_status
assert filesystem.filesystems[0].policy == filesystem_policy
assert efs.filesystems[efs_arn].mount_targets[0].id == "fsmt-123"
assert efs.filesystems[efs_arn].mount_targets[0].subnet_id == "subnet-123"