From aa326341053c379fef82e30fed5de1e0b4997ebe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rub=C3=A9n=20De=20la=20Torre=20Vico?= Date: Tue, 8 Oct 2024 19:27:37 +0200 Subject: [PATCH] chore(guardduty): mock failing tests using moto (#5334) --- .../services/guardduty/guardduty_service.py | 101 +++--- .../guardduty_centrally_managed_test.py | 210 +++++++------ .../guardduty_is_enabled_test.py | 290 ++++++++++-------- ...uardduty_no_high_severity_findings_test.py | 150 +++++---- 4 files changed, 407 insertions(+), 344 deletions(-) diff --git a/prowler/providers/aws/services/guardduty/guardduty_service.py b/prowler/providers/aws/services/guardduty/guardduty_service.py index 01ce461919..5669ecfec6 100644 --- a/prowler/providers/aws/services/guardduty/guardduty_service.py +++ b/prowler/providers/aws/services/guardduty/guardduty_service.py @@ -33,7 +33,10 @@ class GuardDuty(AWSService): ): self.detectors.append( Detector( - id=detector, arn=arn, region=regional_client.region + id=detector, + arn=arn, + region=regional_client.region, + enabled_in_account=True, ) ) if not detectors: @@ -53,58 +56,50 @@ 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 - ) - - 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}" + if detector.id and detector.enabled_in_account: + detector_info = self.regional_clients[detector.region].get_detector( + DetectorId=detector.id ) + if detector_info.get("Status", "DISABLED") == "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 + 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}" @@ -219,7 +214,7 @@ class Detector(BaseModel): id: str arn: str region: str - enabled_in_account: bool = True + enabled_in_account: bool status: bool = None findings: list = [] member_accounts: list = [] diff --git a/tests/providers/aws/services/guardduty/guardduty_centrally_managed/guardduty_centrally_managed_test.py b/tests/providers/aws/services/guardduty/guardduty_centrally_managed/guardduty_centrally_managed_test.py index 7ff4d46a82..739d9728df 100644 --- a/tests/providers/aws/services/guardduty/guardduty_centrally_managed/guardduty_centrally_managed_test.py +++ b/tests/providers/aws/services/guardduty/guardduty_centrally_managed/guardduty_centrally_managed_test.py @@ -1,46 +1,79 @@ -from unittest import mock -from uuid import uuid4 +from unittest.mock import patch -from prowler.providers.aws.services.guardduty.guardduty_service import Detector -from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1 +import botocore +from boto3 import client +from moto import mock_aws -AWS_ACCOUNT_NUMBER_ADMIN = "123456789013" -DETECTOR_ID = str(uuid4()) -DETECTOR_ARN = f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{DETECTOR_ID}" +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_admin_enabled(self, operation_name, api_params): + if operation_name == "GetAdministratorAccount": + return { + "Administrator": { + "AccountId": "210987654321", + } + } + return orig(self, operation_name, api_params) + + +def mock_make_api_call_members_managers(self, operation_name, api_params): + if operation_name == "ListMembers": + return { + "Members": [ + { + "AccountId": "210987654321", + "RelationshipStatus": "Enabled", + } + ] + } + return orig(self, operation_name, api_params) class Test_guardduty_centrally_managed: + @mock_aws def test_no_detectors(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, + 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_centrally_managed.guardduty_centrally_managed.guardduty_client", + new=GuardDuty(aws_provider), ): - from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import ( - guardduty_no_high_severity_findings, + from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import ( + guardduty_centrally_managed, ) - check = guardduty_no_high_severity_findings() + check = guardduty_centrally_managed() result = check.execute() assert len(result) == 0 + @mock_aws def test_detector_no_centralized_managed(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - region=AWS_REGION_EU_WEST_1, - arn=DETECTOR_ARN, - status=False, - findings=[str(uuid4())], - ) - ) + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, + detector_id = guardduty_client.create_detector(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_centrally_managed.guardduty_centrally_managed.guardduty_client", + new=GuardDuty(aws_provider), ): # Test Check from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import ( @@ -53,54 +86,36 @@ class Test_guardduty_centrally_managed: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"GuardDuty detector {DETECTOR_ID} is not centrally managed." + == f"GuardDuty detector {detector_id} is not centrally managed." ) - assert result[0].resource_id == DETECTOR_ID + assert result[0].resource_id == detector_id assert result[0].region == AWS_REGION_EU_WEST_1 - assert result[0].resource_arn == DETECTOR_ARN - - def test_not_enabled_account_detector(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=AWS_ACCOUNT_NUMBER, - region=AWS_REGION_EU_WEST_1, - arn=DETECTOR_ARN, - enabled_in_account=False, + assert ( + result[0].resource_arn + == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}" ) - ) - - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, - ): - # Test Check - from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import ( - guardduty_centrally_managed, - ) - - check = guardduty_centrally_managed() - result = check.execute() - assert len(result) == 0 + assert result[0].resource_tags == [] + @patch( + "botocore.client.BaseClient._make_api_call", + new=mock_make_api_call_admin_enabled, + ) + @mock_aws def test_detector_centralized_managed(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - region=AWS_REGION_EU_WEST_1, - arn=DETECTOR_ARN, - status=False, - findings=[str(uuid4())], - administrator_account=AWS_ACCOUNT_NUMBER_ADMIN, - ) - ) + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, + detector_id = guardduty_client.create_detector(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_centrally_managed.guardduty_centrally_managed.guardduty_client", + new=GuardDuty(aws_provider), ): # Test Check from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import ( @@ -113,29 +128,35 @@ class Test_guardduty_centrally_managed: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"GuardDuty detector {DETECTOR_ID} is centrally managed by account {AWS_ACCOUNT_NUMBER_ADMIN}." + == f"GuardDuty detector {detector_id} is centrally managed by account 210987654321." ) - assert result[0].resource_id == DETECTOR_ID + assert result[0].resource_id == detector_id assert result[0].region == AWS_REGION_EU_WEST_1 - assert result[0].resource_arn == DETECTOR_ARN - - def test_detector_administrator(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - region=AWS_REGION_EU_WEST_1, - arn=DETECTOR_ARN, - status=False, - findings=[str(uuid4())], - member_accounts=[AWS_ACCOUNT_NUMBER_ADMIN], + assert ( + result[0].resource_arn + == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}" ) - ) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, + @patch( + "botocore.client.BaseClient._make_api_call", + new=mock_make_api_call_members_managers, + ) + @mock_aws + def test_detector_members_accounts(self): + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) + + detector_id = guardduty_client.create_detector(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_centrally_managed.guardduty_centrally_managed.guardduty_client", + new=GuardDuty(aws_provider), ): # Test Check from prowler.providers.aws.services.guardduty.guardduty_centrally_managed.guardduty_centrally_managed import ( @@ -148,8 +169,11 @@ class Test_guardduty_centrally_managed: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"GuardDuty detector {DETECTOR_ID} is administrator account with 1 member accounts." + == f"GuardDuty detector {detector_id} is administrator account with 1 member accounts." ) - assert result[0].resource_id == DETECTOR_ID + assert result[0].resource_id == detector_id assert result[0].region == AWS_REGION_EU_WEST_1 - assert result[0].resource_arn == DETECTOR_ARN + assert ( + result[0].resource_arn + == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}" + ) diff --git a/tests/providers/aws/services/guardduty/guardduty_is_enabled/guardduty_is_enabled_test.py b/tests/providers/aws/services/guardduty/guardduty_is_enabled/guardduty_is_enabled_test.py index 3564ad3dbb..25e6aaa18d 100644 --- a/tests/providers/aws/services/guardduty/guardduty_is_enabled/guardduty_is_enabled_test.py +++ b/tests/providers/aws/services/guardduty/guardduty_is_enabled/guardduty_is_enabled_test.py @@ -1,171 +1,199 @@ -from unittest import mock -from uuid import uuid4 +from unittest.mock import patch + +from boto3 import client +from moto import mock_aws -from prowler.providers.aws.services.guardduty.guardduty_service import Detector from tests.providers.aws.utils import ( - AWS_ACCOUNT_ARN, AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1, + set_mocked_aws_provider, ) -DETECTOR_ID = str(uuid4()) -DETECTOR_ARN = f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{DETECTOR_ID}" - class Test_guardduty_is_enabled: + @mock_aws def test_no_detectors(self): - guardduty_client = mock.MagicMock - guardduty_client.region = AWS_REGION_EU_WEST_1 - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=AWS_ACCOUNT_NUMBER, - region=AWS_REGION_EU_WEST_1, - arn=AWS_ACCOUNT_ARN, - enabled_in_account=False, - ) - ) - guardduty_client.audited_account_arn = AWS_ACCOUNT_ARN - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, - ): + 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_is_enabled.guardduty_is_enabled.guardduty_client", + new=GuardDuty(aws_provider), + ) as guardduty_client: from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import ( guardduty_is_enabled, ) - check = guardduty_is_enabled() - result = check.execute() - assert len(result) == 1 - assert result[0].status == "FAIL" - assert result[0].status_extended == "GuardDuty is not enabled." - assert result[0].resource_id == AWS_ACCOUNT_NUMBER - assert result[0].resource_arn == AWS_ACCOUNT_ARN - assert result[0].region == AWS_REGION_EU_WEST_1 + guardduty_client.detectors = [] + check = guardduty_is_enabled() + results = check.execute() + assert len(results) == 0 + + @mock_aws def test_guardduty_enabled(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - region=AWS_REGION_EU_WEST_1, - arn=DETECTOR_ARN, - status=True, - ) - ) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) + + detector_id = guardduty_client.create_detector(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_is_enabled.guardduty_is_enabled.guardduty_client", + new=GuardDuty(aws_provider), ): from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import ( guardduty_is_enabled, ) check = guardduty_is_enabled() - result = check.execute() - assert len(result) == 1 - assert result[0].status == "PASS" - assert ( - result[0].status_extended - == f"GuardDuty detector {DETECTOR_ID} enabled." - ) - assert result[0].resource_id == DETECTOR_ID - assert result[0].resource_arn == DETECTOR_ARN - assert result[0].region == AWS_REGION_EU_WEST_1 + results = check.execute() + assert len(results) == 29 + for result in results: + if result.region == AWS_REGION_EU_WEST_1: + assert result.status == "PASS" + assert ( + result.status_extended + == f"GuardDuty detector {result.resource_id} enabled." + ) + assert result.resource_id == detector_id + assert ( + result.resource_arn + == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}" + ) + assert result.resource_tags == [] + @mock_aws def test_guardduty_configured_but_suspended(self): - guardduty_client = mock.MagicMock - guardduty_client.region = AWS_REGION_EU_WEST_1 - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - arn=DETECTOR_ARN, - region=AWS_REGION_EU_WEST_1, - status=False, - ) - ) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, - ): + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) + + detector_id = guardduty_client.create_detector(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_is_enabled.guardduty_is_enabled.guardduty_client", + new=GuardDuty(aws_provider), + ) as mock_guardduty_client: from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import ( guardduty_is_enabled, ) - check = guardduty_is_enabled() - result = check.execute() - assert len(result) == 1 - assert result[0].status == "FAIL" - assert ( - result[0].status_extended - == f"GuardDuty detector {DETECTOR_ID} configured but suspended." - ) - assert result[0].resource_id == DETECTOR_ID - assert result[0].resource_arn == DETECTOR_ARN - assert result[0].region == AWS_REGION_EU_WEST_1 + for detector in mock_guardduty_client.detectors: + if detector.region == AWS_REGION_EU_WEST_1: + detector.status = False + check = guardduty_is_enabled() + results = check.execute() + assert len(results) == 29 + for result in results: + if result.region == AWS_REGION_EU_WEST_1: + assert result.status == "FAIL" + assert ( + result.status_extended + == f"GuardDuty detector {result.resource_id} configured but suspended." + ) + assert result.resource_id == detector_id + assert ( + result.resource_arn + == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}" + ) + assert result.resource_tags == [] + + @mock_aws def test_guardduty_not_configured(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.region = AWS_REGION_EU_WEST_1 - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - arn=DETECTOR_ARN, - region=AWS_REGION_EU_WEST_1, - ) - ) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, - ): + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) + + detector_id = guardduty_client.create_detector(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_is_enabled.guardduty_is_enabled.guardduty_client", + new=GuardDuty(aws_provider), + ) as mock_guardduty_client: from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import ( guardduty_is_enabled, ) - check = guardduty_is_enabled() - result = check.execute() - assert len(result) == 1 - assert result[0].status == "FAIL" - assert ( - result[0].status_extended - == f"GuardDuty detector {DETECTOR_ID} not configured." - ) - assert result[0].resource_id == DETECTOR_ID - assert result[0].resource_arn == DETECTOR_ARN - assert result[0].region == AWS_REGION_EU_WEST_1 + for detector in mock_guardduty_client.detectors: + if detector.region == AWS_REGION_EU_WEST_1: + detector.status = None + check = guardduty_is_enabled() + results = check.execute() + assert len(results) == 29 + for result in results: + if result.region == AWS_REGION_EU_WEST_1: + assert result.status == "FAIL" + assert ( + result.status_extended + == f"GuardDuty detector {result.resource_id} not configured." + ) + assert result.resource_id == detector_id + assert ( + result.resource_arn + == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}" + ) + assert result.resource_tags == [] + + @mock_aws def test_guardduty_not_configured_muted(self): - guardduty_client = mock.MagicMock - guardduty_client.audit_config = {"mute_non_default_regions": True} - guardduty_client.region = "eu-south-2" - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - arn=DETECTOR_ARN, - region=AWS_REGION_EU_WEST_1, - ) - ) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, - ): + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) + + detector_id = guardduty_client.create_detector(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_is_enabled.guardduty_is_enabled.guardduty_client", + new=GuardDuty(aws_provider), + ) as mock_guardduty_client: from prowler.providers.aws.services.guardduty.guardduty_is_enabled.guardduty_is_enabled import ( guardduty_is_enabled, ) + mock_guardduty_client.audit_config = {"mute_non_default_regions": True} + check = guardduty_is_enabled() - result = check.execute() - assert len(result) == 1 - assert result[0].status == "FAIL" - assert result[0].muted - assert ( - result[0].status_extended - == f"GuardDuty detector {DETECTOR_ID} not configured." - ) - assert result[0].resource_id == DETECTOR_ID - assert result[0].resource_arn == DETECTOR_ARN - assert result[0].region == AWS_REGION_EU_WEST_1 + results = check.execute() + assert len(results) == 29 + for result in results: + if result.region == AWS_REGION_EU_WEST_1: + assert result.status == "FAIL" + assert result.muted + assert ( + result.status_extended + == f"GuardDuty detector {result.resource_id} not configured." + ) + assert result.resource_id == detector_id + assert ( + result.resource_arn + == f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{detector_id}" + ) + assert result.resource_tags == [] + assert result.muted diff --git a/tests/providers/aws/services/guardduty/guardduty_no_high_severity_findings/guardduty_no_high_severity_findings_test.py b/tests/providers/aws/services/guardduty/guardduty_no_high_severity_findings/guardduty_no_high_severity_findings_test.py index 402cf57c79..b068b89225 100644 --- a/tests/providers/aws/services/guardduty/guardduty_no_high_severity_findings/guardduty_no_high_severity_findings_test.py +++ b/tests/providers/aws/services/guardduty/guardduty_no_high_severity_findings/guardduty_no_high_severity_findings_test.py @@ -1,21 +1,43 @@ -from re import search -from unittest import mock -from uuid import uuid4 +from unittest.mock import patch -from prowler.providers.aws.services.guardduty.guardduty_service import Detector -from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1 +import botocore +from boto3 import client +from moto import mock_aws -DETECTOR_ID = str(uuid4()) -DETECTOR_ARN = f"arn:aws:guardduty:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:detector/{DETECTOR_ID}" +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 == "ListFindings": + return { + "FindingIds": [ + "f1", + "f2", + ] + } + # If we don't want to patch the API call + return orig(self, operation_name, kwarg) class Test_guardduty_no_high_severity_findings: + @mock_aws def test_no_detectors(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, + 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_no_high_severity_findings.guardduty_no_high_severity_findings.guardduty_client", + new=GuardDuty(aws_provider), ): from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import ( guardduty_no_high_severity_findings, @@ -25,19 +47,22 @@ class Test_guardduty_no_high_severity_findings: result = check.execute() assert len(result) == 0 + @mock_aws def test_no_high_findings(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - arn=DETECTOR_ARN, - region=AWS_REGION_EU_WEST_1, - ) - ) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) + + detector_id = guardduty_client.create_detector(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_no_high_severity_findings.guardduty_no_high_severity_findings.guardduty_client", + new=GuardDuty(aws_provider), ): from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import ( guardduty_no_high_severity_findings, @@ -47,51 +72,35 @@ class Test_guardduty_no_high_severity_findings: result = check.execute() assert len(result) == 1 assert result[0].status == "PASS" - assert search( - "does not have high severity findings.", result[0].status_extended + assert ( + result[0].status_extended + == f"GuardDuty detector {detector_id} does not have high severity findings." ) - assert result[0].resource_id == DETECTOR_ID - assert result[0].resource_arn == DETECTOR_ARN + assert result[0].resource_id == detector_id assert result[0].region == AWS_REGION_EU_WEST_1 - - def test_not_enabled_account_detector(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=AWS_ACCOUNT_NUMBER, - arn=DETECTOR_ARN, - region=AWS_REGION_EU_WEST_1, - enabled_in_account=False, - ) - ) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, - ): - from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import ( - guardduty_no_high_severity_findings, + 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 == [] - check = guardduty_no_high_severity_findings() - result = check.execute() - assert len(result) == 0 - + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + @mock_aws def test_high_findings(self): - guardduty_client = mock.MagicMock - guardduty_client.detectors = [] - guardduty_client.detectors.append( - Detector( - id=DETECTOR_ID, - region=AWS_REGION_EU_WEST_1, - arn=DETECTOR_ARN, - status=False, - findings=[str(uuid4())], - ) - ) - with mock.patch( - "prowler.providers.aws.services.guardduty.guardduty_service.GuardDuty", - guardduty_client, + guardduty_client = client("guardduty", region_name=AWS_REGION_EU_WEST_1) + + detector_id = guardduty_client.create_detector(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_no_high_severity_findings.guardduty_no_high_severity_findings.guardduty_client", + new=GuardDuty(aws_provider), ): from prowler.providers.aws.services.guardduty.guardduty_no_high_severity_findings.guardduty_no_high_severity_findings import ( guardduty_no_high_severity_findings, @@ -101,7 +110,14 @@ class Test_guardduty_no_high_severity_findings: result = check.execute() assert len(result) == 1 assert result[0].status == "FAIL" - assert search("has 1 high severity findings", result[0].status_extended) - assert result[0].resource_id == DETECTOR_ID - assert result[0].resource_arn == DETECTOR_ARN + assert ( + result[0].status_extended + == f"GuardDuty detector {detector_id} has 2 high severity findings." + ) + 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 == []