ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 12:31:54 +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:08:54 +02:00
committed by GitHub
parent a2362b4bbc
commit a18dd76a5a
7 changed files with 340 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/backend/api/models.py
+2 -2
View File
@@ -218,9 +218,9 @@ class Provider(RowLevelSecurityProtectedModel):
@staticmethod
def validate_m365_uid(value):
if not re.match(r"^[a-zA-Z0-9-]+\.onmicrosoft\.com$", value):
if not re.match(r"^[a-zA-Z0-9-]+\.com$", value):
raise ModelValidationError(
detail="M365 tenant ID must be a valid domain.",
detail="M365 domain ID must be a valid domain.",
code="m365-uid",
pointer="/data/attributes/uid",
)
prowler/providers/m365/lib/powershell/m365_powershell.py
+15 -10
View File
@@ -7,7 +7,7 @@ from prowler.lib.powershell.powershell import PowerShellSession
from prowler.providers.m365.exceptions.exceptions import (
M365UserNotBelongingToTenantError,
)
from prowler.providers.m365.models import M365Credentials
from prowler.providers.m365.models import M365Credentials, M365IdentityInfo
class M365PowerShell(PowerShellSession):
@@ -34,7 +34,7 @@ class M365PowerShell(PowerShellSession):
to be installed and available in the PowerShell environment.
"""
def __init__(self, credentials: M365Credentials):
def __init__(self, credentials: M365Credentials, identity: M365IdentityInfo):
"""
Initialize a Microsoft 365 PowerShell session.
@@ -46,6 +46,7 @@ class M365PowerShell(PowerShellSession):
for authentication.
"""
super().__init__()
self.tenant_identity = identity
self.init_credential(credentials)
def init_credential(self, credentials: M365Credentials) -> None:
@@ -95,12 +96,24 @@ class M365PowerShell(PowerShellSession):
'Write-Output "$($credential.GetNetworkCredential().Password)"'
)
# Validate user belongs to tenant
user_domain = credentials.user.split("@")[1]
if not any(
user_domain.endswith(domain)
for domain in self.tenant_identity.tenant_domains
):
raise M365UserNotBelongingToTenantError(
file=os.path.basename(__file__),
message=f"The user domain {user_domain} does not match any of the tenant domains: {', '.join(self.tenant_identity.tenant_domains)}",
)
app = msal.ConfidentialClientApplication(
client_id=credentials.client_id,
client_credential=credentials.client_secret,
authority=f"https://login.microsoftonline.com/{credentials.tenant_id}",
)
# Validate credentials
result = app.acquire_token_by_username_password(
username=credentials.user,
password=decrypted_password, # Needs to be in plain text
@@ -113,14 +126,6 @@ class M365PowerShell(PowerShellSession):
if "access_token" not in result:
return False
# Validate user credentials belong to tenant
user_domain = credentials.user.split("@")[1]
if not credentials.provider_id.endswith(user_domain):
raise M365UserNotBelongingToTenantError(
file=os.path.basename(__file__),
message="The provided M365 User does not belong to the specified tenant.",
)
return True
def connect_microsoft_teams(self) -> dict:
prowler/providers/m365/lib/service/service.py
+3 -1
View File
@@ -15,5 +15,7 @@ class M365Service:
# Initialize PowerShell client only if credentials are available
self.powershell = (
M365PowerShell(provider.credentials) if provider.credentials else None
M365PowerShell(provider.credentials, provider.identity)
if provider.credentials and provider.identity
else None
)
prowler/providers/m365/m365_provider.py
+90 -86
View File
@@ -194,20 +194,17 @@ class M365Provider(Provider):
# Set up the identity
self._identity = self.setup_identity(
az_cli_auth,
sp_env_auth,
env_auth,
browser_auth,
client_id,
self._session,
)
# Set up PowerShell session credentials
self._credentials = self.setup_powershell(
env_auth=env_auth,
m365_credentials=m365_credentials,
provider_id=self.identity.tenant_domain,
init_modules=init_modules,
identity=self.identity,
init_modules=init_modules,
)
# Audit Config
@@ -381,9 +378,8 @@ class M365Provider(Provider):
def setup_powershell(
env_auth: bool = False,
m365_credentials: dict = {},
provider_id: str = None,
init_modules: bool = False,
identity: M365IdentityInfo = None,
init_modules: bool = False,
) -> M365Credentials:
"""Gets the M365 credentials.
@@ -404,7 +400,7 @@ class M365Provider(Provider):
client_id=m365_credentials.get("client_id", ""),
client_secret=m365_credentials.get("client_secret", ""),
tenant_id=m365_credentials.get("tenant_id", ""),
provider_id=provider_id,
tenant_domains=identity.tenant_domains,
)
elif env_auth:
m365_user = getenv("M365_USER")
@@ -422,19 +418,18 @@ class M365Provider(Provider):
message="Missing M365_USER or M365_ENCRYPTED_PASSWORD environment variables required for credentials authentication.",
)
credentials = M365Credentials(
user=m365_user,
passwd=m365_password,
client_id=client_id,
client_secret=client_secret,
tenant_id=tenant_id,
provider_id=provider_id,
tenant_domains=identity.tenant_domains,
user=m365_user,
passwd=m365_password,
)
if credentials:
if identity:
identity.identity_type = "Service Principal and User Credentials"
identity.user = credentials.user
test_session = M365PowerShell(credentials)
test_session = M365PowerShell(credentials, identity)
try:
if test_session.test_credentials(credentials):
if init_modules:
@@ -704,7 +699,7 @@ class M365Provider(Provider):
)
# Set up the M365 session
credentials = M365Provider.setup_session(
session = M365Provider.setup_session(
az_cli_auth,
sp_env_auth,
env_auth,
@@ -714,16 +709,35 @@ class M365Provider(Provider):
region_config,
)
GraphServiceClient(credentials=credentials)
GraphServiceClient(credentials=session)
logger.info("M365 provider: Connection to MSGraph successful")
# Set up Identity
identity = M365Provider.setup_identity(
sp_env_auth,
env_auth,
session,
)
if not identity:
raise M365GetTokenIdentityError(
file=os.path.basename(__file__),
message="Failed to retrieve M365 identity",
)
if provider_id not in identity.tenant_domains:
raise M365InvalidProviderIdError(
file=os.path.basename(__file__),
message=f"The provider ID {provider_id} does not match any of the service principal tenant domains: {', '.join(identity.tenant_domains)}",
)
# Set up PowerShell credentials
if user and encrypted_password:
M365Provider.setup_powershell(
env_auth,
m365_credentials,
provider_id,
identity,
)
else:
logger.info(
@@ -732,15 +746,6 @@ class M365Provider(Provider):
logger.info("M365 provider: Connection to PowerShell successful")
# Check that user domain, provider_id and Graph client tenant_domain are the same
if user and encrypted_password:
user_domain = user.split("@")[1]
if provider_id and user_domain != provider_id:
raise M365InvalidProviderIdError(
file=os.path.basename(__file__),
message=f"Provider ID {provider_id} does not match Application tenant domain {user_domain}",
)
return Connection(is_connected=True)
# Exceptions from setup_region_config
@@ -867,13 +872,11 @@ class M365Provider(Provider):
message=f"Missing environment variable {env_var} required to authenticate.",
)
@staticmethod
def setup_identity(
self,
az_cli_auth,
sp_env_auth,
env_auth,
browser_auth,
client_id,
session,
):
"""
Sets up the identity for the M365 provider.
@@ -888,7 +891,7 @@ class M365Provider(Provider):
Returns:
M365IdentityInfo: An instance of M365IdentityInfo containing the identity information.
"""
credentials = self.session
logger.info("M365 provider: Setting up identity ...")
# TODO: fill this object with real values not default and set to none
identity = M365IdentityInfo()
@@ -896,74 +899,75 @@ class M365Provider(Provider):
# the identity can access AAD and retrieve the tenant domain name.
# With cli also should be possible but right now it does not work, m365 python package issue is coming
# At the time of writting this with az cli creds is not working, despite that is included
if env_auth or az_cli_auth or sp_env_auth or browser_auth or client_id:
async def get_m365_identity():
# Trying to recover tenant domain info
async def get_m365_identity(identity):
# Trying to recover tenant domain info
try:
logger.info(
"Trying to retrieve tenant domain from AAD to populate identity structure ..."
)
client = GraphServiceClient(credentials=session)
domain_result = await client.domains.get()
if getattr(domain_result, "value"):
if getattr(domain_result.value[0], "id"):
identity.tenant_domain = domain_result.value[0].id
for domain in domain_result.value:
identity.tenant_domains.append(domain.id)
except HttpResponseError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise M365HTTPResponseError(
file=os.path.basename(__file__),
original_exception=error,
)
except ClientAuthenticationError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise M365GetTokenIdentityError(
file=os.path.basename(__file__),
original_exception=error,
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
# since that exception is not considered as critical, we keep filling another identity fields
identity.identity_id = (
getenv("AZURE_CLIENT_ID") or "Unknown user id (Missing AAD permissions)"
)
if sp_env_auth:
identity.identity_type = "Service Principal"
elif env_auth:
identity.identity_type = "Service Principal and User Credentials"
else:
identity.identity_type = "User"
try:
logger.info(
"Trying to retrieve tenant domain from AAD to populate identity structure ..."
"Trying to retrieve user information from AAD to populate identity structure ..."
)
client = GraphServiceClient(credentials=credentials)
client = GraphServiceClient(credentials=session)
domain_result = await client.domains.get()
if getattr(domain_result, "value"):
if getattr(domain_result.value[0], "id"):
identity.tenant_domain = domain_result.value[0].id
me = await client.me.get()
if me:
if getattr(me, "user_principal_name"):
identity.identity_id = me.user_principal_name
except HttpResponseError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise M365HTTPResponseError(
file=os.path.basename(__file__),
original_exception=error,
)
except ClientAuthenticationError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise M365GetTokenIdentityError(
file=os.path.basename(__file__),
original_exception=error,
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
# since that exception is not considered as critical, we keep filling another identity fields
if sp_env_auth or env_auth or client_id:
# The id of the sp can be retrieved from environment variables
identity.identity_id = getenv("AZURE_CLIENT_ID")
identity.identity_type = "Service Principal"
# Same here, if user can access AAD, some fields are retrieved if not, default value, for az cli
# should work but it doesn't, pending issue
else:
identity.identity_id = "Unknown user id (Missing AAD permissions)"
identity.identity_type = "User"
try:
logger.info(
"Trying to retrieve user information from AAD to populate identity structure ..."
)
client = GraphServiceClient(credentials=credentials)
me = await client.me.get()
if me:
if getattr(me, "user_principal_name"):
identity.identity_id = me.user_principal_name
# Retrieve tenant id from the client
client = GraphServiceClient(credentials=session)
organization_info = await client.organization.get()
identity.tenant_id = organization_info.value[0].id
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
# Retrieve tenant id from the client
client = GraphServiceClient(credentials=credentials)
organization_info = await client.organization.get()
identity.tenant_id = organization_info.value[0].id
asyncio.get_event_loop().run_until_complete(get_m365_identity())
return identity
asyncio.get_event_loop().run_until_complete(get_m365_identity(identity))
return identity
@staticmethod
def validate_static_credentials(
prowler/providers/m365/models.py
+5 -4
View File
@@ -8,7 +8,8 @@ class M365IdentityInfo(BaseModel):
identity_id: str = ""
identity_type: str = ""
tenant_id: str = ""
tenant_domain: str = "Unknown tenant domain (missing AAD permissions)"
tenant_domain: str = "Unknown tenant domain (missing Entra permissions)"
tenant_domains: list[str] = []
location: str = ""
user: str = None
@@ -21,12 +22,12 @@ class M365RegionConfig(BaseModel):
class M365Credentials(BaseModel):
user: str = ""
passwd: str = ""
client_id: str = ""
client_secret: str = ""
tenant_id: str = ""
provider_id: str = ""
tenant_domains: list[str] = []
user: str = ""
passwd: str = ""
class M365OutputOptions(ProviderOutputOptions):
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)
)