Merge remote-tracking branch 'origin/master' into fix/oraclecloud-idp-group-mapping-events-10411

This commit is contained in:
Hugo P.Brito committed 2026-03-24 12:54:31 +00:00
commit f58e9cfb0f
69 files changed
+5280 -620

No files matched your search

@@ -142,6 +142,86 @@ class Test_entra_non_privileged_user_has_mfa:
assert result[0].resource_id == user_id
assert result[0].subscription == f"Tenant: {DOMAIN}"
def test_entra_disabled_user_no_privileged_no_mfa(self):
entra_client = mock.MagicMock
user_id = str(uuid4())
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(
"prowler.providers.azure.services.entra.entra_non_privileged_user_has_mfa.entra_non_privileged_user_has_mfa.entra_client",
new=entra_client,
),
):
from prowler.providers.azure.services.entra.entra_non_privileged_user_has_mfa.entra_non_privileged_user_has_mfa import (
entra_non_privileged_user_has_mfa,
)
from prowler.providers.azure.services.entra.entra_service import (
DirectoryRole,
User,
)
user = User(
id=user_id,
name="foo",
is_mfa_capable=False,
account_enabled=False,
)
entra_client.users = {DOMAIN: {f"foo@{DOMAIN}": user}}
entra_client.directory_roles = {
DOMAIN: {
"Global Administrator": DirectoryRole(id=str(uuid4()), members=[])
}
}
check = entra_non_privileged_user_has_mfa()
result = check.execute()
assert len(result) == 0
def test_entra_disabled_user_no_privileged_mfa(self):
entra_client = mock.MagicMock
user_id = str(uuid4())
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(
"prowler.providers.azure.services.entra.entra_non_privileged_user_has_mfa.entra_non_privileged_user_has_mfa.entra_client",
new=entra_client,
),
):
from prowler.providers.azure.services.entra.entra_non_privileged_user_has_mfa.entra_non_privileged_user_has_mfa import (
entra_non_privileged_user_has_mfa,
)
from prowler.providers.azure.services.entra.entra_service import (
DirectoryRole,
User,
)
user = User(
id=user_id,
name="foo",
is_mfa_capable=True,
account_enabled=False,
)
entra_client.users = {DOMAIN: {f"foo@{DOMAIN}": user}}
entra_client.directory_roles = {
DOMAIN: {
"Global Administrator": DirectoryRole(id=str(uuid4()), members=[])
}
}
check = entra_non_privileged_user_has_mfa()
result = check.execute()
assert len(result) == 0
def test_entra_user_privileged_no_mfa(self):
entra_client = mock.MagicMock
user_id = str(uuid4())
@@ -147,6 +147,7 @@ class Test_Entra_Service:
assert entra_client.users[DOMAIN]["user-1@tenant1.es"].id == "id-1"
assert entra_client.users[DOMAIN]["user-1@tenant1.es"].name == "User 1"
assert entra_client.users[DOMAIN]["user-1@tenant1.es"].is_mfa_capable is False
assert entra_client.users[DOMAIN]["user-1@tenant1.es"].account_enabled is True
def test_get_authorization_policy(self):
entra_client = Entra(set_mocked_azure_provider())
@@ -229,8 +230,8 @@ def test_azure_entra__get_users_handles_pagination():
entra_service = Entra.__new__(Entra)
users_page_one = [
SimpleNamespace(id="user-1", display_name="User 1"),
SimpleNamespace(id="user-2", display_name="User 2"),
SimpleNamespace(id="user-1", display_name="User 1", account_enabled=False),
SimpleNamespace(id="user-2", display_name="User 2", account_enabled=True),
]
users_page_two = [
SimpleNamespace(id="user-3", display_name="User 3"),
@@ -288,9 +289,18 @@ def test_azure_entra__get_users_handles_pagination():
assert len(users["tenant-1"]) == 3
assert users_builder.get.await_count == 1
request_configuration = users_builder.get.await_args.kwargs["request_configuration"]
assert request_configuration.query_parameters.select == [
"id",
"displayName",
"accountEnabled",
]
with_url_mock.assert_called_once_with("next-link")
registration_details_builder.get.assert_awaited()
registration_details_builder.with_url.assert_not_called()
assert users["tenant-1"]["user-1"].is_mfa_capable is True
assert users["tenant-1"]["user-1"].account_enabled is False
assert users["tenant-1"]["user-2"].is_mfa_capable is True
assert users["tenant-1"]["user-2"].account_enabled is True
assert users["tenant-1"]["user-3"].is_mfa_capable is False
assert users["tenant-1"]["user-3"].account_enabled is True
@@ -56,6 +56,57 @@ class Test_mysql_flexible_server_audit_log_connection_activated:
result = check.execute()
assert len(result) == 0
def test_mysql_audit_log_connection_activated_lowercase(self):
server_name = str(uuid4())
mysql_client = mock.MagicMock
mysql_client.flexible_servers = {
AZURE_SUBSCRIPTION_ID: {
"/subscriptions/resource_id": FlexibleServer(
resource_id="/subscriptions/resource_id",
name=server_name,
location="location",
version="version",
configurations={
"audit_log_events": Configuration(
resource_id=f"/subscriptions/{server_name}/configurations/audit_log_events",
description="description",
value="connection",
)
},
)
}
}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(
"prowler.providers.azure.services.mysql.mysql_flexible_server_audit_log_connection_activated.mysql_flexible_server_audit_log_connection_activated.mysql_client",
new=mysql_client,
),
):
from prowler.providers.azure.services.mysql.mysql_flexible_server_audit_log_connection_activated.mysql_flexible_server_audit_log_connection_activated import (
mysql_flexible_server_audit_log_connection_activated,
)
check = mysql_flexible_server_audit_log_connection_activated()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].subscription == AZURE_SUBSCRIPTION_ID
assert result[0].resource_name == server_name
assert result[0].location == "location"
assert (
result[0].resource_id
== f"/subscriptions/{server_name}/configurations/audit_log_events"
)
assert (
result[0].status_extended
== f"Audit log is enabled for server {server_name} in subscription {AZURE_SUBSCRIPTION_ID}."
)
def test_mysql_audit_log_connection_not_connection(self):
server_name = str(uuid4())
mysql_client = mock.MagicMock
@@ -56,6 +56,57 @@ class Test_mysql_flexible_server_audit_log_enabled:
result = check.execute()
assert len(result) == 0
def test_mysql_audit_log_enabled_lowercase(self):
server_name = str(uuid4())
mysql_client = mock.MagicMock
mysql_client.flexible_servers = {
AZURE_SUBSCRIPTION_ID: {
"/subscriptions/resource_id": FlexibleServer(
resource_id="/subscriptions/resource_id",
name=server_name,
location="location",
version="version",
configurations={
"audit_log_enabled": Configuration(
resource_id=f"/subscriptions/{server_name}/configurations/audit_log_enabled",
description="description",
value="on",
)
},
)
}
}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(
"prowler.providers.azure.services.mysql.mysql_flexible_server_audit_log_enabled.mysql_flexible_server_audit_log_enabled.mysql_client",
new=mysql_client,
),
):
from prowler.providers.azure.services.mysql.mysql_flexible_server_audit_log_enabled.mysql_flexible_server_audit_log_enabled import (
mysql_flexible_server_audit_log_enabled,
)
check = mysql_flexible_server_audit_log_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].subscription == AZURE_SUBSCRIPTION_ID
assert result[0].resource_name == server_name
assert result[0].location == "location"
assert (
result[0].resource_id
== f"/subscriptions/{server_name}/configurations/audit_log_enabled"
)
assert (
result[0].status_extended
== f"Audit log is enabled for server {server_name} in subscription {AZURE_SUBSCRIPTION_ID}."
)
def test_mysql_audit_log_disabled(self):
server_name = str(uuid4())
mysql_client = mock.MagicMock
@@ -107,6 +107,57 @@ class Test_mysql_flexible_server_ssl_connection_enabled:
== f"SSL connection is enabled for server {server_name} in subscription {AZURE_SUBSCRIPTION_ID}."
)
def test_mysql_connection_enabled_lowercase(self):
server_name = str(uuid4())
mysql_client = mock.MagicMock
mysql_client.flexible_servers = {
AZURE_SUBSCRIPTION_ID: {
"/subscriptions/resource_id": FlexibleServer(
resource_id="/subscriptions/resource_id",
name=server_name,
location="location",
version="version",
configurations={
"require_secure_transport": Configuration(
resource_id=f"/subscriptions/{server_name}/configurations/require_secure_transport",
description="description",
value="on",
)
},
)
}
}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(
"prowler.providers.azure.services.mysql.mysql_flexible_server_ssl_connection_enabled.mysql_flexible_server_ssl_connection_enabled.mysql_client",
new=mysql_client,
),
):
from prowler.providers.azure.services.mysql.mysql_flexible_server_ssl_connection_enabled.mysql_flexible_server_ssl_connection_enabled import (
mysql_flexible_server_ssl_connection_enabled,
)
check = mysql_flexible_server_ssl_connection_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].subscription == AZURE_SUBSCRIPTION_ID
assert result[0].resource_name == server_name
assert result[0].location == "location"
assert (
result[0].resource_id
== f"/subscriptions/{server_name}/configurations/require_secure_transport"
)
assert (
result[0].status_extended
== f"SSL connection is enabled for server {server_name} in subscription {AZURE_SUBSCRIPTION_ID}."
)
def test_mysql_ssl_connection_disabled(self):
server_name = str(uuid4())
mysql_client = mock.MagicMock