ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-04 02:04:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(dms): lazy load ec2 for public access check (#11899)

Browse Source
This commit is contained in:
Hugo Pereira Brito
2026-07-08 16:42:23 +01:00
committed by GitHub
parent 20aad80a78
commit 25bcbac309
3 changed files with 65 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/changelog.d/dms-instance-no-public-access-ec2-lazy.fixed.md
+1
View File
@@ -0,0 +1 @@
`dms_instance_no_public_access` no longer initializes the full EC2 service when there are no DMS replication instances
prowler/providers/aws/services/dms/dms_instance_no_public_access/dms_instance_no_public_access.py
+7 -2
View File
@@ -1,9 +1,14 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.dms.dms_client import dms_client
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
def _get_ec2_client():
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
return ec2_client
class dms_instance_no_public_access(Check):
def execute(self):
findings = []
@@ -19,7 +24,7 @@ class dms_instance_no_public_access(Check):
if instance.security_groups:
report.status = "PASS"
report.status_extended = f"DMS Replication Instance {instance.id} is set as publicly accessible but filtered with security groups."
for security_group in ec2_client.security_groups.values():
for security_group in _get_ec2_client().security_groups.values():
if security_group.id in instance.security_groups:
for ingress_rule in security_group.ingress_rules:
if check_security_group(
tests/providers/aws/services/dms/dms_instance_no_public_access/dms_no_public_access_test.py
+57 -4
View File
@@ -1,3 +1,5 @@
import ast
from pathlib import Path
from unittest import mock
import botocore
@@ -61,6 +63,42 @@ def mock_make_api_call_private(self, operation_name, kwargs):
class Test_dms_instance_no_public_access:
def test_ec2_client_is_not_imported_at_module_level(self):
repo_root = Path(__file__).parents[6]
check_source = repo_root / (
"prowler/providers/aws/services/dms/dms_instance_no_public_access/dms_instance_no_public_access.py"
)
check_tree = ast.parse(check_source.read_text())
top_level_imports = [
node
for node in check_tree.body
if isinstance(node, (ast.Import, ast.ImportFrom))
]
top_level_ec2_client_imports = [
node
for node in top_level_imports
if (
isinstance(node, ast.ImportFrom)
and node.module == "prowler.providers.aws.services.ec2.ec2_client"
)
or (
isinstance(node, ast.ImportFrom)
and node.module == "prowler.providers.aws.services.ec2"
and any(alias.name == "ec2_client" for alias in node.names)
)
or (
isinstance(node, ast.Import)
and any(
alias.name == "prowler.providers.aws.services.ec2.ec2_client"
for alias in node.names
)
)
]
assert top_level_ec2_client_imports == []
@mock_aws
def test_dms_no_instances(self):
dms_client = client("dms", region_name=AWS_REGION_US_EAST_1)
@@ -79,6 +117,10 @@ class Test_dms_instance_no_public_access:
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access.dms_client",
new=DMS(aws_provider),
),
mock.patch(
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access._get_ec2_client",
side_effect=AssertionError("EC2 client should not be loaded"),
) as get_ec2_client_mock,
):
from prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access import (
dms_instance_no_public_access,
@@ -87,6 +129,7 @@ class Test_dms_instance_no_public_access:
check = dms_instance_no_public_access()
result = check.execute()
assert len(result) == 0
get_ec2_client_mock.assert_not_called()
@mock_aws
def test_dms_private(self):
@@ -108,6 +151,10 @@ class Test_dms_instance_no_public_access:
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access.dms_client",
new=DMS(aws_provider),
),
mock.patch(
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access._get_ec2_client",
side_effect=AssertionError("EC2 client should not be loaded"),
) as get_ec2_client_mock,
):
from prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access import (
dms_instance_no_public_access,
@@ -125,6 +172,7 @@ class Test_dms_instance_no_public_access:
assert result[0].resource_id == DMS_INSTANCE_NAME
assert result[0].resource_arn == DMS_INSTANCE_ARN
assert result[0].resource_tags == []
get_ec2_client_mock.assert_not_called()
@mock_aws
def test_dms_public(self):
@@ -146,6 +194,10 @@ class Test_dms_instance_no_public_access:
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access.dms_client",
new=DMS(aws_provider),
),
mock.patch(
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access._get_ec2_client",
side_effect=AssertionError("EC2 client should not be loaded"),
) as get_ec2_client_mock,
):
from prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access import (
dms_instance_no_public_access,
@@ -163,6 +215,7 @@ class Test_dms_instance_no_public_access:
assert result[0].resource_id == DMS_INSTANCE_NAME
assert result[0].resource_arn == DMS_INSTANCE_ARN
assert result[0].resource_tags == []
get_ec2_client_mock.assert_not_called()
@mock_aws
def test_dms_public_with_public_sg(self):
@@ -218,8 +271,8 @@ class Test_dms_instance_no_public_access:
new=dms_client,
),
mock.patch(
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access.ec2_client",
new=EC2(aws_provider),
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access._get_ec2_client",
return_value=EC2(aws_provider),
),
):
# Test Check
@@ -298,8 +351,8 @@ class Test_dms_instance_no_public_access:
new=dms_client,
),
mock.patch(
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access.ec2_client",
new=EC2(aws_provider),
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access._get_ec2_client",
return_value=EC2(aws_provider),
),
):
# Test Check