ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 12:31:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(m365): automate PowerShell modules installation (#7618)

Browse Source 
Co-authored-by: Andoni A <14891798+andoniaf@users.noreply.github.com>
Co-authored-by: Adrián Jesús Peña Rodríguez <adrianjpr@gmail.com>
This commit is contained in:
Hugo Pereira Brito
2025-04-30 16:41:59 +02:00
committed by Pepe Fagoaga
parent 81cf5303a1
commit e25ff209b3
9 changed files with 479 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+31 -10
View File
@@ -1,25 +1,43 @@
FROM python:3.12.10-alpine3.20
FROM python:3.12.10-slim-bookworm AS build
LABEL maintainer="https://github.com/prowler-cloud/prowler"
LABEL org.opencontainers.image.source="https://github.com/prowler-cloud/prowler"
# Update system dependencies and install essential tools
#hadolint ignore=DL3018
RUN apk --no-cache upgrade && apk --no-cache add curl git gcc python3-dev musl-dev linux-headers
ARG POWERSHELL_VERSION=7.5.0
# hadolint ignore=DL3008
RUN apt-get update && apt-get install -y --no-install-recommends wget libicu72 \
&& rm -rf /var/lib/apt/lists/*
# Install PowerShell
RUN ARCH=$(uname -m) && \
if [ "$ARCH" = "x86_64" ]; then \
wget --progress=dot:giga https://github.com/PowerShell/PowerShell/releases/download/v${POWERSHELL_VERSION}/powershell-${POWERSHELL_VERSION}-linux-x64.tar.gz -O /tmp/powershell.tar.gz ; \
elif [ "$ARCH" = "aarch64" ]; then \
wget --progress=dot:giga https://github.com/PowerShell/PowerShell/releases/download/v${POWERSHELL_VERSION}/powershell-${POWERSHELL_VERSION}-linux-arm64.tar.gz -O /tmp/powershell.tar.gz ; \
else \
echo "Unsupported architecture: $ARCH" && exit 1 ; \
fi && \
mkdir -p /opt/microsoft/powershell/7 && \
tar zxf /tmp/powershell.tar.gz -C /opt/microsoft/powershell/7 && \
chmod +x /opt/microsoft/powershell/7/pwsh && \
ln -s /opt/microsoft/powershell/7/pwsh /usr/bin/pwsh && \
rm /tmp/powershell.tar.gz
# Add prowler user
RUN addgroup --gid 1000 prowler && \
adduser --uid 1000 --gid 1000 --disabled-password --gecos "" prowler
# Create non-root user
RUN mkdir -p /home/prowler && \
echo 'prowler:x:1000:1000:prowler:/home/prowler:' > /etc/passwd && \
echo 'prowler:x:1000:' > /etc/group && \
chown -R prowler:prowler /home/prowler
USER prowler
# Copy necessary files
WORKDIR /home/prowler
# Copy necessary files
COPY prowler/ /home/prowler/prowler/
COPY dashboard/ /home/prowler/dashboard/
COPY pyproject.toml /home/prowler
COPY README.md /home/prowler/
COPY prowler/providers/m365/lib/powershell/m365_powershell.py /home/prowler/prowler/providers/m365/lib/powershell/m365_powershell.py
# Install Python dependencies
ENV HOME='/home/prowler'
@@ -35,6 +53,9 @@ RUN pip install --no-cache-dir --upgrade pip && \
RUN poetry install --compile && \
rm -rf ~/.cache/pip
# Install PowerShell modules
RUN poetry run python prowler/providers/m365/lib/powershell/m365_powershell.py
# Remove deprecated dash dependencies
RUN pip uninstall dash-html-components -y && \
pip uninstall dash-core-components -y
api/Dockerfile
+28 -13
View File
@@ -1,13 +1,33 @@
FROM python:3.12.8-alpine3.20 AS build
FROM python:3.12.10-slim-bookworm AS build
LABEL maintainer="https://github.com/prowler-cloud/api"
# hadolint ignore=DL3018
RUN apk --no-cache add gcc python3-dev musl-dev linux-headers curl-dev
ARG POWERSHELL_VERSION=7.5.0
ENV POWERSHELL_VERSION=${POWERSHELL_VERSION}
# hadolint ignore=DL3008
RUN apt-get update && apt-get install -y --no-install-recommends wget libicu72 \
&& rm -rf /var/lib/apt/lists/*
# Install PowerShell
RUN ARCH=$(uname -m) && \
if [ "$ARCH" = "x86_64" ]; then \
wget --progress=dot:giga https://github.com/PowerShell/PowerShell/releases/download/v${POWERSHELL_VERSION}/powershell-${POWERSHELL_VERSION}-linux-x64.tar.gz -O /tmp/powershell.tar.gz ; \
elif [ "$ARCH" = "aarch64" ]; then \
wget --progress=dot:giga https://github.com/PowerShell/PowerShell/releases/download/v${POWERSHELL_VERSION}/powershell-${POWERSHELL_VERSION}-linux-arm64.tar.gz -O /tmp/powershell.tar.gz ; \
else \
echo "Unsupported architecture: $ARCH" && exit 1 ; \
fi && \
mkdir -p /opt/microsoft/powershell/7 && \
tar zxf /tmp/powershell.tar.gz -C /opt/microsoft/powershell/7 && \
chmod +x /opt/microsoft/powershell/7/pwsh && \
ln -s /opt/microsoft/powershell/7/pwsh /usr/bin/pwsh && \
rm /tmp/powershell.tar.gz
# Add prowler user
RUN addgroup --gid 1000 prowler && \
adduser --uid 1000 --gid 1000 --disabled-password --gecos "" prowler
RUN apk --no-cache upgrade && \
addgroup -g 1000 prowler && \
adduser -D -u 1000 -G prowler prowler
USER prowler
WORKDIR /home/prowler
@@ -27,18 +47,13 @@ RUN poetry install --no-root && \
COPY docker-entrypoint.sh ./docker-entrypoint.sh
RUN poetry run python "$(poetry env info --path)/src/prowler/prowler/providers/m365/lib/powershell/m365_powershell.py"
WORKDIR /home/prowler/backend
# Development image
# hadolint ignore=DL3006
FROM build AS dev
USER 0
# hadolint ignore=DL3018
RUN apk --no-cache add curl vim
USER prowler
ENTRYPOINT ["../docker-entrypoint.sh", "dev"]
# Production image
prowler/lib/powershell/powershell.py
+5 -2
View File
@@ -191,8 +191,11 @@ class PowerShellSession:
error_thread.daemon = True
error_thread.start()
result = result_queue.get(timeout=timeout) or default
error_result = error_queue.get(timeout=1)
try:
result = result_queue.get(timeout=timeout) or default
error_result = error_queue.get(timeout=1) or None
except queue.Empty:
result = default
if error_result:
logger.error(f"PowerShell error output: {error_result}")
prowler/providers/common/provider.py
+1
View File
@@ -221,6 +221,7 @@ class Provider(ABC):
az_cli_auth=arguments.az_cli_auth,
browser_auth=arguments.browser_auth,
tenant_id=arguments.tenant_id,
init_modules=arguments.init_modules,
fixer_config=fixer_config,
)
elif "nhn" in provider_class_name.lower():
prowler/providers/m365/lib/arguments/arguments.py
+3 -10
View File
@@ -35,16 +35,9 @@ def init_parser(self):
help="Microsoft 365 Tenant ID to be used with --browser-auth option",
)
m365_parser.add_argument(
"--user",
nargs="?",
default=None,
help="Microsoft 365 user email",
)
m365_parser.add_argument(
"--encypted-password",
nargs="?",
default=None,
help="Microsoft 365 encrypted password",
"--init-modules",
action="store_true",
help="Initialize Microsoft 365 PowerShell modules",
)
# Regions
m365_regions_subparser = m365_parser.add_argument_group("Regions")
prowler/providers/m365/lib/powershell/m365_powershell.py
+78 -14
View File
@@ -2,6 +2,7 @@ import os
import msal
from prowler.lib.logger import logger
from prowler.lib.powershell.powershell import PowerShellSession
from prowler.providers.m365.exceptions.exceptions import (
M365UserNotBelongingToTenantError,
@@ -26,6 +27,7 @@ class M365PowerShell(PowerShellSession):
Attributes:
credentials (M365Credentials): The Microsoft 365 credentials used for authentication.
required_modules (list): List of required PowerShell modules for M365 operations.
Note:
This class requires the Microsoft Teams and Exchange Online PowerShell modules
@@ -371,6 +373,24 @@ class M365PowerShell(PowerShellSession):
"Get-MailboxAuditBypassAssociation | ConvertTo-Json", json_parse=True
)
def get_mailbox_policy(self) -> dict:
"""
Get Mailbox Policy.
Retrieves the current mailbox policy settings for Exchange Online.
Returns:
dict: Mailbox policy settings in JSON format.
Example:
>>> get_mailbox_policy()
{
"Id": "OwaMailboxPolicy-Default",
"AdditionalStorageProvidersAvailable": True
}
"""
return self.execute("Get-OwaMailboxPolicy | ConvertTo-Json", json_parse=True)
def get_external_mail_config(self) -> dict:
"""
Get Exchange Online External Mail Configuration.
@@ -467,20 +487,64 @@ class M365PowerShell(PowerShellSession):
"Get-HostedContentFilterPolicy | ConvertTo-Json", json_parse=True
)
def get_mailbox_policy(self) -> dict:
"""
Get Mailbox Policy.
Retrieves the current mailbox policy settings for Exchange Online.
# This function is used to install the required M365 PowerShell modules in Docker containers
def initialize_m365_powershell_modules():
"""
Initialize required PowerShell modules.
Returns:
dict: Mailbox policy settings in JSON format.
Checks if the required PowerShell modules are installed and installs them if necessary.
This method ensures that all required modules for M365 operations are available.
Example:
>>> get_mailbox_policy()
{
"Id": "OwaMailboxPolicy-Default",
"AdditionalStorageProvidersAvailable": True
}
"""
return self.execute("Get-OwaMailboxPolicy | ConvertTo-Json", json_parse=True)
Returns:
bool: True if all modules were successfully initialized, False otherwise
"""
REQUIRED_MODULES = [
"ExchangeOnlineManagement",
"MicrosoftTeams",
]
pwsh = PowerShellSession()
try:
for module in REQUIRED_MODULES:
try:
# Check if module is already installed
result = pwsh.execute(
f"Get-Module -ListAvailable -Name {module}", timeout=5
)
# Install module if not installed
if not result:
install_result = pwsh.execute(
f'Install-Module -Name "{module}" -Force -AllowClobber -Scope CurrentUser',
timeout=30,
)
if install_result:
logger.warning(
f"Unexpected output while installing module {module}: {install_result}"
)
else:
logger.info(f"Successfully installed module {module}")
# Import module
pwsh.execute(f'Import-Module -Name "{module}" -Force', timeout=1)
except Exception as error:
logger.error(f"Failed to initialize module {module}: {str(error)}")
return False
return True
finally:
pwsh.close()
def main():
if initialize_m365_powershell_modules():
logger.info("M365 PowerShell modules initialized successfully")
else:
logger.error("Failed to initialize M365 PowerShell modules")
if __name__ == "__main__":
main()
prowler/providers/m365/m365_provider.py
+12 -2
View File
@@ -53,7 +53,10 @@ from prowler.providers.m365.exceptions.exceptions import (
M365TenantIdAndClientSecretNotBelongingToClientIdError,
)
from prowler.providers.m365.lib.mutelist.mutelist import M365Mutelist
from prowler.providers.m365.lib.powershell.m365_powershell import M365PowerShell
from prowler.providers.m365.lib.powershell.m365_powershell import (
M365PowerShell,
initialize_m365_powershell_modules,
)
from prowler.providers.m365.lib.regions.regions import get_regions_config
from prowler.providers.m365.models import (
M365Credentials,
@@ -115,6 +118,7 @@ class M365Provider(Provider):
client_secret: str = None,
user: str = None,
encrypted_password: str = None,
init_modules: bool = False,
region: str = "M365Global",
config_content: dict = None,
config_path: str = None,
@@ -202,6 +206,7 @@ class M365Provider(Provider):
env_auth=env_auth,
m365_credentials=m365_credentials,
provider_id=self.identity.tenant_domain,
init_modules=init_modules,
)
# Audit Config
@@ -373,7 +378,10 @@ class M365Provider(Provider):
@staticmethod
def setup_powershell(
env_auth: bool = False, m365_credentials: dict = {}, provider_id: str = None
env_auth: bool = False,
m365_credentials: dict = {},
provider_id: str = None,
init_modules: bool = False,
) -> M365Credentials:
"""Gets the M365 credentials.
@@ -423,6 +431,8 @@ class M365Provider(Provider):
test_session = M365PowerShell(credentials)
try:
if test_session.test_credentials(credentials):
if init_modules:
initialize_m365_powershell_modules()
return credentials
raise M365EnvironmentUserCredentialsError(
file=os.path.basename(__file__),
tests/providers/m365/lib/powershell/m365_powershell_test.py
+232 -1
View File
@@ -1,7 +1,8 @@
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock, call, patch
import pytest
from prowler.lib.powershell.powershell import PowerShellSession
from prowler.providers.m365.exceptions.exceptions import (
M365UserNotBelongingToTenantError,
)
@@ -265,6 +266,7 @@ class Testm365PowerShell:
- Error in stderr
- Timeout in stdout
- Empty output
- Empty queue handling
"""
# Setup
mock_process = MagicMock()
@@ -308,6 +310,89 @@ class Testm365PowerShell:
result = session.read_output()
assert result == ""
# Test 5: Empty queue handling
mock_process.stdout.readline.side_effect = [] # No output at all
mock_process.stderr.readline.return_value = f"Write-Error: {session.END}\n"
result = session.read_output(timeout=0.1, default="empty_queue")
assert result == "empty_queue"
# Test 6: Empty error queue handling
mock_process.stdout.readline.side_effect = ["test output\n", f"{session.END}\n"]
mock_process.stderr.readline.side_effect = [] # No error output
with patch("prowler.lib.logger.logger.error") as mock_error:
result = session.read_output()
assert result == "test output"
mock_error.assert_not_called()
# Test 7: Both queues empty
mock_process.stdout.readline.side_effect = [] # No output
mock_process.stderr.readline.side_effect = [] # No error output
result = session.read_output(timeout=0.1, default="both_empty")
assert result == "both_empty"
session.close()
@patch("subprocess.Popen")
def test_read_output_queue_empty(self, mock_popen):
"""Test read_output when both queues are empty"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
# Mock process to return empty queues
mock_process.stdout.readline.side_effect = [] # No output
mock_process.stderr.readline.side_effect = [] # No error output
# Test with default value
result = session.read_output(timeout=0.1, default="empty_queue")
assert result == "empty_queue"
# Test without default value
result = session.read_output(timeout=0.1)
assert result == ""
session.close()
@patch("subprocess.Popen")
def test_read_output_error_queue_empty(self, mock_popen):
"""Test read_output when error queue is empty but stdout has content"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
# Mock process to return content in stdout but empty stderr
mock_process.stdout.readline.side_effect = ["test output\n", f"{session.END}\n"]
mock_process.stderr.readline.side_effect = [] # No error output
with patch("prowler.lib.logger.logger.error") as mock_error:
result = session.read_output()
assert result == "test output"
mock_error.assert_not_called()
session.close()
@patch("subprocess.Popen")
def test_read_output_result_queue_empty(self, mock_popen):
"""Test read_output when result queue is empty but stderr has content"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
credentials = M365Credentials(user="test@example.com", passwd="test_password")
session = M365PowerShell(credentials)
# Mock process to return empty stdout but content in stderr
mock_process.stdout.readline.side_effect = [] # No output
mock_process.stderr.readline.side_effect = [
"Error message\n",
f"Write-Error: {session.END}\n",
]
with patch("prowler.lib.logger.logger.error") as mock_error:
result = session.read_output(timeout=0.1, default="default")
assert result == "default"
mock_error.assert_called_once_with("PowerShell error output: Error message")
session.close()
@patch("subprocess.Popen")
@@ -346,3 +431,149 @@ class Testm365PowerShell:
mock_process.stdin.flush.assert_called_once()
mock_process.terminate.assert_called_once()
@patch("subprocess.Popen")
def test_initialize_m365_powershell_modules_success(self, mock_popen):
"""Test initialize_m365_powershell_modules when all modules are successfully initialized"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
# Mock the execute method to simulate successful module installation
def mock_execute(command, *args, **kwargs):
if "Get-Module" in command:
return None # Module not installed
elif "Install-Module" in command:
return None # Installation successful
elif "Import-Module" in command:
return None # Import successful
return None
with (
patch.object(
PowerShellSession, "execute", side_effect=mock_execute
) as mock_execute_obj,
patch("prowler.lib.logger.logger.info") as mock_info,
):
from prowler.providers.m365.lib.powershell.m365_powershell import (
initialize_m365_powershell_modules,
)
result = initialize_m365_powershell_modules()
# Verify successful initialization
assert result is True
# Verify that execute was called for each module
assert mock_execute_obj.call_count == 6 # 2 modules * 3 commands each
# Verify success messages were logged
mock_info.assert_any_call(
"Successfully installed module ExchangeOnlineManagement"
)
mock_info.assert_any_call("Successfully installed module MicrosoftTeams")
@patch("subprocess.Popen")
def test_initialize_m365_powershell_modules_failure(self, mock_popen):
"""Test initialize_m365_powershell_modules when module initialization fails"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
# Mock the execute method to simulate installation failure
def mock_execute(command, *args, **kwargs):
if "Get-Module" in command:
return None # Module not installed
elif "Install-Module" in command:
raise Exception("Installation failed")
return None
with (
patch.object(
PowerShellSession, "execute", side_effect=mock_execute
) as mock_execute_obj,
patch("prowler.lib.logger.logger.error") as mock_error,
):
from prowler.providers.m365.lib.powershell.m365_powershell import (
initialize_m365_powershell_modules,
)
result = initialize_m365_powershell_modules()
# Verify failed initialization
assert result is False
# Verify that execute was called at least twice
assert mock_execute_obj.call_count >= 2
# Verify error was logged
mock_error.assert_called_with(
"Failed to initialize module ExchangeOnlineManagement: Installation failed"
)
@patch("subprocess.Popen")
def test_main_success(self, mock_popen):
"""Test main() function when module initialization is successful"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
# Mock the execute method to simulate successful module installation
def mock_execute(command, *args, **kwargs):
if "Get-Module" in command:
return None # Module not installed
elif "Install-Module" in command:
return None # Installation successful
elif "Import-Module" in command:
return None # Import successful
return None
with (
patch.object(PowerShellSession, "execute", side_effect=mock_execute),
patch("prowler.lib.logger.logger.info") as mock_info,
patch("prowler.lib.logger.logger.error") as mock_error,
):
from prowler.providers.m365.lib.powershell.m365_powershell import main
main()
# Verify all info messages were logged in the correct order
assert mock_info.call_count == 3
mock_info.assert_has_calls(
[
call("Successfully installed module ExchangeOnlineManagement"),
call("Successfully installed module MicrosoftTeams"),
call("M365 PowerShell modules initialized successfully"),
]
)
# Verify no error was logged
mock_error.assert_not_called()
@patch("subprocess.Popen")
def test_main_failure(self, mock_popen):
"""Test main() function when module initialization fails"""
mock_process = MagicMock()
mock_popen.return_value = mock_process
# Mock the execute method to simulate installation failure
def mock_execute(command, *args, **kwargs):
if "Get-Module" in command:
return None # Module not installed
elif "Install-Module" in command:
raise Exception("Installation failed")
return None
with (
patch.object(PowerShellSession, "execute", side_effect=mock_execute),
patch("prowler.lib.logger.logger.info") as mock_info,
patch("prowler.lib.logger.logger.error") as mock_error,
):
from prowler.providers.m365.lib.powershell.m365_powershell import main
main()
# Verify all error messages were logged in the correct order
assert mock_error.call_count == 2
mock_error.assert_has_calls(
[
call(
"Failed to initialize module ExchangeOnlineManagement: Installation failed"
),
call("Failed to initialize M365 PowerShell modules"),
]
)
# Verify no info messages were logged
mock_info.assert_not_called()
tests/providers/m365/m365_provider_test.py
+89 -3
View File
@@ -449,9 +449,11 @@ class TestM365Provider:
"client_secret": "test_client_secret",
}
with patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials",
return_value=True,
with (
patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials",
return_value=True,
),
):
result = M365Provider.setup_powershell(
env_auth=False,
@@ -622,3 +624,87 @@ class TestM365Provider:
f"Provider ID {provider_id} does not match Application tenant domain {user_domain}"
in str(exception.value)
)
def test_provider_init_modules_false(self):
"""Test that initialize_m365_powershell_modules is not called when init_modules is False"""
credentials_dict = {
"user": "test@example.com",
"encrypted_password": "test_password",
"client_id": "test_client_id",
"tenant_id": "test_tenant_id",
"client_secret": "test_client_secret",
}
with (
patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials",
return_value=True,
),
patch(
"prowler.providers.m365.m365_provider.initialize_m365_powershell_modules"
) as mock_init_modules,
):
M365Provider.setup_powershell(
env_auth=False,
m365_credentials=credentials_dict,
provider_id="test_provider_id",
init_modules=False,
)
mock_init_modules.assert_not_called()
def test_provider_init_modules_true(self):
"""Test that initialize_m365_powershell_modules is called when init_modules is True"""
credentials_dict = {
"user": "test@example.com",
"encrypted_password": "test_password",
"client_id": "test_client_id",
"tenant_id": "test_tenant_id",
"client_secret": "test_client_secret",
}
with (
patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials",
return_value=True,
),
patch(
"prowler.providers.m365.m365_provider.initialize_m365_powershell_modules"
) as mock_init_modules,
):
M365Provider.setup_powershell(
env_auth=False,
m365_credentials=credentials_dict,
provider_id="test_provider_id",
init_modules=True,
)
mock_init_modules.assert_called_once()
def test_setup_powershell_init_modules_failure(self):
"""Test that setup_powershell handles initialization failures correctly"""
credentials_dict = {
"user": "test@example.com",
"encrypted_password": "test_password",
"client_id": "test_client_id",
"tenant_id": "test_tenant_id",
"client_secret": "test_client_secret",
}
with (
patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.test_credentials",
return_value=True,
),
patch(
"prowler.providers.m365.m365_provider.initialize_m365_powershell_modules",
side_effect=Exception("Module initialization failed"),
),
):
with pytest.raises(Exception) as exc_info:
M365Provider.setup_powershell(
env_auth=False,
m365_credentials=credentials_dict,
provider_id="test_provider_id",
init_modules=True,
)
assert str(exc_info.value) == "Module initialization failed"