feat(azure): New Azure checks related to CosmosDB (#3386)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
This commit is contained in:
Pedro Martín
2024-02-13 13:53:36 +01:00
committed by GitHub
parent 4740a7b930
commit 355f589e5a
18 changed files with 686 additions and 1 deletions

View File

@@ -171,6 +171,42 @@ expected_packages = [
name="prowler.providers.azure.services.sqlserver.sqlserver_va_emails_notifications_admins_enabled.sqlserver_va_emails_notifications_admins_enabled",
ispkg=False,
),
ModuleInfo(
module_finder=FileFinder("/root_dir/prowler/providers/azure/services/cosmosdb"),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks",
ispkg=True,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_firewall_use_selected_networks"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks.cosmosdb_account_firewall_use_selected_networks",
ispkg=False,
),
ModuleInfo(
module_finder=FileFinder("/root_dir/prowler/providers/azure/services/cosmosdb"),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints",
ispkg=True,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_use_private_endpoints"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints.cosmosdb_account_use_private_endpoints",
ispkg=False,
),
ModuleInfo(
module_finder=FileFinder("/root_dir/prowler/providers/azure/services/cosmosdb"),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac",
ispkg=True,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_use_aad_and_rbac"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac.cosmosdb_account_use_aad_and_rbac",
ispkg=False,
),
]
@@ -320,6 +356,48 @@ def mock_list_modules(*_):
name="prowler.providers.azure.services.sqlserver.sqlserver_va_emails_notifications_admins_enabled.sqlserver_va_emails_notifications_admins_enabled",
ispkg=False,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks",
ispkg=True,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_firewall_use_selected_networks"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks.cosmosdb_account_firewall_use_selected_networks",
ispkg=False,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints",
ispkg=True,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_use_private_endpoints"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints.cosmosdb_account_use_private_endpoints",
ispkg=False,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac",
ispkg=True,
),
ModuleInfo(
module_finder=FileFinder(
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_use_aad_and_rbac"
),
name="prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac.cosmosdb_account_use_aad_and_rbac",
ispkg=False,
),
]
return modules
@@ -729,6 +807,18 @@ class Test_Check:
"sqlserver_va_emails_notifications_admins_enabled",
"/root_dir/prowler/providers/azure/services/sqlserver/sqlserver_va_emails_notifications_admins_enabled",
),
(
"cosmosdb_account_firewall_use_selected_networks",
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_firewall_use_selected_networks",
),
(
"cosmosdb_account_use_private_endpoints",
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_use_private_endpoints",
),
(
"cosmosdb_account_use_aad_and_rbac",
"/root_dir/prowler/providers/azure/services/cosmosdb/cosmosdb_account_use_aad_and_rbac",
),
]
returned_checks = recover_checks_from_provider(provider, service)
assert returned_checks == expected_checks

View File

@@ -0,0 +1,102 @@
from unittest import mock
from uuid import uuid4
from prowler.providers.azure.services.cosmosdb.cosmosdb_service import Account
AZURE_SUBSCRIPTION = str(uuid4())
class Test_cosmosdb_account_firewall_use_selected_networks:
def test_no_accounts(self):
cosmosdb_client = mock.MagicMock
cosmosdb_client.accounts = {}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks.cosmosdb_account_firewall_use_selected_networks.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks.cosmosdb_account_firewall_use_selected_networks import (
cosmosdb_account_firewall_use_selected_networks,
)
check = cosmosdb_account_firewall_use_selected_networks()
result = check.execute()
assert len(result) == 0
def test_accounts_no_virtual_network_filter_enabled(self):
cosmosdb_client = mock.MagicMock
account_name = "Account Name"
account_id = str(uuid4())
cosmosdb_client.accounts = {
AZURE_SUBSCRIPTION: [
Account(
id=account_id,
name=account_name,
kind=None,
location=None,
type=None,
tags=None,
disable_local_auth=None,
is_virtual_network_filter_enabled=False,
)
]
}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks.cosmosdb_account_firewall_use_selected_networks.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks.cosmosdb_account_firewall_use_selected_networks import (
cosmosdb_account_firewall_use_selected_networks,
)
check = cosmosdb_account_firewall_use_selected_networks()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"CosmosDB account {account_name} from subscription {AZURE_SUBSCRIPTION} has firewall rules that allow access from all networks."
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert result[0].resource_name == account_name
assert result[0].resource_id == account_id
def test_accounts_virtual_network_filter_enabled(self):
cosmosdb_client = mock.MagicMock
account_name = "Account Name"
account_id = str(uuid4())
cosmosdb_client.accounts = {
AZURE_SUBSCRIPTION: [
Account(
id=account_id,
name=account_name,
kind=None,
location=None,
type=None,
tags=None,
disable_local_auth=None,
is_virtual_network_filter_enabled=True,
)
]
}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks.cosmosdb_account_firewall_use_selected_networks.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_firewall_use_selected_networks.cosmosdb_account_firewall_use_selected_networks import (
cosmosdb_account_firewall_use_selected_networks,
)
check = cosmosdb_account_firewall_use_selected_networks()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"CosmosDB account {account_name} from subscription {AZURE_SUBSCRIPTION} has firewall rules that allow access only from selected networks."
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert result[0].resource_name == account_name
assert result[0].resource_id == account_id

View File

@@ -0,0 +1,104 @@
from unittest import mock
from uuid import uuid4
from prowler.providers.azure.services.cosmosdb.cosmosdb_service import Account
AZURE_SUBSCRIPTION = str(uuid4())
class Test_cosmosdb_account_use_aad_and_rbac:
def test_no_accounts(self):
cosmosdb_client = mock.MagicMock
cosmosdb_client.accounts = {}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac.cosmosdb_account_use_aad_and_rbac.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac.cosmosdb_account_use_aad_and_rbac import (
cosmosdb_account_use_aad_and_rbac,
)
check = cosmosdb_account_use_aad_and_rbac()
result = check.execute()
assert len(result) == 0
def test_accounts_disable_local_auth_false(self):
cosmosdb_client = mock.MagicMock
account_name = "Account Name"
account_id = str(uuid4())
cosmosdb_client.accounts = {
AZURE_SUBSCRIPTION: [
Account(
id=account_id,
name=account_name,
kind=None,
location=None,
type=None,
tags=None,
is_virtual_network_filter_enabled=None,
private_endpoint_connections=None,
disable_local_auth=False,
)
]
}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac.cosmosdb_account_use_aad_and_rbac.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac.cosmosdb_account_use_aad_and_rbac import (
cosmosdb_account_use_aad_and_rbac,
)
check = cosmosdb_account_use_aad_and_rbac()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"CosmosDB account {account_name} from subscription {AZURE_SUBSCRIPTION} is not using AAD and RBAC"
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert result[0].resource_name == account_name
assert result[0].resource_id == account_id
def test_accounts_disable_local_auth_true(self):
cosmosdb_client = mock.MagicMock
account_name = "Account Name"
account_id = str(uuid4())
cosmosdb_client.accounts = {
AZURE_SUBSCRIPTION: [
Account(
id=account_id,
name=account_name,
kind=None,
location=None,
type=None,
tags=None,
is_virtual_network_filter_enabled=None,
private_endpoint_connections=None,
disable_local_auth=True,
)
]
}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac.cosmosdb_account_use_aad_and_rbac.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_aad_and_rbac.cosmosdb_account_use_aad_and_rbac import (
cosmosdb_account_use_aad_and_rbac,
)
check = cosmosdb_account_use_aad_and_rbac()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"CosmosDB account {account_name} from subscription {AZURE_SUBSCRIPTION} is using AAD and RBAC"
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert result[0].resource_name == account_name
assert result[0].resource_id == account_id

View File

@@ -0,0 +1,110 @@
from unittest import mock
from uuid import uuid4
from azure.mgmt.cosmosdb.models import PrivateEndpointConnection
from prowler.providers.azure.services.cosmosdb.cosmosdb_service import Account
AZURE_SUBSCRIPTION = str(uuid4())
class Test_cosmosdb_account_use_private_endpoints:
def test_no_accounts(self):
cosmosdb_client = mock.MagicMock
cosmosdb_client.accounts = {}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints.cosmosdb_account_use_private_endpoints.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints.cosmosdb_account_use_private_endpoints import (
cosmosdb_account_use_private_endpoints,
)
check = cosmosdb_account_use_private_endpoints()
result = check.execute()
assert len(result) == 0
def test_accounts_no_private_endpoints_connections(self):
cosmosdb_client = mock.MagicMock
account_name = "Account Name"
account_id = str(uuid4())
cosmosdb_client.accounts = {
AZURE_SUBSCRIPTION: [
Account(
id=account_id,
name=account_name,
kind=None,
location=None,
type=None,
tags=None,
is_virtual_network_filter_enabled=None,
private_endpoint_connections=None,
disable_local_auth=None,
)
]
}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints.cosmosdb_account_use_private_endpoints.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints.cosmosdb_account_use_private_endpoints import (
cosmosdb_account_use_private_endpoints,
)
check = cosmosdb_account_use_private_endpoints()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"CosmosDB account {account_name} from subscription {AZURE_SUBSCRIPTION} is not using private endpoints connections"
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert result[0].resource_name == account_name
assert result[0].resource_id == account_id
def test_accounts_private_endpoints_connections(self):
cosmosdb_client = mock.MagicMock
account_name = "Account Name"
account_id = str(uuid4())
cosmosdb_client.accounts = {
AZURE_SUBSCRIPTION: [
Account(
id=account_id,
name=account_name,
kind=None,
location=None,
type=None,
tags=None,
is_virtual_network_filter_enabled=None,
private_endpoint_connections=[
PrivateEndpointConnection(
id="private_endpoint", name="private_name"
)
],
disable_local_auth=None,
)
]
}
with mock.patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints.cosmosdb_account_use_private_endpoints.cosmosdb_client",
new=cosmosdb_client,
):
from prowler.providers.azure.services.cosmosdb.cosmosdb_account_use_private_endpoints.cosmosdb_account_use_private_endpoints import (
cosmosdb_account_use_private_endpoints,
)
check = cosmosdb_account_use_private_endpoints()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"CosmosDB account {account_name} from subscription {AZURE_SUBSCRIPTION} is using private endpoints connections"
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert result[0].resource_name == account_name
assert result[0].resource_id == account_id

View File

@@ -0,0 +1,52 @@
from unittest.mock import patch
from prowler.providers.azure.services.cosmosdb.cosmosdb_service import Account, CosmosDB
from tests.providers.azure.azure_fixtures import (
AZURE_SUBSCRIPTION,
set_mocked_azure_audit_info,
)
def mock_cosmosdb_get_accounts(_):
return {
AZURE_SUBSCRIPTION: [
Account(
id="account_id",
name="account_name",
kind=None,
location=None,
type=None,
tags=None,
is_virtual_network_filter_enabled=None,
disable_local_auth=None,
)
]
}
@patch(
"prowler.providers.azure.services.cosmosdb.cosmosdb_service.CosmosDB.__get_accounts__",
new=mock_cosmosdb_get_accounts,
)
class Test_CosmosDB_Service:
def test__get_client__(self):
account = CosmosDB(set_mocked_azure_audit_info())
assert (
account.clients[AZURE_SUBSCRIPTION].__class__.__name__
== "CosmosDBManagementClient"
)
def test__get_accounts__(self):
account = CosmosDB(set_mocked_azure_audit_info())
assert account.accounts[AZURE_SUBSCRIPTION][0].__class__.__name__ == "Account"
assert account.accounts[AZURE_SUBSCRIPTION][0].id == "account_id"
assert account.accounts[AZURE_SUBSCRIPTION][0].name == "account_name"
assert account.accounts[AZURE_SUBSCRIPTION][0].kind is None
assert account.accounts[AZURE_SUBSCRIPTION][0].location is None
assert account.accounts[AZURE_SUBSCRIPTION][0].type is None
assert account.accounts[AZURE_SUBSCRIPTION][0].tags is None
assert (
account.accounts[AZURE_SUBSCRIPTION][0].is_virtual_network_filter_enabled
is None
)
assert account.accounts[AZURE_SUBSCRIPTION][0].disable_local_auth is None