此页面由 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 对象
Syslog （日志类型：POWERSHELL）	从 Syslog 中提取优先级和严重性	将 Syslog Facility 和 Severity 值提取到 UDM 安全结果优先级和严重程度字段。	基于 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` 对象中	例如，提取并存储“Platform Version”值，以便报告和搜索过时的平台版本。在此示例中，没有合适的标准 UDM 字段，因此 `additional` 对象用于将信息存储为自定义键值对。	`additional` 对象用于将信息存储为自定义键值对。
XML （日志类型：WINDOWS_DEFENDER_AV）	将任意字段提取到主机名	从 FQDN 中提取主机名。条件处理用于确定是否应覆盖 `principal.hostname` 字段。 Grok 语句使用正则表达式 (regex) 提取 `hostname` 字段。正则表达式本身使用命名的捕获组，这意味着，括号内匹配的所有内容都将存储在名为 `hostname` 的字段中，并匹配一个或多个字符，直到遇到一个圆点为止。这只会捕获 FQDN 中的 `hostname`。 Grok `overwrite` 语句但是，运行预览 UDM 输出时，系统会返回以下错误：“LOG_PARSING_CBN_ERROR: Field `hostname` already exists in data and is not overwritable”。在 Grok 语句中，除非使用 `overwrite` 语句明确指定，否则命名捕获组无法覆盖现有变量。在这种情况下，我们可以为 Grok 语句中的命名捕获组使用不同的变量名称，也可以（如本示例所示）使用覆盖语句显式覆盖现有的主机名变量。	基于 Grok 条件处理用于确定是否应覆盖字段。使用正则表达式 (regex) 的 Grok 语句。 Grok `overwrite` 语句

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"
}

该示例使用无代码方法，通过以下数据字段映射创建解析器扩展程序：

前提条件路径	前提条件运算符	Precondition Value	原始数据路径	目标字段*
`jsonPayload.total_bytes`	NOT_EQUAL	""	`jsonPayload.total_bytes`	`udm.principal.network.received_bytes`
`jsonPayload.elapsed_time`	NOT_EQUAL	""	`jsonPayload.elapsed_time`	`udm.principal.network.session_duration.seconds`
`jsonPayload.total_packets`	NOT_EQUAL	""	`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 对象中重复字段的所有值），或
append（将提取的值附加到重复字段）。

如需了解详情，请参阅重复字段部分。

此示例会替换标准化 principal.user.email_address 字段中的所有现有电子邮件地址。

借助前提条件，您可以在执行提取操作之前执行条件检查。在大多数情况下，前提条件字段与要提取的原始数据字段相同，前提条件运算符为 not Null，例如 foo != ""。

不过，有时（如我们的示例所示），您要提取的原始数据字段值并非存在于所有日志条目中。在这种情况下，您可以使用其他前提条件字段来过滤提取操作。在我们的示例中，您要提取的原始 triggeringUserEmail 字段仅存在于 type = Data Loss Prevention 为 true 的日志中。

以下是应输入到无代码解析器扩展字段中的示例值：

前提条件路径	前提条件运算符	Precondition Value	原始数据路径	目标字段*
`type`	EQUALS	数据泄露防护	`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）- 事件装饰、临时变量名称和数据类型转换

示例属性：

日志来源格式：带有 Syslog 标头的 JSON
数据映射方法：使用 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 生成报告和进行搜索。

查看重要 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"
            }
        }
    }
}

运行预览 UDM 输出结果显示，新字段已成功添加。

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"
      }
    }
  }
}

再次运行预览 UDM 输出后，您会看到系统从 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 示例

无代码 - 提取字段

无代码 - 提取具有前提条件值的字段

代码段 - 添加 HTTP 用户代理

CSV 示例

代码段 - 将任意字段提取到 additional 对象中

Grok 示例

代码段（和 Grok）- 提取优先级和严重程度

代码段（和 Grok）- 事件装饰、临时变量名称和数据类型转换

基于条件语句的装饰

数据类型转换

临时变量名称（以提高可读性）

重复字段

XML 示例

代码段 - 将任意字段提取到 additional 对象中

代码段（和 Grok）- 将任意字段提取到主机名

Grok overwrite 语句

代码段 - 将任意字段提取到 `additional` 对象中

代码段 - 将任意字段提取到 `additional` 对象中

Grok `overwrite` 语句