feat(guardduty): add new check guardduty_ec2_malware_protection_enabled (#5297)

Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Rubén De la Torre Vico
2024-10-07 19:03:36 +02:00
committed by GitHub
parent 3b64bbd3a8
commit 2f8a3d2ef8
6 changed files with 278 additions and 36 deletions
@@ -0,0 +1,32 @@
{
"Provider": "aws",
"CheckID": "guardduty_ec2_malware_protection_enabled",
"CheckTitle": "Ensure that GuardDuty Malware Protection for EC2 is enabled.",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "guardduty",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:guardduty:region:account-id/detector-id",
"Severity": "high",
"ResourceType": "AwsGuardDutyDetector",
"Description": "GuardDuty Malware Protection for EC2 helps you detect the potential presence of malware by scanning the Amazon Elastic Block Store (Amazon EBS) volumes that are attached to Amazon Elastic Compute Cloud (Amazon EC2) instances and container workloads.",
"Risk": "Malware can compromise your EC2 instances and container workloads, leading to data breaches, data exfiltration, and other security incidents.",
"RelatedUrl": "https://docs.aws.amazon.com/guardduty/latest/ug/malware-protection.html",
"Remediation": {
"Code": {
"CLI": "aws guardduty update-detector --detector-id <detector-id> --data-sources MalwareProtection={ScanEc2InstanceWithFindings={EbsVolumes=true}}",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/guardduty-controls.html#guardduty-8",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable Malware Protection for EC2 in GuardDuty.",
"Url": "https://docs.aws.amazon.com/guardduty/latest/ug/configure-malware-protection-single-account.html"
}
},
"Categories": [],
"Notes": "",
"DependsOn": [],
"RelatedTo": []
}
@@ -0,0 +1,21 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.guardduty.guardduty_client import guardduty_client
class guardduty_ec2_malware_protection_enabled(Check):
def execute(self):
findings = []
for detector in guardduty_client.detectors:
if detector.status:
report = Check_Report_AWS(self.metadata())
report.region = detector.region
report.resource_id = detector.id
report.resource_arn = detector.arn
report.resource_tags = detector.tags
report.status = "FAIL"
report.status_extended = f"GuardDuty detector {detector.id} does not have Malware Protection for EC2 enabled."
if detector.ec2_malware_protection:
report.status = "PASS"
report.status_extended = f"GuardDuty detector {detector.id} has Malware Protection for EC2 enabled."
findings.append(report)
return findings
@@ -53,43 +53,44 @@ class GuardDuty(AWSService):
def _get_detector(self, detector):
logger.info("GuardDuty - getting detector info...")
try:
try:
if detector.id and detector.enabled_in_account:
detector_info = self.regional_clients[detector.region].get_detector(
DetectorId=detector.id
)
if (
"Status" in detector_info
and detector_info["Status"] == "ENABLED"
):
detector.status = True
data_sources = detector_info.get("DataSources", {})
s3_logs = data_sources.get("S3Logs", {})
if s3_logs.get("Status", "DISABLED") == "ENABLED":
detector.s3_protection = True
detector.eks_audit_log_protection = (
True
if data_sources.get("Kubernetes", {})
.get("AuditLogs", {})
.get("Status", "DISABLED")
== "ENABLED"
else False
)
for feat in detector_info.get("Features", []):
if (
feat.get("Name") == "RDS_LOGIN_EVENTS"
and feat.get("Status", "DISABLED") == "ENABLED"
):
detector.rds_protection = True
except Exception as error:
logger.error(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
if detector.id and detector.enabled_in_account:
detector_info = self.regional_clients[detector.region].get_detector(
DetectorId=detector.id
)
if "Status" in detector_info and detector_info["Status"] == "ENABLED":
detector.status = True
data_sources = detector_info.get("DataSources", {})
s3_logs = data_sources.get("S3Logs", {})
if s3_logs.get("Status", "DISABLED") == "ENABLED":
detector.s3_protection = True
detector.eks_audit_log_protection = (
True
if data_sources.get("Kubernetes", {})
.get("AuditLogs", {})
.get("Status", "DISABLED")
== "ENABLED"
else False
)
detector.ec2_malware_protection = (
True
if data_sources.get("MalwareProtection", {})
.get("ScanEc2InstanceWithFindings", {})
.get("EbsVolumes", {})
.get("Status", "DISABLED")
== "ENABLED"
else False
)
for feat in detector_info.get("Features", []):
if (
feat.get("Name") == "RDS_LOGIN_EVENTS"
and feat.get("Status", "DISABLED") == "ENABLED"
):
detector.rds_protection = True
except Exception as error:
logger.error(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
@@ -213,3 +214,4 @@ class Detector(BaseModel):
s3_protection: bool = False
rds_protection: bool = False
eks_audit_log_protection: bool = False
ec2_malware_protection: bool = False
@@ -0,0 +1,186 @@
from unittest.mock import patch
import botocore
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
set_mocked_aws_provider,
)
orig = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "GetDetector":
return {
"CreatedAt": "2021-01-01T00:00:00Z",
"FindingPublishingFrequency": "FIFTEEN_MINUTES",
"ServiceRole": "AWSServiceRoleForAmazonGuardDuty",
"Status": "ENABLED",
"UpdatedAt": "2021-01-01T00:00:00Z",
"DataSources": {
"S3Logs": {
"Enable": False,
},
"CloudTrail": {
"Enable": False,
},
"DNSLogs": {
"Enable": False,
},
"MalwareProtection": {
"ScanEc2InstanceWithFindings": {
"EbsVolumes": {"Status": "ENABLED"},
},
},
},
}
# If we don't want to patch the API call
return orig(self, operation_name, kwarg)
class Test_guardduty_ec2_malware_protection_enabled:
def test_no_detectors(self):
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_ec2_malware_protection_enabled.guardduty_ec2_malware_protection_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_ec2_malware_protection_enabled.guardduty_ec2_malware_protection_enabled import (
guardduty_ec2_malware_protection_enabled,
)
check = guardduty_ec2_malware_protection_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_detector_disabled(self):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
guardduty_client.create_detector(Enable=False)
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_ec2_malware_protection_enabled.guardduty_ec2_malware_protection_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_ec2_malware_protection_enabled.guardduty_ec2_malware_protection_enabled import (
guardduty_ec2_malware_protection_enabled,
)
check = guardduty_ec2_malware_protection_enabled()
result = check.execute()
assert len(result) == 0
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
@mock_aws
def test_detector_malware_protection_enabled(self):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(
Enable=True,
DataSources={
"MalwareProtection": {
"ScanEc2InstanceWithFindings": {"EbsVolumes": True}
}
},
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_ec2_malware_protection_enabled.guardduty_ec2_malware_protection_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_ec2_malware_protection_enabled.guardduty_ec2_malware_protection_enabled import (
guardduty_ec2_malware_protection_enabled,
)
check = guardduty_ec2_malware_protection_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"GuardDuty detector {detector_id} has Malware Protection for EC2 enabled."
)
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result[0].resource_tags == []
@mock_aws
def test_detector_malware_protection_disabled(self):
guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
detector_id = guardduty_client.create_detector(
Enable=True,
DataSources={
"MalwareProtection": {
"ScanEc2InstanceWithFindings": {"EbsVolumes": False}
}
},
)["DetectorId"]
aws_provider = set_mocked_aws_provider()
from prowler.providers.aws.services.guardduty.guardduty_service import GuardDuty
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), patch(
"prowler.providers.aws.services.guardduty.guardduty_ec2_malware_protection_enabled.guardduty_ec2_malware_protection_enabled.guardduty_client",
new=GuardDuty(aws_provider),
):
# Test Check
from prowler.providers.aws.services.guardduty.guardduty_ec2_malware_protection_enabled.guardduty_ec2_malware_protection_enabled import (
guardduty_ec2_malware_protection_enabled,
)
check = guardduty_ec2_malware_protection_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"GuardDuty detector {detector_id} does not have Malware Protection for EC2 enabled."
)
assert result[0].resource_id == detector_id
assert result[0].region == AWS_REGION_EU_WEST_1
assert (
result[0].resource_arn
== f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}"
)
assert result[0].resource_tags == []
@@ -133,6 +133,7 @@ class Test_GuardDuty_Service:
assert guardduty.detectors[0].s3_protection
assert not guardduty.detectors[0].rds_protection
assert guardduty.detectors[0].eks_audit_log_protection
assert not guardduty.detectors[0].ec2_malware_protection
assert guardduty.detectors[0].region == AWS_REGION_EU_WEST_1
assert guardduty.detectors[0].tags == [{"test": "test"}]