mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 20:42:02 +00:00
feat(fixers): add first version of M365 fixers
This commit is contained in:
@@ -1,34 +1,31 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any, Dict, Optional
|
||||
from typing import Optional
|
||||
|
||||
from colorama import Style
|
||||
|
||||
from prowler.config.config import orange_color
|
||||
from prowler.lib.check.models import CheckReportM365
|
||||
from prowler.lib.fix.fixer import Fixer, FixerMetadata
|
||||
from prowler.lib.fix.fixer import Fixer
|
||||
from prowler.lib.logger import logger
|
||||
|
||||
|
||||
class M365Fixer(Fixer, ABC):
|
||||
class M365Fixer(Fixer):
|
||||
"""M365 specific fixer implementation"""
|
||||
|
||||
def __init__(
|
||||
self, credentials: Optional[Dict] = None, session_config: Optional[Dict] = None
|
||||
self,
|
||||
description: str,
|
||||
cost_impact: bool = False,
|
||||
cost_description: Optional[str] = None,
|
||||
service: str = "",
|
||||
):
|
||||
"""
|
||||
Initialize M365 fixer with optional credentials and session configuration.
|
||||
super().__init__(description, cost_impact, cost_description)
|
||||
self.service = service
|
||||
|
||||
Args:
|
||||
credentials (Optional[Dict]): Optional M365 credentials for authentication
|
||||
session_config (Optional[Dict]): Optional M365 session configuration
|
||||
"""
|
||||
super().__init__(credentials, session_config)
|
||||
self.service: str = ""
|
||||
self.client: Any = None
|
||||
|
||||
@abstractmethod
|
||||
def _get_metadata(self) -> FixerMetadata:
|
||||
def _get_fixer_info(self):
|
||||
"""Each fixer must define its metadata"""
|
||||
fixer_info = super()._get_fixer_info()
|
||||
fixer_info["service"] = self.service
|
||||
return fixer_info
|
||||
|
||||
def fix(self, finding: Optional[CheckReportM365] = None, **kwargs) -> bool:
|
||||
"""
|
||||
@@ -50,7 +47,7 @@ class M365Fixer(Fixer, ABC):
|
||||
resource_id = (
|
||||
finding.resource_id if hasattr(finding, "resource_id") else None
|
||||
)
|
||||
else:
|
||||
elif kwargs.get("resource_id"):
|
||||
resource_id = kwargs.get("resource_id")
|
||||
|
||||
# Print the appropriate message based on available information
|
||||
@@ -59,10 +56,8 @@ class M365Fixer(Fixer, ABC):
|
||||
f"\t{orange_color}FIXING Resource {resource_id}...{Style.RESET_ALL}"
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
"Either finding or required kwargs (resource_id) must be provided"
|
||||
)
|
||||
return False
|
||||
# If no resource_id is provided, we'll still try to proceed
|
||||
print(f"\t{orange_color}FIXING...{Style.RESET_ALL}")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
@@ -53,3 +53,6 @@ class M365OutputOptions(ProviderOutputOptions):
|
||||
)
|
||||
else:
|
||||
self.output_filename = arguments.output_filename
|
||||
|
||||
# Add fixer mode to the output options
|
||||
self.fixer = arguments.fixer
|
||||
|
||||
+72
@@ -0,0 +1,72 @@
|
||||
from typing import Optional
|
||||
|
||||
from prowler.lib.check.models import CheckReportM365
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.m365.lib.fix.fixer import M365Fixer
|
||||
from prowler.providers.m365.services.purview.purview_client import purview_client
|
||||
|
||||
|
||||
class PurviewAuditLogSearchEnabledFixer(M365Fixer):
|
||||
"""
|
||||
Fixer for Purview audit log search.
|
||||
This fixer enables the audit log search using PowerShell.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""
|
||||
Initialize Purview audit log search fixer.
|
||||
"""
|
||||
print("Inicializando PurviewAuditLogSearchEnabledFixer")
|
||||
super().__init__(
|
||||
description="Enable Purview audit log search",
|
||||
cost_impact=False,
|
||||
cost_description=None,
|
||||
service="purview",
|
||||
)
|
||||
|
||||
def fix(self, finding: Optional[CheckReportM365] = None, **kwargs) -> bool:
|
||||
"""
|
||||
Enable Purview audit log search using PowerShell.
|
||||
This fixer executes the Set-AdminAuditLogConfig cmdlet to enable the audit log search.
|
||||
|
||||
Args:
|
||||
finding (Optional[CheckReportM365]): Finding to fix
|
||||
**kwargs: Additional arguments
|
||||
|
||||
Returns:
|
||||
bool: True if the operation is successful (audit log search is enabled), False otherwise
|
||||
"""
|
||||
try:
|
||||
print("Iniciando fix de Purview Audit Log")
|
||||
# Show the fixing message
|
||||
super().fix()
|
||||
|
||||
print(f"Estado de purview_client: {purview_client}")
|
||||
print(f"Estado de purview_client.powershell: {purview_client.powershell}")
|
||||
|
||||
# Connect to Exchange Online
|
||||
if purview_client.powershell:
|
||||
print("Conectando a Exchange Online")
|
||||
purview_client.powershell.connect_exchange_online()
|
||||
try:
|
||||
# Execute the command to enable audit log search
|
||||
print("Ejecutando comando pwsh")
|
||||
purview_client.powershell.execute(
|
||||
"Set-AdminAuditLogConfig -UnifiedAuditLogIngestionEnabled $true"
|
||||
)
|
||||
return True
|
||||
finally:
|
||||
# Always close the PowerShell session
|
||||
print("Cerrando sesión de PowerShell")
|
||||
purview_client.powershell.close()
|
||||
else:
|
||||
logger.error("PowerShell session could not be initialized")
|
||||
print("Error al inicializar la sesión de PowerShell")
|
||||
return False
|
||||
|
||||
except Exception as error:
|
||||
print(f"Error en el fixer: {str(error)}")
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
return False
|
||||
@@ -13,7 +13,8 @@ class Purview(M365Service):
|
||||
if self.powershell:
|
||||
self.powershell.connect_exchange_online()
|
||||
self.audit_log_config = self._get_audit_log_config()
|
||||
self.powershell.close()
|
||||
if not provider.output_options.fixer:
|
||||
self.powershell.close()
|
||||
|
||||
def _get_audit_log_config(self):
|
||||
logger.info("M365 - Getting Admin Audit Log settings...")
|
||||
|
||||
Reference in New Issue
Block a user