feat(guardduty): add new check guardduty_lambda_protection_enabled (#5299)

Co-authored-by: Sergio <sergio@prowler.com>
2026-07-24 04:51:51 +00:00 · 2024-10-08 14:20:23 +02:00
parent 5bf85366e0
commit 27cd9b22df
6 changed files with 254 additions and 36 deletions
@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "guardduty_lambda_protection_enabled",
+  "CheckTitle": "Check if GuardDuty Lambda Protection 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": "",
+  "Description": "GuardDuty Lambda Protection helps you identify potential security threats when an AWS Lambda function gets invoked. After you enable Lambda Protection, GuardDuty starts monitoring Lambda network activity logs associated with the Lambda functions in your AWS account.",
+  "Risk": "If Lambda Protection is not enabled, GuardDuty will not be able to monitor Lambda network activity logs and may miss potential security threats.",
+  "RelatedUrl": "https://docs.aws.amazon.com/guardduty/latest/ug/lambda-protection.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws guardduty update-detector --detector-id <detector-id> --features Name=LAMBDA_NETWORK_LOGS,Status=ENABLED",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/guardduty-controls.html#guardduty-6",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable Lambda Protection in your GuardDuty detector to start monitoring Lambda Network Activity in your account.",
+      "Url": "https://docs.aws.amazon.com/guardduty/latest/ug/configure-lambda-protection-standalone-acc.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_lambda_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 Lambda Protection enabled."
+                if detector.lambda_protection:
+                    report.status = "PASS"
+                    report.status_extended = f"GuardDuty detector {detector.id} has Lambda Protection enabled."
+                findings.append(report)
+        return findings
@@ -53,44 +53,58 @@ class GuardDuty(AWSService):
    def _get_detector(self, detector):
        logger.info("GuardDuty - getting detector info...")
        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
-                )
-
-                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", []):
+            try:
+                if detector.id and detector.enabled_in_account:
+                    detector_info = self.regional_clients[detector.region].get_detector(
+                        DetectorId=detector.id
+                    )
                    if (
-                        feat.get("Name") == "RDS_LOGIN_EVENTS"
-                        and feat.get("Status", "DISABLED") == "ENABLED"
+                        "Status" in detector_info
+                        and detector_info["Status"] == "ENABLED"
                    ):
-                        detector.rds_protection = True
+                        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
+                        elif (
+                            feat.get("Name", "") == "LAMBDA_NETWORK_LOGS"
+                            and feat.get("Status", "DISABLED") == "ENABLED"
+                        ):
+                            detector.lambda_protection = True
+
+            except Exception as error:
+                logger.error(
+                    f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
+                )
        except Exception as error:
            logger.error(
                f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
@@ -214,4 +228,5 @@ class Detector(BaseModel):
    s3_protection: bool = False
    rds_protection: bool = False
    eks_audit_log_protection: bool = False
+    lambda_protection: bool = False
    ec2_malware_protection: bool = False
@@ -0,0 +1,148 @@
+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
+
+
+class Test_guardduty_lambda_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_lambda_protection_enabled.guardduty_lambda_protection_enabled.guardduty_client",
+            new=GuardDuty(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled import (
+                guardduty_lambda_protection_enabled,
+            )
+
+            check = guardduty_lambda_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_lambda_protection_enabled.guardduty_lambda_protection_enabled.guardduty_client",
+            new=GuardDuty(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled import (
+                guardduty_lambda_protection_enabled,
+            )
+
+            check = guardduty_lambda_protection_enabled()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    @mock_aws
+    def test_detector_lambda_protection_enabled(self):
+        guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+
+        detector_id = guardduty_client.create_detector(
+            Enable=True,
+            Features=[{"Name": "LAMBDA_NETWORK_LOGS", "Status": "ENABLED"}],
+        )["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_lambda_protection_enabled.guardduty_lambda_protection_enabled.guardduty_client",
+            new=GuardDuty(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled import (
+                guardduty_lambda_protection_enabled,
+            )
+
+            check = guardduty_lambda_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 Lambda Protection 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_lambda_protection_disabled(self):
+        guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+
+        detector_id = guardduty_client.create_detector(
+            Enable=True,
+            Features=[{"Name": "LAMBDA_NETWORK_LOGS", "Status": "DISABLED"}],
+        )["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_lambda_protection_enabled.guardduty_lambda_protection_enabled.guardduty_client",
+            new=GuardDuty(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.guardduty.guardduty_lambda_protection_enabled.guardduty_lambda_protection_enabled import (
+                guardduty_lambda_protection_enabled,
+            )
+
+            check = guardduty_lambda_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 Lambda Protection 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 == []
@@ -115,6 +115,7 @@ class Test_GuardDuty_Service:
                "S3Logs": {"Enable": True},
                "Kubernetes": {"AuditLogs": {"Enable": True}},
            },
+            Features=[{"Name": "LAMBDA_NETWORK_LOGS", "Status": "ENABLED"}],
        )

        aws_provider = set_mocked_aws_provider()
@@ -133,6 +134,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 guardduty.detectors[0].lambda_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"}]