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

feat(dms): add new check dms_endpoint_mongodb_authentication_enabled (#5578)

Browse Source 
Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
This commit is contained in:
Daniel Barranquero
2024-11-05 15:58:55 +01:00
committed by GitHub
parent ea038085ba
commit 9802fc141a
7 changed files with 357 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/dms/dms_endpoint_mongodb_authentication_enabled/__init__.py
View File
prowler/providers/aws/services/dms/dms_endpoint_mongodb_authentication_enabled/dms_endpoint_mongodb_authentication_enabled.metadata.json
+32
View File
@@ -0,0 +1,32 @@
{
"Provider": "aws",
"CheckID": "dms_endpoint_mongodb_authentication_enabled",
"CheckTitle": "Check if DMS endpoints for MongoDB have an authentication mechanism enabled.",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices"
],
"ServiceName": "dms",
"SubServiceName": "",
"ResourceIdTemplate": "arn:aws:dms:region:account-id:endpoint/endpoint-id",
"Severity": "medium",
"ResourceType": "AwsDmsEndpoint",
"Description": "This control checks whether an AWS DMS endpoint for MongoDB is configured with an authentication mechanism. The control fails if an authentication type isn't set for the endpoint.",
"Risk": "Without an authentication mechanism enabled, unauthorized users may gain access to sensitive data during migration, increasing the risk of data breaches and security incidents.",
"RelatedUrl": "https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.MongoDB.html",
"Remediation": {
"Code": {
"CLI": "aws dms modify-endpoint --endpoint-arn <endpoint-arn> --username <username> --password <password> --authentication-type <authentication-type>",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/dms-controls.html#dms-11",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable an authentication mechanism on DMS endpoints for MongoDB to ensure secure access control during migration.",
"Url": "https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.MongoDB.html"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
prowler/providers/aws/services/dms/dms_endpoint_mongodb_authentication_enabled/dms_endpoint_mongodb_authentication_enabled.py
+42
View File
@@ -0,0 +1,42 @@
from typing import List
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.dms.dms_client import dms_client
class dms_endpoint_mongodb_authentication_enabled(Check):
"""
Check if AWS DMS Endpoints for MongoDB have an authentication mechanism enabled.
This class verifies whether each AWS DMS Endpoint configured for MongoDB has an authentication
mechanism enabled by checking the `AuthType` property in the endpoint's configuration. The check
ensures that the `AuthType` is not set to "no", indicating that an authentication method is in place.
"""
def execute(self) -> List[Check_Report_AWS]:
"""
Execute the DMS MongoDB authentication type configured check.
Iterates over all DMS Endpoints and generates a report indicating whether
each MongoDB endpoint has an authentication mechanism enabled.
Returns:
List[Check_Report_AWS]: A list of report objects with the results of the check.
"""
findings = []
for endpoint_arn, endpoint in dms_client.endpoints.items():
if endpoint.engine_name == "mongodb":
report = Check_Report_AWS(self.metadata())
report.resource_id = endpoint.id
report.resource_arn = endpoint_arn
report.region = endpoint.region
report.resource_tags = endpoint.tags
report.status = "FAIL"
report.status_extended = f"DMS Endpoint '{endpoint.id}' for MongoDB does not have an authentication mechanism enabled."
if endpoint.mongodb_auth_type != "no":
report.status = "PASS"
report.status_extended = f"DMS Endpoint '{endpoint.id}' for MongoDB has {endpoint.mongodb_auth_type} as the authentication mechanism."
findings.append(report)
return findings
prowler/providers/aws/services/dms/dms_service.py
+4
View File
@@ -71,6 +71,9 @@ class DMS(AWSService):
id=endpoint["EndpointIdentifier"],
region=regional_client.region,
ssl_mode=endpoint.get("SslMode", False),
mongodb_auth_type=endpoint.get("MongoDbSettings", {}).get(
"AuthType", "no"
),
neptune_iam_auth_enabled=endpoint.get(
"NeptuneSettings", {}
).get("IamAuthEnabled", False),
@@ -98,6 +101,7 @@ class Endpoint(BaseModel):
region: str
ssl_mode: str
tags: Optional[list]
mongodb_auth_type: str
neptune_iam_auth_enabled: bool = False
engine_name: str
tests/providers/aws/services/dms/dms_endpoint_mongodb_authentication_enabled/dms_endpoint_mongodb_authentication_enabled_test.py
+271
View File
@@ -0,0 +1,271 @@
from unittest import mock
import botocore
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
make_api_call = botocore.client.BaseClient._make_api_call
DMS_ENDPOINT_NAME = "dms-endpoint"
DMS_ENDPOINT_ARN = f"arn:aws:dms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:endpoint:{DMS_ENDPOINT_NAME}"
DMS_INSTANCE_NAME = "rep-instance"
DMS_INSTANCE_ARN = (
f"arn:aws:dms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:rep:{DMS_INSTANCE_NAME}"
)
def mock_make_api_call_enabled_not_mongodb(self, operation_name, kwarg):
if operation_name == "DescribeEndpoints":
return {
"Endpoints": [
{
"EndpointIdentifier": DMS_ENDPOINT_NAME,
"EndpointArn": DMS_ENDPOINT_ARN,
"SslMode": "require",
"MongoDbSettings": {
"AuthType": "password",
},
"EngineName": "oracle",
}
]
}
elif operation_name == "ListTagsForResource":
if kwarg["ResourceArn"] == DMS_INSTANCE_ARN:
return {
"TagList": [
{"Key": "Name", "Value": "rep-instance"},
{"Key": "Owner", "Value": "admin"},
]
}
elif kwarg["ResourceArn"] == DMS_ENDPOINT_ARN:
return {
"TagList": [
{"Key": "Name", "Value": "dms-endpoint"},
{"Key": "Owner", "Value": "admin"},
]
}
return make_api_call(self, operation_name, kwarg)
def mock_make_api_call_enabled(self, operation_name, kwarg):
if operation_name == "DescribeEndpoints":
return {
"Endpoints": [
{
"EndpointIdentifier": DMS_ENDPOINT_NAME,
"EndpointArn": DMS_ENDPOINT_ARN,
"SslMode": "require",
"MongoDbSettings": {
"AuthType": "password",
},
"EngineName": "mongodb",
}
]
}
elif operation_name == "ListTagsForResource":
if kwarg["ResourceArn"] == DMS_INSTANCE_ARN:
return {
"TagList": [
{"Key": "Name", "Value": "rep-instance"},
{"Key": "Owner", "Value": "admin"},
]
}
elif kwarg["ResourceArn"] == DMS_ENDPOINT_ARN:
return {
"TagList": [
{"Key": "Name", "Value": "dms-endpoint"},
{"Key": "Owner", "Value": "admin"},
]
}
return make_api_call(self, operation_name, kwarg)
def mock_make_api_call_not_enabled(self, operation_name, kwarg):
if operation_name == "DescribeEndpoints":
return {
"Endpoints": [
{
"EndpointIdentifier": DMS_ENDPOINT_NAME,
"EndpointArn": DMS_ENDPOINT_ARN,
"SslMode": "require",
"MongoDbSettings": {
"AuthType": "no",
},
"EngineName": "mongodb",
}
]
}
elif operation_name == "ListTagsForResource":
if kwarg["ResourceArn"] == DMS_INSTANCE_ARN:
return {
"TagList": [
{"Key": "Name", "Value": "rep-instance"},
{"Key": "Owner", "Value": "admin"},
]
}
elif kwarg["ResourceArn"] == DMS_ENDPOINT_ARN:
return {
"TagList": [
{"Key": "Name", "Value": "dms-endpoint"},
{"Key": "Owner", "Value": "admin"},
]
}
return make_api_call(self, operation_name, kwarg)
class Test_dms_endpoint_mongodb_authentication_enabled:
@mock_aws
def test_no_dms_endpoints(self):
dms_client = client("dms", region_name=AWS_REGION_US_EAST_1)
dms_client.endpoints = {}
from prowler.providers.aws.services.dms.dms_service import DMS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.dms.dms_endpoint_mongodb_authentication_enabled.dms_endpoint_mongodb_authentication_enabled.dms_client",
new=DMS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dms.dms_endpoint_mongodb_authentication_enabled.dms_endpoint_mongodb_authentication_enabled import (
dms_endpoint_mongodb_authentication_enabled,
)
check = dms_endpoint_mongodb_authentication_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_dms_not_mongodb_auth_mecanism_enabled(self):
with mock.patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_enabled_not_mongodb,
):
from prowler.providers.aws.services.dms.dms_service import DMS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.dms.dms_endpoint_mongodb_authentication_enabled.dms_endpoint_mongodb_authentication_enabled.dms_client",
new=DMS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dms.dms_endpoint_mongodb_authentication_enabled.dms_endpoint_mongodb_authentication_enabled import (
dms_endpoint_mongodb_authentication_enabled,
)
check = dms_endpoint_mongodb_authentication_enabled()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_dms_mongodb_auth_mecanism_not_enabled(self):
with mock.patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_not_enabled,
):
from prowler.providers.aws.services.dms.dms_service import DMS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.dms.dms_endpoint_mongodb_authentication_enabled.dms_endpoint_mongodb_authentication_enabled.dms_client",
new=DMS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dms.dms_endpoint_mongodb_authentication_enabled.dms_endpoint_mongodb_authentication_enabled import (
dms_endpoint_mongodb_authentication_enabled,
)
check = dms_endpoint_mongodb_authentication_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended == (
"DMS Endpoint 'dms-endpoint' for MongoDB does not have an authentication mechanism enabled."
)
assert result[0].resource_id == "dms-endpoint"
assert (
result[0].resource_arn
== "arn:aws:dms:us-east-1:123456789012:endpoint:dms-endpoint"
)
assert result[0].resource_tags == [
{
"Key": "Name",
"Value": "dms-endpoint",
},
{
"Key": "Owner",
"Value": "admin",
},
]
assert result[0].region == "us-east-1"
@mock_aws
def test_dms_mongodb_auth_mecanism_enabled(self):
with mock.patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_enabled,
):
from prowler.providers.aws.services.dms.dms_service import DMS
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.dms.dms_endpoint_mongodb_authentication_enabled.dms_endpoint_mongodb_authentication_enabled.dms_client",
new=DMS(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dms.dms_endpoint_mongodb_authentication_enabled.dms_endpoint_mongodb_authentication_enabled import (
dms_endpoint_mongodb_authentication_enabled,
)
check = dms_endpoint_mongodb_authentication_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended == (
"DMS Endpoint 'dms-endpoint' for MongoDB has password as the authentication mechanism."
)
assert result[0].resource_id == "dms-endpoint"
assert (
result[0].resource_arn
== "arn:aws:dms:us-east-1:123456789012:endpoint:dms-endpoint"
)
assert result[0].resource_tags == [
{
"Key": "Name",
"Value": "dms-endpoint",
},
{
"Key": "Owner",
"Value": "admin",
},
]
assert result[0].region == "us-east-1"
tests/providers/aws/services/dms/dms_endpoint_ssl_enabled_test/dms_endpoint_ssl_enabled_test.py
+4
View File
@@ -31,6 +31,7 @@ class Test_dms_endpoint_ssl_enabled:
endpoint_arn: Endpoint(
arn=endpoint_arn,
id="test-endpoint-no-ssl",
mongodb_auth_type="no",
engine_name="test-engine",
region=AWS_REGION_US_EAST_1,
ssl_mode="none",
@@ -78,6 +79,7 @@ class Test_dms_endpoint_ssl_enabled:
endpoint_arn: Endpoint(
arn=endpoint_arn,
id="test-endpoint-ssl-require",
mongodb_auth_type="no",
engine_name="test-engine",
region=AWS_REGION_US_EAST_1,
ssl_mode="require",
@@ -123,6 +125,7 @@ class Test_dms_endpoint_ssl_enabled:
arn=endpoint_arn,
id="test-endpoint-ssl-verify-ca",
engine_name="test-engine",
mongodb_auth_type="no",
region=AWS_REGION_US_EAST_1,
ssl_mode="verify-ca",
tags=[{"Key": "Name", "Value": "test-endpoint-ssl-verify-ca"}],
@@ -166,6 +169,7 @@ class Test_dms_endpoint_ssl_enabled:
endpoint_arn: Endpoint(
arn=endpoint_arn,
id="test-endpoint-ssl-verify-full",
mongodb_auth_type="no",
engine_name="test-engine",
region=AWS_REGION_US_EAST_1,
ssl_mode="verify-full",
tests/providers/aws/services/dms/dms_service_test.py
+4
View File
@@ -44,6 +44,9 @@ def mock_make_api_call(self, operation_name, kwargs):
"EndpointIdentifier": DMS_ENDPOINT_NAME,
"EndpointArn": DMS_ENDPOINT_ARN,
"SslMode": "require",
"MongoDbSettings": {
"AuthType": "password",
},
"NeptuneSettings": {
"IamAuthEnabled": True,
},
@@ -125,6 +128,7 @@ class Test_DMS_Service:
assert len(dms.endpoints) == 1
assert dms.endpoints[DMS_ENDPOINT_ARN].id == DMS_ENDPOINT_NAME
assert dms.endpoints[DMS_ENDPOINT_ARN].ssl_mode == "require"
assert dms.endpoints[DMS_ENDPOINT_ARN].mongodb_auth_type == "password"
assert dms.endpoints[DMS_ENDPOINT_ARN].neptune_iam_auth_enabled
assert dms.endpoints[DMS_ENDPOINT_ARN].engine_name == "neptune"