Panther rules: gcp

import json
from unittest.mock import MagicMock

from panther_base_helpers import deep_walk
from panther_gcp_helpers import get_k8s_info

# This is a list of principals that are allowed to exec into pods
# in various namespaces and projects.
ALLOW_LIST = [
    {
        # If empty, then no principals
        "principals": [
            # "system:serviceaccount:example-namespace:example-namespace-service-account",
        ],
        # If empty, then all namespaces
        "namespaces": [],
        # If projects empty then all projects
        "projects": [],
    },
    # Add more allowed principals here
    # {
    #     "principals": [],
    #     "namespaces": [],
    #     "projects": [],
    # },
]


def rule(event):
    # pylint: disable=not-callable
    # pylint: disable=global-statement
    global ALLOW_LIST
    if isinstance(ALLOW_LIST, MagicMock):
        ALLOW_LIST = json.loads(ALLOW_LIST())

    # Defaults to False (no alert) unless method is exec and principal not allowed
    if not all(
        [
            event.deep_walk("protoPayload", "methodName") == "io.k8s.core.v1.pods.exec.create",
            event.deep_walk("resource", "type") == "k8s_cluster",
        ]
    ):
        return False

    k8s_info = get_k8s_info(event)
    principal = deep_walk(k8s_info, "principal", default="<NO PRINCIPAL>")
    namespace = deep_walk(k8s_info, "namespace", default="<NO NAMESPACE>")
    project_id = deep_walk(k8s_info, "project_id", default="<NO PROJECT_ID>")
    # rule_exceptions that are allowed temporarily are defined in gcp_environment.py
    # Some execs have principal which is long numerical UUID, appears to be k8s internals
    for allowed_principal in ALLOW_LIST:
        allowed_principals = deep_walk(allowed_principal, "principals", default=[])
        allowed_namespaces = deep_walk(allowed_principal, "namespaces", default=[])
        allowed_project_ids = deep_walk(allowed_principal, "projects", default=[])
        if (
            principal in allowed_principals
            and (namespace in allowed_namespaces or allowed_namespaces == [])
            and (project_id in allowed_project_ids or allowed_project_ids == [])
        ):
            if "@" not in principal:
                return False
    return True


def title(event):
    # TODO: use unified data model field in title for actor
    k8s_info = get_k8s_info(event)
    principal = deep_walk(k8s_info, "principal", default="<NO PRINCIPAL>")
    project_id = deep_walk(
        k8s_info,
        "project_id",
        default="",
    )
    pod = deep_walk(k8s_info, "pod", default="")
    namespace = deep_walk(k8s_info, "namespace", default="")
    return f"Exec into pod namespace/{namespace}/pod/{pod} by {principal} in {project_id}"


def alert_context(event):
    return get_k8s_info(event)

AnalysisType: rule
Filename: gcp_k8s_exec_into_pod.py
RuleID: "GCP.K8s.ExecIntoPod"
DisplayName: "Exec into Pod"
Enabled: false
LogTypes:
  - GCP.AuditLog
Tags:
  - GCP
  - Security Control
  - Configuration Required
Severity: Medium
Description: >
  Alerts when users exec into pod. Possible to specify specific projects and allowed users.
Runbook: >
  Investigate the user and determine why. Advise that it is discouraged practice. Create ticket if appropriate.
Reference: https://cloud.google.com/migrate/containers/docs/troubleshooting/executing-shell-commands

def rule(event):
    return all(
        [
            event.deep_get("resource", "type", default="") == "http_load_balancer",
            event.deep_get("jsonPayload", "statusDetails", default="")
            == "handled_by_identity_aware_proxy",
            not any(
                [
                    str(event.deep_get("httprequest", "status", default=000)).startswith("2"),
                    str(event.deep_get("httprequest", "status", default=000)).startswith("3"),
                ]
            ),
        ]
    )


def title(event):
    source = event.deep_get("jsonPayload", "remoteIp", default="<SRC_IP_NOT_FOUND>")
    request_url = event.deep_get("httprequest", "requestUrl", default="<REQUEST_URL_NOT_FOUND>")
    return f"GCP: Request Violating IAP controls from [{source}] to [{request_url}]"

AnalysisType: rule
Description: GCP Access Attempts Violating IAP Access Controls
DisplayName: "GCP Access Attempts Violating IAP Access Controls"
Enabled: true
Filename: gcp_access_attempts_violating_iap_access_controls.py
Reference: https://cloud.google.com/iap/docs/concepts-overview
Severity: Medium
DedupPeriodMinutes: 60
LogTypes:
  - GCP.HTTPLoadBalancer
RuleID: "GCP.Access.Attempts.Violating.IAP.Access.Controls"
Threshold: 1

def rule(event):
    severity = event.get("severity", "")
    status_code = event.deep_get("protoPayload", "status", "code", default="")
    violation_types = event.deep_walk(
        "protoPayload", "status", "details", "violations", "type", default=[]
    )
    if all(
        [
            severity == "ERROR",
            status_code == 7,
            "VPC_SERVICE_CONTROLS" in violation_types,
        ]
    ):
        return True
    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    method = event.deep_get("protoPayload", "methodName", default="<METHOD_NOT_FOUND>")
    return f"GCP: [{actor}] performed a [{method}] request that violates VPC Service Controls"

AnalysisType: rule
Description: An access attempt violating VPC service controls (such as Perimeter controls) has been made.
DisplayName: "GCP Access Attempts Violating VPC Service Controls"
Enabled: true
Filename: gcp_access_attempts_violating_vpc_service_controls.py
Reference: https://cloud.google.com/vpc-service-controls/docs/troubleshooting#debugging
Severity: Medium
DedupPeriodMinutes: 60
LogTypes:
  - GCP.AuditLog
RuleID: "GCP.Access.Attempts.Violating.VPC.Service.Controls"
Threshold: 1

# 1.07 GB
QUERY_THRESHOLD_BYTES = 1073741824


def rule(event):
    return all(
        [
            event.deep_get("resource", "type", default="<type not found>").startswith("bigquery"),
            event.deep_get("operation", "last") is True,
            event.deep_get("protoPayload", "metadata", "jobChange", "job", "jobConfig", "type")
            == "QUERY",
            event.deep_get(
                "protoPayload",
                "metadata",
                "jobChange",
                "job",
                "jobConfig",
                "queryConfig",
                "statementType",
            )
            == "SELECT",
            int(
                event.deep_get(
                    "protoPayload",
                    "metadata",
                    "jobChange",
                    "job",
                    "jobStats",
                    "queryStats",
                    "totalBilledBytes",
                    default=0,
                )
            )
            > QUERY_THRESHOLD_BYTES,
        ]
    )


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    return f"GCP: [{actor}] ran a large BigQuery query exceeding 1.07 GB threshold."


def alert_context(event):
    return {
        "query": event.deep_get(
            "protoPayload",
            "metadata",
            "jobChange",
            "job",
            "jobConfig",
            "queryConfig",
            "query",
            default="<QUERY_NOT_FOUND>",
        ),
        "actor": event.deep_get(
            "protoPayload",
            "authenticationInfo",
            "principalEmail",
            default="<ACTOR_NOT_FOUND>",
        ),
        "query_size": event.deep_get(
            "protoPayload",
            "metadata",
            "jobChange",
            "job",
            "jobStats",
            "queryStats",
            "totalBilledBytes",
            default=0,
        ),
    }

AnalysisType: rule
Description: Detect any BigQuery query that is doing a very large scan (> 1 GB).
DisplayName: "GCP BigQuery Large Scan"
Enabled: true
Filename: gcp_bigquery_large_scan.py
Reference: https://cloud.google.com/bigquery/docs/running-queries
Severity: Info
DedupPeriodMinutes: 60
LogTypes:
  - GCP.AuditLog
RuleID: "GCP.BigQuery.Large.Scan"
Threshold: 1

# CloudArmor signature IDs for CVE-2025-55182
REACT2SHELL_SIGNATURES = [
    "google-mrs-v202512-id000001-rce",
    "google-mrs-v202512-id000002-rce",
]


def rule(event):
    # Check that this is an HTTP load balancer event
    if event.deep_get("resource", "type") != "http_load_balancer":
        return False

    # Check enforced policy match
    enforced_policy = event.deep_get("jsonPayload", "enforcedSecurityPolicy", default={})
    enforced_sigs = enforced_policy.get("preconfiguredExprIds", [])
    if any(sig in enforced_sigs for sig in REACT2SHELL_SIGNATURES):
        return True

    # Check preview policy for non-blocking WAF matches
    preview_policy = event.deep_get("jsonPayload", "previewSecurityPolicy", default={})
    preview_sigs = preview_policy.get("preconfiguredExprIds", [])
    if any(sig in preview_sigs for sig in REACT2SHELL_SIGNATURES):
        return True

    return False


def title(event):
    remote_ip = event.deep_get("httpRequest", "remoteIp", default="<UNKNOWN_IP>")
    return f"Cloud Armor React2Shell (CVE-2025-55182) Exploit Detected from {remote_ip}"


def alert_context(event):
    enforced_policy = event.deep_get("jsonPayload", "enforcedSecurityPolicy", default={})
    preview_policy = event.deep_get("jsonPayload", "previewSecurityPolicy", default={})
    http_request = event.get("httpRequest", {})
    status_details = event.deep_get("jsonPayload", "statusDetails", default="<UNKNOWN_STATUS>")

    context = {
        "vulnerability": "CVE-2025-55182 (React2Shell)",
        "status_details": status_details,
        "remote_ip": http_request.get("remoteIp"),
        "request_url": http_request.get("requestUrl"),
        "request_method": http_request.get("requestMethod"),
        "user_agent": http_request.get("userAgent"),
        "status_code": http_request.get("status"),
        "referer": http_request.get("referer"),
        "enforced_policy": {
            "name": enforced_policy.get("name"),
            "configured_action": enforced_policy.get("configuredAction"),
            "outcome": enforced_policy.get("outcome"),
            "priority": enforced_policy.get("priority"),
            "signature_ids": enforced_policy.get("preconfiguredExprIds", []),
            "matched_field_type": enforced_policy.get("matchedFieldType"),
            "matched_field_name": enforced_policy.get("matchedFieldName"),
            "matched_field_value": enforced_policy.get("matchedFieldValue"),
            "matched_length": enforced_policy.get("matchedLength"),
        },
        "project_id": event.deep_get("resource", "labels", "project_id"),
        "backend_service": event.deep_get("resource", "labels", "backend_service_name"),
        "forwarding_rule": event.deep_get("resource", "labels", "forwarding_rule_name"),
    }

    # Include preview policy details if present
    if preview_policy:
        context["preview_policy"] = {
            "configured_action": preview_policy.get("configuredAction"),
            "outcome": preview_policy.get("outcome"),
            "priority": preview_policy.get("priority"),
            "signature_ids": preview_policy.get("preconfiguredExprIds", []),
            "matched_field_type": preview_policy.get("matchedFieldType"),
            "matched_field_name": preview_policy.get("matchedFieldName"),
            "matched_field_value": preview_policy.get("matchedFieldValue"),
            "matched_length": preview_policy.get("matchedLength"),
        }

    return context

AnalysisType: rule
Description: >
  Detects when GCP Cloud Armor detects HTTP requests matching specific Remote Code Execution (RCE)
  vulnerability signatures (google-mrs-v202512-id000001-rce and google-mrs-v202512-id000002-rce) which match React2Shell exploit attempts.
  These rules indicate active exploitation attempts against known RCE vulnerabilities.
DisplayName: GCP Cloud Armor RCE Attempt Detected
Enabled: true
Filename: gcp_cloud_armor_r2s_rce_attempt.py
RuleID: GCP.CloudArmor.React2Shell.RCE.Attempt
Severity: High
LogTypes:
  - GCP.HTTPLoadBalancer
Tags:
  - GCP
  - Cloud Armor
  - WAF
  - RCE
  - React2Shell
Reference: https://cloud.google.com/armor/docs/waf-rules#cves_and_other_vulnerabilities
Runbook: |
  1. Query GCP HTTP Load Balancer logs for all requests from the httpRequest:remoteIp in the 6 hours before and after the alert to identify attack patterns and other exploit attempts
  2. Check if the httpRequest:remoteIp is associated with known threat actors, scanning infrastructure, VPN services, or proxy networks using IP enrichment
  3. Search for other Cloud Armor denials (jsonPayload:statusDetails = denied_by_security_policy) from this IP or targeting the same resource:labels:backend_service_name in the past 7 days to assess campaign scope

from panther_gcp_helpers import gcp_alert_context


def rule(event):
    if event.get("severity") == "ERROR":
        return False

    method_name = event.deep_get("protoPayload", "methodName", default="")
    if not method_name.endswith("Services.CreateService"):
        return False

    authorization_info = event.deep_walk("protoPayload", "authorizationInfo")
    if not authorization_info:
        return False

    for auth in authorization_info:
        if auth.get("permission") == "run.services.create" and auth.get("granted") is True:
            return True
    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    project_id = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")

    return f"[GCP]: [{actor}] created new Run Service in project [{project_id}]"


def alert_context(event):
    context = gcp_alert_context(event)
    context["service_account"] = event.deep_get(
        "protoPayload",
        "request",
        "service",
        "spec",
        "template",
        "spec",
        default="<SERVICE_ACCOUNT_NOT_FOUND>",
    )
    return context

AnalysisType: rule
LogTypes:
  - GCP.AuditLog
Description:
  Detects creation of new Cloud Run Service, which, if configured maliciously, may be part of the attack
  aimed to invoke the service and retrieve the access token.
DisplayName: "GCP Cloud Run Service Created"
RuleID: "GCP.Cloud.Run.Service.Created"
Filename: gcp_cloud_run_service_created.py
Enabled: true
CreateAlert: false
Reference: https://cloud.google.com/run/docs/quickstarts/deploy-container
Runbook: Confirm this was authorized and necessary behavior
Severity: Low
DedupPeriodMinutes: 60
Threshold: 1

AnalysisType: correlation_rule
RuleID: "GCP.Cloud.Run.Service.Created.WITH.Set.IAM.Policy"
DisplayName: "GCP Cloud Run Service Created WITH Set IAM Policy"
Enabled: false
Severity: High
Description: Detects run.services.create method for privilege escalation in GCP. The exploit creates a new Cloud Run 
  Service that, when invoked, returns the Service Account's access token by accessing the metadata API of the server 
  it is running on.
Reference: https://rhinosecuritylabs.com/gcp/privilege-escalation-google-cloud-platform-part-1/
Runbook: Confirm this was authorized and necessary behavior
Reports:
    MITRE ATT&CK:
        - TA0004:T1548  # Abuse Elevation Control Mechanism
Detection:
    - Group:
        - ID: ServiceCreated
          RuleID: GCP.Cloud.Run.Service.Created
        - ID: SetIAMPolicy
          RuleID: GCP.Cloud.Run.Set.IAM.Policy
      MatchCriteria:
        field_name:
          - GroupID: ServiceCreated
            Match: p_alert_context.caller_ip
          - GroupID: SetIAMPolicy
            Match: p_alert_context.caller_ip
      LookbackWindowMinutes: 1800
      Schedule:
        RateMinutes: 1440
        TimeoutMinutes: 5

from panther_gcp_helpers import gcp_alert_context


def rule(event):
    if event.get("severity") == "ERROR":
        return False

    method_name = event.deep_get("protoPayload", "methodName", default="")
    if not method_name.endswith("Services.SetIamPolicy"):
        return False

    authorization_info = event.deep_walk("protoPayload", "authorizationInfo")
    if not authorization_info:
        return False

    for auth in authorization_info:
        if auth.get("permission") == "run.services.setIamPolicy" and auth.get("granted") is True:
            return True
    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    resource = event.deep_get("protoPayload", "resourceName", default="<RESOURCE_NOT_FOUND>")
    service_name = resource.split("/")[-1] if "/" in resource else resource

    # Extract roles from the response bindings - there could be multiple
    bindings = event.deep_get("protoPayload", "response", "bindings", default=[])

    # Handle multiple roles if present
    roles = []
    for binding in bindings:
        if binding.get("role"):
            roles.append(binding.get("role"))

    # Format roles for title
    if not roles:
        roles_str = "<NO_ROLES_FOUND>"
    elif len(roles) == 1:
        roles_str = roles[0]
    else:
        # If multiple roles, mention the count and list the first one
        roles_str = f"{len(roles)} roles including {roles[0]}"

    project_id = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")

    return (
        f"[GCP]: [{actor}] modified IAM policy for Cloud Run service [{service_name}] "
        f"with [{roles_str}] in project [{project_id}]"
    )


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

    # Extract the service name from the resource path for better context
    resource = event.deep_get("protoPayload", "resourceName", default="")
    if resource:
        context["service_name"] = resource.split("/")[-1] if "/" in resource else resource

    # Get bindings and role information
    bindings = event.deep_get("protoPayload", "response", "bindings", default=[])

    # Collect all roles and members
    all_roles = []
    all_members = []
    role_to_members = {}

    for binding in bindings:
        role = binding.get("role")
        members = binding.get("members", [])

        if role:
            all_roles.append(role)

        if members:
            all_members.extend(members)

            # Create mapping of role to members
            if role:
                role_to_members[role] = members

    # Store all collected information in the context
    context["assigned_roles"] = all_roles
    context["members_granted"] = all_members
    context["role_to_members_mapping"] = role_to_members

    return context

AnalysisType: rule
LogTypes:
  - GCP.AuditLog
Description:
  Detects new roles granted to users to Cloud Run Services. This could potentially allow the user to perform
  actions within the project and its resources, which could pose a security risk.
DisplayName: "GCP Cloud Run Set IAM Policy"
RuleID: "GCP.Cloud.Run.Set.IAM.Policy"
Enabled: true
Filename: gcp_cloud_run_set_iam_policy.py
Reference: https://cloud.google.com/run/docs/securing/managing-access
Runbook: Confirm this was authorized and necessary behavior
Severity: High
DedupPeriodMinutes: 60
Threshold: 1

BUCKET_OPERATIONS = ["storage.buckets.delete", "storage.buckets.update"]


def rule(event):
    return all(
        [
            event.deep_get("protoPayload", "serviceName", default="") == "storage.googleapis.com",
            event.deep_get("protoPayload", "methodName", default="") in BUCKET_OPERATIONS,
            event.get("severity") != "ERROR",
        ]
    )


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    operation = event.deep_get("protoPayload", "methodName", default="<OPERATION_NOT_FOUND>")
    project = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")
    bucket = event.deep_get("resource", "labels", "bucket_name", default="<BUCKET_NOT_FOUND>")

    return f"GCP: [{actor}] performed a [{operation}] on bucket [{bucket}] in project [{project}]."


def alert_context(event):
    return {
        "actor": event.deep_get("protoPayload", "authenticationInfo", "principalEmail"),
        "bucket": event.deep_get("resource", "labels", "bucket_name"),
        "source_ip": event.deep_get("protoPayload", "requestMetadata", "callerIp"),
        "user_agent": event.deep_get("protoPayload", "requestMetadata", "callerSuppliedUserAgent"),
        "project": event.deep_get("resource", "labels", "project_id"),
    }

AnalysisType: rule
DisplayName: "GCP Cloud Storage Buckets Modified Or Deleted"
Enabled: true
Filename: gcp_cloud_storage_buckets_modified_or_deleted.py
Description: >
  Detects when a GCS bucket configuration is updated or deleted. Bucket configuration changes can be
  part of a ransomware attack, such as disabling security settings to prevent data recovery.
Runbook: |
  1. Query GCP Audit logs for all bucket operations by the principal email in the 24 hours before and after this alert
  2. Check if the source IP is associated with known cloud provider IP ranges, VPN endpoints, or matches the user's typical access patterns
  3. Identify what specific configuration was changed (encryption, retention, versioning, lifecycle policies)
  4. Search for related KMS key changes or IAM policy modifications on the same bucket in the past 6 hours
  5. Look for bulk object operations (rewrite, copy, delete) on this bucket following the configuration change
Reference: https://cloud.google.com/storage/docs/json_api/v1/buckets/update
Tags:
  - GCP
  - Google Cloud Storage
  - Defense Evasion:Impair Defenses
  - Impact:Data Encrypted for Impact
  - Ransomware
Reports:
  MITRE ATT&CK:
    - TA0005:T1562
    - TA0040:T1486
SummaryAttributes:
  - severity
  - p_any_ip_addresses
  - p_any_emails
Severity: Low
DedupPeriodMinutes: 60
LogTypes:
  - GCP.AuditLog
RuleID: "GCP.Cloud.Storage.Buckets.Modified.Or.Deleted"
Threshold: 1

from panther_gcp_helpers import gcp_alert_context


def rule(event):
    if not event.deep_get("protoPayload", "methodName", default="METHOD_NOT_FOUND").endswith(
        "CloudBuild.CreateBuild"
    ):
        return False

    authorization_info = event.deep_walk("protoPayload", "authorizationInfo")
    if not authorization_info:
        return False

    # Get the principal (actor) email
    principal = event.deep_get("protoPayload", "authenticationInfo", "principalEmail", default="")

    # Skip whitelisted service accounts
    if principal.endswith("@gcf-admin-robot.iam.gserviceaccount.com"):
        return False

    # Check if build.create permission was granted
    for auth in authorization_info:
        if auth.get("permission") == "cloudbuild.builds.create" and auth.get("granted") is True:
            return True
    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    operation = event.deep_get("protoPayload", "methodName", default="<OPERATION_NOT_FOUND>")
    project_id = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")

    return f"[GCP]: [{actor}] performed [{operation}] on project [{project_id}]"


def alert_context(event):
    return gcp_alert_context(event)

AnalysisType: rule
LogTypes:
  - GCP.AuditLog
Description:
  Detects privilege escalation attacks designed to gain access to the Cloud Build Service Account.
  A user with permissions to start a new build with Cloud Build can gain access to the Cloud Build Service Account
  and abuse it for more access to the environment.
DisplayName: "GCP CloudBuild Potential Privilege Escalation"
RuleID: "GCP.CloudBuild.Potential.Privilege.Escalation"
Enabled: true
Filename: gcp_cloudbuild_potential_privilege_escalation.py
Reference: https://rhinosecuritylabs.com/gcp/iam-privilege-escalation-gcp-cloudbuild/
Runbook:
  Confirm this was authorized and necessary behavior. To defend against this privilege escalation attack,
  it is necessary to restrict the permissions granted to the Cloud Build Service Account and to be careful granting
  the cloudbuild.builds.create permission to any users in your Organization. Most importantly, you need to know that
  any user who is granted cloudbuild.builds.create, is also indirectly granted all the permissions granted to the
  Cloud Build Service Account. If that’s alright with you, then you may not need to worry about this attack vector,
  but it is still highly recommended to modify the default permissions granted to the Cloud Build Service Account.
Reports:
  MITRE ATT&CK:
    - TA0004:T1548 # Abuse Elevation Control Mechanism
Severity: High
DedupPeriodMinutes: 60
Threshold: 1

from panther_gcp_helpers import gcp_alert_context


def rule(event):
    authorization_info = event.deep_walk("protoPayload", "authorizationInfo")
    if not authorization_info:
        return False

    for auth in authorization_info:
        if (
            auth.get("permission") == "cloudfunctions.functions.create"
            and auth.get("granted") is True
        ):
            return True
    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    operation = event.deep_get("protoPayload", "methodName", default="<OPERATION_NOT_FOUND>")
    project_id = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")

    return f"[GCP]: [{actor}] performed [{operation}] on project [{project_id}]"


def alert_context(event):
    return gcp_alert_context(event)

AnalysisType: rule
RuleID: "GCP.Cloudfunctions.Functions.Create"
DisplayName: "GCP cloudfunctions functions create"
Description:
  The Identity and Access Management (IAM) service manages authorization and authentication for
  a GCP environment. This means that there are very likely multiple privilege escalation methods that use
  the IAM service and/or its permissions.
Enabled: true
Filename: gcp_cloudfunctions_functions_create.py
LogTypes:
  - GCP.AuditLog
Severity: High
DedupPeriodMinutes: 60
Threshold: 1
Reference: https://rhinosecuritylabs.com/gcp/privilege-escalation-google-cloud-platform-part-1/
Runbook:
  Confirm this was authorized and necessary behavior. This is not a vulnerability in GCP, it is a vulnerability
  in how GCP environment is configured, so it is necessary to be aware of these attack vectors and to defend against
  them. It’s also important to remember that privilege escalation does not necessarily need to pass through the
  IAM service to be effective. Make sure to follow the principle of least-privilege in your environments to help
  mitigate these security risks.
Reports:
  MITRE ATT&CK:
    - TA0004:T1548

from panther_gcp_helpers import gcp_alert_context


def rule(event):
    authorization_info = event.deep_walk("protoPayload", "authorizationInfo")
    if not authorization_info:
        return False

    for auth in authorization_info:
        if (
            auth.get("permission") == "cloudfunctions.functions.update"
            and auth.get("granted") is True
        ):
            return True
    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    operation = event.deep_get("protoPayload", "methodName", default="<OPERATION_NOT_FOUND>")
    project_id = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")

    return f"[GCP]: [{actor}] performed [{operation}] on project [{project_id}]"


def alert_context(event):
    return gcp_alert_context(event)

AnalysisType: rule
RuleID: "GCP.Cloudfunctions.Functions.Update"
DisplayName: "GCP cloudfunctions functions update"
Description:
  The Identity and Access Management (IAM) service manages authorization and authentication for
  a GCP environment. This means that there are very likely multiple privilege escalation methods that use
  the IAM service and/or its permissions.
Enabled: true
Filename: gcp_cloudfunctions_functions_update.py
LogTypes:
  - GCP.AuditLog
Severity: High
DedupPeriodMinutes: 60
Threshold: 1
Reference: https://rhinosecuritylabs.com/gcp/privilege-escalation-google-cloud-platform-part-1/
Runbook:
  Confirm this was authorized and necessary behavior. This is not a vulnerability in GCP, it is a vulnerability
  in how GCP environment is configured, so it is necessary to be aware of these attack vectors and to defend against
  them. It’s also important to remember that privilege escalation does not necessarily need to pass through the
  IAM service to be effective. Make sure to follow the principle of least-privilege in your environments to help
  mitigate these security risks.
Reports:
  MITRE ATT&CK:
    - TA0004:T1548

from panther_gcp_helpers import gcp_alert_context, get_binding_deltas

SUSPICIOUS_ACTIONS = [
    "v1.compute.disks.setIamPolicy",
    "v1.compute.images.setIamPolicy",
    "v1.compute.snapshots.setIamPolicy",
]


def rule(event):
    if event.deep_get("protoPayload", "response", "error"):
        return False

    method = event.deep_get("protoPayload", "methodName", default="METHOD_NOT_FOUND")
    if method in SUSPICIOUS_ACTIONS:
        return True

    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )

    items = event.deep_get("protoPayload", "methodName", default="ITEMS_NOT_FOUND. ").split(".")[-2]

    project = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")
    return f"[GCP]: [{actor}] updated IAM policy for [{items}] on project [{project}]"


def alert_context(event):
    context = gcp_alert_context(event)
    service_accounts = event.deep_get("protoPayload", "request", "serviceAccounts")
    if not service_accounts:
        service_account_emails = "<SERVICE_ACCOUNT_EMAILS_NOT_FOUND>"
    else:
        service_account_emails = [service_acc["email"] for service_acc in service_accounts]
    context["serviceAccount"] = service_account_emails
    context["binding_deltas"] = get_binding_deltas(event)
    return context

AnalysisType: rule
DedupPeriodMinutes: 60
DisplayName: GCP Compute IAM Policy Update Detection
Enabled: true
Filename: gcp_compute_set_iam_policy.py
RuleID: "GCP.Compute.IAM.Policy.Update"
Severity: Medium
LogTypes:
  - GCP.AuditLog
Description: >
  This rule detects updates to IAM policies for Compute Disks, Images, and Snapshots.
Runbook: >
  Ensure that the IAM policy update was expected. Unauthorized changes can lead to security risks.
Reference: https://cloud.google.com/compute/docs/access/iam

from panther_core import PantherEvent
from panther_gcp_helpers import gcp_alert_context


def rule(event: PantherEvent) -> bool:
    service_name = event.deep_get("protoPayload", "serviceName", default="")
    method_name = event.deep_get("protoPayload", "methodName", default="")

    if service_name == "iap.googleapis.com" and method_name == "AuthorizeUser":
        return True

    if service_name == "oslogin.googleapis.com":
        if any([method_name.endswith(".CheckPolicy"), method_name.endswith(".ContinueSession")]):
            return True

    if service_name == "compute.googleapis.com":
        # Check attempts to add SSH keys to the VM
        # setCommonInstanceMetadata is triggered when SSHing from a remote device
        # setMetadata is triggered when SSHing from the GCP Console
        ssh_keys = {"ssh-keys", "sshKeys"}  # Fields indicating the SSH keys were modified
        if any(
            [
                method_name.endswith(".setCommonInstanceMetadata"),
                method_name.endswith(".setMetadata"),
            ]
        ):
            # The metadata delta field could be for the project or the instance, and could indicate
            # something was removed or modified. We need to check all possible paths to the field
            # we need.
            modified_keys = set()
            for field1 in ["projectMetadataDelta", "instanceMetadataDelta"]:
                for field2 in ["addedMetadataKeys", "modifiedMetadataKeys"]:
                    modified_keys.update(
                        set(event.deep_get("protoPayload", "metadata", field1, field2, default=[]))
                    )
            return bool(modified_keys & ssh_keys)

    # Check direct connections to the serial console
    # The actual service name has the region included, so we just so an easy check here
    if service_name.endswith("ssh-serialport.googleapis.com"):
        if method_name == "google.ssh-serialport.v1.connect":
            return (
                "succeeded"
                in event.deep_get("protoPayload", "status", "message", default="").lower()
            )

    return False


def alert_context(event: PantherEvent) -> dict:
    instance_info = get_instance_info(event)
    context = {
        "instance_id": instance_info.get("id", "UNKNOWN INSTANCE ID"),
        "instance_name": instance_info.get("name", "UNKNOWN INSTANCE NAME"),
    }
    return gcp_alert_context(event) | context


def get_instance_info(event: PantherEvent) -> dict:
    service_name = event.deep_get("protoPayload", "serviceName", default="")

    context = {
        "id": "UNKNOWN INSTANCE ID",
        "name": "UNKNOWN INSTANCE NAME",
    }
    match service_name:
        case "iap.googleapis.com":
            # Name is not included in the event
            context |= {
                "id": event.deep_get(
                    "resource", "labels", "instance_id", default="UNKNOWN INSTANCE ID"
                )
            }
        case "oslogin.googleapis.com":
            context |= {
                "id": event.deep_get("labels", "instance_id", default="UNKNOWN INSTANCE ID"),
                "name": event.deep_get(
                    "protoPayload", "request", "instance", default="UNKNOWN INSTANCE NAME"
                ),
            }
        case "compute.googleapis.com":
            if event.deep_get("protoPayload", "methodName", default="").endswith(
                ".setCommonInstanceMetadata"
            ):
                # These events are targeted prokect-wide, so they don't have information about
                # specific instances.
                pass
            else:
                context |= {
                    # Will look like: projects/project-name/zones/zone-name/instances/instance-name
                    "name": event.deep_get(
                        "protoPayload", "resourceName", default="/UNKNOWN INSTANCE NAME"
                    ).split("/")[-1],
                }

    if "serialport" in service_name:
        context |= {
            "id": event.deep_get(
                "resource", "labels", "instance_id", default="UNKNOWN INSTANCE ID"
            ),
            # Will look like:
            # projects/projectName/zones/zoneName/instances/instanceName/SerialPort/portNum
            "name": event.deep_get(
                "protoPayload", "resourceName", default="/UNKNOWN INSTANCE NAME//"
            ).split("/")[-3],
        }
    return context

AnalysisType: rule
Filename: gcp_compute_ssh_connection.py
RuleID: "GCP.Compute.SSHConnection"
DisplayName: GCP Compute SSH Connection
Enabled: true
LogTypes:
  - GCP.AuditLog
Severity: Info
CreateAlert: true
Description: >
  Detect any SSH connections to a Compute Instance.
Reference: >
  https://cloud.google.com/compute/docs/connect/ssh-best-practices/auditing
Tags:
  - GCP
  - GCP.AuditLog
  - SSH
  - Compute
Status: Experimental

from panther_gcp_helpers import gcp_alert_context

REQUIRED_PERMISSIONS = [
    "compute.disks.create",
    "compute.instances.create",
    "compute.instances.setMetadata",
    "compute.instances.setServiceAccount",
    "compute.subnetworks.use",
    "compute.subnetworks.useExternalIp",
]


def rule(event):
    if event.deep_get("protoPayload", "response", "error"):
        return False

    method = event.deep_get("protoPayload", "methodName", default="METHOD_NOT_FOUND")
    if not method.endswith("compute.instances.insert"):
        return False

    # Skip allowlisted actors
    principal = event.deep_get("protoPayload", "authenticationInfo", "principalEmail", default="")
    if principal.endswith("@cloudservices.gserviceaccount.com"):
        return False

    granted_permissions = {}
    for auth in event.deep_walk("protoPayload", "authorizationInfo") or []:
        granted_permissions[auth.get("permission")] = auth.get("granted")
    for permission in REQUIRED_PERMISSIONS:
        if not granted_permissions.get(permission):
            return False

    return True


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )

    service_accounts = event.deep_get("protoPayload", "request", "serviceAccounts")
    if not service_accounts:
        service_account_emails = "<SERVICE_ACCOUNT_EMAILS_NOT_FOUND>"
    else:
        service_account_emails = [service_acc["email"] for service_acc in service_accounts]

    project = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")
    return (
        f"[GCP]: [{actor}] created a new Compute Engine instance with [{service_account_emails}] "
        f"Service Account on project [{project}]"
    )


def alert_context(event):
    context = gcp_alert_context(event)
    service_accounts = event.deep_get("protoPayload", "request", "serviceAccounts")
    if not service_accounts:
        service_account_emails = "<SERVICE_ACCOUNT_EMAILS_NOT_FOUND>"
    else:
        service_account_emails = [service_acc["email"] for service_acc in service_accounts]
    context["serviceAccount"] = service_account_emails
    return context

AnalysisType: rule
LogTypes:
  - GCP.AuditLog
Description: >
  Detects compute.instances.create method for privilege escalation in GCP. This rule identifies when users
  create compute instances with service accounts that may lead to privilege escalation. Known good service accounts
  (GKE, Kubernetes, compute automation) are excluded to reduce false positives.
DisplayName: "GCP compute.instances.create Privilege Escalation"
RuleID: "GCP.compute.instances.create.Privilege.Escalation"
Enabled: true
Reference: https://rhinosecuritylabs.com/gcp/privilege-escalation-google-cloud-platform-part-1/
Runbook: |
  1. Validate whether this compute instance creation with service account was authorized.
  2. Check if the service account attached has excessive privileges.
  3. Verify if the user creating the instance has a legitimate need for the service account permissions.
  4. If unauthorized, revoke the instance access and investigate for compromise.
Reports:
  MITRE ATT&CK:
    - TA0004:T1548 # Abuse Elevation Control Mechanism
Severity: High
Filename: gcp_computeinstances_create_privilege_escalation.py
DedupPeriodMinutes: 1440
Threshold: 1

from fnmatch import fnmatch

from panther_base_helpers import deep_get

# These patterns indicate members which might be added by default by some GCP services
ACCEPTED_MEMBER_PATTERNS = [
    "serviceAccount:*@*.gserviceaccount.com",
    "serviceAccount:*.svc.id.goog[*",
    "principalSet://iam.googleapis.com/projects/*/workloadIdentityPools/*",
]


def rule(event):
    if event.deep_get("protoPayload", "methodName") != "SetIamPolicy":
        return False

    service_data = event.deep_get("protoPayload", "serviceData")
    if not service_data:
        return False

    authenticated = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default=""
    )
    expected_domain = authenticated.split("@")[-1]

    binding_deltas = deep_get(service_data, "policyDelta", "bindingDeltas")
    if not binding_deltas:
        return False

    for delta in binding_deltas:
        if delta.get("action") != "ADD":
            continue
        member = delta.get("member", "")
        if any(fnmatch(member, pattern) for pattern in ACCEPTED_MEMBER_PATTERNS):
            continue  # Skip this member, check others
        if member.endswith(f"@{expected_domain}"):
            continue  # Skip this member, check others
        return True  # Found a suspicious member - alert
    return False  # No suspicious members found


def title(event):
    return (
        f"A GCP IAM account has been created with an unexpected email domain in "
        f"{event.deep_get('resource', 'labels', 'project_id', default='<UNKNOWN_PROJECT>')}"
    )

AnalysisType: rule
Filename: gcp_iam_corp_email.py
RuleID: "GCP.IAM.CorporateEmail"
DisplayName: "GCP Corporate Email Not Used"
Enabled: true
DedupPeriodMinutes: 720 # 12 hours
LogTypes:
  - GCP.AuditLog
Tags:
  - GCP
  - Identity & Access Management
  - Persistence:Create Account
Reports:
  MITRE ATT&CK:
    - TA0003:T1136
  CIS:
    - 1.1
Severity: Low
Description: Unexpected domain is being used instead of a corporate email
Runbook: Remove the user
Reference: https://cloud.google.com/iam/docs/service-account-overview
SummaryAttributes:
  - severity
  - p_any_ip_addresses
  - p_any_domain_names

DESTRUCTIVE_STATEMENTS = ["UPDATE", "DELETE", "DROP_TABLE", "ALTER_TABLE", "TRUNCATE_TABLE"]


def rule(event):
    if all(
        [
            event.deep_get("resource", "type", default="<RESOURCE_NOT_FOUND>").startswith(
                "bigquery"
            ),
            event.deep_get("protoPayload", "metadata", "jobChange", "job", "jobConfig", "type")
            == "QUERY",
            event.deep_get(
                "protoPayload",
                "metadata",
                "jobChange",
                "job",
                "jobConfig",
                "queryConfig",
                "statementType",
                default="<STATEMENT_NOT_FOUND>",
            )
            in DESTRUCTIVE_STATEMENTS,
        ]
    ):
        return True

    if event.deep_get("protoPayload", "metadata", "tableDeletion"):
        return True

    if event.deep_get("protoPayload", "metadata", "datasetDeletion"):
        return True

    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    statement = event.deep_get(
        "protoPayload",
        "metadata",
        "jobChange",
        "job",
        "jobConfig",
        "queryConfig",
        "statementType",
        default="<STATEMENT_NOT_FOUND>",
    )
    if (
        event.deep_get("protoPayload", "metadata", "jobChange", "job", "jobConfig", "type")
        == "QUERY"
    ):
        return f"GCP: [{actor}] performed a destructive BigQuery [{statement}] query"

    if event.deep_get("protoPayload", "metadata", "tableDeletion"):
        return f"GCP: [{actor}] deleted a table in BigQuery"

    if event.deep_get("protoPayload", "metadata", "datasetDeletion"):
        return f"GCP: [{actor}] deleted a dataset in BigQuery"

    # Default return value
    return f"GCP: [{actor}] performed a destructive BigQuery query"


def severity(event):
    statement = event.deep_get(
        "protoPayload",
        "metadata",
        "jobChange",
        "job",
        "jobConfig",
        "queryConfig",
        "statementType",
        default="<STATEMENT_NOT_FOUND>",
    )
    if statement in ("UPDATE", "DELETE"):
        return "INFO"
    return "DEFAULT"


def alert_context(event):
    return {
        "query": event.deep_get(
            "protoPayload",
            "metadata",
            "jobChange",
            "job",
            "jobConfig",
            "queryConfig",
            "query",
            default="<QUERY_NOT_FOUND>",
        ),
        "actor": event.deep_get(
            "protoPayload",
            "authenticationInfo",
            "principalEmail",
            default="<ACTOR_NOT_FOUND>",
        ),
        "statement": event.deep_get(
            "protoPayload",
            "metadata",
            "jobChange",
            "job",
            "jobConfig",
            "queryConfig",
            "statementType",
            default="<STATEMENT_NOT_FOUND>",
        ),
        "table": event.deep_get(
            "protoPayload",
            "metadata",
            "jobChange",
            "job",
            "jobConfig",
            "queryConfig",
            "destinationTable",
        )
        or event.deep_get("protoPayload", "metadata", "resourceName", default="<TABLE_NOT_FOUND>"),
    }

AnalysisType: rule
Description: Detect any destructive BigQuery queries or jobs such as update, delete, drop, alter or truncate.
DisplayName: "GCP Destructive Queries"
Enabled: true
Filename: gcp_destructive_queries.py
Reference: https://cloud.google.com/bigquery/docs/managing-tables
Severity: Info
SummaryAttributes:
  - p_alert_context.table
DedupPeriodMinutes: 60
LogTypes:
  - GCP.AuditLog
RuleID: "GCP.Destructive.Queries"
Threshold: 1