feat(m365): add Microsoft 365 provider (#5902)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>
Co-authored-by: HugoPBrito <hugopbrit@gmail.com>
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>
This commit is contained in:
Mario Rodriguez Lopez
2025-01-24 19:14:17 +01:00
committed by GitHub
parent bcc246d950
commit ccdb54d7c3
50 changed files with 3173 additions and 8 deletions
+36 -1
View File
@@ -51,6 +51,7 @@ from prowler.lib.outputs.compliance.cis.cis_aws import AWSCIS
from prowler.lib.outputs.compliance.cis.cis_azure import AzureCIS
from prowler.lib.outputs.compliance.cis.cis_gcp import GCPCIS
from prowler.lib.outputs.compliance.cis.cis_kubernetes import KubernetesCIS
from prowler.lib.outputs.compliance.cis.cis_microsoft365 import Microsoft365CIS
from prowler.lib.outputs.compliance.compliance import display_compliance_table
from prowler.lib.outputs.compliance.ens.ens_aws import AWSENS
from prowler.lib.outputs.compliance.ens.ens_azure import AzureENS
@@ -78,6 +79,7 @@ from prowler.providers.common.provider import Provider
from prowler.providers.common.quick_inventory import run_provider_quick_inventory
from prowler.providers.gcp.models import GCPOutputOptions
from prowler.providers.kubernetes.models import KubernetesOutputOptions
from prowler.providers.microsoft365.models import Microsoft365OutputOptions
def prowler():
@@ -259,6 +261,10 @@ def prowler():
output_options = KubernetesOutputOptions(
args, bulk_checks_metadata, global_provider.identity
)
elif provider == "microsoft365":
output_options = Microsoft365OutputOptions(
args, bulk_checks_metadata, global_provider.identity
)
# Run the quick inventory for the provider if available
if hasattr(args, "quick_inventory") and args.quick_inventory:
@@ -307,7 +313,6 @@ def prowler():
if "SLACK_API_TOKEN" in environ and (
"SLACK_CHANNEL_NAME" in environ or "SLACK_CHANNEL_ID" in environ
):
token = environ["SLACK_API_TOKEN"]
channel = (
environ["SLACK_CHANNEL_NAME"]
@@ -629,6 +634,36 @@ def prowler():
generated_outputs["compliance"].append(generic_compliance)
generic_compliance.batch_write_data_to_file()
elif provider == "microsoft365":
for compliance_name in input_compliance_frameworks:
if compliance_name.startswith("cis_"):
# Generate CIS Finding Object
filename = (
f"{output_options.output_directory}/compliance/"
f"{output_options.output_filename}_{compliance_name}.csv"
)
cis = Microsoft365CIS(
findings=finding_outputs,
compliance=bulk_compliance_frameworks[compliance_name],
create_file_descriptor=True,
file_path=filename,
)
generated_outputs["compliance"].append(cis)
cis.batch_write_data_to_file()
else:
filename = (
f"{output_options.output_directory}/compliance/"
f"{output_options.output_filename}_{compliance_name}.csv"
)
generic_compliance = GenericCompliance(
findings=finding_outputs,
compliance=bulk_compliance_frameworks[compliance_name],
create_file_descriptor=True,
file_path=filename,
)
generated_outputs["compliance"].append(generic_compliance)
generic_compliance.batch_write_data_to_file()
# AWS Security Hub Integration
if provider == "aws":
# Send output to S3 if needed (-B / -D) for all the output formats
@@ -0,0 +1,134 @@
{
"Framework": "CIS",
"Version": "4.0",
"Provider": "Microsoft365",
"Description": "The CIS Microsoft 365 Foundations Benchmark provides prescriptive guidance for establishing a secure configuration posture for Microsoft 365 Cloud offerings running on any OS.",
"Requirements": [
{
"Id": "1.1.1",
"Description": "Ensure Administrative accounts are cloud-only",
"Checks": [],
"Attributes": [
{
"Section": "1.Microsoft 365 admin center",
"Profile": "Level 1",
"AssessmentStatus": "Manual",
"Description": "Administrative accounts are special privileged accounts that could have varying levels of access to data, users, and settings. Regular user accounts should never be utilized for administrative tasks and care should be taken, in the case of a hybrid environment, to keep Administrative accounts separated from on-prem accounts. Administrative accounts should not have applications assigned so that they have no access to potentially vulnerable services (e.g., email, Teams, SharePoint, etc.) and only access to perform tasks as needed for administrative purposes. Ensure administrative accounts are not On-premises sync enabled.",
"RationaleStatement": "In a hybrid environment, having separate accounts will help ensure that in the event of a breach in the cloud, that the breach does not affect the on-prem environment and vice versa.",
"ImpactStatement": "Administrative users will have to switch accounts and utilize login/logout functionality when performing administrative tasks, as well as not benefiting from SSO. This will require a migration process from the 'daily driver' account to a dedicated admin account. When migrating permissions to the admin account, both M365 and Azure RBAC roles should be migrated as well. Once the new admin accounts are created, both of these permission sets should be moved from the daily driver account to the new admin account. Failure to migrate Azure RBAC roles can cause an admin to not be able to see their subscriptions/resources while using their admin accounts.",
"RemediationProcedure": "Review all administrative accounts and ensure they are configured as cloud-only. Remove any on-premises synchronization for these accounts. Assign necessary roles and permissions exclusively to the dedicated cloud administrative accounts.",
"AuditProcedure": "Log in to the Microsoft 365 Admin Center and review the list of administrative accounts. Verify that none of them are on-premises sync enabled.",
"AdditionalInformation": "This recommendation is particularly relevant for hybrid environments and is aimed at enhancing the security of administrative accounts by isolating them from on-prem infrastructure.",
"DefaultValue": "By default, administrative accounts may be either cloud-only or hybrid. This setting needs to be verified and adjusted according to the recommendation.",
"References": "CIS Microsoft 365 Foundations Benchmark v4.0, Section 1.1.1"
}
]
},
{
"Id": "1.1.2",
"Description": "Ensure two emergency access accounts have been defined",
"Checks": [],
"Attributes": [
{
"Section": "1.Microsoft 365 admin center",
"Profile": "Level 1",
"AssessmentStatus": "Manual",
"Description": "Emergency access or 'break glass' accounts are limited for emergency scenarios where normal administrative accounts are unavailable. They are not assigned to a specific user and will have a combination of physical and technical controls to prevent them from being accessed outside a true emergency. These emergencies could include technical failures of a cellular provider or Microsoft-related service such as MFA, or the last remaining Global Administrator account becoming inaccessible. Ensure two Emergency Access accounts have been defined.",
"RationaleStatement": "In various situations, an organization may require the use of a break glass account to gain emergency access. Losing access to administrative functions could result in a significant loss of support capability, reduced visibility into the security posture, and potential financial losses.",
"ImpactStatement": "Improper implementation of emergency access accounts could weaken the organization's security posture. To mitigate this, at least one account should be excluded from all conditional access rules, and strong authentication mechanisms (e.g., long, high-entropy passwords or FIDO2 security keys) must be used to secure the accounts.",
"RemediationProcedure": "Create two emergency access accounts and configure them according to Microsoft's recommendations. Ensure that these accounts are not assigned to specific users and are excluded from all conditional access rules. Secure the accounts with strong passwords or passwordless authentication methods, such as FIDO2 security keys. Regularly review and test these accounts to confirm their functionality.",
"AuditProcedure": "Log in to the Microsoft 365 Admin Center and verify the existence of at least two emergency access accounts. Check their configurations to ensure they comply with Microsoft's recommendations, including exclusion from conditional access rules and proper security settings.",
"AdditionalInformation": "Emergency access accounts are critical for maintaining administrative control during unexpected events. Regular audits and strict access controls are recommended to prevent misuse.",
"DefaultValue": "By default, emergency access accounts are not configured. Organizations must create and secure these accounts proactively.",
"References": "CIS Microsoft 365 Foundations Benchmark v4.0, Section 1.1.2; Microsoft documentation on emergency access accounts."
}
]
},
{
"Id": "1.1.3",
"Description": "Ensure that between two and four global admins are designated",
"Checks": [
"admincenter_users_between_two_and_four_global_admins"
],
"Attributes": [
{
"Section": "1.Microsoft 365 admin center",
"Profile": "Level 1",
"AssessmentStatus": "Automated",
"Description": "More than one global administrator should be designated so a single admin can be monitored and to provide redundancy should a single admin leave an organization. Additionally, there should be no more than four global admins set for any tenant. Ideally, global administrators will have no licenses assigned to them.",
"RationaleStatement": "If there is only one global tenant administrator, he or she can perform malicious activity without the possibility of being discovered by another admin. If there are numerous global tenant administrators, the more likely it is that one of their accounts will be successfully breached by an external attacker.",
"ImpactStatement": "If there is only one global administrator in a tenant, an additional global administrator will need to be identified and configured. If there are more than four global administrators, a review of role requirements for current global administrators will be required to identify which of the users require global administrator access.",
"RemediationProcedure": "Review the list of global administrators in the tenant and ensure there are at least two but no more than four accounts assigned this role. Remove excess global administrator accounts and create additional ones if necessary. Avoid assigning licenses to these accounts.",
"AuditProcedure": "Log in to the Microsoft 365 Admin Center and review the list of global administrators. Verify that there are at least two but no more than four global administrators configured.",
"AdditionalInformation": "Global administrators play a critical role in tenant management. Ensuring a proper number of global administrators improves redundancy and security.",
"DefaultValue": "By default, there may be a single global administrator configured for the tenant. Organizations need to manually adjust the count as per best practices.",
"References": "CIS Microsoft 365 Foundations Benchmark v4.0, Section 1.1.3"
}
]
},
{
"Id": "1.1.4",
"Description": "Ensure administrative accounts use licenses with a reduced application footprint",
"Checks": [
"admincenter_users_admins_reduced_license_footprint"
],
"Attributes": [
{
"Section": "1.Microsoft 365 admin center",
"Profile": "Level 1",
"AssessmentStatus": "Automated",
"Description": "Administrative accounts are special privileged accounts with varying levels of access to data, users, and settings. It is recommended that privileged accounts either not be licensed or use Microsoft Entra ID P1 or Microsoft Entra ID P2 licenses to minimize application exposure.",
"RationaleStatement": "Ensuring administrative accounts do not use licenses with applications assigned to them reduces the attack surface of high-privileged identities. This minimizes the likelihood of these accounts being targeted by social engineering attacks or exposed to malicious content via licensed applications. Administrative activities should be restricted to dedicated accounts without access to collaborative tools like mailboxes.",
"ImpactStatement": "Administrative users will need to switch accounts to perform privileged actions, requiring login/logout functionality and potentially losing the convenience of SSO. Alerts sent to Global Administrators or TenantAdmins by default might not be received if these accounts lack application-based licenses. Proper alert routing must be configured to avoid missed notifications.",
"RemediationProcedure": "Review the licenses assigned to administrative accounts. Remove licenses granting access to collaborative applications and assign Microsoft Entra ID P1 or P2 licenses if participation in Microsoft 365 security services is required. Configure alerts to be sent to valid email addresses for monitoring.",
"AuditProcedure": "Log in to the Microsoft 365 Admin Center and review the licenses assigned to administrative accounts. Confirm that administrative accounts either have no licenses or are limited to Microsoft Entra ID P1 or P2 licenses without collaborative applications enabled.",
"AdditionalInformation": "Reducing the application footprint of administrative accounts improves security by minimizing potential attack vectors. Special care should be taken to configure alert routing properly to ensure critical notifications are not missed.",
"DefaultValue": "By default, administrative accounts may have licenses assigned based on organizational setup. Manual review and adjustment are necessary to comply with this recommendation.",
"References": "CIS Microsoft 365 Foundations Benchmark v4.0, Section 1.1.4; Microsoft documentation on Entra ID licenses and privileged account security."
}
]
},
{
"Id": "1.2.1",
"Description": "Ensure that only organizationally managed/approved public groups exist",
"Checks": [
"admincenter_groups_not_public_visibility"
],
"Attributes": [
{
"Section": "1.Microsoft 365 admin center",
"Profile": "Level 2",
"AssessmentStatus": "Automated",
"Description": "Microsoft 365 Groups enable shared resource access across Microsoft 365 services. The default privacy setting for groups is 'Public,' which allows users within the organization to access the group's resources. Ensure that only organizationally managed and approved public groups exist to prevent unauthorized access to sensitive information.",
"RationaleStatement": "Public groups can be accessed by any user within the organization via several methods, such as self-adding through the Azure portal, sending an access request, or directly using the SharePoint URL. Without control over group privacy, sensitive organizational data might be exposed to unintended users.",
"ImpactStatement": "Implementing this recommendation may result in an increased volume of access requests for group owners, particularly for groups previously intended to be public.",
"RemediationProcedure": "Audit all Microsoft 365 public groups and ensure they are organizationally managed and approved. Convert unnecessary public groups to private groups and enforce strict policies for creating and approving public groups. Group owners should be instructed to monitor and review access requests.",
"AuditProcedure": "Log in to the Microsoft 365 Admin Center and review the list of public groups. Verify that all public groups have been approved and are necessary for organizational purposes.",
"AdditionalInformation": "Public groups expose data to all users within the organization, increasing the risk of accidental or unauthorized access. Periodic reviews of group privacy settings are recommended.",
"DefaultValue": "By default, groups created in Microsoft 365 are set to 'Public' privacy.",
"References": "CIS Microsoft 365 Foundations Benchmark v4.0, Section 1.2.1; Microsoft documentation on managing group privacy."
}
]
},
{
"Id": "1.2.2",
"Description": "Ensure sign-in to shared mailboxes is blocked",
"Checks": [],
"Attributes": [
{
"Section": "1.Microsoft 365 admin center",
"Profile": "Level 1",
"AssessmentStatus": "Manuel",
"Description": "Shared mailboxes are used when multiple people need access to the same mailbox for functions such as support or reception. These mailboxes are created with a corresponding user account that includes a system-generated password. To enhance security, sign-in should be blocked for these shared mailbox accounts, ensuring access is granted only through delegation.",
"RationaleStatement": "Blocking sign-in for shared mailbox accounts prevents unauthorized access or direct sign-in, ensuring that all interactions with the shared mailbox are through authorized delegation. This reduces the risk of attackers exploiting shared mailboxes for malicious purposes such as sending emails with spoofed identities.",
"ImpactStatement": "Blocking sign-in to shared mailboxes requires users to access these mailboxes only through delegation. Administrators will need to monitor and ensure proper access permissions are assigned.",
"RemediationProcedure": "Log in to the Microsoft 365 Admin Center and locate the shared mailboxes. For each shared mailbox, verify that sign-in is blocked by reviewing the associated user account settings. If sign-in is not blocked, adjust the account settings to enforce this configuration.",
"AuditProcedure": "Review all shared mailboxes in the Microsoft 365 Admin Center. Confirm that the user accounts associated with these mailboxes have sign-in blocked.",
"AdditionalInformation": "Shared mailboxes are often a target for exploitation due to their broad access and functional role. Blocking sign-in significantly reduces the attack surface.",
"DefaultValue": "By default, shared mailboxes may have sign-in enabled unless explicitly configured otherwise.",
"References": "CIS Microsoft 365 Foundations Benchmark v4.0, Section 1.2.2; Microsoft documentation on managing shared mailboxes."
}
]
}
]
}
+1
View File
@@ -28,6 +28,7 @@ class Provider(str, Enum):
GCP = "gcp"
AZURE = "azure"
KUBERNETES = "kubernetes"
MICROSOFT365 = "microsoft365"
# Compliance
@@ -0,0 +1,44 @@
### Account, Check and/or Region can be * to apply for all the cases.
### Account == Microsoft365 Tenant and Region == Microsoft365 Location
### Resources and tags are lists that can have either Regex or Keywords.
### Tags is an optional list that matches on tuples of 'key=value' and are "ANDed" together.
### Use an alternation Regex to match one of multiple tags with "ORed" logic.
### For each check you can except Accounts, Regions, Resources and/or Tags.
########################### MUTELIST EXAMPLE ###########################
Mutelist:
Accounts:
"*":
Checks:
"admincenter_groups_not_public_visibility":
Regions:
- "westeurope"
Resources:
- "sqlserver1" # Will ignore sqlserver1 in check sqlserver_tde_encryption_enabled located in westeurope
Description: "Findings related with the check sqlserver_tde_encryption_enabled is muted for westeurope region and sqlserver1 resource"
"defender_*":
Regions:
- "*"
Resources:
- "*" # Will ignore every Defender check in every location
"*":
Regions:
- "*"
Resources:
- "test"
Tags:
- "test=test" # Will ignore every resource containing the string "test" and the tags 'test=test' and
- "project=test|project=stage" # either of ('project=test' OR project=stage) in Azure subscription 1 and every location
"*":
Checks:
"admincenter_*":
Regions:
- "*"
Resources:
- "*"
Exceptions:
Accounts:
- "Tenant1"
Regions:
- "eastus"
- "eastus2" # Will ignore every resource in admincenter checks except the ones in Tenant1 located in eastus or eastus2
+23
View File
@@ -535,6 +535,29 @@ class Check_Report_Kubernetes(Check_Report):
self.namespace = "cluster-wide"
@dataclass
class Check_Report_Microsoft365(Check_Report):
"""Contains the Microsoft365 Check's finding information."""
resource_name: str
resource_id: str
location: str
def __init__(self, metadata: Dict, resource: Any) -> None:
"""Initialize the Microsoft365 Check's finding information.
Args:
metadata: The metadata of the check.
resource: Basic information about the resource. Defaults to None.
"""
super().__init__(metadata, resource)
self.resource_name = getattr(
resource, "name", getattr(resource, "resource_name", "")
)
self.resource_id = getattr(resource, "id", getattr(resource, "resource_id", ""))
self.location = getattr(resource, "location", "global")
# Testing Pending
def load_check_metadata(metadata_file: str) -> CheckMetadata:
"""
+3 -2
View File
@@ -26,7 +26,7 @@ class ProwlerArgumentParser:
self.parser = argparse.ArgumentParser(
prog="prowler",
formatter_class=RawTextHelpFormatter,
usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,dashboard} ...",
usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,microsoft365,dashboard} ...",
epilog="""
Available Cloud Providers:
{aws,azure,gcp,kubernetes}
@@ -34,6 +34,7 @@ Available Cloud Providers:
azure Azure Provider
gcp GCP Provider
kubernetes Kubernetes Provider
microsoft365 Microsoft 365 Provider
Available components:
dashboard Local dashboard
@@ -72,7 +73,7 @@ Detailed documentation at https://docs.prowler.com
# Init Providers Arguments
init_providers_parser(self)
# Dahboard Parser
# Dashboard Parser
init_dashboard_parser(self)
def parse(self, args=None) -> argparse.Namespace:
@@ -0,0 +1,99 @@
from prowler.lib.check.compliance_models import Compliance
from prowler.lib.outputs.compliance.cis.models import Microsoft365CISModel
from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput
from prowler.lib.outputs.finding import Finding
class Microsoft365CIS(ComplianceOutput):
"""
This class represents the Azure CIS compliance output.
Attributes:
- _data (list): A list to store transformed data from findings.
- _file_descriptor (TextIOWrapper): A file descriptor to write data to a file.
Methods:
- transform: Transforms findings into Azure CIS compliance format.
"""
def transform(
self,
findings: list[Finding],
compliance: Compliance,
compliance_name: str,
) -> None:
"""
Transforms a list of findings into Azure CIS compliance format.
Parameters:
- findings (list): A list of findings.
- compliance (Compliance): A compliance model.
- compliance_name (str): The name of the compliance model.
Returns:
- None
"""
for finding in findings:
# Get the compliance requirements for the finding
finding_requirements = finding.compliance.get(compliance_name, [])
for requirement in compliance.Requirements:
if requirement.Id in finding_requirements:
for attribute in requirement.Attributes:
compliance_row = Microsoft365CISModel(
Provider=finding.provider,
Description=compliance.Description,
SubscriptionId=finding.account_uid,
Location=finding.region,
AssessmentDate=str(finding.timestamp),
Requirements_Id=requirement.Id,
Requirements_Description=requirement.Description,
Requirements_Attributes_Section=attribute.Section,
Requirements_Attributes_Profile=attribute.Profile,
Requirements_Attributes_AssessmentStatus=attribute.AssessmentStatus,
Requirements_Attributes_Description=attribute.Description,
Requirements_Attributes_RationaleStatement=attribute.RationaleStatement,
Requirements_Attributes_ImpactStatement=attribute.ImpactStatement,
Requirements_Attributes_RemediationProcedure=attribute.RemediationProcedure,
Requirements_Attributes_AuditProcedure=attribute.AuditProcedure,
Requirements_Attributes_AdditionalInformation=attribute.AdditionalInformation,
Requirements_Attributes_DefaultValue=attribute.DefaultValue,
Requirements_Attributes_References=attribute.References,
Status=finding.status,
StatusExtended=finding.status_extended,
ResourceId=finding.resource_uid,
ResourceName=finding.resource_name,
CheckId=finding.check_id,
Muted=finding.muted,
)
self._data.append(compliance_row)
# Add manual requirements to the compliance output
for requirement in compliance.Requirements:
if not requirement.Checks:
for attribute in requirement.Attributes:
compliance_row = Microsoft365CISModel(
Provider=compliance.Provider.lower(),
Description=compliance.Description,
SubscriptionId="",
Location="",
AssessmentDate=str(finding.timestamp),
Requirements_Id=requirement.Id,
Requirements_Description=requirement.Description,
Requirements_Attributes_Section=attribute.Section,
Requirements_Attributes_Profile=attribute.Profile,
Requirements_Attributes_AssessmentStatus=attribute.AssessmentStatus,
Requirements_Attributes_Description=attribute.Description,
Requirements_Attributes_RationaleStatement=attribute.RationaleStatement,
Requirements_Attributes_ImpactStatement=attribute.ImpactStatement,
Requirements_Attributes_RemediationProcedure=attribute.RemediationProcedure,
Requirements_Attributes_AuditProcedure=attribute.AuditProcedure,
Requirements_Attributes_AdditionalInformation=attribute.AdditionalInformation,
Requirements_Attributes_DefaultValue=attribute.DefaultValue,
Requirements_Attributes_References=attribute.References,
Status="MANUAL",
StatusExtended="Manual check",
ResourceId="manual_check",
ResourceName="Manual check",
CheckId="manual",
Muted=False,
)
self._data.append(compliance_row)
@@ -66,6 +66,37 @@ class AzureCISModel(BaseModel):
Muted: bool
class Microsoft365CISModel(BaseModel):
"""
Microsoft365CISModel generates a finding's output in Microsoft365 CIS Compliance format.
"""
Provider: str
Description: str
SubscriptionId: str
Location: str
AssessmentDate: str
Requirements_Id: str
Requirements_Description: str
Requirements_Attributes_Section: str
Requirements_Attributes_Profile: str
Requirements_Attributes_AssessmentStatus: str
Requirements_Attributes_Description: str
Requirements_Attributes_RationaleStatement: str
Requirements_Attributes_ImpactStatement: str
Requirements_Attributes_RemediationProcedure: str
Requirements_Attributes_AuditProcedure: str
Requirements_Attributes_AdditionalInformation: str
Requirements_Attributes_DefaultValue: str
Requirements_Attributes_References: str
Status: str
StatusExtended: str
ResourceId: str
ResourceName: str
CheckId: str
Muted: bool
class GCPCISModel(BaseModel):
"""
GCPCISModel generates a finding's output in GCP CIS Compliance format.
+14
View File
@@ -234,6 +234,20 @@ class Finding(BaseModel):
)
output_data["region"] = f"namespace: {check_output.namespace}"
elif provider.type == "microsoft365":
output_data["auth_method"] = (
f"{provider.identity.identity_type}: {provider.identity.identity_id}"
)
output_data["account_uid"] = get_nested_attribute(
provider, "identity.tenant_id"
)
output_data["account_name"] = get_nested_attribute(
provider, "identity.tenant_domain"
)
output_data["resource_name"] = check_output.resource_name
output_data["resource_uid"] = check_output.resource_id
output_data["region"] = check_output.location
# check_output Unique ID
# TODO: move this to a function
# TODO: in Azure, GCP and K8s there are fidings without resource_name
+46
View File
@@ -538,6 +538,52 @@ class HTML(Output):
)
return ""
@staticmethod
def get_microsoft365_assessment_summary(provider: Provider) -> str:
"""
get_microsoft365_assessment_summary gets the HTML assessment summary for the provider
Args:
provider (Provider): the provider object
Returns:
str: the HTML assessment summary
"""
try:
return f"""
<div class="col-md-2">
<div class="card">
<div class="card-header">
Microsoft365 Assessment Summary
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Microsoft365 Tenant Domain:</b> {provider.identity.tenant_domain}
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
Microsoft365 Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Microsoft365 Identity Type:</b> {provider.identity.identity_type}
</li>
<li class="list-group-item">
<b>Microsoft365 Identity ID:</b> {provider.identity.identity_id}
</li>
</ul>
</div>
</div>"""
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
return ""
@staticmethod
def get_assessment_summary(provider: Provider) -> str:
"""
+2
View File
@@ -13,6 +13,8 @@ def stdout_report(finding, color, verbose, status, fix):
details = finding.location.lower()
if finding.check_metadata.Provider == "kubernetes":
details = finding.namespace.lower()
if finding.check_metadata.Provider == "microsoft365":
details = finding.location
if (verbose or fix) and (not status or finding.status in status):
if finding.muted:
+3
View File
@@ -40,6 +40,9 @@ def display_summary_table(
elif provider.type == "kubernetes":
entity_type = "Context"
audited_entities = provider.identity.context
elif provider.type == "microsoft365":
entity_type = "Tenant Domain"
audited_entities = provider.identity.tenant_domain
# Check if there are findings and that they are not all MANUAL
if findings and not all(finding.status == "MANUAL" for finding in findings):
+11
View File
@@ -211,6 +211,17 @@ class Provider(ABC):
mutelist_path=arguments.mutelist_file,
fixer_config=fixer_config,
)
elif "microsoft365" in provider_class_name.lower():
provider_class(
region=arguments.region,
config_path=arguments.config_file,
mutelist_path=arguments.mutelist_file,
sp_env_auth=arguments.sp_env_auth,
az_cli_auth=arguments.az_cli_auth,
browser_auth=arguments.browser_auth,
tenant_id=arguments.tenant_id,
fixer_config=fixer_config,
)
except TypeError as error:
logger.critical(
@@ -0,0 +1,279 @@
from prowler.exceptions.exceptions import ProwlerException
# Exceptions codes from 5000 to 5999 are reserved for Microsoft365 exceptions
class Microsoft365BaseException(ProwlerException):
"""Base class for Microsoft365 Errors."""
MICROSOFT365_ERROR_CODES = {
(6000, "Microsoft365EnvironmentVariableError"): {
"message": "Microsoft365 environment variable error",
"remediation": "Check the Microsoft365 environment variables and ensure they are properly set.",
},
(6001, "Microsoft365ArgumentTypeValidationError"): {
"message": "Microsoft365 argument type validation error",
"remediation": "Check the provided argument types specific to Microsoft365 and ensure they meet the required format.",
},
(6002, "Microsoft365SetUpRegionConfigError"): {
"message": "Microsoft365 region configuration setup error",
"remediation": "Check the Microsoft365 region configuration and ensure it is properly set up.",
},
(6003, "Microsoft365HTTPResponseError"): {
"message": "Error in HTTP response from Microsoft365",
"remediation": "",
},
(6004, "Microsoft365CredentialsUnavailableError"): {
"message": "Error trying to configure Microsoft365 credentials because they are unavailable",
"remediation": "Check the dictionary and ensure it is properly set up for Microsoft365 credentials. TENANT_ID, CLIENT_ID and CLIENT_SECRET are required.",
},
(6005, "Microsoft365GetTokenIdentityError"): {
"message": "Error trying to get token from Microsoft365 Identity",
"remediation": "Check the Microsoft365 Identity and ensure it is properly set up.",
},
(6006, "Microsoft365ClientAuthenticationError"): {
"message": "Error in client authentication",
"remediation": "Check the client authentication and ensure it is properly set up.",
},
(6007, "Microsoft365NotValidTenantIdError"): {
"message": "The provided tenant ID is not valid",
"remediation": "Check the tenant ID and ensure it is a valid ID.",
},
(6008, "Microsoft365NotValidClientIdError"): {
"message": "The provided client ID is not valid",
"remediation": "Check the client ID and ensure it is a valid ID.",
},
(6009, "Microsoft365NotValidClientSecretError"): {
"message": "The provided client secret is not valid",
"remediation": "Check the client secret and ensure it is a valid secret.",
},
(6010, "Microsoft365ConfigCredentialsError"): {
"message": "Error in configuration of Microsoft365 credentials",
"remediation": "Check the configuration of Microsoft365 credentials and ensure it is properly set up.",
},
(6011, "Microsoft365ClientIdAndClientSecretNotBelongingToTenantIdError"): {
"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.",
},
(6012, "Microsoft365TenantIdAndClientSecretNotBelongingToClientIdError"): {
"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.",
},
(6013, "Microsoft365TenantIdAndClientIdNotBelongingToClientSecretError"): {
"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.",
},
(6014, "Microsoft365InvalidProviderIdError"): {
"message": "The provided provider_id does not match with the available subscriptions",
"remediation": "Check the provider_id and ensure it is a valid subscription for the given credentials.",
},
(6015, "Microsoft365NoAuthenticationMethodError"): {
"message": "No Microsoft365 authentication method found",
"remediation": "Check that any authentication method is properly set up for Microsoft365.",
},
(6016, "Microsoft365SetUpSessionError"): {
"message": "Error setting up session",
"remediation": "Check the session setup and ensure it is properly set up.",
},
(6017, "Microsoft365DefaultAzureCredentialError"): {
"message": "Error with DefaultAzureCredential",
"remediation": "Ensure DefaultAzureCredential is correctly configured.",
},
(6018, "Microsoft365InteractiveBrowserCredentialError"): {
"message": "Error with InteractiveBrowserCredential",
"remediation": "Ensure InteractiveBrowserCredential is correctly configured.",
},
(6019, "Microsoft365BrowserAuthNoTenantIDError"): {
"message": "Microsoft365 Tenant ID (--tenant-id) is required for browser authentication mode",
"remediation": "Check the Microsoft365 Tenant ID and ensure it is properly set up.",
},
(6020, "Microsoft365BrowserAuthNoFlagError"): {
"message": "Microsoft365 tenant ID error: browser authentication flag (--browser-auth) not found",
"remediation": "To use browser authentication, ensure the tenant ID is properly set.",
},
(6021, "Microsoft365NotTenantIdButClientIdAndClienSecretError"): {
"message": "Tenant Id is required for Microsoft365 static credentials. Make sure you are using the correct credentials.",
"remediation": "Check the Microsoft365 Tenant ID and ensure it is properly set up.",
},
}
def __init__(self, code, file=None, original_exception=None, message=None):
provider = "Microsoft365"
error_info = self.MICROSOFT365_ERROR_CODES.get((code, self.__class__.__name__))
if message:
error_info["message"] = message
super().__init__(
code=code,
source=provider,
file=file,
original_exception=original_exception,
error_info=error_info,
)
class Microsoft365CredentialsError(Microsoft365BaseException):
"""Base class for Microsoft365 credentials errors."""
def __init__(self, code, file=None, original_exception=None, message=None):
super().__init__(code, file, original_exception, message)
class Microsoft365EnvironmentVariableError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6000, file=file, original_exception=original_exception, message=message
)
class Microsoft365ArgumentTypeValidationError(Microsoft365BaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6001, file=file, original_exception=original_exception, message=message
)
class Microsoft365SetUpRegionConfigError(Microsoft365BaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6002, file=file, original_exception=original_exception, message=message
)
class Microsoft365HTTPResponseError(Microsoft365BaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6003, file=file, original_exception=original_exception, message=message
)
class Microsoft365CredentialsUnavailableError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6004, file=file, original_exception=original_exception, message=message
)
class Microsoft365GetTokenIdentityError(Microsoft365BaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6005, file=file, original_exception=original_exception, message=message
)
class Microsoft365ClientAuthenticationError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6006, file=file, original_exception=original_exception, message=message
)
class Microsoft365NotValidTenantIdError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6007, file=file, original_exception=original_exception, message=message
)
class Microsoft365NotValidClientIdError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6008, file=file, original_exception=original_exception, message=message
)
class Microsoft365NotValidClientSecretError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6009, file=file, original_exception=original_exception, message=message
)
class Microsoft365ConfigCredentialsError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6010, file=file, original_exception=original_exception, message=message
)
class Microsoft365ClientIdAndClientSecretNotBelongingToTenantIdError(
Microsoft365CredentialsError
):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6011, file=file, original_exception=original_exception, message=message
)
class Microsoft365TenantIdAndClientSecretNotBelongingToClientIdError(
Microsoft365CredentialsError
):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6012, file=file, original_exception=original_exception, message=message
)
class Microsoft365TenantIdAndClientIdNotBelongingToClientSecretError(
Microsoft365CredentialsError
):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6013, file=file, original_exception=original_exception, message=message
)
class Microsoft365InvalidProviderIdError(Microsoft365BaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6014, file=file, original_exception=original_exception, message=message
)
class Microsoft365NoAuthenticationMethodError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6015, file=file, original_exception=original_exception, message=message
)
class Microsoft365SetUpSessionError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6016, file=file, original_exception=original_exception, message=message
)
class Microsoft365DefaultAzureCredentialError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6017, file=file, original_exception=original_exception, message=message
)
class Microsoft365InteractiveBrowserCredentialError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6018, file=file, original_exception=original_exception, message=message
)
class Microsoft365BrowserAuthNoTenantIDError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6019, file=file, original_exception=original_exception, message=message
)
class Microsoft365BrowserAuthNoFlagError(Microsoft365CredentialsError):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6020, file=file, original_exception=original_exception, message=message
)
class Microsoft365NotTenantIdButClientIdAndClienSecretError(
Microsoft365CredentialsError
):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6021, file=file, original_exception=original_exception, message=message
)
@@ -0,0 +1,48 @@
def init_parser(self):
"""Init the Microsoft365 Provider CLI parser"""
microsoft365_parser = self.subparsers.add_parser(
"microsoft365",
parents=[self.common_providers_parser],
help="Microsoft365 Provider",
)
# Authentication Modes
microsoft365_auth_subparser = microsoft365_parser.add_argument_group(
"Authentication Modes"
)
microsoft365_auth_modes_group = (
microsoft365_auth_subparser.add_mutually_exclusive_group()
)
microsoft365_auth_modes_group.add_argument(
"--az-cli-auth",
action="store_true",
help="Use Azure CLI authentication to log in against Microsoft365",
)
microsoft365_auth_modes_group.add_argument(
"--sp-env-auth",
action="store_true",
help="Use Azure Service Principal environment variables authentication to log in against Microsoft365",
)
microsoft365_auth_modes_group.add_argument(
"--browser-auth",
action="store_true",
help="Use Azure interactive browser authentication to log in against Microsoft365",
)
microsoft365_parser.add_argument(
"--tenant-id",
nargs="?",
default=None,
help="Microsoft365 Tenant ID to be used with --browser-auth option",
)
# Regions
microsoft365_regions_subparser = microsoft365_parser.add_argument_group("Regions")
microsoft365_regions_subparser.add_argument(
"--region",
nargs="?",
default="Microsoft365Global",
choices=[
"Microsoft365Global",
"Microsoft365GlobalChina",
"Microsoft365USGovernment",
],
help="Microsoft365 region to be used, default is Microsoft365Global",
)
@@ -0,0 +1,17 @@
from prowler.lib.check.models import Check_Report_Microsoft365
from prowler.lib.mutelist.mutelist import Mutelist
from prowler.lib.outputs.utils import unroll_dict, unroll_tags
class Microsoft365Mutelist(Mutelist):
def is_finding_muted(
self,
finding: Check_Report_Microsoft365,
) -> bool:
return self.is_muted(
finding.tenant_id,
finding.check_metadata.CheckID,
finding.location,
finding.resource_name,
unroll_dict(unroll_tags(finding.resource_tags)),
)
@@ -0,0 +1,27 @@
from azure.identity import AzureAuthorityHosts
MICROSOFT365_CHINA_CLOUD = "https://microsoftgraph.chinacloudapi.cn"
MICROSOFT365_US_GOV_CLOUD = "https://graph.microsoft.us"
MICROSOFT365_US_DOD_CLOUD = "https://graph.microsoftmil.us"
MICROSOFT365_GENERIC_CLOUD = "https://graph.microsoft.com"
def get_regions_config(region):
allowed_regions = {
"Microsoft365Global": {
"authority": None,
"base_url": MICROSOFT365_GENERIC_CLOUD,
"credential_scopes": [MICROSOFT365_GENERIC_CLOUD + "/.default"],
},
"Microsoft365China": {
"authority": AzureAuthorityHosts.AZURE_CHINA,
"base_url": MICROSOFT365_CHINA_CLOUD,
"credential_scopes": [MICROSOFT365_CHINA_CLOUD + "/.default"],
},
"Microsoft365USGovernment": {
"authority": AzureAuthorityHosts.AZURE_GOVERNMENT,
"base_url": MICROSOFT365_US_GOV_CLOUD,
"credential_scopes": [MICROSOFT365_US_GOV_CLOUD + "/.default"],
},
}
return allowed_regions[region]
@@ -0,0 +1,14 @@
from msgraph import GraphServiceClient
from prowler.providers.microsoft365.microsoft365_provider import Microsoft365Provider
class Microsoft365Service:
def __init__(
self,
provider: Microsoft365Provider,
):
self.client = GraphServiceClient(credentials=provider.session)
self.audit_config = provider.audit_config
self.fixer_config = provider.fixer_config
@@ -0,0 +1,925 @@
import asyncio
import os
import re
from argparse import ArgumentTypeError
from os import getenv
from uuid import UUID
from azure.core.exceptions import ClientAuthenticationError, HttpResponseError
from azure.identity import (
ClientSecretCredential,
CredentialUnavailableError,
DefaultAzureCredential,
InteractiveBrowserCredential,
)
from colorama import Fore, Style
from msal import ConfidentialClientApplication
from msgraph import GraphServiceClient
from prowler.config.config import (
default_config_file_path,
get_default_mute_file_path,
load_and_validate_config_file,
)
from prowler.lib.logger import logger
from prowler.lib.utils.utils import print_boxes
from prowler.providers.common.models import Audit_Metadata, Connection
from prowler.providers.common.provider import Provider
from prowler.providers.microsoft365.exceptions.exceptions import (
Microsoft365ArgumentTypeValidationError,
Microsoft365BrowserAuthNoFlagError,
Microsoft365BrowserAuthNoTenantIDError,
Microsoft365ClientAuthenticationError,
Microsoft365ClientIdAndClientSecretNotBelongingToTenantIdError,
Microsoft365ConfigCredentialsError,
Microsoft365CredentialsUnavailableError,
Microsoft365DefaultAzureCredentialError,
Microsoft365EnvironmentVariableError,
Microsoft365GetTokenIdentityError,
Microsoft365HTTPResponseError,
Microsoft365InteractiveBrowserCredentialError,
Microsoft365InvalidProviderIdError,
Microsoft365NoAuthenticationMethodError,
Microsoft365NotTenantIdButClientIdAndClienSecretError,
Microsoft365NotValidClientIdError,
Microsoft365NotValidClientSecretError,
Microsoft365NotValidTenantIdError,
Microsoft365SetUpRegionConfigError,
Microsoft365SetUpSessionError,
Microsoft365TenantIdAndClientIdNotBelongingToClientSecretError,
Microsoft365TenantIdAndClientSecretNotBelongingToClientIdError,
)
from prowler.providers.microsoft365.lib.mutelist.mutelist import Microsoft365Mutelist
from prowler.providers.microsoft365.lib.regions.microsoft365_regions import (
get_regions_config,
)
from prowler.providers.microsoft365.models import (
Microsoft365IdentityInfo,
Microsoft365RegionConfig,
)
class Microsoft365Provider(Provider):
"""
Represents an Microsoft365 provider.
This class provides functionality to interact with the Microsoft365 resources.
It handles authentication, region configuration, and provides access to various properties and methods
related to the Microsoft365 provider.
Attributes:
_type (str): The type of the provider, which is set to "microsoft365".
_session (DefaultMicrosoft365Credential): The session object associated with the Microsoft365 provider.
_identity (Microsoft365IdentityInfo): The identity information for the Microsoft365 provider.
_audit_config (dict): The audit configuration for the Microsoft365 provider.
_region_config (Microsoft365RegionConfig): The region configuration for the Microsoft365 provider.
_mutelist (Microsoft365Mutelist): The mutelist object associated with the Microsoft365 provider.
audit_metadata (Audit_Metadata): The audit metadata for the Microsoft365 provider.
Methods:
__init__ -> Initializes the Microsoft365 provider.
identity(self): Returns the identity of the Microsoft365 provider.
type(self): Returns the type of the Microsoft365 provider.
session(self): Returns the session object associated with the Microsoft365 provider.
region_config(self): Returns the region configuration for the Microsoft365 provider.
audit_config(self): Returns the audit configuration for the Microsoft365 provider.
fixer_config(self): Returns the fixer configuration.
output_options(self, options: tuple): Sets the output options for the Microsoft365 provider.
mutelist(self) -> Microsoft365Mutelist: Returns the mutelist object associated with the Microsoft365 provider.
setup_region_config(cls, region): Sets up the region configuration for the Microsoft365 provider.
print_credentials(self): Prints the Microsoft365 credentials information.
setup_session(cls, az_cli_auth, app_env_auth, browser_auth, managed_identity_auth, tenant_id, region_config): Set up the Microsoft365 session with the specified authentication method.
"""
_type: str = "microsoft365"
_session: DefaultAzureCredential # Must be used besides being named for Azure
_identity: Microsoft365IdentityInfo
_audit_config: dict
_region_config: Microsoft365RegionConfig
_mutelist: Microsoft365Mutelist
# TODO: this is not optional, enforce for all providers
audit_metadata: Audit_Metadata
def __init__(
self,
sp_env_auth: bool,
az_cli_auth: bool,
browser_auth: bool,
tenant_id: str = None,
client_id: str = None,
client_secret: str = None,
region: str = "Microsoft365Global",
config_content: dict = None,
config_path: str = None,
mutelist_path: str = None,
mutelist_content: dict = None,
fixer_config: dict = {},
):
"""
Initializes the Microsoft365 provider.
Args:
tenant_id (str): The Microsoft365 Active Directory tenant ID.
region (str): The Microsoft365 region.
client_id (str): The Microsoft365 client ID.
client_secret (str): The Microsoft365 client secret.
config_path (str): The path to the configuration file.
config_content (dict): The configuration content.
fixer_config (dict): The fixer configuration.
mutelist_path (str): The path to the mutelist file.
mutelist_content (dict): The mutelist content.
Returns:
None
Raises:
Microsoft365ArgumentTypeValidationError: If there is an error in the argument type validation.
Microsoft365SetUpRegionConfigError: If there is an error in setting up the region configuration.
Microsoft365ConfigCredentialsError: If there is an error in configuring the Microsoft365 credentials from a dictionary.
Microsoft365GetTokenIdentityError: If there is an error in getting the token from the Microsoft365 identity.
Microsoft365HTTPResponseError: If there is an HTTP response error.
"""
logger.info("Setting Microsoft365 provider ...")
logger.info("Checking if any credentials mode is set ...")
# Validate the authentication arguments
self.validate_arguments(
az_cli_auth,
sp_env_auth,
browser_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
microsoft365_credentials = None
if tenant_id and client_id and client_secret:
microsoft365_credentials = self.validate_static_credentials(
tenant_id=tenant_id, client_id=client_id, client_secret=client_secret
)
# Set up the Microsoft365 session
self._session = self.setup_session(
az_cli_auth,
sp_env_auth,
browser_auth,
tenant_id,
microsoft365_credentials,
self._region_config,
)
# Set up the identity
self._identity = self.setup_identity(
az_cli_auth,
sp_env_auth,
browser_auth,
client_id,
)
# Audit Config
if config_content:
self._audit_config = config_content
else:
if not config_path:
config_path = default_config_file_path
self._audit_config = load_and_validate_config_file(self._type, config_path)
# Fixer Config
self._fixer_config = fixer_config
# Mutelist
if mutelist_content:
self._mutelist = Microsoft365Mutelist(
mutelist_content=mutelist_content,
)
else:
if not mutelist_path:
mutelist_path = get_default_mute_file_path(self.type)
self._mutelist = Microsoft365Mutelist(
mutelist_path=mutelist_path,
)
Provider.set_global_provider(self)
@property
def identity(self):
"""Returns the identity of the Microsoft365 provider."""
return self._identity
@property
def type(self):
"""Returns the type of the Microsoft365 provider."""
return self._type
@property
def session(self):
"""Returns the session object associated with the Microsoft365 provider."""
return self._session
@property
def region_config(self):
"""Returns the region configuration for the Microsoft365 provider."""
return self._region_config
@property
def audit_config(self):
"""Returns the audit configuration for the Microsoft365 provider."""
return self._audit_config
@property
def fixer_config(self):
"""Returns the fixer configuration."""
return self._fixer_config
@property
def mutelist(self) -> Microsoft365Mutelist:
"""Mutelist object associated with this Microsoft365 provider."""
return self._mutelist
@staticmethod
def validate_arguments(
az_cli_auth: bool,
sp_env_auth: bool,
browser_auth: bool,
tenant_id: str,
client_id: str,
client_secret: str,
):
"""
Validates the authentication arguments for the Microsoft365 provider.
Args:
az_cli_auth (bool): Flag indicating whether Azure CLI authentication is enabled.
sp_env_auth (bool): Flag indicating whether application authentication with environment variables is enabled.
browser_auth (bool): Flag indicating whether browser authentication is enabled.
tenant_id (str): The Microsoft365 Tenant ID.
client_id (str): The Microsoft365 Client ID.
client_secret (str): The Microsoft365 Client Secret.
Raises:
Microsoft365BrowserAuthNoTenantIDError: If browser authentication is enabled but the tenant ID is not found.
"""
if not client_id and not client_secret:
if not browser_auth and tenant_id:
raise Microsoft365BrowserAuthNoFlagError(
file=os.path.basename(__file__),
message="Microsoft365 tenant ID error: browser authentication flag (--browser-auth) not found",
)
elif not az_cli_auth and not sp_env_auth and not browser_auth:
raise Microsoft365NoAuthenticationMethodError(
file=os.path.basename(__file__),
message="Microsoft365 provider requires at least one authentication method set: [--az-cli-auth | --sp-env-auth | --browser-auth]",
)
elif browser_auth and not tenant_id:
raise Microsoft365BrowserAuthNoTenantIDError(
file=os.path.basename(__file__),
message="Microsoft365 Tenant ID (--tenant-id) is required for browser authentication mode",
)
else:
if not tenant_id:
raise Microsoft365NotTenantIdButClientIdAndClienSecretError(
file=os.path.basename(__file__),
message="Tenant Id is required for Microsoft365 static credentials. Make sure you are using the correct credentials.",
)
@staticmethod
def setup_region_config(region):
"""
Sets up the region configuration for the Microsoft365 provider.
Args:
region (str): The name of the region.
Returns:
Microsoft365RegionConfig: The region configuration object.
"""
try:
config = get_regions_config(region)
return Microsoft365RegionConfig(
name=region,
authority=config["authority"],
base_url=config["base_url"],
credential_scopes=config["credential_scopes"],
)
except ArgumentTypeError as validation_error:
logger.error(
f"{validation_error.__class__.__name__}[{validation_error.__traceback__.tb_lineno}]: {validation_error}"
)
raise Microsoft365ArgumentTypeValidationError(
file=os.path.basename(__file__),
original_exception=validation_error,
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
raise Microsoft365SetUpRegionConfigError(
file=os.path.basename(__file__),
original_exception=error,
)
def print_credentials(self):
"""Microsoft365 credentials information.
This method prints the Microsoft365 Tenant Domain, Microsoft365 Tenant ID, Microsoft365 Region,
Microsoft365 Subscriptions, Microsoft365 Identity Type, and Microsoft365 Identity ID.
Args:
None
Returns:
None
"""
report_lines = [
f"Microsoft365 Region: {Fore.YELLOW}{self.region_config.name}{Style.RESET_ALL}",
f"Microsoft365 Tenant Domain: {Fore.YELLOW}{self._identity.tenant_domain}{Style.RESET_ALL} Microsoft365 Tenant ID: {Fore.YELLOW}{self._identity.tenant_id}{Style.RESET_ALL}",
f"Microsoft365 Identity Type: {Fore.YELLOW}{self._identity.identity_type}{Style.RESET_ALL} Microsoft365 Identity ID: {Fore.YELLOW}{self._identity.identity_id}{Style.RESET_ALL}",
]
report_title = (
f"{Style.BRIGHT}Using the Microsoft365 credentials below:{Style.RESET_ALL}"
)
print_boxes(report_lines, report_title)
# TODO: setup_session or setup_credentials?
# This should be setup_credentials, since it is setting up the credentials for the provider
@staticmethod
def setup_session(
az_cli_auth: bool,
sp_env_auth: bool,
browser_auth: bool,
tenant_id: str,
microsoft365_credentials: dict,
region_config: Microsoft365RegionConfig,
):
"""Returns the Microsoft365 credentials object.
Set up the Microsoft365 session with the specified authentication method.
Args:
az_cli_auth (bool): Flag indicating whether to use Azure CLI authentication.
sp_env_auth (bool): Flag indicating whether to use application authentication with environment variables.
browser_auth (bool): Flag indicating whether to use interactive browser authentication.
tenant_id (str): The Microsoft365 Active Directory tenant ID.
microsoft365_credentials (dict): The Microsoft365 configuration object. It contains the following keys:
- tenant_id: The Microsoft365 Active Directory tenant ID.
- client_id: The Microsoft365 client ID.
- client_secret: The Microsoft365 client secret
region_config (Microsoft365RegionConfig): The region configuration object.
Returns:
credentials: The Microsoft365 credentials object.
Raises:
Exception: If failed to retrieve Microsoft365 credentials.
"""
if not browser_auth:
if sp_env_auth:
try:
Microsoft365Provider.check_service_principal_creds_env_vars()
except (
Microsoft365EnvironmentVariableError
) as environment_credentials_error:
logger.critical(
f"{environment_credentials_error.__class__.__name__}[{environment_credentials_error.__traceback__.tb_lineno}] -- {environment_credentials_error}"
)
raise environment_credentials_error
try:
if microsoft365_credentials:
try:
credentials = ClientSecretCredential(
tenant_id=microsoft365_credentials["tenant_id"],
client_id=microsoft365_credentials["client_id"],
client_secret=microsoft365_credentials["client_secret"],
)
return credentials
except ClientAuthenticationError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise Microsoft365ClientAuthenticationError(
file=os.path.basename(__file__), original_exception=error
)
except CredentialUnavailableError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise Microsoft365CredentialsUnavailableError(
file=os.path.basename(__file__), original_exception=error
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise Microsoft365ConfigCredentialsError(
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,
# Microsoft365 Auth using Managed Identity is not supported
exclude_managed_identity_credential=True,
# Microsoft365 Auth using Visual Studio is not supported
exclude_visual_studio_code_credential=True,
# Microsoft365 Auth using Shared Token Cache is not supported
exclude_shared_token_cache_credential=True,
# Microsoft365 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 Microsoft365ClientAuthenticationError(
file=os.path.basename(__file__), original_exception=error
)
except CredentialUnavailableError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise Microsoft365CredentialsUnavailableError(
file=os.path.basename(__file__), original_exception=error
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise Microsoft365DefaultAzureCredentialError(
file=os.path.basename(__file__), original_exception=error
)
except Exception as error:
logger.critical("Failed to retrieve Microsoft365 credentials")
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise Microsoft365SetUpSessionError(
file=os.path.basename(__file__), original_exception=error
)
else:
try:
credentials = InteractiveBrowserCredential(tenant_id=tenant_id)
except Exception as error:
logger.critical(
"Failed to retrieve Microsoft365 credentials using browser authentication"
)
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise Microsoft365InteractiveBrowserCredentialError(
file=os.path.basename(__file__), original_exception=error
)
return credentials
@staticmethod
def test_connection(
az_cli_auth: bool = False,
sp_env_auth: bool = False,
browser_auth: bool = False,
tenant_id: str = None,
region: str = "Microsoft365Global",
raise_on_exception=True,
client_id=None,
client_secret=None,
) -> Connection:
"""Test connection to Microsoft365 subscription.
Test the connection to an Microsoft365 subscription using the provided credentials.
Args:
az_cli_auth (bool): Flag indicating whether to use Azure CLI authentication.
sp_env_auth (bool): Flag indicating whether to use application authentication with environment variables.
browser_auth (bool): Flag indicating whether to use interactive browser authentication.
tenant_id (str): The Microsoft365 Active Directory tenant ID.
region (str): The Microsoft365 region.
raise_on_exception (bool): Flag indicating whether to raise an exception if the connection fails.
client_id (str): The Microsoft365 client ID.
client_secret (str): The Microsoft365 client secret.
Returns:
bool: True if the connection is successful, False otherwise.
Raises:
Exception: If failed to test the connection to Microsoft365 subscription.
Microsoft365ArgumentTypeValidationError: If there is an error in the argument type validation.
Microsoft365SetUpRegionConfigError: If there is an error in setting up the region configuration.
Microsoft365InteractiveBrowserCredentialError: If there is an error in retrieving the Microsoft365 credentials using browser authentication.
Microsoft365HTTPResponseError: If there is an HTTP response error.
Microsoft365ConfigCredentialsError: If there is an error in configuring the Microsoft365 credentials from a dictionary.
Examples:
>>> Microsoft365Provider.test_connection(az_cli_auth=True)
True
>>> Microsoft365Provider.test_connection(sp_env_auth=False, browser_auth=True, tenant_id=None)
False, ArgumentTypeError: Microsoft365 Tenant ID is required only for browser authentication mode
>>> Microsoft365Provider.test_connection(tenant_id="XXXXXXXXXX", client_id="XXXXXXXXXX", client_secret="XXXXXXXXXX")
True
"""
try:
Microsoft365Provider.validate_arguments(
az_cli_auth,
sp_env_auth,
browser_auth,
tenant_id,
client_id,
client_secret,
)
region_config = Microsoft365Provider.setup_region_config(region)
# Get the dict from the static credentials
microsoft365_credentials = None
if tenant_id and client_id and client_secret:
microsoft365_credentials = (
Microsoft365Provider.validate_static_credentials(
tenant_id=tenant_id,
client_id=client_id,
client_secret=client_secret,
)
)
# Set up the Microsoft365 session
credentials = Microsoft365Provider.setup_session(
az_cli_auth,
sp_env_auth,
browser_auth,
tenant_id,
microsoft365_credentials,
region_config,
)
GraphServiceClient(credentials=credentials)
logger.info("Microsoft365 provider: Connection to Microsoft365 successful")
return Connection(is_connected=True)
# Exceptions from setup_region_config
except Microsoft365ArgumentTypeValidationError as type_validation_error:
logger.error(
f"{type_validation_error.__class__.__name__}[{type_validation_error.__traceback__.tb_lineno}]: {type_validation_error}"
)
if raise_on_exception:
raise type_validation_error
return Connection(error=type_validation_error)
except Microsoft365SetUpRegionConfigError as region_config_error:
logger.error(
f"{region_config_error.__class__.__name__}[{region_config_error.__traceback__.tb_lineno}]: {region_config_error}"
)
if raise_on_exception:
raise region_config_error
return Connection(error=region_config_error)
# Exceptions from setup_session
except Microsoft365EnvironmentVariableError as environment_credentials_error:
logger.error(
f"{environment_credentials_error.__class__.__name__}[{environment_credentials_error.__traceback__.tb_lineno}]: {environment_credentials_error}"
)
if raise_on_exception:
raise environment_credentials_error
return Connection(error=environment_credentials_error)
except Microsoft365ConfigCredentialsError as config_credentials_error:
logger.error(
f"{config_credentials_error.__class__.__name__}[{config_credentials_error.__traceback__.tb_lineno}]: {config_credentials_error}"
)
if raise_on_exception:
raise config_credentials_error
return Connection(error=config_credentials_error)
except Microsoft365ClientAuthenticationError as client_auth_error:
logger.error(
f"{client_auth_error.__class__.__name__}[{client_auth_error.__traceback__.tb_lineno}]: {client_auth_error}"
)
if raise_on_exception:
raise client_auth_error
return Connection(error=client_auth_error)
except Microsoft365CredentialsUnavailableError as credential_unavailable_error:
logger.error(
f"{credential_unavailable_error.__class__.__name__}[{credential_unavailable_error.__traceback__.tb_lineno}]: {credential_unavailable_error}"
)
if raise_on_exception:
raise credential_unavailable_error
return Connection(error=credential_unavailable_error)
except (
Microsoft365ClientIdAndClientSecretNotBelongingToTenantIdError
) as tenant_id_error:
logger.error(
f"{tenant_id_error.__class__.__name__}[{tenant_id_error.__traceback__.tb_lineno}]: {tenant_id_error}"
)
if raise_on_exception:
raise tenant_id_error
return Connection(error=tenant_id_error)
except (
Microsoft365TenantIdAndClientSecretNotBelongingToClientIdError
) as client_id_error:
logger.error(
f"{client_id_error.__class__.__name__}[{client_id_error.__traceback__.tb_lineno}]: {client_id_error}"
)
if raise_on_exception:
raise client_id_error
return Connection(error=client_id_error)
except (
Microsoft365TenantIdAndClientIdNotBelongingToClientSecretError
) as client_secret_error:
logger.error(
f"{client_secret_error.__class__.__name__}[{client_secret_error.__traceback__.tb_lineno}]: {client_secret_error}"
)
if raise_on_exception:
raise client_secret_error
return Connection(error=client_secret_error)
# Exceptions from provider_id validation
except Microsoft365InvalidProviderIdError as invalid_credentials_error:
logger.error(
f"{invalid_credentials_error.__class__.__name__}[{invalid_credentials_error.__traceback__.tb_lineno}]: {invalid_credentials_error}"
)
if raise_on_exception:
raise invalid_credentials_error
return Connection(error=invalid_credentials_error)
# Exceptions from SubscriptionClient
except HttpResponseError as http_response_error:
logger.error(
f"{http_response_error.__class__.__name__}[{http_response_error.__traceback__.tb_lineno}]: {http_response_error}"
)
if raise_on_exception:
raise Microsoft365HTTPResponseError(
file=os.path.basename(__file__),
original_exception=http_response_error,
)
return Connection(error=http_response_error)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if raise_on_exception:
# Raise directly the exception
raise error
return Connection(error=error)
@staticmethod
def check_service_principal_creds_env_vars():
"""
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
- AZURE_CLIENT_SECRET: Azure client secret
If any of the environment variables is missing, it logs a critical error and exits the program.
"""
logger.info(
"Microsoft365 provider: checking service principal environment variables ..."
)
for env_var in ["AZURE_CLIENT_ID", "AZURE_TENANT_ID", "AZURE_CLIENT_SECRET"]:
if not getenv(env_var):
logger.critical(
f"Microsoft365 provider: Missing environment variable {env_var} needed to authenticate against Microsoft365"
)
raise Microsoft365EnvironmentVariableError(
file=os.path.basename(__file__),
message=f"Missing environment variable {env_var} required to authenticate.",
)
def setup_identity(
self,
az_cli_auth,
sp_env_auth,
browser_auth,
client_id,
):
"""
Sets up the identity for the Microsoft365 provider.
Args:
az_cli_auth (bool): Flag indicating if Azure CLI authentication is used.
sp_env_auth (bool): Flag indicating if application authentication with environment variables is used.
browser_auth (bool): Flag indicating if interactive browser authentication is used.
client_id (str): The Microsoft365 client ID.
Returns:
Microsoft365IdentityInfo: An instance of Microsoft365IdentityInfo containing the identity information.
"""
credentials = self.session
# TODO: fill this object with real values not default and set to none
identity = Microsoft365IdentityInfo()
# If credentials comes from service principal or browser, if the required permissions are assigned
# the identity can access AAD and retrieve the tenant domain name.
# With cli also should be possible but right now it does not work, microsoft365 python package issue is coming
# At the time of writting this with az cli creds is not working, despite that is included
if az_cli_auth or sp_env_auth or browser_auth or client_id:
async def get_microsoft365_identity():
# Trying to recover tenant domain info
try:
logger.info(
"Trying to retrieve tenant domain from AAD to populate identity structure ..."
)
client = GraphServiceClient(credentials=credentials)
domain_result = await client.domains.get()
if getattr(domain_result, "value"):
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 Microsoft365HTTPResponseError(
file=os.path.basename(__file__),
original_exception=error,
)
except ClientAuthenticationError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
raise Microsoft365GetTokenIdentityError(
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 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"
# Same here, if user can access AAD, some fields are retrieved if not, default value, for az cli
# should work but it doesn't, pending issue
else:
identity.identity_id = "Unknown user id (Missing AAD permissions)"
identity.identity_type = "User"
try:
logger.info(
"Trying to retrieve user information from AAD to populate identity structure ..."
)
client = GraphServiceClient(credentials=credentials)
me = await client.me.get()
if me:
if getattr(me, "user_principal_name"):
identity.identity_id = me.user_principal_name
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
# Retrieve tenant id from the client
client = GraphServiceClient(credentials=credentials)
organization_info = await client.organization.get()
identity.tenant_id = organization_info.value[0].id
asyncio.get_event_loop().run_until_complete(get_microsoft365_identity())
return identity
@staticmethod
def validate_static_credentials(
tenant_id: str = None, client_id: str = None, client_secret: str = None
) -> dict:
"""
Validates the static credentials for the Microsoft365 provider.
Args:
tenant_id (str): The Microsoft365 Active Directory tenant ID.
client_id (str): The Microsoft365 client ID.
client_secret (str): The Microsoft365 client secret.
Raises:
Microsoft365NotValidTenantIdError: If the provided Microsoft365 Tenant ID is not valid.
Microsoft365NotValidClientIdError: If the provided Microsoft365 Client ID is not valid.
Microsoft365NotValidClientSecretError: If the provided Microsoft365 Client Secret is not valid.
Microsoft365ClientIdAndClientSecretNotBelongingToTenantIdError: If the provided Microsoft365 Client ID and Client Secret do not belong to the specified Tenant ID.
Microsoft365TenantIdAndClientSecretNotBelongingToClientIdError: If the provided Microsoft365 Tenant ID and Client Secret do not belong to the specified Client ID.
Microsoft365TenantIdAndClientIdNotBelongingToClientSecretError: If the provided Microsoft365 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 Microsoft365NotValidTenantIdError(
file=os.path.basename(__file__),
message="The provided Microsoft365 Tenant ID is not valid.",
)
# Validate the Client ID
try:
UUID(client_id)
except ValueError:
raise Microsoft365NotValidClientIdError(
file=os.path.basename(__file__),
message="The provided Microsoft365 Client ID is not valid.",
)
# Validate the Client Secret
if not re.match("^[a-zA-Z0-9._~-]+$", client_secret):
raise Microsoft365NotValidClientSecretError(
file=os.path.basename(__file__),
message="The provided Microsoft365 Client Secret is not valid.",
)
try:
Microsoft365Provider.verify_client(tenant_id, client_id, client_secret)
return {
"tenant_id": tenant_id,
"client_id": client_id,
"client_secret": client_secret,
}
except Microsoft365NotValidTenantIdError as tenant_id_error:
logger.error(
f"{tenant_id_error.__class__.__name__}[{tenant_id_error.__traceback__.tb_lineno}]: {tenant_id_error}"
)
raise Microsoft365ClientIdAndClientSecretNotBelongingToTenantIdError(
file=os.path.basename(__file__),
message="The provided Microsoft365 Client ID and Client Secret do not belong to the specified Tenant ID.",
)
except Microsoft365NotValidClientIdError as client_id_error:
logger.error(
f"{client_id_error.__class__.__name__}[{client_id_error.__traceback__.tb_lineno}]: {client_id_error}"
)
raise Microsoft365TenantIdAndClientSecretNotBelongingToClientIdError(
file=os.path.basename(__file__),
message="The provided Microsoft365 Tenant ID and Client Secret do not belong to the specified Client ID.",
)
except Microsoft365NotValidClientSecretError as client_secret_error:
logger.error(
f"{client_secret_error.__class__.__name__}[{client_secret_error.__traceback__.tb_lineno}]: {client_secret_error}"
)
raise Microsoft365TenantIdAndClientIdNotBelongingToClientSecretError(
file=os.path.basename(__file__),
message="The provided Microsoft365 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 Microsoft365 client credentials using the specified tenant ID, client ID, and client secret.
Args:
tenant_id (str): The Microsoft365 Active Directory tenant ID.
client_id (str): The Microsoft365 client ID.
client_secret (str): The Microsoft365 client secret.
Raises:
Microsoft365NotValidTenantIdError: If the provided Microsoft365 Tenant ID is not valid.
Microsoft365NotValidClientIdError: If the provided Microsoft365 Client ID is not valid.
Microsoft365NotValidClientSecretError: If the provided Microsoft365 Client Secret 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,
)
# 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 Microsoft365NotValidTenantIdError(
file=os.path.basename(__file__),
message="The provided Microsoft 365 Tenant ID is not valid for the specified Client ID and Client Secret.",
)
if f"Application with identifier '{client_id}'" in error_description:
raise Microsoft365NotValidClientIdError(
file=os.path.basename(__file__),
message="The provided Microsoft 365 Client ID is not valid for the specified Tenant ID and Client Secret.",
)
if "Invalid client secret provided" in error_description:
raise Microsoft365NotValidClientSecretError(
file=os.path.basename(__file__),
message="The provided Microsoft 365 Client Secret is not valid for the specified Tenant ID and Client ID.",
)
except Exception as e:
# Generic exception handling (if needed)
raise RuntimeError(f"An unexpected error occurred: {str(e)}")
+44
View File
@@ -0,0 +1,44 @@
from pydantic import BaseModel
from prowler.config.config import output_file_timestamp
from prowler.providers.common.models import ProviderOutputOptions
class Microsoft365IdentityInfo(BaseModel):
identity_id: str = ""
identity_type: str = ""
tenant_id: str = ""
tenant_domain: str = "Unknown tenant domain (missing AAD permissions)"
location: str = ""
class Microsoft365RegionConfig(BaseModel):
name: str = ""
authority: str = None
base_url: str = ""
credential_scopes: list = []
class Microsoft365OutputOptions(ProviderOutputOptions):
def __init__(self, arguments, bulk_checks_metadata, identity):
# First call Provider_Output_Options init
super().__init__(arguments, bulk_checks_metadata)
# Check if custom output filename was input, if not, set the default
if (
not hasattr(arguments, "output_filename")
or arguments.output_filename is None
):
if (
identity.tenant_domain
!= "Unknown tenant domain (missing AAD permissions)"
):
self.output_filename = (
f"prowler-output-{identity.tenant_domain}-{output_file_timestamp}"
)
else:
self.output_filename = (
f"prowler-output-{identity.tenant_id}-{output_file_timestamp}"
)
else:
self.output_filename = arguments.output_filename
@@ -0,0 +1,6 @@
from prowler.providers.common.provider import Provider
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
AdminCenter,
)
admincenter_client = AdminCenter(Provider.get_global_provider())
@@ -0,0 +1,30 @@
{
"Provider": "microsoft365",
"CheckID": "admincenter_groups_not_public_visibility",
"CheckTitle": "Ensure that only organizationally managed/approved public groups exist",
"CheckType": [],
"ServiceName": "admincenter",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "high",
"ResourceType": "Microsoft365Group",
"Description": "Ensure that only organizationally managed and approved public groups exist to prevent unauthorized access to sensitive group resources like SharePoint, Teams, or other shared assets.",
"Risk": "Unmanaged public groups can allow unauthorized access to organizational resources, posing a risk of data leakage or misuse through easily guessable SharePoint URLs or self-adding to groups via the Azure portal.",
"RelatedUrl": "https://learn.microsoft.com/en-us/microsoft-365/admin/create-groups/manage-groups?view=o365-worldwide",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "",
"Terraform": ""
},
"Recommendation": {
"Text": "Review and adjust the privacy settings of Microsoft 365 Groups to ensure only organizationally managed and approved public groups exist.",
"Url": "https://learn.microsoft.com/en-us/microsoft-365/security/office-365-security/microsoft-365-groups-governance"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,25 @@
from prowler.lib.check.models import Check, Check_Report_Microsoft365
from prowler.providers.microsoft365.services.admincenter.admincenter_client import (
admincenter_client,
)
class admincenter_groups_not_public_visibility(Check):
def execute(self) -> Check_Report_Microsoft365:
findings = []
for group in admincenter_client.groups.values():
report = Check_Report_Microsoft365(metadata=self.metadata(), resource=group)
report.resource_id = group.id
report.resource_name = group.name
report.status = "FAIL"
report.status_extended = f"Group {group.name} has {group.visibility} visibility and should be Private."
if group.visibility != "Public":
report.status = "PASS"
report.status_extended = (
f"Group {group.name} has {group.visibility} visibility."
)
findings.append(report)
return findings
@@ -0,0 +1,151 @@
from asyncio import gather, get_event_loop
from typing import List, Optional
from msgraph.generated.models.o_data_errors.o_data_error import ODataError
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.providers.microsoft365.lib.service.service import Microsoft365Service
from prowler.providers.microsoft365.microsoft365_provider import Microsoft365Provider
class AdminCenter(Microsoft365Service):
def __init__(self, provider: Microsoft365Provider):
super().__init__(provider)
loop = get_event_loop()
# Get users first alone because it is a dependency for other attributes
self.users = loop.run_until_complete(self._get_users())
attributes = loop.run_until_complete(
gather(
self._get_directory_roles(),
self._get_groups(),
)
)
self.directory_roles = attributes[0]
self.groups = attributes[1]
async def _get_users(self):
logger.info("Microsoft365 - Getting users...")
users = {}
try:
users_list = await self.client.users.get()
users.update({})
for user in users_list.value:
license_details = await self.client.users.by_user_id(
user.id
).license_details.get()
try:
mailbox_settings = await self.client.users.by_user_id(
user.id
).mailbox_settings.get()
mailbox_settings.user_purpose
except ODataError as error:
if error.error.code == "MailboxNotEnabledForRESTAPI":
logger.warning(
f"MailboxNotEnabledForRESTAPI for user {user.id}"
)
else:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
users.update(
{
user.id: User(
id=user.id,
name=user.display_name,
license=(
license_details.value[0].sku_part_number
if license_details.value
else None
),
)
}
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return users
async def _get_directory_roles(self):
logger.info("Microsoft365 - Getting directory roles...")
directory_roles_with_members = {}
try:
directory_roles_with_members.update({})
directory_roles = await self.client.directory_roles.get()
for directory_role in directory_roles.value:
directory_role_members = (
await self.client.directory_roles.by_directory_role_id(
directory_role.id
).members.get()
)
members_with_roles = []
for member in directory_role_members.value:
user = self.users.get(member.id, None)
if user:
user.directory_roles.append(directory_role.display_name)
members_with_roles.append(user)
directory_roles_with_members.update(
{
directory_role.display_name: DirectoryRole(
id=directory_role.id,
name=directory_role.display_name,
members=members_with_roles,
)
}
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return directory_roles_with_members
async def _get_groups(self):
logger.info("Microsoft365 - Getting groups...")
groups = {}
try:
groups_list = await self.client.groups.get()
groups.update({})
for group in groups_list.value:
groups.update(
{
group.id: Group(
id=group.id,
name=group.display_name,
visibility=group.visibility,
)
}
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return groups
class User(BaseModel):
id: str
name: str
directory_roles: List[str] = []
license: Optional[str] = None
user_type: Optional[str] = None
class DirectoryRole(BaseModel):
id: str
name: str
members: List[User]
class Group(BaseModel):
id: str
name: str
visibility: str
@@ -0,0 +1,30 @@
{
"Provider": "microsoft365",
"CheckID": "admincenter_users_admins_reduced_license_footprint",
"CheckTitle": "Ensure administrative accounts use licenses with a reduced application footprint",
"CheckType": [],
"ServiceName": "admincenter",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "AdministrativeAccount",
"Description": "Administrative accounts must use licenses with a reduced application footprint, such as Microsoft Entra ID P1 or P2, or avoid licenses entirely when possible. This minimizes the attack surface associated with privileged identities.",
"Risk": "Licensing administrative accounts with applications like email or collaborative tools increases their exposure to social engineering attacks and malicious content, putting privileged accounts at risk.",
"RelatedUrl": "https://learn.microsoft.com/en-us/microsoft-365/enterprise/protect-your-global-administrator-accounts?view=o365-worldwide",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "",
"Terraform": ""
},
"Recommendation": {
"Text": "Assign Microsoft Entra ID P1 or P2 licenses to administrative accounts to participate in essential security services without enabling access to vulnerable applications.",
"Url": "https://learn.microsoft.com/en-us/microsoft-365/admin/add-users/add-users?view=o365-worldwide"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,35 @@
from prowler.lib.check.models import Check, Check_Report_Microsoft365
from prowler.providers.microsoft365.services.admincenter.admincenter_client import (
admincenter_client,
)
class admincenter_users_admins_reduced_license_footprint(Check):
def execute(self) -> Check_Report_Microsoft365:
findings = []
allowed_licenses = ["AAD_PREMIUM", "AAD_PREMIUM_P2"]
for user in admincenter_client.users.values():
admin_roles = ", ".join(
[
role
for role in user.directory_roles
if "Administrator" in role or "Globar Reader" in role
]
)
if admin_roles:
report = Check_Report_Microsoft365(
metadata=self.metadata(), resource=user
)
report.resource_id = user.id
report.resource_name = user.name
report.status = "FAIL"
report.status_extended = f"User {user.name} has administrative roles {admin_roles} and an invalid license {user.license if user.license else ''}."
if user.license in allowed_licenses:
report.status = "PASS"
report.status_extended = f"User {user.name} has administrative roles {admin_roles} and a valid license: {user.license}."
findings.append(report)
return findings
@@ -0,0 +1,30 @@
{
"Provider": "microsoft365",
"CheckID": "admincenter_users_between_two_and_four_global_admins",
"CheckTitle": "Ensure that between two and four global admins are designated",
"CheckType": [],
"ServiceName": "admincenter",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "AdministrativeRole",
"Description": "Ensure that there are between two and four global administrators designated in your tenant. This ensures monitoring, redundancy, and reduces the risk associated with having too many privileged accounts.",
"Risk": "Having only one global administrator increases the risk of unmonitored actions and operational disruptions if that administrator is unavailable. Having more than four increases the likelihood of a breach through one of these highly privileged accounts.",
"RelatedUrl": "https://learn.microsoft.com/en-us/entra/identity/role-based-access-control/best-practices#5-limit-the-number-of-global-administrators-to-less-than-5",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "",
"Terraform": ""
},
"Recommendation": {
"Text": "Review the number of global administrators in your tenant. Add or remove global admins as necessary to ensure compliance with the recommended range of two to four.",
"Url": "https://learn.microsoft.com/en-us/entra/identity/role-based-access-control/manage-roles-portal"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,39 @@
from prowler.lib.check.models import Check, Check_Report_Microsoft365
from prowler.providers.microsoft365.services.admincenter.admincenter_client import (
admincenter_client,
)
class admincenter_users_between_two_and_four_global_admins(Check):
def execute(self) -> Check_Report_Microsoft365:
findings = []
directory_roles = admincenter_client.directory_roles
report = Check_Report_Microsoft365(
metadata=self.metadata(), resource=admincenter_client.directory_roles
)
report.status = "FAIL"
report.resource_name = "Global Administrator"
if "Global Administrator" in directory_roles:
report.resource_id = getattr(
directory_roles["Global Administrator"],
"id",
"Global Administrator",
)
num_global_admins = len(
getattr(directory_roles["Global Administrator"], "members", [])
)
if num_global_admins >= 2 and num_global_admins < 5:
report.status = "PASS"
report.status_extended = (
f"There are {num_global_admins} global administrators."
)
else:
report.status_extended = f"There are {num_global_admins} global administrators. It should be more than one and less than five."
findings.append(report)
return findings
+2 -4
View File
@@ -16,13 +16,11 @@ prowler_command = "prowler"
# capsys
# https://docs.pytest.org/en/7.1.x/how-to/capture-stdout-stderr.html
prowler_default_usage_error = (
"usage: prowler [-h] [--version] {aws,azure,gcp,kubernetes,dashboard} ..."
)
prowler_default_usage_error = "usage: prowler [-h] [--version] {aws,azure,gcp,kubernetes,microsoft365,dashboard} ..."
def mock_get_available_providers():
return ["aws", "azure", "gcp", "kubernetes"]
return ["aws", "azure", "gcp", "kubernetes", "microsoft365"]
@pytest.mark.arg_parser
+1 -1
View File
@@ -438,7 +438,7 @@ class TestAzureProvider:
raise_on_exception=True,
)
assert exception.type == Exception
assert exception.type is Exception
assert exception.value.args[0] == "Simulated Exception"
@pytest.mark.parametrize(
@@ -0,0 +1,16 @@
### Account, Check and/or Region can be * to apply for all the cases.
### Resources and tags are lists that can have either Regex or Keywords.
### Tags is an optional list that matches on tuples of 'key=value' and are "ANDed" together.
### Use an alternation Regex to match one of multiple tags with "ORed" logic.
### For each check you can except Accounts, Regions, Resources and/or Tags.
########################### MUTELIST EXAMPLE ###########################
Mutelist:
Accounts:
"subscription_1":
Checks:
"admincenter_users_between_two_and_four_global_admins":
Regions:
- "*"
Resources:
- "resource_1"
- "resource_2"
@@ -0,0 +1,103 @@
import yaml
from mock import MagicMock
from prowler.providers.microsoft365.lib.mutelist.mutelist import Microsoft365Mutelist
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
MUTELIST_FIXTURE_PATH = (
"tests/providers/microsoft365/lib/mutelist/fixtures/microsoft365_mutelist.yaml"
)
class TestMicrosoft365Mutelist:
def test_get_mutelist_file_from_local_file(self):
mutelist = Microsoft365Mutelist(mutelist_path=MUTELIST_FIXTURE_PATH)
with open(MUTELIST_FIXTURE_PATH) as f:
mutelist_fixture = yaml.safe_load(f)["Mutelist"]
assert mutelist.mutelist == mutelist_fixture
assert mutelist.mutelist_file_path == MUTELIST_FIXTURE_PATH
def test_get_mutelist_file_from_local_file_non_existent(self):
mutelist_path = "tests/lib/mutelist/fixtures/not_present"
mutelist = Microsoft365Mutelist(mutelist_path=mutelist_path)
assert mutelist.mutelist == {}
assert mutelist.mutelist_file_path == mutelist_path
def test_validate_mutelist_not_valid_key(self):
mutelist_path = MUTELIST_FIXTURE_PATH
with open(mutelist_path) as f:
mutelist_fixture = yaml.safe_load(f)["Mutelist"]
mutelist_fixture["Accounts1"] = mutelist_fixture["Accounts"]
del mutelist_fixture["Accounts"]
mutelist = Microsoft365Mutelist(mutelist_content=mutelist_fixture)
assert not mutelist.validate_mutelist()
assert mutelist.mutelist == {}
assert mutelist.mutelist_file_path is None
def test_is_finding_muted(self):
# Mutelist
mutelist_content = {
"Accounts": {
"subscription_1": {
"Checks": {
"check_test": {
"Regions": ["*"],
"Resources": ["test_resource"],
}
}
}
}
}
mutelist = Microsoft365Mutelist(mutelist_content=mutelist_content)
finding = MagicMock
finding.tenant_id = "subscription_1"
finding.check_metadata = MagicMock
finding.check_metadata.CheckID = "check_test"
finding.status = "FAIL"
finding.location = "global"
finding.resource_name = "test_resource"
finding.tenant_domain = "test_domain"
finding.resource_tags = []
assert mutelist.is_finding_muted(finding)
def test_mute_finding(self):
# Mutelist
mutelist_content = {
"Accounts": {
"subscription_1": {
"Checks": {
"check_test": {
"Regions": ["*"],
"Resources": ["test_resource"],
}
}
}
}
}
mutelist = Microsoft365Mutelist(mutelist_content=mutelist_content)
finding_1 = generate_finding_output(
check_id="check_test",
status="FAIL",
account_uid="subscription_1",
region="subscription_1",
resource_uid="test_resource",
resource_tags=[],
muted=False,
)
muted_finding = mutelist.mute_finding(finding=finding_1)
assert muted_finding.status == "MUTED"
assert muted_finding.muted is True
assert muted_finding.raw["status"] == "FAIL"
@@ -0,0 +1,38 @@
from azure.identity import AzureAuthorityHosts
from prowler.providers.microsoft365.lib.regions.microsoft365_regions import (
MICROSOFT365_CHINA_CLOUD,
MICROSOFT365_GENERIC_CLOUD,
MICROSOFT365_US_GOV_CLOUD,
get_regions_config,
)
class Test_microsoft365_regions:
def test_get_regions_config(self):
allowed_regions = [
"Microsoft365Global",
"Microsoft365China",
"Microsoft365USGovernment",
]
expected_output = {
"Microsoft365Global": {
"authority": None,
"base_url": MICROSOFT365_GENERIC_CLOUD,
"credential_scopes": [MICROSOFT365_GENERIC_CLOUD + "/.default"],
},
"Microsoft365China": {
"authority": AzureAuthorityHosts.AZURE_CHINA,
"base_url": MICROSOFT365_CHINA_CLOUD,
"credential_scopes": [MICROSOFT365_CHINA_CLOUD + "/.default"],
},
"Microsoft365USGovernment": {
"authority": AzureAuthorityHosts.AZURE_GOVERNMENT,
"base_url": MICROSOFT365_US_GOV_CLOUD,
"credential_scopes": [MICROSOFT365_US_GOV_CLOUD + "/.default"],
},
}
for region in allowed_regions:
region_config = get_regions_config(region)
assert region_config == expected_output[region]
@@ -0,0 +1,40 @@
from azure.identity import DefaultAzureCredential
from mock import MagicMock
from prowler.providers.microsoft365.microsoft365_provider import Microsoft365Provider
from prowler.providers.microsoft365.models import (
Microsoft365IdentityInfo,
Microsoft365RegionConfig,
)
# Azure Identity
IDENTITY_ID = "00000000-0000-0000-0000-000000000000"
IDENTITY_TYPE = "Application"
TENANT_ID = "00000000-0000-0000-0000-000000000000"
CLIENT_ID = "00000000-0000-0000-0000-000000000000"
CLIENT_SECRET = "00000000-0000-0000-0000-000000000000"
DOMAIN = "user.onmicrosoft.com"
LOCATION = "global"
# Mocked Azure Audit Info
def set_mocked_microsoft365_provider(
credentials: DefaultAzureCredential = DefaultAzureCredential(),
identity: Microsoft365IdentityInfo = Microsoft365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type=IDENTITY_TYPE,
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
),
audit_config: dict = None,
azure_region_config: Microsoft365RegionConfig = Microsoft365RegionConfig(),
) -> Microsoft365Provider:
provider = MagicMock()
provider.type = "microsoft365"
provider.session.credentials = credentials
provider.identity = identity
provider.audit_config = audit_config
provider.region_config = azure_region_config
return provider
@@ -0,0 +1,314 @@
from unittest.mock import patch
from uuid import uuid4
import pytest
from azure.core.credentials import AccessToken
from azure.identity import (
ClientSecretCredential,
DefaultAzureCredential,
InteractiveBrowserCredential,
)
from mock import MagicMock
from prowler.config.config import (
default_config_file_path,
default_fixer_config_file_path,
load_and_validate_config_file,
)
from prowler.providers.common.models import Connection
from prowler.providers.microsoft365.exceptions.exceptions import (
Microsoft365HTTPResponseError,
Microsoft365NoAuthenticationMethodError,
)
from prowler.providers.microsoft365.microsoft365_provider import Microsoft365Provider
from prowler.providers.microsoft365.models import (
Microsoft365IdentityInfo,
Microsoft365RegionConfig,
)
from tests.providers.microsoft365.microsoft365_fixtures import (
CLIENT_ID,
CLIENT_SECRET,
DOMAIN,
IDENTITY_ID,
IDENTITY_TYPE,
LOCATION,
TENANT_ID,
)
class TestMicrosoft365Provider:
def test_microsoft365_provider(self):
tenant_id = None
client_id = None
client_secret = None
fixer_config = load_and_validate_config_file(
"microsoft365", default_fixer_config_file_path
)
azure_region = "Microsoft365Global"
with (
patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_session",
return_value=ClientSecretCredential(
client_id=CLIENT_ID,
tenant_id=TENANT_ID,
client_secret=CLIENT_SECRET,
),
),
patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_identity",
return_value=Microsoft365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type=IDENTITY_TYPE,
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
location=LOCATION,
),
),
):
microsoft365_provider = Microsoft365Provider(
sp_env_auth=True,
az_cli_auth=False,
browser_auth=False,
tenant_id=tenant_id,
client_id=client_id,
client_secret=client_secret,
region=azure_region,
config_path=default_config_file_path,
fixer_config=fixer_config,
)
assert microsoft365_provider.region_config == Microsoft365RegionConfig(
name="Microsoft365Global",
authority=None,
base_url="https://graph.microsoft.com",
credential_scopes=["https://graph.microsoft.com/.default"],
)
assert microsoft365_provider.identity == Microsoft365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type=IDENTITY_TYPE,
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
location=LOCATION,
)
def test_microsoft365_provider_cli_auth(self):
"""Test Microsoft365 Provider initialization with CLI authentication"""
azure_region = "Microsoft365Global"
fixer_config = load_and_validate_config_file(
"microsoft365", default_fixer_config_file_path
)
with (
patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_session",
return_value=DefaultAzureCredential(
exclude_environment_credential=True,
exclude_cli_credential=False,
exclude_managed_identity_credential=True,
exclude_visual_studio_code_credential=True,
exclude_shared_token_cache_credential=True,
exclude_powershell_credential=True,
exclude_browser_credential=True,
),
),
patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_identity",
return_value=Microsoft365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
location=LOCATION,
),
),
):
microsoft365_provider = Microsoft365Provider(
sp_env_auth=False,
az_cli_auth=True,
browser_auth=False,
region=azure_region,
config_path=default_config_file_path,
fixer_config=fixer_config,
)
assert microsoft365_provider.region_config == Microsoft365RegionConfig(
name="Microsoft365Global",
authority=None,
base_url="https://graph.microsoft.com",
credential_scopes=["https://graph.microsoft.com/.default"],
)
assert microsoft365_provider.identity == Microsoft365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
location=LOCATION,
)
assert isinstance(microsoft365_provider.session, DefaultAzureCredential)
def test_microsoft365_provider_browser_auth(self):
"""Test Microsoft365 Provider initialization with Browser authentication"""
azure_region = "Microsoft365Global"
fixer_config = load_and_validate_config_file(
"microsoft365", default_fixer_config_file_path
)
with (
patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_session",
return_value=InteractiveBrowserCredential(
tenant_id=TENANT_ID,
),
),
patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_identity",
return_value=Microsoft365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
location=LOCATION,
),
),
):
microsoft365_provider = Microsoft365Provider(
sp_env_auth=False,
az_cli_auth=False,
browser_auth=True,
tenant_id=TENANT_ID,
region=azure_region,
config_path=default_config_file_path,
fixer_config=fixer_config,
)
assert microsoft365_provider.region_config == Microsoft365RegionConfig(
name="Microsoft365Global",
authority=None,
base_url="https://graph.microsoft.com",
credential_scopes=["https://graph.microsoft.com/.default"],
)
assert microsoft365_provider.identity == Microsoft365IdentityInfo(
identity_id=IDENTITY_ID,
identity_type="User",
tenant_id=TENANT_ID,
tenant_domain=DOMAIN,
location=LOCATION,
)
assert isinstance(
microsoft365_provider.session, InteractiveBrowserCredential
)
def test_test_connection_browser_auth(self):
with (
patch(
"prowler.providers.microsoft365.microsoft365_provider.DefaultAzureCredential"
) as mock_default_credential,
patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_session"
) as mock_setup_session,
patch(
"prowler.providers.microsoft365.microsoft365_provider.GraphServiceClient"
) as mock_graph_client,
):
# 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 = mock_credentials
# Mock setup_session to return a mocked session object
mock_session = MagicMock()
mock_setup_session.return_value = mock_session
# Mock GraphServiceClient to avoid real API calls
mock_client = MagicMock()
mock_graph_client.return_value = mock_client
test_connection = Microsoft365Provider.test_connection(
browser_auth=True,
tenant_id=str(uuid4()),
region="Microsoft365Global",
raise_on_exception=False,
)
assert isinstance(test_connection, Connection)
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.microsoft365.microsoft365_provider.Microsoft365Provider.setup_session"
) as mock_setup_session,
patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.validate_static_credentials"
) as mock_validate_static_credentials,
):
# 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
test_connection = Microsoft365Provider.test_connection(
tenant_id=str(uuid4()),
region="Microsoft365Global",
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_httpresponseerror(self):
with patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_session"
) as mock_setup_session:
mock_setup_session.side_effect = Microsoft365HTTPResponseError(
file="test_file", original_exception="Simulated HttpResponseError"
)
with pytest.raises(Microsoft365HTTPResponseError) as exception:
Microsoft365Provider.test_connection(
az_cli_auth=True,
raise_on_exception=True,
)
assert exception.type == Microsoft365HTTPResponseError
assert (
exception.value.args[0]
== "[6003] Error in HTTP response from Microsoft365 - Simulated HttpResponseError"
)
def test_test_connection_with_exception(self):
with patch(
"prowler.providers.microsoft365.microsoft365_provider.Microsoft365Provider.setup_session"
) as mock_setup_session:
mock_setup_session.side_effect = Exception("Simulated Exception")
with pytest.raises(Exception) as exception:
Microsoft365Provider.test_connection(
sp_env_auth=True,
raise_on_exception=True,
)
assert exception.type is Exception
assert exception.value.args[0] == "Simulated Exception"
def test_test_connection_without_any_method(self):
with pytest.raises(Microsoft365NoAuthenticationMethodError) as exception:
Microsoft365Provider.test_connection()
assert exception.type == Microsoft365NoAuthenticationMethodError
assert (
"Microsoft365 provider requires at least one authentication method set: [--az-cli-auth | --sp-env-auth | --browser-auth]"
in exception.value.args[0]
)
@@ -0,0 +1,97 @@
from unittest import mock
from uuid import uuid4
from tests.providers.microsoft365.microsoft365_fixtures import (
DOMAIN,
set_mocked_microsoft365_provider,
)
class Test_admincenter_groups_not_public_visibility:
def test_admincenter_no_groups(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
), mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_groups_not_public_visibility.admincenter_groups_not_public_visibility.admincenter_client",
new=admincenter_client,
):
from prowler.providers.microsoft365.services.admincenter.admincenter_groups_not_public_visibility.admincenter_groups_not_public_visibility import (
admincenter_groups_not_public_visibility,
)
admincenter_client.groups = {}
check = admincenter_groups_not_public_visibility()
result = check.execute()
assert len(result) == 0
def test_admincenter_user_no_admin(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
), mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_groups_not_public_visibility.admincenter_groups_not_public_visibility.admincenter_client",
new=admincenter_client,
):
from prowler.providers.microsoft365.services.admincenter.admincenter_groups_not_public_visibility.admincenter_groups_not_public_visibility import (
admincenter_groups_not_public_visibility,
)
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
Group,
)
id_group1 = str(uuid4())
admincenter_client.groups = {
id_group1: Group(id=id_group1, name="Group1", visibility="Private"),
}
check = admincenter_groups_not_public_visibility()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended == "Group Group1 has Private visibility."
assert result[0].resource_name == "Group1"
assert result[0].resource_id == id_group1
def test_admincenter_user_admin_compliant_license(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
), mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_groups_not_public_visibility.admincenter_groups_not_public_visibility.admincenter_client",
new=admincenter_client,
):
from prowler.providers.microsoft365.services.admincenter.admincenter_groups_not_public_visibility.admincenter_groups_not_public_visibility import (
admincenter_groups_not_public_visibility,
)
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
Group,
)
id_group1 = str(uuid4())
admincenter_client.groups = {
id_group1: Group(id=id_group1, name="Group1", visibility="Private"),
}
check = admincenter_groups_not_public_visibility()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended == "Group Group1 has Private visibility."
assert result[0].resource_name == "Group1"
assert result[0].resource_id == id_group1
@@ -0,0 +1,84 @@
from unittest.mock import patch
from prowler.providers.microsoft365.models import Microsoft365IdentityInfo
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
AdminCenter,
DirectoryRole,
Group,
User,
)
from tests.providers.microsoft365.microsoft365_fixtures import (
DOMAIN,
set_mocked_microsoft365_provider,
)
async def mock_admincenter_get_users(_):
return {
"user-1@tenant1.es": User(
id="id-1",
name="User 1",
directory_roles=[],
),
}
async def mock_admincenter_get_directory_roles(_):
return {
"GlobalAdministrator": DirectoryRole(
id="id-directory-role",
name="GlobalAdministrator",
members=[],
)
}
async def mock_admincenter_get_groups(_):
return {
"id-1": Group(id="id-1", name="Test", visibility="Public"),
}
@patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_service.AdminCenter._get_users",
new=mock_admincenter_get_users,
)
@patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_service.AdminCenter._get_directory_roles",
new=mock_admincenter_get_directory_roles,
)
@patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_service.AdminCenter._get_groups",
new=mock_admincenter_get_groups,
)
class Test_AdminCenter_Service:
def test_get_client(self):
admincenter_client = AdminCenter(
set_mocked_microsoft365_provider(
identity=Microsoft365IdentityInfo(tenant_domain=DOMAIN)
)
)
assert admincenter_client.client.__class__.__name__ == "GraphServiceClient"
def test_get_users(self):
admincenter_client = AdminCenter(set_mocked_microsoft365_provider())
assert len(admincenter_client.users) == 1
assert admincenter_client.users["user-1@tenant1.es"].id == "id-1"
assert admincenter_client.users["user-1@tenant1.es"].name == "User 1"
def test_get_group_settings(self):
admincenter_client = AdminCenter(set_mocked_microsoft365_provider())
assert len(admincenter_client.groups) == 1
assert admincenter_client.groups["id-1"].id == "id-1"
assert admincenter_client.groups["id-1"].name == "Test"
assert admincenter_client.groups["id-1"].visibility == "Public"
def test_get_directory_roles(self):
admincenter_client = AdminCenter(set_mocked_microsoft365_provider())
assert (
admincenter_client.directory_roles["GlobalAdministrator"].id
== "id-directory-role"
)
assert (
len(admincenter_client.directory_roles["GlobalAdministrator"].members) == 0
)
@@ -0,0 +1,159 @@
from unittest import mock
from uuid import uuid4
from tests.providers.microsoft365.microsoft365_fixtures import (
DOMAIN,
set_mocked_microsoft365_provider,
)
class Test_admincenter_users_admins_reduced_license_footprint:
def test_admincenter_no_users(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
),
mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_users_admins_reduced_license_footprint.admincenter_users_admins_reduced_license_footprint.admincenter_client",
new=admincenter_client,
),
):
from prowler.providers.microsoft365.services.admincenter.admincenter_users_admins_reduced_license_footprint.admincenter_users_admins_reduced_license_footprint import (
admincenter_users_admins_reduced_license_footprint,
)
admincenter_client.users = {}
check = admincenter_users_admins_reduced_license_footprint()
result = check.execute()
assert len(result) == 0
def test_admincenter_user_no_admin(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
),
mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_users_admins_reduced_license_footprint.admincenter_users_admins_reduced_license_footprint.admincenter_client",
new=admincenter_client,
),
):
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
User,
)
from prowler.providers.microsoft365.services.admincenter.admincenter_users_admins_reduced_license_footprint.admincenter_users_admins_reduced_license_footprint import (
admincenter_users_admins_reduced_license_footprint,
)
id_user1 = str(uuid4())
admincenter_client.users = {
id_user1: User(
id=id_user1,
name="User1",
directory_roles=["Exchange User"],
license="O365 BUSINESS",
),
}
check = admincenter_users_admins_reduced_license_footprint()
result = check.execute()
assert len(result) == 0
def test_admincenter_user_admin_compliant_license(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
),
mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_users_admins_reduced_license_footprint.admincenter_users_admins_reduced_license_footprint.admincenter_client",
new=admincenter_client,
),
):
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
User,
)
from prowler.providers.microsoft365.services.admincenter.admincenter_users_admins_reduced_license_footprint.admincenter_users_admins_reduced_license_footprint import (
admincenter_users_admins_reduced_license_footprint,
)
id_user1 = str(uuid4())
admincenter_client.users = {
id_user1: User(
id=id_user1,
name="User1",
directory_roles=["Global Administrator"],
license="AAD_PREMIUM",
),
}
check = admincenter_users_admins_reduced_license_footprint()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "User User1 has administrative roles Global Administrator and a valid license: AAD_PREMIUM."
)
assert result[0].resource_name == "User1"
assert result[0].resource_id == id_user1
def test_admincenter_user_admin_non_compliant_license(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
),
mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_users_admins_reduced_license_footprint.admincenter_users_admins_reduced_license_footprint.admincenter_client",
new=admincenter_client,
),
):
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
User,
)
from prowler.providers.microsoft365.services.admincenter.admincenter_users_admins_reduced_license_footprint.admincenter_users_admins_reduced_license_footprint import (
admincenter_users_admins_reduced_license_footprint,
)
id_user1 = str(uuid4())
admincenter_client.users = {
id_user1: User(
id=id_user1,
name="User1",
directory_roles=["Global Administrator"],
license="O365 BUSINESS",
),
}
check = admincenter_users_admins_reduced_license_footprint()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "User User1 has administrative roles Global Administrator and an invalid license O365 BUSINESS."
)
assert result[0].resource_name == "User1"
assert result[0].resource_id == id_user1
@@ -0,0 +1,172 @@
from unittest import mock
from uuid import uuid4
from tests.providers.microsoft365.microsoft365_fixtures import (
DOMAIN,
set_mocked_microsoft365_provider,
)
class Test_admincenter_users_between_two_and_four_global_admins:
def test_admincenter_no_directory_roles(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
), mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_users_between_two_and_four_global_admins.admincenter_users_between_two_and_four_global_admins.admincenter_client",
new=admincenter_client,
):
from prowler.providers.microsoft365.services.admincenter.admincenter_users_between_two_and_four_global_admins.admincenter_users_between_two_and_four_global_admins import (
admincenter_users_between_two_and_four_global_admins,
)
admincenter_client.directory_roles = {}
check = admincenter_users_between_two_and_four_global_admins()
result = check.execute()
assert len(result) == 0
def test_admincenter_less_than_five_global_admins(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
), mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_users_between_two_and_four_global_admins.admincenter_users_between_two_and_four_global_admins.admincenter_client",
new=admincenter_client,
):
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
DirectoryRole,
User,
)
from prowler.providers.microsoft365.services.admincenter.admincenter_users_between_two_and_four_global_admins.admincenter_users_between_two_and_four_global_admins import (
admincenter_users_between_two_and_four_global_admins,
)
id = str(uuid4())
id_user1 = str(uuid4())
id_user2 = str(uuid4())
admincenter_client.directory_roles = {
"Global Administrator": DirectoryRole(
id=id,
name="Global Administrator",
members=[
User(id=id_user1, name="User1"),
User(id=id_user2, name="User2"),
],
)
}
check = admincenter_users_between_two_and_four_global_admins()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].status_extended == "There are 2 global administrators."
assert result[0].resource_name == "Global Administrator"
assert result[0].resource_id == id
def test_admincenter_more_than_five_global_admins(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
), mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_users_between_two_and_four_global_admins.admincenter_users_between_two_and_four_global_admins.admincenter_client",
new=admincenter_client,
):
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
DirectoryRole,
User,
)
from prowler.providers.microsoft365.services.admincenter.admincenter_users_between_two_and_four_global_admins.admincenter_users_between_two_and_four_global_admins import (
admincenter_users_between_two_and_four_global_admins,
)
id = str(uuid4())
id_user1 = str(uuid4())
id_user2 = str(uuid4())
id_user3 = str(uuid4())
id_user4 = str(uuid4())
id_user5 = str(uuid4())
id_user6 = str(uuid4())
admincenter_client.directory_roles = {
"Global Administrator": DirectoryRole(
id=id,
name="Global Administrator",
members=[
User(id=id_user1, name="User1"),
User(id=id_user2, name="User2"),
User(id=id_user3, name="User3"),
User(id=id_user4, name="User4"),
User(id=id_user5, name="User5"),
User(id=id_user6, name="User6"),
],
)
}
check = admincenter_users_between_two_and_four_global_admins()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "There are 6 global administrators. It should be more than one and less than five."
)
assert result[0].resource_name == "Global Administrator"
assert result[0].resource_id == id
def test_admincenter_one_global_admin(self):
admincenter_client = mock.MagicMock
admincenter_client.audited_tenant = "audited_tenant"
admincenter_client.audited_domain = DOMAIN
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_microsoft365_provider(),
), mock.patch(
"prowler.providers.microsoft365.services.admincenter.admincenter_users_between_two_and_four_global_admins.admincenter_users_between_two_and_four_global_admins.admincenter_client",
new=admincenter_client,
):
from prowler.providers.microsoft365.services.admincenter.admincenter_service import (
DirectoryRole,
User,
)
from prowler.providers.microsoft365.services.admincenter.admincenter_users_between_two_and_four_global_admins.admincenter_users_between_two_and_four_global_admins import (
admincenter_users_between_two_and_four_global_admins,
)
id = str(uuid4())
id_user1 = str(uuid4())
admincenter_client.directory_roles = {
"Global Administrator": DirectoryRole(
id=id,
name="Global Administrator",
members=[
User(id=id_user1, name="User1"),
],
)
}
check = admincenter_users_between_two_and_four_global_admins()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "There are 1 global administrators. It should be more than one and less than five."
)
assert result[0].resource_name == "Global Administrator"
assert result[0].resource_id == id