From ad25a8fe82e95e11eedc9ee157f2ca843e4e6804 Mon Sep 17 00:00:00 2001 From: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com> Date: Fri, 25 Apr 2025 16:44:25 +0200 Subject: [PATCH] fix(powershell): handle m365 provider execution and logging (#7602) Co-authored-by: MrCloudSec --- prowler/lib/powershell/powershell.py | 83 ++++++++-- .../m365/lib/powershell/m365_powershell.py | 55 ++++--- prowler/providers/m365/lib/service/service.py | 6 +- prowler/providers/m365/m365_provider.py | 7 +- .../services/defender/defender_service.py | 30 ++-- .../services/exchange/exchange_service.py | 18 ++- .../m365/services/purview/purview_service.py | 9 +- .../m365/services/teams/teams_service.py | 15 +- tests/lib/powershell/powershell_test.py | 150 ++++++++++++++++-- .../lib/powershell/m365_powershell_test.py | 94 ++++++++--- 10 files changed, 365 insertions(+), 102 deletions(-) diff --git a/prowler/lib/powershell/powershell.py b/prowler/lib/powershell/powershell.py index 373fc97183..26cd2fe952 100644 --- a/prowler/lib/powershell/powershell.py +++ b/prowler/lib/powershell/powershell.py @@ -4,6 +4,9 @@ import queue import re import subprocess import threading +from typing import Union + +from prowler.lib.logger import logger class PowerShellSession: @@ -100,7 +103,9 @@ class PowerShellSession: ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") return ansi_escape.sub("", text) - def execute(self, command: str) -> dict: + def execute( + self, command: str, json_parse: bool = False, timeout: int = 10 + ) -> Union[str, dict]: """ Send a command to PowerShell and retrieve its output. @@ -120,31 +125,41 @@ class PowerShellSession: """ self.process.stdin.write(f"{command}\n") self.process.stdin.write(f"Write-Output '{self.END}'\n") - return self.json_parse_output(self.read_output()) + self.process.stdin.write(f"Write-Error '{self.END}'\n") + return ( + self.json_parse_output(self.read_output(timeout=timeout)) + if json_parse + else self.read_output(timeout=timeout) + ) def read_output(self, timeout: int = 10, default: str = "") -> str: """ Read output from a process with timeout functionality. - This method reads lines from process stdout until it encounters the END marker - or the stream ends. If reading takes longer than the timeout period, the method - returns a default value while allowing the reading to continue in the background. + This method reads lines from process stdout and stderr in separate threads until it encounters + the END marker for each stream. If reading stdout takes longer than the timeout period, + the method returns a default value while allowing the reading to continue in the background. + + Any errors from stderr are logged but do not affect the return value. Args: - timeout (int, optional): Maximum time in seconds to wait for output. + timeout (int, optional): Maximum time in seconds to wait for stdout output. Defaults to 10. - default (str, optional): Value to return if timeout occurs. + default (str, optional): Value to return if stdout timeout occurs. Defaults to empty string. Returns: - str: Concatenated output lines or default value if timeout occurs. + str: The stdout output if available, otherwise the default value. + Errors from stderr are logged but not returned. Note: - This method uses a daemon thread to read the output asynchronously, + This method uses daemon threads to read stdout and stderr asynchronously, ensuring that the main thread is not blocked. """ output_lines = [] result_queue = queue.Queue() + error_lines = [] + error_queue = queue.Queue() def reader_thread(): try: @@ -154,17 +169,35 @@ class PowerShellSession: break output_lines.append(line) result_queue.put("\n".join(output_lines)) - except Exception as e: - result_queue.put(str(e)) + except Exception as error: + result_queue.put(str(error)) + + def error_reader_thread(): + try: + while True: + line = self.remove_ansi(self.process.stderr.readline().strip()) + if line == f"Write-Error: {self.END}": + break + error_lines.append(line) + error_queue.put("\n".join(error_lines)) + except Exception as error: + error_queue.put(str(error)) thread = threading.Thread(target=reader_thread) thread.daemon = True thread.start() - try: - return result_queue.get(timeout=timeout) - except queue.Empty: - return default + error_thread = threading.Thread(target=error_reader_thread) + error_thread.daemon = True + error_thread.start() + + result = result_queue.get(timeout=timeout) or default + error_result = error_queue.get(timeout=1) + + if error_result: + logger.error(f"PowerShell error output: {error_result}") + + return result def json_parse_output(self, output: str) -> dict: """ @@ -179,13 +212,29 @@ class PowerShellSession: Returns: dict: Parsed JSON object if found, otherwise an empty dictionary. + Raises: + JSONDecodeError: If the JSON parsing fails. + Example: >>> json_parse_output('Some text {"key": "value"} more text') {"key": "value"} """ + if output == "": + return {} + json_match = re.search(r"(\[.*\]|\{.*\})", output, re.DOTALL) - if json_match: - return json.loads(json_match.group(1)) + if not json_match: + logger.error( + f"Unexpected PowerShell output: {output}\n", + ) + else: + try: + return json.loads(json_match.group(1)) + except json.JSONDecodeError as error: + logger.error( + f"Error parsing PowerShell output as JSON: {str(error)}\n", + ) + return {} def close(self) -> None: diff --git a/prowler/providers/m365/lib/powershell/m365_powershell.py b/prowler/providers/m365/lib/powershell/m365_powershell.py index b3a7db9689..982e26672b 100644 --- a/prowler/providers/m365/lib/powershell/m365_powershell.py +++ b/prowler/providers/m365/lib/powershell/m365_powershell.py @@ -84,16 +84,14 @@ class M365PowerShell(PowerShellSession): bool: True if credentials are valid and authentication succeeds, False otherwise. """ self.execute( - f'$securePassword = "{credentials.passwd}" | ConvertTo-SecureString\n' + f'$securePassword = "{credentials.passwd}" | ConvertTo-SecureString' ) self.execute( f'$credential = New-Object System.Management.Automation.PSCredential("{credentials.user}", $securePassword)\n' ) - self.process.stdin.write( - 'Write-Output "$($credential.GetNetworkCredential().Password)"\n' + decrypted_password = self.execute( + 'Write-Output "$($credential.GetNetworkCredential().Password)"' ) - self.process.stdin.write(f"Write-Output '{self.END}'\n") - decrypted_password = self.read_output() app = msal.ConfidentialClientApplication( client_id=credentials.client_id, @@ -154,7 +152,9 @@ class M365PowerShell(PowerShellSession): "AllowGoogleDrive": true } """ - return self.execute("Get-CsTeamsClientConfiguration | ConvertTo-Json") + return self.execute( + "Get-CsTeamsClientConfiguration | ConvertTo-Json", json_parse=True + ) def get_global_meeting_policy(self) -> dict: """ @@ -172,7 +172,8 @@ class M365PowerShell(PowerShellSession): } """ return self.execute( - "Get-CsTeamsMeetingPolicy -Identity Global | ConvertTo-Json" + "Get-CsTeamsMeetingPolicy -Identity Global | ConvertTo-Json", + json_parse=True, ) def get_user_settings(self) -> dict: @@ -190,7 +191,9 @@ class M365PowerShell(PowerShellSession): "AllowExternalAccess": true } """ - return self.execute("Get-CsTenantFederationConfiguration | ConvertTo-Json") + return self.execute( + "Get-CsTenantFederationConfiguration | ConvertTo-Json", json_parse=True + ) def connect_exchange_online(self) -> dict: """ @@ -222,7 +225,8 @@ class M365PowerShell(PowerShellSession): } """ return self.execute( - "Get-AdminAuditLogConfig | Select-Object UnifiedAuditLogIngestionEnabled | ConvertTo-Json" + "Get-AdminAuditLogConfig | Select-Object UnifiedAuditLogIngestionEnabled | ConvertTo-Json", + json_parse=True, ) def get_malware_filter_policy(self) -> dict: @@ -241,7 +245,7 @@ class M365PowerShell(PowerShellSession): "Identity": "Default" } """ - return self.execute("Get-MalwareFilterPolicy | ConvertTo-Json") + return self.execute("Get-MalwareFilterPolicy | ConvertTo-Json", json_parse=True) def get_outbound_spam_filter_policy(self) -> dict: """ @@ -261,7 +265,9 @@ class M365PowerShell(PowerShellSession): "NotifyOutboundSpamRecipients": [] } """ - return self.execute("Get-HostedOutboundSpamFilterPolicy | ConvertTo-Json") + return self.execute( + "Get-HostedOutboundSpamFilterPolicy | ConvertTo-Json", json_parse=True + ) def get_outbound_spam_filter_rule(self) -> dict: """ @@ -278,7 +284,9 @@ class M365PowerShell(PowerShellSession): "State": "Enabled" } """ - return self.execute("Get-HostedOutboundSpamFilterRule | ConvertTo-Json") + return self.execute( + "Get-HostedOutboundSpamFilterRule | ConvertTo-Json", json_parse=True + ) def get_antiphishing_policy(self) -> dict: """ @@ -303,7 +311,7 @@ class M365PowerShell(PowerShellSession): "IsDefault": false } """ - return self.execute("Get-AntiPhishPolicy | ConvertTo-Json") + return self.execute("Get-AntiPhishPolicy | ConvertTo-Json", json_parse=True) def get_antiphishing_rules(self) -> dict: """ @@ -321,7 +329,7 @@ class M365PowerShell(PowerShellSession): "State": Enabled, } """ - return self.execute("Get-AntiPhishRule | ConvertTo-Json") + return self.execute("Get-AntiPhishRule | ConvertTo-Json", json_parse=True) def get_organization_config(self) -> dict: """ @@ -340,7 +348,7 @@ class M365PowerShell(PowerShellSession): "AuditDisabled": false } """ - return self.execute("Get-OrganizationConfig | ConvertTo-Json") + return self.execute("Get-OrganizationConfig | ConvertTo-Json", json_parse=True) def get_mailbox_audit_config(self) -> dict: """ @@ -359,7 +367,9 @@ class M365PowerShell(PowerShellSession): "AuditBypassEnabled": false } """ - return self.execute("Get-MailboxAuditBypassAssociation | ConvertTo-Json") + return self.execute( + "Get-MailboxAuditBypassAssociation | ConvertTo-Json", json_parse=True + ) def get_external_mail_config(self) -> dict: """ @@ -377,7 +387,7 @@ class M365PowerShell(PowerShellSession): "ExternalMailTagEnabled": true } """ - return self.execute("Get-ExternalInOutlook | ConvertTo-Json") + return self.execute("Get-ExternalInOutlook | ConvertTo-Json", json_parse=True) def get_transport_rules(self) -> dict: """ @@ -396,7 +406,7 @@ class M365PowerShell(PowerShellSession): "SenderDomainIs": ["example.com"] } """ - return self.execute("Get-TransportRule | ConvertTo-Json") + return self.execute("Get-TransportRule | ConvertTo-Json", json_parse=True) def get_connection_filter_policy(self) -> dict: """ @@ -415,7 +425,8 @@ class M365PowerShell(PowerShellSession): } """ return self.execute( - "Get-HostedConnectionFilterPolicy -Identity Default | ConvertTo-Json" + "Get-HostedConnectionFilterPolicy -Identity Default | ConvertTo-Json", + json_parse=True, ) def get_dkim_config(self) -> dict: @@ -434,7 +445,7 @@ class M365PowerShell(PowerShellSession): "Enabled": true } """ - return self.execute("Get-DkimSigningConfig | ConvertTo-Json") + return self.execute("Get-DkimSigningConfig | ConvertTo-Json", json_parse=True) def get_inbound_spam_filter_policy(self) -> dict: """ @@ -452,4 +463,6 @@ class M365PowerShell(PowerShellSession): "AllowedSenderDomains": "[]" } """ - return self.execute("Get-HostedContentFilterPolicy | ConvertTo-Json") + return self.execute( + "Get-HostedContentFilterPolicy | ConvertTo-Json", json_parse=True + ) diff --git a/prowler/providers/m365/lib/service/service.py b/prowler/providers/m365/lib/service/service.py index 02aeb2762c..b218aac283 100644 --- a/prowler/providers/m365/lib/service/service.py +++ b/prowler/providers/m365/lib/service/service.py @@ -13,5 +13,7 @@ class M365Service: self.audit_config = provider.audit_config self.fixer_config = provider.fixer_config - if provider.credentials: - self.powershell = M365PowerShell(provider.credentials) + # Initialize PowerShell client only if credentials are available + self.powershell = ( + M365PowerShell(provider.credentials) if provider.credentials else None + ) diff --git a/prowler/providers/m365/m365_provider.py b/prowler/providers/m365/m365_provider.py index 8608c88180..4898b5c311 100644 --- a/prowler/providers/m365/m365_provider.py +++ b/prowler/providers/m365/m365_provider.py @@ -100,7 +100,7 @@ class M365Provider(Provider): _audit_config: dict _region_config: M365RegionConfig _mutelist: M365Mutelist - _credentials: M365Credentials + _credentials: M365Credentials = {} # TODO: this is not optional, enforce for all providers audit_metadata: Audit_Metadata @@ -447,8 +447,11 @@ class M365Provider(Provider): f"M365 Region: {Fore.YELLOW}{self.region_config.name}{Style.RESET_ALL}", f"M365 Tenant Domain: {Fore.YELLOW}{self._identity.tenant_domain}{Style.RESET_ALL} M365 Tenant ID: {Fore.YELLOW}{self._identity.tenant_id}{Style.RESET_ALL}", f"M365 Identity Type: {Fore.YELLOW}{self._identity.identity_type}{Style.RESET_ALL} M365 Identity ID: {Fore.YELLOW}{self._identity.identity_id}{Style.RESET_ALL}", - f"M365 User: {Fore.YELLOW}{self.credentials.user}{Style.RESET_ALL}", ] + if self.credentials and self.credentials.user: + report_lines.append( + f"M365 User: {Fore.YELLOW}{self.credentials.user}{Style.RESET_ALL}" + ) report_title = ( f"{Style.BRIGHT}Using the M365 credentials below:{Style.RESET_ALL}" ) diff --git a/prowler/providers/m365/services/defender/defender_service.py b/prowler/providers/m365/services/defender/defender_service.py index 4fc86aa51c..3ed0eee58b 100644 --- a/prowler/providers/m365/services/defender/defender_service.py +++ b/prowler/providers/m365/services/defender/defender_service.py @@ -10,16 +10,26 @@ from prowler.providers.m365.m365_provider import M365Provider class Defender(M365Service): def __init__(self, provider: M365Provider): super().__init__(provider) - self.powershell.connect_exchange_online() - self.malware_policies = self._get_malware_filter_policy() - 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_antiphising_policy() - self.antiphising_rules = self._get_antiphising_rules() - self.inbound_spam_policies = self._get_inbound_spam_filter_policy() - self.connection_filter_policy = self._get_connection_filter_policy() - self.dkim_configurations = self._get_dkim_config() - self.powershell.close() + self.malware_policies = [] + self.outbound_spam_policies = {} + self.outbound_spam_rules = {} + self.antiphishing_policies = {} + self.antiphising_rules = {} + self.connection_filter_policy = None + self.dkim_configurations = [] + self.inbound_spam_policies = [] + + if self.powershell: + self.powershell.connect_exchange_online() + self.malware_policies = self._get_malware_filter_policy() + 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_antiphising_policy() + self.antiphising_rules = self._get_antiphising_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.powershell.close() def _get_malware_filter_policy(self): logger.info("M365 - Getting Defender malware filter policy...") diff --git a/prowler/providers/m365/services/exchange/exchange_service.py b/prowler/providers/m365/services/exchange/exchange_service.py index c1d544787f..5bd8cada6a 100644 --- a/prowler/providers/m365/services/exchange/exchange_service.py +++ b/prowler/providers/m365/services/exchange/exchange_service.py @@ -10,12 +10,18 @@ from prowler.providers.m365.m365_provider import M365Provider class Exchange(M365Service): def __init__(self, provider: M365Provider): super().__init__(provider) - 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.powershell.close() + self.organization_config = None + self.mailboxes_config = [] + self.external_mail_config = [] + self.transport_rules = [] + + 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.powershell.close() def _get_organization_config(self): logger.info("Microsoft365 - Getting Exchange Organization configuration...") diff --git a/prowler/providers/m365/services/purview/purview_service.py b/prowler/providers/m365/services/purview/purview_service.py index 2323b28dce..138305c899 100644 --- a/prowler/providers/m365/services/purview/purview_service.py +++ b/prowler/providers/m365/services/purview/purview_service.py @@ -8,9 +8,12 @@ from prowler.providers.m365.m365_provider import M365Provider class Purview(M365Service): def __init__(self, provider: M365Provider): super().__init__(provider) - self.powershell.connect_exchange_online() - self.audit_log_config = self._get_audit_log_config() - self.powershell.close() + self.audit_log_config = None + + if self.powershell: + self.powershell.connect_exchange_online() + self.audit_log_config = self._get_audit_log_config() + self.powershell.close() def _get_audit_log_config(self): logger.info("M365 - Getting Admin Audit Log settings...") diff --git a/prowler/providers/m365/services/teams/teams_service.py b/prowler/providers/m365/services/teams/teams_service.py index 22c4c3d6c0..df65cf400b 100644 --- a/prowler/providers/m365/services/teams/teams_service.py +++ b/prowler/providers/m365/services/teams/teams_service.py @@ -8,11 +8,16 @@ from prowler.providers.m365.m365_provider import M365Provider class Teams(M365Service): def __init__(self, provider: M365Provider): super().__init__(provider) - self.powershell.connect_microsoft_teams() - self.teams_settings = self._get_teams_client_configuration() - self.global_meeting_policy = self._get_global_meeting_policy() - self.user_settings = self._get_user_settings() - self.powershell.close() + self.teams_settings = None + self.global_meeting_policy = None + 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.user_settings = self._get_user_settings() + self.powershell.close() def _get_teams_client_configuration(self): logger.info("M365 - Getting Teams settings...") diff --git a/tests/lib/powershell/powershell_test.py b/tests/lib/powershell/powershell_test.py index 6dab9349a0..49108e1cd2 100644 --- a/tests/lib/powershell/powershell_test.py +++ b/tests/lib/powershell/powershell_test.py @@ -58,34 +58,109 @@ class TestPowerShellSession: @patch("subprocess.Popen") def test_execute(self, mock_popen): + """Test the execute method with various scenarios: + - Normal command execution + - JSON parsing enabled + - Timeout handling + - Error handling + """ + # Setup mock_process = MagicMock() mock_popen.return_value = mock_process session = PowerShellSession() - command = "Get-Command" - expected_output = '{"Name": "Get-Command"}' - with patch.object(session, "read_output", return_value=expected_output): - result = session.execute(command) - - mock_process.stdin.write.assert_any_call(f"{command}\n") + # Test 1: Normal command execution + mock_process.stdout.readline.side_effect = ["Hello World\n", f"{session.END}\n"] + mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n" + with patch.object(session, "remove_ansi", side_effect=lambda x: x): + result = session.execute("Get-Command") + assert result == "Hello World" + mock_process.stdin.write.assert_any_call("Get-Command\n") mock_process.stdin.write.assert_any_call(f"Write-Output '{session.END}'\n") - assert result == {"Name": "Get-Command"} + mock_process.stdin.write.assert_any_call(f"Write-Error '{session.END}'\n") + + # Test 2: JSON parsing enabled + mock_process.stdout.readline.side_effect = [ + '{"key": "value"}\n', + f"{session.END}\n", + ] + mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n" + with patch.object(session, "remove_ansi", side_effect=lambda x: x): + with patch.object( + session, "json_parse_output", return_value={"key": "value"} + ) as mock_json_parse: + result = session.execute("Get-Command", json_parse=True) + assert result == {"key": "value"} + mock_json_parse.assert_called_once_with('{"key": "value"}') + + # Test 3: Timeout handling + mock_process.stdout.readline.side_effect = ["test output\n"] # No END marker + mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n" + result = session.execute("Get-Command", timeout=0.1) + assert result == "" + + # Test 4: Error handling + mock_process.stdout.readline.side_effect = ["\n", f"{session.END}\n"] + mock_process.stderr.readline.side_effect = [ + "Write-Error: This is an error\n", + f"Write-Error: {session.END}\n", + ] + with patch.object(session, "remove_ansi", side_effect=lambda x: x): + with patch("prowler.lib.logger.logger.error") as mock_error: + result = session.execute("Get-Command") + assert result == "" + mock_error.assert_called_once_with( + "PowerShell error output: Write-Error: This is an error" + ) + session.close() @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 + """ + # Setup mock_process = MagicMock() mock_popen.return_value = mock_process session = PowerShellSession() - # Test normal output - with patch.object(session, "read_output", return_value="test@example.com"): - assert session.read_output() == "test@example.com" + # Test 1: Normal stdout output + mock_process.stdout.readline.side_effect = ["Hello World\n", f"{session.END}\n"] + mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n" + with patch.object(session, "remove_ansi", side_effect=lambda x: x): + result = session.read_output() + assert result == "Hello World" + + # Test 2: Error in stderr + mock_process.stdout.readline.side_effect = ["\n", f"{session.END}\n"] + mock_process.stderr.readline.side_effect = [ + "Write-Error: This is an error\n", + f"Write-Error: {session.END}\n", + ] + with patch.object(session, "remove_ansi", side_effect=lambda x: x): + with patch("prowler.lib.logger.logger.error") as mock_error: + result = session.read_output() + assert result == "" + mock_error.assert_called_once_with( + "PowerShell error output: Write-Error: This is an error" + ) + + # Test 3: Timeout in stdout + mock_process.stdout.readline.side_effect = ["test output\n"] # No END marker + mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n" + 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 timeout - mock_process.stdout.readline.return_value = "test output\n" - with patch.object(session, "remove_ansi", return_value="test output"): - assert session.read_output(timeout=0.1, default="") == "" session.close() @patch("subprocess.Popen") @@ -112,6 +187,53 @@ class TestPowerShellSession: assert result == expected session.close() + @patch("subprocess.Popen") + def test_json_parse_output_logging(self, mock_popen): + mock_process = MagicMock() + mock_popen.return_value = mock_process + session = PowerShellSession() + + # Test warning for non-JSON output + with patch("prowler.lib.logger.logger.error") as mock_error: + result = session.json_parse_output("some text without json") + assert result == {} + mock_error.assert_called_once_with( + "Unexpected PowerShell output: some text without json\n" + ) + + session.close() + + @patch("subprocess.Popen") + def test_json_parse_output_with_text_around_json(self, mock_popen): + mock_process = MagicMock() + mock_popen.return_value = mock_process + session = PowerShellSession() + + # Test JSON extraction from text with surrounding content + result = session.json_parse_output('some text {"key": "value"} more text') + assert result == {"key": "value"} + + result = session.json_parse_output('prefix [{"key": "value"}] suffix') + assert result == [{"key": "value"}] + + # Test non-JSON text returns empty dict + result = session.json_parse_output("just some text") + assert result == {} + + session.close() + + @patch("subprocess.Popen") + def test_json_parse_output_empty(self, mock_popen): + mock_process = MagicMock() + mock_popen.return_value = mock_process + session = PowerShellSession() + + # Test empty string + result = session.json_parse_output("") + assert result == {} + + session.close() + @patch("subprocess.Popen") def test_close(self, mock_popen): mock_process = MagicMock() diff --git a/tests/providers/m365/lib/powershell/m365_powershell_test.py b/tests/providers/m365/lib/powershell/m365_powershell_test.py index 56cc28ef33..72f5e72810 100644 --- a/tests/providers/m365/lib/powershell/m365_powershell_test.py +++ b/tests/providers/m365/lib/powershell/m365_powershell_test.py @@ -98,23 +98,28 @@ class Testm365PowerShell: ) session = M365PowerShell(credentials) - session.execute = MagicMock() - session.process.stdin.write = MagicMock() + # Mock read_output to return the decrypted password session.read_output = MagicMock(return_value="decrypted_password") - assert session.test_credentials(credentials) is True + # Mock execute to return the result of read_output + session.execute = MagicMock(side_effect=lambda _: session.read_output()) + # Execute the test + result = session.test_credentials(credentials) + assert result is True + + # Verify execute was called with the correct commands session.execute.assert_any_call( - f'$securePassword = "{credentials.passwd}" | ConvertTo-SecureString\n' + f'$securePassword = "{credentials.passwd}" | ConvertTo-SecureString' ) session.execute.assert_any_call( f'$credential = New-Object System.Management.Automation.PSCredential("{credentials.user}", $securePassword)\n' ) - session.process.stdin.write.assert_any_call( - 'Write-Output "$($credential.GetNetworkCredential().Password)"\n' + session.execute.assert_any_call( + 'Write-Output "$($credential.GetNetworkCredential().Password)"' ) - session.process.stdin.write.assert_any_call(f"Write-Output '{session.END}'\n") + # Verify MSAL was called with the correct parameters mock_msal.assert_called_once_with( client_id="test_client_id", client_credential="test_client_secret", @@ -148,7 +153,13 @@ class Testm365PowerShell: ) session = M365PowerShell(credentials) - session.execute = MagicMock() + # 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 + + session.execute = MagicMock(side_effect=mock_execute) session.process.stdin.write = MagicMock() session.read_output = MagicMock(return_value="decrypted_password") @@ -191,7 +202,13 @@ class Testm365PowerShell: ) session = M365PowerShell(credentials) - session.execute = MagicMock() + # 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 + + session.execute = MagicMock(side_effect=mock_execute) session.process.stdin.write = MagicMock() session.read_output = MagicMock(return_value="decrypted_password") @@ -207,6 +224,7 @@ class Testm365PowerShell: password="decrypted_password", scopes=["https://graph.microsoft.com/.default"], ) + session.close() @patch("subprocess.Popen") @@ -233,31 +251,63 @@ class Testm365PowerShell: credentials = M365Credentials(user="test@example.com", passwd="test_password") session = M365PowerShell(credentials) command = "Get-Command" - expected_output = '{"Name": "Get-Command"}' + expected_output = {"Name": "Get-Command"} - with patch.object(session, "read_output", return_value=expected_output): + with patch.object(session, "execute", return_value=expected_output): result = session.execute(command) - - mock_process.stdin.write.assert_any_call(f"{command}\n") - mock_process.stdin.write.assert_any_call(f"Write-Output '{session.END}'\n") - assert result == {"Name": "Get-Command"} + assert result == expected_output session.close() @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 + """ + # Setup mock_process = MagicMock() mock_popen.return_value = mock_process credentials = M365Credentials(user="test@example.com", passwd="test_password") session = M365PowerShell(credentials) - # Test normal output - with patch.object(session, "read_output", return_value="test@example.com"): - assert session.read_output() == "test@example.com" + # Test 1: Normal stdout output + mock_process.stdout.readline.side_effect = [ + "test@example.com\n", + f"{session.END}\n", + ] + mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n" + with patch.object(session, "remove_ansi", side_effect=lambda x: x): + result = session.read_output() + assert result == "test@example.com" + + # Test 2: Error in stderr + mock_process.stdout.readline.side_effect = ["\n", f"{session.END}\n"] + mock_process.stderr.readline.side_effect = [ + "Write-Error: Authentication failed\n", + f"Write-Error: {session.END}\n", + ] + with patch.object(session, "remove_ansi", side_effect=lambda x: x): + with patch("prowler.lib.logger.logger.error") as mock_error: + result = session.read_output() + assert result == "" + mock_error.assert_called_once_with( + "PowerShell error output: Write-Error: Authentication failed" + ) + + # Test 3: Timeout in stdout + mock_process.stdout.readline.side_effect = ["test output\n"] # No END marker + mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n" + 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 timeout - mock_process.stdout.readline.return_value = "test output\n" - with patch.object(session, "remove_ansi", return_value="test output"): - assert session.read_output(timeout=0.1, default="") == "" session.close() @patch("subprocess.Popen")