From b63f70ac82df5dbfcd464a811b508fb48c6b5650 Mon Sep 17 00:00:00 2001 From: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com> Date: Wed, 30 Jul 2025 08:54:27 +0200 Subject: [PATCH] fix(m365): enhance execution to avoid multiple error calls (#8353) --- prowler/CHANGELOG.md | 1 + prowler/providers/m365/lib/jwt/jwt_decoder.py | 68 +++++ .../m365/lib/powershell/m365_powershell.py | 176 ++++++------ .../admincenter/admincenter_service.py | 6 +- .../services/defender/defender_service.py | 24 +- .../services/exchange/exchange_service.py | 18 +- .../m365/services/purview/purview_service.py | 4 +- .../m365/services/teams/teams_service.py | 10 +- .../m365/lib/jwt/jwt_decoder_test.py | 259 ++++++++++++++++++ .../lib/powershell/m365_powershell_test.py | 192 ++++++------- 10 files changed, 538 insertions(+), 220 deletions(-) create mode 100644 prowler/providers/m365/lib/jwt/jwt_decoder.py create mode 100644 tests/providers/m365/lib/jwt/jwt_decoder_test.py diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index fa390aa664..5e70737670 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -31,6 +31,7 @@ All notable changes to the **Prowler SDK** are documented in this file. - `sns_topics_not_publicly_accessible` false positive with `aws:SourceArn` conditions [(#8326)](https://github.com/prowler-cloud/prowler/issues/8326) - Remove typo from description req 1.2.3 - Prowler ThreatScore m365 [(#8384)](https://github.com/prowler-cloud/prowler/pull/8384) - Way of counting FAILED/PASS reqs from `kisa_isms_p_2023_aws` table [(#8382)](https://github.com/prowler-cloud/prowler/pull/8382) +- Avoid multiple module error calls in M365 provider [(#8353)](https://github.com/prowler-cloud/prowler/pull/8353) --- diff --git a/prowler/providers/m365/lib/jwt/jwt_decoder.py b/prowler/providers/m365/lib/jwt/jwt_decoder.py new file mode 100644 index 0000000000..8f3484f987 --- /dev/null +++ b/prowler/providers/m365/lib/jwt/jwt_decoder.py @@ -0,0 +1,68 @@ +import base64 +import json +import re + + +def decode_jwt(token: str) -> dict: + """ + Decodes the payload of a JWT without verifying its signature. + + Args: + token (str): JWT string in the format 'header.payload.signature' + + Returns: + dict: A dictionary containing the decoded payload (claims), or an empty dict on failure. + """ + try: + # Split the JWT into its 3 parts + parts = token.split(".") + if len(parts) != 3: + raise ValueError( + "The token does not have the expected three-part structure." + ) + + # Extract and decode the payload (second part) + payload_b64 = parts[1] + + # Add padding if necessary for base64 decoding + padding = "=" * (-len(payload_b64) % 4) + payload_b64 += padding + + payload_bytes = base64.urlsafe_b64decode(payload_b64) + payload_json = payload_bytes.decode("utf-8") + payload = json.loads(payload_json) + + return payload + + except Exception as e: + print(f"Failed to decode the token: {e}") + return {} + + +def decode_msal_token(text: str) -> dict: + """ + Extracts and decodes the payload of a MSAL token from a given string. + + Args: + text (str): A string that contains the MSAL token, possibly over multiple lines. + + Returns: + dict: A dictionary containing the decoded payload (claims), or an empty dict on failure. + """ + try: + # Join all lines and remove whitespace + flattened = "".join(text.split()) + + # Search for a valid JWT pattern (three base64url parts separated by dots) + match = re.search( + r"([A-Za-z0-9-_]+\.[A-Za-z0-9-_]+\.[A-Za-z0-9-_]+)", flattened + ) + if not match: + raise ValueError("No valid JWT found in the input.") + + token = match.group(1) + return decode_jwt(token) + + except Exception as e: + print(f"Failed to extract and decode the token: {e}") + return {} diff --git a/prowler/providers/m365/lib/powershell/m365_powershell.py b/prowler/providers/m365/lib/powershell/m365_powershell.py index a9e3a0fdf9..f8c4ec5ff0 100644 --- a/prowler/providers/m365/lib/powershell/m365_powershell.py +++ b/prowler/providers/m365/lib/powershell/m365_powershell.py @@ -1,16 +1,14 @@ import os import platform -import msal - from prowler.lib.logger import logger from prowler.lib.powershell.powershell import PowerShellSession from prowler.providers.m365.exceptions.exceptions import ( - M365ExchangeConnectionError, M365GraphConnectionError, - M365TeamsConnectionError, + M365UserCredentialsError, M365UserNotBelongingToTenantError, ) +from prowler.providers.m365.lib.jwt.jwt_decoder import decode_jwt, decode_msal_token from prowler.providers.m365.models import M365Credentials, M365IdentityInfo @@ -162,26 +160,29 @@ class M365PowerShell(PowerShellSession): 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=credentials.passwd, - scopes=["https://graph.microsoft.com/.default"], - ) - - if result is None: - raise Exception( - "Unexpected error: Acquiring token in behalf of user did not return a result." - ) - - if "access_token" not in result: - raise Exception(f"MsGraph Error {result.get('error_description')}") + result = self.execute("Connect-ExchangeOnline -Credential $credential") + if "https://aka.ms/exov3-module" not in result: + if "AADSTS" in result: # Entra Security Token Service Error + raise M365UserCredentialsError( + file=os.path.basename(__file__), + message=result, + ) + else: # Could not connect to Exchange Online, try Microsoft Teams + result = self.execute( + "Connect-MicrosoftTeams -Credential $credential" + ) + if self.tenant_identity.tenant_id not in result: + if "AADSTS" in result: # Entra Security Token Service Error + raise M365UserCredentialsError( + file=os.path.basename(__file__), + message=result, + ) + else: # Unknown error, could be a permission issue or modules not installed + raise Exception( + file=os.path.basename(__file__), + message=f"Error connecting to PowerShell modules: {result}", + ) return True @@ -191,6 +192,7 @@ class M365PowerShell(PowerShellSession): logger.info("Testing Microsoft Graph connection...") self.test_graph_connection() logger.info("Microsoft Graph connection successful") + return True except Exception as e: logger.error(f"Microsoft Graph connection failed: {e}") raise M365GraphConnectionError( @@ -199,34 +201,6 @@ class M365PowerShell(PowerShellSession): message="Check your Microsoft Application credentials and ensure the app has proper permissions", ) - # Test Microsoft Teams connection - try: - logger.info("Testing Microsoft Teams connection...") - self.test_teams_connection() - logger.info("Microsoft Teams connection successful") - except Exception as e: - logger.error(f"Microsoft Teams connection failed: {e}") - raise M365TeamsConnectionError( - file=os.path.basename(__file__), - original_exception=e, - message="Ensure the application has proper permission granted to access Microsoft Teams.", - ) - - # Test Exchange Online connection - try: - logger.info("Testing Exchange Online connection...") - self.test_exchange_connection() - logger.info("Exchange Online connection successful") - except Exception as e: - logger.error(f"Exchange Online connection failed: {e}") - raise M365ExchangeConnectionError( - file=os.path.basename(__file__), - original_exception=e, - message="Ensure the application has proper permission granted to access Exchange Online.", - ) - - return True - def test_graph_connection(self) -> bool: """Test Microsoft Graph API connection and raise exception if it fails.""" try: @@ -253,19 +227,20 @@ class M365PowerShell(PowerShellSession): self.execute( '$teamsToken = Invoke-RestMethod -Uri "https://login.microsoftonline.com/$tenantID/oauth2/v2.0/token" -Method POST -Body $teamstokenBody | Select-Object -ExpandProperty Access_Token' ) - if self.execute("Write-Output $teamsToken") == "": - raise M365TeamsConnectionError( - file=os.path.basename(__file__), - message="Microsoft Teams token is empty or invalid.", + permissions = decode_jwt(self.execute("Write-Output $teamsToken")).get( + "roles", [] + ) + if "application_access" not in permissions: + logger.error( + "Microsoft Teams connection failed: Please check your permissions and try again." ) + return False return True except Exception as e: - logger.error(f"Microsoft Teams connection failed: {e}") - raise M365TeamsConnectionError( - file=os.path.basename(__file__), - original_exception=e, - message=f"Failed to connect to Microsoft Teams API: {str(e)}", + logger.error( + f"Microsoft Teams connection failed: {e}. Please check your permissions and try again." ) + return False def test_exchange_connection(self) -> bool: """Test Exchange Online API connection and raise exception if it fails.""" @@ -276,19 +251,19 @@ class M365PowerShell(PowerShellSession): self.execute( '$exchangeToken = Get-MsalToken -clientID "$clientID" -tenantID "$tenantID" -clientSecret $SecureSecret -Scopes "https://outlook.office365.com/.default"' ) - if self.execute("Write-Output $exchangeToken") == "": - raise M365ExchangeConnectionError( - file=os.path.basename(__file__), - message="Exchange Online token is empty or invalid.", + token = decode_msal_token(self.execute("Write-Output $exchangeToken")) + permissions = token.get("roles", []) + if "Exchange.ManageAsApp" not in permissions: + logger.error( + "Exchange Online connection failed: Please check your permissions and try again." ) + return False return True except Exception as e: - logger.error(f"Exchange Online connection failed: {e}") - raise M365ExchangeConnectionError( - file=os.path.basename(__file__), - original_exception=e, - message=f"Failed to connect to Exchange Online API: {str(e)}", + logger.error( + f"Exchange Online connection failed: {e}. Please check your permissions and try again." ) + return False def connect_microsoft_teams(self) -> dict: """ @@ -302,18 +277,26 @@ class M365PowerShell(PowerShellSession): Note: This method requires the Microsoft Teams PowerShell module to be installed. """ - if self.execute("Write-Output $credential") != "": # User Auth - return self.execute("Connect-MicrosoftTeams -Credential $credential") - else: # Application Auth - self.execute( - '$teamstokenBody = @{ Grant_Type = "client_credentials"; Scope = "48ac35b8-9aa8-4d74-927d-1f4a14a0b239/.default"; Client_Id = $clientID; Client_Secret = $clientSecret }' - ) - self.execute( - '$teamsToken = Invoke-RestMethod -Uri "https://login.microsoftonline.com/$tenantID/oauth2/v2.0/token" -Method POST -Body $teamstokenBody | Select-Object -ExpandProperty Access_Token' - ) - return self.execute( - 'Connect-MicrosoftTeams -AccessTokens @("$graphToken","$teamsToken")' - ) + # User Auth + if self.execute("Write-Output $credential") != "": + self.execute("Connect-MicrosoftTeams -Credential $credential") + # Test connection with a simple call + connection = self.execute("Get-CsTeamsClientConfiguration") + if connection: + return True + else: + logger.error( + "Microsoft Teams connection failed: Please check your permissions and try again." + ) + return connection + # Application Auth + else: + connection = self.test_teams_connection() + if connection: + self.execute( + 'Connect-MicrosoftTeams -AccessTokens @("$graphToken","$teamsToken")' + ) + return connection def get_teams_settings(self) -> dict: """ @@ -407,18 +390,25 @@ class M365PowerShell(PowerShellSession): Note: This method requires the Exchange Online PowerShell module to be installed. """ - if self.execute("Write-Output $credential") != "": # User Auth - return self.execute("Connect-ExchangeOnline -Credential $credential") - else: # Application Auth - self.execute( - '$SecureSecret = ConvertTo-SecureString "$clientSecret" -AsPlainText -Force' - ) - self.execute( - '$exchangeToken = Get-MsalToken -clientID "$clientID" -tenantID "$tenantID" -clientSecret $SecureSecret -Scopes "https://outlook.office365.com/.default"' - ) - return self.execute( - 'Connect-ExchangeOnline -AccessToken $exchangeToken.AccessToken -Organization "$tenantID"' - ) + # User Auth + if self.execute("Write-Output $credential") != "": + self.execute("Connect-ExchangeOnline -Credential $credential") + connection = self.execute("Get-OrganizationConfig") + if connection: + return True + else: + logger.error( + "Exchange Online connection failed: Please check your permissions and try again." + ) + return False + # Application Auth + else: + connection = self.test_exchange_connection() + if connection: + self.execute( + 'Connect-ExchangeOnline -AccessToken $exchangeToken.AccessToken -Organization "$tenantID"' + ) + return connection def get_audit_log_config(self) -> dict: """ diff --git a/prowler/providers/m365/services/admincenter/admincenter_service.py b/prowler/providers/m365/services/admincenter/admincenter_service.py index 7c28ff15a2..3a3e1f3284 100644 --- a/prowler/providers/m365/services/admincenter/admincenter_service.py +++ b/prowler/providers/m365/services/admincenter/admincenter_service.py @@ -15,9 +15,9 @@ class AdminCenter(M365Service): self.organization_config = None self.sharing_policy = None if self.powershell: - self.powershell.connect_exchange_online() - self.organization_config = self._get_organization_config() - self.sharing_policy = self._get_sharing_policy() + if self.powershell.connect_exchange_online(): + self.organization_config = self._get_organization_config() + self.sharing_policy = self._get_sharing_policy() self.powershell.close() loop = get_event_loop() diff --git a/prowler/providers/m365/services/defender/defender_service.py b/prowler/providers/m365/services/defender/defender_service.py index 433d5b5422..cdfd53a6b7 100644 --- a/prowler/providers/m365/services/defender/defender_service.py +++ b/prowler/providers/m365/services/defender/defender_service.py @@ -21,18 +21,18 @@ class Defender(M365Service): self.inbound_spam_rules = {} self.report_submission_policy = None if self.powershell: - self.powershell.connect_exchange_online() - self.malware_policies = self._get_malware_filter_policy() - self.malware_rules = self._get_malware_filter_rule() - self.outbound_spam_policies = self._get_outbound_spam_filter_policy() - self.outbound_spam_rules = self._get_outbound_spam_filter_rule() - self.antiphishing_policies = self._get_antiphishing_policy() - self.antiphishing_rules = self._get_antiphishing_rules() - self.connection_filter_policy = self._get_connection_filter_policy() - self.dkim_configurations = self._get_dkim_config() - self.inbound_spam_policies = self._get_inbound_spam_filter_policy() - self.inbound_spam_rules = self._get_inbound_spam_filter_rule() - self.report_submission_policy = self._get_report_submission_policy() + if self.powershell.connect_exchange_online(): + self.malware_policies = self._get_malware_filter_policy() + self.malware_rules = self._get_malware_filter_rule() + self.outbound_spam_policies = self._get_outbound_spam_filter_policy() + self.outbound_spam_rules = self._get_outbound_spam_filter_rule() + self.antiphishing_policies = self._get_antiphishing_policy() + self.antiphishing_rules = self._get_antiphishing_rules() + self.connection_filter_policy = self._get_connection_filter_policy() + self.dkim_configurations = self._get_dkim_config() + self.inbound_spam_policies = self._get_inbound_spam_filter_policy() + self.inbound_spam_rules = self._get_inbound_spam_filter_rule() + self.report_submission_policy = self._get_report_submission_policy() self.powershell.close() def _get_malware_filter_policy(self): diff --git a/prowler/providers/m365/services/exchange/exchange_service.py b/prowler/providers/m365/services/exchange/exchange_service.py index 07bbbe2ae6..96bbd4e13c 100644 --- a/prowler/providers/m365/services/exchange/exchange_service.py +++ b/prowler/providers/m365/services/exchange/exchange_service.py @@ -21,15 +21,15 @@ class Exchange(M365Service): self.mailbox_audit_properties = [] if self.powershell: - self.powershell.connect_exchange_online() - self.organization_config = self._get_organization_config() - self.mailboxes_config = self._get_mailbox_audit_config() - self.external_mail_config = self._get_external_mail_config() - self.transport_rules = self._get_transport_rules() - self.transport_config = self._get_transport_config() - self.mailbox_policy = self._get_mailbox_policy() - self.role_assignment_policies = self._get_role_assignment_policies() - self.mailbox_audit_properties = self._get_mailbox_audit_properties() + if self.powershell.connect_exchange_online(): + self.organization_config = self._get_organization_config() + self.mailboxes_config = self._get_mailbox_audit_config() + self.external_mail_config = self._get_external_mail_config() + self.transport_rules = self._get_transport_rules() + self.transport_config = self._get_transport_config() + self.mailbox_policy = self._get_mailbox_policy() + self.role_assignment_policies = self._get_role_assignment_policies() + self.mailbox_audit_properties = self._get_mailbox_audit_properties() self.powershell.close() def _get_organization_config(self): diff --git a/prowler/providers/m365/services/purview/purview_service.py b/prowler/providers/m365/services/purview/purview_service.py index c950b9e5cc..e93f495353 100644 --- a/prowler/providers/m365/services/purview/purview_service.py +++ b/prowler/providers/m365/services/purview/purview_service.py @@ -11,8 +11,8 @@ class Purview(M365Service): self.audit_log_config = None if self.powershell: - self.powershell.connect_exchange_online() - self.audit_log_config = self._get_audit_log_config() + if self.powershell.connect_exchange_online(): + self.audit_log_config = self._get_audit_log_config() self.powershell.close() def _get_audit_log_config(self): diff --git a/prowler/providers/m365/services/teams/teams_service.py b/prowler/providers/m365/services/teams/teams_service.py index 66970b0ec7..4045834c43 100644 --- a/prowler/providers/m365/services/teams/teams_service.py +++ b/prowler/providers/m365/services/teams/teams_service.py @@ -14,11 +14,11 @@ class Teams(M365Service): self.user_settings = None if self.powershell: - self.powershell.connect_microsoft_teams() - self.teams_settings = self._get_teams_client_configuration() - self.global_meeting_policy = self._get_global_meeting_policy() - self.global_messaging_policy = self._get_global_messaging_policy() - self.user_settings = self._get_user_settings() + if self.powershell.connect_microsoft_teams(): + self.teams_settings = self._get_teams_client_configuration() + self.global_meeting_policy = self._get_global_meeting_policy() + self.global_messaging_policy = self._get_global_messaging_policy() + self.user_settings = self._get_user_settings() self.powershell.close() def _get_teams_client_configuration(self): diff --git a/tests/providers/m365/lib/jwt/jwt_decoder_test.py b/tests/providers/m365/lib/jwt/jwt_decoder_test.py new file mode 100644 index 0000000000..29d0a00397 --- /dev/null +++ b/tests/providers/m365/lib/jwt/jwt_decoder_test.py @@ -0,0 +1,259 @@ +import base64 +import json +from unittest.mock import patch + +from prowler.providers.m365.lib.jwt.jwt_decoder import decode_jwt, decode_msal_token + + +class TestJwtDecoder: + def test_decode_jwt_valid_token(self): + """Test decode_jwt with a valid JWT token""" + # Create a mock JWT token + header = {"alg": "HS256", "typ": "JWT"} + payload = { + "sub": "1234567890", + "name": "John Doe", + "iat": 1516239022, + "roles": ["application_access", "user_read"], + } + + # Encode header and payload + header_b64 = ( + base64.urlsafe_b64encode(json.dumps(header).encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + payload_b64 = ( + base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + # Create JWT with dummy signature + token = f"{header_b64}.{payload_b64}.dummy_signature" + + result = decode_jwt(token) + + assert result == payload + assert result["sub"] == "1234567890" + assert result["name"] == "John Doe" + assert result["roles"] == ["application_access", "user_read"] + + def test_decode_jwt_valid_token_with_padding(self): + """Test decode_jwt with a token that needs base64 padding""" + # Create mock payload that will need padding + payload = {"test": "data"} + payload_json = json.dumps(payload) + + # Encode mock payload without padding + payload_b64 = ( + base64.urlsafe_b64encode(payload_json.encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + token = f"header.{payload_b64}.signature" + + result = decode_jwt(token) + + assert result == payload + + def test_decode_jwt_invalid_structure_two_parts(self): + """Test decode_jwt with token that has only 2 parts""" + token = "header.payload" # Missing signature + + result = decode_jwt(token) + + assert result == {} + + def test_decode_jwt_invalid_structure_four_parts(self): + """Test decode_jwt with token that has 4 parts""" + token = "header.payload.signature.extra" + + result = decode_jwt(token) + + assert result == {} + + def test_decode_jwt_invalid_base64(self): + """Test decode_jwt with invalid base64 in payload""" + token = "header.invalid_base64!@#.signature" + + result = decode_jwt(token) + + assert result == {} + + def test_decode_jwt_invalid_json(self): + """Test decode_jwt with invalid JSON in payload""" + # Create invalid JSON base64 + invalid_json = "{'invalid': json,}" + payload_b64 = ( + base64.urlsafe_b64encode(invalid_json.encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + token = f"header.{payload_b64}.signature" + + result = decode_jwt(token) + + assert result == {} + + def test_decode_jwt_empty_token(self): + """Test decode_jwt with empty token""" + result = decode_jwt("") + assert result == {} + + def test_decode_jwt_none_token(self): + """Test decode_jwt with None token""" + assert decode_jwt(None) == {} + + @patch("builtins.print") + def test_decode_jwt_prints_error_on_failure(self, mock_print): + """Test that decode_jwt prints error message on failure""" + token = "invalid.token" + + result = decode_jwt(token) + + assert result == {} + mock_print.assert_called_once() + assert "Failed to decode the token:" in mock_print.call_args[0][0] + + def test_decode_msal_token_valid_single_line(self): + """Test decode_msal_token with valid JWT in single line""" + # Create a valid JWT + payload = {"roles": ["Exchange.ManageAsApp"], "tenant": "test-tenant"} + payload_b64 = ( + base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + jwt_token = f"header.{payload_b64}.signature" + text = f"Some text before {jwt_token} some text after" + + result = decode_msal_token(text) + + assert result == payload + assert result["roles"] == ["Exchange.ManageAsApp"] + + def test_decode_msal_token_valid_multiline(self): + """Test decode_msal_token with valid JWT across multiple lines""" + payload = {"roles": ["application_access"], "user": "test@contoso.com"} + payload_b64 = ( + base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + jwt_token = f"header.{payload_b64}.signature" + text = f"""Line 1 + Line 2 with {jwt_token} + Line 3""" + + result = decode_msal_token(text) + + assert result == payload + assert result["user"] == "test@contoso.com" + + def test_decode_msal_token_with_whitespace(self): + """Test decode_msal_token with JWT containing whitespace""" + payload = {"test": "data"} + payload_b64 = ( + base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + jwt_token = f"header.{payload_b64}.signature" + text = f" Token: {jwt_token} " + + result = decode_msal_token(text) + + assert result == payload + + def test_decode_msal_token_no_jwt_found(self): + """Test decode_msal_token when no JWT pattern is found""" + text = "This text contains no JWT tokens at all" + + result = decode_msal_token(text) + + assert result == {} + + def test_decode_msal_token_invalid_jwt_pattern(self): + """Test decode_msal_token with text that looks like JWT but isn't""" + text = "header.payload" # Only 2 parts, not valid JWT + + result = decode_msal_token(text) + + assert result == {} + + def test_decode_msal_token_empty_text(self): + """Test decode_msal_token with empty text""" + result = decode_msal_token("") + assert result == {} + + def test_decode_msal_token_none_text(self): + """Test decode_msal_token with None text""" + assert decode_msal_token(None) == {} + + @patch("builtins.print") + def test_decode_msal_token_prints_error_on_failure(self, mock_print): + """Test that decode_msal_token prints error message on failure""" + text = "No JWT here" + + result = decode_msal_token(text) + + assert result == {} + mock_print.assert_called_once() + assert "Failed to extract and decode the token:" in mock_print.call_args[0][0] + + def test_decode_msal_token_real_world_scenario(self): + """Test decode_msal_token with a realistic PowerShell output scenario""" + # Simulate output from Get-MsalToken or similar + payload = { + "aud": "https://graph.microsoft.com", + "iss": "https://sts.windows.net/tenant-id/", + "iat": 1640995200, + "exp": 1641081600, + "roles": ["Application.ReadWrite.All"], + "sub": "app-subject-id", + } + payload_b64 = ( + base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + jwt_token = f"eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.{payload_b64}.signature123" + + # Simulate PowerShell output format + powershell_output = f""" + AccessToken : {jwt_token} + TokenType : Bearer + ExpiresOn : 1/2/2022 12:00:00 AM +00:00 + ExtendedExpiresOn : 1/2/2022 12:00:00 AM +00:00 + """ + + result = decode_msal_token(powershell_output) + + assert result == payload + assert result["roles"] == ["Application.ReadWrite.All"] + assert result["aud"] == "https://graph.microsoft.com" + + def test_decode_msal_token_with_jwt_in_json(self): + """Test decode_msal_token with JWT embedded in JSON-like structure""" + payload = {"tenant": "test", "scope": "https://graph.microsoft.com/.default"} + payload_b64 = ( + base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")) + .decode("utf-8") + .rstrip("=") + ) + + jwt_token = f"header.{payload_b64}.signature" + + json_like_text = f'{{"access_token": "{jwt_token}", "token_type": "Bearer"}}' + + result = decode_msal_token(json_like_text) + + assert result == payload diff --git a/tests/providers/m365/lib/powershell/m365_powershell_test.py b/tests/providers/m365/lib/powershell/m365_powershell_test.py index f02f283d1f..2e6a34a032 100644 --- a/tests/providers/m365/lib/powershell/m365_powershell_test.py +++ b/tests/providers/m365/lib/powershell/m365_powershell_test.py @@ -4,9 +4,8 @@ import pytest from prowler.lib.powershell.powershell import PowerShellSession from prowler.providers.m365.exceptions.exceptions import ( - M365ExchangeConnectionError, M365GraphConnectionError, - M365TeamsConnectionError, + M365UserCredentialsError, M365UserNotBelongingToTenantError, ) from prowler.providers.m365.lib.powershell.m365_powershell import M365PowerShell @@ -113,15 +112,9 @@ class Testm365PowerShell: session.close() @patch("subprocess.Popen") - @patch("msal.ConfidentialClientApplication") - def test_test_credentials(self, mock_msal, mock_popen): + def test_test_credentials(self, 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="test@contoso.onmicrosoft.com", @@ -143,7 +136,11 @@ class Testm365PowerShell: # Mock encrypt_password to return a known value session.encrypt_password = MagicMock(return_value="encrypted_password") - session.execute = MagicMock() + + # Mock execute to simulate successful Connect-ExchangeOnline + session.execute = MagicMock( + return_value="Connected successfully https://aka.ms/exov3-module" + ) # Execute the test result = session.test_credentials(credentials) @@ -156,18 +153,10 @@ class Testm365PowerShell: session.execute.assert_any_call( f'$credential = New-Object System.Management.Automation.PSCredential("{session.sanitize(credentials.user)}", $securePassword)' ) + session.execute.assert_any_call( + "Connect-ExchangeOnline -Credential $credential" + ) - # Verify MSAL was called with the correct parameters - 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="test_password", # Original password, not encrypted - scopes=["https://graph.microsoft.com/.default"], - ) session.close() @patch("subprocess.Popen") @@ -255,13 +244,9 @@ class Testm365PowerShell: session.close() @patch("subprocess.Popen") - @patch("msal.ConfidentialClientApplication") - def test_test_credentials_auth_failure(self, mock_msal, mock_popen): + def test_test_credentials_auth_failure_aadsts_error(self, 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", @@ -281,46 +266,37 @@ class Testm365PowerShell: ) session = M365PowerShell(credentials, identity) - # Mock the execute method to return the decrypted password - def mock_execute(command, *args, **kwargs): - if "Write-Output" in command: - return "decrypted_password" - return None + # Mock encrypt_password and execute to simulate AADSTS error + session.encrypt_password = MagicMock(return_value="encrypted_password") + session.execute = MagicMock( + return_value="AADSTS50126: Error validating credentials due to invalid username or password" + ) - session.execute = MagicMock(side_effect=mock_execute) - session.process.stdin.write = MagicMock() - session.read_output = MagicMock(return_value="decrypted_password") - - with pytest.raises(Exception) as exc_info: + with pytest.raises(M365UserCredentialsError) as exc_info: session.test_credentials(credentials) + assert ( - "Unexpected error: Acquiring token in behalf of user did not return a result." + "AADSTS50126: Error validating credentials due to invalid username or password" in str(exc_info.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", + # Verify execute was called with the correct commands + session.execute.assert_any_call( + f'$securePassword = "{credentials.encrypted_passwd}" | ConvertTo-SecureString' ) - mock_msal_instance.acquire_token_by_username_password.assert_called_once_with( - username="test@contoso.onmicrosoft.com", - password="test_password", - scopes=["https://graph.microsoft.com/.default"], + session.execute.assert_any_call( + f'$credential = New-Object System.Management.Automation.PSCredential("{session.sanitize(credentials.user)}", $securePassword)' + ) + session.execute.assert_any_call( + "Connect-ExchangeOnline -Credential $credential" ) session.close() @patch("subprocess.Popen") - @patch("msal.ConfidentialClientApplication") - def test_test_credentials_auth_failure_no_access_token(self, mock_msal, mock_popen): + def test_test_credentials_auth_failure_no_access_token(self, 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 = { - "error_description": "invalid_grant: authentication failed" - } credentials = M365Credentials( user="test@contoso.onmicrosoft.com", @@ -340,31 +316,29 @@ class Testm365PowerShell: ) session = M365PowerShell(credentials, identity) - # Mock the execute method to return the decrypted password - def mock_execute(command, *args, **kwargs): - if "Write-Output" in command: - return "decrypted_password" - return None + # Mock encrypt_password and execute to simulate AADSTS invalid grant error + session.encrypt_password = MagicMock(return_value="encrypted_password") + session.execute = MagicMock( + return_value="AADSTS70002: The request body must contain the following parameter: 'client_secret' or 'client_assertion'." + ) - session.execute = MagicMock(side_effect=mock_execute) - session.process.stdin.write = MagicMock() - session.read_output = MagicMock(return_value="decrypted_password") - - with pytest.raises(Exception) as exc_info: + with pytest.raises(M365UserCredentialsError) as exc_info: session.test_credentials(credentials) - assert "MsGraph Error invalid_grant: authentication failed" in str( - exc_info.value + + assert ( + "AADSTS70002: The request body must contain the following parameter: 'client_secret' or 'client_assertion'." + in str(exc_info.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", + # Verify execute was called with the correct commands + session.execute.assert_any_call( + f'$securePassword = "{credentials.encrypted_passwd}" | ConvertTo-SecureString' ) - mock_msal_instance.acquire_token_by_username_password.assert_called_once_with( - username="test@contoso.onmicrosoft.com", - password="test_password", - scopes=["https://graph.microsoft.com/.default"], + session.execute.assert_any_call( + f'$credential = New-Object System.Management.Automation.PSCredential("{session.sanitize(credentials.user)}", $securePassword)' + ) + session.execute.assert_any_call( + "Connect-ExchangeOnline -Credential $credential" ) session.close() @@ -744,7 +718,8 @@ class Testm365PowerShell: session.close() @patch("subprocess.Popen") - def test_test_teams_connection_success(self, mock_popen): + @patch("prowler.providers.m365.lib.powershell.m365_powershell.decode_jwt") + def test_test_teams_connection_success(self, mock_decode_jwt, mock_popen): """Test test_teams_connection when token is valid""" mock_process = MagicMock() mock_popen.return_value = mock_process @@ -766,17 +741,23 @@ class Testm365PowerShell: return None session.execute = MagicMock(side_effect=mock_execute) + # Mock JWT decode to return proper permissions + mock_decode_jwt.return_value = {"roles": ["application_access"]} result = session.test_teams_connection() assert result is True # Verify all expected PowerShell commands were called assert session.execute.call_count == 3 + mock_decode_jwt.assert_called_once_with("valid_teams_token") session.close() @patch("subprocess.Popen") - def test_test_teams_connection_empty_token(self, mock_popen): - """Test test_teams_connection when token is empty""" + @patch("prowler.providers.m365.lib.powershell.m365_powershell.decode_jwt") + def test_test_teams_connection_missing_permissions( + self, mock_decode_jwt, mock_popen + ): + """Test test_teams_connection when token lacks required permissions""" mock_process = MagicMock() mock_popen.return_value = mock_process credentials = M365Credentials(user="test@example.com", passwd="test_password") @@ -790,18 +771,23 @@ class Testm365PowerShell: ) session = M365PowerShell(credentials, identity) - # Mock execute to return empty token when checking + # Mock execute to return valid token but decode returns no permissions def mock_execute(command, *args, **kwargs): if "Write-Output $teamsToken" in command: - return "" + return "valid_teams_token" return None session.execute = MagicMock(side_effect=mock_execute) + # Mock JWT decode to return missing required permission + mock_decode_jwt.return_value = {"roles": ["other_permission"]} - with pytest.raises(M365TeamsConnectionError) as exc_info: - session.test_teams_connection() + with patch("prowler.lib.logger.logger.error") as mock_error: + result = session.test_teams_connection() - assert "Microsoft Teams token is empty or invalid" in str(exc_info.value) + assert result is False + mock_error.assert_called_once_with( + "Microsoft Teams connection failed: Please check your permissions and try again." + ) session.close() @patch("subprocess.Popen") @@ -823,16 +809,18 @@ class Testm365PowerShell: # Mock execute to raise an exception session.execute = MagicMock(side_effect=Exception("Teams API error")) - with pytest.raises(M365TeamsConnectionError) as exc_info: - session.test_teams_connection() + with patch("prowler.lib.logger.logger.error") as mock_error: + result = session.test_teams_connection() - assert "Failed to connect to Microsoft Teams API: Teams API error" in str( - exc_info.value + assert result is False + mock_error.assert_called_once_with( + "Microsoft Teams connection failed: Teams API error. Please check your permissions and try again." ) session.close() @patch("subprocess.Popen") - def test_test_exchange_connection_success(self, mock_popen): + @patch("prowler.providers.m365.lib.powershell.m365_powershell.decode_msal_token") + def test_test_exchange_connection_success(self, mock_decode_msal_token, mock_popen): """Test test_exchange_connection when token is valid""" mock_process = MagicMock() mock_popen.return_value = mock_process @@ -854,17 +842,23 @@ class Testm365PowerShell: return None session.execute = MagicMock(side_effect=mock_execute) + # Mock MSAL token decode to return proper permissions + mock_decode_msal_token.return_value = {"roles": ["Exchange.ManageAsApp"]} result = session.test_exchange_connection() assert result is True # Verify all expected PowerShell commands were called assert session.execute.call_count == 3 + mock_decode_msal_token.assert_called_once_with("valid_exchange_token") session.close() @patch("subprocess.Popen") - def test_test_exchange_connection_empty_token(self, mock_popen): - """Test test_exchange_connection when token is empty""" + @patch("prowler.providers.m365.lib.powershell.m365_powershell.decode_msal_token") + def test_test_exchange_connection_missing_permissions( + self, mock_decode_msal_token, mock_popen + ): + """Test test_exchange_connection when token lacks required permissions""" mock_process = MagicMock() mock_popen.return_value = mock_process credentials = M365Credentials(user="test@example.com", passwd="test_password") @@ -878,18 +872,23 @@ class Testm365PowerShell: ) session = M365PowerShell(credentials, identity) - # Mock execute to return empty token when checking + # Mock execute to return valid token but decode returns no permissions def mock_execute(command, *args, **kwargs): if "Write-Output $exchangeToken" in command: - return "" + return "valid_exchange_token" return None session.execute = MagicMock(side_effect=mock_execute) + # Mock MSAL token decode to return missing required permission + mock_decode_msal_token.return_value = {"roles": ["other_permission"]} - with pytest.raises(M365ExchangeConnectionError) as exc_info: - session.test_exchange_connection() + with patch("prowler.lib.logger.logger.error") as mock_error: + result = session.test_exchange_connection() - assert "Exchange Online token is empty or invalid" in str(exc_info.value) + assert result is False + mock_error.assert_called_once_with( + "Exchange Online connection failed: Please check your permissions and try again." + ) session.close() @patch("subprocess.Popen") @@ -911,11 +910,12 @@ class Testm365PowerShell: # Mock execute to raise an exception session.execute = MagicMock(side_effect=Exception("Exchange API error")) - with pytest.raises(M365ExchangeConnectionError) as exc_info: - session.test_exchange_connection() + with patch("prowler.lib.logger.logger.error") as mock_error: + result = session.test_exchange_connection() - assert "Failed to connect to Exchange Online API: Exchange API error" in str( - exc_info.value + assert result is False + mock_error.assert_called_once_with( + "Exchange Online connection failed: Exchange API error. Please check your permissions and try again." ) session.close()