feat(guardduty): add new check guardduty_eks_audit_log_enabled (#5293)

Co-authored-by: Sergio <sergio@prowler.com>
2026-07-23 12:31:54 +00:00 · 2024-10-04 19:43:04 +02:00
parent aa3425a7de
commit 2b66368cf2
8 changed files with 244 additions and 33 deletions
@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "guardduty_eks_audit_log_enabled",
+  "CheckTitle": "GuardDuty EKS Audit Log Monitoring Enabled",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices/Runtime Behavior Analysis"
+  ],
+  "ServiceName": "guardduty",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:guardduty:region:account-id/detector-id",
+  "Severity": "high",
+  "ResourceType": "AwsGuardDutyDetector",
+  "Description": "Checks whether GuardDuty EKS Audit Log Monitoring is enabled as source in a detector.",
+  "Risk": "Without GuardDuty EKS Audit Log Monitoring enabled, you may not be able to detect potentially suspicious activities in your Amazon Elastic Kubernetes Service (Amazon EKS) clusters.",
+  "RelatedUrl": "https://docs.aws.amazon.com/guardduty/latest/ug/kubernetes-protection.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws guardduty update-detector --detector-id <detector-id> --data-sources Kubernetes={AuditLogs={Enable=true}}",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/guardduty-controls.html#guardduty-5",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable GuardDuty EKS Audit Log Monitoring to detect potentially suspicious activities in your Amazon Elastic Kubernetes Service (Amazon EKS) clusters.",
+      "Url": "https://docs.aws.amazon.com/guardduty/latest/ug/eks-protection-enable-standalone-account.html"
+    }
+  },
+  "Categories": [
+    "logging"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -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_eks_audit_log_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 EKS Audit Log Monitoring enabled."
+                if detector.eks_audit_log_protection:
+                    report.status = "PASS"
+                    report.status_extended = f"GuardDuty detector {detector.id} has EKS Audit Log Monitoring enabled."
+                findings.append(report)
+        return findings
@@ -15,7 +15,7 @@
  "RelatedUrl": "https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/guard-duty-rds-protection.html",
  "Remediation": {
    "Code": {
-      "CLI": "",
+      "CLI": "aws guardduty update-detector --detector-id <detector-id> --features Name=RDS_LOGIN_EVENTS,Status=ENABLED",
      "NativeIaC": "",
      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/guardduty-controls.html#guardduty-9",
      "Terraform": ""
@@ -15,7 +15,7 @@
  "RelatedUrl": "https://docs.aws.amazon.com/guardduty/latest/ug/s3_detection.html",
  "Remediation": {
    "Code": {
-      "CLI": "aws guardduty update-detector --detector-id <detector-id> --data-sources '{\"S3Logs\": {\"Enable\": true}}'",
+      "CLI": "aws guardduty update-detector --detector-id <detector-id> --data-sources S3Logs={Enable=true}}'",
      "NativeIaC": "",
      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/guardduty-controls.html#guardduty-10",
      "Terraform": ""
@@ -13,7 +13,7 @@ class GuardDuty(AWSService):
        super().__init__(__class__.__name__, provider)
        self.detectors = []
        self.__threading_call__(self._list_detectors)
-        self._get_detector()
+        self.__threading_call__(self._get_detector, self.detectors)
        self._list_findings()
        self._list_members()
        self._get_administrator_account()
@@ -50,38 +50,46 @@ class GuardDuty(AWSService):
                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

-    def _get_detector(self):
+    def _get_detector(self, detector):
        logger.info("GuardDuty - getting detector info...")
        try:
-            for detector in self.detectors:
-                try:
-                    if detector.id and detector.enabled_in_account:
-                        regional_client = self.regional_clients[detector.region]
-                        detector_info = regional_client.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") == "ENABLED":
-                            detector.s3_protection = True
-
-                        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}"
+            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}"
+                )
        except Exception as error:
            logger.error(
                f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
@@ -204,3 +212,4 @@ class Detector(BaseModel):
    tags: Optional[list] = []
    s3_protection: bool = False
    rds_protection: bool = False
+    eks_audit_log_protection: bool = False
@@ -0,0 +1,143 @@
+from unittest.mock import patch
+
+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,
+)
+
+
+class Test_guardduty_eks_audit_log_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_eks_audit_log_enabled.guardduty_eks_audit_log_enabled.guardduty_client",
+            new=GuardDuty(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.guardduty.guardduty_eks_audit_log_enabled.guardduty_eks_audit_log_enabled import (
+                guardduty_eks_audit_log_enabled,
+            )
+
+            check = guardduty_eks_audit_log_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_eks_audit_log_enabled.guardduty_eks_audit_log_enabled.guardduty_client",
+            new=GuardDuty(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.guardduty.guardduty_eks_audit_log_enabled.guardduty_eks_audit_log_enabled import (
+                guardduty_eks_audit_log_enabled,
+            )
+
+            check = guardduty_eks_audit_log_enabled()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    @mock_aws
+    def test_detector_eks_audit_log_enabled(self):
+        guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+
+        detector_id = guardduty_client.create_detector(
+            Enable=True, DataSources={"Kubernetes": {"AuditLogs": {"Enable": 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_eks_audit_log_enabled.guardduty_eks_audit_log_enabled.guardduty_client",
+            new=GuardDuty(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.guardduty.guardduty_eks_audit_log_enabled.guardduty_eks_audit_log_enabled import (
+                guardduty_eks_audit_log_enabled,
+            )
+
+            check = guardduty_eks_audit_log_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"GuardDuty detector {detector_id} has EKS Audit Log Monitoring 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_eks_audit_log_disabled(self):
+        guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
+
+        detector_id = guardduty_client.create_detector(
+            Enable=True, DataSources={"Kubernetes": {"AuditLogs": {"Enable": 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_eks_audit_log_enabled.guardduty_eks_audit_log_enabled.guardduty_client",
+            new=GuardDuty(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.guardduty.guardduty_eks_audit_log_enabled.guardduty_eks_audit_log_enabled import (
+                guardduty_eks_audit_log_enabled,
+            )
+
+            check = guardduty_eks_audit_log_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 EKS Audit Log Monitoring 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 == []
@@ -111,7 +111,10 @@ class Test_GuardDuty_Service:
        guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1)
        response = guardduty_client.create_detector(
            Enable=True,
-            DataSources={"S3Logs": {"Enable": True}},
+            DataSources={
+                "S3Logs": {"Enable": True},
+                "Kubernetes": {"AuditLogs": {"Enable": True}},
+            },
        )

        aws_provider = set_mocked_aws_provider()
@@ -129,6 +132,7 @@ class Test_GuardDuty_Service:
        assert guardduty.detectors[0].administrator_account == "123456789013"
        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].region == AWS_REGION_EU_WEST_1
        assert guardduty.detectors[0].tags == [{"test": "test"}]