ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-04 02:04:06 +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 GitHub
co-authored by Andoni A Adrián Jesús Peña Rodríguez
parent de01087246
commit 3fd9c51086
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
+4 -1
View File
@@ -191,8 +191,11 @@ class PowerShellSession:
error_thread.daemon = True
error_thread.start()
try:
result = result_queue.get(timeout=timeout) or default
error_result = error_queue.get(timeout=1)
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
+77 -13
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.
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.
Returns:
dict: Mailbox policy settings in JSON format.
Example:
>>> get_mailbox_policy()
{
"Id": "OwaMailboxPolicy-Default",
"AdditionalStorageProvidersAvailable": True
}
bool: True if all modules were successfully initialized, False otherwise
"""
return self.execute("Get-OwaMailboxPolicy | ConvertTo-Json", json_parse=True)
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
+87 -1
View File
@@ -449,9 +449,11 @@ class TestM365Provider:
"client_secret": "test_client_secret",
}
with patch(
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"