Panther rules: gsuite

import datetime

from panther_base_helpers import pattern_match, pattern_match_list

COMPANY_DOMAIN = "your-company-name.com"
EXCEPTION_PATTERNS = {
    # The glob pattern for the document title (lowercased)
    "1 document title p*": {  # allow any title "all"
        "allowed_to_send": {
            "alice@acme.com",
            "samuel@acme.com",
            "nathan@acme.com",
            "barry@acme.com",
            # Allow any user
            # "all"
            # Allow any user in a specific domain
            # "*@acme.com"
        },
        "allowed_to_receive": {
            "alice@abc.com",
            "samuel@abc.com",
            "nathan@abc.com",
            "barry@abc.com",
            # Allow any user
            # "all"
            # Allow any user in a specific domain
            # "*@acme.com"
        },
        # The time limit for how long the file share stays valid
        "allowed_until": datetime.datetime(year=2030, month=6, day=2),
    },
    "2 document title p*": {
        "allowed_to_send": {
            "alice@abc.com",
        },
        "allowed_to_receive": {
            "*@acme.com",
        },
        # The time limit for how long the file share stays valid
        "allowed_until": datetime.datetime(year=2030, month=6, day=2),
    },
}


def _check_acl_change_event(actor_email, event):
    # For GSuite.ActivityEvent, parameters is a dict, not an array
    parameters = event.get("parameters", {})

    doc_title = parameters.get("doc_title", "TITLE_UNKNOWN")
    old_visibility = parameters.get("old_visibility", "OLD_VISIBILITY_UNKNOWN")
    new_visibility = parameters.get("visibility", "NEW_VISIBILITY_UNKNOWN")
    target_user = parameters.get("target_user") or parameters.get("target_domain") or "USER_UNKNOWN"
    current_time = datetime.datetime.now()

    if (
        new_visibility == "shared_externally"
        and old_visibility == "private"
        and not target_user.endswith(f"@{COMPANY_DOMAIN}")
    ):
        # This is a dangerous share, check exceptions:

        for pattern, details in EXCEPTION_PATTERNS.items():
            proper_title = pattern_match(doc_title.lower(), pattern) or pattern == "all"

            proper_sender = pattern_match_list(
                actor_email, details.get("allowed_to_send")
            ) or details.get("allowed_to_send") == {"all"}

            proper_receiver = pattern_match_list(
                target_user, details.get("allowed_to_receive")
            ) or details.get("allowed_to_receive") == {"all"}

            if (
                proper_title
                and proper_sender
                and proper_receiver
                and current_time < details.get("allowed_until")
            ):
                return False
        # No exceptions match.
        # Return the event summary (which is True) to alert & use in title.
        return {
            "actor": actor_email,
            "doc_title": doc_title,
            "target_user": target_user,
        }
    return False


def rule(event):
    application_name = event.deep_get("id", "applicationName")
    actor_email = event.deep_get("actor", "email", default="EMAIL_UNKNOWN")

    # For GSuite.ActivityEvent, each log is a single event (no events array)
    if application_name == "drive" and event.get("type") == "acl_change":
        # If this event is a dangerous file share, alert:
        return bool(_check_acl_change_event(actor_email, event))
    return False


def title(event):
    actor_email = event.deep_get("actor", "email", default="EMAIL_UNKNOWN")
    matching_event = _check_acl_change_event(actor_email, event)

    if matching_event:
        actor = matching_event.get("actor", "ACTOR_UNKNOWN")
        doc_title = matching_event.get("doc_title", "DOC_TITLE_UNKNOWN")
        target_user = matching_event.get("target_user", "USER_UNKNOWN")
        return f'Dangerous file share by [{actor}]: "{doc_title}" to {target_user}'
    return "No matching events, but DangerousShares still fired"

AnalysisType: rule
Filename: gsuite_drive_external_share.py
RuleID: "GSuite.Drive.ExternalFileShare"
DisplayName: "External GSuite File Share"
Enabled: false
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
  - Security Control
  - Configuration Required
  - Collection:Data from Information Repositories
Reports:
  MITRE ATT&CK:
    - TA0009:T1213
Severity: High
Description: An employee shared a sensitive file externally with another organization
Runbook: |
  Contact the employee who made the share and make sure they redact the access.
  If the share was legitimate, add to the EXCEPTION_PATTERNS in the detection.
Reference: https://support.google.com/docs/answer/2494822?hl=en&co=GENIE.Platform%3DiOS&sjid=864417124752637253-EU

from panther_gsuite_helpers import gsuite_activityevent_alert_context

# SMTP response reasons that indicate security threats
MALICIOUS_SMTP_RESPONSES = {
    3: "Malware",
    13: "Blatant Spam",
    14: "Denial of Service",
    15: "Malicious or Spam Links",
    16: "Low IP Reputation",
    17: "Low Domain Reputation",
    18: "IP address listed in public real-time block list",
}


def rule(event):
    if event.deep_get("id", "applicationName", default="<UNKNOWN_APPLICATION>") != "gmail":
        return False

    smtp_response_reason = event.deep_get(
        "parameters", "message_info", "connection_info", "smtp_response_reason", default=0
    )

    return smtp_response_reason in MALICIOUS_SMTP_RESPONSES


def title(event):
    user = event.deep_get("actor", "email", default="<UNKNOWN_USER>")
    smtp_response_reason = event.deep_get(
        "parameters", "message_info", "connection_info", "smtp_response_reason", default=0
    )
    reason_description = MALICIOUS_SMTP_RESPONSES.get(
        smtp_response_reason, f"Unknown ({smtp_response_reason})"
    )

    sender = event.deep_get(
        "parameters", "message_info", "source", "address", default="<UNKNOWN_SENDER>"
    )

    return f"Gmail blocked email to [{user}] from [{sender}] due to: {reason_description}"


def alert_context(event):
    context = gsuite_activityevent_alert_context(event)

    # Add specific SMTP response information
    smtp_response_reason = event.deep_get(
        "parameters", "message_info", "connection_info", "smtp_response_reason", default=0
    )

    context.update(
        {
            "smtp_response_reason_code": smtp_response_reason,
            "smtp_response_reason": MALICIOUS_SMTP_RESPONSES.get(
                smtp_response_reason, f"Unknown ({smtp_response_reason})"
            ),
            "smtp_reply_code": event.deep_get(
                "parameters", "message_info", "connection_info", "smtp_reply_code", default=0
            ),
        }
    )

    return context

AnalysisType: rule
Filename: gsuite_malicious_smtp_response.py
RuleID: "GSuite.Gmail.Malicious.SMTP.Response"
DisplayName: "Gmail Malicious SMTP Response"
Enabled: true
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
  - Gmail
  - Email Security
  - Malware
  - Spam
Reports:
  MITRE ATT&CK:
    - TA0001:T1566 # Initial Access: Phishing
Severity: High
Description: >
  Detects when Gmail blocks or rejects emails due to malicious SMTP response reasons including
  malware detection, spam/phishing links, low sender reputation, RBL listings, or denial of service attempts.
  This rule monitors inbound SMTP connections for security threats that Gmail's filters identify.
Reference: https://support.google.com/a/answer/12384955
Runbook: |
  1. Review the sender's email address and domain
  2. Check the SMTP response reason and reply code for details
  3. Investigate the sender's IP address for additional context (geolocation, reputation)
  4. Review authentication status (SPF, DKIM, DMARC)
  5. If malware was detected, check if similar messages were received by other users
  6. Consider adding sender to blocklist if pattern of malicious activity is confirmed
  7. For DoS attempts, review firewall and rate limiting configurations
DedupPeriodMinutes: 60
SummaryAttributes:
  - user_email
  - sender_address
  - smtp_response_reason

from panther_gsuite_helpers import gsuite_activityevent_alert_context


def rule(event):

    if event.deep_get("id", "applicationName", default="<UNKNOWN_APPLICATION>") != "gmail":
        return False

    dmarc_passed = event.deep_get(
        "parameters",
        "message_info",
        "connection_info",
        "dmarc_pass",
        default="<UNKNOWN_DMARC_PASS>",
    )
    spf_passed = event.deep_get(
        "parameters", "message_info", "connection_info", "spf_pass", default="<UNKNOWN_SPF_PASS>"
    )
    dkim_passed = event.deep_get(
        "parameters",
        "message_info",
        "connection_info",
        "dkim_pass",
        default="<UNKNOWN_DKIM_PASS>",
    )

    event_success = event.deep_get("parameters", "event_info", "success", default=False)

    if event_success is True:  # Message was delivered despite failures
        if dmarc_passed is False:
            return True
        if spf_passed is False and dkim_passed is False:
            return True
    return False


def title(event):
    user = event.deep_get("actor", "email", default="<UNKNOWN_USER>")
    return f"[{user}] received a potentially spoofed email"


def alert_context(event):
    context = gsuite_activityevent_alert_context(event)
    return context

AnalysisType: rule
Filename: gsuite_potential_spoofed_email.py
RuleID: "GSuite.Gmail.Potential.Spoofed.Email"
DisplayName: "Gmail Potential Spoofed Email Delivered"
Enabled: true
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
  - Gmail
  - Email Security
  - Spoofing
  - Phishing
Reports:
  MITRE ATT&CK:
    - TA0001:T1566.001 # Initial Access: Phishing - Spearphishing Attachment
    - TA0001:T1566.002 # Initial Access: Phishing - Spearphishing Link
Severity: High
Description: >
  Detects when a potentially spoofed email was successfully delivered to a user's inbox despite
  failing email authentication checks. This rule triggers when:

  1. DMARC authentication fails, OR
  2. Both SPF and DKIM authentication fail simultaneously

  These authentication failures indicate the sender may be impersonating a legitimate domain,
  which is a common tactic in phishing and business email compromise (BEC) attacks.
Reference: https://support.google.com/a/answer/12384955
Runbook: |
  1. Review the sender's email address and compare with the From: header display name
  2. Check if the sender domain is impersonating an internal or partner domain
  3. Verify the authentication status details (SPF, DKIM, DMARC)
  4. Review the message subject and content if available
  5. Check the sender's IP geolocation and reputation
  6. Search for similar messages from the same sender to other users
  7. If confirmed as spoofing:
     - Add sender domain/IP to blocklist
     - Remove the message from user's inbox
     - Notify affected users not to interact with the email
  8. Consider strengthening DMARC policy (quarantine/reject) if not already enforced
DedupPeriodMinutes: 60
SummaryAttributes:
  - user_email
  - sender_address
  - dmarc_pass
  - spf_pass
  - dkim_pass

def rule(event):
    if event.deep_get("id", "applicationName") != "access_transparency":
        return False

    return bool(event.get("type") == "GSUITE_RESOURCE")

AnalysisType: rule
Filename: gsuite_google_access.py
RuleID: "GSuite.GoogleAccess"
DisplayName: "Google Accessed a GSuite Resource"
Enabled: true
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
Severity: Low
Description: >
  Google accessed one of your GSuite resources directly, most likely in response to a support incident.
Reference: https://support.google.com/a/answer/9230474?hl=en
Runbook: >
  Your GSuite Super Admin can visit the Access Transparency report in the GSuite Admin Dashboard to see more details about the access.
SummaryAttributes:
  - actor:email

def rule(_):
    return True


def title(event):
    return (
        f"GSuite: [{event.get('user', '<user_not_found>')}] "
        f"downloaded [{event.get('download_count', '<count_not_found>')}] "
        "files from Google Drive."
    )


def alert_context(event):
    return event.to_dict()

AnalysisType: scheduled_rule
Description: Scheduled rule for the High Google Drive Download Count query which looks for incidents of more than 10 (tunable) downloads by a user in the past day.
DisplayName: "Google Drive High Download Count"
Status: Deprecated
Enabled: false
Filename: gsuite_drive_many_docs_downloaded.py
Reference: https://support.google.com/drive/answer/2423534?hl=en&co=GENIE.Platform%3DDesktop
Severity: Medium
Tags:
  - Deprecated
DedupPeriodMinutes: 60
RuleID: "Google.Drive.High.Download.Count"
Threshold: 1
ScheduledQueries:
  - GSuite Many Docs Downloaded Query

AnalysisType: scheduled_query
QueryName: "Google Workspace Login Type Anomaly"
Description: |
  Detects users authenticating with login types they haven't used in the past 30 days.
  May indicate GAIA credential theft where attackers use stolen tokens with different
  authentication methods than the victim's normal pattern (e.g., google_password instead of SAML).
Enabled: false
Query: |
  -- pragma: template
  {% import 'anomalies' new_unique_values %}

  WITH subquery AS (
    SELECT
      actor:email AS email,
      parameters:login_type AS login_type,
      ipAddress,
      p_event_time
    FROM panther_logs.public.gsuite_activityevent
    WHERE p_occurs_since('30 days')
      AND id:applicationName = 'login'
      AND name = 'login_success'
      AND parameters:login_type IS NOT NULL
      AND parameters:login_type != 'reauth'
  ),
  {{ new_unique_values('subquery', 'email', 'login_type', '1 day') }}
Schedule:
  RateMinutes: 360
  TimeoutMinutes: 3

import re


def normalize_username(email):
    if not email:
        return None
    # Extract username before @ symbol
    username = email.split("@")[0] if "@" in email else email
    # Remove all non-alphanumeric characters and convert to lowercase
    return re.sub(r"[^a-z0-9]", "", username.lower())


def rule(_):
    return True


def title(event):
    user = event.get("email", "<UNKNOWN_USER>")
    login_type = event.get("login_type", "<UNKNOWN_TYPE>")
    return f"Google Workspace: User [{user}] used anomalous login type [{login_type}]"


def severity(event):
    login_type = event.get("login_type", "")

    # Higher severity for password-based auth
    if login_type == "google_password":
        return "HIGH"

    return "MEDIUM"


def alert_context(event):
    email = event.get("email")
    return {
        "user_email": email,
        "username_normalized": normalize_username(email),
        "anomalous_login_type": event.get("login_type"),
        "ip_address": event.get("ipAddress"),
        "description": ("User authenticated with a login type not seen in the previous 30 days"),
    }

AnalysisType: scheduled_rule
DisplayName: "Google Workspace Login Type Anomaly"
DedupPeriodMinutes: 360
RuleID: "Google.Workspace.Login.Type.Anomaly"
Description: >
  Alerts when users authenticate with login types they haven't used in the past 30 days.
  This may indicate GAIA credential theft where attackers use stolen OAuth tokens with
  different authentication methods (e.g., google_password instead of SAML). Particularly
  suspicious when a user who normally uses SSO/SAML suddenly authenticates via password.
ScheduledQueries:
  - Google Workspace Login Type Anomaly
Enabled: false
Filename: gsuite_login_type_anomaly_rule.py
Reference: https://businessinsights.bitdefender.com/the-chain-reaction-new-methods-for-extending-local-breaches-in-google-workspace
Runbook: |
  1. Query GSuite.ActivityEvent for all login events by the user email in the 24 hours before and after the alert to identify login patterns, IP addresses used, and the context around the anomalous login_type
  2. Check if the source IP addresses from recent logins are associated with known VPN services, cloud providers, or corporate network ranges, and compare to the user's typical login locations in the past 30 days
  3. Search for other authentication anomalies or alerts for this user in the past 7 days, including failed logins, OAuth token authorizations, password changes, or suspicious activity warnings
Severity: Medium
Tags:
  - GSuite
  - Lateral Movement
  - Valid Accounts
  - GAIA
Reports:
  MITRE ATT&CK:
    - TA0008:T1078.004
    - TA0006:T1550
SummaryAttributes:
  - email

AnalysisType: scheduled_query
QueryName: "Google Workspace OAuth Anomalous Privileged Request"
Description: |
  Detects new OAuth applications authorized with privileged scopes in Google Workspace.
  Uses anomaly detection to identify users authorizing OAuth apps they haven't used in
  the past 7 days.
Enabled: false
Query: |
  -- pragma: template

  {% import 'anomalies' new_unique_values %}
  with subquery as (
      SELECT
        p_event_time,
        actor:email AS actor_email,
        parameters:client_id AS client_id,
        parameters:app_name AS app_name,
        parameters:scope AS scopes,
        ipAddress,
        id:applicationName AS application_name,
        name AS event_name
      FROM panther_logs.public.gsuite_activityevent
      WHERE p_occurs_since('7 day')
      AND id:applicationName = 'token'
      AND name = 'authorize'
      AND (
        LOWER(ARRAY_TO_STRING(parameters:scope, ' ')) LIKE '%admin.directory.user%'
        OR LOWER(ARRAY_TO_STRING(parameters:scope, ' ')) LIKE '%ediscovery%'
        OR LOWER(ARRAY_TO_STRING(parameters:scope, ' ')) LIKE '%drive%'
        OR LOWER(ARRAY_TO_STRING(parameters:scope, ' ')) LIKE '%cloud_search.query%'
      )
      LIMIT 10000
    ),
    {{ new_unique_values('subquery', 'actor_email', 'client_id', '1d') }}
Schedule:
    RateMinutes: 360
    TimeoutMinutes: 3

from panther_gsuite_helpers import gsuite_parameter_lookup

PRIVILEGED_SCOPES = [
    "admin.directory.user",
    "admin.directory.group",
    "admin.directory.domain",
    "ediscovery",
    "vault",
    "cloud_search.query",
]


def rule(event):
    scopes = event.deep_get("parameters", "scope", default=[])
    app_name = event.deep_get("id", "applicationName", default="")
    event_name = event.get("name")

    if app_name != "token" or event_name != "authorize":
        return False

    # Handle both list and string formats
    if scopes and isinstance(scopes, str):
        scopes = [scopes]

    # Check if any scope matches privileged scopes
    privileged_scopes_lower = [ps.lower() for ps in PRIVILEGED_SCOPES]

    for scope_url in scopes:
        # Extract the last part of the scope URL (e.g., "admin.directory.user" from full URL)
        scope_name = scope_url.split("/")[-1].lower()
        if scope_name in privileged_scopes_lower:
            return True

    return False


def title(event):
    actor = event.deep_get("actor", "email", default="<UNKNOWN_USER>")
    app_name = event.deep_get("parameters", "app_name", default="<UNKNOWN_APP>")

    return (
        f"Google Workspace: User [{actor}] authorized OAuth app [{app_name}] with privileged scopes"
    )


def alert_context(event):
    parameters = event.get("parameters", {})

    return {
        "actor": event.deep_get("actor", "email", default=""),
        "app_name": gsuite_parameter_lookup(parameters, "app_name"),
        "client_id": gsuite_parameter_lookup(parameters, "client_id"),
        "client_type": gsuite_parameter_lookup(parameters, "client_type"),
        "scopes": gsuite_parameter_lookup(parameters, "scope"),
        "scope_data": gsuite_parameter_lookup(parameters, "scope_data"),
        "event_type": event.get("name"),
        "ip_address": event.get("ipAddress"),
    }

AnalysisType: rule
RuleID: "Google.Workspace.OAuth.Privileged.Scopes"
DisplayName: "Google Workspace OAuth Application Authorized with Privileged Scopes"
Filename: gsuite_oauth_privileged_scopes.py
LogTypes:
  - GSuite.ActivityEvent
Enabled: true
Severity: Info
DedupPeriodMinutes: 60
Status: Experimental
Reference: https://businessinsights.bitdefender.com/the-chain-reaction-new-methods-for-extending-local-breaches-in-google-workspace
Description: >
  Detects when a user authorizes an OAuth application with privileged scopes in Google Workspace.
  Privileged scopes grant broad access to sensitive data and administrative functions.
Runbook: |
  1. Query GSuite.ActivityEvent logs for all OAuth token authorize events by actor:email in the 7 days before and after this alert to identify if this is part of a pattern of suspicious OAuth grants
  2. Search for the parameters:client_id across all users in the organization to determine if other users also authorized the same application
  3. Review audit logs for any actions taken using the OAuth token between the authorization timestamp and now, filtering by parameters:app_name and the authorized scopes
Tags:
  - GSuite
  - Initial Access
  - Persistence
  - Account Manipulation
Reports:
  MITRE ATT&CK:
    - TA0001:T1078.004
    - TA0003:T1098
SummaryAttributes:
  - actor:email
  - p_any_ip_addresses

import re


def normalize_username(email):
    if not email:
        return None
    # Extract username before @ symbol
    username = email.split("@")[0] if "@" in email else email
    # Remove all non-alphanumeric characters and convert to lowercase
    return re.sub(r"[^a-z0-9]", "", username.lower())


def rule(_):
    return True


def title(event):
    user = event.get("user", "<UNKNOWN_USER>")
    new_ip = event.get("new_ip", "<UNKNOWN_IP>")
    request_count = event.get("request_count", 0)
    return (
        f"Google Workspace: User [{user}] made {request_count} OAuth token requests "
        f"from new IP [{new_ip}]"
    )


def alert_context(event):
    user = event.get("user")
    return {
        "user": user,
        "username_normalized": normalize_username(user),
        "new_ip": event.get("new_ip"),
        "request_count": event.get("request_count"),
        "app_names": event.get("app_names"),
        "client_ids": event.get("client_ids"),
        "first_seen": event.get("first_seen"),
        "last_seen": event.get("last_seen"),
        "description": (
            "User requested OAuth tokens from an IP address not seen in the past 30 days, "
            "with multiple requests indicating active usage"
        ),
    }

AnalysisType: scheduled_rule
DisplayName: "Google Workspace OAuth Token Requests from New IP"
DedupPeriodMinutes: 1440
RuleID: "Google.Workspace.OAuth.Token.New.IP"
Description: |
  Alerts when users request OAuth tokens from IP addresses they haven't used in the past 30 days,
  with 3+ requests indicating active usage. This may indicate GAIA credential theft where attackers
  use stolen refresh tokens to request access tokens from their infrastructure.
ScheduledQueries:
  - Google Workspace OAuth Token Requests from New IPs
Enabled: false
Filename: gsuite_oauth_token_new_ip_rule.py
Reference: https://businessinsights.bitdefender.com/the-chain-reaction-new-methods-for-extending-local-breaches-in-google-workspace
Runbook: |
  1. Query GSuite.ActivityEvent for all OAuth token requests (applicationName: "token") by the user in the 24 hours before and after the alert to identify the full scope of token activity from the new IP address
  2. Check if the new IP address is associated with cloud providers, VPN services, proxy networks, or residential ISPs, and compare its geographic location to the user's typical login locations in the past 30 days
  3. Search for other authentication anomalies for this user in the past 7 days, including login type changes, rapid multi-IP authentication, password changes, or device compromise warnings
Severity: Medium
Tags:
  - GSuite
  - Initial Access
  - Valid Accounts
  - GAIA
  - Credential Theft
  - OAuth
Reports:
  MITRE ATT&CK:
    - TA0001:T1078.004
    - TA0006:T1550
SummaryAttributes:
  - user
  - new_ip
  - app_names

AnalysisType: scheduled_query
QueryName: "Google Workspace OAuth Token Requests from New IPs"
Description: |
  Detects users requesting OAuth tokens from IPv4 addresses they haven't used in the past 30 days,
  with 3+ requests indicating active usage. May indicate GAIA credential theft where attackers use
  stolen refresh tokens from their infrastructure.
Enabled: false
SnowflakeQuery: |
  WITH baseline_ips AS (
    SELECT
      actor:email AS user,
      ARRAY_AGG(DISTINCT ipAddress) AS normal_ips
    FROM panther_logs.public.gsuite_activityevent
    WHERE p_event_time >= DATEADD(day, -30, CURRENT_TIMESTAMP)
      AND p_event_time < DATEADD(day, -1, CURRENT_TIMESTAMP)
      AND id:applicationName = 'token'
      AND name = 'authorize'
      AND ipAddress NOT LIKE '%:%'
    GROUP BY actor:email
  ),
  recent_tokens AS (
    SELECT
      actor:email AS user,
      ipAddress,
      parameters:app_name AS app_name,
      parameters:client_id AS client_id,
      p_event_time
    FROM panther_logs.public.gsuite_activityevent
    WHERE p_occurs_since('1 day')
      AND id:applicationName = 'token'
      AND name = 'authorize'
      AND ipAddress NOT LIKE '%:%'
  )
  SELECT
    r.user,
    r.ipAddress AS new_ip,
    COUNT(*) AS request_count,
    ARRAY_UNIQUE_AGG(r.app_name) AS app_names,
    ARRAY_UNIQUE_AGG(r.client_id) AS client_ids,
    MIN(r.p_event_time) AS first_seen,
    MAX(r.p_event_time) AS last_seen
  FROM recent_tokens r
  LEFT JOIN baseline_ips b ON r.user = b.user
  WHERE NOT ARRAY_CONTAINS(r.ipAddress::VARIANT, COALESCE(b.normal_ips, ARRAY_CONSTRUCT()))
  GROUP BY r.user, r.ipAddress
  HAVING COUNT(*) >= 3
  ORDER BY request_count DESC

DatabricksQuery: |
  WITH baseline_ips AS (
    SELECT
      actor:email AS user,
      COLLECT_SET(ipAddress) AS normal_ips
    FROM panther_logs.gsuite_activityevent
    WHERE p_event_time >= CURRENT_TIMESTAMP - INTERVAL 30 DAYS
      AND p_event_time < CURRENT_TIMESTAMP - INTERVAL 1 DAY
      AND id:applicationName = 'token'
      AND name = 'authorize'
      AND ipAddress NOT LIKE '%:%'
    GROUP BY actor:email
  ),
  recent_tokens AS (
    SELECT
      actor:email AS user,
      ipAddress,
      parameters:app_name AS app_name,
      parameters:client_id AS client_id,
      p_event_time
    FROM panther_logs.gsuite_activityevent
    WHERE p_occurs_since('1 day')
      AND id:applicationName = 'token'
      AND name = 'authorize'
      AND ipAddress NOT LIKE '%:%'
  )
  SELECT
    r.user,
    r.ipAddress AS new_ip,
    COUNT(*) AS request_count,
    COLLECT_SET(r.app_name) AS app_names,
    COLLECT_SET(r.client_id) AS client_ids,
    MIN(r.p_event_time) AS first_seen,
    MAX(r.p_event_time) AS last_seen
  FROM recent_tokens r
  LEFT JOIN baseline_ips b ON r.user = b.user
  WHERE NOT ARRAY_CONTAINS(COALESCE(b.normal_ips, ARRAY()), r.ipAddress)
  GROUP BY r.user, r.ipAddress
  HAVING COUNT(*) >= 3
  ORDER BY request_count DESC
Schedule:
  RateMinutes: 1440
  TimeoutMinutes: 5

AnalysisType: scheduled_query
QueryName: "Google Workspace OAuthLogin Scope Anomalous Application Access"
Description: |
  Detects apps requesting OAuth tokens with the OAuthLogin scope when they haven't requested
  this scope in the previous 14 days. This scope can be used to access Google's device password
  escrow endpoint for GAIA credential theft.
Enabled: false
Query: |
  -- pragma: template
  {% import 'anomalies' new_unique_values %}

  WITH subquery AS (
    SELECT
      parameters:app_name AS app_name,
      parameters:client_id AS client_id,
      actor:email AS user_email,
      ipAddress,
      p_event_time
    FROM panther_logs.public.gsuite_activityevent
    WHERE p_occurs_since('14 days')
      AND id:applicationName = 'token'
      AND name = 'authorize'
      AND ARRAY_TO_STRING(parameters:scope, ',') ILIKE '%accounts/OAuthLogin%'
  ),
  {{ new_unique_values('subquery', 'app_name', 'user_email', '1 day') }}
Schedule:
  RateMinutes: 360
  TimeoutMinutes: 3

AnalysisType: scheduled_query
QueryName: "Google Workspace Rapid Multi-IP Authentication"
Description: |
  Detects users authenticating from 3+ distinct IPv4 addresses within 6 hours.
  May indicate GAIA credential theft where stolen OAuth tokens are used across
  multiple compromised machines simultaneously. IPv6 addresses are excluded to
  avoid false positives from dual-stack networking.
Enabled: false
SnowflakeQuery: |
  SELECT
    actor:email AS user,
    COUNT(DISTINCT
      CASE
        WHEN ipAddress LIKE '%:%' THEN NULL
        ELSE ipAddress
      END
    ) AS unique_ip_count,
    COUNT(*) AS total_logins,
    ARRAY_UNIQUE_AGG(
      CASE
        WHEN ipAddress LIKE '%:%' THEN NULL
        ELSE ipAddress
      END
    ) AS ip_addresses,
    ARRAY_UNIQUE_AGG(parameters:login_type) AS login_types,
    MIN(p_event_time) AS first_login,
    MAX(p_event_time) AS last_login,
    DATEDIFF('minute', MIN(p_event_time), MAX(p_event_time)) AS time_span_minutes
  FROM panther_logs.public.gsuite_activityevent
  WHERE p_occurs_since('6 hours')
    AND id:applicationName = 'login'
    AND name = 'login_success'
  GROUP BY actor:email
  HAVING COUNT(DISTINCT
    CASE
      WHEN ipAddress LIKE '%:%' THEN NULL
      ELSE ipAddress
    END
  ) >= 3
  ORDER BY unique_ip_count DESC
  LIMIT 10000

DatabricksQuery: |
  SELECT
    actor:email AS user,
    COUNT(DISTINCT
      CASE
        WHEN ipAddress LIKE '%:%' THEN NULL
        ELSE ipAddress
      END
    ) AS unique_ip_count,
    COUNT(*) AS total_logins,
    COLLECT_SET(
      CASE
        WHEN ipAddress LIKE '%:%' THEN NULL
        ELSE ipAddress
      END
    ) AS ip_addresses,
    COLLECT_SET(parameters:login_type) AS login_types,
    MIN(p_event_time) AS first_login,
    MAX(p_event_time) AS last_login,
    TIMESTAMPDIFF(MINUTE, MIN(p_event_time), MAX(p_event_time)) AS time_span_minutes
  FROM panther_logs.gsuite_activityevent
  WHERE p_occurs_since('6 hours')
    AND id:applicationName = 'login'
    AND name = 'login_success'
  GROUP BY actor:email
  HAVING COUNT(DISTINCT
    CASE
      WHEN ipAddress LIKE '%:%' THEN NULL
      ELSE ipAddress
    END
  ) >= 3
  ORDER BY unique_ip_count DESC
  LIMIT 10000
Schedule:
  RateMinutes: 360
  TimeoutMinutes: 3

def rule(_):
    return True


def title(event):
    user = event.get("user", "<UNKNOWN_USER>")
    ip_count = event.get("unique_ip_count", 0)
    return f"Google Workspace: User [{user}] authenticated from {ip_count} distinct IPs in 6 hours"


def severity(event):
    ip_count = event.get("unique_ip_count", 0)

    # Very high IP count suggests active credential spread
    if ip_count >= 4:
        return "HIGH"

    return "MEDIUM"


def alert_context(event):
    return {
        "user": event.get("user"),
        "unique_ip_count": event.get("unique_ip_count"),
        "ip_addresses": event.get("ip_addresses"),
        "login_types": event.get("login_types"),
        "first_login": event.get("first_login"),
        "last_login": event.get("last_login"),
        "time_span_minutes": event.get("time_span_minutes"),
        "total_logins": event.get("total_logins"),
        "description": (
            "User authenticated from multiple distinct IPv4 addresses in a short time window"
        ),
    }

AnalysisType: scheduled_rule
DisplayName: "Google Workspace Rapid Multi-IP Authentication"
DedupPeriodMinutes: 360
RuleID: "Google.Workspace.Rapid.Multi.IP.Authentication"
Description: |
  Alerts when users authenticate from 3+ distinct IPv4 addresses within 6 hours.
  This pattern may indicate GAIA credential theft where attackers use stolen OAuth
  tokens across multiple compromised machines simultaneously. IPv6 addresses are
  excluded to avoid false positives from dual-stack networking environments.
ScheduledQueries:
  - Google Workspace Rapid Multi-IP Authentication
Enabled: false
Filename: gsuite_rapid_multi_ip_authentication_rule.py
Reference: https://businessinsights.bitdefender.com/the-chain-reaction-new-methods-for-extending-local-breaches-in-google-workspace
Runbook: |
  1. Query GSuite.ActivityEvent for all login events by the user in the 12 hours before and after the alert to establish the full timeline of authentication activity and identify all source IP addresses used
  2. Check if the source IP addresses are associated with cloud providers, VPN services, proxy networks, or known corporate infrastructure, and compare the geographic locations of the IPs to identify impossible travel patterns
  3. Search for other authentication anomalies for this user in the past 7 days, including login type changes, OAuth token authorizations with privileged scopes, failed authentication attempts, or device compromise warnings
Severity: Medium
Tags:
  - GSuite
  - Lateral Movement
  - Valid Accounts
  - GAIA
  - Credential Theft
Reports:
  MITRE ATT&CK:
    - TA0008:T1078.004
    - TA0006:T1550
SummaryAttributes:
  - user
  - ip_addresses

from panther_gsuite_helpers import gsuite_activityevent_alert_context


def rule(event):
    if event.deep_get("id", "applicationName", default="<UNKNOWN_APPLICATION>") != "gmail":
        return False
    return event.deep_get(
        "parameters", "message_info", "is_spam", default=False
    ) is True and event.deep_get("parameters", "event_info", "mail_event_type", default=0) in (
        17,
        18,
        19,
    )


def title(event):
    user = event.deep_get("actor", "email", default="<UNKNOWN_USER>")
    return f"[{user}] has downloaded potentially malicious attachments from a spam email"


def alert_context(event):
    return gsuite_activityevent_alert_context(event)

AnalysisType: rule
Filename: gsuite_attachments_downloaded_from_spam_email.py
RuleID: "GSuite.Gmail.SpamEmail.AttachmentDownload"
DisplayName: "Gsuite Attachments Downloaded from Spam Email"
Enabled: true
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
Reports:
  MITRE ATT&CK:
    - TA0001:T1566.001 # Initial Access: Phishing - Spearphishing Attachment
    - TA0011:T1204.002 # Execution: User Execution - Malicious File
Severity: High
Description: Detects when a user downloads or saves to Google Drive one or more attachments that are classified as spam.
Threshold: 1
DedupPeriodMinutes: 60

def rule(event):
    return (
        event.get("name") == "change_calendar_acls"
        and event.get("parameters", {}).get("grantee_email")
        == "__public_principal__@public.calendar.google.com"
    )


def title(event):
    return (
        f"GSuite calendar "
        f"[{event.deep_get('parameters', 'calendar_id', default='<NO_CALENDAR_ID>')}] made "
        f"{public_or_private(event)} by "
        f"[{event.deep_get('actor', 'email', default='<NO_ACTOR_FOUND>')}]"
    )


def severity(event):
    return "LOW" if public_or_private(event) == "private" else "MEDIUM"


def public_or_private(event):
    return "private" if event.deep_get("parameters", "access_level") == "none" else "public"

AnalysisType: rule
Filename: gsuite_calendar_made_public.py
RuleID: "GSuite.CalendarMadePublic"
DisplayName: "GSuite Calendar Has Been Made Public"
Enabled: true
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
Reports:
  MITRE ATT&CK:
    - TA0007:T1087
Severity: Medium
Description: >
  A User or Admin Has Modified A Calendar To Be Public
Reference: https://support.google.com/calendar/answer/37083?hl=en&sjid=864417124752637253-EU
Runbook: >
  Follow up with user about this calendar share.
SummaryAttributes:
  - actor:email

def rule(event):
    if event.deep_get("id", "applicationName") != "mobile":
        return False

    return bool(event.get("name") == "SUSPICIOUS_ACTIVITY_EVENT")


def title(event):
    return (
        f"User [{event.deep_get('actor', 'email', default='<UNKNOWN_USER>')}]"
        f"'s device was compromised"
    )

AnalysisType: rule
Filename: gsuite_mobile_device_suspicious_activity.py
RuleID: "GSuite.DeviceSuspiciousActivity"
DisplayName: "GSuite Device Suspicious Activity"
Enabled: true
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
Severity: Low
Description: >
  GSuite reported a suspicious activity on a user's device.
Reference: https://support.google.com/a/answer/7562460?hl=en&sjid=864417124752637253-EU
Runbook: >
  Validate that the suspicious activity was expected by the user.
SummaryAttributes:
  - actor:email

def rule(event):
    if event.get("name") != "change_owner":
        return False

    if event.deep_get("parameters", "visibility") in (
        "shared_internally",
        "people_within_domain_with_link",
        "private",
    ):
        return False

    previous_owner = event.deep_get("parameters", "owner", default="<UNKNOWN USER>")
    new_owner = event.deep_get("parameters", "new_owner", default="<UNKNOWN USER>")

    previous_owner_domain = previous_owner.split("@")[1] if "@" in previous_owner else None
    new_owner_domain = new_owner.split("@")[1] if "@" in new_owner else None

    if previous_owner_domain is None or new_owner_domain is None:
        return False

    if previous_owner_domain != new_owner_domain:
        return True

    return False


def title(event):
    actor = event.deep_get("actor", "email", default="<UNKNOWN USER>")
    previous_owner = event.deep_get("parameters", "owner", default="<UNKNOWN USER>")
    new_owner = event.deep_get("parameters", "new_owner", default="<UNKNOWN USER>")

    return f"User [{actor}] transferred document ownership from [{previous_owner}] to [{new_owner}]"

AnalysisType: rule
Filename: gsuite_doc_ownership_transfer.py
RuleID: "GSuite.DocOwnershipTransfer"
DisplayName: "GSuite Document External Ownership Transfer"
Enabled: true
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
  - Collection:Data from Information Repositories
Reports:
  MITRE ATT&CK:
    - TA0009:T1213
Severity: Low
Description: >
  A GSuite document's ownership was transferred to an external party.
Reference: https://support.google.com/drive/answer/2494892?hl=en&co=GENIE.Platform%3DDesktop&sjid=864417124752637253-EU
Runbook: >
  Verify that this document did not contain sensitive or private company information.
SummaryAttributes:
  - actor:email

def rule(_):
    return True


def title(event):
    return (
        f"GSuite: [{event.get('user', '<user_not_found>')}] "
        f"has deleted [{event.get('delete_count', '<count_not_found>')}] "
        "documents from Google Drive."
    )


def alert_context(event):
    return event.to_dict()

AnalysisType: scheduled_rule
Description: Scheduled rule for the GSuite Drive Many Documents Deleted query. Looks for users who have deleted more than 10 (tunable) documents the past day.
DisplayName: "GSuite Drive Many Documents Deleted"
Enabled: false
Status: Deprecated
Filename: gsuite_drive_many_docs_deleted.py
Reference: https://support.google.com/drive/answer/2375102?hl=en&co=GENIE.Platform%3DAndroid#:~:text=To%20delete%20your%20Google%20Drive,them%20to%20empty%20your%20trash.
Severity: Medium
DedupPeriodMinutes: 60
RuleID: "GSuite.Drive.Many.Documents.Deleted"
Threshold: 1
ScheduledQueries:
  - GSuite Many Docs Deleted Query

def rule(event):
    if event.deep_get("id", "applicationName") != "drive":
        return False
    return event.get("name") == "trash"


def title(event):
    user = event.deep_get("actor", "email", default="<UNKNOWN_USER>")
    return f"Google Workspace: User [{user}] deleted many documents from Google Drive"


def dedup(event):
    return event.deep_get("actor", "email", default="")


def unique(event):
    return event.deep_get("parameters", "doc_id") or None


def severity(event):
    visibility = event.deep_get("parameters", "visibility", default="")
    if visibility == "shared_externally":
        return "HIGH"
    return "DEFAULT"


def alert_context(event):
    return {
        "user": event.deep_get("actor", "email"),
        "doc_title": event.deep_get("parameters", "doc_title"),
        "doc_type": event.deep_get("parameters", "doc_type"),
        "visibility": event.deep_get("parameters", "visibility"),
    }

AnalysisType: rule
Filename: gsuite_drive_many_docs_deleted.py
RuleID: "GSuite.Drive.BulkDocumentDeletion"
DisplayName: "GSuite Drive Many Documents Deleted"
Status: Experimental
Enabled: true
Severity: Medium
DedupPeriodMinutes: 60
Threshold: 11
LogTypes:
  - GSuite.ActivityEvent
Description: >
  Detects when a user moves more than 10 distinct documents to the trash in Google Drive
  within 60 minutes. This may indicate accidental or malicious bulk deletion of files.
Reference: https://support.google.com/drive/answer/2375102
Reports:
  MITRE ATT&CK:
    - TA0040:T1485
Tags:
  - GSuite
  - Impact
  - Data Destruction
Runbook: |
  1. Query GSuite.ActivityEvent for all trash events by actor:email in the 2 hours around this alert to identify the full list of parameters:doc_title values deleted and whether they belong to shared drives
  2. Check parameters:visibility on the deleted documents to determine if externally or internally shared files were affected, and assess the business impact of the deletions
  3. Search for other suspicious drive activity by this user in the past 24 hours, including bulk downloads, sharing changes, or access to sensitive documents prior to deletion

from panther_gsuite_helpers import gsuite_activityevent_alert_context


def rule(event):
    if event.deep_get("id", "applicationName", default="<UNKNOWN_APPLICATION>") != "gmail":
        return False
    return event.deep_get("parameters", "message_info", "message_set", "type", default=0) == 46


def title(event):
    user = event.deep_get("actor", "email", default="<UNKNOWN_USER>")
    subject = event.deep_get("parameters", "message_info", "subject", default="<UNKNOWN_SUBJECT>")
    return (
        f"Message [{subject}] received by user [{user}] "
        f"has bypassed your organization's spam filter"
    )


def alert_context(event):
    return gsuite_activityevent_alert_context(event)

AnalysisType: rule
Filename: gsuite_bypass_spam_filter_email.py
RuleID: "GSuite.Gmail.Email.SpamFilter.Bypass"
DisplayName: "Gsuite Email Bypassed Spam Filter"
Enabled: true
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
Reports:
  MITRE ATT&CK:
    - TA0001:T1566 # Initial Access: Phishing
Severity: Medium
Description: >
  Detects if an email received by a user has bypassed the organization's spam filter.
Threshold: 1
DedupPeriodMinutes: 60

import json
from unittest.mock import MagicMock

from panther_gsuite_helpers import gsuite_parameter_lookup as param_lookup

# Add any domain name(s) that you expect to share documents with in the ALLOWED_DOMAINS set
ALLOWED_DOMAINS = set()

PUBLIC_PROVIDERS = {
    "gmail.com",
    "yahoo.com",
    "outlook.com",
    "aol.com",
    "yandex.com",
    "protonmail.com",
    "pm.me",
    "icloud.com",
    "tutamail.com",
    "tuta.io",
    "keemail.me",
    "mail.com",
    "zohomail.com",
    "hotmail.com",
    "msn.com",
}

VISIBILITY = {
    "people_with_link",
    "people_within_domain_with_link",
    "public_on_the_web",
    "shared_externally",
    "unknown",
}

ALERT_DETAILS = {}

# Events where documents have changed perms due to parent folder change
INHERITANCE_EVENTS = {
    "change_user_access_hierarchy_reconciled",
    "change_document_access_scope_hierarchy_reconciled",
}


def init_alert_details(log):
    global ALERT_DETAILS  # pylint: disable=global-statement
    ALERT_DETAILS[log] = {
        "ACCESS_SCOPE": "<UNKNOWN_ACCESS_SCOPE>",
        "DOC_TITLE": "<UNKNOWN_TITLE>",
        "NEW_VISIBILITY": "<UNKNOWN_VISIBILITY>",
        "TARGET_USER_EMAILS": ["<UNKNOWN_USER>"],
        "TARGET_DOMAIN": "<UNKNOWN_DOMAIN>",
    }


def user_is_external(target_user):
    global ALLOWED_DOMAINS  # pylint: disable=global-statement
    # We need to type-cast ALLOWED_DOMAINS for unit testing mocks
    if isinstance(ALLOWED_DOMAINS, MagicMock):
        ALLOWED_DOMAINS = set(json.loads(ALLOWED_DOMAINS()))  # pylint: disable=not-callable
    for domain in ALLOWED_DOMAINS:
        if domain in target_user:
            return False
    return True


def rule(event):
    # pylint: disable=too-complex
    global ALLOWED_DOMAINS  # pylint: disable=global-statement
    if event.deep_get("id", "applicationName") != "drive":
        return False

    # Events that have the types in INHERITANCE_EVENTS are
    # changes to documents and folders that occur due to
    # a change in the parent folder's permission. We ignore
    # these events to prevent every folder change from
    # generating multiple alerts.
    if event.get("name") in INHERITANCE_EVENTS:
        return False

    log = event.get("p_row_id")
    init_alert_details(log)

    # We need to type-cast ALLOWED_DOMAINS for unit testing mocks
    if isinstance(ALLOWED_DOMAINS, MagicMock):
        ALLOWED_DOMAINS = set(json.loads(ALLOWED_DOMAINS()))  # pylint: disable=not-callable

    # For GSuite.ActivityEvent, each log is a single event.
    # Check if this event is a visibility change for a domain
    if (
        event.get("type") == "acl_change"
        and event.get("name") == "change_document_visibility"
        and param_lookup(event.get("parameters", {}), "new_value") != ["private"]
        and not param_lookup(event.get("parameters", {}), "target_domain") in ALLOWED_DOMAINS
        and param_lookup(event.get("parameters", {}), "visibility") in VISIBILITY
    ):
        ALERT_DETAILS[log]["TARGET_DOMAIN"] = param_lookup(
            event.get("parameters", {}), "target_domain"
        )
        ALERT_DETAILS[log]["NEW_VISIBILITY"] = param_lookup(
            event.get("parameters", {}), "visibility"
        )
        ALERT_DETAILS[log]["DOC_TITLE"] = param_lookup(event.get("parameters", {}), "doc_title")
        if param_lookup(event.get("parameters", {}), "new_value") != ["none"]:
            ALERT_DETAILS[log]["ACCESS_SCOPE"] = param_lookup(
                event.get("parameters", {}), "new_value"
            )
        return True

    # For visibility changes that apply to a user
    if (
        event.get("type") == "acl_change"
        and event.get("name") == "change_user_access"
        and param_lookup(event.get("parameters", {}), "new_value") != ["none"]
        and user_is_external(param_lookup(event.get("parameters", {}), "target_user"))
    ):
        if ALERT_DETAILS[log]["TARGET_USER_EMAILS"] != ["<UNKNOWN_USER>"]:
            ALERT_DETAILS[log]["TARGET_USER_EMAILS"].append(
                param_lookup(event.get("parameters", {}), "target_user")
            )
        else:
            ALERT_DETAILS[log]["TARGET_USER_EMAILS"] = [
                param_lookup(event.get("parameters", {}), "target_user")
            ]
            ALERT_DETAILS[log]["DOC_TITLE"] = param_lookup(event.get("parameters", {}), "doc_title")
            ALERT_DETAILS[log]["ACCESS_SCOPE"] = param_lookup(
                event.get("parameters", {}), "new_value"
            )
        return True

    return False


def alert_context(event):
    log = event.get("p_row_id")
    if ALERT_DETAILS[log]["TARGET_USER_EMAILS"] != ["<UNKNOWN_USER>"]:
        return {"target users": ALERT_DETAILS[log]["TARGET_USER_EMAILS"]}
    return {}


def dedup(event):
    return event.deep_get("actor", "email", default="<UNKNOWN_USER>")


def title(event):
    log = event.get("p_row_id")
    if ALERT_DETAILS[log]["TARGET_USER_EMAILS"] != ["<UNKNOWN_USER>"]:
        if len(ALERT_DETAILS[log]["TARGET_USER_EMAILS"]) == 1:
            sharing_scope = ALERT_DETAILS[log]["TARGET_USER_EMAILS"][0]
        else:
            sharing_scope = "multiple users"
        if ALERT_DETAILS[log]["NEW_VISIBILITY"] == "shared_externally":
            sharing_scope += " (outside the document's current domain)"
    elif ALERT_DETAILS[log]["TARGET_DOMAIN"] == "all":
        sharing_scope = "the entire internet"
        if ALERT_DETAILS[log]["NEW_VISIBILITY"] == "people_with_link":
            sharing_scope += " (anyone with the link)"
        elif ALERT_DETAILS[log]["NEW_VISIBILITY"] == "public_on_the_web":
            sharing_scope += " (link not required)"
    else:
        sharing_scope = f"the {ALERT_DETAILS[log]['TARGET_DOMAIN']} domain"
        if ALERT_DETAILS[log]["NEW_VISIBILITY"] == "people_within_domain_with_link":
            sharing_scope += f" (anyone in {ALERT_DETAILS[log]['TARGET_DOMAIN']} with the link)"
        elif ALERT_DETAILS[log]["NEW_VISIBILITY"] == "public_in_the_domain":
            sharing_scope += f" (anyone in {ALERT_DETAILS[log]['TARGET_DOMAIN']})"

    # alert_access_scope = ALERT_DETAILS[log]["ACCESS_SCOPE"][0].replace("can_", "")

    return (
        f"User [{event.deep_get('actor', 'email', default='<UNKNOWN_USER>')}] made documents "
        f"externally visible"
    )


def severity(event):
    log = event.get("p_row_id")
    if ALERT_DETAILS[log]["TARGET_USER_EMAILS"] != ["<UNKNOWN_USER>"]:
        for address in ALERT_DETAILS[log]["TARGET_USER_EMAILS"]:
            domain = address.split("@")[1]
            if domain in PUBLIC_PROVIDERS:
                return "LOW"
    return "INFO"

AnalysisType: rule
Filename: gsuite_drive_visibility_change.py
RuleID: "GSuite.DriveVisibilityChanged"
DisplayName: "GSuite External Drive Document"
Enabled: false
LogTypes:
  - GSuite.ActivityEvent
Tags:
  - GSuite
  - Collection:Data from Information Repositories
  - Configuration Required
Reports:
  MITRE ATT&CK:
    - TA0009:T1213
Severity: Low
Description: >
  A Google drive resource became externally accessible.
Reference: https://support.google.com/a/users/answer/12380484?hl=en&sjid=864417124752637253-EU
Runbook: >
  Investigate whether the drive document is appropriate to be publicly accessible.
SummaryAttributes:
  - actor:email
DedupPeriodMinutes: 360 # 6 hours

def rule(event):
    if event.deep_get("id", "applicationName") != "login":
        return False

    return bool(event.get("name") == "gov_attack_warning")


def title(event):
    return (
        f"User [{event.deep_get('actor', 'email', default='<UNKNOWN_EMAIL>')}] may have been "
        f"targeted by a government attack"
    )