feat(googleworkspace): pass app checks when the service is off for everyone (#13117)

This commit is contained in:
Pepe Fagoaga authored and GitHub committed 2026-10-22 01:07:19 +02:00
1 parent a59eae7b24
commit eae17a5e17
124 files changed
+3447 -618

No files matched your search

@@ -0,0 +1 @@
Google Workspace checks now read the policy Google applies (greatest sortOrder) instead of the last policy the Cloud Identity Policy API lists, which reported default values over admin-configured ones
@@ -0,0 +1 @@
Google Workspace checks for Chat, Drive, Gmail, Calendar and Groups pass with the reason stated when the app is turned off for everyone and no organizational unit or group turns it back on. The rule covers settings that only act while the app is on: Drive sharing settings and the settings of existing groups keep being evaluated because existing shares and groups stay reachable, and only group creation is exempt when Groups for Business is off
@@ -1,5 +1,8 @@
from typing import Optional
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.providers.googleworkspace.googleworkspace_provider import (
@@ -11,6 +14,52 @@ CUSTOMER_SCOPE = "customer"
OVERRIDE_SCOPE = "override"
UNKNOWN_SCOPE = "unknown"
# Settings whose policies are items of a list rather than competing values for
# one setting: every rule is a distinct policy and none outranks another.
LIST_SETTING_NAMESPACES = ("rule.",)
def _setting_type(policy: dict) -> str:
"""Return the setting type of a policy without the "settings/" prefix."""
return (policy.get("setting") or {}).get("type", "").removeprefix("settings/")
def _service_state(policy: dict) -> Optional[str]:
"""Return the serviceState a service status policy carries, if any."""
return (((policy.get("setting") or {}).get("value")) or {}).get("serviceState")
def _sort_order(policy: dict) -> float:
"""Return the policy's sortOrder, or 0 when the API did not give one."""
try:
return float((policy.get("policyQuery") or {}).get("sortOrder") or 0)
except (TypeError, ValueError):
return 0.0
class ServicePolicies(BaseModel):
"""Base of the per-service policy models: carries the service status.
`service_state` is the customer-level value of the service's
`<service>.service_status` setting. `service_state_overridden` records
that an organizational unit or group turns the service back on, in
which case "off" at the customer level does not mean off for everyone.
"""
service_state: Optional[str] = None
service_state_overridden: bool = False
@property
def off_for_everyone(self) -> bool:
"""Whether the service is off at the customer level with no override.
Settings that only act while the service is on have no effect then,
so the checks on them pass with that reason. Settings that govern
what already exists, such as Drive shares or Groups, keep being
evaluated: Google keeps existing shares and groups reachable.
"""
return self.service_state == "DISABLED" and not self.service_state_overridden
class GoogleWorkspaceService:
def __init__(
@@ -71,6 +120,88 @@ class GoogleWorkspaceService:
"""Whether a policy applies to the whole domain."""
return self._policy_scope(policy) == CUSTOMER_SCOPE
def _list_policies(
self, service, filter_str: str, context: str
) -> tuple[list[dict], bool]:
"""Read every Cloud Identity policy matching a filter, across pages.
Returns the policies read and whether every page was read. A failing
page is logged, the pages read before it are kept, and the caller
decides what a partial read means for its checks.
"""
policies: list[dict] = []
try:
request = service.policies().list(pageSize=100, filter=filter_str)
while request is not None:
response = request.execute()
policies.extend(response.get("policies", []))
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(error, context, self.provider.identity.customer_id)
return policies, False
return policies, True
def _effective_policies(self, policies: list[dict]) -> list[dict]:
"""Keep, per setting type, the customer-level policy Google applies.
The API returns every policy that exists for a setting: Google's
SYSTEM defaults next to the ADMIN policy saved in the Admin console.
The value in force is the one with the greatest policyQuery.sortOrder
(Policy API concepts, Max and Merge reducers), not the last one
listed. Policies without a sort order keep the API order, so the last
one wins. Rules are not reduced: each policy is a distinct rule.
"""
kept: list[dict] = []
position: dict[str, int] = {}
for policy in policies:
if not self._is_customer_level_policy(policy):
continue
setting_type = _setting_type(policy)
if setting_type.startswith(LIST_SETTING_NAMESPACES):
kept.append(policy)
continue
index = position.get(setting_type)
if index is None:
position[setting_type] = len(kept)
kept.append(policy)
elif _sort_order(policy) >= _sort_order(kept[index]):
kept[index] = policy
return kept
def _service_status(
self, policies: list[dict], setting_type: str
) -> tuple[Optional[str], bool]:
"""Return a service's customer-level state and whether anything overrides it.
The state comes from the effective customer-level policy of the given
`<service>.service_status` setting. A policy of that setting outside
the customer scope counts as an override when it turns the service on,
or when its scope could not be resolved: "off" then holds only for
part of the tenant.
"""
status_policies = [
policy for policy in policies if _setting_type(policy) == setting_type
]
overridden = any(
self._overrides_service_status(policy) for policy in status_policies
)
effective = self._effective_policies(status_policies)
state = _service_state(effective[0]) if effective else None
return state, overridden
def _overrides_service_status(self, policy: dict) -> bool:
"""Whether a status policy turns the service back on for part of the tenant.
A sub-OU or group policy counts only when it does not say DISABLED: an
OU that also turns the service off changes nothing, and Google lets an
access group only turn a service on. A policy whose scope cannot be
resolved counts as well, since it may be either.
"""
scope = self._policy_scope(policy)
if scope == CUSTOMER_SCOPE:
return False
return scope == UNKNOWN_SCOPE or _service_state(policy) != "DISABLED"
def _handle_api_error(self, error, context: str, resource_name: str = ""):
"""
Centralized Google Workspace API error handling.
@@ -31,43 +31,23 @@ class AdditionalServices(GoogleWorkspaceService):
logger.error("Failed to build Cloud Identity service")
return
request = service.policies().list(
pageSize=100,
filter='setting.type.matches("groups.service_status")',
policies, fetch_succeeded = self._list_policies(
service,
'setting.type.matches("groups.service_status")',
"fetching Additional Services policies",
)
fetch_succeeded = True
while request is not None:
try:
response = request.execute()
for policy in self._effective_policies(policies):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
value = setting.get("value", {})
for policy in response.get("policies", []):
if not self._is_customer_level_policy(policy):
continue
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
value = setting.get("value", {})
if setting_type == "groups.service_status":
self.policies.groups_service_state = value.get(
"serviceState"
)
logger.debug(
"Additional Services - Groups service state: "
f"{self.policies.groups_service_state}"
)
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
"fetching Additional Services policies",
self.provider.identity.customer_id,
if setting_type == "groups.service_status":
self.policies.groups_service_state = value.get("serviceState")
logger.debug(
"Additional Services - Groups service state: "
f"{self.policies.groups_service_state}"
)
fetch_succeeded = False
break
self.policies_fetched = fetch_succeeded
@@ -36,5 +36,5 @@
"calendar_external_sharing_primary_calendar",
"calendar_external_sharing_secondary_calendar"
],
"Notes": ""
"Notes": "When Google Calendar is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -28,7 +28,13 @@ class calendar_external_invitations_warning(Check):
warning_enabled = calendar_client.policies.external_invitations_warning
if warning_enabled is True:
if calendar_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Calendar is turned off for everyone in domain "
f"{calendar_client.provider.identity.domain}, so no invitation can be sent."
)
elif warning_enabled is True:
report.status = "PASS"
report.status_extended = (
f"External invitation warnings for Google Calendar are enabled "
@@ -36,5 +36,5 @@
"calendar_external_sharing_secondary_calendar",
"calendar_external_invitations_warning"
],
"Notes": ""
"Notes": "When Google Calendar is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -28,7 +28,13 @@ class calendar_external_sharing_primary_calendar(Check):
sharing = calendar_client.policies.primary_calendar_external_sharing
if sharing == "EXTERNAL_FREE_BUSY_ONLY":
if calendar_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Calendar is turned off for everyone in domain "
f"{calendar_client.provider.identity.domain}, so no calendar can be shared outside the organization."
)
elif sharing == "EXTERNAL_FREE_BUSY_ONLY":
report.status = "PASS"
report.status_extended = (
f"Primary calendar external sharing in domain "
@@ -36,5 +36,5 @@
"calendar_external_sharing_primary_calendar",
"calendar_external_invitations_warning"
],
"Notes": ""
"Notes": "When Google Calendar is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -28,7 +28,13 @@ class calendar_external_sharing_secondary_calendar(Check):
sharing = calendar_client.policies.secondary_calendar_external_sharing
if sharing == "EXTERNAL_FREE_BUSY_ONLY":
if calendar_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Calendar is turned off for everyone in domain "
f"{calendar_client.provider.identity.domain}, so no calendar can be shared outside the organization."
)
elif sharing == "EXTERNAL_FREE_BUSY_ONLY":
report.status = "PASS"
report.status_extended = (
f"Secondary calendar external sharing in domain "
@@ -1,9 +1,10 @@
from typing import Optional
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.providers.googleworkspace.lib.service.service import GoogleWorkspaceService
from prowler.providers.googleworkspace.lib.service.service import (
GoogleWorkspaceService,
ServicePolicies,
)
class Calendar(GoogleWorkspaceService):
@@ -30,66 +31,21 @@ class Calendar(GoogleWorkspaceService):
logger.error("Failed to build Cloud Identity service")
return
request = service.policies().list(
pageSize=100,
filter='setting.type.matches("calendar.*")',
policies, fetch_succeeded = self._list_policies(
service,
'setting.type.matches("calendar.*")',
"fetching calendar policies",
)
fetch_succeeded = True
while request is not None:
try:
response = request.execute()
(
self.policies.service_state,
self.policies.service_state_overridden,
) = self._service_status(policies, "calendar.service_status")
for policy in response.get("policies", []):
if not self._is_customer_level_policy(policy):
continue
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
value = setting.get("value", {})
if (
setting_type
== "calendar.primary_calendar_max_allowed_external_sharing"
):
self.policies.primary_calendar_external_sharing = value.get(
"maxAllowedExternalSharing"
)
logger.debug(
"Primary calendar external sharing: "
f"{self.policies.primary_calendar_external_sharing}"
)
elif (
setting_type
== "calendar.secondary_calendar_max_allowed_external_sharing"
):
self.policies.secondary_calendar_external_sharing = (
value.get("maxAllowedExternalSharing")
)
logger.debug(
"Secondary calendar external sharing: "
f"{self.policies.secondary_calendar_external_sharing}"
)
elif setting_type == "calendar.external_invitations":
self.policies.external_invitations_warning = value.get(
"warnOnInvite"
)
logger.debug(
"External invitations warning: "
f"{self.policies.external_invitations_warning}"
)
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
"fetching calendar policies",
self.provider.identity.customer_id,
)
fetch_succeeded = False
break
for policy in self._effective_policies(policies):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
self._process_setting(setting_type, setting.get("value", {}))
self.policies_fetched = fetch_succeeded
@@ -108,8 +64,35 @@ class Calendar(GoogleWorkspaceService):
)
self.policies_fetched = False
def _process_setting(self, setting_type: str, value: dict):
"""Populate the model from a single policy setting."""
if setting_type == "calendar.primary_calendar_max_allowed_external_sharing":
self.policies.primary_calendar_external_sharing = value.get(
"maxAllowedExternalSharing"
)
logger.debug(
"Primary calendar external sharing: "
f"{self.policies.primary_calendar_external_sharing}"
)
class CalendarPolicies(BaseModel):
elif setting_type == "calendar.secondary_calendar_max_allowed_external_sharing":
self.policies.secondary_calendar_external_sharing = value.get(
"maxAllowedExternalSharing"
)
logger.debug(
"Secondary calendar external sharing: "
f"{self.policies.secondary_calendar_external_sharing}"
)
elif setting_type == "calendar.external_invitations":
self.policies.external_invitations_warning = value.get("warnOnInvite")
logger.debug(
"External invitations warning: "
f"{self.policies.external_invitations_warning}"
)
class CalendarPolicies(ServicePolicies):
"""Model for domain-level Calendar policy settings."""
primary_calendar_external_sharing: Optional[str] = None
@@ -35,5 +35,5 @@
"RelatedTo": [
"chat_incoming_webhooks_disabled"
],
"Notes": ""
"Notes": "When Google Chat is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class chat_apps_installation_disabled(Check):
apps_enabled = chat_client.policies.enable_apps
if apps_enabled is False:
if chat_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Chat is turned off for everyone in domain "
f"{chat_client.provider.identity.domain}, so Chat apps cannot be installed."
)
elif apps_enabled is False:
report.status = "PASS"
report.status_extended = (
f"Chat apps installation is disabled "
@@ -36,5 +36,5 @@
"chat_internal_file_sharing_disabled",
"drive_sharing_allowlisted_domains"
],
"Notes": ""
"Notes": "When Google Chat is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class chat_external_file_sharing_disabled(Check):
external_sharing = chat_client.policies.external_file_sharing
if external_sharing == "NO_FILES":
if chat_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Chat is turned off for everyone in domain "
f"{chat_client.provider.identity.domain}, so no files can be shared externally through Chat."
)
elif external_sharing == "NO_FILES":
report.status = "PASS"
report.status_extended = (
f"External file sharing in Chat is disabled "
@@ -36,5 +36,5 @@
"chat_external_spaces_restricted",
"drive_sharing_allowlisted_domains"
],
"Notes": ""
"Notes": "When Google Chat is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -27,7 +27,13 @@ class chat_external_messaging_restricted(Check):
allow_external = chat_client.policies.allow_external_chat
restriction = chat_client.policies.external_chat_restriction
if allow_external is False:
if chat_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Chat is turned off for everyone in domain "
f"{chat_client.provider.identity.domain}, so no external messaging can take place."
)
elif allow_external is False:
report.status = "PASS"
report.status_extended = (
f"External Chat messaging is disabled "
@@ -36,5 +36,5 @@
"chat_external_messaging_restricted",
"drive_sharing_allowlisted_domains"
],
"Notes": ""
"Notes": "When Google Chat is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -27,7 +27,13 @@ class chat_external_spaces_restricted(Check):
spaces_enabled = chat_client.policies.external_spaces_enabled
allowlist_mode = chat_client.policies.external_spaces_domain_allowlist_mode
if spaces_enabled is False:
if chat_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Chat is turned off for everyone in domain "
f"{chat_client.provider.identity.domain}, so no external spaces can be created or joined."
)
elif spaces_enabled is False:
report.status = "PASS"
report.status_extended = (
f"External spaces are disabled "
@@ -35,5 +35,5 @@
"RelatedTo": [
"chat_apps_installation_disabled"
],
"Notes": ""
"Notes": "When Google Chat is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class chat_incoming_webhooks_disabled(Check):
webhooks_enabled = chat_client.policies.enable_webhooks
if webhooks_enabled is False:
if chat_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Chat is turned off for everyone in domain "
f"{chat_client.provider.identity.domain}, so incoming webhooks cannot be used."
)
elif webhooks_enabled is False:
report.status = "PASS"
report.status_extended = (
f"Incoming webhooks are disabled "
@@ -35,5 +35,5 @@
"RelatedTo": [
"chat_external_file_sharing_disabled"
],
"Notes": ""
"Notes": "When Google Chat is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class chat_internal_file_sharing_disabled(Check):
internal_sharing = chat_client.policies.internal_file_sharing
if internal_sharing == "NO_FILES":
if chat_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Chat is turned off for everyone in domain "
f"{chat_client.provider.identity.domain}, so no files can be shared through Chat."
)
elif internal_sharing == "NO_FILES":
report.status = "PASS"
report.status_extended = (
f"Internal file sharing in Chat is disabled "
@@ -1,9 +1,10 @@
from typing import Optional
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.providers.googleworkspace.lib.service.service import GoogleWorkspaceService
from prowler.providers.googleworkspace.lib.service.service import (
GoogleWorkspaceService,
ServicePolicies,
)
class Chat(GoogleWorkspaceService):
@@ -30,68 +31,21 @@ class Chat(GoogleWorkspaceService):
logger.error("Failed to build Cloud Identity service")
return
request = service.policies().list(
pageSize=100,
filter='setting.type.matches("chat.*")',
policies, fetch_succeeded = self._list_policies(
service,
'setting.type.matches("chat.*")',
"fetching Chat policies",
)
fetch_succeeded = True
while request is not None:
try:
response = request.execute()
(
self.policies.service_state,
self.policies.service_state_overridden,
) = self._service_status(policies, "chat.service_status")
for policy in response.get("policies", []):
if not self._is_customer_level_policy(policy):
continue
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
logger.debug(f"Processing setting type: {setting_type}")
value = setting.get("value", {})
if setting_type == "chat.chat_file_sharing":
self.policies.external_file_sharing = value.get(
"externalFileSharing"
)
self.policies.internal_file_sharing = value.get(
"internalFileSharing"
)
logger.debug("Chat file sharing settings fetched.")
elif setting_type == "chat.external_chat_restriction":
self.policies.allow_external_chat = value.get(
"allowExternalChat"
)
self.policies.external_chat_restriction = value.get(
"externalChatRestriction"
)
logger.debug(
"Chat external chat restriction settings fetched."
)
elif setting_type == "chat.chat_external_spaces":
self.policies.external_spaces_enabled = value.get("enabled")
self.policies.external_spaces_domain_allowlist_mode = (
value.get("domainAllowlistMode")
)
logger.debug("Chat external spaces settings fetched.")
elif setting_type == "chat.chat_apps_access":
self.policies.enable_apps = value.get("enableApps")
self.policies.enable_webhooks = value.get("enableWebhooks")
logger.debug("Chat apps access settings fetched.")
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
"fetching Chat policies",
self.provider.identity.customer_id,
)
fetch_succeeded = False
break
for policy in self._effective_policies(policies):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
self._process_setting(setting_type, setting.get("value", {}))
self.policies_fetched = fetch_succeeded
logger.info("Chat policies fetched successfully.")
@@ -104,8 +58,36 @@ class Chat(GoogleWorkspaceService):
)
self.policies_fetched = False
def _process_setting(self, setting_type: str, value: dict):
"""Populate the model from a single policy setting."""
logger.debug(f"Processing setting type: {setting_type}")
class ChatPolicies(BaseModel):
if setting_type == "chat.chat_file_sharing":
self.policies.external_file_sharing = value.get("externalFileSharing")
self.policies.internal_file_sharing = value.get("internalFileSharing")
logger.debug("Chat file sharing settings fetched.")
elif setting_type == "chat.external_chat_restriction":
self.policies.allow_external_chat = value.get("allowExternalChat")
self.policies.external_chat_restriction = value.get(
"externalChatRestriction"
)
logger.debug("Chat external chat restriction settings fetched.")
elif setting_type == "chat.chat_external_spaces":
self.policies.external_spaces_enabled = value.get("enabled")
self.policies.external_spaces_domain_allowlist_mode = value.get(
"domainAllowlistMode"
)
logger.debug("Chat external spaces settings fetched.")
elif setting_type == "chat.chat_apps_access":
self.policies.enable_apps = value.get("enableApps")
self.policies.enable_webhooks = value.get("enableWebhooks")
logger.debug("Chat apps access settings fetched.")
class ChatPolicies(ServicePolicies):
"""Model for domain-level Chat policy settings."""
# chat.chat_file_sharing
@@ -36,5 +36,5 @@
"drive_external_sharing_warn_users",
"drive_publishing_files_disabled"
],
"Notes": ""
"Notes": "When Google Drive is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -27,7 +27,13 @@ class drive_access_checker_recipients_only(Check):
access_checker = drive_client.policies.access_checker_suggestions
if access_checker == "RECIPIENTS_ONLY":
if drive_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Drive is turned off for everyone in domain "
f"{drive_client.provider.identity.domain}, so Access Checker cannot widen any share."
)
elif access_checker == "RECIPIENTS_ONLY":
report.status = "PASS"
report.status_extended = (
f"Drive and Docs Access Checker in domain "
@@ -31,5 +31,5 @@
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
"Notes": "When Google Drive is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -28,7 +28,13 @@ class drive_desktop_access_disabled(Check):
allow_desktop = drive_client.policies.allow_drive_for_desktop
if allow_desktop is False:
if drive_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Drive is turned off for everyone in domain "
f"{drive_client.provider.identity.domain}, so Drive for desktop cannot sync any content."
)
elif allow_desktop is False:
report.status = "PASS"
report.status_extended = (
f"Google Drive for desktop is disabled in domain "
@@ -1,9 +1,10 @@
from typing import Optional
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.providers.googleworkspace.lib.service.service import GoogleWorkspaceService
from prowler.providers.googleworkspace.lib.service.service import (
GoogleWorkspaceService,
ServicePolicies,
)
class Drive(GoogleWorkspaceService):
@@ -31,88 +32,21 @@ class Drive(GoogleWorkspaceService):
logger.error("Failed to build Cloud Identity service")
return
request = service.policies().list(
pageSize=100,
filter='setting.type.matches("drive_and_docs.*")',
policies, fetch_succeeded = self._list_policies(
service,
'setting.type.matches("drive_and_docs.*")',
"fetching Drive and Docs policies",
)
fetch_succeeded = True
while request is not None:
try:
response = request.execute()
(
self.policies.service_state,
self.policies.service_state_overridden,
) = self._service_status(policies, "drive_and_docs.service_status")
for policy in response.get("policies", []):
if not self._is_customer_level_policy(policy):
continue
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
value = setting.get("value", {})
if setting_type == "drive_and_docs.external_sharing":
self.policies.external_sharing_mode = value.get(
"externalSharingMode"
)
self.policies.warn_for_external_sharing = value.get(
"warnForExternalSharing"
)
self.policies.warn_for_sharing_outside_allowlisted_domains = value.get(
"warnForSharingOutsideAllowlistedDomains"
)
self.policies.allow_publishing_files = value.get(
"allowPublishingFiles"
)
self.policies.access_checker_suggestions = value.get(
"accessCheckerSuggestions"
)
self.policies.allowed_parties_for_distributing_content = (
value.get("allowedPartiesForDistributingContent")
)
logger.debug(
"Drive external sharing settings fetched: "
f"mode={self.policies.external_sharing_mode}, "
f"warn={self.policies.warn_for_external_sharing}, "
f"publish={self.policies.allow_publishing_files}"
)
elif setting_type == "drive_and_docs.shared_drive_creation":
self.policies.allow_shared_drive_creation = value.get(
"allowSharedDriveCreation"
)
self.policies.allow_managers_to_override_settings = (
value.get("allowManagersToOverrideSettings")
)
self.policies.allow_non_member_access = value.get(
"allowNonMemberAccess"
)
self.policies.allowed_parties_for_download_print_copy = (
value.get("allowedPartiesForDownloadPrintCopy")
)
logger.debug(
"Drive shared drive creation settings fetched: "
f"creation={self.policies.allow_shared_drive_creation}, "
f"managers_override={self.policies.allow_managers_to_override_settings}"
)
elif setting_type == "drive_and_docs.drive_for_desktop":
self.policies.allow_drive_for_desktop = value.get(
"allowDriveForDesktop"
)
logger.debug(
"Drive for desktop setting fetched: "
f"{self.policies.allow_drive_for_desktop}"
)
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
"fetching Drive and Docs policies",
self.provider.identity.customer_id,
)
fetch_succeeded = False
break
for policy in self._effective_policies(policies):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
self._process_setting(setting_type, setting.get("value", {}))
self.policies_fetched = fetch_succeeded
@@ -131,8 +65,56 @@ class Drive(GoogleWorkspaceService):
)
self.policies_fetched = False
def _process_setting(self, setting_type: str, value: dict):
"""Populate the model from a single policy setting."""
if setting_type == "drive_and_docs.external_sharing":
self.policies.external_sharing_mode = value.get("externalSharingMode")
self.policies.warn_for_external_sharing = value.get(
"warnForExternalSharing"
)
self.policies.warn_for_sharing_outside_allowlisted_domains = value.get(
"warnForSharingOutsideAllowlistedDomains"
)
self.policies.allow_publishing_files = value.get("allowPublishingFiles")
self.policies.access_checker_suggestions = value.get(
"accessCheckerSuggestions"
)
self.policies.allowed_parties_for_distributing_content = value.get(
"allowedPartiesForDistributingContent"
)
logger.debug(
"Drive external sharing settings fetched: "
f"mode={self.policies.external_sharing_mode}, "
f"warn={self.policies.warn_for_external_sharing}, "
f"publish={self.policies.allow_publishing_files}"
)
class DrivePolicies(BaseModel):
elif setting_type == "drive_and_docs.shared_drive_creation":
self.policies.allow_shared_drive_creation = value.get(
"allowSharedDriveCreation"
)
self.policies.allow_managers_to_override_settings = value.get(
"allowManagersToOverrideSettings"
)
self.policies.allow_non_member_access = value.get("allowNonMemberAccess")
self.policies.allowed_parties_for_download_print_copy = value.get(
"allowedPartiesForDownloadPrintCopy"
)
logger.debug(
"Drive shared drive creation settings fetched: "
f"creation={self.policies.allow_shared_drive_creation}, "
f"managers_override={self.policies.allow_managers_to_override_settings}"
)
elif setting_type == "drive_and_docs.drive_for_desktop":
self.policies.allow_drive_for_desktop = value.get("allowDriveForDesktop")
logger.debug(
"Drive for desktop setting fetched: "
f"{self.policies.allow_drive_for_desktop}"
)
class DrivePolicies(ServicePolicies):
"""Model for domain-level Drive and Docs policy settings."""
# drive_and_docs.external_sharing
@@ -35,5 +35,5 @@
"drive_shared_drive_members_only_access",
"drive_shared_drive_disable_download_print_copy"
],
"Notes": ""
"Notes": "When Google Drive is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -28,7 +28,13 @@ class drive_shared_drive_creation_allowed(Check):
allow_creation = drive_client.policies.allow_shared_drive_creation
if allow_creation is True:
if drive_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Google Drive is turned off for everyone in domain "
f"{drive_client.provider.identity.domain}, so no shared drive can be created."
)
elif allow_creation is True:
report.status = "PASS"
report.status_extended = (
f"Users in domain {drive_client.provider.identity.domain} "
@@ -36,5 +36,5 @@
"gmail_encrypted_attachment_protection_enabled",
"gmail_script_attachment_protection_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -29,7 +29,13 @@ class gmail_anomalous_attachment_protection_enabled(Check):
gmail_client.policies.anomalous_attachment_protection_consequence
)
if enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif enabled is False:
report.status = "FAIL"
report.status_extended = (
f"Protection against anomalous attachment types in emails "
@@ -36,5 +36,5 @@
"gmail_pop_imap_access_disabled",
"gmail_per_user_outbound_gateway_disabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class gmail_auto_forwarding_disabled(Check):
forwarding_enabled = gmail_client.policies.enable_auto_forwarding
if forwarding_enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no email can be forwarded."
)
elif forwarding_enabled is False:
report.status = "PASS"
report.status_extended = (
f"Automatic email forwarding is disabled "
@@ -33,5 +33,5 @@
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class gmail_comprehensive_mail_storage_enabled(Check):
storage_enabled = gmail_client.policies.comprehensive_mail_storage_enabled
if storage_enabled is True:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so there is no Gmail mail to store."
)
elif storage_enabled is True:
report.status = "PASS"
report.status_extended = (
f"Comprehensive mail storage is enabled "
@@ -38,5 +38,5 @@
"gmail_unauthenticated_email_protection_enabled",
"gmail_groups_spoofing_protection_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -34,7 +34,13 @@ class gmail_domain_spoofing_protection_enabled(Check):
consequence = gmail_client.policies.domain_spoofing_consequence
domain = gmail_client.provider.identity.domain
if enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif enabled is False:
report.status = "FAIL"
report.status_extended = (
f"Protection against domain spoofing based on similar domain names "
@@ -38,5 +38,5 @@
"gmail_unauthenticated_email_protection_enabled",
"gmail_groups_spoofing_protection_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -34,7 +34,13 @@ class gmail_employee_name_spoofing_protection_enabled(Check):
consequence = gmail_client.policies.employee_name_spoofing_consequence
domain = gmail_client.provider.identity.domain
if enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif enabled is False:
report.status = "FAIL"
report.status_extended = (
f"Protection against spoofing of employee names "
@@ -36,5 +36,5 @@
"gmail_script_attachment_protection_enabled",
"gmail_anomalous_attachment_protection_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -29,7 +29,13 @@ class gmail_encrypted_attachment_protection_enabled(Check):
gmail_client.policies.encrypted_attachment_protection_consequence
)
if enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif enabled is False:
report.status = "FAIL"
report.status_extended = (
f"Protection against encrypted attachments from untrusted "
@@ -33,5 +33,5 @@
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -28,7 +28,13 @@ class gmail_enhanced_pre_delivery_scanning_enabled(Check):
gmail_client.policies.enable_enhanced_pre_delivery_scanning
)
if scanning_enabled is True:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif scanning_enabled is True:
report.status = "PASS"
report.status_extended = (
f"Enhanced pre-delivery message scanning is enabled "
@@ -36,5 +36,5 @@
"gmail_shortener_scanning_enabled",
"gmail_untrusted_link_warnings_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class gmail_external_image_scanning_enabled(Check):
scanning_enabled = gmail_client.policies.enable_external_image_scanning
if scanning_enabled is True:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif scanning_enabled is True:
report.status = "PASS"
report.status_extended = (
f"Scanning of linked images for malicious content is enabled "
@@ -38,5 +38,5 @@
"gmail_inbound_domain_spoofing_protection_enabled",
"gmail_unauthenticated_email_protection_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -41,7 +41,13 @@ class gmail_groups_spoofing_protection_enabled(Check):
else "all groups"
)
if enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif enabled is False:
report.status = "FAIL"
report.status_extended = (
f"Protection of groups from inbound emails spoofing your "
@@ -38,5 +38,5 @@
"gmail_unauthenticated_email_protection_enabled",
"gmail_groups_spoofing_protection_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -34,7 +34,13 @@ class gmail_inbound_domain_spoofing_protection_enabled(Check):
consequence = gmail_client.policies.inbound_domain_spoofing_consequence
domain = gmail_client.provider.identity.domain
if enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif enabled is False:
report.status = "FAIL"
report.status_extended = (
f"Protection against inbound emails spoofing your domain "
@@ -33,5 +33,5 @@
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class gmail_mail_delegation_disabled(Check):
delegation_enabled = gmail_client.policies.enable_mail_delegation
if delegation_enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no mailbox can be delegated."
)
elif delegation_enabled is False:
report.status = "PASS"
report.status_extended = (
f"Mail delegation is disabled "
@@ -36,5 +36,5 @@
"gmail_pop_imap_access_disabled",
"gmail_auto_forwarding_disabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class gmail_per_user_outbound_gateway_disabled(Check):
gateway_allowed = gmail_client.policies.allow_per_user_outbound_gateway
if gateway_allowed is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no outbound gateway can be used."
)
elif gateway_allowed is False:
report.status = "PASS"
report.status_extended = (
f"Per-user outbound gateways are disabled "
@@ -36,5 +36,5 @@
"gmail_auto_forwarding_disabled",
"gmail_per_user_outbound_gateway_disabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -27,7 +27,13 @@ class gmail_pop_imap_access_disabled(Check):
pop_enabled = gmail_client.policies.enable_pop_access
imap_enabled = gmail_client.policies.enable_imap_access
if pop_enabled is False and imap_enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no mailbox can be reached over POP or IMAP."
)
elif pop_enabled is False and imap_enabled is False:
report.status = "PASS"
report.status_extended = (
f"POP and IMAP access are both disabled "
@@ -36,5 +36,5 @@
"gmail_encrypted_attachment_protection_enabled",
"gmail_anomalous_attachment_protection_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -27,7 +27,13 @@ class gmail_script_attachment_protection_enabled(Check):
enabled = gmail_client.policies.enable_script_attachment_protection
consequence = gmail_client.policies.script_attachment_protection_consequence
if enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif enabled is False:
report.status = "FAIL"
report.status_extended = (
f"Protection against attachments with scripts from "
@@ -1,9 +1,10 @@
from typing import Optional
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.providers.googleworkspace.lib.service.service import GoogleWorkspaceService
from prowler.providers.googleworkspace.lib.service.service import (
GoogleWorkspaceService,
ServicePolicies,
)
class Gmail(GoogleWorkspaceService):
@@ -30,160 +31,21 @@ class Gmail(GoogleWorkspaceService):
logger.error("Failed to build Cloud Identity service")
return
request = service.policies().list(
pageSize=100,
filter='setting.type.matches("gmail.*")',
policies, fetch_succeeded = self._list_policies(
service,
'setting.type.matches("gmail.*")',
"fetching Gmail policies",
)
fetch_succeeded = True
while request is not None:
try:
response = request.execute()
(
self.policies.service_state,
self.policies.service_state_overridden,
) = self._service_status(policies, "gmail.service_status")
for policy in response.get("policies", []):
if not self._is_customer_level_policy(policy):
continue
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
logger.debug(f"Processing setting type: {setting_type}")
value = setting.get("value", {})
if setting_type == "gmail.mail_delegation":
self.policies.enable_mail_delegation = value.get(
"enableMailDelegation"
)
logger.debug("Gmail mail delegation setting fetched.")
elif setting_type == "gmail.email_attachment_safety":
self.policies.enable_encrypted_attachment_protection = (
value.get("enableEncryptedAttachmentProtection")
)
self.policies.encrypted_attachment_protection_consequence = value.get(
"encryptedAttachmentProtectionConsequence"
)
self.policies.enable_script_attachment_protection = (
value.get("enableAttachmentWithScriptsProtection")
)
self.policies.script_attachment_protection_consequence = (
value.get("scriptAttachmentProtectionConsequence")
)
self.policies.enable_anomalous_attachment_protection = (
value.get("enableAnomalousAttachmentProtection")
)
self.policies.anomalous_attachment_protection_consequence = value.get(
"anomalousAttachmentProtectionConsequence"
)
logger.debug("Gmail attachment safety settings fetched.")
elif setting_type == "gmail.links_and_external_images":
self.policies.enable_shortener_scanning = value.get(
"enableShortenerScanning"
)
self.policies.enable_external_image_scanning = value.get(
"enableExternalImageScanning"
)
self.policies.enable_aggressive_warnings_on_untrusted_links = value.get(
"enableAggressiveWarningsOnUntrustedLinks"
)
logger.debug(
"Gmail links and external images settings fetched."
)
elif setting_type == "gmail.spoofing_and_authentication":
self.policies.detect_domain_name_spoofing = value.get(
"detectDomainNameSpoofing"
)
self.policies.domain_spoofing_consequence = value.get(
"domainSpoofingConsequence"
)
self.policies.detect_employee_name_spoofing = value.get(
"detectEmployeeNameSpoofing"
)
self.policies.employee_name_spoofing_consequence = (
value.get("employeeNameSpoofingConsequence")
)
self.policies.detect_inbound_domain_spoofing = value.get(
"detectDomainSpoofingFromUnauthenticatedSenders"
)
self.policies.inbound_domain_spoofing_consequence = (
value.get("inboundDomainSpoofingConsequence")
)
self.policies.detect_unauthenticated_emails = value.get(
"detectUnauthenticatedEmails"
)
self.policies.unauthenticated_email_consequence = value.get(
"unauthenticatedEmailConsequence"
)
self.policies.detect_groups_spoofing = value.get(
"detectGroupsSpoofing"
)
self.policies.groups_spoofing_visibility_type = value.get(
"groupsSpoofingVisibilityType"
)
self.policies.groups_spoofing_consequence = value.get(
"groupsSpoofingConsequence"
)
logger.debug(
"Gmail spoofing and authentication settings fetched."
)
elif setting_type == "gmail.pop_access":
self.policies.enable_pop_access = value.get(
"enablePopAccess"
)
logger.debug("Gmail POP access setting fetched.")
elif setting_type == "gmail.imap_access":
self.policies.enable_imap_access = value.get(
"enableImapAccess"
)
logger.debug("Gmail IMAP access setting fetched.")
elif setting_type == "gmail.auto_forwarding":
self.policies.enable_auto_forwarding = value.get(
"enableAutoForwarding"
)
logger.debug("Gmail auto-forwarding setting fetched.")
elif setting_type == "gmail.per_user_outbound_gateway":
self.policies.allow_per_user_outbound_gateway = value.get(
"allowUsersToUseExternalSmtpServers"
)
logger.debug(
"Gmail per-user outbound gateway setting fetched."
)
elif (
setting_type
== "gmail.enhanced_pre_delivery_message_scanning"
):
self.policies.enable_enhanced_pre_delivery_scanning = (
value.get("enableImprovedSuspiciousContentDetection")
)
logger.debug(
"Gmail enhanced pre-delivery scanning setting fetched."
)
elif setting_type == "gmail.comprehensive_mail_storage":
self.policies.comprehensive_mail_storage_enabled = (
value.get("ruleId") is not None
)
logger.debug(
"Gmail comprehensive mail storage setting fetched."
)
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
"fetching Gmail policies",
self.provider.identity.customer_id,
)
fetch_succeeded = False
break
for policy in self._effective_policies(policies):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
self._process_setting(setting_type, setting.get("value", {}))
self.policies_fetched = fetch_succeeded
logger.info("Gmail policies fetched successfully.")
@@ -196,8 +58,113 @@ class Gmail(GoogleWorkspaceService):
)
self.policies_fetched = False
def _process_setting(self, setting_type: str, value: dict):
"""Populate the model from a single policy setting."""
logger.debug(f"Processing setting type: {setting_type}")
class GmailPolicies(BaseModel):
if setting_type == "gmail.mail_delegation":
self.policies.enable_mail_delegation = value.get("enableMailDelegation")
logger.debug("Gmail mail delegation setting fetched.")
elif setting_type == "gmail.email_attachment_safety":
self.policies.enable_encrypted_attachment_protection = value.get(
"enableEncryptedAttachmentProtection"
)
self.policies.encrypted_attachment_protection_consequence = value.get(
"encryptedAttachmentProtectionConsequence"
)
self.policies.enable_script_attachment_protection = value.get(
"enableAttachmentWithScriptsProtection"
)
self.policies.script_attachment_protection_consequence = value.get(
"scriptAttachmentProtectionConsequence"
)
self.policies.enable_anomalous_attachment_protection = value.get(
"enableAnomalousAttachmentProtection"
)
self.policies.anomalous_attachment_protection_consequence = value.get(
"anomalousAttachmentProtectionConsequence"
)
logger.debug("Gmail attachment safety settings fetched.")
elif setting_type == "gmail.links_and_external_images":
self.policies.enable_shortener_scanning = value.get(
"enableShortenerScanning"
)
self.policies.enable_external_image_scanning = value.get(
"enableExternalImageScanning"
)
self.policies.enable_aggressive_warnings_on_untrusted_links = value.get(
"enableAggressiveWarningsOnUntrustedLinks"
)
logger.debug("Gmail links and external images settings fetched.")
elif setting_type == "gmail.spoofing_and_authentication":
self.policies.detect_domain_name_spoofing = value.get(
"detectDomainNameSpoofing"
)
self.policies.domain_spoofing_consequence = value.get(
"domainSpoofingConsequence"
)
self.policies.detect_employee_name_spoofing = value.get(
"detectEmployeeNameSpoofing"
)
self.policies.employee_name_spoofing_consequence = value.get(
"employeeNameSpoofingConsequence"
)
self.policies.detect_inbound_domain_spoofing = value.get(
"detectDomainSpoofingFromUnauthenticatedSenders"
)
self.policies.inbound_domain_spoofing_consequence = value.get(
"inboundDomainSpoofingConsequence"
)
self.policies.detect_unauthenticated_emails = value.get(
"detectUnauthenticatedEmails"
)
self.policies.unauthenticated_email_consequence = value.get(
"unauthenticatedEmailConsequence"
)
self.policies.detect_groups_spoofing = value.get("detectGroupsSpoofing")
self.policies.groups_spoofing_visibility_type = value.get(
"groupsSpoofingVisibilityType"
)
self.policies.groups_spoofing_consequence = value.get(
"groupsSpoofingConsequence"
)
logger.debug("Gmail spoofing and authentication settings fetched.")
elif setting_type == "gmail.pop_access":
self.policies.enable_pop_access = value.get("enablePopAccess")
logger.debug("Gmail POP access setting fetched.")
elif setting_type == "gmail.imap_access":
self.policies.enable_imap_access = value.get("enableImapAccess")
logger.debug("Gmail IMAP access setting fetched.")
elif setting_type == "gmail.auto_forwarding":
self.policies.enable_auto_forwarding = value.get("enableAutoForwarding")
logger.debug("Gmail auto-forwarding setting fetched.")
elif setting_type == "gmail.per_user_outbound_gateway":
self.policies.allow_per_user_outbound_gateway = value.get(
"allowUsersToUseExternalSmtpServers"
)
logger.debug("Gmail per-user outbound gateway setting fetched.")
elif setting_type == "gmail.enhanced_pre_delivery_message_scanning":
self.policies.enable_enhanced_pre_delivery_scanning = value.get(
"enableImprovedSuspiciousContentDetection"
)
logger.debug("Gmail enhanced pre-delivery scanning setting fetched.")
elif setting_type == "gmail.comprehensive_mail_storage":
self.policies.comprehensive_mail_storage_enabled = (
value.get("ruleId") is not None
)
logger.debug("Gmail comprehensive mail storage setting fetched.")
class GmailPolicies(ServicePolicies):
"""Model for domain-level Gmail policy settings."""
# gmail.mail_delegation
@@ -36,5 +36,5 @@
"gmail_external_image_scanning_enabled",
"gmail_untrusted_link_warnings_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -26,7 +26,13 @@ class gmail_shortener_scanning_enabled(Check):
scanning_enabled = gmail_client.policies.enable_shortener_scanning
if scanning_enabled is True:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif scanning_enabled is True:
report.status = "PASS"
report.status_extended = (
f"Identification of links behind shortened URLs is enabled "
@@ -38,5 +38,5 @@
"gmail_inbound_domain_spoofing_protection_enabled",
"gmail_groups_spoofing_protection_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -27,7 +27,13 @@ class gmail_unauthenticated_email_protection_enabled(Check):
enabled = gmail_client.policies.detect_unauthenticated_emails
consequence = gmail_client.policies.unauthenticated_email_consequence
if enabled is False:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif enabled is False:
report.status = "FAIL"
report.status_extended = (
f"Protection against unauthenticated emails is disabled "
@@ -36,5 +36,5 @@
"gmail_shortener_scanning_enabled",
"gmail_external_image_scanning_enabled"
],
"Notes": ""
"Notes": "When Gmail is turned off for everyone, the check passes: the setting has no effect while the service is off."
}
@@ -28,7 +28,13 @@ class gmail_untrusted_link_warnings_enabled(Check):
gmail_client.policies.enable_aggressive_warnings_on_untrusted_links
)
if warnings_enabled is True:
if gmail_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Gmail is turned off for everyone in domain "
f"{gmail_client.provider.identity.domain}, so no user mailbox receives mail through Gmail."
)
elif warnings_enabled is True:
report.status = "PASS"
report.status_extended = (
f"Warning prompts for clicks on untrusted domain links are enabled "
@@ -36,5 +36,5 @@
"groups_external_access_restricted",
"groups_view_conversations_restricted"
],
"Notes": ""
"Notes": "When Groups for Business is turned off for everyone, only the group creation setting is exempt: no group can be created in the Google Groups app, but existing groups stay active, so the external members and incoming email settings are still evaluated."
}
@@ -13,6 +13,10 @@ class groups_creation_restricted(Check):
- Only organization admins can create groups (not all users)
- Group owners cannot allow external members
- Group owners cannot allow incoming email from outside the organization
When Groups for Business is off for everyone, nobody can create a group in
the Google Groups app, so the creation setting is moot. Existing groups
stay active and keep receiving mail, so the other two settings still count.
"""
def execute(self) -> List[CheckReportGoogleWorkspace]:
@@ -35,10 +39,12 @@ class groups_creation_restricted(Check):
incoming_mail = policies.owners_can_allow_incoming_mail_from_public
issues = []
service_off = policies.off_for_everyone
# Check creation access level
# Default is USERS_IN_DOMAIN (insecure) — only ADMIN_ONLY is compliant
if access_level is None or access_level != "ADMIN_ONLY":
# Default is USERS_IN_DOMAIN (insecure) — only ADMIN_ONLY is compliant.
# Moot while the service is off: nobody can create a group then.
if not service_off and access_level != "ADMIN_ONLY":
effective = access_level or "USERS_IN_DOMAIN (default)"
issues.append(
f"group creation is set to {effective} instead of ADMIN_ONLY"
@@ -56,13 +62,28 @@ class groups_creation_restricted(Check):
"group owners can allow incoming email from outside the organization"
)
if not issues:
if not issues and service_off:
report.status = "PASS"
report.status_extended = (
f"Groups for Business is turned off for everyone in domain "
f"{domain}, so no group can be created, and existing groups "
f"allow no external members and no incoming email from outside."
)
elif not issues:
report.status = "PASS"
report.status_extended = (
f"Group creation is properly restricted in domain {domain}: "
f"admin-only creation, no external members, "
f"no incoming email from outside."
)
elif service_off:
report.status = "FAIL"
issues_text = "; ".join(issues)
report.status_extended = (
f"Groups for Business is turned off for everyone in domain "
f"{domain}, so no group can be created, but existing groups "
f"remain active and are not fully restricted: {issues_text}."
)
else:
report.status = "FAIL"
issues_text = "; ".join(issues)
@@ -1,9 +1,10 @@
from typing import Optional
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.providers.googleworkspace.lib.service.service import GoogleWorkspaceService
from prowler.providers.googleworkspace.lib.service.service import (
GoogleWorkspaceService,
ServicePolicies,
)
class Groups(GoogleWorkspaceService):
@@ -30,62 +31,21 @@ class Groups(GoogleWorkspaceService):
logger.error("Failed to build Cloud Identity service")
return
request = service.policies().list(
pageSize=100,
filter='setting.type.matches("groups_for_business.*")',
policies, fetch_succeeded = self._list_policies(
service,
'setting.type.matches("groups_for_business.*")',
"fetching Groups for Business policies",
)
fetch_succeeded = True
while request is not None:
try:
response = request.execute()
(
self.policies.service_state,
self.policies.service_state_overridden,
) = self._service_status(policies, "groups_for_business.service_status")
for policy in response.get("policies", []):
if not self._is_customer_level_policy(policy):
continue
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
logger.debug(f"Processing setting type: {setting_type}")
value = setting.get("value", {})
if setting_type == "groups_for_business.groups_sharing":
self.policies.collaboration_capability = value.get(
"collaborationCapability"
)
self.policies.create_groups_access_level = value.get(
"createGroupsAccessLevel"
)
self.policies.owners_can_allow_external_members = value.get(
"ownersCanAllowExternalMembers"
)
self.policies.owners_can_allow_incoming_mail_from_public = (
value.get("ownersCanAllowIncomingMailFromPublic")
)
self.policies.view_topics_default_access_level = value.get(
"viewTopicsDefaultAccessLevel"
)
self.policies.owners_can_hide_groups = value.get(
"ownersCanHideGroups"
)
self.policies.new_groups_are_hidden = value.get(
"newGroupsAreHidden"
)
logger.debug(
"Groups for Business sharing settings fetched."
)
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
"fetching Groups for Business policies",
self.provider.identity.customer_id,
)
fetch_succeeded = False
break
for policy in self._effective_policies(policies):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
self._process_setting(setting_type, setting.get("value", {}))
self.policies_fetched = fetch_succeeded
@@ -104,8 +64,32 @@ class Groups(GoogleWorkspaceService):
)
self.policies_fetched = False
def _process_setting(self, setting_type: str, value: dict):
"""Populate the model from a single policy setting."""
logger.debug(f"Processing setting type: {setting_type}")
class GroupsPolicies(BaseModel):
if setting_type == "groups_for_business.groups_sharing":
self.policies.collaboration_capability = value.get(
"collaborationCapability"
)
self.policies.create_groups_access_level = value.get(
"createGroupsAccessLevel"
)
self.policies.owners_can_allow_external_members = value.get(
"ownersCanAllowExternalMembers"
)
self.policies.owners_can_allow_incoming_mail_from_public = value.get(
"ownersCanAllowIncomingMailFromPublic"
)
self.policies.view_topics_default_access_level = value.get(
"viewTopicsDefaultAccessLevel"
)
self.policies.owners_can_hide_groups = value.get("ownersCanHideGroups")
self.policies.new_groups_are_hidden = value.get("newGroupsAreHidden")
logger.debug("Groups for Business sharing settings fetched.")
class GroupsPolicies(ServicePolicies):
"""Model for domain-level Groups for Business policy settings."""
# groups_for_business.groups_sharing
@@ -36,5 +36,5 @@
"groups_external_access_restricted",
"groups_creation_restricted"
],
"Notes": ""
"Notes": "When Groups for Business is turned off for everyone, the check passes: the setting only governs who can read conversations in the Google Groups app, which is off. Existing groups keep delivering mail to their members."
}
@@ -29,7 +29,13 @@ class groups_view_conversations_restricted(Check):
view_access = groups_client.policies.view_topics_default_access_level
domain = groups_client.provider.identity.domain
if view_access == "GROUP_MEMBERS":
if groups_client.policies.off_for_everyone:
report.status = "PASS"
report.status_extended = (
f"Groups for Business is turned off for everyone in domain "
f"{groups_client.provider.identity.domain}, so no group conversation can be viewed in the Google Groups app."
)
elif view_access == "GROUP_MEMBERS":
report.status = "PASS"
report.status_extended = (
f"Default permission to view conversations is set to "
@@ -30,41 +30,22 @@ class Marketplace(GoogleWorkspaceService):
logger.error("Failed to build Cloud Identity service")
return
request = service.policies().list(
pageSize=100,
filter='setting.type.matches("workspace_marketplace.*")',
policies, fetch_succeeded = self._list_policies(
service,
'setting.type.matches("workspace_marketplace.*")',
"fetching Marketplace policies",
)
fetch_succeeded = True
while request is not None:
try:
response = request.execute()
for policy in self._effective_policies(policies):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
value = setting.get("value", {})
for policy in response.get("policies", []):
if not self._is_customer_level_policy(policy):
continue
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
value = setting.get("value", {})
if setting_type == "workspace_marketplace.apps_access_options":
self.policies.access_level = value.get("accessLevel")
logger.debug(
"Marketplace access level: "
f"{self.policies.access_level}"
)
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
"fetching Marketplace policies",
self.provider.identity.customer_id,
if setting_type == "workspace_marketplace.apps_access_options":
self.policies.access_level = value.get("accessLevel")
logger.debug(
"Marketplace access level: " f"{self.policies.access_level}"
)
fetch_succeeded = False
break
self.policies_fetched = fetch_succeeded
@@ -78,52 +78,33 @@ class Security(GoogleWorkspaceService):
def _fetch_namespace(self, service, filter_str: str, fetch_succeeded: bool) -> bool:
"""Fetch policies for a single namespace filter."""
try:
request = service.policies().list(
pageSize=100,
filter=filter_str,
policies, succeeded = self._list_policies(
service, filter_str, f"fetching policies with filter {filter_str}"
)
customer_level = []
for policy in policies:
setting_type = (
policy.get("setting", {}).get("type", "").removeprefix("settings/")
)
while request is not None:
try:
response = request.execute()
scope = self._policy_scope(policy)
if scope != CUSTOMER_SCOPE:
if scope == UNKNOWN_SCOPE:
self.policies.unresolved_scope = True
elif setting_type:
self._overridden.add(setting_type)
continue
for policy in response.get("policies", []):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
self._observed.add(setting_type)
customer_level.append(policy)
scope = self._policy_scope(policy)
if scope != CUSTOMER_SCOPE:
if scope == UNKNOWN_SCOPE:
self.policies.unresolved_scope = True
elif setting_type:
self._overridden.add(setting_type)
continue
for policy in self._effective_policies(customer_level):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
self._process_setting(setting_type, setting.get("value", {}))
self._observed.add(setting_type)
value = setting.get("value", {})
self._process_setting(setting_type, value)
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
f"fetching policies with filter {filter_str}",
self.provider.identity.customer_id,
)
return False
except Exception as error:
self._handle_api_error(
error,
f"listing policies with filter {filter_str}",
self.provider.identity.customer_id,
)
return False
return fetch_succeeded
return fetch_succeeded and succeeded
def _process_setting(self, setting_type: str, value: dict):
"""Process a single policy setting and populate the model."""
@@ -30,40 +30,22 @@ class Sites(GoogleWorkspaceService):
logger.error("Failed to build Cloud Identity service")
return
request = service.policies().list(
pageSize=100,
filter='setting.type.matches("sites.*")',
policies, fetch_succeeded = self._list_policies(
service,
'setting.type.matches("sites.*")',
"fetching Sites policies",
)
fetch_succeeded = True
while request is not None:
try:
response = request.execute()
for policy in self._effective_policies(policies):
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
value = setting.get("value", {})
for policy in response.get("policies", []):
if not self._is_customer_level_policy(policy):
continue
setting = policy.get("setting", {})
setting_type = setting.get("type", "").removeprefix("settings/")
value = setting.get("value", {})
if setting_type == "sites.service_status":
self.policies.service_state = value.get("serviceState")
logger.debug(
"Sites service state: " f"{self.policies.service_state}"
)
request = service.policies().list_next(request, response)
except Exception as error:
self._handle_api_error(
error,
"fetching Sites policies",
self.provider.identity.customer_id,
if setting_type == "sites.service_status":
self.policies.service_state = value.get("serviceState")
logger.debug(
"Sites service state: " f"{self.policies.service_state}"
)
fetch_succeeded = False
break
self.policies_fetched = fetch_succeeded
@@ -7,6 +7,7 @@ from prowler.providers.googleworkspace.lib.service.service import (
OVERRIDE_SCOPE,
UNKNOWN_SCOPE,
GoogleWorkspaceService,
ServicePolicies,
)
ROOT_OU_ID = "03ph8a2z1234"
@@ -125,3 +126,236 @@ class TestPolicyScope:
svc._policy_scope({"policyQuery": {"group": "groups/xyz"}})
== OVERRIDE_SCOPE
)
def _policy(
setting_type,
value,
sort_order=None,
org_unit=f"orgUnits/{ROOT_OU_ID}",
group=None,
):
"""Build a policy as the Cloud Identity Policy API returns it."""
policy = {"setting": {"type": f"settings/{setting_type}", "value": value}}
query = {}
if org_unit:
query["orgUnit"] = org_unit
if group:
query["group"] = group
if sort_order is not None:
query["sortOrder"] = sort_order
if query:
policy["policyQuery"] = query
return policy
class TestEffectivePolicies:
def test_the_greatest_sort_order_wins_whatever_the_api_order(self):
"""The ADMIN policy saved in the console outranks Google's SYSTEM defaults, even when the API returns it first"""
admin = _policy("security.password", {"minimumLength": 12}, 201.00131)
default_a = _policy("security.password", {"minimumLength": 8}, 100.00001)
default_b = _policy("security.password", {"minimumLength": 8}, 100.00003)
svc = _make_service()
assert svc._effective_policies([admin, default_a, default_b]) == [admin]
assert svc._effective_policies([default_a, default_b, admin]) == [admin]
def test_each_setting_type_is_reduced_on_its_own(self):
password_admin = _policy("security.password", {"minimumLength": 12}, 201.1)
password_default = _policy("security.password", {"minimumLength": 8}, 100.1)
session_default = _policy(
"security.session_controls", {"webSessionDuration": "1209600s"}, 100.3
)
assert _make_service()._effective_policies(
[password_admin, password_default, session_default]
) == [password_admin, session_default]
def test_a_single_policy_is_kept_as_is(self):
policy = _policy("sites.service_status", {"serviceState": "ENABLED"})
assert _make_service()._effective_policies([policy]) == [policy]
def test_without_a_sort_order_the_last_policy_wins(self):
"""No sort order means no documented precedence, so the API order is kept"""
first = _policy("sites.service_status", {"serviceState": "ENABLED"})
last = _policy("sites.service_status", {"serviceState": "DISABLED"})
assert _make_service()._effective_policies([first, last]) == [last]
def test_policies_outside_the_customer_scope_are_dropped(self):
root = _policy("security.password", {"minimumLength": 8}, 100.1)
sub_ou = _policy(
"security.password", {"minimumLength": 14}, 300.1, org_unit="orgUnits/sub"
)
group = _policy(
"security.password", {"minimumLength": 16}, 400.1, group="groups/g1"
)
assert _make_service()._effective_policies([sub_ou, root, group]) == [root]
def test_rules_are_items_rather_than_values_so_every_one_is_kept(self):
"""Each rule.* policy is a distinct rule, not a competing value for one setting"""
dlp_a = _policy("rule.dlp", {"displayName": "cards", "state": "ACTIVE"}, 1)
dlp_b = _policy("rule.dlp", {"displayName": "ids", "state": "INACTIVE"}, 2)
alert = _policy("rule.system_defined_alerts", {"displayName": "Leaked"}, 1)
assert _make_service()._effective_policies([dlp_a, dlp_b, alert]) == [
dlp_a,
dlp_b,
alert,
]
def test_no_policies(self):
assert _make_service()._effective_policies([]) == []
class TestServiceStatus:
def test_off_at_the_customer_level_with_no_override(self):
status = _policy("chat.service_status", {"serviceState": "DISABLED"}, 201.1)
other = _policy("chat.chat_apps_access", {"enableApps": True}, 100.1)
assert _make_service()._service_status(
[other, status], "chat.service_status"
) == (
"DISABLED",
False,
)
def test_the_greatest_sort_order_gives_the_state(self):
default = _policy("chat.service_status", {"serviceState": "ENABLED"}, 100.1)
admin = _policy("chat.service_status", {"serviceState": "DISABLED"}, 201.1)
assert _make_service()._service_status(
[admin, default], "chat.service_status"
) == (
"DISABLED",
False,
)
def test_an_org_unit_or_group_with_its_own_status_is_an_override(self):
root = _policy("chat.service_status", {"serviceState": "DISABLED"}, 201.1)
sub_ou = _policy(
"chat.service_status",
{"serviceState": "ENABLED"},
301.1,
org_unit="orgUnits/sub",
)
group = _policy(
"chat.service_status", {"serviceState": "ENABLED"}, 401.1, group="groups/g1"
)
svc = _make_service()
assert svc._service_status([root, sub_ou], "chat.service_status") == (
"DISABLED",
True,
)
assert svc._service_status([root, group], "chat.service_status") == (
"DISABLED",
True,
)
def test_a_sub_org_unit_that_is_also_off_is_not_an_override(self):
"""An org unit that turns the service off as well changes nothing for anyone"""
root = _policy("chat.service_status", {"serviceState": "DISABLED"}, 201.1)
sub_ou = _policy(
"chat.service_status",
{"serviceState": "DISABLED"},
301.1,
org_unit="orgUnits/sub",
)
assert _make_service()._service_status(
[root, sub_ou], "chat.service_status"
) == (
"DISABLED",
False,
)
def test_an_unresolved_scope_counts_as_an_override(self):
"""Without the root OU id a scoped policy may be the root or a sub-OU; neither may be assumed"""
status = _policy("chat.service_status", {"serviceState": "DISABLED"}, 201.1)
assert _make_service(root_org_unit_id=None)._service_status(
[status], "chat.service_status"
) == (None, True)
def test_no_status_policy(self):
other = _policy("chat.chat_apps_access", {"enableApps": True}, 100.1)
assert _make_service()._service_status([other], "chat.service_status") == (
None,
False,
)
class TestServicePolicies:
@pytest.mark.parametrize(
"state, overridden, expected",
[
("DISABLED", False, True),
("DISABLED", True, False),
("ENABLED", False, False),
(None, False, False),
],
)
def test_off_for_everyone(self, state, overridden, expected):
policies = ServicePolicies(
service_state=state, service_state_overridden=overridden
)
assert policies.off_for_everyone is expected
class TestListPolicies:
def test_collects_every_page(self):
svc = _make_service()
page_1 = {"policies": [_policy("gmail.pop_access", {"enablePopAccess": True})]}
page_2 = {
"policies": [_policy("gmail.imap_access", {"enableImapAccess": True})]
}
request_1, request_2 = MagicMock(), MagicMock()
request_1.execute.return_value = page_1
request_2.execute.return_value = page_2
service = MagicMock()
service.policies().list.return_value = request_1
service.policies().list_next.side_effect = [request_2, None]
policies, succeeded = svc._list_policies(
service, 'setting.type.matches("gmail.*")', "fetching Gmail policies"
)
assert succeeded is True
assert policies == page_1["policies"] + page_2["policies"]
service.policies().list.assert_called_once_with(
pageSize=100, filter='setting.type.matches("gmail.*")'
)
def test_a_failing_page_keeps_what_was_read_and_reports_the_failure(self):
svc = _make_service()
page_1 = {"policies": [_policy("gmail.pop_access", {"enablePopAccess": True})]}
request_1, request_2 = MagicMock(), MagicMock()
request_1.execute.return_value = page_1
request_2.execute.side_effect = Exception("boom")
service = MagicMock()
service.policies().list.return_value = request_1
service.policies().list_next.side_effect = [request_2, None]
policies, succeeded = svc._list_policies(
service, 'setting.type.matches("gmail.*")', "fetching Gmail policies"
)
assert succeeded is False
assert policies == page_1["policies"]
def test_a_failing_list_call_reports_the_failure(self):
svc = _make_service()
service = MagicMock()
service.policies().list.side_effect = Exception("boom")
policies, succeeded = svc._list_policies(
service, 'setting.type.matches("gmail.*")', "fetching Gmail policies"
)
assert succeeded is False
assert policies == []
@@ -97,6 +97,59 @@ class TestAdditionalServicesService:
assert additional_services.policies_fetched is True
assert additional_services.policies.groups_service_state == "ENABLED"
def test_the_admin_policy_outranks_a_system_default_returned_after_it(self):
"""The greatest sortOrder is the value in force, whatever order the API lists them in"""
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": [
{
"type": "ADMIN",
"policyQuery": {"sortOrder": 201.00131},
"setting": {
"type": "settings/groups.service_status",
"value": {"serviceState": "DISABLED"},
},
},
{
"type": "SYSTEM",
"policyQuery": {"sortOrder": 100.00002},
"setting": {
"type": "settings/groups.service_status",
"value": {"serviceState": "ENABLED"},
},
},
]
}
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.additionalservices.additionalservices_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.additionalservices.additionalservices_service import (
AdditionalServices,
)
additional_services = AdditionalServices(mock_provider)
assert additional_services.policies_fetched is True
assert additional_services.policies.groups_service_state == "DISABLED"
def test_fetch_policies_empty_response(self):
"""Test handling empty policies response"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -5,6 +5,7 @@ from prowler.providers.googleworkspace.services.calendar.calendar_service import
)
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -102,6 +103,38 @@ class TestCalendarExternalInvitationsWarning:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning.calendar_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_invitations_warning.calendar_external_invitations_warning import (
calendar_external_invitations_warning,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = CalendarPolicies(
service_state="DISABLED", external_invitations_warning=False
)
findings = calendar_external_invitations_warning().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Calendar is turned off for everyone in domain {DOMAIN}, "
f"so no invitation can be sent."
)
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -5,6 +5,7 @@ from prowler.providers.googleworkspace.services.calendar.calendar_service import
)
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -140,6 +141,39 @@ class TestCalendarExternalSharingPrimaryCalendar:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar.calendar_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_primary_calendar.calendar_external_sharing_primary_calendar import (
calendar_external_sharing_primary_calendar,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = CalendarPolicies(
service_state="DISABLED",
primary_calendar_external_sharing="EXTERNAL_ALL_INFO_READ_WRITE",
)
findings = calendar_external_sharing_primary_calendar().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Calendar is turned off for everyone in domain {DOMAIN}, "
f"so no calendar can be shared outside the organization."
)
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -5,6 +5,7 @@ from prowler.providers.googleworkspace.services.calendar.calendar_service import
)
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -133,6 +134,39 @@ class TestCalendarExternalSharingSecondaryCalendar:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar.calendar_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.calendar.calendar_external_sharing_secondary_calendar.calendar_external_sharing_secondary_calendar import (
calendar_external_sharing_secondary_calendar,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = CalendarPolicies(
service_state="DISABLED",
secondary_calendar_external_sharing="EXTERNAL_ALL_INFO_READ_WRITE",
)
findings = calendar_external_sharing_secondary_calendar().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Calendar is turned off for everyone in domain {DOMAIN}, "
f"so no calendar can be shared outside the organization."
)
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -1,6 +1,7 @@
from unittest.mock import MagicMock, patch
from tests.providers.googleworkspace.googleworkspace_fixtures import (
ROOT_ORG_UNIT_ID,
set_mocked_googleworkspace_provider,
)
@@ -77,6 +78,171 @@ class TestCalendarService:
)
assert calendar.policies.external_invitations_warning is True
def test_the_admin_policy_outranks_a_system_default_returned_after_it(self):
"""The greatest sortOrder is the value in force, whatever order the API lists them in"""
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": [
{
"type": "ADMIN",
"policyQuery": {"sortOrder": 201.00131},
"setting": {
"type": "settings/calendar.secondary_calendar_max_allowed_external_sharing",
"value": {
"maxAllowedExternalSharing": "EXTERNAL_FREE_BUSY_ONLY"
},
},
},
{
"type": "SYSTEM",
"policyQuery": {"sortOrder": 100.00004},
"setting": {
"type": "settings/calendar.secondary_calendar_max_allowed_external_sharing",
"value": {
"maxAllowedExternalSharing": "EXTERNAL_ALL_INFO_READ_ONLY"
},
},
},
]
}
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.calendar.calendar_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
Calendar,
)
calendar = Calendar(mock_provider)
assert calendar.policies_fetched is True
assert (
calendar.policies.secondary_calendar_external_sharing
== "EXTERNAL_FREE_BUSY_ONLY"
)
def test_off_for_everyone_when_off_at_the_root_with_no_override(self):
"""A root-level DISABLED status with no org unit or group override is off for everyone"""
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": [
{
"type": "ADMIN",
"policyQuery": {
"orgUnit": f"orgUnits/{ROOT_ORG_UNIT_ID}",
"sortOrder": 201.00122,
},
"setting": {
"type": "settings/calendar.service_status",
"value": {"serviceState": "DISABLED"},
},
},
]
}
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.calendar.calendar_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
Calendar,
)
fetched = Calendar(mock_provider)
assert fetched.policies_fetched is True
assert fetched.policies.service_state == "DISABLED"
assert fetched.policies.service_state_overridden is False
assert fetched.policies.off_for_everyone is True
def test_reads_the_service_status_and_its_overrides(self):
"""Off at the customer level only counts when no org unit or group turns the service back on"""
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": [
{
"type": "ADMIN",
"policyQuery": {"sortOrder": 201.00122},
"setting": {
"type": "settings/calendar.service_status",
"value": {"serviceState": "DISABLED"},
},
},
{
"type": "ADMIN",
"policyQuery": {"orgUnit": "orgUnits/sub", "sortOrder": 301.00122},
"setting": {
"type": "settings/calendar.service_status",
"value": {"serviceState": "ENABLED"},
},
},
]
}
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.calendar.calendar_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.calendar.calendar_service import (
Calendar,
)
fetched = Calendar(mock_provider)
assert fetched.policies_fetched is True
assert fetched.policies.service_state == "DISABLED"
assert fetched.policies.service_state_overridden is True
assert fetched.policies.off_for_everyone is False
def test_calendar_fetch_policies_empty_response(self):
"""Test handling empty policies response"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.chat.chat_service import ChatPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -93,6 +94,69 @@ class TestChatAppsInstallationDisabled:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.chat.chat_apps_installation_disabled.chat_apps_installation_disabled.chat_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.chat.chat_apps_installation_disabled.chat_apps_installation_disabled import (
chat_apps_installation_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = ChatPolicies(
service_state="DISABLED", enable_apps=True
)
findings = chat_apps_installation_disabled().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Chat is turned off for everyone in domain {DOMAIN}, "
f"so Chat apps cannot be installed."
)
def test_evaluates_the_setting_when_an_override_turns_the_service_on(self):
"""Off at the customer level with an org unit turned back on is not off for everyone"""
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.chat.chat_apps_installation_disabled.chat_apps_installation_disabled.chat_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.chat.chat_apps_installation_disabled.chat_apps_installation_disabled import (
chat_apps_installation_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = ChatPolicies(
service_state="DISABLED",
service_state_overridden=True,
enable_apps=True,
)
findings = chat_apps_installation_disabled().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.chat.chat_service import ChatPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -123,6 +124,38 @@ class TestChatExternalFileSharingDisabled:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.chat.chat_external_file_sharing_disabled.chat_external_file_sharing_disabled.chat_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.chat.chat_external_file_sharing_disabled.chat_external_file_sharing_disabled import (
chat_external_file_sharing_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = ChatPolicies(
service_state="DISABLED", external_file_sharing="ALL_FILES"
)
findings = chat_external_file_sharing_disabled().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Chat is turned off for everyone in domain {DOMAIN}, "
f"so no files can be shared externally through Chat."
)
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.chat.chat_service import ChatPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -128,6 +129,40 @@ class TestChatExternalMessagingRestricted:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.chat.chat_external_messaging_restricted.chat_external_messaging_restricted.chat_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.chat.chat_external_messaging_restricted.chat_external_messaging_restricted import (
chat_external_messaging_restricted,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = ChatPolicies(
service_state="DISABLED",
allow_external_chat=True,
external_chat_restriction="NO_RESTRICTION",
)
findings = chat_external_messaging_restricted().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Chat is turned off for everyone in domain {DOMAIN}, "
f"so no external messaging can take place."
)
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.chat.chat_service import ChatPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -129,6 +130,40 @@ class TestChatExternalSpacesRestricted:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.chat.chat_external_spaces_restricted.chat_external_spaces_restricted.chat_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.chat.chat_external_spaces_restricted.chat_external_spaces_restricted import (
chat_external_spaces_restricted,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = ChatPolicies(
service_state="DISABLED",
external_spaces_enabled=True,
external_spaces_domain_allowlist_mode="ALL_DOMAINS",
)
findings = chat_external_spaces_restricted().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Chat is turned off for everyone in domain {DOMAIN}, "
f"so no external spaces can be created or joined."
)
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.chat.chat_service import ChatPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -93,6 +94,38 @@ class TestChatIncomingWebhooksDisabled:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.chat.chat_incoming_webhooks_disabled.chat_incoming_webhooks_disabled.chat_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.chat.chat_incoming_webhooks_disabled.chat_incoming_webhooks_disabled import (
chat_incoming_webhooks_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = ChatPolicies(
service_state="DISABLED", enable_webhooks=True
)
findings = chat_incoming_webhooks_disabled().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Chat is turned off for everyone in domain {DOMAIN}, "
f"so incoming webhooks cannot be used."
)
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.chat.chat_service import ChatPolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -96,6 +97,38 @@ class TestChatInternalFileSharingDisabled:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.chat.chat_internal_file_sharing_disabled.chat_internal_file_sharing_disabled.chat_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.chat.chat_internal_file_sharing_disabled.chat_internal_file_sharing_disabled import (
chat_internal_file_sharing_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = ChatPolicies(
service_state="DISABLED", internal_file_sharing="ALL_FILES"
)
findings = chat_internal_file_sharing_disabled().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Chat is turned off for everyone in domain {DOMAIN}, "
f"so no files can be shared through Chat."
)
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
@@ -93,6 +93,165 @@ class TestChatService:
assert chat.policies.enable_apps is False
assert chat.policies.enable_webhooks is False
def test_the_admin_policy_outranks_a_system_default_returned_after_it(self):
"""The greatest sortOrder is the value in force, whatever order the API lists them in"""
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": [
{
"type": "ADMIN",
"policyQuery": {"sortOrder": 201.00131},
"setting": {
"type": "settings/chat.chat_apps_access",
"value": {"enableApps": False, "enableWebhooks": False},
},
},
{
"type": "SYSTEM",
"policyQuery": {"sortOrder": 100.00005},
"setting": {
"type": "settings/chat.chat_apps_access",
"value": {"enableApps": True, "enableWebhooks": 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.chat.chat_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.chat.chat_service import (
Chat,
)
chat = Chat(mock_provider)
assert chat.policies_fetched is True
assert chat.policies.enable_apps is False
assert chat.policies.enable_webhooks is False
def test_off_for_everyone_when_off_at_the_root_with_no_override(self):
"""A root-level DISABLED status with no org unit or group override is off for everyone"""
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": [
{
"type": "ADMIN",
"policyQuery": {
"orgUnit": f"orgUnits/{ROOT_ORG_UNIT_ID}",
"sortOrder": 201.00122,
},
"setting": {
"type": "settings/chat.service_status",
"value": {"serviceState": "DISABLED"},
},
},
]
}
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.chat.chat_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.chat.chat_service import (
Chat,
)
fetched = Chat(mock_provider)
assert fetched.policies_fetched is True
assert fetched.policies.service_state == "DISABLED"
assert fetched.policies.service_state_overridden is False
assert fetched.policies.off_for_everyone is True
def test_reads_the_service_status_and_its_overrides(self):
"""Off at the customer level only counts when no org unit or group turns the service back on"""
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": [
{
"type": "ADMIN",
"policyQuery": {"sortOrder": 201.00122},
"setting": {
"type": "settings/chat.service_status",
"value": {"serviceState": "DISABLED"},
},
},
{
"type": "ADMIN",
"policyQuery": {"orgUnit": "orgUnits/sub", "sortOrder": 301.00122},
"setting": {
"type": "settings/chat.service_status",
"value": {"serviceState": "ENABLED"},
},
},
]
}
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.chat.chat_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.chat.chat_service import (
Chat,
)
fetched = Chat(mock_provider)
assert fetched.policies_fetched is True
assert fetched.policies.service_state == "DISABLED"
assert fetched.policies.service_state_overridden is True
assert fetched.policies.off_for_everyone is False
def test_chat_fetch_policies_empty_response(self):
"""Test handling empty policies response"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.drive.drive_service import DrivePolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -128,6 +129,39 @@ class TestDriveAccessCheckerRecipientsOnly:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.drive.drive_access_checker_recipients_only.drive_access_checker_recipients_only.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_access_checker_recipients_only.drive_access_checker_recipients_only import (
drive_access_checker_recipients_only,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED",
access_checker_suggestions="RECIPIENTS_OR_AUDIENCE_OR_PUBLIC",
)
findings = drive_access_checker_recipients_only().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Drive is turned off for everyone in domain {DOMAIN}, "
f"so Access Checker cannot widen any share."
)
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.drive.drive_service import DrivePolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -94,6 +95,38 @@ class TestDriveDesktopAccessDisabled:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.drive.drive_desktop_access_disabled.drive_desktop_access_disabled.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_desktop_access_disabled.drive_desktop_access_disabled import (
drive_desktop_access_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED", allow_drive_for_desktop=True
)
findings = drive_desktop_access_disabled().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Drive is turned off for everyone in domain {DOMAIN}, "
f"so Drive for desktop cannot sync any content."
)
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -99,6 +99,35 @@ class TestDriveExternalSharingWarnUsers:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_evaluates_the_setting_when_the_service_is_off_for_everyone(self):
"""Existing shares stay reachable when Drive is off, so the stored setting still counts"""
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.drive.drive_external_sharing_warn_users.drive_external_sharing_warn_users.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_external_sharing_warn_users.drive_external_sharing_warn_users import (
drive_external_sharing_warn_users,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED", warn_for_external_sharing=False
)
findings = drive_external_sharing_warn_users().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -130,6 +130,36 @@ class TestDriveInternalUsersDistributeContent:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_evaluates_the_setting_when_the_service_is_off_for_everyone(self):
"""Existing shares stay reachable when Drive is off, so the stored setting still counts"""
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.drive.drive_internal_users_distribute_content.drive_internal_users_distribute_content.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_internal_users_distribute_content.drive_internal_users_distribute_content import (
drive_internal_users_distribute_content,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED",
allowed_parties_for_distributing_content="ALL_ELIGIBLE_USERS",
)
findings = drive_internal_users_distribute_content().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -94,6 +94,35 @@ class TestDrivePublishingFilesDisabled:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_evaluates_the_setting_when_the_service_is_off_for_everyone(self):
"""Existing shares stay reachable when Drive is off, so the stored setting still counts"""
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.drive.drive_publishing_files_disabled.drive_publishing_files_disabled.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_publishing_files_disabled.drive_publishing_files_disabled import (
drive_publishing_files_disabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED", allow_publishing_files=True
)
findings = drive_publishing_files_disabled().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -1,6 +1,7 @@
from unittest.mock import MagicMock, patch
from tests.providers.googleworkspace.googleworkspace_fixtures import (
ROOT_ORG_UNIT_ID,
set_mocked_googleworkspace_provider,
)
@@ -92,6 +93,164 @@ class TestDriveService:
)
assert drive.policies.allow_drive_for_desktop is False
def test_the_admin_policy_outranks_a_system_default_returned_after_it(self):
"""The greatest sortOrder is the value in force, whatever order the API lists them in"""
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": [
{
"type": "ADMIN",
"policyQuery": {"sortOrder": 201.00131},
"setting": {
"type": "settings/drive_and_docs.external_sharing",
"value": {"externalSharingMode": "ALLOWLISTED_DOMAINS"},
},
},
{
"type": "SYSTEM",
"policyQuery": {"sortOrder": 100.0001},
"setting": {
"type": "settings/drive_and_docs.external_sharing",
"value": {"externalSharingMode": "ALLOWED"},
},
},
]
}
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.drive.drive_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.drive.drive_service import (
Drive,
)
drive = Drive(mock_provider)
assert drive.policies_fetched is True
assert drive.policies.external_sharing_mode == "ALLOWLISTED_DOMAINS"
def test_off_for_everyone_when_off_at_the_root_with_no_override(self):
"""A root-level DISABLED status with no org unit or group override is off for everyone"""
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": [
{
"type": "ADMIN",
"policyQuery": {
"orgUnit": f"orgUnits/{ROOT_ORG_UNIT_ID}",
"sortOrder": 201.00122,
},
"setting": {
"type": "settings/drive_and_docs.service_status",
"value": {"serviceState": "DISABLED"},
},
},
]
}
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.drive.drive_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.drive.drive_service import (
Drive,
)
fetched = Drive(mock_provider)
assert fetched.policies_fetched is True
assert fetched.policies.service_state == "DISABLED"
assert fetched.policies.service_state_overridden is False
assert fetched.policies.off_for_everyone is True
def test_reads_the_service_status_and_its_overrides(self):
"""Off at the customer level only counts when no org unit or group turns the service back on"""
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": [
{
"type": "ADMIN",
"policyQuery": {"sortOrder": 201.00122},
"setting": {
"type": "settings/drive_and_docs.service_status",
"value": {"serviceState": "DISABLED"},
},
},
{
"type": "ADMIN",
"policyQuery": {"orgUnit": "orgUnits/sub", "sortOrder": 301.00122},
"setting": {
"type": "settings/drive_and_docs.service_status",
"value": {"serviceState": "ENABLED"},
},
},
]
}
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.drive.drive_service.GoogleWorkspaceService._build_service",
return_value=mock_service,
),
):
from prowler.providers.googleworkspace.services.drive.drive_service import (
Drive,
)
fetched = Drive(mock_provider)
assert fetched.policies_fetched is True
assert fetched.policies.service_state == "DISABLED"
assert fetched.policies.service_state_overridden is True
assert fetched.policies.off_for_everyone is False
def test_drive_fetch_policies_empty_response(self):
"""Test handling empty policies response"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ from unittest.mock import patch
from prowler.providers.googleworkspace.services.drive.drive_service import DrivePolicies
from tests.providers.googleworkspace.googleworkspace_fixtures import (
CUSTOMER_ID,
DOMAIN,
set_mocked_googleworkspace_provider,
)
@@ -96,6 +97,38 @@ class TestDriveSharedDriveCreationAllowed:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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.drive.drive_shared_drive_creation_allowed.drive_shared_drive_creation_allowed.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_shared_drive_creation_allowed.drive_shared_drive_creation_allowed import (
drive_shared_drive_creation_allowed,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED", allow_shared_drive_creation=False
)
findings = drive_shared_drive_creation_allowed().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Google Drive is turned off for everyone in domain {DOMAIN}, "
f"so no shared drive can be created."
)
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -130,6 +130,35 @@ class TestDriveSharedDriveDisableDownloadPrintCopy:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_evaluates_the_setting_when_the_service_is_off_for_everyone(self):
"""Existing shares stay reachable when Drive is off, so the stored setting still counts"""
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.drive.drive_shared_drive_disable_download_print_copy.drive_shared_drive_disable_download_print_copy.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_shared_drive_disable_download_print_copy.drive_shared_drive_disable_download_print_copy import (
drive_shared_drive_disable_download_print_copy,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED", allowed_parties_for_download_print_copy="ALL"
)
findings = drive_shared_drive_disable_download_print_copy().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -100,6 +100,35 @@ class TestDriveSharedDriveManagersCannotOverride:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_evaluates_the_setting_when_the_service_is_off_for_everyone(self):
"""Existing shares stay reachable when Drive is off, so the stored setting still counts"""
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.drive.drive_shared_drive_managers_cannot_override.drive_shared_drive_managers_cannot_override.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_shared_drive_managers_cannot_override.drive_shared_drive_managers_cannot_override import (
drive_shared_drive_managers_cannot_override,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED", allow_managers_to_override_settings=True
)
findings = drive_shared_drive_managers_cannot_override().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -94,6 +94,35 @@ class TestDriveSharedDriveMembersOnlyAccess:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_evaluates_the_setting_when_the_service_is_off_for_everyone(self):
"""Existing shares stay reachable when Drive is off, so the stored setting still counts"""
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.drive.drive_shared_drive_members_only_access.drive_shared_drive_members_only_access.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_shared_drive_members_only_access.drive_shared_drive_members_only_access import (
drive_shared_drive_members_only_access,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED", allow_non_member_access=True
)
findings = drive_shared_drive_members_only_access().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -126,6 +126,35 @@ class TestDriveSharingAllowlistedDomains:
assert findings[0].status == "FAIL"
assert "not explicitly configured" in findings[0].status_extended
def test_evaluates_the_setting_when_the_service_is_off_for_everyone(self):
"""Existing shares stay reachable when Drive is off, so the stored setting still counts"""
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.drive.drive_sharing_allowlisted_domains.drive_sharing_allowlisted_domains.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_sharing_allowlisted_domains.drive_sharing_allowlisted_domains import (
drive_sharing_allowlisted_domains,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED", external_sharing_mode="ALLOWED"
)
findings = drive_sharing_allowlisted_domains().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -100,6 +100,36 @@ class TestDriveWarnSharingWithAllowlistedDomains:
assert findings[0].status == "PASS"
assert "secure default" in findings[0].status_extended
def test_evaluates_the_setting_when_the_service_is_off_for_everyone(self):
"""Existing shares stay reachable when Drive is off, so the stored setting still counts"""
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.drive.drive_warn_sharing_with_allowlisted_domains.drive_warn_sharing_with_allowlisted_domains.drive_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.drive.drive_warn_sharing_with_allowlisted_domains.drive_warn_sharing_with_allowlisted_domains import (
drive_warn_sharing_with_allowlisted_domains,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = DrivePolicies(
service_state="DISABLED",
warn_for_sharing_outside_allowlisted_domains=False,
)
findings = drive_warn_sharing_with_allowlisted_domains().execute()
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "turned off" not in findings[0].status_extended
def test_no_findings_when_fetch_failed(self):
"""Test no findings returned when the API fetch failed"""
mock_provider = set_mocked_googleworkspace_provider()
@@ -3,6 +3,7 @@ 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,
)
@@ -127,6 +128,38 @@ class TestGmailAnomalousAttachmentProtectionEnabled:
assert findings[0].status == "FAIL"
assert "insecure default" in findings[0].status_extended
def test_pass_when_the_service_is_off_for_everyone(self):
"""A setting that only acts while the service is on has no effect once the service is off"""
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_anomalous_attachment_protection_enabled.gmail_anomalous_attachment_protection_enabled.gmail_client"
) as mock_client,
):
from prowler.providers.googleworkspace.services.gmail.gmail_anomalous_attachment_protection_enabled.gmail_anomalous_attachment_protection_enabled import (
gmail_anomalous_attachment_protection_enabled,
)
mock_client.provider = mock_provider
mock_client.policies_fetched = True
mock_client.policies = GmailPolicies(
service_state="DISABLED", enable_anomalous_attachment_protection=False
)
findings = gmail_anomalous_attachment_protection_enabled().execute()
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].status_extended == (
f"Gmail is turned off for everyone in domain {DOMAIN}, "
f"so no user mailbox receives mail through Gmail."
)
def test_no_findings_when_fetch_failed(self):
mock_provider = set_mocked_googleworkspace_provider()
Loaded 100 of 124 files, more files were not shown because too many files have changed in this diff. Show more