mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
chore(guardduty): mock failing tests using moto (#5334)
This commit is contained in:
committed by
GitHub
parent
b27898de1d
commit
aa32634105
@@ -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 = []
|
||||
|
||||
+117
-93
@@ -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}"
|
||||
)
|
||||
|
||||
+159
-131
@@ -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
|
||||
|
||||
+83
-67
@@ -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 == []
|
||||
|
||||
Reference in New Issue
Block a user