ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-09 21:14:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(m365): enhance execution to avoid multiple error calls (#8353)

Browse Source
This commit is contained in:
Hugo Pereira Brito
2025-07-30 14:54:27 +08:00
committed by GitHub
parent 2c86b3a990
commit b63f70ac82
10 changed files with 538 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/providers/m365/lib/jwt/jwt_decoder_test.py
+259
View File
@@ -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
tests/providers/m365/lib/powershell/m365_powershell_test.py
+96 -96
View File
@@ -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()