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

chore(m365): accept all tenant domains in authentication (#7746)

Browse Source
This commit is contained in:
Hugo Pereira Brito
2025-05-19 10:47:40 +02:00
committed by César Arroba
parent 91b1feffcb
commit f6bb6efbf1
7 changed files with 340 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/providers/m365/lib/powershell/m365_powershell_test.py
+108 -124
View File
@@ -7,7 +7,7 @@ from prowler.providers.m365.exceptions.exceptions import (
M365UserNotBelongingToTenantError,
)
from prowler.providers.m365.lib.powershell.m365_powershell import M365PowerShell
from prowler.providers.m365.models import M365Credentials
from prowler.providers.m365.models import M365Credentials, M365IdentityInfo
class Testm365PowerShell:
@@ -16,9 +16,17 @@ class Testm365PowerShell:
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="example.com",
tenant_domains=["example.com"],
location="test_location",
)
with patch.object(M365PowerShell, "init_credential") as mock_init_credential:
session = M365PowerShell(credentials)
session = M365PowerShell(credentials, identity)
mock_popen.assert_called_once()
mock_init_credential.assert_called_once_with(credentials)
@@ -29,7 +37,15 @@ class Testm365PowerShell:
@patch("subprocess.Popen")
def test_sanitize(self, _):
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="example.com",
tenant_domains=["example.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
test_cases = [
("test@example.com", "test@example.com"),
@@ -63,7 +79,15 @@ class Testm365PowerShell:
client_secret="test_client_secret",
tenant_id="test_tenant_id",
)
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="example.com",
tenant_domains=["example.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
session.execute = MagicMock()
@@ -95,9 +119,16 @@ class Testm365PowerShell:
client_id="test_client_id",
client_secret="test_client_secret",
tenant_id="test_tenant_id",
provider_id="contoso.onmicrosoft.com",
)
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="contoso.onmicrosoft.com",
tenant_domains=["contoso.onmicrosoft.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
# Mock read_output to return the decrypted password
session.read_output = MagicMock(return_value="decrypted_password")
@@ -150,9 +181,16 @@ class Testm365PowerShell:
client_id="test_client_id",
client_secret="test_client_secret",
tenant_id="test_tenant_id",
provider_id="contoso.onmicrosoft.com",
)
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="contoso.onmicrosoft.com",
tenant_domains=["contoso.onmicrosoft.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
# Mock the execute method to return the decrypted password
def mock_execute(command, *args, **kwargs):
@@ -168,20 +206,15 @@ class Testm365PowerShell:
session.test_credentials(credentials)
assert exception.type == M365UserNotBelongingToTenantError
assert "The provided M365 User does not belong to the specified tenant." in str(
exception.value
assert (
"The user domain otherdomain.com does not match any of the tenant domains: contoso.onmicrosoft.com"
in str(exception.value)
)
mock_msal.assert_called_once_with(
client_id="test_client_id",
client_credential="test_client_secret",
authority="https://login.microsoftonline.com/test_tenant_id",
)
mock_msal_instance.acquire_token_by_username_password.assert_called_once_with(
username="user@otherdomain.com",
password="decrypted_password",
scopes=["https://graph.microsoft.com/.default"],
)
# Verify MSAL was not called since domain validation failed first
mock_msal.assert_not_called()
mock_msal_instance.acquire_token_by_username_password.assert_not_called()
session.close()
@patch("subprocess.Popen")
@@ -199,9 +232,16 @@ class Testm365PowerShell:
client_id="test_client_id",
client_secret="test_client_secret",
tenant_id="test_tenant_id",
provider_id="contoso.onmicrosoft.com",
)
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="contoso.onmicrosoft.com",
tenant_domains=["contoso.onmicrosoft.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
# Mock the execute method to return the decrypted password
def mock_execute(command, *args, **kwargs):
@@ -231,7 +271,15 @@ class Testm365PowerShell:
@patch("subprocess.Popen")
def test_remove_ansi(self, mock_popen):
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="example.com",
tenant_domains=["example.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
test_cases = [
("\x1b[32mSuccess\x1b[0m", "Success"),
@@ -250,7 +298,15 @@ class Testm365PowerShell:
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="example.com",
tenant_domains=["example.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
command = "Get-Command"
expected_output = {"Name": "Get-Command"}
@@ -261,18 +317,19 @@ class Testm365PowerShell:
@patch("subprocess.Popen")
def test_read_output(self, mock_popen):
"""Test the read_output method with various scenarios:
- Normal stdout output
- Error in stderr
- Timeout in stdout
- Empty output
- Empty queue handling
"""
# Setup
"""Test the read_output method with various scenarios"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="example.com",
tenant_domains=["example.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
# Test 1: Normal stdout output
mock_process.stdout.readline.side_effect = [
@@ -304,95 +361,6 @@ class Testm365PowerShell:
result = session.read_output(timeout=0.1, default="timeout")
assert result == "timeout"
# Test 4: Empty output
mock_process.stdout.readline.side_effect = [f"{session.END}\n"]
mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n"
result = session.read_output()
assert result == ""
# Test 5: Empty queue handling
mock_process.stdout.readline.side_effect = [] # No output at all
mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n"
result = session.read_output(timeout=0.1, default="empty_queue")
assert result == "empty_queue"
# Test 6: Empty error queue handling
mock_process.stdout.readline.side_effect = ["test output\n", f"{session.END}\n"]
mock_process.stderr.readline.side_effect = [] # No error output
with patch("prowler.lib.logger.logger.error") as mock_error:
result = session.read_output()
assert result == "test output"
mock_error.assert_not_called()
# Test 7: Both queues empty
mock_process.stdout.readline.side_effect = [] # No output
mock_process.stderr.readline.side_effect = [] # No error output
result = session.read_output(timeout=0.1, default="both_empty")
assert result == "both_empty"
session.close()
@patch("subprocess.Popen")
def test_read_output_queue_empty(self, mock_popen):
"""Test read_output when both queues are empty"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
# Mock process to return empty queues
mock_process.stdout.readline.side_effect = [] # No output
mock_process.stderr.readline.side_effect = [] # No error output
# Test with default value
result = session.read_output(timeout=0.1, default="empty_queue")
assert result == "empty_queue"
# Test without default value
result = session.read_output(timeout=0.1)
assert result == ""
session.close()
@patch("subprocess.Popen")
def test_read_output_error_queue_empty(self, mock_popen):
"""Test read_output when error queue is empty but stdout has content"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
# Mock process to return content in stdout but empty stderr
mock_process.stdout.readline.side_effect = ["test output\n", f"{session.END}\n"]
mock_process.stderr.readline.side_effect = [] # No error output
with patch("prowler.lib.logger.logger.error") as mock_error:
result = session.read_output()
assert result == "test output"
mock_error.assert_not_called()
session.close()
@patch("subprocess.Popen")
def test_read_output_result_queue_empty(self, mock_popen):
"""Test read_output when result queue is empty but stderr has content"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
# Mock process to return empty stdout but content in stderr
mock_process.stdout.readline.side_effect = [] # No output
mock_process.stderr.readline.side_effect = [
"Error message\n",
f"Write-Error: {session.END}\n",
]
with patch("prowler.lib.logger.logger.error") as mock_error:
result = session.read_output(timeout=0.1, default="default")
assert result == "default"
mock_error.assert_called_once_with("PowerShell error output: Error message")
session.close()
@patch("subprocess.Popen")
@@ -400,7 +368,15 @@ class Testm365PowerShell:
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="example.com",
tenant_domains=["example.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
test_cases = [
('{"key": "value"}', {"key": "value"}),
@@ -425,7 +401,15 @@ class Testm365PowerShell:
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
identity = M365IdentityInfo(
identity_id="test_id",
identity_type="User",
tenant_id="test_tenant",
tenant_domain="example.com",
tenant_domains=["example.com"],
location="test_location",
)
session = M365PowerShell(credentials, identity)
session.close()
tests/providers/m365/m365_provider_test.py
+117 -7
View File
@@ -286,6 +286,17 @@ class TestM365Provider:
patch(
"prowler.providers.m365.m365_provider.GraphServiceClient"
) as mock_graph_client,
patch(
"prowler.providers.m365.m365_provider.M365Provider.setup_identity",
return_value=M365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
tenant_domains=["test.onmicrosoft.com"],
location=LOCATION,
),
),
):
# Mock the return value of DefaultAzureCredential
mock_credentials = MagicMock()
@@ -298,7 +309,7 @@ class TestM365Provider:
mock_session = MagicMock()
mock_setup_session.return_value = mock_session
# Mock GraphServiceClient to avoid real API calls
# Mock GraphServiceClient
mock_client = MagicMock()
mock_graph_client.return_value = mock_client
@@ -307,6 +318,7 @@ class TestM365Provider:
tenant_id=str(uuid4()),
region="M365Global",
raise_on_exception=False,
provider_id="test.onmicrosoft.com",
)
assert isinstance(test_connection, Connection)
@@ -321,6 +333,17 @@ class TestM365Provider:
patch(
"prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials"
) as mock_validate_static_credentials,
patch(
"prowler.providers.m365.m365_provider.M365Provider.setup_identity",
return_value=M365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
tenant_domains=["test.onmicrosoft.com"],
location=LOCATION,
),
),
):
# Mock setup_session to return a mocked session object
mock_session = MagicMock()
@@ -335,6 +358,7 @@ class TestM365Provider:
raise_on_exception=False,
client_id=str(uuid4()),
client_secret=str(uuid4()),
provider_id="test.onmicrosoft.com",
)
assert isinstance(test_connection, Connection)
@@ -458,9 +482,15 @@ class TestM365Provider:
result = M365Provider.setup_powershell(
env_auth=False,
m365_credentials=credentials_dict,
provider_id="test_provider_id",
identity=M365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
tenant_domains=["test.onmicrosoft.com"],
location=LOCATION,
),
)
assert result.user == credentials_dict["user"]
assert result.passwd == credentials_dict["encrypted_password"]
@@ -595,6 +625,17 @@ class TestM365Provider:
patch(
"prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials"
) as mock_validate_static_credentials,
patch(
"prowler.providers.m365.m365_provider.M365Provider.setup_identity",
return_value=M365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain="contoso.com",
tenant_domains=["contoso.com"],
location=LOCATION,
),
),
):
# Mock setup_session to return a mocked session object
mock_session = MagicMock()
@@ -620,7 +661,7 @@ class TestM365Provider:
assert exception.type == M365InvalidProviderIdError
assert (
f"Provider ID {provider_id} does not match Application tenant domain {user_domain}"
f"The provider ID {provider_id} does not match any of the service principal tenant domains: {user_domain}"
in str(exception.value)
)
@@ -646,7 +687,14 @@ class TestM365Provider:
M365Provider.setup_powershell(
env_auth=False,
m365_credentials=credentials_dict,
provider_id="test_provider_id",
identity=M365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
tenant_domains=["test.onmicrosoft.com"],
location=LOCATION,
),
init_modules=False,
)
mock_init_modules.assert_not_called()
@@ -673,7 +721,14 @@ class TestM365Provider:
M365Provider.setup_powershell(
env_auth=False,
m365_credentials=credentials_dict,
provider_id="test_provider_id",
identity=M365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
tenant_domains=["test.onmicrosoft.com"],
location=LOCATION,
),
init_modules=True,
)
mock_init_modules.assert_called_once()
@@ -702,8 +757,63 @@ class TestM365Provider:
M365Provider.setup_powershell(
env_auth=False,
m365_credentials=credentials_dict,
provider_id="test_provider_id",
identity=M365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
tenant_domains=["test.onmicrosoft.com"],
location=LOCATION,
),
init_modules=True,
)
assert str(exc_info.value) == "Module initialization failed"
def test_test_connection_provider_id_not_in_tenant_domains(self):
"""Test that an exception is raised when provider_id is not in tenant_domains"""
with (
patch(
"prowler.providers.m365.m365_provider.M365Provider.setup_session"
) as mock_setup_session,
patch(
"prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials"
) as mock_validate_static_credentials,
patch(
"prowler.providers.m365.m365_provider.M365Provider.setup_identity",
return_value=M365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain="contoso.onmicrosoft.com",
tenant_domains=["contoso.onmicrosoft.com", "contoso.com"],
location=LOCATION,
),
),
):
# Mock setup_session to return a mocked session object
mock_session = MagicMock()
mock_setup_session.return_value = mock_session
# Mock ValidateStaticCredentials to avoid real API calls
mock_validate_static_credentials.return_value = None
provider_id = "test.onmicrosoft.com"
with pytest.raises(M365InvalidProviderIdError) as exception:
M365Provider.test_connection(
tenant_id=str(uuid4()),
region="M365Global",
raise_on_exception=True,
client_id=str(uuid4()),
client_secret=str(uuid4()),
user="user@contoso.onmicrosoft.com",
encrypted_password="test_password",
provider_id=provider_id,
)
assert exception.type == M365InvalidProviderIdError
assert (
f"The provider ID {provider_id} does not match any of the service principal tenant domains: contoso.onmicrosoft.com, contoso.com"
in str(exception.value)
)