feat(mongodbatlas): initial checks

This commit is contained in:
Andoni A.
2025-07-17 12:53:30 +02:00
parent e4d3a51b9e
commit 946f49df44
40 changed files with 1504 additions and 0 deletions
@@ -0,0 +1,34 @@
{
"Provider": "mongodbatlas",
"CheckID": "clusters_authentication_enabled",
"CheckTitle": "Ensure MongoDB Atlas clusters have authentication enabled",
"CheckType": [
"Authentication"
],
"ServiceName": "clusters",
"SubServiceName": "",
"ResourceIdTemplate": "arn:mongodbatlas:cluster:{project_id}:{cluster_name}",
"Severity": "high",
"ResourceType": "Cluster",
"Description": "Ensure MongoDB Atlas clusters have authentication enabled to prevent unauthorized access",
"Risk": "Without authentication enabled, MongoDB Atlas clusters may be vulnerable to unauthorized access, potentially exposing sensitive data or allowing malicious actions",
"RelatedUrl": "https://www.mongodb.com/docs/atlas/reference/api/clusters-get-one/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.atlas.mongodb.com/security-authentication/",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable authentication for MongoDB Atlas clusters by setting authEnabled to true in the cluster configuration.",
"Url": "https://docs.atlas.mongodb.com/security-authentication/"
}
},
"Categories": [
"authentication"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies that MongoDB Atlas clusters have authentication enabled (authEnabled=true) to prevent unauthorized access to the database."
}
@@ -0,0 +1,45 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportMongoDBAtlas
from prowler.providers.mongodbatlas.services.clusters.clusters_client import (
clusters_client,
)
class clusters_authentication_enabled(Check):
"""Check if MongoDB Atlas clusters have authentication enabled
This class verifies that MongoDB Atlas clusters have authentication
enabled to prevent unauthorized access to the database.
"""
def execute(self) -> List[CheckReportMongoDBAtlas]:
"""Execute the MongoDB Atlas cluster authentication enabled check
Iterates over all clusters and checks if they have authentication
enabled (authEnabled=true).
Returns:
List[CheckReportMongoDBAtlas]: A list of reports for each cluster
"""
findings = []
for cluster in clusters_client.clusters.values():
report = CheckReportMongoDBAtlas(metadata=self.metadata(), resource=cluster)
if cluster.auth_enabled:
report.status = "PASS"
report.status_extended = (
f"Cluster {cluster.name} in project {cluster.project_name} "
f"has authentication enabled."
)
else:
report.status = "FAIL"
report.status_extended = (
f"Cluster {cluster.name} in project {cluster.project_name} "
f"does not have authentication enabled."
)
findings.append(report)
return findings
@@ -0,0 +1,34 @@
{
"Provider": "mongodbatlas",
"CheckID": "clusters_backup_enabled",
"CheckTitle": "Ensure MongoDB Atlas clusters have backup enabled",
"CheckType": [
"Backup"
],
"ServiceName": "clusters",
"SubServiceName": "",
"ResourceIdTemplate": "arn:mongodbatlas:cluster:{project_id}:{cluster_name}",
"Severity": "high",
"ResourceType": "Cluster",
"Description": "Ensure MongoDB Atlas clusters have backup enabled to protect against data loss",
"Risk": "Without backup enabled, MongoDB Atlas clusters are vulnerable to data loss in case of failures, corruption, or accidental deletion",
"RelatedUrl": "https://www.mongodb.com/docs/atlas/reference/api/clusters-get-one/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.atlas.mongodb.com/backup/",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable backup for MongoDB Atlas clusters by setting backupEnabled to true in the cluster configuration.",
"Url": "https://docs.atlas.mongodb.com/backup/"
}
},
"Categories": [
"backup"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies that MongoDB Atlas clusters have backup enabled (backupEnabled=true) to ensure data protection and recovery capabilities."
}
@@ -0,0 +1,45 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportMongoDBAtlas
from prowler.providers.mongodbatlas.services.clusters.clusters_client import (
clusters_client,
)
class clusters_backup_enabled(Check):
"""Check if MongoDB Atlas clusters have backup enabled
This class verifies that MongoDB Atlas clusters have backup enabled
to protect against data loss.
"""
def execute(self) -> List[CheckReportMongoDBAtlas]:
"""Execute the MongoDB Atlas cluster backup enabled check
Iterates over all clusters and checks if they have backup
enabled (backupEnabled=true).
Returns:
List[CheckReportMongoDBAtlas]: A list of reports for each cluster
"""
findings = []
for cluster in clusters_client.clusters.values():
report = CheckReportMongoDBAtlas(metadata=self.metadata(), resource=cluster)
if cluster.backup_enabled:
report.status = "PASS"
report.status_extended = (
f"Cluster {cluster.name} in project {cluster.project_name} "
f"has backup enabled."
)
else:
report.status = "FAIL"
report.status_extended = (
f"Cluster {cluster.name} in project {cluster.project_name} "
f"does not have backup enabled."
)
findings.append(report)
return findings
@@ -18,6 +18,8 @@ class Cluster(BaseModel):
state_name: str
encryption_at_rest_provider: Optional[str] = None
backup_enabled: bool = False
auth_enabled: bool = False
ssl_enabled: bool = False
provider_settings: Optional[dict] = {}
replication_specs: Optional[List[dict]] = []
disk_size_gb: Optional[float] = None
@@ -134,6 +136,8 @@ class Clusters(MongoDBAtlasService):
state_name=cluster_data.get("stateName", ""),
encryption_at_rest_provider=encryption_provider,
backup_enabled=backup_enabled,
auth_enabled=cluster_data.get("authEnabled", False),
ssl_enabled=cluster_data.get("sslEnabled", False),
provider_settings=provider_settings,
replication_specs=replication_specs,
disk_size_gb=cluster_data.get("diskSizeGB"),
@@ -0,0 +1,34 @@
{
"Provider": "mongodbatlas",
"CheckID": "clusters_tls_enabled",
"CheckTitle": "Ensure MongoDB Atlas clusters have TLS authentication required",
"CheckType": [
"Encryption"
],
"ServiceName": "clusters",
"SubServiceName": "",
"ResourceIdTemplate": "arn:mongodbatlas:cluster:{project_id}:{cluster_name}",
"Severity": "high",
"ResourceType": "Cluster",
"Description": "Ensure MongoDB Atlas clusters have TLS authentication required to secure data in transit",
"Risk": "Without TLS enabled, MongoDB Atlas clusters are vulnerable to man-in-the-middle attacks and data interception during transmission",
"RelatedUrl": "https://www.mongodb.com/docs/atlas/reference/api/clusters-get-one/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.atlas.mongodb.com/security-connection/",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable TLS for MongoDB Atlas clusters by setting sslEnabled to true in the cluster configuration.",
"Url": "https://docs.atlas.mongodb.com/security-connection/"
}
},
"Categories": [
"encryption"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies that MongoDB Atlas clusters have TLS enabled (sslEnabled=true) to ensure secure data transmission."
}
@@ -0,0 +1,45 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportMongoDBAtlas
from prowler.providers.mongodbatlas.services.clusters.clusters_client import (
clusters_client,
)
class clusters_tls_enabled(Check):
"""Check if MongoDB Atlas clusters have TLS authentication required
This class verifies that MongoDB Atlas clusters have TLS authentication
required to secure data in transit.
"""
def execute(self) -> List[CheckReportMongoDBAtlas]:
"""Execute the MongoDB Atlas cluster TLS enabled check
Iterates over all clusters and checks if they have TLS
enabled (sslEnabled=true).
Returns:
List[CheckReportMongoDBAtlas]: A list of reports for each cluster
"""
findings = []
for cluster in clusters_client.clusters.values():
report = CheckReportMongoDBAtlas(metadata=self.metadata(), resource=cluster)
if cluster.ssl_enabled:
report.status = "PASS"
report.status_extended = (
f"Cluster {cluster.name} in project {cluster.project_name} "
f"has TLS authentication enabled."
)
else:
report.status = "FAIL"
report.status_extended = (
f"Cluster {cluster.name} in project {cluster.project_name} "
f"does not have TLS authentication enabled."
)
findings.append(report)
return findings
@@ -0,0 +1,34 @@
{
"Provider": "mongodbatlas",
"CheckID": "organizations_api_access_list_required",
"CheckTitle": "Ensure organization requires API access list",
"CheckType": [
"Access Control"
],
"ServiceName": "organizations",
"SubServiceName": "",
"ResourceIdTemplate": "arn:mongodbatlas:organization:{org_id}",
"Severity": "medium",
"ResourceType": "Organization",
"Description": "Ensure organization requires API operations to originate from an IP Address added to the API access list",
"Risk": "Without API access list requirement, API operations can originate from any IP address, increasing the risk of unauthorized access",
"RelatedUrl": "https://www.mongodb.com/docs/atlas/reference/api/organization-get-settings/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.atlas.mongodb.com/security-api-access-list/",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable API access list requirement for the organization by setting apiAccessListRequired to true in the organization settings.",
"Url": "https://docs.atlas.mongodb.com/security-api-access-list/"
}
},
"Categories": [
"iam"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies that the organization requires API operations to originate from an IP Address added to the API access list (apiAccessListRequired=true)."
}
@@ -0,0 +1,51 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportMongoDBAtlas
from prowler.providers.mongodbatlas.services.organizations.organizations_client import (
organizations_client,
)
class organizations_api_access_list_required(Check):
"""Check if organization requires API access list
This class verifies that MongoDB Atlas organizations require API operations
to originate from an IP Address added to the API access list.
"""
def execute(self) -> List[CheckReportMongoDBAtlas]:
"""Execute the MongoDB Atlas organization API access list required check
Iterates over all organizations and checks if they require API operations
to originate from an IP Address added to the API access list.
Returns:
List[CheckReportMongoDBAtlas]: A list of reports for each organization
"""
findings = []
for organization in organizations_client.organizations.values():
report = CheckReportMongoDBAtlas(
metadata=self.metadata(), resource=organization
)
api_access_list_required = organization.settings.get(
"apiAccessListRequired", False
)
if api_access_list_required:
report.status = "PASS"
report.status_extended = (
f"Organization {organization.name} requires API operations "
f"to originate from an IP Address added to the API access list."
)
else:
report.status = "FAIL"
report.status_extended = (
f"Organization {organization.name} does not require API operations "
f"to originate from an IP Address added to the API access list."
)
findings.append(report)
return findings
@@ -0,0 +1,6 @@
from prowler.providers.common.provider import Provider
from prowler.providers.mongodbatlas.services.organizations.organizations_service import (
Organizations,
)
organizations_client = Organizations(Provider.get_global_provider())
@@ -0,0 +1,34 @@
{
"Provider": "mongodbatlas",
"CheckID": "organizations_mfa_required",
"CheckTitle": "Ensure organization requires MFA",
"CheckType": [
"Authentication"
],
"ServiceName": "organizations",
"SubServiceName": "",
"ResourceIdTemplate": "arn:mongodbatlas:organization:{org_id}",
"Severity": "high",
"ResourceType": "Organization",
"Description": "Ensure organization requires users to set up Multi-Factor Authentication (MFA) before accessing the organization",
"Risk": "Without MFA requirement, user accounts are vulnerable to credential-based attacks and unauthorized access",
"RelatedUrl": "https://www.mongodb.com/docs/atlas/reference/api/organization-get-settings/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.atlas.mongodb.com/security-multi-factor-authentication/",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable MFA requirement for the organization by setting multiFactorAuthRequired to true in the organization settings.",
"Url": "https://docs.atlas.mongodb.com/security-multi-factor-authentication/"
}
},
"Categories": [
"iam"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies that the organization requires users to set up Multi-Factor Authentication (MFA) before accessing the organization (multiFactorAuthRequired=true)."
}
@@ -0,0 +1,49 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportMongoDBAtlas
from prowler.providers.mongodbatlas.services.organizations.organizations_client import (
organizations_client,
)
class organizations_mfa_required(Check):
"""Check if organization requires MFA
This class verifies that MongoDB Atlas organizations require users
to set up Multi-Factor Authentication (MFA) before accessing the organization.
"""
def execute(self) -> List[CheckReportMongoDBAtlas]:
"""Execute the MongoDB Atlas organization MFA required check
Iterates over all organizations and checks if they require users
to set up Multi-Factor Authentication (MFA) before accessing the organization.
Returns:
List[CheckReportMongoDBAtlas]: A list of reports for each organization
"""
findings = []
for organization in organizations_client.organizations.values():
report = CheckReportMongoDBAtlas(
metadata=self.metadata(), resource=organization
)
mfa_required = organization.settings.get("multiFactorAuthRequired", False)
if mfa_required:
report.status = "PASS"
report.status_extended = (
f"Organization {organization.name} requires users to set up "
f"Multi-Factor Authentication (MFA) before accessing the organization."
)
else:
report.status = "FAIL"
report.status_extended = (
f"Organization {organization.name} does not require users to set up "
f"Multi-Factor Authentication (MFA) before accessing the organization."
)
findings.append(report)
return findings
@@ -0,0 +1,34 @@
{
"Provider": "mongodbatlas",
"CheckID": "organizations_security_contact_defined",
"CheckTitle": "Ensure organization has a Security Contact defined",
"CheckType": [
"Security Contact"
],
"ServiceName": "organizations",
"SubServiceName": "",
"ResourceIdTemplate": "arn:mongodbatlas:organization:{org_id}",
"Severity": "medium",
"ResourceType": "Organization",
"Description": "Ensure organization has a security contact defined to receive security-related notifications",
"Risk": "Without a security contact, the organization may not receive important security notifications and alerts",
"RelatedUrl": "https://www.mongodb.com/docs/atlas/reference/api/organization-get-settings/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.atlas.mongodb.com/security-contact/",
"Terraform": ""
},
"Recommendation": {
"Text": "Set a security contact email address in the organization settings to receive security-related notifications.",
"Url": "https://docs.atlas.mongodb.com/security-contact/"
}
},
"Categories": [
"security-contacts"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies that the organization has a security contact defined (securityContact field) to receive security-related notifications."
}
@@ -0,0 +1,49 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportMongoDBAtlas
from prowler.providers.mongodbatlas.services.organizations.organizations_client import (
organizations_client,
)
class organizations_security_contact_defined(Check):
"""Check if organization has a Security Contact defined
This class verifies that MongoDB Atlas organizations have a security contact
defined to receive security-related notifications.
"""
def execute(self) -> List[CheckReportMongoDBAtlas]:
"""Execute the MongoDB Atlas organization security contact defined check
Iterates over all organizations and checks if they have a security contact
defined to receive security-related notifications.
Returns:
List[CheckReportMongoDBAtlas]: A list of reports for each organization
"""
findings = []
for organization in organizations_client.organizations.values():
report = CheckReportMongoDBAtlas(
metadata=self.metadata(), resource=organization
)
security_contact = organization.settings.get("securityContact")
if security_contact:
report.status = "PASS"
report.status_extended = (
f"Organization {organization.name} has a security contact defined: "
f"{security_contact}"
)
else:
report.status = "FAIL"
report.status_extended = (
f"Organization {organization.name} does not have a security contact "
f"defined to receive security-related notifications."
)
findings.append(report)
return findings
@@ -0,0 +1,94 @@
from typing import Dict, Optional
from pydantic.v1 import BaseModel
from prowler.lib.logger import logger
from prowler.providers.mongodbatlas.lib.service.service import MongoDBAtlasService
class Organization(BaseModel):
"""MongoDB Atlas Organization model"""
id: str
name: str
settings: Optional[dict] = {}
class Organizations(MongoDBAtlasService):
"""MongoDB Atlas Organizations service"""
def __init__(self, provider):
super().__init__(__class__.__name__, provider)
self.organizations = self._list_organizations()
def _list_organizations(self) -> Dict[str, Organization]:
"""
List all MongoDB Atlas organizations
Returns:
Dict[str, Organization]: Dictionary of organizations indexed by organization ID
"""
logger.info("Organizations - Listing MongoDB Atlas organizations...")
organizations = {}
try:
# If organization_id filter is set, only get that organization
if self.provider.organization_id:
org_data = self._make_request(
"GET", f"/orgs/{self.provider.organization_id}"
)
organizations[org_data["id"]] = self._process_organization(org_data)
else:
# Get all organizations with pagination
all_orgs = self._paginate_request("/orgs")
for org_data in all_orgs:
organizations[org_data["id"]] = self._process_organization(org_data)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
logger.info(f"Found {len(organizations)} MongoDB Atlas organizations")
return organizations
def _process_organization(self, org_data: dict) -> Organization:
"""
Process a single organization and fetch additional details
Args:
org_data: Raw organization data from API
Returns:
Organization: Processed organization object
"""
org_id = org_data["id"]
# Get organization settings
org_settings = self._get_organization_settings(org_id)
return Organization(
id=org_id,
name=org_data.get("name", ""),
settings=org_settings,
)
def _get_organization_settings(self, org_id: str) -> dict:
"""
Get organization settings
Args:
org_id: Organization ID
Returns:
dict: Organization settings
"""
try:
settings = self._make_request("GET", f"/orgs/{org_id}/settings")
return settings
except Exception as error:
logger.error(
f"Error getting organization settings for organization {org_id}: {error}"
)
return {}
@@ -0,0 +1,34 @@
{
"Provider": "mongodbatlas",
"CheckID": "organizations_service_account_secrets_expiration",
"CheckTitle": "Ensure organization has maximum period expiration for Admin API Service Account Secrets",
"CheckType": [
"Secrets Management"
],
"ServiceName": "organizations",
"SubServiceName": "",
"ResourceIdTemplate": "arn:mongodbatlas:organization:{org_id}",
"Severity": "medium",
"ResourceType": "Organization",
"Description": "Ensure organization has a maximum period before expiry for new Atlas Admin API Service Account secrets",
"Risk": "Without proper expiration limits, service account secrets may remain valid for extended periods, increasing security risks",
"RelatedUrl": "https://www.mongodb.com/docs/atlas/reference/api/organization-get-settings/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.atlas.mongodb.com/security-service-accounts/",
"Terraform": ""
},
"Recommendation": {
"Text": "Set maxServiceAccountSecretValidityInHours to 8 hours or less in the organization settings to ensure service account secrets expire regularly.",
"Url": "https://docs.atlas.mongodb.com/security-service-accounts/"
}
},
"Categories": [
"secrets-management"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies that the organization has a maximum period expiration for Admin API Service Account secrets set to 8 hours or less (configurable)."
}
@@ -0,0 +1,62 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportMongoDBAtlas
from prowler.providers.mongodbatlas.services.organizations.organizations_client import (
organizations_client,
)
class organizations_service_account_secrets_expiration(Check):
"""Check if organization has maximum period expiration for Admin API Service Account Secrets
This class verifies that MongoDB Atlas organizations have a maximum period
before expiry for new Atlas Admin API Service Account secrets.
"""
def execute(self) -> List[CheckReportMongoDBAtlas]:
"""Execute the MongoDB Atlas organization service account secrets expiration check
Iterates over all organizations and checks if they have a maximum period
expiration for Admin API Service Account secrets set to 8 hours or less.
Returns:
List[CheckReportMongoDBAtlas]: A list of reports for each organization
"""
findings = []
# Default maximum hours as specified in the ticket (configurable)
max_hours_threshold = 8
for organization in organizations_client.organizations.values():
report = CheckReportMongoDBAtlas(
metadata=self.metadata(), resource=organization
)
max_validity_hours = organization.settings.get(
"maxServiceAccountSecretValidityInHours"
)
if max_validity_hours is None:
report.status = "FAIL"
report.status_extended = (
f"Organization {organization.name} does not have a maximum period "
f"expiration configured for Admin API Service Account secrets."
)
elif max_validity_hours <= max_hours_threshold:
report.status = "PASS"
report.status_extended = (
f"Organization {organization.name} has a maximum period expiration "
f"of {max_validity_hours} hours for Admin API Service Account secrets, "
f"which is within the recommended threshold of {max_hours_threshold} hours."
)
else:
report.status = "FAIL"
report.status_extended = (
f"Organization {organization.name} has a maximum period expiration "
f"of {max_validity_hours} hours for Admin API Service Account secrets, "
f"which exceeds the recommended threshold of {max_hours_threshold} hours."
)
findings.append(report)
return findings
@@ -0,0 +1,34 @@
{
"Provider": "mongodbatlas",
"CheckID": "projects_auditing_enabled",
"CheckTitle": "Ensure database auditing is enabled",
"CheckType": [
"Auditing"
],
"ServiceName": "projects",
"SubServiceName": "",
"ResourceIdTemplate": "arn:mongodbatlas:project:{project_id}",
"Severity": "medium",
"ResourceType": "Project",
"Description": "Ensure database auditing is enabled to track database operations and security events",
"Risk": "Without auditing enabled, security events and database operations are not logged, making it difficult to detect unauthorized access or troubleshoot issues",
"RelatedUrl": "https://www.mongodb.com/docs/atlas/reference/api/auditing-get-configuration/",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.atlas.mongodb.com/database-auditing/",
"Terraform": ""
},
"Recommendation": {
"Text": "Enable database auditing for the MongoDB Atlas project by configuring audit filters and destinations.",
"Url": "https://docs.atlas.mongodb.com/database-auditing/"
}
},
"Categories": [
"logging"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "This check verifies that database auditing is enabled by checking the audit configuration for the project."
}
@@ -0,0 +1,53 @@
from typing import List
from prowler.lib.check.models import Check, CheckReportMongoDBAtlas
from prowler.providers.mongodbatlas.services.projects.projects_client import (
projects_client,
)
class projects_auditing_enabled(Check):
"""Check if database auditing is enabled for MongoDB Atlas projects
This class verifies that MongoDB Atlas projects have database auditing
enabled to track database operations and security events.
"""
def execute(self) -> List[CheckReportMongoDBAtlas]:
"""Execute the MongoDB Atlas project auditing enabled check
Iterates over all projects and checks if they have database auditing
enabled by examining the audit configuration.
Returns:
List[CheckReportMongoDBAtlas]: A list of reports for each project
"""
findings = []
for project in projects_client.projects.values():
report = CheckReportMongoDBAtlas(metadata=self.metadata(), resource=project)
if not project.audit_config:
report.status = "FAIL"
report.status_extended = f"Project {project.name} does not have audit configuration available."
else:
# Check if audit configuration is enabled
enabled = project.audit_config.get("enabled", False)
audit_filter = project.audit_config.get("auditFilter")
if enabled:
report.status = "PASS"
report.status_extended = (
f"Project {project.name} has database auditing enabled."
)
if audit_filter:
report.status_extended += (
f" Audit filter configured: {audit_filter}"
)
else:
report.status = "FAIL"
report.status_extended = f"Project {project.name} does not have database auditing enabled."
findings.append(report)
return findings
@@ -17,6 +17,7 @@ class Project(BaseModel):
cluster_count: int
network_access_entries: List[MongoDBAtlasNetworkAccessEntry] = []
project_settings: Optional[dict] = {}
audit_config: Optional[dict] = {}
class Projects(MongoDBAtlasService):
@@ -84,6 +85,9 @@ class Projects(MongoDBAtlasService):
# Get project settings
project_settings = self._get_project_settings(project_id)
# Get audit configuration
audit_config = self._get_audit_config(project_id)
return Project(
id=project_id,
name=project_data.get("name", ""),
@@ -92,6 +96,7 @@ class Projects(MongoDBAtlasService):
cluster_count=cluster_count,
network_access_entries=network_access_entries,
project_settings=project_settings,
audit_config=audit_config,
)
def _get_cluster_count(self, project_id: str) -> int:
@@ -165,3 +170,22 @@ class Projects(MongoDBAtlasService):
f"Error getting project settings for project {project_id}: {error}"
)
return {}
def _get_audit_config(self, project_id: str) -> dict:
"""
Get audit configuration for a project
Args:
project_id: Project ID
Returns:
dict: Audit configuration
"""
try:
audit_config = self._make_request("GET", f"/groups/{project_id}/auditLog")
return audit_config
except Exception as error:
logger.error(
f"Error getting audit configuration for project {project_id}: {error}"
)
return {}
@@ -0,0 +1,73 @@
from unittest.mock import MagicMock, patch
from prowler.providers.mongodbatlas.services.clusters.clusters_service import Cluster
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
CLUSTER_ID,
CLUSTER_NAME,
CLUSTER_TYPE,
MONGO_VERSION,
PROJECT_ID,
PROJECT_NAME,
STATE_NAME,
set_mocked_mongodbatlas_provider,
)
class TestClustersAuthenticationEnabled:
def _create_cluster(self, auth_enabled=False):
"""Helper method to create a cluster with authentication settings"""
return Cluster(
id=CLUSTER_ID,
name=CLUSTER_NAME,
project_id=PROJECT_ID,
project_name=PROJECT_NAME,
mongo_db_version=MONGO_VERSION,
cluster_type=CLUSTER_TYPE,
state_name=STATE_NAME,
auth_enabled=auth_enabled,
ssl_enabled=False,
backup_enabled=False,
encryption_at_rest_provider=None,
provider_settings={},
replication_specs=[],
)
def _execute_check_with_cluster(self, cluster):
"""Helper method to execute check with a cluster"""
clusters_client = MagicMock()
clusters_client.clusters = {f"{PROJECT_ID}:{CLUSTER_NAME}": cluster}
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
),
patch(
"prowler.providers.mongodbatlas.services.clusters.clusters_authentication_enabled.clusters_authentication_enabled.clusters_client",
new=clusters_client,
),
):
from prowler.providers.mongodbatlas.services.clusters.clusters_authentication_enabled.clusters_authentication_enabled import (
clusters_authentication_enabled,
)
check = clusters_authentication_enabled()
return check.execute()
def test_check_with_authentication_enabled(self):
"""Test check with authentication enabled"""
cluster = self._create_cluster(auth_enabled=True)
reports = self._execute_check_with_cluster(cluster)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert "has authentication enabled" in reports[0].status_extended
def test_check_with_authentication_disabled(self):
"""Test check with authentication disabled"""
cluster = self._create_cluster(auth_enabled=False)
reports = self._execute_check_with_cluster(cluster)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert "does not have authentication enabled" in reports[0].status_extended
@@ -0,0 +1,73 @@
from unittest.mock import MagicMock, patch
from prowler.providers.mongodbatlas.services.clusters.clusters_service import Cluster
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
CLUSTER_ID,
CLUSTER_NAME,
CLUSTER_TYPE,
MONGO_VERSION,
PROJECT_ID,
PROJECT_NAME,
STATE_NAME,
set_mocked_mongodbatlas_provider,
)
class TestClustersBackupEnabled:
def _create_cluster(self, backup_enabled=False):
"""Helper method to create a cluster with backup settings"""
return Cluster(
id=CLUSTER_ID,
name=CLUSTER_NAME,
project_id=PROJECT_ID,
project_name=PROJECT_NAME,
mongo_db_version=MONGO_VERSION,
cluster_type=CLUSTER_TYPE,
state_name=STATE_NAME,
auth_enabled=False,
ssl_enabled=False,
backup_enabled=backup_enabled,
encryption_at_rest_provider=None,
provider_settings={},
replication_specs=[],
)
def _execute_check_with_cluster(self, cluster):
"""Helper method to execute check with a cluster"""
clusters_client = MagicMock()
clusters_client.clusters = {f"{PROJECT_ID}:{CLUSTER_NAME}": cluster}
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
),
patch(
"prowler.providers.mongodbatlas.services.clusters.clusters_backup_enabled.clusters_backup_enabled.clusters_client",
new=clusters_client,
),
):
from prowler.providers.mongodbatlas.services.clusters.clusters_backup_enabled.clusters_backup_enabled import (
clusters_backup_enabled,
)
check = clusters_backup_enabled()
return check.execute()
def test_check_with_backup_enabled(self):
"""Test check with backup enabled"""
cluster = self._create_cluster(backup_enabled=True)
reports = self._execute_check_with_cluster(cluster)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert "has backup enabled" in reports[0].status_extended
def test_check_with_backup_disabled(self):
"""Test check with backup disabled"""
cluster = self._create_cluster(backup_enabled=False)
reports = self._execute_check_with_cluster(cluster)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert "does not have backup enabled" in reports[0].status_extended
@@ -31,6 +31,8 @@ class TestClustersEncryptionAtRestEnabled:
state_name=STATE_NAME,
encryption_at_rest_provider=encryption_at_rest_provider,
backup_enabled=False,
auth_enabled=False,
ssl_enabled=False,
provider_settings=provider_settings,
replication_specs=[],
paused=paused,
@@ -0,0 +1,73 @@
from unittest.mock import MagicMock, patch
from prowler.providers.mongodbatlas.services.clusters.clusters_service import Cluster
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
CLUSTER_ID,
CLUSTER_NAME,
CLUSTER_TYPE,
MONGO_VERSION,
PROJECT_ID,
PROJECT_NAME,
STATE_NAME,
set_mocked_mongodbatlas_provider,
)
class TestClustersTlsEnabled:
def _create_cluster(self, ssl_enabled=False):
"""Helper method to create a cluster with TLS settings"""
return Cluster(
id=CLUSTER_ID,
name=CLUSTER_NAME,
project_id=PROJECT_ID,
project_name=PROJECT_NAME,
mongo_db_version=MONGO_VERSION,
cluster_type=CLUSTER_TYPE,
state_name=STATE_NAME,
auth_enabled=False,
ssl_enabled=ssl_enabled,
backup_enabled=False,
encryption_at_rest_provider=None,
provider_settings={},
replication_specs=[],
)
def _execute_check_with_cluster(self, cluster):
"""Helper method to execute check with a cluster"""
clusters_client = MagicMock()
clusters_client.clusters = {f"{PROJECT_ID}:{CLUSTER_NAME}": cluster}
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
),
patch(
"prowler.providers.mongodbatlas.services.clusters.clusters_tls_enabled.clusters_tls_enabled.clusters_client",
new=clusters_client,
),
):
from prowler.providers.mongodbatlas.services.clusters.clusters_tls_enabled.clusters_tls_enabled import (
clusters_tls_enabled,
)
check = clusters_tls_enabled()
return check.execute()
def test_check_with_tls_enabled(self):
"""Test check with TLS enabled"""
cluster = self._create_cluster(ssl_enabled=True)
reports = self._execute_check_with_cluster(cluster)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert "has TLS authentication enabled" in reports[0].status_extended
def test_check_with_tls_disabled(self):
"""Test check with TLS disabled"""
cluster = self._create_cluster(ssl_enabled=False)
reports = self._execute_check_with_cluster(cluster)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert "does not have TLS authentication enabled" in reports[0].status_extended
@@ -0,0 +1,81 @@
from unittest.mock import MagicMock, patch
from prowler.providers.mongodbatlas.services.organizations.organizations_service import (
Organization,
)
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
ORG_ID,
set_mocked_mongodbatlas_provider,
)
class TestOrganizationsApiAccessListRequired:
def _create_organization(self, api_access_list_required=False):
"""Helper method to create an organization with API access list settings"""
return Organization(
id=ORG_ID,
name="Test Organization",
settings={"apiAccessListRequired": api_access_list_required},
)
def _execute_check_with_organization(self, organization):
"""Helper method to execute check with an organization"""
organizations_client = MagicMock()
organizations_client.organizations = {ORG_ID: organization}
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
),
patch(
"prowler.providers.mongodbatlas.services.organizations.organizations_api_access_list_required.organizations_api_access_list_required.organizations_client",
new=organizations_client,
),
):
from prowler.providers.mongodbatlas.services.organizations.organizations_api_access_list_required.organizations_api_access_list_required import (
organizations_api_access_list_required,
)
check = organizations_api_access_list_required()
return check.execute()
def test_check_with_api_access_list_required(self):
"""Test check with API access list required"""
organization = self._create_organization(api_access_list_required=True)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert (
"requires API operations to originate from an IP Address added to the API access list"
in reports[0].status_extended
)
def test_check_with_api_access_list_not_required(self):
"""Test check with API access list not required"""
organization = self._create_organization(api_access_list_required=False)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert (
"does not require API operations to originate from an IP Address added to the API access list"
in reports[0].status_extended
)
def test_check_with_no_api_access_list_setting(self):
"""Test check with no API access list setting"""
organization = Organization(
id=ORG_ID,
name="Test Organization",
settings={},
)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert (
"does not require API operations to originate from an IP Address added to the API access list"
in reports[0].status_extended
)
@@ -0,0 +1,81 @@
from unittest.mock import MagicMock, patch
from prowler.providers.mongodbatlas.services.organizations.organizations_service import (
Organization,
)
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
ORG_ID,
set_mocked_mongodbatlas_provider,
)
class TestOrganizationsMfaRequired:
def _create_organization(self, mfa_required=False):
"""Helper method to create an organization with MFA settings"""
return Organization(
id=ORG_ID,
name="Test Organization",
settings={"multiFactorAuthRequired": mfa_required},
)
def _execute_check_with_organization(self, organization):
"""Helper method to execute check with an organization"""
organizations_client = MagicMock()
organizations_client.organizations = {ORG_ID: organization}
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
),
patch(
"prowler.providers.mongodbatlas.services.organizations.organizations_mfa_required.organizations_mfa_required.organizations_client",
new=organizations_client,
),
):
from prowler.providers.mongodbatlas.services.organizations.organizations_mfa_required.organizations_mfa_required import (
organizations_mfa_required,
)
check = organizations_mfa_required()
return check.execute()
def test_check_with_mfa_required(self):
"""Test check with MFA required"""
organization = self._create_organization(mfa_required=True)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert (
"requires users to set up Multi-Factor Authentication"
in reports[0].status_extended
)
def test_check_with_mfa_not_required(self):
"""Test check with MFA not required"""
organization = self._create_organization(mfa_required=False)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert (
"does not require users to set up Multi-Factor Authentication"
in reports[0].status_extended
)
def test_check_with_no_mfa_setting(self):
"""Test check with no MFA setting"""
organization = Organization(
id=ORG_ID,
name="Test Organization",
settings={},
)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert (
"does not require users to set up Multi-Factor Authentication"
in reports[0].status_extended
)
@@ -0,0 +1,77 @@
from unittest.mock import MagicMock, patch
from prowler.providers.mongodbatlas.services.organizations.organizations_service import (
Organization,
)
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
ORG_ID,
set_mocked_mongodbatlas_provider,
)
class TestOrganizationsSecurityContactDefined:
def _create_organization(self, security_contact=None):
"""Helper method to create an organization with security contact settings"""
settings = {}
if security_contact is not None:
settings["securityContact"] = security_contact
return Organization(
id=ORG_ID,
name="Test Organization",
settings=settings,
)
def _execute_check_with_organization(self, organization):
"""Helper method to execute check with an organization"""
organizations_client = MagicMock()
organizations_client.organizations = {ORG_ID: organization}
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
),
patch(
"prowler.providers.mongodbatlas.services.organizations.organizations_security_contact_defined.organizations_security_contact_defined.organizations_client",
new=organizations_client,
),
):
from prowler.providers.mongodbatlas.services.organizations.organizations_security_contact_defined.organizations_security_contact_defined import (
organizations_security_contact_defined,
)
check = organizations_security_contact_defined()
return check.execute()
def test_check_with_security_contact_defined(self):
"""Test check with security contact defined"""
organization = self._create_organization(
security_contact="security@example.com"
)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert (
"has a security contact defined: security@example.com"
in reports[0].status_extended
)
def test_check_with_no_security_contact(self):
"""Test check with no security contact"""
organization = self._create_organization()
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert "does not have a security contact defined" in reports[0].status_extended
def test_check_with_empty_security_contact(self):
"""Test check with empty security contact"""
organization = self._create_organization(security_contact="")
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert "does not have a security contact defined" in reports[0].status_extended
@@ -0,0 +1,90 @@
from unittest.mock import MagicMock, patch
from prowler.providers.mongodbatlas.services.organizations.organizations_service import (
Organization,
)
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
ORG_ID,
set_mocked_mongodbatlas_provider,
)
class TestOrganizationsServiceAccountSecretsExpiration:
def _create_organization(self, max_validity_hours=None):
"""Helper method to create an organization with service account secrets expiration settings"""
settings = {}
if max_validity_hours is not None:
settings["maxServiceAccountSecretValidityInHours"] = max_validity_hours
return Organization(
id=ORG_ID,
name="Test Organization",
settings=settings,
)
def _execute_check_with_organization(self, organization):
"""Helper method to execute check with an organization"""
organizations_client = MagicMock()
organizations_client.organizations = {ORG_ID: organization}
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
),
patch(
"prowler.providers.mongodbatlas.services.organizations.organizations_service_account_secrets_expiration.organizations_service_account_secrets_expiration.organizations_client",
new=organizations_client,
),
):
from prowler.providers.mongodbatlas.services.organizations.organizations_service_account_secrets_expiration.organizations_service_account_secrets_expiration import (
organizations_service_account_secrets_expiration,
)
check = organizations_service_account_secrets_expiration()
return check.execute()
def test_check_with_valid_expiration_hours(self):
"""Test check with valid expiration hours (8 hours)"""
organization = self._create_organization(max_validity_hours=8)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert (
"within the recommended threshold of 8 hours" in reports[0].status_extended
)
def test_check_with_valid_expiration_hours_lower(self):
"""Test check with valid expiration hours (4 hours)"""
organization = self._create_organization(max_validity_hours=4)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert (
"within the recommended threshold of 8 hours" in reports[0].status_extended
)
def test_check_with_invalid_expiration_hours(self):
"""Test check with invalid expiration hours (24 hours)"""
organization = self._create_organization(max_validity_hours=24)
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert (
"exceeds the recommended threshold of 8 hours" in reports[0].status_extended
)
def test_check_with_no_expiration_setting(self):
"""Test check with no expiration setting"""
organization = self._create_organization()
reports = self._execute_check_with_organization(organization)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert (
"does not have a maximum period expiration configured"
in reports[0].status_extended
)
@@ -0,0 +1,64 @@
from unittest.mock import patch
from prowler.providers.mongodbatlas.services.organizations.organizations_service import (
Organizations,
)
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
ORG_ID,
set_mocked_mongodbatlas_provider,
)
class TestOrganizationsService:
def test_organizations_service_initialization(self):
"""Test Organizations service initialization"""
with patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
):
with patch.object(
Organizations, "_list_organizations", return_value={}
) as mock_list:
service = Organizations(set_mocked_mongodbatlas_provider())
assert service.organizations == {}
mock_list.assert_called_once()
def test_process_organization(self):
"""Test organization processing"""
provider = set_mocked_mongodbatlas_provider()
with patch.object(Organizations, "_list_organizations", return_value={}):
service = Organizations(provider)
# Mock the settings request
mock_settings = {
"apiAccessListRequired": True,
"multiFactorAuthRequired": True,
"maxServiceAccountSecretValidityInHours": 8,
"securityContact": "security@example.com",
}
with patch.object(
service, "_get_organization_settings", return_value=mock_settings
):
org_data = {"id": ORG_ID, "name": "Test Organization"}
organization = service._process_organization(org_data)
assert organization.id == ORG_ID
assert organization.name == "Test Organization"
assert organization.settings == mock_settings
def test_get_organization_settings_error_handling(self):
"""Test error handling in get organization settings"""
provider = set_mocked_mongodbatlas_provider()
with patch.object(Organizations, "_list_organizations", return_value={}):
service = Organizations(provider)
# Mock the request to raise an exception
with patch.object(
service, "_make_request", side_effect=Exception("API Error")
):
settings = service._get_organization_settings(ORG_ID)
assert settings == {}
@@ -0,0 +1,90 @@
from unittest.mock import MagicMock, patch
from prowler.providers.mongodbatlas.services.projects.projects_service import Project
from tests.providers.mongodbatlas.mongodbatlas_fixtures import (
ORG_ID,
PROJECT_ID,
PROJECT_NAME,
set_mocked_mongodbatlas_provider,
)
class TestProjectsAuditingEnabled:
def _create_project(self, audit_config=None):
"""Helper method to create a project with audit settings"""
if audit_config is None:
audit_config = {}
return Project(
id=PROJECT_ID,
name=PROJECT_NAME,
org_id=ORG_ID,
created="2024-01-01T00:00:00Z",
cluster_count=1,
network_access_entries=[],
project_settings={},
audit_config=audit_config,
)
def _execute_check_with_project(self, project):
"""Helper method to execute check with a project"""
projects_client = MagicMock()
projects_client.projects = {PROJECT_ID: project}
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_mongodbatlas_provider(),
),
patch(
"prowler.providers.mongodbatlas.services.projects.projects_auditing_enabled.projects_auditing_enabled.projects_client",
new=projects_client,
),
):
from prowler.providers.mongodbatlas.services.projects.projects_auditing_enabled.projects_auditing_enabled import (
projects_auditing_enabled,
)
check = projects_auditing_enabled()
return check.execute()
def test_check_with_auditing_enabled(self):
"""Test check with auditing enabled"""
project = self._create_project(audit_config={"enabled": True})
reports = self._execute_check_with_project(project)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert "has database auditing enabled" in reports[0].status_extended
def test_check_with_auditing_enabled_and_filter(self):
"""Test check with auditing enabled and filter configured"""
project = self._create_project(
audit_config={"enabled": True, "auditFilter": "{'action': 'authenticate'}"}
)
reports = self._execute_check_with_project(project)
assert len(reports) == 1
assert reports[0].status == "PASS"
assert "has database auditing enabled" in reports[0].status_extended
assert "Audit filter configured" in reports[0].status_extended
def test_check_with_auditing_disabled(self):
"""Test check with auditing disabled"""
project = self._create_project(audit_config={"enabled": False})
reports = self._execute_check_with_project(project)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert "does not have database auditing enabled" in reports[0].status_extended
def test_check_with_no_audit_config(self):
"""Test check with no audit configuration"""
project = self._create_project(audit_config={})
reports = self._execute_check_with_project(project)
assert len(reports) == 1
assert reports[0].status == "FAIL"
assert (
"does not have audit configuration available" in reports[0].status_extended
)
@@ -21,6 +21,7 @@ class TestProjectsNetworkAccessListNotOpenToWorld:
cluster_count=0,
network_access_entries=network_entries,
project_settings={},
audit_config={},
)
def _execute_check_with_project(self, project):