本頁面由 Cloud Translation API 翻譯而成。

剖析器擴充功能範例

支援的國家/地區：

Google SecOps SIEM

這份文件提供不同情境下的剖析器擴充功能建立範例。如要進一步瞭解剖析器擴充功能，請參閱「建立剖析器擴充功能」。

剖析器擴充功能範例

請使用下列屬性表格，快速找出所需的範例程式碼。

無程式碼範例

記錄來源格式	標題範例	說明	本例中的剖析器概念
JSON (記錄類型：`GCP_IDS`)	擷取欄位	從 JSON 格式的記錄中擷取欄位。	無程式碼
JSON (記錄類型：`WORKSPACE_ALERTS`)	使用前提條件值擷取欄位	從 JSON 格式的記錄中擷取欄位，並在符合前提條件的情況下，將其正規化為重複的 UDM 欄位。	重複的 UDM 欄位無程式碼先決條件

程式碼片段範例

記錄來源格式	標題範例	說明	本例中的剖析器概念
JSON (記錄類型：`GCP_IDS`)	新增 HTTP 使用者代理程式	擷取網路 HTTP 剖析器使用者代理程式，並從 `requestUrl` 建立 `target hostname`。指派命名空間，允許使用別名和擴充功能。
CSV (記錄類型：MISP_IOC)	將任意欄位擷取至 `additional` UDM 物件	將欄位擷取到 UDM > 實體 > `additional` UDM 物件 > 鍵/值組合	`additional` UDM 物件
系統記錄檔 (記錄類型：POWERSHELL)	從系統記錄中擷取優先順序和嚴重程度	將 Syslog Facility 和 Severity 值擷取至「UDM Security Result Priority」和「Severity」欄位。	以 Grok 為基礎
含有 Syslog 標頭的 JSON (記錄類型：WINDOWS_SYSMON)	根據條件陳述式套用裝飾	根據條件陳述式，並瞭解程式碼片段中的資料型別，在 `metadata.description` 欄位中加入裝飾 (內容資訊)。使用擷取篩選器時，系統可能會保留原始資料類型。 Grok 條件陳述式應使用原始資料類型評估欄位。	以 Grok 為基礎 Grok 條件陳述式系統可能會保留擷取欄位的原始資料類型。 Grok 條件陳述式應使用原始資料類型評估欄位。
含有 Syslog 標頭的 JSON (記錄類型：WINDOWS_SYSMON)	轉換資料類型	使用 `convert` 函式，在剖析器擴充功能中轉換資料類型。使用 `on_error` 陳述式妥善處理錯誤，避免剖析器擴充功能失敗。	以 Grok 為基礎轉換資料類型使用 `on_error` 陳述式提供錯誤處理機制。
含有 Syslog 標頭的 JSON (記錄類型：WINDOWS_SYSMON)	方便閱讀的暫時變數名稱	您可以在程式碼片段中使用暫時變數名稱，之後再重新命名，與最終輸出 UDM 事件物件名稱相符。這有助於提升整體可讀性。	以 Grok 為基礎使用暫時變數名稱，稍後再將其重新命名為最終輸出 UDM 名稱。
含有 Syslog 標頭的 JSON (記錄類型：WINDOWS_SYSMON)	重複欄位	在程式碼片段中使用重複欄位時，請務必謹慎，例如 security_result 欄位。	以 Grok 為基礎重複的 UDM 欄位
XML (記錄類型：WINDOWS_DEFENDER_AV)	將任意欄位擷取至 `additional` 物件	擷取並儲存平台版本值，例如，以便回報及搜尋過時的平台版本。在本例中，沒有合適的標準 UDM 欄位，因此 `additional` 物件會將資訊儲存為自訂鍵/值配對。	`additional` 物件用於以自訂鍵/值組合的形式儲存資訊。
XML (記錄類型：WINDOWS_DEFENDER_AV)	將任意欄位擷取至主體主機名稱	從 FQDN 擷取主機名稱。條件式處理用於判斷是否應覆寫 `principal.hostname` 欄位。 Grok 陳述式會使用規則運算式 (regex) 擷取 `hostname` 欄位。規則運算式本身會使用具名擷取群組，也就是說，括號內相符的任何內容都會儲存在名為 `hostname` 的欄位中，比對一或多個字元，直到遇到半形句號為止。這只會擷取 FQDN 內的 `hostname`。 Grok `overwrite` 陳述不過，執行「PREVIEW UDM OUTPUT」(預覽 UDM 輸出) 時，系統會傳回錯誤：「LOG_PARSING_CBN_ERROR: Field `hostname` already exists in data and is not overwritable」(LOG_PARSING_CBN_ERROR：資料中已存在欄位 `hostname`，且無法覆寫)。在 Grok 陳述式中，具名擷取群組無法覆寫現有變數，除非使用 `overwrite` 陳述式明確指定。在這種情況下，我們可以在 Grok 陳述式中為具名擷取群組使用不同的變數名稱，也可以 (如本例所示) 使用覆寫陳述式，明確覆寫現有的主機名稱變數。	以 Grok 為基礎條件式處理用於判斷是否應覆寫欄位。使用規則運算式 (regex) 的 Grok 陳述式。 Grok `overwrite` 陳述式
JSON、CSV、XML、Syslog 和 KV	移除現有對應	如要移除現有對應，請移除 UDM 欄位的值。

JSON 範例

下列範例說明如何建立剖析器擴充功能，其中記錄來源採用 JSON 格式。

無程式碼 - 擷取欄位

屬性範例：

記錄來源格式：JSON
資料對應方法：無程式碼
記錄類型：GCP_IDS
剖析器擴充功能的用途：擷取欄位。

說明：

系統不會擷取幾個與網路相關的欄位。由於這個記錄檔範例是 JSON 格式的結構化記錄，我們可以採用無程式碼 (對應資料欄位) 方法建立剖析器擴充功能。

我們要擷取的原始欄位如下：

total_packets (字串)
elapsed_time (字串)
total_bytes (字串)

原始記錄項目範例如下：

{
"insertId": "625a41542d64c124e7db097ae0906ccb-1@a3",
"jsonPayload": {
  "destination_port": "80",
  "application": "incomplete",
  "ip_protocol": "tcp",
  "network": "projects/prj-p-shared-base/global/networks/shared-vpc-production",
  "start_time": "2024-10-29T21:14:59Z",
  "source_port": "41936",
  "source_ip_address": "35.191.200.157",
  "total_packets": "6",
  "elapsed_time": "0",
  "destination_ip_address": "192.168.0.11",
  "total_bytes": "412",
  "repeat_count": "1",
  "session_id": "1289742"
},
"resource": {
  "type": "ids.googleapis.com/Endpoint",
  "labels": {
    "resource_container": "projects/12345678910",
    "location": "europe-west4-a",
    "id": "p-europe-west4"
  }
},
"timestamp": "2024-10-29T21:15:21Z",
"logName": "projects/prj-p-shared-base/logs/ids.googleapis.com%2Ftraffic",
"receiveTimestamp": "2024-10-29T21:15:24.051990717Z"
}

這個範例採用無程式碼方法，使用下列資料欄位對應建立剖析器擴充功能：

先決條件路徑	前置條件運算子	先決條件值	原始資料路徑	目的地欄位*
`jsonPayload.total_bytes`	NOT_EQUALS	""	`jsonPayload.total_bytes`	`udm.principal.network.received_bytes`
`jsonPayload.elapsed_time`	NOT_EQUALS	""	`jsonPayload.elapsed_time`	`udm.principal.network.session_duration.seconds`
`jsonPayload.total_packets`	NOT_EQUALS	""	`jsonPayload.total_packets`	`udm.principal.network.received_packets`

執行剖析器擴充功能後，系統會成功將三個擷取的欄位新增至 principal.network 物件。

metadata.product_log_id = "625a41542d64c124e7db097ae0906ccb-1@a3"
metadata.event_timestamp = "2024-10-29T21:14:59Z"
metadata.event_type = "NETWORK_CONNECTION"
metadata.vendor_name = "Google Cloud"
metadata.product_name = "IDS"
metadata.ingestion_labels[0].key = "label"
metadata.ingestion_labels[0].value = "GCP_IDS"
metadata.log_type = "GCP_IDS"
principal.ip[0] = "35.191.200.157"
principal.port = 41936
principal.network.received_bytes = 412
principal.network.session_duration.seconds = "0s"
principal.network.received_packets = 6
target.ip[0] = "192.168.0.11"
target.port = 80
target.application = "incomplete"
observer.location.country_or_region = "EUROPE"
observer.location.name = "europe-west4-a"
observer.resource.name = "projects/12345678910"
observer.resource.resource_type = "CLOUD_PROJECT"
observer.resource.attribute.cloud.environment = "GOOGLE_CLOUD_PLATFORM"
observer.resource.product_object_id = "p-europe-west4"
network.ip_protocol = "TCP"
network.session_id = "1289742"

免程式碼 - 使用前置條件值擷取欄位

屬性範例：

記錄來源格式：JSON
資料對應方法：無程式碼
記錄類型：WORKSPACE_ALERTS
剖析器擴充功能用途：使用前置條件值擷取欄位。

說明：

原始剖析器不會擷取受 DLP (資料遺失防護) 快訊影響的主要使用者 email address。

這個範例使用無程式碼剖析器擴充功能擷取 email address，並透過前置條件將其正規化為重複 UDM 欄位。

在無程式碼剖析器擴充功能中使用重複欄位時，您必須指出要：

replace (覆寫現有 UDM 物件中重複欄位的所有值)，或
附加 (將擷取的值附加至重複欄位)。

詳情請參閱「重複欄位」一節。

這個範例會取代標準化 principal.user.email_address 欄位中的所有現有電子郵件地址。

前提條件可讓您在執行擷取作業前，先執行條件式檢查。在大多數情況下，「前置條件欄位」會與您要擷取的「原始資料欄位」相同，且「前置條件運算子」為 not Null，例如 foo != ""。

不過，有時您想擷取的「原始資料欄位」值並非出現在所有記錄項目中，如我們的範例所示。在這種情況下，您可以使用另一個先決條件欄位來篩選擷取作業。在我們的範例中，您要擷取的原始 triggeringUserEmail 欄位只會出現在 type = Data Loss Prevention 的記錄中。

以下是範例值，請輸入至無程式碼剖析器擴充功能欄位：

先決條件路徑	前置條件運算子	先決條件值	原始資料路徑	目的地欄位*
`type`	等於	資料遺失防護	`data.ruleViolationInfo.triggeringUserEmail`	`udm.principal.user.email_addresses`

以下範例顯示以範例值填入的無程式碼剖析器擴充功能欄位：

成功執行剖析器擴充功能後，系統會將 email_address 新增至 principal.user 物件。

metadata.product_log_id = "Ug71LGqBr6Q="
metadata.event_timestamp = "2022-12-18T12:17:35.154368Z"
metadata.event_type = "USER_UNCATEGORIZED"
metadata.vendor_name = "Google Workspace"
metadata.product_name = "Google Workspace Alerts"
metadata.product_event_type = "DlpRuleViolation"
metadata.log_type = "WORKSPACE_ALERTS"
additional.fields["resource_title"] = "bq-results-20221215-112933-1671103787123.csv"
principal.user.email_addresses[0] = "foo.bar@altostrat.com"
target.resource.name = "DRIVE"
target.resource.resource_type = "STORAGE_OBJECT"
target.resource.product_object_id = "1wLteoF3VHljS_8_ABCD_VVbhFTfcTQplJ5k1k7cL4r8"
target.labels[0].key = "resource_title"
target.labels[0].value = "bq-results-20221321-112933-1671103787697.csv"
about[0].resource.resource_type = "CLOUD_ORGANIZATION"
about[0].resource.product_object_id = "C01abcde2"
security_result[0].about.object_reference.id = "ODU2NjEwZTItMWE2YS0xMjM0LWJjYzAtZTJlMWU2YWQzNzE3"
security_result[0].category_details[0] = "Data Loss Prevention"
security_result[0].rule_name = "Sensitive Projects Match"
security_result[0].summary = "Data Loss Prevention"
security_result[0].action[0] = "ALLOW"
security_result[0].severity = "MEDIUM"
security_result[0].rule_id = "rules/00abcdxs183abcd"
security_result[0].action_details = "ALERT, DRIVE_WARN_ON_EXTERNAL_SHARING"
security_result[0].alert_state = "ALERTING"
security_result[0].detection_fields[0].key = "start_time"
security_result[0].detection_fields[0].value = "2022-12-18T12:17:35.154368Z"
security_result[0].detection_fields[1].key = "status"
security_result[0].detection_fields[1].value = "NOT_STARTED"
security_result[0].detection_fields[2].key = "trigger"
security_result[0].detection_fields[2].value = "DRIVE_SHARE"
security_result[0].rule_labels[0].key = "detector_name"
security_result[0].rule_labels[0].value = "EMAIL_ADDRESS"
network.email.to[0] = "foo.bar@altostrat.com"

程式碼片段 - 新增 HTTP 使用者代理程式

屬性範例：

記錄來源格式：JSON
資料對應方法：程式碼片段
記錄類型：GCP_IDS
剖析器擴充功能用途：新增 HTTP 使用者代理程式。

說明：

這是非標準 UDM 物件類型的範例，無程式碼方法不支援這類物件，因此需要使用程式碼片段。預設剖析器不會擷取 Network HTTP Parser User Agent 分析。此外，為保持一致性：

系統會從 requestUrl 建立 Target Hostname。
系統會指派 Namespace，確保執行以資產為基礎的別名和擴充功能。

# GCP_LOADBALANCING
# owner: @owner
# updated: 2022-12-23
# Custom parser extension that:
# 1) adds consistent Namespace 
# 2) adds Parsed User Agent Object 
filter {
    # Initialize placeholder
    mutate {
        replace => {
            "httpRequest.userAgent" => ""
            "httpRequest.requestUrl" => ""
        }
    }
    json {
        on_error => "not_json"
        source => "message"
        array_function => "split_columns"
    }
    if ![not_json] {
      #1 - Override Namespaces
        mutate {
            replace => {
                "event1.idm.read_only_udm.principal.namespace" => "TMO"
            }
        }
        mutate {
            replace => {
                "event1.idm.read_only_udm.target.namespace" => "TMO"
            }
        }
        mutate {
            replace => {
                "event1.idm.read_only_udm.src.namespace" => "TMO"
            }
        }
        #2 - Parsed User Agent
        if [httpRequest][requestUrl]!= "" {
            grok {
                match => {
                    "httpRequest.requestUrl" => ["\/\/(?P<_hostname>.*?)\/"]
                }
                on_error => "_grok_hostname_failed"
            }
            if ![_grok_hostname_failed] {
                mutate {
                    replace => {
                        "event1.idm.read_only_udm.target.hostname" => "%{_hostname}"
                    }
                }
            }
        }
        if [httpRequest][userAgent] != "" {
            mutate {
                convert => {
                    "httpRequest.userAgent" => "parseduseragent"
                }
            }
            #Map the converted "user_agent" to the new UDM field "http.parsed_user_agent".
            mutate {
                rename => {
                    "httpRequest.userAgent" => "event1.idm.read_only_udm.network.http.parsed_user_agent"
                }
            }
        }
        mutate {
            merge => {
                "@output" => "event1"
            }
        }
    }
}

CSV 範例

以下範例說明如何建立剖析器擴充功能，其中記錄來源採用 CSV 格式。

程式碼片段 - 將任意欄位擷取到 `additional` 物件中

屬性範例：

記錄來源格式：CSV
資料對應方法：程式碼片段
記錄類型：MISP_IOC
剖析器擴充功能用途：將任意欄位擷取到 additional 物件中。

說明：

本例使用 MISP_IOC UDM 實體內容整合。additional 鍵/值對 UDM 物件可用來擷取預設剖析器未擷取的脈絡資訊，並新增每個機構專屬的欄位。例如，返回特定 MISP 執行個體的網址。

以下是本範例的 CSV 記錄來源：

`1`	`9d66d38a-14e1-407f-a4d1-90b82aa1d59f`
`2`	`3908`
`3`	`Network activity`
`4`	`ip-dst`
`5`	`117.253.154.123`
`6`
`7`
`8`	`1687894564`
`9`
`10`
`11`
`12`
`13`
`14`	`DigitalSide Malware report\: MD5\: 59ce0baba11893f90527fc951ac69912`
`15`	`ORGNAME`
`16`	`DIGITALSIDE.IT`
`17`	`0`
`18`	`Medium`
`19`	`0`
`20`	`2023-06-23`
`21`	`tlp:white,type:OSINT,source:DigitalSide.IT,source:urlhaus.abuse.ch`
`22`	`1698036218`

# MISP_IOC
# owner: @owner
# updated: 2024-06-21
# Custom parser extension that:
# 1) adds a link back to internal MISP tenant 
# 2) extracts missing fields into UDM > Entity > Additional fields
filter {
    # Set the base URL for MISP. Remember to replace this placeholder!
    mutate {
        replace => {
            "misp_base_url" => "https://<YOUR_MISP_URL>"
        }
    }
    # Parse the CSV data from the 'message' field. Uses a comma as the separator.
    # The 'on_error' option handles lines that are not properly formatted CSV.
    csv {
        source => "message"
        separator => ","
        on_error => "broken_csv"
    }
    # If the CSV parsing was successful...
    if ![broken_csv] {
        # Rename the CSV columns to more descriptive names.
        mutate {
            rename => {
                "column2" => "event_id"
                "column8" => "object_timestamp"
                "column16" => "event_source_org"
                "column17" => "event_distribution"
                "column19" => "event_analysis"
                "column22" => "attribute_timestamp"
            }
        }
    }
    # Add a link to view the event in MISP, if an event ID is available.
    # "column2" => "event_id"
    if [event_id] != "" {
        mutate {
            replace => {
                "additional_url.key" => "view_in_misp"
                "additional_url.value.string_value" => "%{misp_base_url}/events/view/%{event_id}"
            }
        }
        mutate {
            merge => {
                "event.idm.entity.additional.fields" => "additional_url"
            }
        }
    }
    # Add the object timestamp as an additional field, if available.
    # "column8" => "object_timestamp"
    if [object_timestamp] != "" {
        mutate {
            replace => {
                "additional_object_timestamp.key" => "object_timestamp"
                "additional_object_timestamp.value.string_value" => "%{object_timestamp}"
            }
        }
        mutate {
            merge => {
                "event.idm.entity.additional.fields" => "additional_object_timestamp"
            }
        }
    }
    # Add the event source organization as an additional field, if available.
    # "column16" => "event_source_org"
    if [event_source_org] != "" {
        mutate {
            replace => {
                "additional_event_source_org.key" => "event_source_org"
                "additional_event_source_org.value.string_value" => "%{event_source_org}"
            }
        }
        mutate {
            merge => {
                "event.idm.entity.additional.fields" => "additional_event_source_org"
            }
        }
    }
    # Add the event distribution level as an additional field, if available.
    # Maps numerical values to descriptive strings.
    # "column17" => "event_distribution"
    if [event_distribution] != "" {
        if [event_distribution] == "0" {
            mutate {
                replace => {
                    "additional_event_distribution.value.string_value" => "YOUR_ORGANIZATION_ONLY"
                }
            }
        } else if [event_distribution] == "1" {
            mutate {
                replace => {
                    "additional_event_distribution.value.string_value" => "THIS_COMMUNITY_ONLY"
                }
            }
        } else if [event_distribution] == "2" {
            mutate {
                replace => {
                    "additional_event_distribution.value.string_value" => "CONNECTED_COMMUNITIES"
                }
            }
        } else if [event_distribution] == "3" {
            mutate {
                replace => {
                    "additional_event_distribution.value.string_value" => "ALL_COMMUNITIES"
                }
            }
        } else if [event_distribution] == "4" {
            mutate {
                replace => {
                    "additional_event_distribution.value.string_value" => "SHARING_GROUP"
                }
            }
        } else if [event_distribution] == "5" {
            mutate {
                replace => {
                    "additional_event_distribution.value.string_value" => "INHERIT_EVENT"
                }
            }
        }
        mutate {
            replace => {
                "additional_event_distribution.key" => "event_distribution"
            }
        }
        mutate {
            merge => {
                "event.idm.entity.additional.fields" => "additional_event_distribution"
            }
        }
    }
    # Add the event analysis level as an additional field, if available.
    # Maps numerical values to descriptive strings.
    # "column19" => "event_analysis"
    if [event_analysis] != "" {
        if [event_analysis] == "0" {
            mutate {
                replace => {
                    "additional_event_analysis.value.string_value" => "INITIAL"
                }
            }
        } else if [event_analysis] == "1" {
            mutate {
                replace => {
                    "additional_event_analysis.value.string_value" => "ONGOING"
                }
            }
        } else if [event_analysis] == "2" {
            mutate {
                replace => {
                    "additional_event_analysis.value.string_value" => "COMPLETE"
                }
            }
        }
        mutate {
            replace => {
                "additional_event_analysis.key" => "event_analysis"
            }
        }
        mutate {
            merge => {
                "event.idm.entity.additional.fields" => "additional_event_analysis"
            }
        }
    }
    # Add the attribute timestamp as an additional field, if available.
    # "column22" => "attribute_timestamp" 
    if [attribute_timestamp] != "" {
        mutate {
            replace => {
                "additional_attribute_timestamp.key" => "attribute_timestamp"
                "additional_attribute_timestamp.value.string_value" => "%{attribute_timestamp}"
            }
        }
        mutate {
            merge => {
                "event.idm.entity.additional.fields" => "additional_attribute_timestamp"
            }
        }
    }
    # Finally, merge the 'event' data into the '@output' field.
    mutate {
        merge => {
            "@output" => "event"
        }
    }
}

執行剖析器擴充功能後，CSV 中的自訂欄位就會成功新增至 additional 物件。

metadata.product_entity_id = "9d66d38a-14e1-407f-a4d1-90b82aa1d59f"
metadata.collected_timestamp = "2024-10-31T15:16:08Z"
metadata.vendor_name = "MISP"
metadata.product_name = "MISP"
metadata.entity_type = "IP_ADDRESS"
metadata.description = "ip-dst"
metadata.interval.start_time = "2023-06-27T19:36:04Z"
metadata.interval.end_time = "9999-12-31T23:59:59Z"
metadata.threat[0].category_details[0] = "Network activity"
metadata.threat[0].description = "tlp:white,type:OSINT,source:DigitalSide.IT,source:urlhaus.abuse.ch - additional info: DigitalSide Malware report: MD5: 59ce0baba11893f90527fc951ac69912"
metadata.threat[0].severity_details = "Medium"
metadata.threat[0].threat_feed_name = "DIGITALSIDE.IT"
entity.ip[0] = "117.253.154.123"
additional.fields["view_in_misp"] = "https:///events/view/3908"
additional.fields["object_timestamp"] = "1687894564"
additional.fields["event_source_org"] = "DIGITALSIDE.IT"
additional.fields["event_distribution"] = "YOUR_ORGANIZATION_ONLY"
additional.fields["event_analysis"] = "INITIAL"
additional.fields["attribute_timestamp"] = "1698036218"

Grok 範例

下列範例說明如何建立以 Grok 為基礎的剖析器擴充功能。

程式碼片段 (和 Grok) - 擷取優先順序和嚴重性

屬性範例：

記錄來源格式：Syslog
資料對應方法：使用 Grok 的程式碼片段
記錄類型：POWERSHELL
剖析器擴充功能用途：擷取優先順序和嚴重性。

說明：

在本範例中，系統會建立以 Grok 為基礎的剖析器擴充功能，將 Syslog Facility 和 Severity 值擷取到 UDM 安全性結果的 Priority 和 Severity 欄位。

filter {
    # Use grok to parse syslog messages. The on_error clause handles messages that don't match the pattern.
    grok {
        match => {
            "message" => [
                # Extract message with syslog headers.
                "(<%{POSINT:_syslog_priority}>)%{SYSLOGTIMESTAMP:datetime} %{DATA:logginghost}: %{GREEDYDATA:log_data}"
            ]
        }
        on_error => "not_supported_format"
    }
    # If the grok parsing failed, tag the event as unsupported and drop it.
    if ![not_supported_format] {
        if [_syslog_priority] != "" {
            if [_syslog_priority] =~ /0|8|16|24|32|40|48|56|64|72|80|88|96|104|112|120|128|136|144|152|160|168|176|184/ {
                mutate { replace => { "_security_result.severity_details" => "EMERGENCY" } } 
            }
            if [_syslog_priority] =~ /1|9|17|25|33|41|49|57|65|73|81|89|97|105|113|121|129|137|145|153|161|169|177|185/ {
                mutate { replace => { "_security_result.severity_details" => "ALERT" } } 
            }
            if [_syslog_priority] =~ /2|10|18|26|34|42|50|58|66|74|82|90|98|106|114|122|130|138|146|154|162|170|178|186/ {
                mutate { replace => { "_security_result.severity_details" => "CRITICAL" } }
            }
            if [_syslog_priority] =~ /3|11|19|27|35|43|51|59|67|75|83|91|99|107|115|123|131|139|147|155|163|171|179|187/ {
                mutate { replace => { "_security_result.severity_details" => "ERROR" } }
            }
            if [_syslog_priority] =~ /4|12|20|28|36|44|52|60|68|76|84|92|100|108|116|124|132|140|148|156|164|172|180|188/ {
                mutate { replace => { "_security_result.severity_details" => "WARNING" } }
            }
            if [_syslog_priority] =~ /5|13|21|29|37|45|53|61|69|77|85|93|101|109|117|125|133|141|149|157|165|173|181|189/ {
                mutate { replace => { "_security_result.severity_details" => "NOTICE" } }
            }
            if [_syslog_priority] =~ /6|14|22|30|38|46|54|62|70|78|86|94|102|110|118|126|134|142|150|158|166|174|182|190/ {
                mutate { replace => { "_security_result.severity_details" => "INFORMATIONAL" } }
            }
            if [_syslog_priority] =~ /7|15|23|31|39|47|55|63|71|79|87|95|103|111|119|127|135|143|151|159|167|175|183|191/ {
                mutate { replace => { "_security_result.severity_details" => "DEBUG" } }
            }
            # Facilities (mapped to priority)
            if [_syslog_priority] =~ /0|1|2|3|4|5|6|7/ { 
                mutate { replace => { "_security_result.priority_details" => "KERNEL" } } 
            }
            if [_syslog_priority] =~ /8|9|10|11|12|13|14|15/ { 
                mutate { replace => { "_security_result.priority_details" => "USER" } } 
            }
            if [_syslog_priority] =~ /16|17|18|19|20|21|22|23/ { 
                mutate { replace => { "_security_result.priority_details" => "MAIL" } } 
            }
            if [_syslog_priority] =~ /24|25|26|27|28|29|30|31/ { 
                mutate { replace => { "_security_result.priority_details" => "SYSTEM" } } 
            }
            if [_syslog_priority] =~ /32|33|34|35|36|37|38|39/ { 
                mutate { replace => { "_security_result.priority_details" => "SECURITY" } } 
            }
            if [_syslog_priority] =~ /40|41|42|43|44|45|46|47/ { 
                mutate { replace => { "_security_result.priority_details" => "SYSLOG" } } 
            }
            if [_syslog_priority] =~ /48|49|50|51|52|53|54|55/ { 
                mutate { replace => { "_security_result.priority_details" => "LPD" } } 
            }
            if [_syslog_priority] =~ /56|57|58|59|60|61|62|63/ { 
                mutate { replace => { "_security_result.priority_details" => "NNTP" } } 
            }
            if [_syslog_priority] =~ /64|65|66|67|68|69|70|71/ { 
                mutate { replace => { "_security_result.priority_details" => "UUCP" } } 
            }
            if [_syslog_priority] =~ /72|73|74|75|76|77|78|79/ { 
                mutate { replace => { "_security_result.priority_details" => "TIME" } } 
            }
            if [_syslog_priority] =~ /80|81|82|83|84|85|86|87/ { 
                mutate { replace => { "_security_result.priority_details" => "SECURITY" } } 
            }
            if [_syslog_priority] =~ /88|89|90|91|92|93|94|95/ { 
                mutate { replace => { "_security_result.priority_details" => "FTPD" } } 
            }
            if [_syslog_priority] =~ /96|97|98|99|100|101|102|103/ { 
                mutate { replace => { "_security_result.priority_details" => "NTPD" } } 
            }
            if [_syslog_priority] =~ /104|105|106|107|108|109|110|111/ { 
                mutate { replace => { "_security_result.priority_details" => "LOGAUDIT" } } 
            }
            if [_syslog_priority] =~ /112|113|114|115|116|117|118|119/ { 
                mutate { replace => { "_security_result.priority_details" => "LOGALERT" } } 
            }
            if [_syslog_priority] =~ /120|121|122|123|124|125|126|127/ { 
                mutate { replace => { "_security_result.priority_details" => "CLOCK" } } 
            }
            if [_syslog_priority] =~ /128|129|130|131|132|133|134|135/ { 
                mutate { replace => { "_security_result.priority_details" => "LOCAL0" } } 
            }
            if [_syslog_priority] =~ /136|137|138|139|140|141|142|143/ { 
                mutate { replace => { "_security_result.priority_details" => "LOCAL1" } } 
            }
            if [_syslog_priority] =~ /144|145|146|147|148|149|150|151/ { 
                mutate { replace => { "_security_result.priority_details" => "LOCAL2" } } 
            }
            if [_syslog_priority] =~ /152|153|154|155|156|157|158|159/ { 
                mutate { replace => { "_security_result.priority_details" => "LOCAL3" } } 
            }
            if [_syslog_priority] =~ /160|161|162|163|164|165|166|167/ { 
                mutate { replace => { "_security_result.priority_details" => "LOCAL4" } } 
            }
            if [_syslog_priority] =~ /168|169|170|171|172|173|174|175/ { 
                mutate { replace => { "_security_result.priority_details" => "LOCAL5" } } 
            }
            if [_syslog_priority] =~ /176|177|178|179|180|181|182|183/ { 
                mutate { replace => { "_security_result.priority_details" => "LOCAL6" } } 
            }
            if [_syslog_priority] =~ /184|185|186|187|188|189|190|191/ { 
                mutate { replace => { "_security_result.priority_details" => "LOCAL7" } } 
            }
            mutate {
                merge => {
                    "event.idm.read_only_udm.security_result" => "_security_result"
                }
            }
        }
        mutate {
            merge => {
                "@output" => "event"
            }
        }
    }
}

查看剖析器擴充功能的結果時，會顯示人類可讀的格式。

metadata.product_log_id = "6161053"
metadata.event_timestamp = "2024-10-31T15:10:10Z"
metadata.event_type = "PROCESS_LAUNCH"
metadata.vendor_name = "Microsoft"
metadata.product_name = "PowerShell"
metadata.product_event_type = "600"
metadata.description = "Info"
metadata.log_type = "POWERSHELL"
principal.hostname = "win-adfs.lunarstiiiness.com"
principal.resource.name = "in_powershell"
principal.resource.resource_subtype = "im_msvistalog"
principal.asset.hostname = "win-adfs.lunarstiiiness.com"
target.hostname = "Default Host"
target.process.command_line = "C:\Program Files\Microsoft Azure AD Sync\Bin\miiserver.exe"
target.asset.hostname = "Default Host"
target.asset.asset_id = "Host ID:bf203e94-72cf-4649-84a5-fc02baedb75f"
security_result[0].severity_details = "INFORMATIONAL"
security_result[0].priority_details = "USER"

程式碼片段 (和 Grok) - 事件裝飾、暫時變數名稱和資料類型轉換

屬性範例：

記錄來源格式：JSON，並包含 Syslog 標頭
資料對應方法：使用 Grok 的程式碼片段
記錄類型：WINDOWS_SYSMON
剖析器擴充功能用途：裝飾事件、暫時性變數名稱和資料類型。

說明：

這個範例說明建立剖析器擴充功能時，如何執行下列動作：

根據條件陳述式裝飾，並瞭解程式碼片段中的資料類型。
轉換資料類型
方便閱讀的暫時變數名稱
重複欄位

根據條件陳述式裝飾

這個範例會新增每個事件類型在 WINDOWS_SYSMON 中的意義 (背景資訊) 說明。它會使用條件陳述式檢查 EventID，然後新增 Description，例如 EventID 1 是 Process Creation 事件。

使用擷取篩選器 (例如 JSON) 時，系統可能會保留原始資料類型。

在以下範例中，系統會預設將 EventID 值擷取為整數。條件陳述式會將 EventID 值評估為整數，而非字串。

if [EventID] == 1 {
  mutate {
    replace => {
      "_description" => "[1] Process creation"
    }
  }
}

資料類型轉換

您可以使用 convert 函式，在剖析器擴充功能中轉換資料型別。

mutate {
  convert => {
    "EventID" => "string"
  }
  on_error => "_convert_EventID_already_string"
}

方便閱讀的暫時變數名稱

您可以在程式碼片段中使用暫時變數名稱，稍後再重新命名，與最終輸出 UDM 事件物件名稱相符。這有助於提升整體可讀性。

在以下範例中，description 變數會重新命名為 event.idm.read_only_udm.metadata.description：

mutate {
  rename => {
    "_description" => "event.idm.read_only_udm.metadata.description"
  }
}

重複欄位

完整的剖析器擴充功能如下：

filter {
# initialize variable
mutate {
  replace => {
    "EventID" => ""
  }
}
# Use grok to parse syslog messages.
# The on_error clause handles messages that don't match the pattern.
grok {
  match => {
    "message" => [
      "(<%{POSINT:_syslog_priority}>)%{SYSLOGTIMESTAMP:datetime} %{DATA:logginghost}: %{GREEDYDATA:log_data}"
    ]
  }
  on_error => "not_supported_format"
}
if ![not_supported_format] {
  json {
    source => "log_data"
    on_error => "not_json"
  }
  if ![not_json] {
    if [EventID] == 1 {
      mutate {
        replace => {
          "_description" => "[1] Process creation"
        }
      }
    }
    if [EventID] == 2 {
      mutate {
        replace => {
          "_description" => "[2] A process changed a file creation time"
        }
      }
    }
    if [EventID] == 3 {
      mutate {
        replace => {
          "_description" => "[3] Network connection"
        }
      }
    }
    if [EventID] == 4 {
      mutate {
        replace => {
          "_description" => "[4] Sysmon service state changed"
        }
      }
    }
    if [EventID] == 5 {
      mutate {
        replace => {
          "_description" => "[5] Process terminated"
        }
      }
    }
    if [EventID] == 6 {
      mutate {
        replace => {
          "_description" => "[6] Driver loaded"
        }
      }
    }
    if [EventID] == 7 {
      mutate {
        replace => {
          "_description" => "[7] Image loaded"
        }
      }
    }
    if [EventID] == 8 {
      mutate {
        replace => {
          "_description" => "[8] CreateRemoteThread"
        }
      }
    }
    if [EventID] == 9 {
      mutate {
        replace => {
          "_description" => "[9] RawAccessRead"
        }
      }
    }
    if [EventID] == 10 {
      mutate {
        replace => {
          "_description" => "[10] ProcessAccess"
        }
      }
    }
    if [EventID] == 11 {
      mutate {
        replace => {
          "_description" => "[11] FileCreate"
        }
      }
    }
    if [EventID] == 12 {
      mutate {
        replace => {
          "_description" => "[12] RegistryEvent (Object create and delete)"
        }
      }
    }
    if [EventID] == 13 {
      mutate {
        replace => {
          "_description" => "[13] RegistryEvent (Value Set)"
        }
      }
    }
    if [EventID] == 14 {
      mutate {
        replace => {
          "_description" => "[14] RegistryEvent (Key and Value Rename)"
        }
      }
    }
    if [EventID] == 15 {
      mutate {
        replace => {
          "_description" => "[15] FileCreateStreamHash"
        }
      }
    }
    if [EventID] == 16 {
      mutate {
        replace => {
          "_description" => "[16] ServiceConfigurationChange"
        }
      }
    }
    if [EventID] == 17 {
      mutate {
        replace => {
          "_description" => "[17] PipeEvent (Pipe Created)"
        }
      }
    }
    if [EventID] == 18 {
      mutate {
        replace => {
          "_description" => "[18] PipeEvent (Pipe Connected)"
        }
      }
    }
    if [EventID] == 19 {
      mutate {
        replace => {
          "_description" => "[19] WmiEvent (WmiEventFilter activity detected)"
        }
      }
    }
    if [EventID] == 20 {
      mutate {
        replace => {
          "_description" => "[20] WmiEvent (WmiEventConsumer activity detected)"
        }
      }
    }
    if [EventID] == 21 {
      mutate {
        replace => {
          "_description" => "[21] WmiEvent (WmiEventConsumerToFilter activity detected)"
        }
      }
    }
    if [EventID] == 22 {
      mutate {
        replace => {
          "_description" => "[22] DNSEvent (DNS query)"
        }
      }
    }
    if [EventID] == 23 {
      mutate {
        replace => {
          "_description" => "[23] FileDelete (File Delete archived)"
        }
      }
    }
    if [EventID] == 24 {
      mutate {
        replace => {
          "_description" => "[24] ClipboardChange (New content in the clipboard)"
        }
      }
    }
    if [EventID] == 25 {
      mutate {
        replace => {
          "_description" => "[25] ProcessTampering (Process image change)"
        }
      }
    }
    if [EventID] == 26 {
      mutate {
        replace => {
          "_description" => "[26] FileDeleteDetected (File Delete logged)"
        }
      }
    }
    if [EventID] == 255 {
      mutate {
        replace => {
          "_description" => "[255] Error"
        }
      }
    }
    mutate {
      rename => {
        "_description" => "event.idm.read_only_udm.metadata.description"
      }
    }
    statedump{}
    mutate {
      merge => {
        "@output" => "event"
      }
    }
  }
}
}

執行剖析器擴充功能後，裝飾項目會成功新增至 metadata.description 欄位。

metadata.product_log_id = "6008459"
metadata.event_timestamp = "2024-10-31T14:41:53.442Z"
metadata.event_type = "REGISTRY_CREATION"
metadata.vendor_name = "Microsoft"
metadata.product_name = "Microsoft-Windows-Sysmon"
metadata.product_event_type = "12"
metadata.description = "[12] RegistryEvent (Object create and delete)"
metadata.log_type = "WINDOWS_SYSMON"
additional.fields["thread_id"] = "3972"
additional.fields["channel"] = "Microsoft-Windows-Sysmon/Operational"
additional.fields["Keywords"] = "-9223372036854776000"
additional.fields["Opcode"] = "Info"
additional.fields["ThreadID"] = "3972"
principal.hostname = "win-adfs.lunarstiiiness.com"
principal.user.userid = "tim.smith_admin"
principal.user.windows_sid = "S-1-5-18"
principal.process.pid = "6856"
principal.process.file.full_path = "C:\Windows\system32\wsmprovhost.exe"
principal.process.product_specific_process_id = "SYSMON:{927d35bf-a374-6495-f348-000000002900}"
principal.administrative_domain = "LUNARSTIIINESS"
principal.asset.hostname = "win-adfs.lunarstiiiness.com"
target.registry.registry_key = "HKU\S-1-5-21-3263964631-4121654051-1417071188-1116\Software\Policies\Microsoft\SystemCertificates\CA\Certificates"
observer.asset_id = "5770385F:C22A:43E0:BF4C:06F5698FFBD9"
observer.process.pid = "2556"
about[0].labels[0].key = "Category ID"
about[0].labels[0].value = "RegistryEvent"
security_result[0].rule_name = "technique_id=T1553.004,technique_name=Install Root Certificate"
security_result[0].summary = "Registry object added or deleted"
security_result[0].severity = "INFORMATIONAL"
security_result[1].rule_name = "EventID: 12"
security_result[2].summary = "12"

XML 範例

下列範例說明如何建立剖析器擴充功能，其中記錄來源為 XML 格式。

程式碼片段 - 將任意欄位擷取至 `additional` 物件

屬性範例：

記錄來源格式：XML
資料對應方法：程式碼片段
記錄類型：WINDOWS_DEFENDER_AV
剖析器擴充功能用途：將任意欄位擷取到 additional 物件

說明：

本範例的目標是擷取並儲存 Platform Version 值，例如，以便製作 outdated platform versions 的報表及搜尋 outdated platform versions。

審查重要 UDM 欄位文件後，我們未發現合適的標準 UDM 欄位。因此，這個範例會使用 additional 物件，將這項資訊儲存為自訂鍵/值組合。

# Parser Extension for WINDOWS_DEFENDER_AV
# 2024-10-29: cmmartin: Extracting 'Platform Version' into Additional
filter {
    # Uses XPath to target the specific element(s)
    xml {
        source => "message"
            xpath => {
                "/Event/EventData/Data[@Name='Platform version']" => "platform_version"
        }
        on_error => "_xml_error"
    }
    # Conditional processing: Only proceed if XML parsing was successful
    if ![_xml_error] {
        # Prepare the additional field structure using a temporary variable
        mutate{
            replace => {
                "additional_platform_version.key" => "Platform Version"
                "additional_platform_version.value.string_value" => "%{platform_version}"
            }
            on_error => "no_platform_version"
        }
        # Merge the additional field into the event1 structure.
        if ![no_platform_version] {
            mutate {
                merge => {
                    "event1.idm.read_only_udm.additional.fields" => "additional_platform_version"
                }
            }
        }
        mutate {
            merge => {
                "@output" => "event1"
            }
        }
    }
}

執行「PREVIEW UDM OUTPUT」會顯示新欄位已成功新增。

metadata.event_timestamp = "2024-10-29T14:08:52Z"
metadata.event_type = "STATUS_HEARTBEAT"
metadata.vendor_name = "Microsoft"
metadata.product_name = "Windows Defender AV"
metadata.product_event_type = "MALWAREPROTECTION_SERVICE_HEALTH_REPORT"
metadata.description = "Endpoint Protection client health report (time in UTC)."
metadata.log_type = "WINDOWS_DEFENDER_AV"
additional.fields["Platform Version"] = "4.18.24080.9"
principal.hostname = "win-dc-01.ad.1823127835827.altostrat.com"
security_result[0].description = "EventID: 1151"
security_result[0].action[0] = "ALLOW"
security_result[0].severity = "LOW"

程式碼片段 (和 Grok) - 將任意欄位擷取到主體主機名稱

屬性範例：

記錄來源格式：XML
資料對應方法：使用 Grok 的程式碼片段
記錄類型：WINDOWS_DEFENDER_AV
剖析器擴充功能用途：將任意欄位擷取至主體主機名稱

說明：

這個範例的目標是從 FQDN 擷取 Hostname，並覆寫 principal.hostname 欄位。

這個範例會檢查原始記錄的 Computer name 欄位是否包含 FQDN。如果是，系統只會擷取 Hostname 部分，並覆寫 UDM Principal Hostname 欄位。

查看剖析器和重要 UDM 欄位文件後，很明顯應該使用 principal.hostname 欄位。

# Parser Extension for WINDOWS_DEFENDER_AV
# 2024-10-29: Extract Hostname from FQDN and overwrite principal.hostname
filter {
    # Uses XPath to target the specific element(s)
    xml {
        source => "message"
            xpath => {
                "/Event/System/Computer" => "hostname"
        }
        on_error => "_xml_error"
    }
    # Conditional processing: Only proceed if XML parsing was successful
    if ![_xml_error] {
  # Extract all characters before the first dot in the hostname variable
        grok {
            match => { "hostname" => "(?<hostname>[^.]+)" }
        }
        mutate {
            replace => {
                "event1.idm.read_only_udm.principal.hostname" => "%{hostname}"
            }
        }
        mutate {
            merge => {
                "@output" => "event1"
            }
        }
    }
}

這個剖析器擴充功能會使用 Grok 陳述式執行規則運算式 (regex)，藉此擷取 hostname 欄位。規則運算式本身會使用具名擷取群組，也就是說，括號內比對到的任何內容都會儲存在名為 hostname 的欄位中，比對一或多個字元，直到遇到半形句號為止。這只會擷取 FQDN 內的 hostname。

不過，執行「預覽 UDM 輸出內容」時會傳回錯誤。為什麼會這樣？

generic::unknown: pipeline.ParseLogEntry failed:
 LOG_PARSING_CBN_ERROR: "generic::internal: pipeline failed: filter grok (2) failed: 
field\ "hostname\" already exists in data and is not overwritable"

Grok `overwrite` 陳述式

在 Grok 陳述式中，具名擷取群組無法覆寫現有變數，除非使用 overwrite 陳述式明確指定。在這個情境中，我們可以為 Grok 陳述式中的具名擷取群組使用不同的變數名稱，也可以如以下程式碼片段範例所示，使用 overwrite 陳述式明確覆寫現有的 hostname 變數。

# Parser Extension for WINDOWS_DEFENDER_AV
# 2024-10-29: cmmartin: Overwriting principal Hostname
filter {
  xml {
    source => "message"
      xpath => {
        "/Event/System/Computer" => "hostname"
    }
    on_error => "_xml_error"
  }
  if ![_xml_error] {
    grok {
      match => { "hostname" => "(?<hostname>[^.]+)" }
      overwrite => ["hostname"]
      on_error => "_grok_hostname_error"
    }
    mutate {
      replace => {
        "event1.idm.read_only_udm.principal.hostname" => "%{hostname}"
      }
    }
    mutate {
      merge => {
        "@output" => "event1"
      }
    }
  }
}

再次執行「PREVIEW UDM OUTPUT」，會顯示系統已在擷取 FQDN 中的 hostname 後，新增該欄位。

metadata.event_timestamp"2024-10-29T14:08:52Z"
metadata.event_type"STATUS_HEARTBEAT"
metadata.vendor_name"Microsoft"
metadata.product_name"Windows Defender AV"
metadata.product_event_type"MALWAREPROTECTION_SERVICE_HEALTH_REPORT"
metadata.description"Endpoint Protection client health report (time in UTC)."
metadata.log_type"WINDOWS_DEFENDER_AV"
principal.hostname"win-dc-01"
security_result[0].description"EventID: 1151"
security_result[0].action[0]"ALLOW"
security_result[0].severity"LOW"

JSON、CSV、XML、Syslog 和 KV 範例

下列範例說明如何建立剖析器擴充功能，其中記錄來源採用 JSON、CSV、XML、Syslog 或 KV 格式。

程式碼片段 - 移除現有對應

屬性範例：

記錄來源格式：JSON、CSV、Syslog、XML 和 KV
資料對應方法：程式碼片段
剖析器擴充功能用途：移除 UDM 欄位的值

說明：

這些範例的目標是移除 UDM 欄位的值，藉此移除現有對應。

以下範例會移除 string 欄位的值：

filter {
   mutate{
     replace => {
         "event.idm.read_only_udm.metadata.vendor_name" => ""
     }
   }
   mutate {
     merge => {
       "@output" => "event"
     }
  }
}

以下範例會移除 integer 欄位的值：

filter {
   mutate {
     replace => {
       "principal_port" => "0"
     }
 }
   mutate {
     convert => {
       "principal_port" => "integer"
     }
 }
   mutate {
     rename => {
       "principal_port" => "event.idm.read_only_udm.principal.port"
     }
 }
   mutate {
     merge => {
       "@output" => "event"
     }
  }
}

以下範例會移除 float 欄位的值：

filter {
   mutate {
       replace => {
         "security_result_object.risk_score" => "0.0" 
       }
       convert => {
         "security_result_object.risk_score" => "float"
       }
       on_error => "default_risk_score_conversion_failed"
     }
   mutate {
       merge => {
           "event.idm.read_only_udm.security_result" => "security_result_object"
       }
       on_error => "security_result_merge_failed"
     }
   mutate {
     merge => {
       "@output" => "event"
     }
 }
}

以下範例會移除 boolean 欄位的值：

filter {
   mutate{
       replace => {
           "tls_established" => "false"
       }
  }
   mutate {
     convert => {
       "tls_established" => "boolean"
     }
   }
   mutate {
     rename => {
       "tls_established" => "event.idm.read_only_udm.network.tls.established"
     }
   }
   mutate {
     merge => {
       "@output" => "event"
     }
  }
}

以下範例會移除 extension 欄位的值：

filter {
   mutate {
       replace => {
          "event.idm.read_only_udm.extensions.auth.auth_details" => ""
       }
       on_error => "logon_type_not_set"
   }
   mutate {
     merge => {
       "@output" => "event"
     }
  }
}

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。

剖析器擴充功能範例

剖析器擴充功能範例

無程式碼範例

程式碼片段範例

JSON 範例

無程式碼 - 擷取欄位

免程式碼 - 使用前置條件值擷取欄位

程式碼片段 - 新增 HTTP 使用者代理程式

CSV 範例

程式碼片段 - 將任意欄位擷取到 additional 物件中

Grok 範例

程式碼片段 (和 Grok) - 擷取優先順序和嚴重性

程式碼片段 (和 Grok) - 事件裝飾、暫時變數名稱和資料類型轉換

根據條件陳述式裝飾

資料類型轉換

方便閱讀的暫時變數名稱

重複欄位

XML 範例

程式碼片段 - 將任意欄位擷取至 additional 物件

程式碼片段 (和 Grok) - 將任意欄位擷取到主體主機名稱

Grok overwrite 陳述式

JSON、CSV、XML、Syslog 和 KV 範例

程式碼片段 - 移除現有對應

程式碼片段 - 將任意欄位擷取到 `additional` 物件中

程式碼片段 - 將任意欄位擷取至 `additional` 物件

Grok `overwrite` 陳述式