Merge branch 'master' into PROWLER-1391-provider-contract-dynamic-discovery

This commit is contained in:
StylusFrost
2026-04-28 12:42:18 +02:00
committed by GitHub
143 changed files with 9600 additions and 709 deletions
+2 -1
View File
@@ -17,7 +17,7 @@ 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,m365,github,googleworkspace,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel,dashboard,iac,image} ..."
prowler_default_usage_error = "usage: prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,googleworkspace,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,vercel,dashboard,iac,image,llm} ..."
def mock_get_available_providers():
@@ -35,6 +35,7 @@ def mock_get_available_providers():
"mongodbatlas",
"oraclecloud",
"alibabacloud",
"llm",
"cloudflare",
"openstack",
]
+49
View File
@@ -455,6 +455,55 @@ class TestAWSProvider:
aws_provider.organizations_metadata.organization_arn == organization["Arn"]
)
@mock_aws
def test_aws_provider_organizations_uses_assumed_role_session_by_default(self):
# Regression test for issue #10215.
# When only `role_arn` is provided (no `organizations_role_arn`),
# the FIRST attempt to fetch Organizations metadata must use the
# assumed role session (current_session), not the pre-assume
# credentials. This mirrors the CLI: `aws sts assume-role` followed
# by `aws organizations describe-account` uses the assumed identity.
role_arn = create_role(AWS_REGION_EU_WEST_1)
captured_sessions = []
original_get_organizations_info = AwsProvider.get_organizations_info
def capture(self, organizations_session, aws_account_id):
captured_sessions.append(organizations_session)
return original_get_organizations_info(
self, organizations_session, aws_account_id
)
with patch.object(AwsProvider, "get_organizations_info", capture):
aws_provider = AwsProvider(role_arn=role_arn, session_duration=900)
assert captured_sessions[0] is aws_provider.session.current_session
assert captured_sessions[0] is not aws_provider.session.original_session
@mock_aws
def test_aws_provider_organizations_falls_back_to_original_session(self):
# When `role_arn` is provided and the assumed role session cannot
# retrieve Organizations metadata (e.g. management-account ->
# member-account flow where the member account has no Organizations
# permissions), retry with the original (pre-assume) session.
role_arn = create_role(AWS_REGION_EU_WEST_1)
captured_sessions = []
original_get_organizations_info = AwsProvider.get_organizations_info
def capture(self, organizations_session, aws_account_id):
captured_sessions.append(organizations_session)
return original_get_organizations_info(
self, organizations_session, aws_account_id
)
with patch.object(AwsProvider, "get_organizations_info", capture):
aws_provider = AwsProvider(role_arn=role_arn, session_duration=900)
assert len(captured_sessions) == 2
assert captured_sessions[0] is aws_provider.session.current_session
assert captured_sessions[1] is aws_provider.session.original_session
@mock_aws
def test_aws_provider_session_with_mfa(self):
mfa = True
@@ -12,6 +12,7 @@ class Test_GitHubArguments:
self.mock_github_parser = MagicMock()
self.mock_auth_group = MagicMock()
self.mock_scoping_group = MagicMock()
self.mock_actions_group = MagicMock()
# Setup the mock chain
self.mock_parser.add_subparsers.return_value = self.mock_subparsers
@@ -19,6 +20,7 @@ class Test_GitHubArguments:
self.mock_github_parser.add_argument_group.side_effect = [
self.mock_auth_group,
self.mock_scoping_group,
self.mock_actions_group,
]
def test_init_parser_creates_subparser(self):
@@ -47,10 +49,11 @@ class Test_GitHubArguments:
arguments.init_parser(mock_github_args)
# Verify argument groups were created
assert self.mock_github_parser.add_argument_group.call_count == 2
assert self.mock_github_parser.add_argument_group.call_count == 3
calls = self.mock_github_parser.add_argument_group.call_args_list
assert calls[0][0][0] == "Authentication Modes"
assert calls[1][0][0] == "Scan Scoping"
assert calls[2][0][0] == "GitHub Actions Scanning"
def test_init_parser_adds_authentication_arguments(self):
"""Test that init_parser adds all authentication arguments"""
@@ -0,0 +1,367 @@
import io
import json
import sys
from unittest.mock import ANY, MagicMock, patch
from prowler.providers.github.services.githubactions.githubactions_service import (
GithubActions,
GithubActionsWorkflowFinding,
)
class TestGithubActionsService:
def test_should_exclude_workflow_no_patterns(self):
assert not GithubActions._should_exclude_workflow("test.yml", [])
def test_should_exclude_workflow_exact_filename(self):
assert GithubActions._should_exclude_workflow(
".github/workflows/test.yml", ["test.yml"]
)
def test_should_exclude_workflow_wildcard_filename(self):
assert GithubActions._should_exclude_workflow(
".github/workflows/test-api.yml", ["test-*.yml"]
)
def test_should_exclude_workflow_full_path(self):
assert GithubActions._should_exclude_workflow(
".github/workflows/test.yml", [".github/workflows/test.yml"]
)
def test_should_exclude_workflow_full_path_wildcard(self):
assert GithubActions._should_exclude_workflow(
".github/workflows/api-tests.yml", [".github/workflows/api-*.yml"]
)
def test_should_exclude_workflow_no_match(self):
assert not GithubActions._should_exclude_workflow(
".github/workflows/deploy.yml", ["test-*.yml", "api-*.yml"]
)
def test_should_exclude_workflow_multiple_patterns(self):
assert GithubActions._should_exclude_workflow(
".github/workflows/api-test.yml", ["test-*.yml", "api-*.yml"]
)
def test_should_exclude_workflow_filename_in_subdir(self):
assert GithubActions._should_exclude_workflow(
"workflows/subdir/test-deploy.yml", ["test-*.yml"]
)
def test_extract_workflow_file_from_location_v1(self):
location = {
"symbolic": {
"key": {"Local": {"given_path": ".github/workflows/test.yml"}},
}
}
result = GithubActions._extract_workflow_file_from_location(location)
assert result == ".github/workflows/test.yml"
def test_extract_workflow_file_from_location_missing_key(self):
location = {"symbolic": {}}
result = GithubActions._extract_workflow_file_from_location(location)
assert result is None
def test_extract_workflow_file_from_location_empty(self):
result = GithubActions._extract_workflow_file_from_location({})
assert result is None
def test_parse_finding_valid(self):
finding = {
"ident": "template-injection",
"desc": "Template Injection Vulnerability",
"determinations": {"severity": "high", "confidence": "High"},
"url": "https://example.com/docs",
}
location = {
"symbolic": {
"annotation": "High risk of code execution",
"key": {"Local": {"given_path": ".github/workflows/test.yml"}},
},
"concrete": {
"location": {
"start_point": {"row": 10, "column": 5},
"end_point": {"row": 10, "column": 15},
}
},
}
repo = MagicMock()
repo.id = 1
repo.name = "test-repo"
repo.full_name = "owner/test-repo"
repo.owner = "owner"
result = GithubActions._parse_finding(
finding, ".github/workflows/test.yml", location, repo
)
assert isinstance(result, GithubActionsWorkflowFinding)
assert result.finding_id == "githubactions_template_injection"
assert result.ident == "template-injection"
assert result.severity == "high"
assert result.line_range == "line 10"
assert result.workflow_file == ".github/workflows/test.yml"
assert result.repo_name == "test-repo"
assert result.confidence == "High"
def test_parse_finding_multiline_range(self):
finding = {
"ident": "excessive-permissions",
"desc": "Excessive permissions",
"determinations": {"severity": "medium", "confidence": "Medium"},
"url": "https://example.com",
}
location = {
"symbolic": {"annotation": "Excessive permissions detected"},
"concrete": {
"location": {
"start_point": {"row": 5, "column": 1},
"end_point": {"row": 10, "column": 20},
}
},
}
repo = MagicMock()
repo.id = 1
repo.name = "repo"
repo.full_name = "owner/repo"
repo.owner = "owner"
result = GithubActions._parse_finding(finding, "wf.yml", location, repo)
assert result.line_range == "lines 5-10"
def test_parse_finding_unknown_severity(self):
finding = {
"ident": "test",
"desc": "Test",
"determinations": {"severity": "Unknown", "confidence": "Low"},
}
location = {
"symbolic": {},
"concrete": {"location": {}},
}
repo = MagicMock()
repo.id = 1
repo.name = "repo"
repo.full_name = "owner/repo"
repo.owner = "owner"
result = GithubActions._parse_finding(finding, "wf.yml", location, repo)
assert result.severity == "medium"
assert result.line_range == "location unknown"
def test_run_zizmor_no_output(self):
mock_process = MagicMock()
mock_process.stdout = ""
mock_process.stderr = ""
with patch("subprocess.run", return_value=mock_process):
service = GithubActions.__new__(GithubActions)
result = service._run_zizmor("/tmp/test")
assert result == []
def test_run_zizmor_empty_array(self):
mock_process = MagicMock()
mock_process.stdout = "[]"
mock_process.stderr = ""
with patch("subprocess.run", return_value=mock_process):
service = GithubActions.__new__(GithubActions)
result = service._run_zizmor("/tmp/test")
assert result == []
def test_run_zizmor_with_findings(self):
mock_output = [
{
"ident": "excessive-permissions",
"desc": "Workflow has write-all permissions",
"determinations": {"severity": "medium", "confidence": "High"},
"locations": [
{
"symbolic": {
"key": {"Local": {"given_path": ".github/workflows/ci.yml"}}
},
"concrete": {
"location": {
"start_point": {"row": 5, "column": 1},
"end_point": {"row": 5, "column": 20},
}
},
}
],
}
]
mock_process = MagicMock()
mock_process.stdout = json.dumps(mock_output)
mock_process.stderr = ""
with patch("subprocess.run", return_value=mock_process):
service = GithubActions.__new__(GithubActions)
result = service._run_zizmor("/tmp/test")
assert len(result) == 1
assert result[0]["ident"] == "excessive-permissions"
def test_run_zizmor_invalid_json(self):
mock_process = MagicMock()
mock_process.stdout = "not valid json"
mock_process.stderr = ""
with patch("subprocess.run", return_value=mock_process):
service = GithubActions.__new__(GithubActions)
result = service._run_zizmor("/tmp/test")
assert result == []
def test_clone_repository_with_token(self):
with (
patch("tempfile.mkdtemp", return_value="/tmp/test"),
patch("dulwich.porcelain.clone") as mock_clone,
):
service = GithubActions.__new__(GithubActions)
result = service._clone_repository(
"https://github.com/owner/repo", token="mytoken"
)
assert result == "/tmp/test"
mock_clone.assert_called_once_with(
"https://mytoken@github.com/owner/repo",
"/tmp/test",
depth=1,
errstream=ANY,
)
call_kwargs = mock_clone.call_args
assert isinstance(call_kwargs.kwargs["errstream"], io.BytesIO)
def test_clone_repository_without_token(self):
with (
patch("tempfile.mkdtemp", return_value="/tmp/test"),
patch("dulwich.porcelain.clone") as mock_clone,
):
service = GithubActions.__new__(GithubActions)
result = service._clone_repository("https://github.com/owner/repo")
assert result == "/tmp/test"
mock_clone.assert_called_once_with(
"https://github.com/owner/repo",
"/tmp/test",
depth=1,
errstream=ANY,
)
call_kwargs = mock_clone.call_args
assert isinstance(call_kwargs.kwargs["errstream"], io.BytesIO)
def test_clone_repository_failure(self):
with (
patch("tempfile.mkdtemp", return_value="/tmp/test"),
patch("dulwich.porcelain.clone", side_effect=Exception("clone failed")),
):
service = GithubActions.__new__(GithubActions)
result = service._clone_repository("https://github.com/owner/repo")
assert result is None
def test_init_zizmor_missing(self):
mock_provider = MagicMock()
mock_provider.session = MagicMock()
mock_provider.session.token = "test-token"
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_provider.github_actions_enabled = True
with (
patch.object(GithubActions, "__init__", lambda self, provider: None),
patch("shutil.which", return_value=None),
):
service = GithubActions.__new__(GithubActions)
service.provider = mock_provider
service.clients = []
service.audit_config = {}
service.fixer_config = {}
service.findings = {}
# Manually call the part after super().__init__
# Since zizmor is missing, _scan_repositories should not be called
assert service.findings == {}
def test_scan_repositories_strips_temp_dir_prefix(self):
temp_dir = "/var/folders/xx/tmp48xjp_g0"
zizmor_output = [
{
"ident": "template-injection",
"desc": "Template Injection",
"determinations": {"severity": "high", "confidence": "High"},
"url": "https://example.com",
"locations": [
{
"symbolic": {
"key": {
"Local": {
"given_path": f"{temp_dir}/.github/workflows/release.yml"
}
},
"annotation": "Injection risk",
},
"concrete": {
"location": {
"start_point": {"row": 5, "column": 1},
"end_point": {"row": 5, "column": 20},
}
},
}
],
}
]
mock_repo = MagicMock()
mock_repo.id = 1
mock_repo.name = "repo"
mock_repo.full_name = "owner/repo"
mock_repo.owner = "owner"
mock_repo.default_branch = MagicMock()
mock_repo.default_branch.name = "main"
mock_repo_client = MagicMock()
mock_repo_client.repositories = {1: mock_repo}
mock_provider = MagicMock()
mock_provider.session.token = "test-token"
mock_provider.exclude_workflows = []
service = GithubActions.__new__(GithubActions)
service.findings = {}
mock_repo_module = MagicMock()
mock_repo_module.repository_client = mock_repo_client
with (
patch.object(service, "_clone_repository", return_value=temp_dir),
patch.object(service, "_run_zizmor", return_value=zizmor_output),
patch.dict(
sys.modules,
{
"prowler.providers.github.services.repository.repository_client": mock_repo_module,
},
),
patch("shutil.rmtree"),
):
service._scan_repositories(mock_provider)
assert 1 in service.findings
assert len(service.findings[1]) == 1
finding = service.findings[1][0]
assert finding.workflow_file == ".github/workflows/release.yml"
assert (
finding.workflow_url
== "https://github.com/owner/repo/blob/main/.github/workflows/release.yml"
)
def test_init_github_actions_disabled(self):
mock_provider = MagicMock()
mock_provider.github_actions_enabled = False
mock_provider.session = MagicMock()
mock_provider.session.token = "test-token"
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
with patch.object(GithubActions, "__init__", lambda self, provider: None):
service = GithubActions.__new__(GithubActions)
service.findings = {}
# Service created, no scanning happened
assert service.findings == {}
@@ -0,0 +1,375 @@
from datetime import datetime, timezone
from unittest import mock
from prowler.providers.github.services.githubactions.githubactions_service import (
GithubActionsWorkflowFinding,
)
from prowler.providers.github.services.repository.repository_service import Branch, Repo
from tests.providers.github.github_fixtures import set_mocked_github_provider
def _make_repo(repo_id=1, name="repo1", owner="account-name"):
return Repo(
id=repo_id,
name=name,
owner=owner,
full_name=f"{owner}/{name}",
default_branch=Branch(
name="main",
protected=False,
default_branch=True,
require_pull_request=False,
approval_count=0,
required_linear_history=False,
allow_force_pushes=True,
branch_deletion=True,
status_checks=False,
enforce_admins=False,
require_code_owner_reviews=False,
require_signed_commits=False,
conversation_resolution=False,
),
private=False,
securitymd=True,
codeowners_exists=False,
secret_scanning_enabled=True,
archived=False,
pushed_at=datetime.now(timezone.utc),
delete_branch_on_merge=False,
)
def _make_finding(
repo_id=1,
repo_name="repo1",
repo_owner="account-name",
workflow_file=".github/workflows/ci.yml",
):
return GithubActionsWorkflowFinding(
repo_id=repo_id,
repo_name=repo_name,
repo_full_name=f"{repo_owner}/{repo_name}",
repo_owner=repo_owner,
workflow_file=workflow_file,
workflow_url=f"https://github.com/{repo_owner}/{repo_name}/blob/main/{workflow_file}",
line_range="line 10",
finding_id="githubactions_template_injection",
ident="template-injection",
description="Template Injection Vulnerability",
severity="high",
confidence="High",
annotation="High risk of code execution",
url="https://docs.zizmor.sh/",
)
class Test_githubactions_workflow_security_scan:
def test_scan_disabled(self):
repo = _make_repo()
repository_client = mock.MagicMock()
repository_client.repositories = {1: repo}
githubactions_client = mock.MagicMock()
githubactions_client.scan_enabled = False
githubactions_client.findings = {1: [_make_finding()]}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_github_provider(),
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.repository_client",
new=repository_client,
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.githubactions_client",
new=githubactions_client,
),
):
from prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan import (
githubactions_workflow_security_scan,
)
check = githubactions_workflow_security_scan()
result = check.execute()
assert len(result) == 0
def test_no_repositories(self):
repository_client = mock.MagicMock()
repository_client.repositories = {}
githubactions_client = mock.MagicMock()
githubactions_client.findings = {}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_github_provider(),
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.repository_client",
new=repository_client,
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.githubactions_client",
new=githubactions_client,
),
):
from prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan import (
githubactions_workflow_security_scan,
)
check = githubactions_workflow_security_scan()
result = check.execute()
assert len(result) == 0
def test_repository_no_findings_pass(self):
repo = _make_repo()
repository_client = mock.MagicMock()
repository_client.repositories = {1: repo}
githubactions_client = mock.MagicMock()
githubactions_client.findings = {}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_github_provider(),
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.repository_client",
new=repository_client,
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.githubactions_client",
new=githubactions_client,
),
):
from prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan import (
githubactions_workflow_security_scan,
)
check = githubactions_workflow_security_scan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].resource_name == "repo1"
assert (
result[0].check_metadata.CheckID
== "githubactions_workflow_security_scan"
)
assert (
"no GitHub Actions workflow security issues"
in result[0].status_extended
)
def test_repository_with_findings_fail(self):
repo = _make_repo()
finding = _make_finding()
repository_client = mock.MagicMock()
repository_client.repositories = {1: repo}
githubactions_client = mock.MagicMock()
githubactions_client.findings = {1: [finding]}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_github_provider(),
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.repository_client",
new=repository_client,
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.githubactions_client",
new=githubactions_client,
),
):
from prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan import (
githubactions_workflow_security_scan,
)
check = githubactions_workflow_security_scan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].resource_name == ".github/workflows/ci.yml"
assert "Template Injection Vulnerability" in result[0].status_extended
assert "line 10" in result[0].status_extended
assert "High" in result[0].status_extended
assert (
"https://github.com/account-name/repo1/blob/main/.github/workflows/ci.yml"
in result[0].status_extended
)
assert (
result[0].check_metadata.CheckID == "githubactions_template_injection"
)
assert (
result[0].check_metadata.CheckTitle
== "GitHub Actions workflows free of template-injection issues"
)
assert result[0].check_metadata.Severity == "high"
assert result[0].check_metadata.Risk == "Template Injection Vulnerability"
assert "https://docs.zizmor.sh/" in result[0].check_metadata.AdditionalURLs
def test_repository_with_multiple_findings(self):
repo = _make_repo()
finding1 = _make_finding(workflow_file=".github/workflows/ci.yml")
finding2 = _make_finding(workflow_file=".github/workflows/deploy.yml")
repository_client = mock.MagicMock()
repository_client.repositories = {1: repo}
githubactions_client = mock.MagicMock()
githubactions_client.findings = {1: [finding1, finding2]}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_github_provider(),
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.repository_client",
new=repository_client,
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.githubactions_client",
new=githubactions_client,
),
):
from prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan import (
githubactions_workflow_security_scan,
)
check = githubactions_workflow_security_scan()
result = check.execute()
assert len(result) == 2
assert all(r.status == "FAIL" for r in result)
workflow_files = [r.resource_name for r in result]
assert ".github/workflows/ci.yml" in workflow_files
assert ".github/workflows/deploy.yml" in workflow_files
def test_findings_have_independent_metadata(self):
repo = _make_repo()
finding1 = GithubActionsWorkflowFinding(
repo_id=1,
repo_name="repo1",
repo_full_name="account-name/repo1",
repo_owner="account-name",
workflow_file=".github/workflows/ci.yml",
workflow_url="https://github.com/account-name/repo1/blob/main/.github/workflows/ci.yml",
line_range="line 10",
finding_id="githubactions_template_injection",
ident="template-injection",
description="Template Injection",
severity="high",
confidence="High",
annotation="Attacker-controllable code",
url="https://docs.zizmor.sh/audits/#template-injection",
)
finding2 = GithubActionsWorkflowFinding(
repo_id=1,
repo_name="repo1",
repo_full_name="account-name/repo1",
repo_owner="account-name",
workflow_file=".github/workflows/deploy.yml",
workflow_url="https://github.com/account-name/repo1/blob/main/.github/workflows/deploy.yml",
line_range="line 5",
finding_id="githubactions_excessive_permissions",
ident="excessive-permissions",
description="Excessive Permissions",
severity="medium",
confidence="Medium",
annotation="Workflow has overly broad permissions",
url="https://docs.zizmor.sh/audits/#excessive-permissions",
)
repository_client = mock.MagicMock()
repository_client.repositories = {1: repo}
githubactions_client = mock.MagicMock()
githubactions_client.findings = {1: [finding1, finding2]}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_github_provider(),
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.repository_client",
new=repository_client,
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.githubactions_client",
new=githubactions_client,
),
):
from prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan import (
githubactions_workflow_security_scan,
)
check = githubactions_workflow_security_scan()
result = check.execute()
assert len(result) == 2
r1 = next(
r for r in result if r.resource_name == ".github/workflows/ci.yml"
)
r2 = next(
r for r in result if r.resource_name == ".github/workflows/deploy.yml"
)
assert r1.check_metadata.CheckID == "githubactions_template_injection"
assert r1.check_metadata.Severity == "high"
assert r1.check_metadata.Risk == "Template Injection"
assert (
"https://docs.zizmor.sh/audits/#template-injection"
in r1.check_metadata.AdditionalURLs
)
assert r2.check_metadata.CheckID == "githubactions_excessive_permissions"
assert r2.check_metadata.Severity == "medium"
assert r2.check_metadata.Risk == "Excessive Permissions"
assert (
"https://docs.zizmor.sh/audits/#excessive-permissions"
in r2.check_metadata.AdditionalURLs
)
def test_multiple_repos_mixed(self):
repo1 = _make_repo(repo_id=1, name="repo1")
repo2 = _make_repo(repo_id=2, name="repo2")
finding = _make_finding(repo_id=1)
repository_client = mock.MagicMock()
repository_client.repositories = {1: repo1, 2: repo2}
githubactions_client = mock.MagicMock()
githubactions_client.findings = {1: [finding]}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_github_provider(),
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.repository_client",
new=repository_client,
),
mock.patch(
"prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan.githubactions_client",
new=githubactions_client,
),
):
from prowler.providers.github.services.githubactions.githubactions_workflow_security_scan.githubactions_workflow_security_scan import (
githubactions_workflow_security_scan,
)
check = githubactions_workflow_security_scan()
result = check.execute()
assert len(result) == 2
statuses = {r.resource_name: r.status for r in result}
assert statuses[".github/workflows/ci.yml"] == "FAIL"
assert statuses["repo2"] == "PASS"
@@ -2,12 +2,16 @@
from unittest.mock import MagicMock
from prowler.providers.googleworkspace.models import GoogleWorkspaceIdentityInfo
from prowler.providers.googleworkspace.models import (
GoogleWorkspaceIdentityInfo,
GoogleWorkspaceResource,
)
# Google Workspace test constants
DOMAIN = "test-company.com"
CUSTOMER_ID = "C1234567"
DELEGATED_USER = "prowler-reader@test-company.com"
ROOT_ORG_UNIT_ID = "03ph8a2z1234"
# Service Account credentials (mock)
SERVICE_ACCOUNT_CREDENTIALS = {
@@ -81,10 +85,22 @@ def set_mocked_googleworkspace_provider(
domain=DOMAIN,
customer_id=CUSTOMER_ID,
delegated_user=DELEGATED_USER,
root_org_unit_id=ROOT_ORG_UNIT_ID,
profile="default",
),
):
provider = MagicMock()
provider.type = "googleworkspace"
provider.identity = identity
provider.domain_resource = build_googleworkspace_domain_resource()
return provider
def build_googleworkspace_domain_resource() -> GoogleWorkspaceResource:
"""Build the domain-level Google Workspace resource for tests."""
return GoogleWorkspaceResource(
id=CUSTOMER_ID,
name=DOMAIN,
customer_id=CUSTOMER_ID,
)
@@ -24,6 +24,7 @@ from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DELEGATED_USER,
DOMAIN,
ROOT_ORG_UNIT_ID,
SERVICE_ACCOUNT_CREDENTIALS,
)
@@ -68,6 +69,8 @@ class TestGoogleWorkspaceProvider:
delegated_user=DELEGATED_USER,
profile="default",
)
assert provider.domain_resource.id == CUSTOMER_ID
assert provider.domain_resource.name == DOMAIN
assert provider._audit_config == {}
def test_googleworkspace_provider_with_credentials_content(self):
@@ -107,6 +110,7 @@ class TestGoogleWorkspaceProvider:
assert provider.identity.domain == DOMAIN
assert provider.identity.customer_id == CUSTOMER_ID
assert provider.identity.delegated_user == DELEGATED_USER
assert provider.domain_resource.customer_id == CUSTOMER_ID
def test_googleworkspace_provider_missing_delegated_user(self):
"""Test that missing delegated_user raises exception"""
@@ -344,6 +348,62 @@ class TestGoogleWorkspaceProvider:
)
assert "is not configured in this Google Workspace" in str(exc_info.value)
def test_setup_identity_fetches_root_org_unit(self):
"""Test that setup_identity fetches and stores the root org unit ID"""
mock_session = GoogleWorkspaceSession(credentials=MagicMock(spec=Credentials))
with patch(
"prowler.providers.googleworkspace.googleworkspace_provider.build"
) as mock_build:
mock_service = MagicMock()
mock_build.return_value = mock_service
mock_service.customers().get().execute.return_value = {"id": CUSTOMER_ID}
mock_service.domains().list().execute.return_value = {
"domains": [{"domainName": DOMAIN}]
}
mock_service.orgunits().list().execute.return_value = {
"organizationUnits": [
{
"orgUnitPath": "/",
"orgUnitId": f"id:{ROOT_ORG_UNIT_ID}",
"name": "Test Company",
}
]
}
identity = GoogleworkspaceProvider.setup_identity(
session=mock_session,
delegated_user=DELEGATED_USER,
)
assert identity.root_org_unit_id == ROOT_ORG_UNIT_ID
assert identity.customer_id == CUSTOMER_ID
def test_setup_identity_root_org_unit_fetch_failure(self):
"""Test that setup_identity gracefully handles root org unit fetch failure"""
mock_session = GoogleWorkspaceSession(credentials=MagicMock(spec=Credentials))
with patch(
"prowler.providers.googleworkspace.googleworkspace_provider.build"
) as mock_build:
mock_service = MagicMock()
mock_build.return_value = mock_service
mock_service.customers().get().execute.return_value = {"id": CUSTOMER_ID}
mock_service.domains().list().execute.return_value = {
"domains": [{"domainName": DOMAIN}]
}
mock_service.orgunits().list().execute.side_effect = Exception(
"Insufficient permissions"
)
identity = GoogleworkspaceProvider.setup_identity(
session=mock_session,
delegated_user=DELEGATED_USER,
)
assert identity.root_org_unit_id is None
assert identity.customer_id == CUSTOMER_ID
def test_test_connection_raises_exception_when_flag_true(self):
"""Test that test_connection raises exception when raise_on_exception=True"""
credentials_file = "/path/to/credentials.json"
@@ -1,46 +1,82 @@
from unittest.mock import MagicMock
from prowler.providers.googleworkspace.lib.service.service import GoogleWorkspaceService
ROOT_OU_ID = "03ph8a2z1234"
def _make_service(root_org_unit_id=ROOT_OU_ID):
"""Create a GoogleWorkspaceService with a mocked provider."""
provider = MagicMock()
provider.identity.root_org_unit_id = root_org_unit_id
provider.audit_config = {}
provider.fixer_config = {}
provider.session.credentials = MagicMock()
svc = object.__new__(GoogleWorkspaceService)
svc.provider = provider
return svc
class TestIsCustomerLevelPolicy:
def test_no_policy_query(self):
"""Policy without policyQuery is customer-level"""
assert GoogleWorkspaceService._is_customer_level_policy({}) is True
svc = _make_service()
assert svc._is_customer_level_policy({}) is True
def test_empty_policy_query(self):
"""Policy with empty policyQuery is customer-level"""
svc = _make_service()
assert svc._is_customer_level_policy({"policyQuery": {}}) is True
def test_root_org_unit_accepted(self):
"""Policy targeting the root OU is customer-level"""
svc = _make_service()
assert (
GoogleWorkspaceService._is_customer_level_policy({"policyQuery": {}})
svc._is_customer_level_policy(
{"policyQuery": {"orgUnit": f"orgUnits/{ROOT_OU_ID}"}}
)
is True
)
def test_org_unit_targeted(self):
"""Policy targeting a specific OU is not customer-level"""
def test_sub_org_unit_rejected(self):
"""Policy targeting a sub-OU is not customer-level"""
svc = _make_service()
assert (
GoogleWorkspaceService._is_customer_level_policy(
{"policyQuery": {"orgUnit": "orgUnits/abc123"}}
svc._is_customer_level_policy(
{"policyQuery": {"orgUnit": "orgUnits/sub_ou_abc123"}}
)
is False
)
def test_group_targeted(self):
"""Policy targeting a specific group is not customer-level"""
svc = _make_service()
assert (
GoogleWorkspaceService._is_customer_level_policy(
{"policyQuery": {"group": "groups/xyz789"}}
)
svc._is_customer_level_policy({"policyQuery": {"group": "groups/xyz789"}})
is False
)
def test_org_unit_and_group_targeted(self):
"""Policy targeting both OU and group is not customer-level"""
svc = _make_service()
assert (
GoogleWorkspaceService._is_customer_level_policy(
svc._is_customer_level_policy(
{
"policyQuery": {
"orgUnit": "orgUnits/abc123",
"orgUnit": f"orgUnits/{ROOT_OU_ID}",
"group": "groups/xyz789",
}
}
)
is False
)
def test_no_root_org_unit_id_rejects_all_ou(self):
"""When root OU ID is unknown, all OU-targeted policies are rejected"""
svc = _make_service(root_org_unit_id=None)
assert (
svc._is_customer_level_policy(
{"policyQuery": {"orgUnit": f"orgUnits/{ROOT_OU_ID}"}}
)
is False
)
@@ -41,7 +41,9 @@ class TestCalendarExternalSharingPrimaryCalendar:
assert findings[0].status == "PASS"
assert "free/busy information only" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].resource_id == CUSTOMER_ID
assert findings[0].customer_id == CUSTOMER_ID
assert findings[0].resource == mock_provider.domain_resource.dict()
def test_fail_read_only(self):
"""Test FAIL when external sharing allows read-only access"""
@@ -1,6 +1,8 @@
from unittest.mock import MagicMock, patch
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
GROUPS_ADMIN_ROLE_ID,
ROLE_GROUPS_ADMIN,
ROLE_SEED_ADMIN,
@@ -66,6 +68,7 @@ class TestDirectoryService:
directory = Directory(mock_provider)
assert len(directory.users) == 3
assert directory.domain_resource == mock_provider.domain_resource
assert "user1-id" in directory.users
assert "user2-id" in directory.users
assert "user3-id" in directory.users
@@ -119,6 +122,8 @@ class TestDirectoryService:
directory = Directory(mock_provider)
assert len(directory.users) == 0
assert directory.domain_resource.id == CUSTOMER_ID
assert directory.domain_resource.name == DOMAIN
def test_directory_api_error_handling(self):
"""Test handling of API errors"""
@@ -55,7 +55,9 @@ class TestDirectorySuperAdminCount:
assert "2 super administrator(s)" in findings[0].status_extended
assert "within the recommended range" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].resource_id == CUSTOMER_ID
assert findings[0].customer_id == CUSTOMER_ID
assert findings[0].resource == mock_provider.domain_resource.dict()
def test_directory_super_admin_count_pass_4_admins(self):
"""Test PASS when there are 4 super admins (within range)"""
@@ -91,7 +91,9 @@ class TestDirectorySuperAdminOnlyAdminRoles:
assert findings[0].status == "PASS"
assert "used only for super admin activities" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].resource_id == CUSTOMER_ID
assert findings[0].customer_id == CUSTOMER_ID
assert findings[0].resource == mock_provider.domain_resource.dict()
def test_pass_super_admin_with_seed_admin_role(self):
"""Test PASS when a super admin only holds _SEED_ADMIN_ROLE.
@@ -213,7 +215,8 @@ class TestDirectorySuperAdminOnlyAdminRoles:
assert "Groups Administrator" in findings[0].status_extended
assert "_GROUPS_ADMIN_ROLE" not in findings[0].status_extended
assert "used only for super admin activities" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].resource_name == "admin1@test-company.com"
assert findings[0].resource_id == "admin1-id"
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_seed_admin_with_additional_roles(self):
@@ -254,6 +257,8 @@ class TestDirectorySuperAdminOnlyAdminRoles:
assert "Groups Administrator" in findings[0].status_extended
assert "_GROUPS_ADMIN_ROLE" not in findings[0].status_extended
assert "_SEED_ADMIN_ROLE" not in findings[0].status_extended
assert findings[0].resource_name == "playground@prowler.cloud"
assert findings[0].resource_id == "admin1-id"
def test_fail_multiple_super_admins_with_extra_roles(self):
"""Test FAIL lists all super admins that have additional roles"""
@@ -299,11 +304,12 @@ class TestDirectorySuperAdminOnlyAdminRoles:
check = directory_super_admin_only_admin_roles()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "admin1@test-company.com" in findings[0].status_extended
assert "admin2@test-company.com" in findings[0].status_extended
assert len(findings) == 2
assert all(finding.status == "FAIL" for finding in findings)
assert findings[0].resource_name == "admin1@test-company.com"
assert findings[1].resource_name == "admin2@test-company.com"
assert "admin3@test-company.com" not in findings[0].status_extended
assert "admin3@test-company.com" not in findings[1].status_extended
def test_no_findings_when_no_users(self):
"""Test no findings when there are no users"""
@@ -444,3 +450,4 @@ class TestDirectorySuperAdminOnlyAdminRoles:
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "custom-helpdesk-role" in findings[0].status_extended
assert findings[0].resource_name == "admin1@test-company.com"
@@ -37,7 +37,9 @@ class TestDriveExternalSharingWarnUsers:
assert findings[0].status == "PASS"
assert "enabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].resource_id == CUSTOMER_ID
assert findings[0].customer_id == CUSTOMER_ID
assert findings[0].resource == mock_provider.domain_resource.dict()
def test_fail_warning_disabled(self):
"""Test FAIL when external sharing warning is explicitly disabled"""
@@ -0,0 +1,120 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailAutoForwardingDisabled:
def test_pass(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_auto_forwarding_disabled.gmail_auto_forwarding_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_auto_forwarding_disabled.gmail_auto_forwarding_disabled import (
gmail_auto_forwarding_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_auto_forwarding=False)
check = gmail_auto_forwarding_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "disabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].resource_id == CUSTOMER_ID
assert findings[0].customer_id == CUSTOMER_ID
assert findings[0].resource == mock_provider.domain_resource.dict()
def test_fail_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_auto_forwarding_disabled.gmail_auto_forwarding_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_auto_forwarding_disabled.gmail_auto_forwarding_disabled import (
gmail_auto_forwarding_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_auto_forwarding=True)
check = gmail_auto_forwarding_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "enabled" in findings[0].status_extended
def test_fail_no_policy_set(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_auto_forwarding_disabled.gmail_auto_forwarding_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_auto_forwarding_disabled.gmail_auto_forwarding_disabled import (
gmail_auto_forwarding_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_auto_forwarding=None)
check = gmail_auto_forwarding_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_auto_forwarding_disabled.gmail_auto_forwarding_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_auto_forwarding_disabled.gmail_auto_forwarding_disabled import (
gmail_auto_forwarding_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_auto_forwarding_disabled()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,124 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailComprehensiveMailStorageEnabled:
def test_pass(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_comprehensive_mail_storage_enabled.gmail_comprehensive_mail_storage_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_comprehensive_mail_storage_enabled.gmail_comprehensive_mail_storage_enabled import (
gmail_comprehensive_mail_storage_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
comprehensive_mail_storage_enabled=True
)
check = gmail_comprehensive_mail_storage_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "enabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_comprehensive_mail_storage_enabled.gmail_comprehensive_mail_storage_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_comprehensive_mail_storage_enabled.gmail_comprehensive_mail_storage_enabled import (
gmail_comprehensive_mail_storage_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
comprehensive_mail_storage_enabled=False
)
check = gmail_comprehensive_mail_storage_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "disabled" in findings[0].status_extended
def test_fail_no_policy_set(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_comprehensive_mail_storage_enabled.gmail_comprehensive_mail_storage_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_comprehensive_mail_storage_enabled.gmail_comprehensive_mail_storage_enabled import (
gmail_comprehensive_mail_storage_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
comprehensive_mail_storage_enabled=None
)
check = gmail_comprehensive_mail_storage_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_comprehensive_mail_storage_enabled.gmail_comprehensive_mail_storage_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_comprehensive_mail_storage_enabled.gmail_comprehensive_mail_storage_enabled import (
gmail_comprehensive_mail_storage_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_comprehensive_mail_storage_enabled()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,124 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailEnhancedPreDeliveryScanningEnabled:
def test_pass(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_enhanced_pre_delivery_scanning_enabled.gmail_enhanced_pre_delivery_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_enhanced_pre_delivery_scanning_enabled.gmail_enhanced_pre_delivery_scanning_enabled import (
gmail_enhanced_pre_delivery_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_enhanced_pre_delivery_scanning=True
)
check = gmail_enhanced_pre_delivery_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "enabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_enhanced_pre_delivery_scanning_enabled.gmail_enhanced_pre_delivery_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_enhanced_pre_delivery_scanning_enabled.gmail_enhanced_pre_delivery_scanning_enabled import (
gmail_enhanced_pre_delivery_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_enhanced_pre_delivery_scanning=False
)
check = gmail_enhanced_pre_delivery_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "disabled" in findings[0].status_extended
def test_pass_using_default(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_enhanced_pre_delivery_scanning_enabled.gmail_enhanced_pre_delivery_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_enhanced_pre_delivery_scanning_enabled.gmail_enhanced_pre_delivery_scanning_enabled import (
gmail_enhanced_pre_delivery_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_enhanced_pre_delivery_scanning=None
)
check = gmail_enhanced_pre_delivery_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_enhanced_pre_delivery_scanning_enabled.gmail_enhanced_pre_delivery_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_enhanced_pre_delivery_scanning_enabled.gmail_enhanced_pre_delivery_scanning_enabled import (
gmail_enhanced_pre_delivery_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_enhanced_pre_delivery_scanning_enabled()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,118 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailExternalImageScanningEnabled:
def test_pass(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_external_image_scanning_enabled.gmail_external_image_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_external_image_scanning_enabled.gmail_external_image_scanning_enabled import (
gmail_external_image_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_external_image_scanning=True)
check = gmail_external_image_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "enabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_external_image_scanning_enabled.gmail_external_image_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_external_image_scanning_enabled.gmail_external_image_scanning_enabled import (
gmail_external_image_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_external_image_scanning=False)
check = gmail_external_image_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "disabled" in findings[0].status_extended
def test_pass_using_default(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_external_image_scanning_enabled.gmail_external_image_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_external_image_scanning_enabled.gmail_external_image_scanning_enabled import (
gmail_external_image_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_external_image_scanning=None)
check = gmail_external_image_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_external_image_scanning_enabled.gmail_external_image_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_external_image_scanning_enabled.gmail_external_image_scanning_enabled import (
gmail_external_image_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_external_image_scanning_enabled()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,118 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailMailDelegationDisabled:
def test_pass_delegation_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_mail_delegation_disabled.gmail_mail_delegation_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_mail_delegation_disabled.gmail_mail_delegation_disabled import (
gmail_mail_delegation_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_mail_delegation=False)
check = gmail_mail_delegation_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "disabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_delegation_enabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_mail_delegation_disabled.gmail_mail_delegation_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_mail_delegation_disabled.gmail_mail_delegation_disabled import (
gmail_mail_delegation_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_mail_delegation=True)
check = gmail_mail_delegation_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "enabled" in findings[0].status_extended
def test_pass_using_default(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_mail_delegation_disabled.gmail_mail_delegation_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_mail_delegation_disabled.gmail_mail_delegation_disabled import (
gmail_mail_delegation_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_mail_delegation=None)
check = gmail_mail_delegation_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_mail_delegation_disabled.gmail_mail_delegation_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_mail_delegation_disabled.gmail_mail_delegation_disabled import (
gmail_mail_delegation_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_mail_delegation_disabled()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,118 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailPerUserOutboundGatewayDisabled:
def test_pass(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_per_user_outbound_gateway_disabled.gmail_per_user_outbound_gateway_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_per_user_outbound_gateway_disabled.gmail_per_user_outbound_gateway_disabled import (
gmail_per_user_outbound_gateway_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(allow_per_user_outbound_gateway=False)
check = gmail_per_user_outbound_gateway_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "disabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_per_user_outbound_gateway_disabled.gmail_per_user_outbound_gateway_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_per_user_outbound_gateway_disabled.gmail_per_user_outbound_gateway_disabled import (
gmail_per_user_outbound_gateway_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(allow_per_user_outbound_gateway=True)
check = gmail_per_user_outbound_gateway_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "enabled" in findings[0].status_extended
def test_pass_using_default(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_per_user_outbound_gateway_disabled.gmail_per_user_outbound_gateway_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_per_user_outbound_gateway_disabled.gmail_per_user_outbound_gateway_disabled import (
gmail_per_user_outbound_gateway_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(allow_per_user_outbound_gateway=None)
check = gmail_per_user_outbound_gateway_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_per_user_outbound_gateway_disabled.gmail_per_user_outbound_gateway_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_per_user_outbound_gateway_disabled.gmail_per_user_outbound_gateway_disabled import (
gmail_per_user_outbound_gateway_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_per_user_outbound_gateway_disabled()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,183 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailPopImapAccessDisabled:
def test_pass_both_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled import (
gmail_pop_imap_access_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_pop_access=False, enable_imap_access=False
)
check = gmail_pop_imap_access_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "disabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_both_enabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled import (
gmail_pop_imap_access_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_pop_access=True, enable_imap_access=True
)
check = gmail_pop_imap_access_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "POP" in findings[0].status_extended
assert "IMAP" in findings[0].status_extended
def test_fail_pop_enabled_only(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled import (
gmail_pop_imap_access_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_pop_access=True, enable_imap_access=False
)
check = gmail_pop_imap_access_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "POP" in findings[0].status_extended
def test_fail_imap_enabled_only(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled import (
gmail_pop_imap_access_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_pop_access=False, enable_imap_access=True
)
check = gmail_pop_imap_access_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "IMAP" in findings[0].status_extended
def test_fail_no_policy_set(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled import (
gmail_pop_imap_access_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_pop_access=None, enable_imap_access=None
)
check = gmail_pop_imap_access_disabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_pop_imap_access_disabled.gmail_pop_imap_access_disabled import (
gmail_pop_imap_access_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_pop_imap_access_disabled()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,493 @@
from unittest.mock import MagicMock, patch
from googleapiclient.errors import HttpError
from httplib2 import Response as HttpResponse
from tests.providers.googleworkspace.googleworkspace_fixtures import (
ROOT_ORG_UNIT_ID,
set_mocked_googleworkspace_provider,
)
class TestGmailService:
def test_gmail_fetch_policies_all_settings(self):
"""Test fetching all 10 Gmail policy settings from Cloud Identity API"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_credentials = MagicMock()
mock_session = MagicMock()
mock_session.credentials = mock_credentials
mock_provider.session = mock_session
mock_service = MagicMock()
mock_policies_list = MagicMock()
mock_policies_list.execute.return_value = {
"policies": [
{
"setting": {
"type": "settings/gmail.mail_delegation",
"value": {"enableMailDelegation": False},
}
},
{
"setting": {
"type": "settings/gmail.email_attachment_safety",
"value": {
"encryptedAttachmentProtectionConsequence": "SPAM_FOLDER",
"scriptAttachmentProtectionConsequence": "QUARANTINE",
"anomalousAttachmentProtectionConsequence": "WARNING",
},
}
},
{
"setting": {
"type": "settings/gmail.links_and_external_images",
"value": {
"enableShortenerScanning": True,
"enableExternalImageScanning": True,
"enableAggressiveWarningsOnUntrustedLinks": True,
},
}
},
{
"setting": {
"type": "settings/gmail.spoofing_and_authentication",
"value": {
"domainSpoofingConsequence": "SPAM_FOLDER",
"employeeNameSpoofingConsequence": "SPAM_FOLDER",
"inboundDomainSpoofingConsequence": "QUARANTINE",
"unauthenticatedEmailConsequence": "WARNING",
"groupsSpoofingConsequence": "SPAM_FOLDER",
},
}
},
{
"setting": {
"type": "settings/gmail.pop_access",
"value": {"enablePopAccess": False},
}
},
{
"setting": {
"type": "settings/gmail.imap_access",
"value": {"enableImapAccess": False},
}
},
{
"setting": {
"type": "settings/gmail.auto_forwarding",
"value": {"enableAutoForwarding": False},
}
},
{
"setting": {
"type": "settings/gmail.per_user_outbound_gateway",
"value": {"allowUsersToUseExternalSmtpServers": False},
}
},
{
"setting": {
"type": "settings/gmail.enhanced_pre_delivery_message_scanning",
"value": {"enableImprovedSuspiciousContentDetection": True},
}
},
{
"setting": {
"type": "settings/gmail.comprehensive_mail_storage",
"value": {"ruleId": "rule-abc-123"},
}
},
]
}
mock_service.policies().list.return_value = mock_policies_list
mock_service.policies().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
Gmail,
)
gmail = Gmail(mock_provider)
assert gmail.policies_fetched is True
assert gmail.policies.enable_mail_delegation is False
assert (
gmail.policies.encrypted_attachment_protection_consequence
== "SPAM_FOLDER"
)
assert (
gmail.policies.script_attachment_protection_consequence == "QUARANTINE"
)
assert (
gmail.policies.anomalous_attachment_protection_consequence == "WARNING"
)
assert gmail.policies.enable_shortener_scanning is True
assert gmail.policies.enable_external_image_scanning is True
assert gmail.policies.enable_aggressive_warnings_on_untrusted_links is True
assert gmail.policies.domain_spoofing_consequence == "SPAM_FOLDER"
assert gmail.policies.employee_name_spoofing_consequence == "SPAM_FOLDER"
assert gmail.policies.inbound_domain_spoofing_consequence == "QUARANTINE"
assert gmail.policies.unauthenticated_email_consequence == "WARNING"
assert gmail.policies.groups_spoofing_consequence == "SPAM_FOLDER"
assert gmail.policies.enable_pop_access is False
assert gmail.policies.enable_imap_access is False
assert gmail.policies.enable_auto_forwarding is False
assert gmail.policies.allow_per_user_outbound_gateway is False
assert gmail.policies.enable_enhanced_pre_delivery_scanning is True
assert gmail.policies.comprehensive_mail_storage_enabled is True
def test_gmail_fetch_policies_empty_response(self):
"""Test handling empty policies response"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
mock_policies_list = MagicMock()
mock_policies_list.execute.return_value = {"policies": []}
mock_service.policies().list.return_value = mock_policies_list
mock_service.policies().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
Gmail,
)
gmail = Gmail(mock_provider)
assert gmail.policies_fetched is True
assert gmail.policies.enable_mail_delegation is None
assert gmail.policies.encrypted_attachment_protection_consequence is None
assert gmail.policies.enable_pop_access is None
assert gmail.policies.comprehensive_mail_storage_enabled is None
def test_gmail_fetch_policies_api_error(self):
"""Test handling of API errors during policy fetch"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
mock_service.policies().list.side_effect = Exception("API Error")
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
Gmail,
)
gmail = Gmail(mock_provider)
assert gmail.policies_fetched is False
assert gmail.policies.enable_mail_delegation is None
def test_gmail_fetch_policies_build_service_returns_none(self):
"""Test early return when _build_service fails to construct the client"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_service.GoogleWorkspaceService._build_service",
return_value=None,
),
):
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
Gmail,
)
gmail = Gmail(mock_provider)
assert gmail.policies_fetched is False
assert gmail.policies.enable_mail_delegation is None
def test_gmail_fetch_policies_execute_raises(self):
"""Test inner except handler when request.execute() raises during pagination"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
mock_request = MagicMock()
mock_request.execute.side_effect = Exception("Execute failed")
mock_service.policies().list.return_value = mock_request
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
Gmail,
)
gmail = Gmail(mock_provider)
assert gmail.policies_fetched is False
assert gmail.policies.enable_mail_delegation is None
def test_gmail_fetch_policies_ignores_ou_and_group_level(self):
"""Test that OU-level and group-level policies are skipped, only customer-level used"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
mock_policies_list = MagicMock()
mock_policies_list.execute.return_value = {
"policies": [
{
# Customer-level: no policyQuery → should be used
"setting": {
"type": "settings/gmail.mail_delegation",
"value": {"enableMailDelegation": False},
}
},
{
# OU-level: has policyQuery.orgUnit → should be skipped
"policyQuery": {"orgUnit": "orgUnits/sales_team"},
"setting": {
"type": "settings/gmail.mail_delegation",
"value": {"enableMailDelegation": True},
},
},
{
# Group-level: has policyQuery.group → should be skipped
"policyQuery": {"group": "groups/contractors"},
"setting": {
"type": "settings/gmail.auto_forwarding",
"value": {"enableAutoForwarding": True},
},
},
{
# Customer-level: no policyQuery → should be used
"setting": {
"type": "settings/gmail.auto_forwarding",
"value": {"enableAutoForwarding": False},
}
},
]
}
mock_service.policies().list.return_value = mock_policies_list
mock_service.policies().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
Gmail,
)
gmail = Gmail(mock_provider)
assert gmail.policies_fetched is True
assert gmail.policies.enable_mail_delegation is False
assert gmail.policies.enable_auto_forwarding is False
def test_gmail_fetch_policies_accepts_root_ou(self):
"""Test that root-OU-scoped policies are accepted as customer-level"""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
mock_policies_list = MagicMock()
mock_policies_list.execute.return_value = {
"policies": [
{
# Root OU: matches provider's root_org_unit_id → should be accepted
"policyQuery": {"orgUnit": f"orgUnits/{ROOT_ORG_UNIT_ID}"},
"setting": {
"type": "settings/gmail.mail_delegation",
"value": {"enableMailDelegation": True},
},
},
{
# Sub-OU: different orgUnit → should be skipped
"policyQuery": {"orgUnit": "orgUnits/sub_ou_sales"},
"setting": {
"type": "settings/gmail.auto_forwarding",
"value": {"enableAutoForwarding": True},
},
},
]
}
mock_service.policies().list.return_value = mock_policies_list
mock_service.policies().list_next.return_value = None
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
Gmail,
)
gmail = Gmail(mock_provider)
assert gmail.policies_fetched is True
# Root OU policy accepted
assert gmail.policies.enable_mail_delegation is True
# Sub-OU policy skipped
assert gmail.policies.enable_auto_forwarding is None
def test_gmail_partial_fetch_marks_policies_fetched_false(self):
"""Regression: if page 1 returns valid data but page 2 raises an error,
policies_fetched must be False even though some policy values were stored."""
mock_provider = set_mocked_googleworkspace_provider()
mock_provider.audit_config = {}
mock_provider.fixer_config = {}
mock_session = MagicMock()
mock_session.credentials = MagicMock()
mock_provider.session = mock_session
mock_service = MagicMock()
# Page 1: returns valid Gmail data
page1_response = {
"policies": [
{
"setting": {
"type": "settings/gmail.mail_delegation",
"value": {"enableMailDelegation": False},
}
},
]
}
# Page 2 request raises HttpError 429
page1_request = MagicMock()
page1_request.execute.return_value = page1_response
page2_request = MagicMock()
page2_request.execute.side_effect = HttpError(
HttpResponse({"status": "429"}), b"Rate limit exceeded"
)
mock_service.policies().list.return_value = page1_request
mock_service.policies().list_next.return_value = page2_request
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
Gmail,
)
gmail = Gmail(mock_provider)
# Page 1 data was stored
assert gmail.policies.enable_mail_delegation is False
# But policies_fetched must be False because page 2 failed
assert gmail.policies_fetched is False
def test_gmail_policies_model(self):
"""Test GmailPolicies Pydantic model"""
from prowler.providers.googleworkspace.services.gmail.gmail_service import (
GmailPolicies,
)
policies = GmailPolicies(
enable_mail_delegation=False,
encrypted_attachment_protection_consequence="SPAM_FOLDER",
script_attachment_protection_consequence="QUARANTINE",
anomalous_attachment_protection_consequence="WARNING",
enable_shortener_scanning=True,
enable_external_image_scanning=True,
enable_aggressive_warnings_on_untrusted_links=True,
domain_spoofing_consequence="SPAM_FOLDER",
employee_name_spoofing_consequence="SPAM_FOLDER",
inbound_domain_spoofing_consequence="QUARANTINE",
unauthenticated_email_consequence="WARNING",
groups_spoofing_consequence="SPAM_FOLDER",
enable_pop_access=False,
enable_imap_access=False,
enable_auto_forwarding=False,
allow_per_user_outbound_gateway=False,
enable_enhanced_pre_delivery_scanning=True,
comprehensive_mail_storage_enabled=True,
)
assert policies.enable_mail_delegation is False
assert policies.encrypted_attachment_protection_consequence == "SPAM_FOLDER"
assert policies.enable_shortener_scanning is True
assert policies.domain_spoofing_consequence == "SPAM_FOLDER"
assert policies.enable_pop_access is False
assert policies.enable_auto_forwarding is False
assert policies.enable_enhanced_pre_delivery_scanning is True
assert policies.comprehensive_mail_storage_enabled is True
@@ -0,0 +1,118 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailShortenerScanningEnabled:
def test_pass(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_shortener_scanning_enabled.gmail_shortener_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_shortener_scanning_enabled.gmail_shortener_scanning_enabled import (
gmail_shortener_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_shortener_scanning=True)
check = gmail_shortener_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "enabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_shortener_scanning_enabled.gmail_shortener_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_shortener_scanning_enabled.gmail_shortener_scanning_enabled import (
gmail_shortener_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_shortener_scanning=False)
check = gmail_shortener_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "disabled" in findings[0].status_extended
def test_pass_using_default(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_shortener_scanning_enabled.gmail_shortener_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_shortener_scanning_enabled.gmail_shortener_scanning_enabled import (
gmail_shortener_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(enable_shortener_scanning=None)
check = gmail_shortener_scanning_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_shortener_scanning_enabled.gmail_shortener_scanning_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_shortener_scanning_enabled.gmail_shortener_scanning_enabled import (
gmail_shortener_scanning_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_shortener_scanning_enabled()
findings = check.execute()
assert len(findings) == 0
@@ -0,0 +1,124 @@
from unittest.mock import patch
from prowler.providers.googleworkspace.services.gmail.gmail_service import GmailPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
class TestGmailUntrustedLinkWarningsEnabled:
def test_pass(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_untrusted_link_warnings_enabled.gmail_untrusted_link_warnings_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_untrusted_link_warnings_enabled.gmail_untrusted_link_warnings_enabled import (
gmail_untrusted_link_warnings_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_aggressive_warnings_on_untrusted_links=True
)
check = gmail_untrusted_link_warnings_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "enabled" in findings[0].status_extended
assert findings[0].resource_name == DOMAIN
assert findings[0].customer_id == CUSTOMER_ID
def test_fail_disabled(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_untrusted_link_warnings_enabled.gmail_untrusted_link_warnings_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_untrusted_link_warnings_enabled.gmail_untrusted_link_warnings_enabled import (
gmail_untrusted_link_warnings_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_aggressive_warnings_on_untrusted_links=False
)
check = gmail_untrusted_link_warnings_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "disabled" in findings[0].status_extended
def test_pass_using_default(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_untrusted_link_warnings_enabled.gmail_untrusted_link_warnings_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_untrusted_link_warnings_enabled.gmail_untrusted_link_warnings_enabled import (
gmail_untrusted_link_warnings_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
enable_aggressive_warnings_on_untrusted_links=None
)
check = gmail_untrusted_link_warnings_enabled()
findings = check.execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
with (
patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
),
patch(
"prowler.providers.googleworkspace.services.gmail.gmail_untrusted_link_warnings_enabled.gmail_untrusted_link_warnings_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_untrusted_link_warnings_enabled.gmail_untrusted_link_warnings_enabled import (
gmail_untrusted_link_warnings_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = False
mock_client.policies = GmailPolicies()
check = gmail_untrusted_link_warnings_enabled()
findings = check.execute()
assert len(findings) == 0
@@ -61,7 +61,12 @@ class Test_admincenter_groups_not_public_visibility:
id_group1 = str(uuid4())
admincenter_client.groups = {
id_group1: Group(id=id_group1, name="Group1", visibility="Private"),
id_group1: Group(
id=id_group1,
name="Group1",
visibility="Private",
group_types=["Unified"],
),
}
check = admincenter_groups_not_public_visibility()
@@ -102,7 +107,12 @@ class Test_admincenter_groups_not_public_visibility:
id_group1 = str(uuid4())
admincenter_client.groups = {
id_group1: Group(id=id_group1, name="Group1", visibility="Private"),
id_group1: Group(
id=id_group1,
name="Group1",
visibility="Private",
group_types=["Unified"],
),
}
check = admincenter_groups_not_public_visibility()
@@ -143,7 +153,12 @@ class Test_admincenter_groups_not_public_visibility:
id_group1 = str(uuid4())
admincenter_client.groups = {
id_group1: Group(id=id_group1, name="Group1", visibility="Public"),
id_group1: Group(
id=id_group1,
name="Group1",
visibility="Public",
group_types=["Unified"],
),
}
check = admincenter_groups_not_public_visibility()
@@ -187,7 +202,12 @@ class Test_admincenter_groups_not_public_visibility:
id_group1 = str(uuid4())
admincenter_client.groups = {
id_group1: Group(id=id_group1, name="Group1", visibility=None),
id_group1: Group(
id=id_group1,
name="Group1",
visibility=None,
group_types=["Unified"],
),
}
check = admincenter_groups_not_public_visibility()
@@ -202,3 +222,60 @@ class Test_admincenter_groups_not_public_visibility:
assert result[0].resource_name == "Group1"
assert result[0].resource_id == id_group1
assert result[0].location == "global"
def test_admincenter_security_group_ignored(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_m365_provider(),
),
mock.patch(
"prowler.providers.m365.lib.powershell.m365_powershell.M365PowerShell.connect_exchange_online"
),
mock.patch(
"prowler.providers.m365.services.admincenter.admincenter_groups_not_public_visibility.admincenter_groups_not_public_visibility.admincenter_client",
new=admincenter_client,
),
):
from prowler.providers.m365.services.admincenter.admincenter_groups_not_public_visibility.admincenter_groups_not_public_visibility import (
admincenter_groups_not_public_visibility,
)
from prowler.providers.m365.services.admincenter.admincenter_service import (
Group,
)
id_security_group = str(uuid4())
id_distribution_group = str(uuid4())
id_m365_group = str(uuid4())
admincenter_client.groups = {
id_security_group: Group(
id=id_security_group,
name="SecurityGroup",
visibility=None,
group_types=[],
),
id_distribution_group: Group(
id=id_distribution_group,
name="DistributionGroup",
visibility=None,
group_types=[],
),
id_m365_group: Group(
id=id_m365_group,
name="M365Group",
visibility="Private",
group_types=["Unified"],
),
}
check = admincenter_groups_not_public_visibility()
result = check.execute()
assert len(result) == 1
assert result[0].resource_id == id_m365_group
assert result[0].status == "PASS"
assert result[0].resource_name == "M365Group"
@@ -214,3 +214,41 @@ def test_admincenter__get_users_handles_pagination():
with_url_mock.assert_called_once_with("next-link")
assert users["user-1"].license == "SKU-user-1"
assert users["user-3"].license == "SKU-user-3"
def test_admincenter__get_groups_maps_group_types():
admincenter_service = AdminCenter.__new__(AdminCenter)
groups_response = SimpleNamespace(
value=[
SimpleNamespace(
id="id-1",
display_name="Unified Group",
visibility="Private",
group_types=["Unified"],
),
SimpleNamespace(
id="id-2",
display_name="Security Group",
visibility=None,
group_types=[],
),
SimpleNamespace(
id="id-3",
display_name="Legacy Group",
visibility="Public",
),
]
)
groups_builder = SimpleNamespace(get=AsyncMock(return_value=groups_response))
admincenter_service.client = SimpleNamespace(groups=groups_builder)
groups = asyncio.run(admincenter_service._get_groups())
assert len(groups) == 3
assert groups_builder.get.await_count == 1
assert groups["id-1"].group_types == ["Unified"]
assert groups["id-2"].group_types == []
assert groups["id-3"].group_types == []
assert groups["id-3"].visibility == "Public"
@@ -237,3 +237,186 @@ class Test_entra_users_mfa_capable:
assert result[0].resource == entra_client.users[enabled_user_id]
assert result[0].resource_name == "Enabled User"
assert result[0].resource_id == enabled_user_id
def test_disabled_guest_user_not_checked(self):
"""Disabled guest user should not be checked: expected no results.
Regression test for https://github.com/prowler-cloud/prowler/issues/10637.
CIS 5.2.3.4 evaluates only enabled member users; disabled guests must be skipped
even when ``account_enabled`` cannot be derived from Exchange Online.
"""
entra_client = mock.MagicMock
entra_client.audited_tenant = "audited_tenant"
entra_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.services.entra.entra_users_mfa_capable.entra_users_mfa_capable.entra_client",
new=entra_client,
),
):
from prowler.providers.m365.services.entra.entra_users_mfa_capable.entra_users_mfa_capable import (
entra_users_mfa_capable,
)
user_id = str(uuid4())
entra_client.users = {
user_id: User(
id=user_id,
name="Disabled Guest",
on_premises_sync_enabled=False,
directory_roles_ids=[],
is_mfa_capable=False,
account_enabled=False,
user_type="Guest",
)
}
check = entra_users_mfa_capable()
result = check.execute()
assert len(result) == 0
def test_enabled_guest_user_not_checked(self):
"""Enabled guest user is out of scope for CIS 5.2.3.4: expected no results."""
entra_client = mock.MagicMock
entra_client.audited_tenant = "audited_tenant"
entra_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.services.entra.entra_users_mfa_capable.entra_users_mfa_capable.entra_client",
new=entra_client,
),
):
from prowler.providers.m365.services.entra.entra_users_mfa_capable.entra_users_mfa_capable import (
entra_users_mfa_capable,
)
user_id = str(uuid4())
entra_client.users = {
user_id: User(
id=user_id,
name="Guest User",
on_premises_sync_enabled=False,
directory_roles_ids=[],
is_mfa_capable=False,
account_enabled=True,
user_type="Guest",
)
}
check = entra_users_mfa_capable()
result = check.execute()
assert len(result) == 0
def test_member_and_guest_users(self):
"""Mix of member and guest users: only member users should be checked."""
entra_client = mock.MagicMock
entra_client.audited_tenant = "audited_tenant"
entra_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.services.entra.entra_users_mfa_capable.entra_users_mfa_capable.entra_client",
new=entra_client,
),
):
from prowler.providers.m365.services.entra.entra_users_mfa_capable.entra_users_mfa_capable import (
entra_users_mfa_capable,
)
member_user_id = str(uuid4())
guest_user_id = str(uuid4())
entra_client.users = {
member_user_id: User(
id=member_user_id,
name="Member User",
on_premises_sync_enabled=False,
directory_roles_ids=[],
is_mfa_capable=False,
account_enabled=True,
user_type="Member",
),
guest_user_id: User(
id=guest_user_id,
name="Guest User",
on_premises_sync_enabled=False,
directory_roles_ids=[],
is_mfa_capable=False,
account_enabled=True,
user_type="Guest",
),
}
check = entra_users_mfa_capable()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended == "User Member User is not MFA capable."
assert result[0].resource == entra_client.users[member_user_id]
assert result[0].resource_name == "Member User"
assert result[0].resource_id == member_user_id
def test_unknown_user_type_is_evaluated(self):
"""Users without a ``user_type`` reported by Microsoft Graph must not be
silently dropped.
We only skip users that Graph explicitly reports as ``Guest``; for everyone
else (including ``user_type=None``) the check still evaluates MFA capability
so that we never mask findings on accounts whose type cannot be determined.
"""
entra_client = mock.MagicMock
entra_client.audited_tenant = "audited_tenant"
entra_client.audited_domain = DOMAIN
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_m365_provider(),
),
mock.patch(
"prowler.providers.m365.services.entra.entra_users_mfa_capable.entra_users_mfa_capable.entra_client",
new=entra_client,
),
):
from prowler.providers.m365.services.entra.entra_users_mfa_capable.entra_users_mfa_capable import (
entra_users_mfa_capable,
)
user_id = str(uuid4())
entra_client.users = {
user_id: User(
id=user_id,
name="Test User",
on_premises_sync_enabled=False,
directory_roles_ids=[],
is_mfa_capable=False,
account_enabled=True,
user_type=None,
)
}
check = entra_users_mfa_capable()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].status_extended == "User Test User is not MFA capable."
assert result[0].resource == entra_client.users[user_id]
assert result[0].resource_name == "Test User"
assert result[0].resource_id == user_id