From f6bb6efbf15f071af3f29afa45e213c4174f58cd Mon Sep 17 00:00:00 2001 From: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com> Date: Mon, 19 May 2025 10:08:54 +0200 Subject: [PATCH] chore(m365): accept all tenant domains in authentication (#7746) --- api/src/backend/api/models.py | 4 +- .../m365/lib/powershell/m365_powershell.py | 25 +- prowler/providers/m365/lib/service/service.py | 4 +- prowler/providers/m365/m365_provider.py | 176 ++++++------- prowler/providers/m365/models.py | 9 +- .../lib/powershell/m365_powershell_test.py | 232 ++++++++---------- tests/providers/m365/m365_provider_test.py | 124 +++++++++- 7 files changed, 340 insertions(+), 234 deletions(-) diff --git a/api/src/backend/api/models.py b/api/src/backend/api/models.py index 676b3b116c..61e8eb22a2 100644 --- a/api/src/backend/api/models.py +++ b/api/src/backend/api/models.py @@ -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", ) diff --git a/prowler/providers/m365/lib/powershell/m365_powershell.py b/prowler/providers/m365/lib/powershell/m365_powershell.py index f32921b001..9dbda1c01e 100644 --- a/prowler/providers/m365/lib/powershell/m365_powershell.py +++ b/prowler/providers/m365/lib/powershell/m365_powershell.py @@ -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: diff --git a/prowler/providers/m365/lib/service/service.py b/prowler/providers/m365/lib/service/service.py index b218aac283..8806c99728 100644 --- a/prowler/providers/m365/lib/service/service.py +++ b/prowler/providers/m365/lib/service/service.py @@ -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 ) diff --git a/prowler/providers/m365/m365_provider.py b/prowler/providers/m365/m365_provider.py index 2b766171e6..a4135b0692 100644 --- a/prowler/providers/m365/m365_provider.py +++ b/prowler/providers/m365/m365_provider.py @@ -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( diff --git a/prowler/providers/m365/models.py b/prowler/providers/m365/models.py index 198c54e33e..d0d73263b5 100644 --- a/prowler/providers/m365/models.py +++ b/prowler/providers/m365/models.py @@ -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): diff --git a/tests/providers/m365/lib/powershell/m365_powershell_test.py b/tests/providers/m365/lib/powershell/m365_powershell_test.py index 66094aa581..b841602914 100644 --- a/tests/providers/m365/lib/powershell/m365_powershell_test.py +++ b/tests/providers/m365/lib/powershell/m365_powershell_test.py @@ -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() diff --git a/tests/providers/m365/m365_provider_test.py b/tests/providers/m365/m365_provider_test.py index 1d8c0759e6..2f30efabde 100644 --- a/tests/providers/m365/m365_provider_test.py +++ b/tests/providers/m365/m365_provider_test.py @@ -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) + )