mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 20:42:02 +00:00
chore(m365): add test_connection function (#7541)
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>
This commit is contained in:
committed by
Pepe Fagoaga
parent
97dac23d39
commit
de36bddccf
@@ -94,7 +94,7 @@ class M365BaseException(ProwlerException):
|
||||
"message": "Tenant Id is required for Microsoft 365 static credentials. Make sure you are using the correct credentials.",
|
||||
"remediation": "Check the Microsoft 365 Tenant ID and ensure it is properly set up.",
|
||||
},
|
||||
(6022, "M365MissingEnvironmentUserCredentialsError"): {
|
||||
(6022, "M365MissingEnvironmentCredentialsError"): {
|
||||
"message": "User and Password environment variables are needed to use Credentials authentication method.",
|
||||
"remediation": "Ensure your environment variables are properly set up.",
|
||||
},
|
||||
@@ -102,6 +102,18 @@ class M365BaseException(ProwlerException):
|
||||
"message": "User or Password environment variables are not correct.",
|
||||
"remediation": "Ensure you are using the right credentials.",
|
||||
},
|
||||
(6024, "M365NotValidUserError"): {
|
||||
"message": "The provided M365 User is not valid.",
|
||||
"remediation": "Check the M365 User and ensure it is a valid user.",
|
||||
},
|
||||
(6025, "M365NotValidEncryptedPasswordError"): {
|
||||
"message": "The provided M365 Encrypted Password is not valid.",
|
||||
"remediation": "Check the M365 Encrypted Password and ensure it is a valid password.",
|
||||
},
|
||||
(6026, "M365UserNotBelongingToTenantError"): {
|
||||
"message": "The provided M365 User does not belong to the specified tenant.",
|
||||
"remediation": "Check the M365 User email domain and ensure it belongs to the specified tenant.",
|
||||
},
|
||||
}
|
||||
|
||||
def __init__(self, code, file=None, original_exception=None, message=None):
|
||||
@@ -279,7 +291,7 @@ class M365NotTenantIdButClientIdAndClientSecretError(M365CredentialsError):
|
||||
)
|
||||
|
||||
|
||||
class M365MissingEnvironmentUserCredentialsError(M365CredentialsError):
|
||||
class M365MissingEnvironmentCredentialsError(M365CredentialsError):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
6022, file=file, original_exception=original_exception, message=message
|
||||
@@ -291,3 +303,24 @@ class M365EnvironmentUserCredentialsError(M365CredentialsError):
|
||||
super().__init__(
|
||||
6023, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class M365NotValidUserError(M365CredentialsError):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
6024, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class M365NotValidEncryptedPasswordError(M365CredentialsError):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
6025, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
|
||||
class M365UserNotBelongingToTenantError(M365CredentialsError):
|
||||
def __init__(self, file=None, original_exception=None, message=None):
|
||||
super().__init__(
|
||||
6026, file=file, original_exception=original_exception, message=message
|
||||
)
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
import os
|
||||
|
||||
import msal
|
||||
|
||||
from prowler.lib.powershell.powershell import PowerShellSession
|
||||
from prowler.providers.m365.exceptions.exceptions import (
|
||||
M365UserNotBelongingToTenantError,
|
||||
)
|
||||
from prowler.providers.m365.models import M365Credentials
|
||||
|
||||
|
||||
@@ -102,7 +107,21 @@ class M365PowerShell(PowerShellSession):
|
||||
scopes=["https://graph.microsoft.com/.default"],
|
||||
)
|
||||
|
||||
return "access_token" in result
|
||||
if result is None:
|
||||
return False
|
||||
|
||||
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:
|
||||
"""
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import asyncio
|
||||
import os
|
||||
import re
|
||||
from argparse import ArgumentTypeError
|
||||
from os import getenv
|
||||
from uuid import UUID
|
||||
@@ -40,12 +39,14 @@ from prowler.providers.m365.exceptions.exceptions import (
|
||||
M365HTTPResponseError,
|
||||
M365InteractiveBrowserCredentialError,
|
||||
M365InvalidProviderIdError,
|
||||
M365MissingEnvironmentUserCredentialsError,
|
||||
M365MissingEnvironmentCredentialsError,
|
||||
M365NoAuthenticationMethodError,
|
||||
M365NotTenantIdButClientIdAndClientSecretError,
|
||||
M365NotValidClientIdError,
|
||||
M365NotValidClientSecretError,
|
||||
M365NotValidEncryptedPasswordError,
|
||||
M365NotValidTenantIdError,
|
||||
M365NotValidUserError,
|
||||
M365SetUpRegionConfigError,
|
||||
M365SetUpSessionError,
|
||||
M365TenantIdAndClientIdNotBelongingToClientSecretError,
|
||||
@@ -105,10 +106,10 @@ class M365Provider(Provider):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
sp_env_auth: bool,
|
||||
env_auth: bool,
|
||||
az_cli_auth: bool,
|
||||
browser_auth: bool,
|
||||
sp_env_auth: bool = False,
|
||||
env_auth: bool = False,
|
||||
az_cli_auth: bool = False,
|
||||
browser_auth: bool = False,
|
||||
tenant_id: str = None,
|
||||
client_id: str = None,
|
||||
client_secret: str = None,
|
||||
@@ -187,9 +188,6 @@ class M365Provider(Provider):
|
||||
self._region_config,
|
||||
)
|
||||
|
||||
# Set up PowerShell session credentials
|
||||
self._credentials = self.setup_powershell(env_auth, m365_credentials)
|
||||
|
||||
# Set up the identity
|
||||
self._identity = self.setup_identity(
|
||||
az_cli_auth,
|
||||
@@ -199,6 +197,13 @@ class M365Provider(Provider):
|
||||
client_id,
|
||||
)
|
||||
|
||||
# Set up PowerShell session credentials
|
||||
self._credentials = self.setup_powershell(
|
||||
env_auth=env_auth,
|
||||
m365_credentials=m365_credentials,
|
||||
provider_id=self.identity.tenant_domain,
|
||||
)
|
||||
|
||||
# Audit Config
|
||||
if config_content:
|
||||
self._audit_config = config_content
|
||||
@@ -294,8 +299,8 @@ class M365Provider(Provider):
|
||||
M365BrowserAuthNoTenantIDError: If browser authentication is enabled but the tenant ID is not found.
|
||||
"""
|
||||
|
||||
if not client_id and not client_secret and not user and not encrypted_password:
|
||||
if not browser_auth and tenant_id:
|
||||
if not client_id and not client_secret:
|
||||
if not browser_auth and tenant_id and not env_auth:
|
||||
raise M365BrowserAuthNoFlagError(
|
||||
file=os.path.basename(__file__),
|
||||
message="M365 tenant ID error: browser authentication flag (--browser-auth) not found",
|
||||
@@ -315,6 +320,12 @@ class M365Provider(Provider):
|
||||
file=os.path.basename(__file__),
|
||||
message="M365 Tenant ID (--tenant-id) is required for browser authentication mode",
|
||||
)
|
||||
elif env_auth:
|
||||
if not user or not encrypted_password or not tenant_id:
|
||||
raise M365MissingEnvironmentCredentialsError(
|
||||
file=os.path.basename(__file__),
|
||||
message="M365 provider requires AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID, M365_USER and M365_ENCRYPTED_PASSWORD environment variables to be set when using --env-auth",
|
||||
)
|
||||
else:
|
||||
if not tenant_id:
|
||||
raise M365NotTenantIdButClientIdAndClientSecretError(
|
||||
@@ -362,7 +373,7 @@ class M365Provider(Provider):
|
||||
|
||||
@staticmethod
|
||||
def setup_powershell(
|
||||
env_auth: bool = False, m365_credentials: dict = {}
|
||||
env_auth: bool = False, m365_credentials: dict = {}, provider_id: str = None
|
||||
) -> M365Credentials:
|
||||
"""Gets the M365 credentials.
|
||||
|
||||
@@ -382,6 +393,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,
|
||||
)
|
||||
elif env_auth:
|
||||
m365_user = getenv("M365_USER")
|
||||
@@ -394,7 +406,7 @@ class M365Provider(Provider):
|
||||
logger.critical(
|
||||
"M365 provider: Missing M365_USER or M365_ENCRYPTED_PASSWORD environment variables needed for credentials authentication"
|
||||
)
|
||||
raise M365MissingEnvironmentUserCredentialsError(
|
||||
raise M365MissingEnvironmentCredentialsError(
|
||||
file=os.path.basename(__file__),
|
||||
message="Missing M365_USER or M365_ENCRYPTED_PASSWORD environment variables required for credentials authentication.",
|
||||
)
|
||||
@@ -404,6 +416,7 @@ class M365Provider(Provider):
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
tenant_id=tenant_id,
|
||||
provider_id=provider_id,
|
||||
)
|
||||
|
||||
if credentials:
|
||||
@@ -466,6 +479,9 @@ class M365Provider(Provider):
|
||||
- tenant_id: The M365 Active Directory tenant ID.
|
||||
- client_id: The M365 client ID.
|
||||
- client_secret: The M365 client secret
|
||||
- user: The M365 user email
|
||||
- encrypted_password: The M365 encrypted password
|
||||
- provider_id: The M365 provider ID (in this case the Tenant ID).
|
||||
region_config (M365RegionConfig): The region configuration object.
|
||||
|
||||
Returns:
|
||||
@@ -592,10 +608,11 @@ class M365Provider(Provider):
|
||||
client_secret=None,
|
||||
user=None,
|
||||
encrypted_password=None,
|
||||
provider_id=None,
|
||||
) -> Connection:
|
||||
"""Test connection to M365 subscription.
|
||||
"""Test connection to M365 tenant and PowerShell modules.
|
||||
|
||||
Test the connection to an M365 subscription using the provided credentials.
|
||||
Test the connection to an M365 tenant and PowerShell modules using the provided credentials.
|
||||
|
||||
Args:
|
||||
|
||||
@@ -610,6 +627,7 @@ class M365Provider(Provider):
|
||||
client_secret (str): The M365 client secret.
|
||||
user (str): The M365 user email.
|
||||
encrypted_password (str): The M365 encrypted_password.
|
||||
provider_id (str): The M365 provider ID (in this case the Tenant ID).
|
||||
|
||||
|
||||
Returns:
|
||||
@@ -622,7 +640,7 @@ class M365Provider(Provider):
|
||||
M365InteractiveBrowserCredentialError: If there is an error in retrieving the M365 credentials using browser authentication.
|
||||
M365HTTPResponseError: If there is an HTTP response error.
|
||||
M365ConfigCredentialsError: If there is an error in configuring the M365 credentials from a dictionary.
|
||||
|
||||
M365InvalidProviderIdError: If the provider ID does not match the application tenant domain.
|
||||
|
||||
Examples:
|
||||
>>> M365Provider.test_connection(az_cli_auth=True)
|
||||
@@ -649,11 +667,22 @@ class M365Provider(Provider):
|
||||
# Get the dict from the static credentials
|
||||
m365_credentials = None
|
||||
if tenant_id and client_id and client_secret:
|
||||
m365_credentials = M365Provider.validate_static_credentials(
|
||||
tenant_id=tenant_id,
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
)
|
||||
if not user and not encrypted_password:
|
||||
m365_credentials = M365Provider.validate_static_credentials(
|
||||
tenant_id=tenant_id,
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
user="user",
|
||||
encrypted_password="encrypted_password",
|
||||
)
|
||||
else:
|
||||
m365_credentials = M365Provider.validate_static_credentials(
|
||||
tenant_id=tenant_id,
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
user=user,
|
||||
encrypted_password=encrypted_password,
|
||||
)
|
||||
|
||||
# Set up the M365 session
|
||||
credentials = M365Provider.setup_session(
|
||||
@@ -668,7 +697,30 @@ class M365Provider(Provider):
|
||||
|
||||
GraphServiceClient(credentials=credentials)
|
||||
|
||||
logger.info("M365 provider: Connection to M365 successful")
|
||||
logger.info("M365 provider: Connection to MSGraph successful")
|
||||
|
||||
# Set up PowerShell credentials
|
||||
if user and encrypted_password:
|
||||
M365Provider.setup_powershell(
|
||||
env_auth,
|
||||
m365_credentials,
|
||||
provider_id,
|
||||
)
|
||||
else:
|
||||
logger.info(
|
||||
"M365 provider: Connection to PowerShell has not been requested"
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
@@ -896,7 +948,11 @@ class M365Provider(Provider):
|
||||
|
||||
@staticmethod
|
||||
def validate_static_credentials(
|
||||
tenant_id: str = None, client_id: str = None, client_secret: str = None
|
||||
tenant_id: str = None,
|
||||
client_id: str = None,
|
||||
client_secret: str = None,
|
||||
user: str = None,
|
||||
encrypted_password: str = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Validates the static credentials for the M365 provider.
|
||||
@@ -905,6 +961,8 @@ class M365Provider(Provider):
|
||||
tenant_id (str): The M365 Active Directory tenant ID.
|
||||
client_id (str): The M365 client ID.
|
||||
client_secret (str): The M365 client secret.
|
||||
user (str): The M365 user email.
|
||||
encrypted_password (str): The M365 encrypted password.
|
||||
|
||||
Raises:
|
||||
M365NotValidTenantIdError: If the provided M365 Tenant ID is not valid.
|
||||
@@ -934,19 +992,36 @@ class M365Provider(Provider):
|
||||
file=os.path.basename(__file__),
|
||||
message="The provided M365 Client ID is not valid.",
|
||||
)
|
||||
|
||||
# Validate the Client Secret
|
||||
if not re.match("^[a-zA-Z0-9._~-]+$", client_secret):
|
||||
if not client_secret:
|
||||
raise M365NotValidClientSecretError(
|
||||
file=os.path.basename(__file__),
|
||||
message="The provided M365 Client Secret is not valid.",
|
||||
)
|
||||
|
||||
# Validate the User
|
||||
if not user:
|
||||
raise M365NotValidUserError(
|
||||
file=os.path.basename(__file__),
|
||||
message="The provided M365 User is not valid.",
|
||||
)
|
||||
|
||||
# Validate the Encrypted Password
|
||||
if not encrypted_password:
|
||||
raise M365NotValidEncryptedPasswordError(
|
||||
file=os.path.basename(__file__),
|
||||
message="The provided M365 Encrypted Password is not valid.",
|
||||
)
|
||||
|
||||
try:
|
||||
M365Provider.verify_client(tenant_id, client_id, client_secret)
|
||||
return {
|
||||
"tenant_id": tenant_id,
|
||||
"client_id": client_id,
|
||||
"client_secret": client_secret,
|
||||
"user": user,
|
||||
"encrypted_password": encrypted_password,
|
||||
}
|
||||
except M365NotValidTenantIdError as tenant_id_error:
|
||||
logger.error(
|
||||
|
||||
@@ -25,6 +25,7 @@ class M365Credentials(BaseModel):
|
||||
client_id: str = ""
|
||||
client_secret: str = ""
|
||||
tenant_id: str = ""
|
||||
provider_id: str = ""
|
||||
|
||||
|
||||
class M365OutputOptions(ProviderOutputOptions):
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
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
|
||||
|
||||
@@ -84,11 +89,12 @@ class Testm365PowerShell:
|
||||
}
|
||||
|
||||
credentials = M365Credentials(
|
||||
user="test@example.com",
|
||||
user="test@contoso.onmicrosoft.com",
|
||||
passwd="test_password",
|
||||
client_id="test_client_id",
|
||||
client_secret="test_client_secret",
|
||||
tenant_id="test_tenant_id",
|
||||
provider_id="contoso.onmicrosoft.com",
|
||||
)
|
||||
session = M365PowerShell(credentials)
|
||||
|
||||
@@ -115,7 +121,89 @@ class Testm365PowerShell:
|
||||
authority="https://login.microsoftonline.com/test_tenant_id",
|
||||
)
|
||||
mock_msal_instance.acquire_token_by_username_password.assert_called_once_with(
|
||||
username="test@example.com",
|
||||
username="test@contoso.onmicrosoft.com",
|
||||
password="decrypted_password",
|
||||
scopes=["https://graph.microsoft.com/.default"],
|
||||
)
|
||||
session.close()
|
||||
|
||||
@patch("subprocess.Popen")
|
||||
@patch("msal.ConfidentialClientApplication")
|
||||
def test_test_credentials_user_not_belonging_to_tenant(self, mock_msal, mock_popen):
|
||||
mock_process = MagicMock()
|
||||
mock_popen.return_value = mock_process
|
||||
mock_msal_instance = MagicMock()
|
||||
mock_msal.return_value = mock_msal_instance
|
||||
mock_msal_instance.acquire_token_by_username_password.return_value = {
|
||||
"access_token": "test_token"
|
||||
}
|
||||
|
||||
credentials = M365Credentials(
|
||||
user="user@otherdomain.com",
|
||||
passwd="test_password",
|
||||
client_id="test_client_id",
|
||||
client_secret="test_client_secret",
|
||||
tenant_id="test_tenant_id",
|
||||
provider_id="contoso.onmicrosoft.com",
|
||||
)
|
||||
session = M365PowerShell(credentials)
|
||||
|
||||
session.execute = MagicMock()
|
||||
session.process.stdin.write = MagicMock()
|
||||
session.read_output = MagicMock(return_value="decrypted_password")
|
||||
|
||||
with pytest.raises(M365UserNotBelongingToTenantError) as exception:
|
||||
session.test_credentials(credentials)
|
||||
|
||||
assert exception.type == M365UserNotBelongingToTenantError
|
||||
assert "The provided M365 User does not belong to the specified tenant." 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"],
|
||||
)
|
||||
session.close()
|
||||
|
||||
@patch("subprocess.Popen")
|
||||
@patch("msal.ConfidentialClientApplication")
|
||||
def test_test_credentials_auth_failure(self, mock_msal, mock_popen):
|
||||
mock_process = MagicMock()
|
||||
mock_popen.return_value = mock_process
|
||||
mock_msal_instance = MagicMock()
|
||||
mock_msal.return_value = mock_msal_instance
|
||||
mock_msal_instance.acquire_token_by_username_password.return_value = None
|
||||
|
||||
credentials = M365Credentials(
|
||||
user="test@contoso.onmicrosoft.com",
|
||||
passwd="test_password",
|
||||
client_id="test_client_id",
|
||||
client_secret="test_client_secret",
|
||||
tenant_id="test_tenant_id",
|
||||
provider_id="contoso.onmicrosoft.com",
|
||||
)
|
||||
session = M365PowerShell(credentials)
|
||||
|
||||
session.execute = MagicMock()
|
||||
session.process.stdin.write = MagicMock()
|
||||
session.read_output = MagicMock(return_value="decrypted_password")
|
||||
|
||||
assert session.test_credentials(credentials) is False
|
||||
|
||||
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="test@contoso.onmicrosoft.com",
|
||||
password="decrypted_password",
|
||||
scopes=["https://graph.microsoft.com/.default"],
|
||||
)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
from uuid import uuid4
|
||||
|
||||
@@ -18,8 +19,15 @@ from prowler.config.config import (
|
||||
from prowler.providers.common.models import Connection
|
||||
from prowler.providers.m365.exceptions.exceptions import (
|
||||
M365HTTPResponseError,
|
||||
M365MissingEnvironmentUserCredentialsError,
|
||||
M365InvalidProviderIdError,
|
||||
M365MissingEnvironmentCredentialsError,
|
||||
M365NoAuthenticationMethodError,
|
||||
M365NotValidClientIdError,
|
||||
M365NotValidClientSecretError,
|
||||
M365NotValidEncryptedPasswordError,
|
||||
M365NotValidTenantIdError,
|
||||
M365NotValidUserError,
|
||||
M365UserNotBelongingToTenantError,
|
||||
)
|
||||
from prowler.providers.m365.m365_provider import M365Provider
|
||||
from prowler.providers.m365.models import (
|
||||
@@ -333,37 +341,59 @@ class TestM365Provider:
|
||||
assert test_connection.is_connected
|
||||
assert test_connection.error is None
|
||||
|
||||
def test_test_connection_tenant_id_client_id_client_secret_user_encrypted_password(
|
||||
def test_test_connection_tenant_id_client_id_client_secret_no_user_encrypted_password(
|
||||
self,
|
||||
):
|
||||
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,
|
||||
):
|
||||
# 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
|
||||
|
||||
test_connection = M365Provider.test_connection(
|
||||
tenant_id=str(uuid4()),
|
||||
region="M365Global",
|
||||
raise_on_exception=False,
|
||||
client_id=str(uuid4()),
|
||||
client_secret=str(uuid4()),
|
||||
user="user@user.com",
|
||||
encrypted_password="AAAA1111",
|
||||
with patch(
|
||||
"prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials"
|
||||
) as mock_validate_static_credentials:
|
||||
mock_validate_static_credentials.side_effect = M365NotValidUserError(
|
||||
file=os.path.basename(__file__),
|
||||
message="The provided M365 User is not valid.",
|
||||
)
|
||||
|
||||
assert isinstance(test_connection, Connection)
|
||||
assert test_connection.is_connected
|
||||
assert test_connection.error is None
|
||||
with pytest.raises(M365NotValidUserError) as exception:
|
||||
M365Provider.test_connection(
|
||||
tenant_id=str(uuid4()),
|
||||
region="M365Global",
|
||||
raise_on_exception=True,
|
||||
client_id=str(uuid4()),
|
||||
client_secret=str(uuid4()),
|
||||
user=None,
|
||||
encrypted_password="test_password",
|
||||
)
|
||||
|
||||
assert exception.type == M365NotValidUserError
|
||||
assert "The provided M365 User is not valid." in str(exception.value)
|
||||
|
||||
def test_test_connection_tenant_id_client_id_client_secret_user_no_encrypted_password(
|
||||
self,
|
||||
):
|
||||
with patch(
|
||||
"prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials"
|
||||
) as mock_validate_static_credentials:
|
||||
mock_validate_static_credentials.side_effect = (
|
||||
M365NotValidEncryptedPasswordError(
|
||||
file=os.path.basename(__file__),
|
||||
message="The provided M365 Encrypted Password is not valid.",
|
||||
)
|
||||
)
|
||||
|
||||
with pytest.raises(M365NotValidEncryptedPasswordError) as exception:
|
||||
M365Provider.test_connection(
|
||||
tenant_id=str(uuid4()),
|
||||
region="M365Global",
|
||||
raise_on_exception=True,
|
||||
client_id=str(uuid4()),
|
||||
client_secret=str(uuid4()),
|
||||
user="test@example.com",
|
||||
encrypted_password=None,
|
||||
)
|
||||
|
||||
assert exception.type == M365NotValidEncryptedPasswordError
|
||||
assert "The provided M365 Encrypted Password is not valid." in str(
|
||||
exception.value
|
||||
)
|
||||
|
||||
def test_test_connection_with_httpresponseerror(self):
|
||||
with patch(
|
||||
@@ -424,7 +454,9 @@ class TestM365Provider:
|
||||
return_value=True,
|
||||
):
|
||||
result = M365Provider.setup_powershell(
|
||||
env_auth=False, m365_credentials=credentials_dict
|
||||
env_auth=False,
|
||||
m365_credentials=credentials_dict,
|
||||
provider_id="test_provider_id",
|
||||
)
|
||||
|
||||
assert result.user == credentials_dict["user"]
|
||||
@@ -440,7 +472,7 @@ class TestM365Provider:
|
||||
mock_session.test_credentials.return_value = False
|
||||
mock_powershell.return_value = mock_session
|
||||
|
||||
with pytest.raises(M365MissingEnvironmentUserCredentialsError) as exc_info:
|
||||
with pytest.raises(M365MissingEnvironmentCredentialsError) as exc_info:
|
||||
M365Provider.setup_powershell(
|
||||
env_auth=True, m365_credentials=credentials
|
||||
)
|
||||
@@ -450,3 +482,143 @@ class TestM365Provider:
|
||||
in str(exc_info.value)
|
||||
)
|
||||
mock_session.test_credentials.assert_not_called()
|
||||
|
||||
def test_test_connection_user_not_belonging_to_tenant(
|
||||
self,
|
||||
):
|
||||
with patch(
|
||||
"prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials"
|
||||
) as mock_validate_static_credentials:
|
||||
mock_validate_static_credentials.side_effect = M365UserNotBelongingToTenantError(
|
||||
file=os.path.basename(__file__),
|
||||
message="The provided M365 User does not belong to the specified tenant.",
|
||||
)
|
||||
|
||||
with pytest.raises(M365UserNotBelongingToTenantError) as exception:
|
||||
M365Provider.test_connection(
|
||||
tenant_id="contoso.onmicrosoft.com",
|
||||
region="M365Global",
|
||||
raise_on_exception=True,
|
||||
client_id=str(uuid4()),
|
||||
client_secret=str(uuid4()),
|
||||
user="user@otherdomain.com",
|
||||
encrypted_password="test_password",
|
||||
)
|
||||
|
||||
assert exception.type == M365UserNotBelongingToTenantError
|
||||
assert (
|
||||
"The provided M365 User does not belong to the specified tenant."
|
||||
in str(exception.value)
|
||||
)
|
||||
|
||||
def test_validate_static_credentials_invalid_tenant_id(self):
|
||||
with pytest.raises(M365NotValidTenantIdError) as exception:
|
||||
M365Provider.validate_static_credentials(
|
||||
tenant_id="invalid-tenant-id",
|
||||
client_id="12345678-1234-5678-1234-567812345678",
|
||||
client_secret="test_secret",
|
||||
user="test@example.com",
|
||||
encrypted_password="test_password",
|
||||
)
|
||||
assert "The provided M365 Tenant ID is not valid." in str(exception.value)
|
||||
|
||||
def test_validate_static_credentials_missing_client_id(self):
|
||||
with pytest.raises(M365NotValidClientIdError) as exception:
|
||||
M365Provider.validate_static_credentials(
|
||||
tenant_id="12345678-1234-5678-1234-567812345678",
|
||||
client_id="",
|
||||
client_secret="test_secret",
|
||||
user="test@example.com",
|
||||
encrypted_password="test_password",
|
||||
)
|
||||
assert "The provided M365 Client ID is not valid." in str(exception.value)
|
||||
|
||||
def test_validate_static_credentials_missing_client_secret(self):
|
||||
with pytest.raises(M365NotValidClientSecretError) as exception:
|
||||
M365Provider.validate_static_credentials(
|
||||
tenant_id="12345678-1234-5678-1234-567812345678",
|
||||
client_id="12345678-1234-5678-1234-567812345678",
|
||||
client_secret="",
|
||||
user="test@example.com",
|
||||
encrypted_password="test_password",
|
||||
)
|
||||
assert "The provided M365 Client Secret is not valid." in str(exception.value)
|
||||
|
||||
def test_validate_static_credentials_missing_user(self):
|
||||
with pytest.raises(M365NotValidUserError) as exception:
|
||||
M365Provider.validate_static_credentials(
|
||||
tenant_id="12345678-1234-5678-1234-567812345678",
|
||||
client_id="12345678-1234-5678-1234-567812345678",
|
||||
client_secret="test_secret",
|
||||
user="",
|
||||
encrypted_password="test_password",
|
||||
)
|
||||
assert "The provided M365 User is not valid." in str(exception.value)
|
||||
|
||||
def test_validate_static_credentials_missing_encrypted_password(self):
|
||||
with pytest.raises(M365NotValidEncryptedPasswordError) as exception:
|
||||
M365Provider.validate_static_credentials(
|
||||
tenant_id="12345678-1234-5678-1234-567812345678",
|
||||
client_id="12345678-1234-5678-1234-567812345678",
|
||||
client_secret="test_secret",
|
||||
user="test@example.com",
|
||||
encrypted_password="",
|
||||
)
|
||||
assert "The provided M365 Encrypted Password is not valid." in str(
|
||||
exception.value
|
||||
)
|
||||
|
||||
def test_validate_arguments_missing_env_credentials(self):
|
||||
with pytest.raises(M365MissingEnvironmentCredentialsError) as exception:
|
||||
M365Provider.validate_arguments(
|
||||
az_cli_auth=False,
|
||||
sp_env_auth=False,
|
||||
env_auth=True,
|
||||
browser_auth=False,
|
||||
tenant_id=None,
|
||||
client_id="test_client_id",
|
||||
client_secret="test_secret",
|
||||
user=None,
|
||||
encrypted_password=None,
|
||||
)
|
||||
assert (
|
||||
"M365 provider requires AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID, M365_USER and M365_ENCRYPTED_PASSWORD environment variables to be set when using --env-auth"
|
||||
in str(exception.value)
|
||||
)
|
||||
|
||||
def test_test_connection_invalid_provider_id(self):
|
||||
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,
|
||||
):
|
||||
# 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
|
||||
|
||||
user_domain = "contoso.com"
|
||||
provider_id = "Test.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=f"user@{user_domain}",
|
||||
encrypted_password="test_password",
|
||||
provider_id=provider_id,
|
||||
)
|
||||
|
||||
assert exception.type == M365InvalidProviderIdError
|
||||
assert (
|
||||
f"Provider ID {provider_id} does not match Application tenant domain {user_domain}"
|
||||
in str(exception.value)
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user