feat(azure): add authentication method from static credentials (#5358)

This commit is contained in:
Pedro Martín
2024-10-15 09:37:17 +02:00
committed by GitHub
parent c1d061ef70
commit bc1e6c0626
3 changed files with 540 additions and 42 deletions
+358 -42
View File
@@ -1,11 +1,18 @@
import asyncio
import os
import re
from argparse import ArgumentTypeError
from os import getenv
from uuid import UUID
import requests
from azure.core.exceptions import HttpResponseError
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
from azure.core.exceptions import ClientAuthenticationError, HttpResponseError
from azure.identity import (
ClientSecretCredential,
CredentialUnavailableError,
DefaultAzureCredential,
InteractiveBrowserCredential,
)
from azure.mgmt.subscription import SubscriptionClient
from colorama import Fore, Style
from msgraph import GraphServiceClient
@@ -16,14 +23,26 @@ from prowler.lib.utils.utils import print_boxes
from prowler.providers.azure.exceptions.exceptions import (
AzureArgumentTypeValidationError,
AzureBrowserAuthNoTenantIDError,
AzureClientAuthenticationError,
AzureClientIdAndClientSecretNotBelongingToTenantIdError,
AzureConfigCredentialsError,
AzureCredentialsUnavailableError,
AzureDefaultAzureCredentialError,
AzureEnvironmentVariableError,
AzureGetTokenIdentityError,
AzureHTTPResponseError,
AzureInteractiveBrowserCredentialError,
AzureNoAuthenticationMethodError,
AzureNoSubscriptionsError,
AzureNotTenantIdButClientIdAndClienSecret,
AzureNotValidClientIdError,
AzureNotValidClientSecretError,
AzureNotValidTenantIdError,
AzureSetUpIdentityError,
AzureSetUpRegionConfigError,
AzureSetUpSessionError,
AzureTenantIdAndClientIdNotBelongingToClientSecretError,
AzureTenantIdAndClientSecretNotBelongingToClientIdError,
AzureTenantIDNoBrowserAuthError,
)
from prowler.providers.azure.lib.arguments.arguments import validate_azure_region
@@ -91,6 +110,8 @@ class AzureProvider(Provider):
subscription_ids: list = [],
audit_config: dict = {},
fixer_config: dict = {},
client_id: str = None,
client_secret: str = None,
):
"""
Initializes the Azure provider.
@@ -105,6 +126,8 @@ class AzureProvider(Provider):
subscription_ids (list): List of subscription IDs.
audit_config (dict): The audit configuration for the Azure provider.
fixer_config (dict): The fixer configuration.
client_id (str): The Azure client ID.
client_secret (str): The Azure client secret.
Returns:
None
@@ -114,6 +137,9 @@ class AzureProvider(Provider):
AzureSetUpRegionConfigError: If there is an error in setting up the region configuration.
AzureDefaultAzureCredentialError: If there is an error in retrieving the Azure credentials.
AzureInteractiveBrowserCredentialError: If there is an error in retrieving the Azure credentials using browser authentication.
AzureConfigCredentialsError: If there is an error in configuring the Azure credentials from a dictionary.
AzureGetTokenIdentityError: If there is an error in getting the token from the Azure identity.
AzureHTTPResponseError: If there is an HTTP response error.
"""
logger.info("Setting Azure provider ...")
@@ -121,12 +147,25 @@ class AzureProvider(Provider):
# Validate the authentication arguments
self.validate_arguments(
az_cli_auth, sp_env_auth, browser_auth, managed_identity_auth, tenant_id
az_cli_auth,
sp_env_auth,
browser_auth,
managed_identity_auth,
tenant_id,
client_id,
client_secret,
)
logger.info("Checking if region is different than default one")
self._region_config = self.setup_region_config(region)
# Get the dict from the static credentials
azure_credentials = None
if tenant_id and client_id and client_secret:
azure_credentials = self.validate_static_credentials(
tenant_id=tenant_id, client_id=client_id, client_secret=client_secret
)
# Set up the Azure session
self._session = self.setup_session(
az_cli_auth,
@@ -134,6 +173,7 @@ class AzureProvider(Provider):
browser_auth,
managed_identity_auth,
tenant_id,
azure_credentials,
self._region_config,
)
@@ -144,6 +184,7 @@ class AzureProvider(Provider):
browser_auth,
managed_identity_auth,
subscription_ids,
client_id,
)
# TODO: should we keep this here or within the identity?
@@ -236,6 +277,8 @@ class AzureProvider(Provider):
browser_auth: bool,
managed_identity_auth: bool,
tenant_id: str,
client_id: str,
client_secret: str,
):
"""
Validates the authentication arguments for the Azure provider.
@@ -246,30 +289,40 @@ class AzureProvider(Provider):
browser_auth (bool): Flag indicating whether browser authentication is enabled.
managed_identity_auth (bool): Flag indicating whether managed identity authentication is enabled.
tenant_id (str): The Azure Tenant ID.
client_id (str): The Azure Client ID.
client_secret (str): The Azure Client Secret.
Raises:
AzureBrowserAuthNoTenantIDError: If browser authentication is enabled but the tenant ID is not found.
"""
if not browser_auth and tenant_id:
raise AzureTenantIDNoBrowserAuthError(
file=os.path.basename(__file__),
message="Azure Tenant ID (--tenant-id) is required for browser authentication mode",
)
elif (
not az_cli_auth
and not sp_env_auth
and not browser_auth
and not managed_identity_auth
):
raise AzureNoAuthenticationMethodError(
file=os.path.basename(__file__),
message="Azure provider requires at least one authentication method set: [--az-cli-auth | --sp-env-auth | --browser-auth | --managed-identity-auth]",
)
elif browser_auth and not tenant_id:
raise AzureBrowserAuthNoTenantIDError(
file=os.path.basename(__file__),
message="Azure Tenant ID (--tenant-id) is required for browser authentication mode",
)
if not client_id and not client_secret:
if not browser_auth and tenant_id:
raise AzureTenantIDNoBrowserAuthError(
file=os.path.basename(__file__),
message="Azure Tenant ID (--tenant-id) is required for browser authentication mode",
)
elif (
not az_cli_auth
and not sp_env_auth
and not browser_auth
and not managed_identity_auth
):
raise AzureNoAuthenticationMethodError(
file=os.path.basename(__file__),
message="Azure provider requires at least one authentication method set: [--az-cli-auth | --sp-env-auth | --browser-auth | --managed-identity-auth]",
)
elif browser_auth and not tenant_id:
raise AzureBrowserAuthNoTenantIDError(
file=os.path.basename(__file__),
message="Azure Tenant ID (--tenant-id) is required for browser authentication mode",
)
else:
if not tenant_id:
raise AzureNotTenantIdButClientIdAndClienSecret(
file=os.path.basename(__file__),
message="Tenant Id is required for Azure static credentials. Make sure you are using the correct credentials.",
)
@staticmethod
def setup_region_config(region):
@@ -346,6 +399,7 @@ class AzureProvider(Provider):
browser_auth: bool,
managed_identity_auth: bool,
tenant_id: str,
azure_credentials: dict,
region_config: AzureRegionConfig,
):
"""Returns the Azure credentials object.
@@ -358,6 +412,10 @@ class AzureProvider(Provider):
browser_auth (bool): Flag indicating whether to use interactive browser authentication.
managed_identity_auth (bool): Flag indicating whether to use managed identity authentication.
tenant_id (str): The Azure Active Directory tenant ID.
azure_credentials (dict): The Azure configuration object. It contains the following keys:
- tenant_id: The Azure Active Directory tenant ID.
- client_id: The Azure client ID.
- client_secret: The Azure client secret
region_config (AzureRegionConfig): The region configuration object.
Returns:
@@ -378,27 +436,80 @@ class AzureProvider(Provider):
)
raise environment_credentials_error
try:
# Since the input vars come as True when it is wanted to be used, we need to inverse it since
# DefaultAzureCredential sets the auth method excluding the others
credentials = DefaultAzureCredential(
exclude_environment_credential=not sp_env_auth,
exclude_cli_credential=not az_cli_auth,
exclude_managed_identity_credential=not managed_identity_auth,
# Azure Auth using Visual Studio is not supported
exclude_visual_studio_code_credential=True,
# Azure Auth using Shared Token Cache is not supported
exclude_shared_token_cache_credential=True,
# Azure Auth using PowerShell is not supported
exclude_powershell_credential=True,
# set Authority of a Microsoft Entra endpoint
authority=region_config.authority,
)
if azure_credentials:
try:
credentials = ClientSecretCredential(
tenant_id=azure_credentials["tenant_id"],
client_id=azure_credentials["client_id"],
client_secret=azure_credentials["client_secret"],
)
return credentials
except ClientAuthenticationError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureClientAuthenticationError(
file=os.path.basename(__file__), original_exception=error
)
except CredentialUnavailableError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureCredentialsUnavailableError(
file=os.path.basename(__file__), original_exception=error
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureConfigCredentialsError(
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,
exclude_cli_credential=not az_cli_auth,
exclude_managed_identity_credential=not managed_identity_auth,
# Azure Auth using Visual Studio is not supported
exclude_visual_studio_code_credential=True,
# Azure Auth using Shared Token Cache is not supported
exclude_shared_token_cache_credential=True,
# Azure Auth using PowerShell is not supported
exclude_powershell_credential=True,
# set Authority of a Microsoft Entra endpoint
authority=region_config.authority,
)
except ClientAuthenticationError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureClientAuthenticationError(
file=os.path.basename(__file__), original_exception=error
)
except CredentialUnavailableError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureCredentialsUnavailableError(
file=os.path.basename(__file__), original_exception=error
)
except Exception as error:
logger.critical("Failed to retrieve azure credentials")
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureDefaultAzureCredentialError(
file=os.path.basename(__file__), original_exception=error
)
except Exception as error:
logger.critical("Failed to retrieve azure credentials")
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureDefaultAzureCredentialError(
raise AzureSetUpSessionError(
file=os.path.basename(__file__), original_exception=error
)
else:
@@ -426,6 +537,8 @@ class AzureProvider(Provider):
tenant_id=None,
region="AzureCloud",
raise_on_exception=True,
client_id=None,
client_secret=None,
) -> Connection:
"""Test connection to Azure subscription.
@@ -439,6 +552,8 @@ class AzureProvider(Provider):
tenant_id (str): The Azure Active Directory tenant ID.
region (str): The Azure region.
raise_on_exception (bool): Flag indicating whether to raise an exception if the connection fails.
client_id (str): The Azure client ID.
client_secret (str): The Azure client secret.
Returns:
bool: True if the connection is successful, False otherwise.
@@ -450,6 +565,7 @@ class AzureProvider(Provider):
AzureDefaultAzureCredentialError: If there is an error in retrieving the Azure credentials.
AzureInteractiveBrowserCredentialError: If there is an error in retrieving the Azure credentials using browser authentication.
AzureHTTPResponseError: If there is an HTTP response error.
AzureConfigCredentialsError: If there is an error in configuring the Azure credentials from a dictionary.
Examples:
@@ -457,12 +573,30 @@ class AzureProvider(Provider):
True
>>> AzureProvider.test_connection(sp_env_auth=False, browser_auth=True, tenant_id=None)
False, ArgumentTypeError: Azure Tenant ID is required only for browser authentication mode
>>> AzureProvider.test_connection(tenant_id="XXXXXXXXXX", client_id="XXXXXXXXXX", client_secret="XXXXXXXXXX")
True
"""
try:
AzureProvider.validate_arguments(
az_cli_auth, sp_env_auth, browser_auth, managed_identity_auth, tenant_id
az_cli_auth,
sp_env_auth,
browser_auth,
managed_identity_auth,
tenant_id,
client_id,
client_secret,
)
region_config = AzureProvider.setup_region_config(region)
# Get the dict from the static credentials
azure_credentials = None
if tenant_id and client_id and client_secret:
azure_credentials = AzureProvider.validate_static_credentials(
tenant_id=tenant_id,
client_id=client_id,
client_secret=client_secret,
)
# Set up the Azure session
credentials = AzureProvider.setup_session(
az_cli_auth,
@@ -470,6 +604,7 @@ class AzureProvider(Provider):
browser_auth,
managed_identity_auth,
tenant_id,
azure_credentials,
region_config,
)
# Create a SubscriptionClient
@@ -521,6 +656,47 @@ class AzureProvider(Provider):
if raise_on_exception:
raise interactive_browser_error
return Connection(error=interactive_browser_error)
except AzureConfigCredentialsError as config_credentials_error:
logger.error(str(config_credentials_error))
if raise_on_exception:
raise config_credentials_error
return Connection(error=config_credentials_error)
except AzureClientAuthenticationError as client_auth_error:
logger.error(str(client_auth_error))
if raise_on_exception:
raise client_auth_error
return Connection(error=client_auth_error)
except AzureCredentialsUnavailableError as credential_unavailable_error:
logger.error(str(credential_unavailable_error))
if raise_on_exception:
raise credential_unavailable_error
return Connection(error=credential_unavailable_error)
except AzureDefaultAzureCredentialError as default_credentials_error:
logger.error(str(default_credentials_error))
if raise_on_exception:
raise default_credentials_error
return Connection(error=default_credentials_error)
except (
AzureClientIdAndClientSecretNotBelongingToTenantIdError
) as tenant_id_error:
logger.error(str(tenant_id_error))
if raise_on_exception:
raise tenant_id_error
return Connection(error=tenant_id_error)
except (
AzureTenantIdAndClientSecretNotBelongingToClientIdError
) as client_id_error:
logger.error(str(client_id_error))
if raise_on_exception:
raise client_id_error
return Connection(error=client_id_error)
except (
AzureTenantIdAndClientIdNotBelongingToClientSecretError
) as client_secret_error:
logger.error(str(client_secret_error))
if raise_on_exception:
raise client_secret_error
return Connection(error=client_secret_error)
# Exceptions from SubscriptionClient
except HttpResponseError as http_response_error:
logger.error(
@@ -573,6 +749,7 @@ class AzureProvider(Provider):
browser_auth,
managed_identity_auth,
subscription_ids,
client_id,
):
"""
Sets up the identity for the Azure provider.
@@ -595,7 +772,7 @@ class AzureProvider(Provider):
# the identity can access AAD and retrieve the tenant domain name.
# With cli also should be possible but right now it does not work, azure python package issue is coming
# At the time of writting this with az cli creds is not working, despite that is included
if sp_env_auth or browser_auth or az_cli_auth:
if sp_env_auth or browser_auth or az_cli_auth or client_id:
async def get_azure_identity():
# Trying to recover tenant domain info
@@ -610,12 +787,28 @@ class AzureProvider(Provider):
if getattr(domain_result.value[0], "id"):
identity.tenant_domain = domain_result.value[0].id
except HttpResponseError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureHTTPResponseError(
file=os.path.basename(__file__),
original_exception=error,
)
except ClientAuthenticationError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise AzureGetTokenIdentityError(
file=os.path.basename(__file__),
original_exception=error,
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
# since that exception is not considered as critical, we keep filling another identity fields
if sp_env_auth:
if sp_env_auth or client_id:
# The id of the sp can be retrieved from environment variables
identity.identity_id = getenv("AZURE_CLIENT_ID")
identity.identity_type = "Service Principal"
@@ -730,3 +923,126 @@ class AzureProvider(Provider):
for location in data["value"]:
locations[display_name].append(location["name"])
return locations
@staticmethod
def validate_static_credentials(
tenant_id: str = None, client_id: str = None, client_secret: str = None
) -> dict:
"""
Validates the static credentials for the Azure provider.
Args:
tenant_id (str): The Azure Active Directory tenant ID.
client_id (str): The Azure client ID.
client_secret (str): The Azure client secret.
Raises:
AzureNotValidTenantIdError: If the provided Azure Tenant ID is not valid.
AzureNotValidClientIdError: If the provided Azure Client ID is not valid.
AzureNotValidClientSecretError: If the provided Azure Client Secret is not valid.
AzureClientIdAndClientSecretNotBelongingToTenantIdError: If the provided Azure Client ID and Client Secret do not belong to the specified Tenant ID.
AzureTenantIdAndClientSecretNotBelongingToClientIdError: If the provided Azure Tenant ID and Client Secret do not belong to the specified Client ID.
AzureTenantIdAndClientIdNotBelongingToClientSecretError: If the provided Azure Tenant ID and Client ID do not belong to the specified Client Secret.
Returns:
dict: A dictionary containing the validated static credentials.
"""
# Validate the Tenant ID
try:
UUID(tenant_id)
except ValueError:
raise AzureNotValidTenantIdError(
file=os.path.basename(__file__),
message="The provided Azure Tenant ID is not valid.",
)
# Validate the Client ID
try:
UUID(client_id)
except ValueError:
raise AzureNotValidClientIdError(
file=os.path.basename(__file__),
message="The provided Azure Client ID is not valid.",
)
# Validate the Client Secret
if not re.match("^[a-zA-Z0-9._~-]+$", client_secret):
raise AzureNotValidClientSecretError(
file=os.path.basename(__file__),
message="The provided Azure Client Secret is not valid.",
)
try:
AzureProvider.verify_client(tenant_id, client_id, client_secret)
return {
"tenant_id": tenant_id,
"client_id": client_id,
"client_secret": client_secret,
}
except AzureNotValidTenantIdError as tenant_id_error:
logger.error(str(tenant_id_error))
raise AzureClientIdAndClientSecretNotBelongingToTenantIdError(
file=os.path.basename(__file__),
message="The provided Azure Client ID and Client Secret do not belong to the specified Tenant ID.",
)
except AzureNotValidClientIdError as client_id_error:
logger.error(str(client_id_error))
raise AzureTenantIdAndClientSecretNotBelongingToClientIdError(
file=os.path.basename(__file__),
message="The provided Azure Tenant ID and Client Secret do not belong to the specified Client ID.",
)
except AzureNotValidClientSecretError as client_secret_error:
logger.error(str(client_secret_error))
raise AzureTenantIdAndClientIdNotBelongingToClientSecretError(
file=os.path.basename(__file__),
message="The provided Azure Tenant ID and Client ID do not belong to the specified Client Secret.",
)
@staticmethod
def verify_client(tenant_id, client_id, client_secret) -> None:
"""
Verifies the Azure client credentials using the specified tenant ID, client ID, and client secret.
Args:
tenant_id (str): The Azure Active Directory tenant ID.
client_id (str): The Azure client ID.
client_secret (str): The Azure client secret.
Raises:
AzureNotValidTenantIdError: If the provided Azure Tenant ID is not valid.
AzureNotValidClientIdError: If the provided Azure Client ID is not valid.
AzureNotValidClientSecretError: If the provided Azure Client Secret is not valid.
Returns:
None
"""
url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json",
}
data = {
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": "https://graph.microsoft.com/.default",
}
response = requests.post(url, headers=headers, data=data).json()
if "access_token" not in response.keys() and "error_codes" in response.keys():
if f"Tenant '{tenant_id}'" in response["error_description"]:
raise AzureNotValidTenantIdError(
file=os.path.basename(__file__),
message="The provided Azure Tenant ID is not valid for the specified Client ID and Client Secret.",
)
if (
f"Application with identifier '{client_id}'"
in response["error_description"]
):
raise AzureNotValidClientIdError(
file=os.path.basename(__file__),
message="The provided Azure Client ID is not valid for the specified Tenant ID and Client Secret.",
)
if "Invalid client secret provided" in response["error_description"]:
raise AzureNotValidClientSecretError(
file=os.path.basename(__file__),
message="The provided Azure Client Secret is not valid for the specified Tenant ID and Client ID.",
)
@@ -49,6 +49,54 @@ class AzureBaseException(ProwlerException):
"message": "Error in HTTP response from Azure",
"remediation": "",
},
(1925, "AzureCredentialsUnavailableError"): {
"message": "Error trying to configure Azure credentials because they are unavailable",
"remediation": "Check the dictionary and ensure it is properly set up for Azure credentials. TENANT_ID, CLIENT_ID and CLIENT_SECRET are required.",
},
(1926, "AzureGetTokenIdentityError"): {
"message": "Error trying to get token from Azure Identity",
"remediation": "Check the Azure Identity and ensure it is properly set up.",
},
(1927, "AzureNotTenantIdButClientIdAndClienSecret"): {
"message": "The provided credentials are not a tenant ID but a client ID and client secret",
"remediation": "Tenant Id, Client Id and Client Secret are required for Azure credentials. Make sure you are using the correct credentials.",
},
(1928, "AzureClientAuthenticationError"): {
"message": "Error in client authentication",
"remediation": "Check the client authentication and ensure it is properly set up.",
},
(1929, "AzureSetUpSessionError"): {
"message": "Error setting up session",
"remediation": "Check the session setup and ensure it is properly set up.",
},
(1930, "AzureNotValidTenantIdError"): {
"message": "The provided tenant ID is not valid",
"remediation": "Check the tenant ID and ensure it is a valid ID.",
},
(1931, "AzureNotValidClientIdError"): {
"message": "The provided client ID is not valid",
"remediation": "Check the client ID and ensure it is a valid ID.",
},
(1932, "AzureNotValidClientSecretError"): {
"message": "The provided client secret is not valid",
"remediation": "Check the client secret and ensure it is a valid secret.",
},
(1933, "AzureConfigCredentialsError"): {
"message": "Error in configuration of Azure credentials",
"remediation": "Check the configuration of Azure credentials and ensure it is properly set up.",
},
(1934, "AzureClientIdAndClientSecretNotBelongingToTenantIdError"): {
"message": "The provided client ID and client secret do not belong to the provided tenant ID",
"remediation": "Check the client ID and client secret and ensure they belong to the provided tenant ID.",
},
(1935, "AzureTenantIdAndClientSecretNotBelongingToClientIdError"): {
"message": "The provided tenant ID and client secret do not belong to the provided client ID",
"remediation": "Check the tenant ID and client secret and ensure they belong to the provided client ID.",
},
(1936, "AzureTenantIdAndClientIdNotBelongingToClientSecretError"): {
"message": "The provided tenant ID and client ID do not belong to the provided client secret",
"remediation": "Check the tenant ID and client ID and ensure they belong to the provided client secret.",
},
}
def __init__(self, code, file=None, original_exception=None, message=None):
@@ -147,3 +195,87 @@ class AzureHTTPResponseError(AzureBaseException):
super().__init__(
1924, file=file, original_exception=original_exception, message=message
)
class AzureCredentialsUnavailableError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1925, file=file, original_exception=original_exception, message=message
)
class AzureGetTokenIdentityError(AzureBaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1926, file=file, original_exception=original_exception, message=message
)
class AzureNotTenantIdButClientIdAndClienSecret(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1927, file=file, original_exception=original_exception, message=message
)
class AzureClientAuthenticationError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1928, file=file, original_exception=original_exception, message=message
)
class AzureSetUpSessionError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1929, file=file, original_exception=original_exception, message=message
)
class AzureNotValidTenantIdError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1930, file=file, original_exception=original_exception, message=message
)
class AzureNotValidClientIdError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1931, file=file, original_exception=original_exception, message=message
)
class AzureNotValidClientSecretError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1932, file=file, original_exception=original_exception, message=message
)
class AzureConfigCredentialsError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1933, file=file, original_exception=original_exception, message=message
)
class AzureClientIdAndClientSecretNotBelongingToTenantIdError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1934, file=file, original_exception=original_exception, message=message
)
class AzureTenantIdAndClientSecretNotBelongingToClientIdError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1935, file=file, original_exception=original_exception, message=message
)
class AzureTenantIdAndClientIdNotBelongingToClientSecretError(AzureCredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
1936, file=file, original_exception=original_exception, message=message
)
@@ -31,6 +31,8 @@ class TestAzureProvider:
sp_env_auth = None
browser_auth = None
managed_identity_auth = None
client_id = None
client_secret = None
audit_config = load_and_validate_config_file("azure", default_config_file_path)
fixer_config = load_and_validate_config_file(
@@ -55,6 +57,8 @@ class TestAzureProvider:
subscription_id,
audit_config=audit_config,
fixer_config=fixer_config,
client_id=client_id,
client_secret=client_secret,
)
assert azure_provider.region_config == AzureRegionConfig(
@@ -230,6 +234,52 @@ class TestAzureProvider:
assert test_connection.is_connected
assert test_connection.error is None
def test_test_connection_tenant_id_client_id_client_secret(self):
with patch(
"prowler.providers.azure.azure_provider.DefaultAzureCredential"
) as mock_default_credential, patch(
"prowler.providers.azure.azure_provider.AzureProvider.setup_session"
) as mock_setup_session, patch(
"prowler.providers.azure.azure_provider.SubscriptionClient"
) as mock_resource_client, patch(
"prowler.providers.azure.azure_provider.AzureProvider.validate_static_credentials"
) as mock_validate_static_credentials:
# Mock the return value of DefaultAzureCredential
mock_credentials = MagicMock()
mock_credentials.get_token.return_value = AccessToken(
token="fake_token", expires_on=9999999999
)
mock_default_credential.return_value = {
"client_id": str(uuid4()),
"client_secret": str(uuid4()),
"tenant_id": str(uuid4()),
}
# Mock setup_session to return a mocked session object
mock_session = MagicMock()
mock_setup_session.return_value = mock_session
# Mock ValidateStaticCredentials to avoid real API calls
mock_validate_static_credentials.return_value = None
# Mock ResourceManagementClient to avoid real API calls
mock_client = MagicMock()
mock_resource_client.return_value = mock_client
test_connection = AzureProvider.test_connection(
browser_auth=False,
tenant_id=str(uuid4()),
region="AzureCloud",
raise_on_exception=False,
client_id=str(uuid4()),
client_secret=str(uuid4()),
)
assert isinstance(test_connection, Connection)
assert test_connection.is_connected
assert test_connection.error is None
def test_test_connection_with_ClientAuthenticationError(self):
with pytest.raises(AzureHTTPResponseError) as exception:
tenant_id = str(uuid4())