From 85af4ff77c162287ed832e012968ba2bb13f73e6 Mon Sep 17 00:00:00 2001 From: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com> Date: Mon, 11 Aug 2025 09:47:56 +0200 Subject: [PATCH] feat(m365): add certificate auth method to cli (#8404) Co-authored-by: Sergio Garcia --- prowler/CHANGELOG.md | 2 +- prowler/providers/common/provider.py | 2 + .../providers/m365/exceptions/exceptions.py | 33 + .../providers/m365/lib/arguments/arguments.py | 11 + .../m365/lib/powershell/m365_powershell.py | 209 ++- prowler/providers/m365/m365_provider.py | 355 ++++- prowler/providers/m365/models.py | 4 +- .../m365/lib/arguments/m365_arguments_test.py | 641 ++++++++ .../lib/powershell/m365_powershell_test.py | 760 ++++++++- tests/providers/m365/m365_fixtures.py | 3 +- tests/providers/m365/m365_provider_test.py | 1397 ++++++++++++++++- 11 files changed, 3258 insertions(+), 159 deletions(-) create mode 100644 tests/providers/m365/lib/arguments/m365_arguments_test.py diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 07e798ae8b..2a1428ef18 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -2,10 +2,10 @@ All notable changes to the **Prowler SDK** are documented in this file. - ## [v5.11.0] (Prowler UNRELEASED) ### Added +- Certificate authentication for M365 provider [(#8404)](https://github.com/prowler-cloud/prowler/pull/8404) ### Changed diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index f0e233017b..5f3ee07160 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -222,6 +222,8 @@ class Provider(ABC): env_auth=arguments.env_auth, az_cli_auth=arguments.az_cli_auth, browser_auth=arguments.browser_auth, + certificate_auth=arguments.certificate_auth, + certificate_path=arguments.certificate_path, tenant_id=arguments.tenant_id, init_modules=arguments.init_modules, fixer_config=fixer_config, diff --git a/prowler/providers/m365/exceptions/exceptions.py b/prowler/providers/m365/exceptions/exceptions.py index 7f9dae2c31..871607aeff 100644 --- a/prowler/providers/m365/exceptions/exceptions.py +++ b/prowler/providers/m365/exceptions/exceptions.py @@ -126,6 +126,18 @@ class M365BaseException(ProwlerException): "message": "Failed to establish connection to Exchange Online API.", "remediation": "Ensure the application has proper permission granted to access Exchange Online.", }, + (6030, "M365CertificateCreationError"): { + "message": "Failed to create X.509 certificate object from provided certificate content.", + "remediation": "Ensure the certificate content is valid base64 encoded X.509 certificate data and is properly formatted.", + }, + (6031, "M365NotValidCertificateContentError"): { + "message": "The provided certificate content is not valid base64 encoded data.", + "remediation": "Ensure the certificate content is valid base64 encoded X.509 certificate data without line breaks or invalid characters.", + }, + (6032, "M365NotValidCertificatePathError"): { + "message": "The provided certificate path is not valid or the file cannot be accessed.", + "remediation": "Ensure the certificate path exists, is accessible, and points to a valid certificate file.", + }, } def __init__(self, code, file=None, original_exception=None, message=None): @@ -357,3 +369,24 @@ class M365ExchangeConnectionError(M365CredentialsError): super().__init__( 6029, file=file, original_exception=original_exception, message=message ) + + +class M365CertificateCreationError(M365CredentialsError): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 6030, file=file, original_exception=original_exception, message=message + ) + + +class M365NotValidCertificateContentError(M365CredentialsError): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 6031, file=file, original_exception=original_exception, message=message + ) + + +class M365NotValidCertificatePathError(M365CredentialsError): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 6032, file=file, original_exception=original_exception, message=message + ) diff --git a/prowler/providers/m365/lib/arguments/arguments.py b/prowler/providers/m365/lib/arguments/arguments.py index e88ef7982b..fffbc3e5bb 100644 --- a/prowler/providers/m365/lib/arguments/arguments.py +++ b/prowler/providers/m365/lib/arguments/arguments.py @@ -28,6 +28,17 @@ def init_parser(self): action="store_true", help="Use Azure interactive browser authentication to log in against Microsoft 365", ) + m365_auth_modes_group.add_argument( + "--certificate-auth", + action="store_true", + help="Use Certificate authentication to log in against Microsoft 365", + ) + m365_parser.add_argument( + "--certificate-path", + nargs="?", + default=None, + help="Path to the certificate file to be used with --certificate-auth option", + ) m365_parser.add_argument( "--tenant-id", nargs="?", diff --git a/prowler/providers/m365/lib/powershell/m365_powershell.py b/prowler/providers/m365/lib/powershell/m365_powershell.py index 7692700988..3cd2dda4df 100644 --- a/prowler/providers/m365/lib/powershell/m365_powershell.py +++ b/prowler/providers/m365/lib/powershell/m365_powershell.py @@ -4,6 +4,7 @@ import platform from prowler.lib.logger import logger from prowler.lib.powershell.powershell import PowerShellSession from prowler.providers.m365.exceptions.exceptions import ( + M365CertificateCreationError, M365GraphConnectionError, M365UserCredentialsError, M365UserNotBelongingToTenantError, @@ -51,23 +52,71 @@ class M365PowerShell(PowerShellSession): self.tenant_identity = identity self.init_credential(credentials) + def clean_certificate_content(self, cert_content: str) -> str: + """ + Clean certificate content for PowerShell consumption. + + Removes newlines, carriage returns, and extra spaces from base64 content + to ensure proper parsing in PowerShell. + + Args: + cert_content (str): Base64 encoded certificate content + + Returns: + str: Cleaned base64 certificate content + """ + # Clean base64 content - remove any newlines or whitespace + clean_content = ( + cert_content.strip().replace("\n", "").replace("\r", "").replace(" ", "") + ) + logger.info(f"Cleaned certificate content length: {len(clean_content)}") + return clean_content + def init_credential(self, credentials: M365Credentials) -> None: """ Initialize PowerShell credential object for Microsoft 365 authentication. - Sanitizes the username and password, then creates a PSCredential object - in the PowerShell session for use with Microsoft 365 cmdlets. + Supports three authentication methods: + 1. User authentication (username/password) - Will be deprecated in September 2025 + 2. Application authentication (client_id/client_secret) + 3. Certificate authentication (certificate_content in base64/application_id) Args: credentials (M365Credentials): The credentials object containing - username and password. + authentication information. Note: The credentials are sanitized to prevent command injection and stored securely in the PowerShell session. """ + # Certificate Auth + if credentials.certificate_content and credentials.client_id: + # Clean certificate content for PowerShell consumption + clean_cert_content = self.clean_certificate_content( + credentials.certificate_content + ) + + # Sanitize credentials + sanitized_client_id = self.sanitize(credentials.client_id) + sanitized_tenant_id = self.sanitize(credentials.tenant_id) + + self.execute( + f'$certBytes = [Convert]::FromBase64String("{clean_cert_content}")' + ) + error = self.execute( + "$certificate = New-Object System.Security.Cryptography.X509Certificates.X509Certificate2(,$certBytes)" + ) + if error: + raise M365CertificateCreationError( + f"[{os.path.basename(__file__)}] Error creating certificate: {error}" + ) + + self.execute(f'$clientID = "{sanitized_client_id}"') + self.execute(f'$tenantID = "{sanitized_tenant_id}"') + self.execute(f'$tenantDomain = "{credentials.tenant_domains[0]}"') + # User Auth (Will be deprecated in September 2025) - if credentials.user and credentials.passwd: + elif credentials.user and credentials.passwd: credentials.encrypted_passwd = self.encrypt_password(credentials.passwd) # Sanitize user and password @@ -135,14 +184,28 @@ class M365PowerShell(PowerShellSession): """ Test Microsoft 365 credentials by attempting to authenticate against Entra ID. + Supports testing three authentication methods: + 1. User authentication (username/password) + 2. Application authentication (client_id/client_secret) + 3. Certificate authentication (certificate_content in base64/application_id) + Args: credentials (M365Credentials): The credentials object containing - username and password to test. + authentication information to test. Returns: bool: True if credentials are valid and authentication succeeds, False otherwise. """ - if credentials.user and credentials.passwd: + # Test Certificate Auth + if credentials.certificate_content and credentials.client_id: + try: + self.test_teams_certificate_connection() or self.test_exchange_certificate_connection() + return True + except Exception as e: + logger.error(f"Exchange Online Certificate connection failed: {e}") + + # Test User Auth + elif credentials.user and credentials.passwd: self.execute( f'$securePassword = "{credentials.encrypted_passwd}" | ConvertTo-SecureString' # encrypted password already sanitized ) @@ -161,6 +224,7 @@ class M365PowerShell(PowerShellSession): ) # Validate credentials + # Test Exchange Online connection 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 @@ -168,21 +232,19 @@ class M365PowerShell(PowerShellSession): 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}", - ) + # Test Microsoft Teams connection + result = self.execute("Connect-MicrosoftTeams -Credential $credential") + if self.tenant_identity.user 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 M365UserCredentialsError( + file=os.path.basename(__file__), + message=f"Error connecting to PowerShell modules: {result if result else 'Unknown error'}", + ) return True @@ -235,6 +297,9 @@ class M365PowerShell(PowerShellSession): "Microsoft Teams connection failed: Please check your permissions and try again." ) return False + self.execute( + 'Connect-MicrosoftTeams -AccessTokens @("$graphToken","$teamsToken")' + ) return True except Exception as e: logger.error( @@ -242,6 +307,31 @@ class M365PowerShell(PowerShellSession): ) return False + def test_teams_certificate_connection(self) -> bool: + """Test Microsoft Teams API connection using certificate and raise exception if it fails.""" + result = self.execute( + "Connect-MicrosoftTeams -Certificate $certificate -ApplicationId $clientID -TenantId $tenantID" + ) + if self.tenant_identity.identity_id not in result: + logger.error(f"Microsoft Teams Certificate connection failed: {result}") + return False + return True + + def test_teams_user_connection(self) -> bool: + """Test Microsoft Teams API connection using user authentication and raise exception if it fails.""" + result = self.execute("Connect-MicrosoftTeams -Credential $credential") + if self.tenant_identity.user not in result: + logger.error(f"Microsoft Teams User Auth connection failed: {result}.") + return False + + connection = self.execute("Get-CsTeamsClientConfiguration") + if not connection: + logger.error( + "Microsoft Teams User Auth connection failed: Please check your permissions and try again." + ) + return False + return True + def test_exchange_connection(self) -> bool: """Test Exchange Online API connection and raise exception if it fails.""" try: @@ -258,6 +348,9 @@ class M365PowerShell(PowerShellSession): "Exchange Online connection failed: Please check your permissions and try again." ) return False + self.execute( + 'Connect-ExchangeOnline -AccessToken $exchangeToken.AccessToken -Organization "$tenantID"' + ) return True except Exception as e: logger.error( @@ -265,11 +358,40 @@ class M365PowerShell(PowerShellSession): ) return False + def test_exchange_certificate_connection(self) -> bool: + """Test Exchange Online API connection using certificate and raise exception if it fails.""" + result = self.execute( + "Connect-ExchangeOnline -Certificate $certificate -AppId $clientID -Organization $tenantDomain" + ) + if "https://aka.ms/exov3-module" not in result: + logger.error(f"Exchange Online Certificate connection failed: {result}") + return False + return True + + def test_exchange_user_connection(self) -> bool: + """Test Exchange Online API connection using user authentication and raise exception if it fails.""" + result = self.execute("Connect-ExchangeOnline -Credential $credential") + if "https://aka.ms/exov3-module" not in result: + logger.error(f"Exchange Online User Auth connection failed: {result}.") + return False + + connection = self.execute("Get-OrganizationConfig") + if not connection: + logger.error( + "Exchange Online User Auth connection failed: Please check your permissions and try again." + ) + return False + return True + def connect_microsoft_teams(self) -> dict: """ Connect to Microsoft Teams Module PowerShell Module. Establishes a connection to Microsoft Teams using the initialized credentials. + Supports three authentication methods: + 1. User authentication (username/password) + 2. Application authentication (client_id/client_secret) + 3. Certificate authentication (certificate_content in base64/application_id) Returns: dict: Connection status information in JSON format. @@ -277,26 +399,15 @@ class M365PowerShell(PowerShellSession): Note: This method requires the Microsoft Teams PowerShell module to be installed. """ + # Certificate Auth + if self.execute("Write-Output $certificate") != "": + return self.test_teams_certificate_connection() # 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 User Auth connection failed: Please check your permissions and try again." - ) - return connection + return self.test_teams_user_connection() # Application Auth else: - connection = self.test_teams_connection() - if connection: - self.execute( - 'Connect-MicrosoftTeams -AccessTokens @("$graphToken","$teamsToken")' - ) - return connection + return self.test_teams_connection() def get_teams_settings(self) -> dict: """ @@ -383,6 +494,10 @@ class M365PowerShell(PowerShellSession): Connect to Exchange Online PowerShell Module. Establishes a connection to Exchange Online using the initialized credentials. + Supports three authentication methods: + 1. User authentication (username/password) + 2. Application authentication (client_id/client_secret) + 3. Certificate authentication (certificate_content in base64/application_id) Returns: dict: Connection status information in JSON format. @@ -390,25 +505,15 @@ class M365PowerShell(PowerShellSession): Note: This method requires the Exchange Online PowerShell module to be installed. """ + # Certificate Auth + if self.execute("Write-Output $certificate") != "": + return self.test_exchange_certificate_connection() # 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 User Auth connection failed: Please check your permissions and try again." - ) - return False + return self.test_exchange_user_connection() # Application Auth else: - connection = self.test_exchange_connection() - if connection: - self.execute( - 'Connect-ExchangeOnline -AccessToken $exchangeToken.AccessToken -Organization "$tenantID"' - ) - return connection + return self.test_exchange_connection() def get_audit_log_config(self) -> dict: """ diff --git a/prowler/providers/m365/m365_provider.py b/prowler/providers/m365/m365_provider.py index 1910e5de91..4be03f5a70 100644 --- a/prowler/providers/m365/m365_provider.py +++ b/prowler/providers/m365/m365_provider.py @@ -1,4 +1,5 @@ import asyncio +import base64 import os from argparse import ArgumentTypeError from os import getenv @@ -6,6 +7,7 @@ from uuid import UUID from azure.core.exceptions import ClientAuthenticationError, HttpResponseError from azure.identity import ( + CertificateCredential, ClientSecretCredential, CredentialUnavailableError, DefaultAzureCredential, @@ -41,6 +43,8 @@ from prowler.providers.m365.exceptions.exceptions import ( M365MissingEnvironmentCredentialsError, M365NoAuthenticationMethodError, M365NotTenantIdButClientIdAndClientSecretError, + M365NotValidCertificateContentError, + M365NotValidCertificatePathError, M365NotValidClientIdError, M365NotValidClientSecretError, M365NotValidTenantIdError, @@ -111,11 +115,14 @@ class M365Provider(Provider): env_auth: bool = False, az_cli_auth: bool = False, browser_auth: bool = False, + certificate_auth: bool = False, tenant_id: str = None, client_id: str = None, client_secret: str = None, user: str = None, password: str = None, + certificate_content: str = None, + certificate_path: str = None, init_modules: bool = False, region: str = "M365Global", config_content: dict = None, @@ -158,11 +165,14 @@ class M365Provider(Provider): sp_env_auth, env_auth, browser_auth, + certificate_auth, tenant_id, client_id, client_secret, user, password, + certificate_content, + certificate_path, ) logger.info("Checking if region is different than default one") @@ -170,13 +180,15 @@ class M365Provider(Provider): # Get the dict from the static credentials m365_credentials = None - if tenant_id and client_id and client_secret: + if tenant_id and client_id: m365_credentials = self.validate_static_credentials( tenant_id=tenant_id, client_id=client_id, client_secret=client_secret, user=user, password=password, + certificate_content=certificate_content, + certificate_path=certificate_path, ) # Set up the M365 session @@ -185,6 +197,8 @@ class M365Provider(Provider): sp_env_auth, env_auth, browser_auth, + certificate_auth, + certificate_path, tenant_id, m365_credentials, self._region_config, @@ -196,6 +210,7 @@ class M365Provider(Provider): env_auth, browser_auth, az_cli_auth, + certificate_auth, self._session, ) @@ -203,6 +218,8 @@ class M365Provider(Provider): self._credentials = self.setup_powershell( env_auth=env_auth, sp_env_auth=sp_env_auth, + certificate_auth=certificate_auth, + certificate_path=certificate_path, m365_credentials=m365_credentials, identity=self.identity, init_modules=init_modules, @@ -279,11 +296,14 @@ class M365Provider(Provider): sp_env_auth: bool, env_auth: bool, browser_auth: bool, + certificate_auth: bool, tenant_id: str, client_id: str, client_secret: str, user: str, password: str, + certificate_content: str, + certificate_path: str, ): """ Validates the authentication arguments for the M365 provider. @@ -293,11 +313,14 @@ class M365Provider(Provider): sp_env_auth (bool): Flag indicating whether application authentication with environment variables is enabled. env_auth: (bool): Flag indicating whether to use application and PowerShell authentication with environment variables. browser_auth (bool): Flag indicating whether browser authentication is enabled. + certificate_auth (bool): Flag indicating whether certificate authentication is enabled. tenant_id (str): The M365 Tenant ID. client_id (str): The M365 Client ID. client_secret (str): The M365 Client Secret. user (str): The M365 User Account. password (str): The M365 User Password. + certificate_content (str): The M365 Certificate Content. + certificate_path (str): The path to the certificate file. Raises: M365BrowserAuthNoTenantIDError: If browser authentication is enabled but the tenant ID is not found. @@ -314,28 +337,33 @@ class M365Provider(Provider): and not sp_env_auth and not browser_auth and not env_auth + and not certificate_auth ): raise M365NoAuthenticationMethodError( file=os.path.basename(__file__), - message="M365 provider requires at least one authentication method set: [--env-auth | --az-cli-auth | --sp-env-auth | --browser-auth]", + message="M365 provider requires at least one authentication method set: [--env-auth | --az-cli-auth | --sp-env-auth | --browser-auth | --certificate-auth]", ) elif browser_auth and not tenant_id: raise M365BrowserAuthNoTenantIDError( file=os.path.basename(__file__), message="M365 Tenant ID (--tenant-id) is required for browser authentication mode", ) - elif env_auth: - if not user or not password or not tenant_id: - raise M365MissingEnvironmentCredentialsError( - file=os.path.basename(__file__), - message="M365 provider requires AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID, M365_USER and M365_PASSWORD environment variables to be set when using --env-auth", - ) else: if not tenant_id: raise M365NotTenantIdButClientIdAndClientSecretError( file=os.path.basename(__file__), message="Tenant Id is required for M365 static credentials. Make sure you are using the correct credentials.", ) + if ( + not certificate_content + and not certificate_path + and not (user and password) + and not client_secret + ): + raise M365ConfigCredentialsError( + file=os.path.basename(__file__), + message="You must provide a valid set of credentials. Please check your credentials and try again.", + ) @staticmethod def setup_region_config(region): @@ -378,6 +406,8 @@ class M365Provider(Provider): def setup_powershell( env_auth: bool = False, sp_env_auth: bool = False, + certificate_auth: bool = False, + certificate_path: str = None, m365_credentials: dict = {}, identity: M365IdentityInfo = None, init_modules: bool = False, @@ -402,6 +432,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", ""), + certificate_content=m365_credentials.get("certificate_content", ""), tenant_domains=identity.tenant_domains, ) elif env_auth: @@ -440,10 +471,28 @@ class M365Provider(Provider): tenant_domains=identity.tenant_domains, ) + elif certificate_auth: + client_id = getenv("AZURE_CLIENT_ID") + tenant_id = getenv("AZURE_TENANT_ID") + if certificate_path: + with open(certificate_path, "rb") as cert_file: + # Encode the certificate content to base64 since PowerShell expects a base64 string + certificate_content = base64.b64encode(cert_file.read()) + else: + certificate_content = getenv("M365_CERTIFICATE_CONTENT") + credentials = M365Credentials( + client_id=client_id, + tenant_id=tenant_id, + certificate_content=certificate_content, + tenant_domains=identity.tenant_domains, + ) + if credentials: if identity and credentials.user: identity.user = credentials.user identity.identity_type = "Service Principal and User Credentials" + if identity and credentials.certificate_content: + identity.identity_type = "Service Principal with Certificate" test_session = M365PowerShell(credentials, identity) try: if init_modules: @@ -478,6 +527,10 @@ class M365Provider(Provider): report_lines.append( f"M365 User: {Fore.YELLOW}{self.credentials.user}{Style.RESET_ALL}" ) + elif self.credentials and self.credentials.certificate_content: + report_lines.append( + f"M365 Certificate Thumbprint: {Fore.YELLOW}{self._identity.certificate_thumbprint}{Style.RESET_ALL}" + ) report_title = ( f"{Style.BRIGHT}Using the M365 credentials below:{Style.RESET_ALL}" ) @@ -491,6 +544,8 @@ class M365Provider(Provider): sp_env_auth: bool, env_auth: bool, browser_auth: bool, + certificate_auth: bool, + certificate_path: str, tenant_id: str, m365_credentials: dict, region_config: M365RegionConfig, @@ -510,6 +565,8 @@ class M365Provider(Provider): - client_secret: The M365 client secret - user: The M365 user email - password: The M365 user password + - certificate_content: The M365 certificate content + - certificate_path: The path to the certificate file. - provider_id: The M365 provider ID (in this case the Tenant ID). region_config (M365RegionConfig): The region configuration object. @@ -530,16 +587,45 @@ class M365Provider(Provider): f"{environment_credentials_error.__class__.__name__}[{environment_credentials_error.__traceback__.tb_lineno}] -- {environment_credentials_error}" ) raise environment_credentials_error + elif certificate_auth: + try: + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=not certificate_path + ) + except M365EnvironmentVariableError as environment_variable_error: + logger.critical( + f"{environment_variable_error.__class__.__name__}[{environment_variable_error.__traceback__.tb_lineno}] -- {environment_variable_error}" + ) + raise environment_variable_error try: if m365_credentials: try: - credentials = ClientSecretCredential( - tenant_id=m365_credentials["tenant_id"], - client_id=m365_credentials["client_id"], - client_secret=m365_credentials["client_secret"], - user=m365_credentials["user"], - password=m365_credentials["password"], - ) + if m365_credentials["certificate_content"]: + credentials = CertificateCredential( + tenant_id=m365_credentials["tenant_id"], + client_id=m365_credentials["client_id"], + certificate_data=base64.b64decode( + m365_credentials["certificate_content"] + ), + ) + elif m365_credentials["certificate_path"]: + with open( + m365_credentials["certificate_path"], "rb" + ) as cert_file: + certificate_data = cert_file.read() + credentials = CertificateCredential( + tenant_id=m365_credentials["tenant_id"], + client_id=m365_credentials["client_id"], + certificate_data=certificate_data, + ) + else: + credentials = ClientSecretCredential( + tenant_id=m365_credentials["tenant_id"], + client_id=m365_credentials["client_id"], + client_secret=m365_credentials["client_secret"], + user=m365_credentials["user"], + password=m365_credentials["password"], + ) return credentials except ClientAuthenticationError as error: logger.error( @@ -562,13 +648,34 @@ class M365Provider(Provider): raise M365ConfigCredentialsError( file=os.path.basename(__file__), original_exception=error ) + elif certificate_auth: + try: + if certificate_path: + with open(certificate_path, "rb") as cert_file: + certificate_data = cert_file.read() + else: + certificate_data = base64.b64decode( + getenv("M365_CERTIFICATE_CONTENT") + ) + credentials = CertificateCredential( + tenant_id=getenv("AZURE_TENANT_ID"), + client_id=getenv("AZURE_CLIENT_ID"), + certificate_data=certificate_data, + ) + except ClientAuthenticationError as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}" + ) + raise M365ClientAuthenticationError( + file=os.path.basename(__file__), original_exception=error + ) else: # Since the authentication method to be used will come as True, we have to negate it since # DefaultAzureCredential sets just one authentication method, excluding the others try: credentials = DefaultAzureCredential( exclude_environment_credential=not ( - sp_env_auth or env_auth + sp_env_auth or env_auth or certificate_auth ), exclude_cli_credential=not az_cli_auth, # M365 Auth using Managed Identity is not supported @@ -633,6 +740,7 @@ class M365Provider(Provider): sp_env_auth: bool = False, env_auth: bool = False, browser_auth: bool = False, + certificate_auth: bool = False, tenant_id: str = None, region: str = "M365Global", raise_on_exception: bool = True, @@ -640,6 +748,8 @@ class M365Provider(Provider): client_secret: str = None, user: str = None, password: str = None, + certificate_content: str = None, + certificate_path: str = None, provider_id: str = None, ) -> Connection: """Test connection to M365 tenant and PowerShell modules. @@ -652,6 +762,7 @@ class M365Provider(Provider): sp_env_auth (bool): Flag indicating whether to use application authentication with environment variables. env_auth: (bool): Flag indicating whether to use application and PowerShell authentication with environment variables. browser_auth (bool): Flag indicating whether to use interactive browser authentication. + certificate_auth (bool): Flag indicating whether to use certificate authentication. tenant_id (str): The M365 Active Directory tenant ID. region (str): The M365 region. raise_on_exception (bool): Flag indicating whether to raise an exception if the connection fails. @@ -688,11 +799,14 @@ class M365Provider(Provider): sp_env_auth, env_auth, browser_auth, + certificate_auth, tenant_id, client_id, client_secret, user, password, + certificate_content, + certificate_path, ) region_config = M365Provider.setup_region_config(region) @@ -704,8 +818,6 @@ class M365Provider(Provider): tenant_id=tenant_id, client_id=client_id, client_secret=client_secret, - user=None, - password=None, ) else: m365_credentials = M365Provider.validate_static_credentials( @@ -715,6 +827,12 @@ class M365Provider(Provider): user=user, password=password, ) + elif tenant_id and client_id and certificate_content: + m365_credentials = M365Provider.validate_static_credentials( + tenant_id=tenant_id, + client_id=client_id, + certificate_content=certificate_content, + ) # Set up the M365 session session = M365Provider.setup_session( @@ -722,6 +840,8 @@ class M365Provider(Provider): sp_env_auth, env_auth, browser_auth, + certificate_auth, + certificate_path, tenant_id, m365_credentials, region_config, @@ -737,6 +857,7 @@ class M365Provider(Provider): env_auth, browser_auth, az_cli_auth, + certificate_auth, session, ) @@ -758,6 +879,8 @@ class M365Provider(Provider): M365Provider.setup_powershell( env_auth, sp_env_auth, + certificate_auth, + certificate_path, m365_credentials, identity, ) @@ -889,12 +1012,44 @@ class M365Provider(Provider): message=f"Missing environment variable {env_var} required to authenticate.", ) + @staticmethod + def check_certificate_creds_env_vars(check_certificate_content: bool): + """ + Checks the presence of required environment variables for service principal authentication against Azure. + + This method checks for the presence of the following environment variables: + - AZURE_CLIENT_ID: Azure client ID + - AZURE_TENANT_ID: Azure tenant ID + - M365_CERTIFICATE_CONTENT: Azure certificate content + + If any of the environment variables is missing, it logs a critical error and exits the program. + """ + logger.info( + "M365 provider: checking service principal environment variables ..." + ) + env_vars = [ + "AZURE_CLIENT_ID", + "AZURE_TENANT_ID", + ] + if check_certificate_content: + env_vars.append("M365_CERTIFICATE_CONTENT") + for env_var in env_vars: + if not getenv(env_var): + logger.critical( + f"M365 provider: Missing environment variable {env_var} needed to authenticate against M365." + ) + raise M365EnvironmentVariableError( + file=os.path.basename(__file__), + message=f"Missing environment variable {env_var} required to authenticate.", + ) + @staticmethod def setup_identity( sp_env_auth, env_auth, browser_auth, az_cli_auth, + certificate_auth, session, ): """ @@ -968,6 +1123,16 @@ class M365Provider(Provider): or session.credentials[0]._credential.client_id or "Unknown user id (Missing AAD permissions)" ) + elif certificate_auth: + identity.identity_type = "Service Principal with Certificate" + identity.identity_id = ( + getenv("AZURE_CLIENT_ID") + or session.credentials[0]._credential.client_id + or "Unknown user id (Missing AAD permissions)" + ) + identity.certificate_thumbprint = session._client_credential.get( + "thumbprint", "Unknown certificate thumbprint" + ) elif browser_auth or az_cli_auth: identity.identity_type = "User" try: @@ -987,8 +1152,14 @@ class M365Provider(Provider): ) else: # Static Credentials - identity.identity_type = "Service Principal" identity.identity_id = session._client_id + if isinstance(session, CertificateCredential): + identity.identity_type = "Service Principal with Certificate" + identity.certificate_thumbprint = session._client_credential.get( + "thumbprint", "Unknown certificate thumbprint" + ) + else: + identity.identity_type = "Service Principal" # Retrieve tenant id from the client client = GraphServiceClient(credentials=session) @@ -1005,6 +1176,8 @@ class M365Provider(Provider): client_secret: str = None, user: str = None, password: str = None, + certificate_content: str = None, + certificate_path: str = None, ) -> dict: """ Validates the static credentials for the M365 provider. @@ -1015,6 +1188,8 @@ class M365Provider(Provider): client_secret (str): The M365 client secret. user (str): The M365 user email. password (str): The M365 user password. + certificate_content (str): The M365 Certificate Content. + certificate_path (str): The path to the certificate file. Raises: M365NotValidTenantIdError: If the provided M365 Tenant ID is not valid. @@ -1045,21 +1220,47 @@ class M365Provider(Provider): message="The provided Client ID is not valid.", ) - # Validate the Client Secret - if not client_secret: + if not certificate_content and not certificate_path and not client_secret: raise M365NotValidClientSecretError( file=os.path.basename(__file__), - message="The provided Client Secret is not valid.", + message="You must provide a client secret, certificate content or certificate path. Please check your credentials and try again.", ) + if certificate_content: + try: + # Validate that certificate content can be properly decoded from base64 + base64.b64decode(certificate_content) + except Exception as e: + raise M365NotValidCertificateContentError( + file=os.path.basename(__file__), + message=f"The provided certificate content is not valid base64 encoded data: {str(e)}", + ) + if certificate_path: + try: + with open(certificate_path, "rb") as cert_file: + certificate_content = cert_file.read() + except Exception as e: + raise M365NotValidCertificatePathError( + file=os.path.basename(__file__), + message=f"The provided certificate path is not valid: {str(e)}", + ) + try: - M365Provider.verify_client(tenant_id, client_id, client_secret) + M365Provider.verify_client( + tenant_id, + client_id, + client_secret, + certificate_content, + certificate_path, + ) return { "tenant_id": tenant_id, "client_id": client_id, "client_secret": client_secret, "user": user, "password": password, + "certificate_content": certificate_content, + "certificate_path": certificate_path, } except M365NotValidTenantIdError as tenant_id_error: logger.error( @@ -1087,7 +1288,9 @@ class M365Provider(Provider): ) @staticmethod - def verify_client(tenant_id, client_id, client_secret) -> None: + def verify_client( + tenant_id, client_id, client_secret, certificate_content, certificate_path + ) -> None: """ Verifies the M365 client credentials using the specified tenant ID, client ID, and client secret. @@ -1095,52 +1298,106 @@ class M365Provider(Provider): tenant_id (str): The M365 Active Directory tenant ID. client_id (str): The M365 client ID. client_secret (str): The M365 client secret. + certificate_content (str): The M365 certificate content. + certificate_path (str): The path to the certificate file. Raises: M365NotValidTenantIdError: If the provided M365 Tenant ID is not valid. M365NotValidClientIdError: If the provided M365 Client ID is not valid. M365NotValidClientSecretError: If the provided M365 Client Secret is not valid. + M365NotValidCertificateContentError: If the provided M365 Certificate Content is not valid. + M365NotValidCertificatePathError: If the provided M365 Certificate Path is not valid. Returns: None """ authority = f"https://login.microsoftonline.com/{tenant_id}" try: - # Create a ConfidentialClientApplication instance - app = ConfidentialClientApplication( - client_id=client_id, - client_credential=client_secret, - authority=authority, - ) + if client_secret: + # Create a ConfidentialClientApplication instance + app = ConfidentialClientApplication( + client_id=client_id, + client_credential=client_secret, + authority=authority, + ) + # Attempt to acquire a token + result = app.acquire_token_for_client( + scopes=["https://graph.microsoft.com/.default"] + ) - # Attempt to acquire a token - result = app.acquire_token_for_client( - scopes=["https://graph.microsoft.com/.default"] - ) + # Check if token acquisition was successful + if "access_token" not in result: + # Handle specific errors based on the MSAL response + error_description = result.get("error_description", "") + if f"Tenant '{tenant_id}'" in error_description: + raise M365NotValidTenantIdError( + file=os.path.basename(__file__), + message="The provided Tenant ID is not valid for the specified Client ID and Client Secret.", + ) + if ( + f"Application with identifier '{client_id}'" + in error_description + ): + raise M365NotValidClientIdError( + file=os.path.basename(__file__), + message="The provided Client ID is not valid for the specified Tenant ID and Client Secret.", + ) + if "Invalid client secret provided" in error_description: + raise M365NotValidClientSecretError( + file=os.path.basename(__file__), + message="The provided Client Secret is not valid for the specified Tenant ID and Client ID.", + ) + elif certificate_content: + credential = CertificateCredential( + client_id=client_id, + tenant_id=tenant_id, + certificate_data=base64.b64decode(certificate_content), + ) + client = GraphServiceClient(credentials=credential) - # Check if token acquisition was successful - if "access_token" not in result: - # Handle specific errors based on the MSAL response - error_description = result.get("error_description", "") - if f"Tenant '{tenant_id}'" in error_description: - raise M365NotValidTenantIdError( + # Verify that the certificate is valid + async def verify_certificate(): + result = await client.domains.get() + return result.value + + result = asyncio.get_event_loop().run_until_complete( + verify_certificate() + ) + if not result: + raise M365NotValidCertificateContentError( file=os.path.basename(__file__), - message="The provided Tenant ID is not valid for the specified Client ID and Client Secret.", + message="The provided certificate content is not valid.", ) - if f"Application with identifier '{client_id}'" in error_description: - raise M365NotValidClientIdError( + elif certificate_path: + with open(certificate_path, "rb") as cert_file: + certificate_content = cert_file.read() + credential = CertificateCredential( + client_id=client_id, + tenant_id=tenant_id, + certificate_data=certificate_content, + ) + client = GraphServiceClient(credentials=credential) + + # Verify that the certificate is valid + async def verify_certificate(): + result = await client.domains.get() + return result.value + + result = asyncio.get_event_loop().run_until_complete( + verify_certificate() + ) + if not result: + raise M365NotValidCertificatePathError( file=os.path.basename(__file__), - message="The provided Client ID is not valid for the specified Tenant ID and Client Secret.", - ) - if "Invalid client secret provided" in error_description: - raise M365NotValidClientSecretError( - file=os.path.basename(__file__), - message="The provided Client Secret is not valid for the specified Tenant ID and Client ID.", + message="The provided certificate is not valid.", ) + except ( M365NotValidTenantIdError, M365NotValidClientIdError, M365NotValidClientSecretError, + M365NotValidCertificateContentError, + M365NotValidCertificatePathError, ) as m365_error: # M365 specific errors already raised raise RuntimeError(f"{m365_error}") diff --git a/prowler/providers/m365/models.py b/prowler/providers/m365/models.py index ff2cec2fd5..3459b31688 100644 --- a/prowler/providers/m365/models.py +++ b/prowler/providers/m365/models.py @@ -11,6 +11,7 @@ class M365IdentityInfo(BaseModel): identity_type: str = "" tenant_id: str = "" tenant_domain: str = "Unknown tenant domain (missing Entra permissions)" + certificate_thumbprint: str = "" tenant_domains: list[str] = [] location: str = "" user: str = None @@ -28,9 +29,10 @@ class M365Credentials(BaseModel): passwd: Optional[str] = None encrypted_passwd: Optional[str] = None client_id: str = "" - client_secret: str = "" + client_secret: Optional[str] = None tenant_id: str = "" tenant_domains: list[str] = [] + certificate_content: Optional[str] = None class M365OutputOptions(ProviderOutputOptions): diff --git a/tests/providers/m365/lib/arguments/m365_arguments_test.py b/tests/providers/m365/lib/arguments/m365_arguments_test.py new file mode 100644 index 0000000000..220ff40f8f --- /dev/null +++ b/tests/providers/m365/lib/arguments/m365_arguments_test.py @@ -0,0 +1,641 @@ +import argparse +from unittest.mock import MagicMock + +from prowler.providers.m365.lib.arguments import arguments + + +class TestM365Arguments: + def setup_method(self): + """Setup mock ArgumentParser for testing""" + self.mock_parser = MagicMock() + self.mock_subparsers = MagicMock() + self.mock_m365_parser = MagicMock() + self.mock_auth_group = MagicMock() + self.mock_auth_modes_group = MagicMock() + self.mock_regions_group = MagicMock() + + # Setup the mock chain + self.mock_parser.add_subparsers.return_value = self.mock_subparsers + self.mock_subparsers.add_parser.return_value = self.mock_m365_parser + self.mock_m365_parser.add_argument_group.side_effect = [ + self.mock_auth_group, + self.mock_regions_group, + ] + self.mock_auth_group.add_mutually_exclusive_group.return_value = ( + self.mock_auth_modes_group + ) + + def test_init_parser_creates_subparser(self): + """Test that init_parser creates the M365 subparser correctly""" + # Create a mock object that has the necessary attributes + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + # Call init_parser + arguments.init_parser(mock_m365_args) + + # Verify subparser was created + self.mock_subparsers.add_parser.assert_called_once_with( + "m365", + parents=[mock_m365_args.common_providers_parser], + help="M365 Provider", + ) + + def test_init_parser_creates_argument_groups(self): + """Test that init_parser creates the correct argument groups""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Verify argument groups were created + assert self.mock_m365_parser.add_argument_group.call_count == 2 + calls = self.mock_m365_parser.add_argument_group.call_args_list + assert calls[0][0][0] == "Authentication Modes" + assert calls[1][0][0] == "Regions" + + def test_init_parser_creates_mutually_exclusive_auth_group(self): + """Test that init_parser creates mutually exclusive authentication group""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Verify mutually exclusive group was created for authentication modes + self.mock_auth_group.add_mutually_exclusive_group.assert_called_once() + + def test_init_parser_adds_authentication_arguments(self): + """Test that init_parser adds all authentication arguments""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Verify authentication arguments were added to the mutually exclusive group + assert self.mock_auth_modes_group.add_argument.call_count == 5 + + # Check that all authentication arguments are present + calls = self.mock_auth_modes_group.add_argument.call_args_list + auth_args = [call[0][0] for call in calls] + + assert "--az-cli-auth" in auth_args + assert "--env-auth" in auth_args + assert "--sp-env-auth" in auth_args + assert "--browser-auth" in auth_args + assert "--certificate-auth" in auth_args + + def test_init_parser_adds_non_exclusive_arguments(self): + """Test that init_parser adds non-exclusive arguments directly to parser""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Verify non-exclusive arguments were added to main parser + assert self.mock_m365_parser.add_argument.call_count == 3 + + # Check that non-exclusive arguments are present + calls = self.mock_m365_parser.add_argument.call_args_list + non_exclusive_args = [call[0][0] for call in calls] + + assert "--tenant-id" in non_exclusive_args + assert "--init-modules" in non_exclusive_args + assert "--certificate-path" in non_exclusive_args + + def test_init_parser_adds_region_arguments(self): + """Test that init_parser adds region arguments""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Verify region arguments were added to regions group + assert self.mock_regions_group.add_argument.call_count == 1 + + # Check that region argument is present + calls = self.mock_regions_group.add_argument.call_args_list + region_args = [call[0][0] for call in calls] + + assert "--region" in region_args + + def test_az_cli_auth_argument_configuration(self): + """Test that az-cli-auth argument is configured correctly""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Find the az-cli-auth argument call + calls = self.mock_auth_modes_group.add_argument.call_args_list + az_cli_call = None + for call in calls: + if call[0][0] == "--az-cli-auth": + az_cli_call = call + break + + assert az_cli_call is not None + + # Check argument configuration + kwargs = az_cli_call[1] + assert kwargs["action"] == "store_true" + assert "Azure CLI authentication" in kwargs["help"] + + def test_env_auth_argument_configuration(self): + """Test that env-auth argument is configured correctly""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Find the env-auth argument call + calls = self.mock_auth_modes_group.add_argument.call_args_list + env_auth_call = None + for call in calls: + if call[0][0] == "--env-auth": + env_auth_call = call + break + + assert env_auth_call is not None + + # Check argument configuration + kwargs = env_auth_call[1] + assert kwargs["action"] == "store_true" + assert "User and Password environment variables" in kwargs["help"] + + def test_sp_env_auth_argument_configuration(self): + """Test that sp-env-auth argument is configured correctly""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Find the sp-env-auth argument call + calls = self.mock_auth_modes_group.add_argument.call_args_list + sp_env_call = None + for call in calls: + if call[0][0] == "--sp-env-auth": + sp_env_call = call + break + + assert sp_env_call is not None + + # Check argument configuration + kwargs = sp_env_call[1] + assert kwargs["action"] == "store_true" + assert "Azure Service Principal environment variables" in kwargs["help"] + + def test_browser_auth_argument_configuration(self): + """Test that browser-auth argument is configured correctly""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Find the browser-auth argument call + calls = self.mock_auth_modes_group.add_argument.call_args_list + browser_auth_call = None + for call in calls: + if call[0][0] == "--browser-auth": + browser_auth_call = call + break + + assert browser_auth_call is not None + + # Check argument configuration + kwargs = browser_auth_call[1] + assert kwargs["action"] == "store_true" + assert "Azure interactive browser authentication" in kwargs["help"] + + def test_certificate_auth_argument_configuration(self): + """Test that certificate-auth argument is configured correctly""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Find the certificate-auth argument call + calls = self.mock_auth_modes_group.add_argument.call_args_list + cert_auth_call = None + for call in calls: + if call[0][0] == "--certificate-auth": + cert_auth_call = call + break + + assert cert_auth_call is not None + + # Check argument configuration + kwargs = cert_auth_call[1] + assert kwargs["action"] == "store_true" + assert "Certificate authentication" in kwargs["help"] + + def test_tenant_id_argument_configuration(self): + """Test that tenant-id argument is configured correctly""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Find the tenant-id argument call + calls = self.mock_m365_parser.add_argument.call_args_list + tenant_id_call = None + for call in calls: + if call[0][0] == "--tenant-id": + tenant_id_call = call + break + + assert tenant_id_call is not None + + # Check argument configuration + kwargs = tenant_id_call[1] + assert kwargs["nargs"] == "?" + assert kwargs["default"] is None + assert "Microsoft 365 Tenant ID" in kwargs["help"] + assert "--browser-auth" in kwargs["help"] + + def test_init_modules_argument_configuration(self): + """Test that init-modules argument is configured correctly""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Find the init-modules argument call + calls = self.mock_m365_parser.add_argument.call_args_list + init_modules_call = None + for call in calls: + if call[0][0] == "--init-modules": + init_modules_call = call + break + + assert init_modules_call is not None + + # Check argument configuration + kwargs = init_modules_call[1] + assert kwargs["action"] == "store_true" + assert "Initialize Microsoft 365 PowerShell modules" in kwargs["help"] + + def test_region_argument_configuration(self): + """Test that region argument is configured correctly""" + mock_m365_args = MagicMock() + mock_m365_args.subparsers = self.mock_subparsers + mock_m365_args.common_providers_parser = MagicMock() + + arguments.init_parser(mock_m365_args) + + # Find the region argument call + calls = self.mock_regions_group.add_argument.call_args_list + region_call = None + for call in calls: + if call[0][0] == "--region": + region_call = call + break + + assert region_call is not None + + # Check argument configuration + kwargs = region_call[1] + assert kwargs["nargs"] == "?" + assert kwargs["default"] == "M365Global" + assert kwargs["choices"] == [ + "M365Global", + "M365GlobalChina", + "M365USGovernment", + ] + assert "Microsoft 365 region" in kwargs["help"] + assert "M365Global" in kwargs["help"] + + +class TestM365ArgumentsIntegration: + def test_real_argument_parsing_az_cli_auth(self): + """Test parsing arguments with Azure CLI authentication""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + # Create a mock object that mimics the structure used by the init_parser function + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with Azure CLI auth + args = parser.parse_args(["m365", "--az-cli-auth"]) + + assert args.az_cli_auth is True + assert args.env_auth is False + assert args.sp_env_auth is False + assert args.browser_auth is False + assert args.certificate_auth is False + + def test_real_argument_parsing_env_auth(self): + """Test parsing arguments with environment authentication""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with environment auth + args = parser.parse_args(["m365", "--env-auth"]) + + assert args.az_cli_auth is False + assert args.env_auth is True + assert args.sp_env_auth is False + assert args.browser_auth is False + assert args.certificate_auth is False + + def test_real_argument_parsing_sp_env_auth(self): + """Test parsing arguments with service principal environment authentication""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with service principal environment auth + args = parser.parse_args(["m365", "--sp-env-auth"]) + + assert args.az_cli_auth is False + assert args.env_auth is False + assert args.sp_env_auth is True + assert args.browser_auth is False + assert args.certificate_auth is False + + def test_real_argument_parsing_browser_auth_with_tenant_id(self): + """Test parsing arguments with browser authentication and tenant ID""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with browser auth and tenant ID + args = parser.parse_args( + [ + "m365", + "--browser-auth", + "--tenant-id", + "12345678-1234-5678-abcd-123456789012", + ] + ) + + assert args.az_cli_auth is False + assert args.env_auth is False + assert args.sp_env_auth is False + assert args.browser_auth is True + assert args.certificate_auth is False + assert args.tenant_id == "12345678-1234-5678-abcd-123456789012" + + def test_real_argument_parsing_certificate_auth(self): + """Test parsing arguments with certificate authentication""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with certificate auth + args = parser.parse_args(["m365", "--certificate-auth"]) + + assert args.az_cli_auth is False + assert args.env_auth is False + assert args.sp_env_auth is False + assert args.browser_auth is False + assert args.certificate_auth is True + + def test_real_argument_parsing_with_init_modules(self): + """Test parsing arguments with init modules flag""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with init modules + args = parser.parse_args(["m365", "--az-cli-auth", "--init-modules"]) + + assert args.az_cli_auth is True + assert args.init_modules is True + + def test_real_argument_parsing_with_different_regions(self): + """Test parsing arguments with different region options""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Test M365Global (default) + args = parser.parse_args(["m365", "--az-cli-auth"]) + assert args.region == "M365Global" + + # Test M365GlobalChina + args = parser.parse_args( + ["m365", "--az-cli-auth", "--region", "M365GlobalChina"] + ) + assert args.region == "M365GlobalChina" + + # Test M365USGovernment + args = parser.parse_args( + ["m365", "--az-cli-auth", "--region", "M365USGovernment"] + ) + assert args.region == "M365USGovernment" + + def test_real_argument_parsing_no_authentication_defaults(self): + """Test parsing arguments without any authentication flags (should have defaults)""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments without explicit auth (defaults should apply) + args = parser.parse_args(["m365"]) + + assert args.az_cli_auth is False + assert args.env_auth is False + assert args.sp_env_auth is False + assert args.browser_auth is False + assert args.certificate_auth is False + assert args.tenant_id is None + assert args.init_modules is False + assert args.region == "M365Global" + + def test_real_argument_parsing_complete_configuration(self): + """Test parsing arguments with all non-exclusive options""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with complete configuration + args = parser.parse_args( + [ + "m365", + "--browser-auth", + "--tenant-id", + "12345678-1234-5678-abcd-123456789012", + "--init-modules", + "--region", + "M365USGovernment", + ] + ) + + assert args.browser_auth is True + assert args.tenant_id == "12345678-1234-5678-abcd-123456789012" + assert args.init_modules is True + assert args.region == "M365USGovernment" + + def test_mutually_exclusive_authentication_enforcement(self): + """Test that authentication methods are mutually exclusive""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # This should raise SystemExit due to mutually exclusive group + try: + parser.parse_args(["m365", "--az-cli-auth", "--env-auth"]) + assert False, "Expected SystemExit due to mutually exclusive arguments" + except SystemExit: + # This is expected + pass + + def test_tenant_id_without_arguments(self): + """Test that tenant-id can be specified without an argument (optional value)""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with tenant-id but no value (should be None due to nargs="?") + args = parser.parse_args(["m365", "--az-cli-auth", "--tenant-id"]) + + assert args.tenant_id is None + + def test_certificate_path_argument_configuration(self): + """Test that certificate_path argument is properly configured""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with certificate-path + args = parser.parse_args( + ["m365", "--certificate-auth", "--certificate-path", "/path/to/cert.pem"] + ) + + assert args.certificate_auth is True + assert args.certificate_path == "/path/to/cert.pem" + + def test_certificate_path_without_value(self): + """Test certificate_path argument without value (should be None due to nargs='?')""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse arguments with certificate-path but no value + args = parser.parse_args(["m365", "--certificate-auth", "--certificate-path"]) + + assert args.certificate_auth is True + assert args.certificate_path is None + + def test_certificate_auth_with_certificate_path_integration(self): + """Test certificate authentication with certificate path integration""" + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers() + common_parser = argparse.ArgumentParser(add_help=False) + + mock_m365_args = MagicMock() + mock_m365_args.subparsers = subparsers + mock_m365_args.common_providers_parser = common_parser + + arguments.init_parser(mock_m365_args) + + # Parse complete certificate authentication arguments + args = parser.parse_args( + [ + "m365", + "--certificate-auth", + "--certificate-path", + "/home/user/cert.pem", + "--tenant-id", + "12345678-1234-1234-1234-123456789012", + ] + ) + + assert args.certificate_auth is True + assert args.certificate_path == "/home/user/cert.pem" + assert args.tenant_id == "12345678-1234-1234-1234-123456789012" + assert args.az_cli_auth is False + assert args.env_auth is False + assert args.sp_env_auth is False + assert args.browser_auth is False diff --git a/tests/providers/m365/lib/powershell/m365_powershell_test.py b/tests/providers/m365/lib/powershell/m365_powershell_test.py index 2e6a34a032..bd45956de0 100644 --- a/tests/providers/m365/lib/powershell/m365_powershell_test.py +++ b/tests/providers/m365/lib/powershell/m365_powershell_test.py @@ -1,9 +1,11 @@ +import base64 from unittest.mock import MagicMock, call, patch import pytest from prowler.lib.powershell.powershell import PowerShellSession from prowler.providers.m365.exceptions.exceptions import ( + M365CertificateCreationError, M365GraphConnectionError, M365UserCredentialsError, M365UserNotBelongingToTenantError, @@ -112,7 +114,7 @@ class Testm365PowerShell: session.close() @patch("subprocess.Popen") - def test_test_credentials(self, mock_popen): + def test_test_credentials_exchange_success(self, mock_popen): mock_process = MagicMock() mock_popen.return_value = mock_process @@ -131,16 +133,20 @@ class Testm365PowerShell: tenant_domain="contoso.onmicrosoft.com", tenant_domains=["contoso.onmicrosoft.com"], location="test_location", + user="test@contoso.onmicrosoft.com", ) session = M365PowerShell(credentials, identity) # Mock encrypt_password to return a known value session.encrypt_password = MagicMock(return_value="encrypted_password") - # Mock execute to simulate successful Connect-ExchangeOnline - session.execute = MagicMock( - return_value="Connected successfully https://aka.ms/exov3-module" - ) + # Mock execute to simulate successful Exchange connection + def mock_execute_side_effect(command): + if "Connect-ExchangeOnline" in command: + return "Connected successfully https://aka.ms/exov3-module" + return "" + + session.execute = MagicMock(side_effect=mock_execute_side_effect) # Execute the test result = session.test_credentials(credentials) @@ -153,42 +159,110 @@ class Testm365PowerShell: session.execute.assert_any_call( f'$credential = New-Object System.Management.Automation.PSCredential("{session.sanitize(credentials.user)}", $securePassword)' ) + # Exchange connection should be tested session.execute.assert_any_call( "Connect-ExchangeOnline -Credential $credential" ) + # Verify Teams connection was NOT called (since Exchange succeeded) + teams_calls = [ + call + for call in session.execute.call_args_list + if "Connect-MicrosoftTeams" in str(call) + ] + assert ( + len(teams_calls) == 0 + ), "Teams connection should not be called when Exchange succeeds" + + session.close() + + @patch("subprocess.Popen") + def test_test_credentials_exchange_fail_teams_success(self, mock_popen): + mock_process = MagicMock() + mock_popen.return_value = mock_process + + credentials = M365Credentials( + user="test@contoso.onmicrosoft.com", + passwd="test_password", + encrypted_passwd="test_encrypted_password", + client_id="test_client_id", + client_secret="test_client_secret", + tenant_id="test_tenant_id", + ) + 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", + user="test@contoso.onmicrosoft.com", + ) + session = M365PowerShell(credentials, identity) + + # Mock encrypt_password to return a known value + session.encrypt_password = MagicMock(return_value="encrypted_password") + + # Mock execute to simulate Exchange fail and Teams success + def mock_execute_side_effect(command): + if "Connect-ExchangeOnline" in command: + return ( + "Connection failed" # No "https://aka.ms/exov3-module" in response + ) + elif "Connect-MicrosoftTeams" in command: + return "Connected successfully test@contoso.onmicrosoft.com" + return "" + + session.execute = MagicMock(side_effect=mock_execute_side_effect) + + # 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.encrypted_passwd}" | ConvertTo-SecureString' + ) + session.execute.assert_any_call( + f'$credential = New-Object System.Management.Automation.PSCredential("{session.sanitize(credentials.user)}", $securePassword)' + ) + # Both Exchange and Teams connections should be tested + session.execute.assert_any_call( + "Connect-ExchangeOnline -Credential $credential" + ) + session.execute.assert_any_call( + "Connect-MicrosoftTeams -Credential $credential" + ) + session.close() @patch("subprocess.Popen") def test_test_credentials_application_auth(self, mock_popen): mock_process = MagicMock() mock_popen.return_value = mock_process - with patch( - "prowler.providers.m365.lib.powershell.m365_powershell.M365Credentials", - autospec=True, - ) as mock_creds: - credentials = mock_creds.return_value - credentials.user = "" - credentials.passwd = "" - credentials.encrypted_passwd = "" - credentials.client_id = "test_client_id" - credentials.client_secret = "test_client_secret" - credentials.tenant_id = "test_tenant_id" - identity = M365IdentityInfo( - identity_id="test_id", - identity_type="Application", - tenant_id="test_tenant", - tenant_domain="contoso.onmicrosoft.com", - tenant_domains=["contoso.onmicrosoft.com"], - location="test_location", - ) - session = M365PowerShell(credentials, identity) - session.execute = MagicMock(return_value="sometoken") + credentials = M365Credentials( + user="", + passwd="", + encrypted_passwd="", + client_id="test_client_id", + client_secret="test_client_secret", + tenant_id="test_tenant_id", + ) + identity = M365IdentityInfo( + identity_id="test_id", + identity_type="Application", + tenant_id="test_tenant", + tenant_domain="contoso.onmicrosoft.com", + tenant_domains=["contoso.onmicrosoft.com"], + location="test_location", + ) + session = M365PowerShell(credentials, identity) + session.execute = MagicMock(return_value="sometoken") - result = session.test_credentials(credentials) - assert result is True - session.execute.assert_any_call("Write-Output $graphToken") - session.close() + result = session.test_credentials(credentials) + assert result is True + session.execute.assert_any_call("Write-Output $graphToken") + session.close() @patch("subprocess.Popen") @patch("msal.ConfidentialClientApplication") @@ -748,7 +822,8 @@ class Testm365PowerShell: assert result is True # Verify all expected PowerShell commands were called - assert session.execute.call_count == 3 + # 4 calls: teamstokenBody, teamsToken, Write-Output $teamsToken, Connect-MicrosoftTeams + assert session.execute.call_count == 4 mock_decode_jwt.assert_called_once_with("valid_teams_token") session.close() @@ -849,7 +924,8 @@ class Testm365PowerShell: assert result is True # Verify all expected PowerShell commands were called - assert session.execute.call_count == 3 + # 4 calls: SecureSecret, exchangeToken, Write-Output $exchangeToken, Connect-ExchangeOnline + assert session.execute.call_count == 4 mock_decode_msal_token.assert_called_once_with("valid_exchange_token") session.close() @@ -970,3 +1046,623 @@ class Testm365PowerShell: del sys.modules["win32crypt"] session.close() + + @patch("subprocess.Popen") + def test_clean_certificate_content(self, mock_popen): + """Test clean_certificate_content method with various certificate content formats""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + credentials = M365Credentials() + identity = M365IdentityInfo() + session = M365PowerShell(credentials, identity) + + # Test with clean base64 content + clean_cert = "MIIDXTCCAkWgAwIBAgIJAJc1HiIAZAiIMA0GCSqGSIb3DQEBBQUAMEUxCzAJBgNV" + result = session.clean_certificate_content(clean_cert) + assert result == clean_cert + + # Test with newlines + cert_with_newlines = "MIIDXTCCAkWgAwIBAgIJAJc1HiIAZAiIMA0GCSqGSIb3DQEBBQUAMEUxCzAJBgNV\nBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldA==" + expected = "MIIDXTCCAkWgAwIBAgIJAJc1HiIAZAiIMA0GCSqGSIb3DQEBBQUAMEUxCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldA==" + result = session.clean_certificate_content(cert_with_newlines) + assert result == expected + + # Test with carriage returns + cert_with_cr = "MIIDXTCCAkWgAwIBAgIJAJc1HiIAZAiIMA0GCSqGSIb3DQEBBQUAMEUxCzAJBgNV\rBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldA==" + result = session.clean_certificate_content(cert_with_cr) + assert result == expected + + # Test with spaces + cert_with_spaces = "MIIDXTCCAkWgAwIBAgIJAJc1HiIAZAiIMA0GCSqGSIb3DQEBBQUAMEUxCzAJBgNV BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldA==" + result = session.clean_certificate_content(cert_with_spaces) + assert result == expected + + # Test with combination of all whitespace types + cert_mixed = "MIIDXTCCAkWgAwIBAgIJAJc1HiIAZAiIMA0GCSqGSIb3DQEBBQUAMEUxCzAJBgNV\n\r BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldA==" + result = session.clean_certificate_content(cert_mixed) + assert result == expected + + # Test with leading/trailing whitespace + cert_with_whitespace = " MIIDXTCCAkWgAwIBAgIJAJc1HiIAZAiIMA0GCSqGSIb3DQEBBQUAMEUxCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldA== " + result = session.clean_certificate_content(cert_with_whitespace) + assert result == expected + + session.close() + + @patch("subprocess.Popen") + def test_init_credential_certificate_auth(self, mock_popen): + """Test init_credential method with certificate authentication""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + credentials = M365Credentials( + client_id="test_client_id", + tenant_id="test_tenant_id", + certificate_content=certificate_content, + tenant_domains=["example.com"], + ) + identity = M365IdentityInfo(tenant_domains=["example.com"]) + + # Create session without calling init_credential + with patch.object(M365PowerShell, "init_credential"): + session = M365PowerShell(credentials, identity) + + # Mock the execute method + execute_calls = [] + + def mock_execute(command): + execute_calls.append(command) + if ( + "New-Object System.Security.Cryptography.X509Certificates.X509Certificate2" + in command + ): + return "" # No error + return "" + + session.execute = MagicMock(side_effect=mock_execute) + session.sanitize = MagicMock(side_effect=lambda x: x) + session.clean_certificate_content = MagicMock(return_value=certificate_content) + + # Now call init_credential + session.init_credential(credentials) + + # Verify clean_certificate_content was called + session.clean_certificate_content.assert_called_once_with(certificate_content) + + # Verify sanitize was called for client_id and tenant_id + session.sanitize.assert_any_call(credentials.client_id) + session.sanitize.assert_any_call(credentials.tenant_id) + + # Verify execute was called with correct commands + session.execute.assert_any_call( + f'$certBytes = [Convert]::FromBase64String("{certificate_content}")' + ) + session.execute.assert_any_call( + "$certificate = New-Object System.Security.Cryptography.X509Certificates.X509Certificate2(,$certBytes)" + ) + session.execute.assert_any_call(f'$clientID = "{credentials.client_id}"') + session.execute.assert_any_call(f'$tenantID = "{credentials.tenant_id}"') + session.execute.assert_any_call( + f'$tenantDomain = "{credentials.tenant_domains[0]}"' + ) + + session.close() + + @patch("subprocess.Popen") + def test_init_credential_certificate_auth_creation_error(self, mock_popen): + """Test init_credential method with certificate authentication when certificate creation fails""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + certificate_content = base64.b64encode(b"invalid_certificate").decode("utf-8") + credentials = M365Credentials( + client_id="test_client_id", + tenant_id="test_tenant_id", + certificate_content=certificate_content, + tenant_domains=["example.com"], + ) + identity = M365IdentityInfo(tenant_domains=["example.com"]) + + # Create session without calling init_credential + with patch.object(M365PowerShell, "init_credential"): + session = M365PowerShell(credentials, identity) + + # Mock the execute method to simulate certificate creation error + def mock_execute(command): + if ( + "New-Object System.Security.Cryptography.X509Certificates.X509Certificate2" + in command + ): + return "Certificate creation failed: Invalid certificate format" + return "" + + session.execute = MagicMock(side_effect=mock_execute) + session.sanitize = MagicMock(side_effect=lambda x: x) + session.clean_certificate_content = MagicMock(return_value=certificate_content) + + with pytest.raises(M365CertificateCreationError) as exc_info: + session.init_credential(credentials) + + # The actual error message format from the exception + assert "Failed to create X.509 certificate object" in str(exc_info.value) + + session.close() + + @patch("subprocess.Popen") + def test_test_exchange_certificate_connection_success(self, mock_popen): + """Test test_exchange_certificate_connection method with successful connection""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + credentials = M365Credentials() + identity = M365IdentityInfo() + session = M365PowerShell(credentials, identity) + + # Mock successful Exchange connection + session.execute = MagicMock( + return_value="Connected successfully https://aka.ms/exov3-module" + ) + + result = session.test_exchange_certificate_connection() + assert result is True + + session.execute.assert_called_once_with( + "Connect-ExchangeOnline -Certificate $certificate -AppId $clientID -Organization $tenantDomain" + ) + + session.close() + + @patch("subprocess.Popen") + def test_test_exchange_certificate_connection_failure(self, mock_popen): + """Test test_exchange_certificate_connection method with failed connection""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + credentials = M365Credentials() + identity = M365IdentityInfo() + session = M365PowerShell(credentials, identity) + + # Mock failed Exchange connection + session.execute = MagicMock( + return_value="Connection failed: Authentication error" + ) + + result = session.test_exchange_certificate_connection() + assert result is False + + session.execute.assert_called_once_with( + "Connect-ExchangeOnline -Certificate $certificate -AppId $clientID -Organization $tenantDomain" + ) + + session.close() + + @patch("subprocess.Popen") + def test_test_teams_certificate_connection_success(self, mock_popen): + """Test test_teams_certificate_connection method with successful connection""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + credentials = M365Credentials() + identity = M365IdentityInfo(identity_id="test_identity_id") + + # Create session without calling init_credential + with patch.object(M365PowerShell, "init_credential"): + session = M365PowerShell(credentials, identity) + + # Mock successful Teams connection - the method returns bool + def mock_execute_side_effect(command): + if "Connect-MicrosoftTeams" in command: + # Return result that contains the identity_id for success + return "Connected successfully test_identity_id" + return "" + + session.execute = MagicMock(side_effect=mock_execute_side_effect) + + result = session.test_teams_certificate_connection() + assert result is True + + session.execute.assert_called_once_with( + "Connect-MicrosoftTeams -Certificate $certificate -ApplicationId $clientID -TenantId $tenantID" + ) + + session.close() + + @patch("subprocess.Popen") + def test_test_teams_certificate_connection_failure(self, mock_popen): + """Test test_teams_certificate_connection method with failed connection""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + credentials = M365Credentials() + identity = M365IdentityInfo() + session = M365PowerShell(credentials, identity) + + # Mock failed Teams connection + def mock_execute_side_effect(command, json_parse=False): + if "Connect-MicrosoftTeams" in command: + raise Exception("Connection failed: Authentication error") + return "" + + session.execute = MagicMock(side_effect=mock_execute_side_effect) + + # Should raise exception on connection failure + with pytest.raises(Exception) as exc_info: + session.test_teams_certificate_connection() + + assert "Connection failed: Authentication error" in str(exc_info.value) + + session.close() + + @patch("subprocess.Popen") + def test_test_credentials_certificate_auth_success(self, mock_popen): + """Test test_credentials method with certificate authentication - successful""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + credentials = M365Credentials( + client_id="test_client_id", certificate_content=certificate_content + ) + identity = M365IdentityInfo() + + # Create session without calling init_credential + with patch.object(M365PowerShell, "init_credential"): + session = M365PowerShell(credentials, identity) + + # Mock successful certificate connections + # Note: The actual implementation uses "or" so if teams succeeds, exchange won't be called + session.test_teams_certificate_connection = MagicMock(return_value=True) + session.test_exchange_certificate_connection = MagicMock(return_value=True) + + result = session.test_credentials(credentials) + assert result is True + + session.test_teams_certificate_connection.assert_called_once() + # Exchange connection should NOT be called if teams connection succeeds (due to "or" logic) + session.test_exchange_certificate_connection.assert_not_called() + + session.close() + + @patch("subprocess.Popen") + def test_test_credentials_certificate_auth_failure(self, mock_popen): + """Test test_credentials method with certificate authentication - failure""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + credentials = M365Credentials( + client_id="test_client_id", certificate_content=certificate_content + ) + identity = M365IdentityInfo() + + # Create session without calling init_credential + with patch.object(M365PowerShell, "init_credential"): + session = M365PowerShell(credentials, identity) + + # Mock failed certificate connections - teams fails, so exchange is tried + session.test_teams_certificate_connection = MagicMock(return_value=False) + session.test_exchange_certificate_connection = MagicMock(return_value=False) + + result = session.test_credentials(credentials) + assert result is True # Method always returns True after the try block + + session.close() + + @patch("subprocess.Popen") + def test_connect_microsoft_teams_certificate_auth(self, mock_popen): + """Test connect_microsoft_teams method with certificate authentication""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + credentials = M365Credentials() + identity = M365IdentityInfo() + session = M365PowerShell(credentials, identity) + + # Mock certificate variable check and teams connection + execute_calls = [] + + def mock_execute_side_effect(command, json_parse=False): + execute_calls.append(command) + if "Write-Output $certificate" in command: + return "certificate_content" # Non-empty means certificate exists + elif "Connect-MicrosoftTeams" in command: + return {"TenantId": "test-tenant-id", "Account": "test-account"} + return "" + + session.execute = MagicMock(side_effect=mock_execute_side_effect) + session.test_teams_certificate_connection = MagicMock( + return_value={"success": True} + ) + + result = session.connect_microsoft_teams() + assert result == {"success": True} + + session.test_teams_certificate_connection.assert_called_once() + + session.close() + + @patch("subprocess.Popen") + def test_connect_exchange_online_certificate_auth(self, mock_popen): + """Test connect_exchange_online method with certificate authentication""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + + credentials = M365Credentials() + identity = M365IdentityInfo() + session = M365PowerShell(credentials, identity) + + # Mock certificate variable check and exchange connection + execute_calls = [] + + def mock_execute_side_effect(command, json_parse=False): + execute_calls.append(command) + if "Write-Output $certificate" in command: + return "certificate_content" # Non-empty means certificate exists + return "" + + session.execute = MagicMock(side_effect=mock_execute_side_effect) + session.test_exchange_certificate_connection = MagicMock(return_value=True) + + result = session.connect_exchange_online() + assert result is True + + session.test_exchange_certificate_connection.assert_called_once() + + session.close() + + @patch("subprocess.Popen") + def test_clean_certificate_content_basic(self, mock_popen): + """Test clean_certificate_content method with basic certificate content""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + mock_process.returncode = 0 + + session = M365PowerShell( + M365Credentials( + client_id="test_client_id", + client_secret="test_secret", + tenant_id="test_tenant_id", + tenant_domains=["contoso.com"], + ), + M365IdentityInfo( + tenant_id="test_tenant_id", + tenant_domain="contoso.com", + tenant_domains=["contoso.com"], + identity_id="test_identity_id", + identity_type="Service Principal", + ), + ) + + # Test basic certificate content cleaning + cert_content = "LS0tLS1CRUdJTi\nBFUlRJRklD\rQVRFLS0tLS0\n" + expected = "LS0tLS1CRUdJTiBFUlRJRklDQVRFLS0tLS0" + + result = session.clean_certificate_content(cert_content) + + assert result == expected + session.close() + + @patch("subprocess.Popen") + def test_clean_certificate_content_with_spaces(self, mock_popen): + """Test clean_certificate_content method with spaces in certificate content""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + mock_process.returncode = 0 + + session = M365PowerShell( + M365Credentials( + client_id="test_client_id", + client_secret="test_secret", + tenant_id="test_tenant_id", + tenant_domains=["contoso.com"], + ), + M365IdentityInfo( + tenant_id="test_tenant_id", + tenant_domain="contoso.com", + tenant_domains=["contoso.com"], + identity_id="test_identity_id", + identity_type="Service Principal", + ), + ) + + # Test certificate content with spaces + cert_content = " LS0tLS1CRUdJTi BFUlRJRklD QVRFLS0tLS0 " + expected = "LS0tLS1CRUdJTiBFUlRJRklDQVRFLS0tLS0" + + result = session.clean_certificate_content(cert_content) + + assert result == expected + session.close() + + @patch("subprocess.Popen") + def test_clean_certificate_content_empty(self, mock_popen): + """Test clean_certificate_content method with empty certificate content""" + mock_process = MagicMock() + mock_popen.return_value = mock_process + mock_process.returncode = 0 + + session = M365PowerShell( + M365Credentials( + client_id="test_client_id", + client_secret="test_secret", + tenant_id="test_tenant_id", + tenant_domains=["contoso.com"], + ), + M365IdentityInfo( + tenant_id="test_tenant_id", + tenant_domain="contoso.com", + tenant_domains=["contoso.com"], + identity_id="test_identity_id", + identity_type="Service Principal", + ), + ) + + # Test empty certificate content + cert_content = "" + expected = "" + + result = session.clean_certificate_content(cert_content) + + assert result == expected + session.close() + + @patch("subprocess.Popen") + def test_init_credential_certificate_auth_with_domains(self, mock_popen): + """Test init_credential method with certificate authentication and tenant domains""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + mock_process = MagicMock() + mock_popen.return_value = mock_process + mock_process.returncode = 0 + + executed_commands = [] + + def mock_execute(command): + executed_commands.append(command) + if "Error creating certificate" in command: + return None # No error + return "" + + # Create session with non-certificate credentials first + session = M365PowerShell( + M365Credentials( + client_id="test_client_id", + client_secret="test_secret", + tenant_id="test_tenant_id", + tenant_domains=["contoso.com", "subdomain.contoso.com"], + ), + M365IdentityInfo( + tenant_id="test_tenant_id", + tenant_domain="contoso.com", + tenant_domains=["contoso.com", "subdomain.contoso.com"], + identity_id="test_identity_id", + identity_type="Service Principal with Certificate", + ), + ) + + # Mock methods before calling init_credential + session.execute = MagicMock(side_effect=mock_execute) + session.sanitize = MagicMock(side_effect=lambda x: x) + session.clean_certificate_content = MagicMock(return_value=certificate_content) + + # Now test certificate authentication + session.init_credential( + M365Credentials( + client_id="test_client_id", + tenant_id="test_tenant_id", + certificate_content=certificate_content, + tenant_domains=["contoso.com", "subdomain.contoso.com"], + ) + ) + + # Verify that certificate content was cleaned + session.clean_certificate_content.assert_called_once_with(certificate_content) + + # Verify certificate creation commands were executed + assert any( + "$certBytes = [Convert]::FromBase64String" in cmd + for cmd in executed_commands + ) + assert any( + "$certificate = New-Object System.Security.Cryptography.X509Certificates.X509Certificate2" + in cmd + for cmd in executed_commands + ) + assert any('$clientID = "test_client_id"' in cmd for cmd in executed_commands) + assert any('$tenantID = "test_tenant_id"' in cmd for cmd in executed_commands) + assert any('$tenantDomain = "contoso.com"' in cmd for cmd in executed_commands) + + session.close() + + @patch("subprocess.Popen") + def test_test_credentials_certificate_auth_with_or_logic(self, mock_popen): + """Test test_credentials method with certificate auth using OR logic between Teams and Exchange""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + mock_process = MagicMock() + mock_popen.return_value = mock_process + mock_process.returncode = 0 + + # Create session with non-certificate credentials first + session = M365PowerShell( + M365Credentials( + client_id="test_client_id", + client_secret="test_secret", + tenant_id="test_tenant_id", + tenant_domains=["contoso.com"], + ), + M365IdentityInfo( + tenant_id="test_tenant_id", + tenant_domain="contoso.com", + tenant_domains=["contoso.com"], + identity_id="test_identity_id", + identity_type="Service Principal with Certificate", + ), + ) + + # Mock that Teams connection fails but Exchange succeeds + session.test_teams_certificate_connection = MagicMock(return_value=False) + session.test_exchange_certificate_connection = MagicMock(return_value=True) + + result = session.test_credentials( + M365Credentials( + client_id="test_client_id", + tenant_id="test_tenant_id", + certificate_content=certificate_content, + tenant_domains=["contoso.com"], + ) + ) + + assert result is True + session.test_teams_certificate_connection.assert_called_once() + session.test_exchange_certificate_connection.assert_called_once() + + session.close() + + @patch("subprocess.Popen") + def test_test_credentials_certificate_auth_both_fail(self, mock_popen): + """Test test_credentials method with certificate auth when both Teams and Exchange fail""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + mock_process = MagicMock() + mock_popen.return_value = mock_process + mock_process.returncode = 0 + + # Create session with non-certificate credentials first + session = M365PowerShell( + M365Credentials( + client_id="test_client_id", + client_secret="test_secret", + tenant_id="test_tenant_id", + tenant_domains=["contoso.com"], + ), + M365IdentityInfo( + tenant_id="test_tenant_id", + tenant_domain="contoso.com", + tenant_domains=["contoso.com"], + identity_id="test_identity_id", + identity_type="Service Principal with Certificate", + ), + ) + + # Mock that both connections fail + session.test_teams_certificate_connection = MagicMock(return_value=False) + session.test_exchange_certificate_connection = MagicMock(return_value=False) + + # Even when both fail, the method should return True (this is the intended logic) + result = session.test_credentials( + M365Credentials( + client_id="test_client_id", + tenant_id="test_tenant_id", + certificate_content=certificate_content, + tenant_domains=["contoso.com"], + ) + ) + + assert result is True + session.test_teams_certificate_connection.assert_called_once() + session.test_exchange_certificate_connection.assert_called_once() + + session.close() diff --git a/tests/providers/m365/m365_fixtures.py b/tests/providers/m365/m365_fixtures.py index f985f94b6a..2c1a322ece 100644 --- a/tests/providers/m365/m365_fixtures.py +++ b/tests/providers/m365/m365_fixtures.py @@ -1,5 +1,6 @@ +from unittest.mock import MagicMock + from azure.identity import DefaultAzureCredential -from mock import MagicMock from prowler.providers.m365.m365_provider import M365Provider from prowler.providers.m365.models import ( diff --git a/tests/providers/m365/m365_provider_test.py b/tests/providers/m365/m365_provider_test.py index 0f629d21a9..0af2abc7c8 100644 --- a/tests/providers/m365/m365_provider_test.py +++ b/tests/providers/m365/m365_provider_test.py @@ -1,15 +1,16 @@ +import base64 import os -from unittest.mock import patch +from unittest.mock import MagicMock, mock_open, patch from uuid import uuid4 import pytest from azure.core.credentials import AccessToken from azure.identity import ( + CertificateCredential, ClientSecretCredential, DefaultAzureCredential, InteractiveBrowserCredential, ) -from mock import MagicMock from prowler.config.config import ( default_config_file_path, @@ -18,15 +19,29 @@ from prowler.config.config import ( ) from prowler.providers.common.models import Connection from prowler.providers.m365.exceptions.exceptions import ( + M365BrowserAuthNoFlagError, + M365BrowserAuthNoTenantIDError, + M365ClientIdAndClientSecretNotBelongingToTenantIdError, + M365ConfigCredentialsError, + M365CredentialsUnavailableError, + M365DefaultAzureCredentialError, + M365EnvironmentVariableError, + M365GetTokenIdentityError, M365HTTPResponseError, M365InvalidProviderIdError, M365MissingEnvironmentCredentialsError, M365NoAuthenticationMethodError, + M365NotTenantIdButClientIdAndClientSecretError, + M365NotValidCertificateContentError, + M365NotValidCertificatePathError, M365NotValidClientIdError, M365NotValidClientSecretError, M365NotValidPasswordError, M365NotValidTenantIdError, M365NotValidUserError, + M365TenantIdAndClientIdNotBelongingToClientSecretError, + M365TenantIdAndClientSecretNotBelongingToClientIdError, + M365UserCredentialsError, M365UserNotBelongingToTenantError, ) from prowler.providers.m365.m365_provider import M365Provider @@ -296,6 +311,10 @@ class TestM365Provider: patch( "prowler.providers.m365.m365_provider.GraphServiceClient" ) as mock_graph_client, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), patch( "prowler.providers.m365.m365_provider.M365Provider.setup_identity", return_value=M365IdentityInfo( @@ -343,6 +362,10 @@ class TestM365Provider: patch( "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" ) as mock_validate_static_credentials, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), patch( "prowler.providers.m365.m365_provider.M365Provider.setup_identity", return_value=M365IdentityInfo( @@ -378,9 +401,15 @@ class TestM365Provider: def test_test_connection_tenant_id_client_id_client_secret_no_user_password( self, ): - with patch( - "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" - ) as mock_validate_static_credentials: + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" + ) as mock_validate_static_credentials, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), + ): mock_validate_static_credentials.side_effect = M365NotValidUserError( file=os.path.basename(__file__), message="The provided M365 User is not valid.", @@ -403,9 +432,15 @@ class TestM365Provider: def test_test_connection_tenant_id_client_id_client_secret_user_no_password( self, ): - with patch( - "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" - ) as mock_validate_static_credentials: + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" + ) as mock_validate_static_credentials, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), + ): mock_validate_static_credentials.side_effect = M365NotValidPasswordError( file=os.path.basename(__file__), message="The provided M365 Password is not valid.", @@ -426,9 +461,15 @@ class TestM365Provider: assert "The provided M365 Password is not valid." in str(exception.value) def test_test_connection_with_httpresponseerror(self): - with patch( - "prowler.providers.m365.m365_provider.M365Provider.setup_session" - ) as mock_setup_session: + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_session" + ) as mock_setup_session, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), + ): mock_setup_session.side_effect = M365HTTPResponseError( file="test_file", original_exception="Simulated HttpResponseError" ) @@ -446,9 +487,15 @@ class TestM365Provider: ) def test_test_connection_with_exception(self): - with patch( - "prowler.providers.m365.m365_provider.M365Provider.setup_session" - ) as mock_setup_session: + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_session" + ) as mock_setup_session, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), + ): mock_setup_session.side_effect = Exception("Simulated Exception") with pytest.raises(Exception) as exception: @@ -466,7 +513,7 @@ class TestM365Provider: assert exception.type == M365NoAuthenticationMethodError assert ( - "M365 provider requires at least one authentication method set: [--env-auth | --az-cli-auth | --sp-env-auth | --browser-auth]" + "M365 provider requires at least one authentication method set: [--env-auth | --az-cli-auth | --sp-env-auth | --browser-auth | --certificate-auth]" in exception.value.args[0] ) @@ -503,9 +550,15 @@ class TestM365Provider: def test_test_connection_user_not_belonging_to_tenant( self, ): - with patch( - "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" - ) as mock_validate_static_credentials: + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" + ) as mock_validate_static_credentials, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), + ): mock_validate_static_credentials.side_effect = M365UserNotBelongingToTenantError( file=os.path.basename(__file__), message="The provided M365 User does not belong to the specified tenant.", @@ -559,24 +612,30 @@ class TestM365Provider: user="test@example.com", password="test_password", ) - assert "The provided Client Secret is not valid." in str(exception.value) + assert ( + "You must provide a client secret, certificate content or certificate path. Please check your credentials and try again." + in str(exception.value) + ) def test_validate_arguments_missing_env_credentials(self): - with pytest.raises(M365MissingEnvironmentCredentialsError) as exception: + with pytest.raises(M365ConfigCredentialsError) as exception: M365Provider.validate_arguments( az_cli_auth=False, sp_env_auth=False, env_auth=True, browser_auth=False, - tenant_id=None, + certificate_auth=False, + tenant_id="test_tenant_id", client_id="test_client_id", - client_secret="test_secret", + client_secret=None, user=None, password=None, + certificate_content=None, + certificate_path=None, ) assert ( - "M365 provider requires AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID, M365_USER and M365_PASSWORD environment variables to be set when using --env-auth" + "You must provide a valid set of credentials. Please check your credentials and try again." in str(exception.value) ) @@ -588,6 +647,10 @@ class TestM365Provider: patch( "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" ) as mock_validate_static_credentials, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), patch( "prowler.providers.m365.m365_provider.M365Provider.setup_identity", return_value=M365IdentityInfo( @@ -742,6 +805,10 @@ class TestM365Provider: patch( "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials" ) as mock_validate_static_credentials, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), patch( "prowler.providers.m365.m365_provider.M365Provider.setup_identity", return_value=M365IdentityInfo( @@ -780,3 +847,1287 @@ class TestM365Provider: 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) ) + + def test_m365_provider_certificate_auth(self): + """Test M365 Provider initialization with certificate authentication""" + tenant_id = None + client_id = None + client_secret = None + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + fixer_config = load_and_validate_config_file( + "m365", default_fixer_config_file_path + ) + azure_region = "M365Global" + + # Mock certificate credential + mock_cert_credential = MagicMock() + mock_cert_credential.__class__ = CertificateCredential + + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_session", + return_value=mock_cert_credential, + ), + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_identity", + return_value=M365IdentityInfo( + identity_id=IDENTITY_ID, + identity_type="Service Principal with Certificate", + tenant_id=TENANT_ID, + tenant_domain=DOMAIN, + location=LOCATION, + certificate_thumbprint="ABC123", + ), + ), + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_powershell", + return_value=M365Credentials( + client_id=CLIENT_ID, + tenant_id=TENANT_ID, + certificate_content=certificate_content, + ), + ), + ): + m365_provider = M365Provider( + sp_env_auth=False, + az_cli_auth=False, + browser_auth=False, + env_auth=False, + certificate_auth=True, + tenant_id=tenant_id, + client_id=client_id, + client_secret=client_secret, + certificate_content=certificate_content, + region=azure_region, + config_path=default_config_file_path, + fixer_config=fixer_config, + ) + + assert m365_provider.region_config == M365RegionConfig( + name="M365Global", + authority=None, + base_url="https://graph.microsoft.com", + credential_scopes=["https://graph.microsoft.com/.default"], + ) + assert ( + m365_provider.identity.identity_type + == "Service Principal with Certificate" + ) + assert m365_provider.session == mock_cert_credential + + def test_check_service_principal_creds_env_vars_missing_client_id(self): + """Test check_service_principal_creds_env_vars with missing AZURE_CLIENT_ID""" + with ( + patch.dict(os.environ, {}, clear=True), + pytest.raises(M365EnvironmentVariableError) as exception, + ): + M365Provider.check_service_principal_creds_env_vars() + + assert exception.type == M365EnvironmentVariableError + assert ( + "Missing environment variable AZURE_CLIENT_ID required to authenticate." + in str(exception.value) + ) + + def test_check_service_principal_creds_env_vars_missing_tenant_id(self): + """Test check_service_principal_creds_env_vars with missing AZURE_TENANT_ID""" + with ( + patch.dict(os.environ, {"AZURE_CLIENT_ID": "test_client_id"}, clear=True), + pytest.raises(M365EnvironmentVariableError) as exception, + ): + M365Provider.check_service_principal_creds_env_vars() + + assert exception.type == M365EnvironmentVariableError + assert ( + "Missing environment variable AZURE_TENANT_ID required to authenticate." + in str(exception.value) + ) + + def test_check_service_principal_creds_env_vars_missing_client_secret(self): + """Test check_service_principal_creds_env_vars with missing AZURE_CLIENT_SECRET""" + with ( + patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": "test_client_id", + "AZURE_TENANT_ID": "test_tenant_id", + }, + clear=True, + ), + pytest.raises(M365EnvironmentVariableError) as exception, + ): + M365Provider.check_service_principal_creds_env_vars() + + assert exception.type == M365EnvironmentVariableError + assert ( + "Missing environment variable AZURE_CLIENT_SECRET required to authenticate." + in str(exception.value) + ) + + def test_check_service_principal_creds_env_vars_success(self): + """Test check_service_principal_creds_env_vars with all required variables""" + with patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": "test_client_id", + "AZURE_TENANT_ID": "test_tenant_id", + "AZURE_CLIENT_SECRET": "test_client_secret", + }, + ): + # Should not raise any exception + M365Provider.check_service_principal_creds_env_vars() + + def test_check_certificate_creds_env_vars_missing_client_id(self): + """Test check_certificate_creds_env_vars with missing AZURE_CLIENT_ID""" + with ( + patch.dict(os.environ, {}, clear=True), + pytest.raises(M365EnvironmentVariableError) as exception, + ): + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=True + ) + + assert exception.type == M365EnvironmentVariableError + assert ( + "Missing environment variable AZURE_CLIENT_ID required to authenticate." + in str(exception.value) + ) + + def test_check_certificate_creds_env_vars_missing_tenant_id(self): + """Test check_certificate_creds_env_vars with missing AZURE_TENANT_ID""" + with ( + patch.dict(os.environ, {"AZURE_CLIENT_ID": "test_client_id"}, clear=True), + pytest.raises(M365EnvironmentVariableError) as exception, + ): + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=True + ) + + assert exception.type == M365EnvironmentVariableError + assert ( + "Missing environment variable AZURE_TENANT_ID required to authenticate." + in str(exception.value) + ) + + def test_check_certificate_creds_env_vars_missing_certificate_content(self): + """Test check_certificate_creds_env_vars with missing M365_CERTIFICATE_CONTENT""" + with ( + patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": "test_client_id", + "AZURE_TENANT_ID": "test_tenant_id", + }, + clear=True, + ), + pytest.raises(M365EnvironmentVariableError) as exception, + ): + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=True + ) + + assert exception.type == M365EnvironmentVariableError + assert ( + "Missing environment variable M365_CERTIFICATE_CONTENT required to authenticate." + in str(exception.value) + ) + + def test_check_certificate_creds_env_vars_missing_certificate_content_but_certificate_path( + self, + ): + """Test check_certificate_creds_env_vars with missing M365_CERTIFICATE_CONTENT""" + with patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": "test_client_id", + "AZURE_TENANT_ID": "test_tenant_id", + }, + clear=True, + ): + # This should not raise an exception since check_certificate_content=False + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=False + ) + + def test_check_certificate_creds_env_vars_success(self): + """Test check_certificate_creds_env_vars with all required variables""" + with patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": "test_client_id", + "AZURE_TENANT_ID": "test_tenant_id", + "M365_CERTIFICATE_CONTENT": base64.b64encode( + b"fake_certificate" + ).decode("utf-8"), + }, + ): + # Should not raise any exception + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=True + ) + + def test_setup_powershell_env_auth_missing_credentials(self): + """Test setup_powershell with env_auth but missing environment variables""" + with ( + patch.dict(os.environ, {}, clear=True), + pytest.raises(M365MissingEnvironmentCredentialsError) as exception, + ): + M365Provider.setup_powershell( + env_auth=True, + identity=M365IdentityInfo( + identity_id=IDENTITY_ID, + identity_type="User", + tenant_id=TENANT_ID, + tenant_domain=DOMAIN, + tenant_domains=["test.onmicrosoft.com"], + location=LOCATION, + ), + ) + + assert exception.type == M365MissingEnvironmentCredentialsError + assert ( + "Missing M365_USER or M365_PASSWORD environment variables required for credentials authentication." + in str(exception.value) + ) + + def test_setup_powershell_env_auth_success(self): + """Test setup_powershell with env_auth and valid environment variables""" + with ( + patch.dict( + os.environ, + { + "M365_USER": "test@example.com", + "M365_PASSWORD": "password", + "AZURE_CLIENT_ID": CLIENT_ID, + "AZURE_CLIENT_SECRET": CLIENT_SECRET, + "AZURE_TENANT_ID": TENANT_ID, + }, + ), + patch( + "prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials", + return_value=True, + ), + ): + result = M365Provider.setup_powershell( + env_auth=True, + 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 == "test@example.com" + assert result.passwd == "password" + assert result.client_id == CLIENT_ID + + def test_setup_powershell_sp_env_auth_success(self): + """Test setup_powershell with sp_env_auth and valid environment variables""" + with ( + patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": CLIENT_ID, + "AZURE_CLIENT_SECRET": CLIENT_SECRET, + "AZURE_TENANT_ID": TENANT_ID, + }, + ), + patch( + "prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials", + return_value=True, + ), + ): + result = M365Provider.setup_powershell( + sp_env_auth=True, + identity=M365IdentityInfo( + identity_id=IDENTITY_ID, + identity_type="Service Principal", + tenant_id=TENANT_ID, + tenant_domain=DOMAIN, + tenant_domains=["test.onmicrosoft.com"], + location=LOCATION, + ), + ) + + assert result.client_id == CLIENT_ID + assert result.client_secret == CLIENT_SECRET + assert result.tenant_id == TENANT_ID + + def test_setup_powershell_certificate_auth_success(self): + """Test setup_powershell with certificate_auth and valid environment variables""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + with ( + patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": CLIENT_ID, + "AZURE_TENANT_ID": TENANT_ID, + "M365_CERTIFICATE_CONTENT": certificate_content, + }, + ), + patch( + "prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials", + return_value=True, + ), + ): + identity = M365IdentityInfo( + identity_id=IDENTITY_ID, + identity_type="Service Principal with Certificate", + tenant_id=TENANT_ID, + tenant_domain=DOMAIN, + tenant_domains=["test.onmicrosoft.com"], + location=LOCATION, + ) + + result = M365Provider.setup_powershell( + certificate_auth=True, + identity=identity, + ) + + assert result.client_id == CLIENT_ID + assert result.tenant_id == TENANT_ID + assert result.certificate_content == certificate_content + assert identity.identity_type == "Service Principal with Certificate" + + def test_setup_powershell_invalid_credentials(self): + """Test setup_powershell with invalid credentials""" + credentials_dict = { + "user": "test@example.com", + "password": "test_password", + "client_id": "test_client_id", + "tenant_id": "test_tenant_id", + "client_secret": "test_client_secret", + } + + with ( + patch( + "prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials", + return_value=False, + ), + pytest.raises(M365UserCredentialsError) as exception, + ): + M365Provider.setup_powershell( + env_auth=False, + m365_credentials=credentials_dict, + identity=M365IdentityInfo( + identity_id=IDENTITY_ID, + identity_type="User", + tenant_id=TENANT_ID, + tenant_domain=DOMAIN, + tenant_domains=["test.onmicrosoft.com"], + location=LOCATION, + ), + ) + + assert exception.type == M365UserCredentialsError + assert "The provided User credentials are not valid." in str(exception.value) + + def test_validate_arguments_browser_auth_without_tenant_id(self): + """Test validate_arguments with browser_auth but missing tenant_id""" + with pytest.raises(M365BrowserAuthNoTenantIDError) as exception: + M365Provider.validate_arguments( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=True, + certificate_auth=False, + tenant_id=None, + client_id=None, + client_secret=None, + user=None, + password=None, + certificate_content=None, + certificate_path=None, + ) + + assert exception.type == M365BrowserAuthNoTenantIDError + assert ( + "M365 Tenant ID (--tenant-id) is required for browser authentication mode" + in str(exception.value) + ) + + def test_validate_arguments_tenant_id_without_browser_flag(self): + """Test validate_arguments with tenant_id but without browser auth flag""" + with pytest.raises(M365BrowserAuthNoFlagError) as exception: + M365Provider.validate_arguments( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=False, + tenant_id=TENANT_ID, + client_id=None, + client_secret=None, + user=None, + password=None, + certificate_content=None, + certificate_path=None, + ) + + assert exception.type == M365BrowserAuthNoFlagError + assert "browser authentication flag (--browser-auth) not found" in str( + exception.value + ) + + def test_validate_arguments_missing_tenant_id_with_credentials(self): + """Test validate_arguments with client credentials but missing tenant_id""" + with pytest.raises(M365NotTenantIdButClientIdAndClientSecretError) as exception: + M365Provider.validate_arguments( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=False, + tenant_id=None, + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + user=None, + password=None, + certificate_content=None, + certificate_path=None, + ) + + assert exception.type == M365NotTenantIdButClientIdAndClientSecretError + assert "Tenant Id is required for M365 static credentials" in str( + exception.value + ) + + def test_test_connection_certificate_auth(self): + """Test test_connection with certificate authentication""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_session" + ) as mock_setup_session, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch("prowler.providers.m365.m365_provider.M365Provider.setup_powershell"), + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_identity", + return_value=M365IdentityInfo( + identity_id=IDENTITY_ID, + identity_type="Service Principal with Certificate", + tenant_id=TENANT_ID, + tenant_domain=DOMAIN, + tenant_domains=["test.onmicrosoft.com"], + location=LOCATION, + ), + ), + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_static_credentials", + return_value={ + "tenant_id": TENANT_ID, + "client_id": CLIENT_ID, + "client_secret": None, + "user": None, + "password": None, + "certificate_content": certificate_content, + }, + ), + ): + mock_session = MagicMock() + mock_setup_session.return_value = mock_session + + test_connection = M365Provider.test_connection( + certificate_auth=True, + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_content=certificate_content, + region="M365Global", + raise_on_exception=False, + provider_id="test.onmicrosoft.com", + ) + + assert isinstance(test_connection, Connection) + assert test_connection.is_connected + assert test_connection.error is None + + def test_test_connection_get_token_identity_error(self): + """Test test_connection when setup_identity returns None""" + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_session" + ) as mock_setup_session, + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_identity", + return_value=None, + ), + pytest.raises(M365GetTokenIdentityError) as exception, + ): + mock_session = MagicMock() + mock_setup_session.return_value = mock_session + + M365Provider.test_connection( + az_cli_auth=True, + raise_on_exception=True, + ) + + assert exception.type == M365GetTokenIdentityError + assert "Failed to retrieve M365 identity" in str(exception.value) + + def test_validate_static_credentials_client_id_secret_tenant_error(self): + """Test validate_static_credentials with client/secret/tenant mismatch errors""" + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.verify_client", + side_effect=M365NotValidTenantIdError( + file="test", message="Invalid tenant" + ), + ), + pytest.raises( + M365ClientIdAndClientSecretNotBelongingToTenantIdError + ) as exception, + ): + M365Provider.validate_static_credentials( + tenant_id=str(uuid4()), + client_id=str(uuid4()), + client_secret="test_secret", + user="test@example.com", + password="test_password", + ) + + assert exception.type == M365ClientIdAndClientSecretNotBelongingToTenantIdError + assert ( + "The provided Client ID and Client Secret do not belong to the specified Tenant ID." + in str(exception.value) + ) + + def test_validate_static_credentials_tenant_secret_client_error(self): + """Test validate_static_credentials with tenant/secret/client mismatch errors""" + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.verify_client", + side_effect=M365NotValidClientIdError( + file="test", message="Invalid client ID" + ), + ), + pytest.raises( + M365TenantIdAndClientSecretNotBelongingToClientIdError + ) as exception, + ): + M365Provider.validate_static_credentials( + tenant_id=str(uuid4()), + client_id=str(uuid4()), + client_secret="test_secret", + user="test@example.com", + password="test_password", + ) + + assert exception.type == M365TenantIdAndClientSecretNotBelongingToClientIdError + assert ( + "The provided Tenant ID and Client Secret do not belong to the specified Client ID." + in str(exception.value) + ) + + def test_validate_static_credentials_tenant_client_secret_error(self): + """Test validate_static_credentials with tenant/client/secret mismatch errors""" + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.verify_client", + side_effect=M365NotValidClientSecretError( + file="test", message="Invalid client secret" + ), + ), + pytest.raises( + M365TenantIdAndClientIdNotBelongingToClientSecretError + ) as exception, + ): + M365Provider.validate_static_credentials( + tenant_id=str(uuid4()), + client_id=str(uuid4()), + client_secret="test_secret", + user="test@example.com", + password="test_password", + ) + + assert exception.type == M365TenantIdAndClientIdNotBelongingToClientSecretError + assert ( + "The provided Tenant ID and Client ID do not belong to the specified Client Secret." + in str(exception.value) + ) + + def test_test_connection_default_azure_credential_error(self): + """Test test_connection with DefaultAzureCredential error in exception handling""" + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_session", + side_effect=M365DefaultAzureCredentialError( + file="test", + original_exception=Exception("Default credential error"), + ), + ), + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + pytest.raises(M365DefaultAzureCredentialError) as exception, + ): + M365Provider.test_connection( + az_cli_auth=True, + raise_on_exception=True, + ) + + assert exception.type == M365DefaultAzureCredentialError + + def test_test_connection_credentials_unavailable_error_handling(self): + """Test test_connection with CredentialsUnavailableError in exception handling""" + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_session", + side_effect=M365CredentialsUnavailableError( + file="test", original_exception=Exception("Credentials unavailable") + ), + ), + patch( + "prowler.providers.m365.m365_provider.M365Provider.validate_arguments" + ), + pytest.raises(M365CredentialsUnavailableError) as exception, + ): + M365Provider.test_connection( + sp_env_auth=True, + raise_on_exception=True, + ) + + assert exception.type == M365CredentialsUnavailableError + + def test_setup_session_certificate_auth_success(self): + """Test setup_session method with certificate authentication - success""" + with ( + patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": CLIENT_ID, + "AZURE_TENANT_ID": TENANT_ID, + "M365_CERTIFICATE_CONTENT": base64.b64encode( + b"fake_certificate" + ).decode("utf-8"), + }, + ), + patch( + "prowler.providers.m365.m365_provider.CertificateCredential" + ) as mock_cert_credential, + ): + mock_credential_instance = MagicMock() + mock_cert_credential.return_value = mock_credential_instance + + region_config = M365RegionConfig( + name="M365Global", + authority=None, + base_url="https://graph.microsoft.com", + credential_scopes=["https://graph.microsoft.com/.default"], + ) + + result = M365Provider.setup_session( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=True, + certificate_path=None, + tenant_id=None, + m365_credentials=None, + region_config=region_config, + ) + + assert result == mock_credential_instance + mock_cert_credential.assert_called_once_with( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_data=base64.b64decode( + base64.b64encode(b"fake_certificate").decode("utf-8") + ), + ) + + def test_setup_session_certificate_auth_client_authentication_error(self): + """Test setup_session method with certificate authentication - ClientAuthenticationError""" + from azure.core.exceptions import ClientAuthenticationError + + from prowler.providers.m365.exceptions.exceptions import M365SetUpSessionError + + with ( + patch.dict( + os.environ, + { + "AZURE_CLIENT_ID": CLIENT_ID, + "AZURE_TENANT_ID": TENANT_ID, + "M365_CERTIFICATE_CONTENT": base64.b64encode( + b"fake_certificate" + ).decode("utf-8"), + }, + ), + patch( + "prowler.providers.m365.m365_provider.CertificateCredential", + side_effect=ClientAuthenticationError("Authentication failed"), + ), + pytest.raises(M365SetUpSessionError) as exception, + ): + region_config = M365RegionConfig( + name="M365Global", + authority=None, + base_url="https://graph.microsoft.com", + credential_scopes=["https://graph.microsoft.com/.default"], + ) + + M365Provider.setup_session( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=True, + certificate_path=None, + tenant_id=None, + m365_credentials=None, + region_config=region_config, + ) + + # The error should be wrapped in M365SetUpSessionError and contain the ClientAuthenticationError + assert exception.type == M365SetUpSessionError + assert "M365ClientAuthenticationError" in str(exception.value) + + def test_setup_session_certificate_auth_with_static_credentials(self): + """Test setup_session method with certificate authentication using static credentials""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + m365_credentials = { + "tenant_id": TENANT_ID, + "client_id": CLIENT_ID, + "client_secret": CLIENT_SECRET, + "user": None, + "password": None, + "certificate_content": certificate_content, + } + + with ( + patch( + "prowler.providers.m365.m365_provider.CertificateCredential" + ) as mock_credential, + ): + mock_credential_instance = MagicMock() + mock_credential.return_value = mock_credential_instance + + region_config = M365RegionConfig( + name="M365Global", + authority=None, + base_url="https://graph.microsoft.com", + credential_scopes=["https://graph.microsoft.com/.default"], + ) + + result = M365Provider.setup_session( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=False, + certificate_path=None, + tenant_id=None, + m365_credentials=m365_credentials, + region_config=region_config, + ) + + assert result == mock_credential_instance + mock_credential.assert_called_once_with( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_data=base64.b64decode(certificate_content), + ) + + def test_setup_powershell_certificate_auth_missing_env_vars(self): + """Test setup_powershell with certificate_auth but missing environment variables""" + from pydantic.v1.error_wrappers import ValidationError + + with (patch.dict(os.environ, {}, clear=True),): + identity = M365IdentityInfo( + identity_id=IDENTITY_ID, + identity_type="Service Principal with Certificate", + tenant_id=TENANT_ID, + tenant_domain=DOMAIN, + tenant_domains=["test.onmicrosoft.com"], + location=LOCATION, + ) + + # Should raise ValidationError when trying to create credentials with None values + with pytest.raises(ValidationError) as exc_info: + M365Provider.setup_powershell( + certificate_auth=True, + identity=identity, + ) + + # Verify the error is about None values not being allowed + assert "none is not an allowed value" in str(exc_info.value).lower() + + def test_validate_arguments_certificate_auth_valid(self): + """Test validate_arguments method with valid certificate authentication arguments""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + # Should not raise any exception + M365Provider.validate_arguments( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=True, + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + client_secret=None, + user=None, + password=None, + certificate_content=certificate_content, + certificate_path=None, + ) + + def test_validate_arguments_certificate_auth_missing_certificate_content(self): + """Test validate_arguments method with certificate auth but missing certificate content""" + with pytest.raises(M365ConfigCredentialsError) as exception: + M365Provider.validate_arguments( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=False, + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + client_secret=None, + user=None, + password=None, + certificate_content=None, + certificate_path=None, + ) + + assert "You must provide a valid set of credentials" in str(exception.value) + + def test_print_credentials_with_certificate(self): + """Test print_credentials method with certificate authentication""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + # Mock certificate credential + mock_cert_credential = MagicMock() + mock_cert_credential.__class__ = CertificateCredential + + with ( + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_session", + return_value=mock_cert_credential, + ), + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_identity", + return_value=M365IdentityInfo( + identity_id=IDENTITY_ID, + identity_type="Service Principal with Certificate", + tenant_id=TENANT_ID, + tenant_domain=DOMAIN, + location=LOCATION, + certificate_thumbprint="ABC123DEF456", + ), + ), + patch( + "prowler.providers.m365.m365_provider.M365Provider.setup_powershell", + return_value=M365Credentials( + client_id=CLIENT_ID, + tenant_id=TENANT_ID, + certificate_content=certificate_content, + ), + ), + patch( + "prowler.providers.m365.m365_provider.print_boxes" + ) as mock_print_boxes, + ): + m365_provider = M365Provider( + certificate_auth=True, + region="M365Global", + ) + + m365_provider.print_credentials() + + # Verify print_boxes was called + mock_print_boxes.assert_called_once() + args, _ = mock_print_boxes.call_args + report_lines = args[0] + + # Check that certificate thumbprint is in the output + cert_line_found = any( + "Certificate Thumbprint" in line for line in report_lines + ) + assert ( + cert_line_found + ), "Certificate thumbprint should be in printed credentials" + + def test_validate_static_credentials_invalid_certificate_content(self): + """Test validate_static_credentials method with invalid certificate content""" + invalid_cert_content = "invalid_base64_content" + + with pytest.raises(RuntimeError) as exception: + M365Provider.validate_static_credentials( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_content=invalid_cert_content, + ) + + assert "An unexpected error occurred" in str(exception.value) + + def test_validate_static_credentials_invalid_certificate_path(self): + """Test validate_static_credentials method with invalid certificate path""" + invalid_cert_path = "/non/existent/path/cert.pem" + + with pytest.raises(M365NotValidCertificatePathError) as exception: + M365Provider.validate_static_credentials( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_path=invalid_cert_path, + ) + + assert "certificate path is not valid" in str(exception.value) + + def test_validate_static_credentials_certificate_base64_validation_error(self): + """Test validate_static_credentials method with invalid base64 certificate content""" + invalid_cert_content = "not_valid_base64!@#$%" + + with pytest.raises(M365NotValidCertificateContentError) as exception: + M365Provider.validate_static_credentials( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_content=invalid_cert_content, + ) + + assert "certificate content is not valid base64 encoded data" in str( + exception.value + ) + + def test_validate_static_credentials_certificate_path_file_not_found(self): + """Test validate_static_credentials method with certificate path that doesn't exist""" + invalid_cert_path = "/totally/non/existent/path/cert.pem" + + with pytest.raises(M365NotValidCertificatePathError) as exception: + M365Provider.validate_static_credentials( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_path=invalid_cert_path, + ) + + assert "certificate path is not valid" in str(exception.value) + + def test_validate_static_credentials_valid_certificate_content(self): + """Test validate_static_credentials method with valid certificate content""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + with patch( + "prowler.providers.m365.m365_provider.M365Provider.verify_client" + ) as mock_verify: + mock_verify.return_value = None + + result = M365Provider.validate_static_credentials( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_content=certificate_content, + ) + + assert result["tenant_id"] == TENANT_ID + assert result["client_id"] == CLIENT_ID + assert result["certificate_content"] == certificate_content + mock_verify.assert_called_once_with( + TENANT_ID, CLIENT_ID, None, certificate_content, None + ) + + @patch("builtins.open", mock_open(read_data=b"fake_certificate_data")) + def test_validate_static_credentials_valid_certificate_path(self): + """Test validate_static_credentials method with valid certificate path""" + certificate_path = "/path/to/cert.pem" + + with patch( + "prowler.providers.m365.m365_provider.M365Provider.verify_client" + ) as mock_verify: + mock_verify.return_value = None + + result = M365Provider.validate_static_credentials( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_path=certificate_path, + ) + + assert result["tenant_id"] == TENANT_ID + assert result["client_id"] == CLIENT_ID + assert result["certificate_path"] == certificate_path + mock_verify.assert_called_once_with( + TENANT_ID, CLIENT_ID, None, b"fake_certificate_data", certificate_path + ) + + @patch("prowler.providers.m365.m365_provider.asyncio.get_event_loop") + @patch("prowler.providers.m365.m365_provider.GraphServiceClient") + @patch("prowler.providers.m365.m365_provider.CertificateCredential") + def test_verify_client_certificate_content_success( + self, mock_cert_cred, mock_graph, mock_loop + ): + """Test verify_client method with valid certificate content""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + # Mock the async call + mock_loop_instance = MagicMock() + mock_loop.return_value = mock_loop_instance + mock_loop_instance.run_until_complete.return_value = [{"id": "domain.com"}] + + # Mock credential and graph client + mock_credential = MagicMock() + mock_cert_cred.return_value = mock_credential + mock_client = MagicMock() + mock_graph.return_value = mock_client + + # Should not raise any exception + M365Provider.verify_client( + TENANT_ID, CLIENT_ID, None, certificate_content, None + ) + + mock_cert_cred.assert_called_once() + mock_graph.assert_called_once_with(credentials=mock_credential) + + @patch("prowler.providers.m365.m365_provider.asyncio.get_event_loop") + @patch("prowler.providers.m365.m365_provider.GraphServiceClient") + @patch("prowler.providers.m365.m365_provider.CertificateCredential") + def test_verify_client_certificate_content_failure( + self, mock_cert_cred, mock_graph, mock_loop + ): + """Test verify_client method with certificate content that fails validation""" + certificate_content = base64.b64encode(b"fake_certificate").decode("utf-8") + + # Mock the async call to return empty result (invalid certificate) + mock_loop_instance = MagicMock() + mock_loop.return_value = mock_loop_instance + mock_loop_instance.run_until_complete.return_value = None + + # Mock credential and graph client + mock_credential = MagicMock() + mock_cert_cred.return_value = mock_credential + mock_client = MagicMock() + mock_graph.return_value = mock_client + + with pytest.raises(RuntimeError) as exception: + M365Provider.verify_client( + TENANT_ID, CLIENT_ID, None, certificate_content, None + ) + + assert "certificate content is not valid" in str(exception.value) + + @patch("builtins.open", mock_open(read_data=b"fake_certificate_data")) + @patch("prowler.providers.m365.m365_provider.asyncio.get_event_loop") + @patch("prowler.providers.m365.m365_provider.GraphServiceClient") + @patch("prowler.providers.m365.m365_provider.CertificateCredential") + def test_verify_client_certificate_path_success( + self, mock_cert_cred, mock_graph, mock_loop + ): + """Test verify_client method with valid certificate path""" + certificate_path = "/path/to/cert.pem" + + # Mock the async call + mock_loop_instance = MagicMock() + mock_loop.return_value = mock_loop_instance + mock_loop_instance.run_until_complete.return_value = [{"id": "domain.com"}] + + # Mock credential and graph client + mock_credential = MagicMock() + mock_cert_cred.return_value = mock_credential + mock_client = MagicMock() + mock_graph.return_value = mock_client + + # Should not raise any exception + M365Provider.verify_client(TENANT_ID, CLIENT_ID, None, None, certificate_path) + + mock_cert_cred.assert_called_once() + mock_graph.assert_called_once_with(credentials=mock_credential) + + @patch("builtins.open", mock_open(read_data=b"fake_certificate_data")) + @patch("prowler.providers.m365.m365_provider.asyncio.get_event_loop") + @patch("prowler.providers.m365.m365_provider.GraphServiceClient") + @patch("prowler.providers.m365.m365_provider.CertificateCredential") + def test_verify_client_certificate_path_failure( + self, mock_cert_cred, mock_graph, mock_loop + ): + """Test verify_client method with certificate path that fails validation""" + certificate_path = "/path/to/cert.pem" + + # Mock the async call to return empty result (invalid certificate) + mock_loop_instance = MagicMock() + mock_loop.return_value = mock_loop_instance + mock_loop_instance.run_until_complete.return_value = None + + # Mock credential and graph client + mock_credential = MagicMock() + mock_cert_cred.return_value = mock_credential + mock_client = MagicMock() + mock_graph.return_value = mock_client + + with pytest.raises(RuntimeError) as exception: + M365Provider.verify_client( + TENANT_ID, CLIENT_ID, None, None, certificate_path + ) + + assert "certificate is not valid" in str(exception.value) + + def test_setup_session_certificate_auth_with_certificate_path(self): + """Test setup_session method with certificate authentication using certificate path""" + certificate_path = "/path/to/cert.pem" + mock_cert_data = b"fake_certificate_data" + + with ( + patch("builtins.open", mock_open(read_data=mock_cert_data)), + patch("prowler.providers.m365.m365_provider.getenv") as mock_getenv, + patch( + "prowler.providers.m365.m365_provider.CertificateCredential" + ) as mock_cert_cred, + ): + mock_getenv.side_effect = lambda var: { + "AZURE_TENANT_ID": TENANT_ID, + "AZURE_CLIENT_ID": CLIENT_ID, + }.get(var) + + mock_credential = MagicMock() + mock_cert_cred.return_value = mock_credential + + result = M365Provider.setup_session( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=True, + certificate_path=certificate_path, + tenant_id=None, + m365_credentials=None, + region_config=MagicMock(), + ) + + assert result == mock_credential + mock_cert_cred.assert_called_once_with( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_data=mock_cert_data, + ) + + def test_setup_session_certificate_auth_with_static_credentials_certificate_path( + self, + ): + """Test setup_session method with certificate authentication using static credentials with certificate path""" + certificate_path = "/path/to/cert.pem" + mock_cert_data = b"fake_certificate_data" + + m365_credentials = { + "tenant_id": TENANT_ID, + "client_id": CLIENT_ID, + "client_secret": None, + "user": None, + "password": None, + "certificate_content": None, + "certificate_path": certificate_path, + } + + with ( + patch("builtins.open", mock_open(read_data=mock_cert_data)), + patch( + "prowler.providers.m365.m365_provider.CertificateCredential" + ) as mock_cert_cred, + ): + mock_credential = MagicMock() + mock_cert_cred.return_value = mock_credential + + result = M365Provider.setup_session( + az_cli_auth=False, + sp_env_auth=False, + env_auth=False, + browser_auth=False, + certificate_auth=False, + certificate_path=None, + tenant_id=None, + m365_credentials=m365_credentials, + region_config=MagicMock(), + ) + + assert result == mock_credential + mock_cert_cred.assert_called_once_with( + tenant_id=TENANT_ID, + client_id=CLIENT_ID, + certificate_data=mock_cert_data, + ) + + def test_check_certificate_creds_env_vars_with_certificate_content_success(self): + """Test check_certificate_creds_env_vars method with certificate content check""" + with patch("prowler.providers.m365.m365_provider.getenv") as mock_getenv: + mock_getenv.side_effect = lambda var: { + "AZURE_CLIENT_ID": CLIENT_ID, + "AZURE_TENANT_ID": TENANT_ID, + "M365_CERTIFICATE_CONTENT": "fake_certificate_content", + }.get(var) + + # Should not raise any exception + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=True + ) + + def test_check_certificate_creds_env_vars_without_certificate_content_success(self): + """Test check_certificate_creds_env_vars method without certificate content check""" + with patch("prowler.providers.m365.m365_provider.getenv") as mock_getenv: + mock_getenv.side_effect = lambda var: { + "AZURE_CLIENT_ID": CLIENT_ID, + "AZURE_TENANT_ID": TENANT_ID, + }.get(var) + + # Should not raise any exception + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=False + ) + + def test_check_certificate_creds_env_vars_missing_client_id_with_certificate(self): + """Test check_certificate_creds_env_vars method missing client ID with certificate content""" + with patch("prowler.providers.m365.m365_provider.getenv") as mock_getenv: + mock_getenv.side_effect = lambda var: { + "AZURE_TENANT_ID": TENANT_ID, + "M365_CERTIFICATE_CONTENT": "fake_certificate_content", + }.get(var) + + with pytest.raises(M365EnvironmentVariableError) as exception: + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=True + ) + + assert "Missing environment variable AZURE_CLIENT_ID" in str( + exception.value + ) + + def test_check_certificate_creds_env_vars_missing_tenant_id_with_certificate(self): + """Test check_certificate_creds_env_vars method missing tenant ID with certificate content""" + with patch("prowler.providers.m365.m365_provider.getenv") as mock_getenv: + mock_getenv.side_effect = lambda var: { + "AZURE_CLIENT_ID": CLIENT_ID, + "M365_CERTIFICATE_CONTENT": "fake_certificate_content", + }.get(var) + + with pytest.raises(M365EnvironmentVariableError) as exception: + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=True + ) + + assert "Missing environment variable AZURE_TENANT_ID" in str( + exception.value + ) + + def test_check_certificate_creds_env_vars_missing_certificate_content_when_required( + self, + ): + """Test check_certificate_creds_env_vars method missing certificate content when required""" + with patch("prowler.providers.m365.m365_provider.getenv") as mock_getenv: + mock_getenv.side_effect = lambda var: { + "AZURE_CLIENT_ID": CLIENT_ID, + "AZURE_TENANT_ID": TENANT_ID, + }.get(var) + + with pytest.raises(M365EnvironmentVariableError) as exception: + M365Provider.check_certificate_creds_env_vars( + check_certificate_content=True + ) + + assert "Missing environment variable M365_CERTIFICATE_CONTENT" in str( + exception.value + )