Fix SecOps parser review findings

Review findings verified against the parsedmarc serializers, the UDM
field list, Google's parser troubleshooting doc, and the content-hub
parsers:

- Detect failure reports by feedback_type OR arrival_date_utc: the
  text-format failure path synthesizes a feedback report with no
  Feedback-Type field, so those rows had no feedback_type key and were
  dropped as TAG_UNSUPPORTED.
- Detect SMTP TLS rows by policy_type OR result_type, and fall back to
  receiving_mx_hostname for target.hostname, so failure-detail rows from
  parsedmarc versions without the paired serializer fix still map.
- Merge (not replace) network.email.to and network.email.subject: both
  are repeated fields per the UDM field list, so a scalar replace is a
  type mismatch. The unguarded subject replace could have failed parsing
  for every failure event with a subject.
- Guard testing with != "": parsedmarc emits the RFC 9990 t= flag as the
  string "y"/"n", not a boolean, so the == "true"/"false" test could
  never fire and dmarc_testing was never emitted. Drop testing from the
  boolean convert list accordingly.
- Add on_error to the four numeric %{} interpolations (count,
  source_asn, *_session_count) so a tenant where interpolating a
  non-string field errors degrades to a missing additional.fields entry
  instead of _failed_parsing_ (per the official parser tips doc).
- Drop the concatenated security_result.description strings: the tips
  doc says not to pack multiple values into one UDM field, and every
  value already lands individually in additional.fields.
- Use TAG_MALFORMED_MESSAGE for non-JSON input, matching the tips doc's
  canonical not-JSON example (TAG_MALFORMED_ENCODING is for character
  encoding problems).
- Map org_extra_contact_info/errors (aggregate) and the source
  enrichment fields (failure) into additional.fields; the failure branch
  previously dropped enrichment the aggregate branch kept.
- Replace the SMTP TLS success sample, which was not generated from this
  repository's samples, with rfc8460.json output; regenerate all sample
  events with the fixed serializer and name each event's source sample
  file. All six samples are now byte-identical to real syslog output.
- Reorder the validation plan to start with the failure sample: it is
  the only shape that emits JSON nulls, whose json{} interaction with
  the pre-initialized "" fields is undocumented.
- Update Google doc links to docs.cloud.google.com (the old URLs 301)
  and link the parser from the syslog section of the usage docs.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/source/usage.md

@@ -503,6 +503,13 @@ The full set of configuration options are:
   retry_delay = 5
   ```
 
+  :::{note}
+  To ingest the syslog output into Google SecOps (Chronicle), a custom UDM
+  parser is available in the
+  [google_secops_parser](https://github.com/domainaware/parsedmarc/tree/master/google_secops_parser)
+  directory.
+  :::
+
 - `gmail_api`
   - `credentials_file` - str: Path to file containing the
       credentials, None to disable (Default: `None`)

google_secops_parser/README.md

@@ -5,13 +5,18 @@ custom parser (configuration-based normalizer / CBN) that maps the JSON events
 parsedmarc emits through its built-in `[syslog]` output to the Unified Data
 Model (UDM).
 
-This is a **SecOps-side parser only** — it requires no changes to parsedmarc.
-parsedmarc already ships structured JSON over syslog; the DMARC→UDM mapping
-lives here so that a downstream UDM schema change is a parser edit rather than a
-parsedmarc release.
+This is a **SecOps-side parser** — parsedmarc already ships structured JSON
+over syslog, and the DMARC→UDM mapping lives here so that a downstream UDM
+schema change is a parser edit rather than a parsedmarc release. One paired
+library fix ships alongside it: SMTP TLS failure-detail rows carry
+`policy_domain` / `policy_type` as of the parsedmarc release that includes this
+parser. Older parsedmarc versions omit those keys on failure rows; the parser
+still detects such rows via `result_type`, but a row that carries neither an
+MTA IP nor an MX hostname (e.g. `sts-policy-fetch-error`) then has no UDM noun
+and will not produce a valid event.
 
 > **New to SecOps parsers?** SecOps ingests a log source by running a *parser*
-> that turns each raw log line into a [Unified Data Model](https://cloud.google.com/chronicle/docs/event-processing/udm-overview)
+> that turns each raw log line into a [Unified Data Model](https://docs.cloud.google.com/chronicle/docs/event-processing/udm-overview)
 > (UDM) event. These parsers are written in a Logstash-style configuration
 > language Google calls a **configuration-based normalizer (CBN)** — the
 > `parsedmarc.conf` in this directory is one. You attach it to a custom *log
@@ -36,8 +41,14 @@ detects them by a field unique to each and maps them as follows:
 | parsedmarc report | Detected by | UDM `metadata.event_type` |
 |---|---|---|
 | DMARC aggregate | `xml_schema` | `EMAIL_TRANSACTION` |
-| DMARC failure | `feedback_type` | `EMAIL_TRANSACTION` |
-| SMTP TLS (RFC 8460) | `policy_type` | `GENERIC_EVENT` |
+| DMARC failure | `feedback_type` or `arrival_date_utc` | `EMAIL_TRANSACTION` |
+| SMTP TLS (RFC 8460) | `policy_type` or `result_type` | `GENERIC_EVENT` |
+
+The `or` fallbacks matter: text-format failure reports have no `Feedback-Type`
+field (parsedmarc emits no `feedback_type` key for them, but always computes
+`arrival_date_utc`), and SMTP TLS failure-detail rows from parsedmarc versions
+older than this parser lack `policy_type` (every RFC 8460 failure detail has a
+`result_type`).
 
 `EMAIL_TRANSACTION` and `GENERIC_EVENT` are both valid `metadata.event_type`
 values. Note that **`GENERIC_EVENT` events only appear in raw-log and UDM
@@ -50,14 +61,18 @@ DMARC types.
 1. **Unvalidated** — see [Status](#status).
 2. **JSON types** — Chronicle's `json{}` filter **preserves the original JSON
    type**, so parsedmarc's booleans and numbers are handled differently:
-   - **Booleans** (`dmarc_aligned` / `spf_aligned` / `dkim_aligned` / `testing`
-     / `normalized_timespan`) are converted to strings so `[dmarc_aligned] ==
+   - **Booleans** (`dmarc_aligned` / `spf_aligned` / `dkim_aligned` /
+     `normalized_timespan`) are converted to strings so `[dmarc_aligned] ==
      "false"` works, and stored as `string_value` (Google's content-hub parsers
-     never use `bool_value`).
+     never use `bool_value`). Note that `testing` is **not** a boolean —
+     parsedmarc emits the RFC 9990 `t=` flag as the string `"y"`/`"n"` — so it
+     is passed through as a string and guarded with `!= ""`.
    - **Numbers** (`count` / `*_session_count` / `source_asn`) are stored as
      `number_value` — built as a string, `convert`-ed to `uinteger`, then
      renamed — so SecOps can range-query and sort them (parsedmarc's "store
-     numbers as numbers" rule).
+     numbers as numbers" rule). Each numeric interpolation carries `on_error`
+     so a tenant where `%{}` rejects non-string fields degrades to a missing
+     `additional.fields` entry instead of `_failed_parsing_`.
 
    Every `if`-tested field is initialized to `""` *before* `json` and guarded
    with `!= ""`: CBN raises `_failed_parsing_` on a conditional referencing an
@@ -65,12 +80,18 @@ DMARC types.
    DMARC-fail record (`dmarc_aligned=false`) should yield
    `security_result.category = AUTH_VIOLATION` — worth confirming in the
    validation tool.
-3. **Aggregate count** — a DMARC aggregate record summarizes `count` messages
+3. **JSON nulls** — failure-report rows are the only shape that emits JSON
+   `null` values (see the failure sample below: `dkim_domain`, `source_*`).
+   Google's docs cover *absent* fields (the `""` initialization above) but are
+   silent on whether `json{}` overwrites an initialized field with a null.
+   This is why the failure sample should be the **first** event pasted into
+   the validation tool.
+4. **Aggregate count** — a DMARC aggregate record summarizes `count` messages
    from one source IP, not a single message. Each record becomes one
    `EMAIL_TRANSACTION` with `count` carried in `additional.fields`. There is no
    first-class per-message expansion (fanning out `count` copies would
    misrepresent the data).
-4. **Address format** — aggregate reports only carry the From *domain*, so
+5. **Address format** — aggregate reports only carry the From *domain*, so
    `network.email.from` holds a bare domain for aggregate events but a full
    address for failure events. UDM email-address fields are expected to be
    `local-mailbox@domain`; downstream consumers should account for the
@@ -79,8 +100,8 @@ DMARC types.
 ## UDM field mappings
 
 All UDM field names below are from the
-[UDM field list](https://cloud.google.com/chronicle/docs/reference/udm-field-list)
-and [SecurityResult reference](https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/SecurityResult).
+[UDM field list](https://docs.cloud.google.com/chronicle/docs/reference/udm-field-list)
+and [SecurityResult reference](https://docs.cloud.google.com/chronicle/docs/reference/rest/v1alpha/SecurityResult).
 
 ### DMARC aggregate → `EMAIL_TRANSACTION`
 
@@ -95,7 +116,7 @@ and [SecurityResult reference](https://cloud.google.com/chronicle/docs/reference
 | `header_from` | `network.email.from` (domain; see caveat 4) |
 | `disposition` | `security_result.action` (`none`→`ALLOW`, `quarantine`→`QUARANTINE`, `reject`→`BLOCK`) |
 | `dmarc_aligned=false` | `security_result.category = AUTH_VIOLATION` |
-| `org_name`, `org_email`, `count`, `p`, `sp`, `np`, `pct`, `fo`, `adkim`, `aspf`, `testing`, `discovery_method`, `normalized_timespan`, `*_aligned`, `dkim_*`, `spf_*`, `policy_override_*`, `source_base_domain`, `source_name`, `source_type`, `source_asn`, `source_as_name`, `source_as_domain`, `envelope_from`, `envelope_to` | `additional.fields` |
+| `org_name`, `org_email`, `org_extra_contact_info`, `errors`, `count`, `p`, `sp`, `np`, `pct`, `fo`, `adkim`, `aspf`, `testing`, `discovery_method`, `normalized_timespan`, `*_aligned`, `dkim_*`, `spf_*`, `policy_override_*`, `source_base_domain`, `source_name`, `source_type`, `source_asn`, `source_as_name`, `source_as_domain`, `envelope_from`, `envelope_to` | `additional.fields` |
 
 ### DMARC failure → `EMAIL_TRANSACTION`
 
@@ -108,11 +129,10 @@ and [SecurityResult reference](https://cloud.google.com/chronicle/docs/reference
 | `source_country` | `principal.location.country_or_region` |
 | `reported_domain` | `target.hostname` |
 | `original_mail_from` | `network.email.from` |
-| `original_rcpt_to` | `network.email.to` |
-| `subject` | `network.email.subject` |
-| `auth_failure` | `security_result.category = AUTH_VIOLATION` + description |
-| `delivery_result` | `security_result.action` (`reject`→`BLOCK`, `quarantine`→`QUARANTINE`, `delivered`→`ALLOW`) |
-| `feedback_type`, `authentication_results`, `authentication_mechanisms`, `user_agent`, `dkim_domain`, `arrival_date` | `additional.fields` |
+| `original_rcpt_to` | `network.email.to` (repeated field → merged) |
+| `subject` | `network.email.subject` (repeated field → merged) |
+| `delivery_result` | `security_result.action` (`reject`→`BLOCK`, `quarantine`→`QUARANTINE`, `delivered`→`ALLOW`); `security_result.category` is always `AUTH_VIOLATION` |
+| `feedback_type`, `auth_failure`, `delivery_result`, `authentication_results`, `authentication_mechanisms`, `user_agent`, `dkim_domain`, `arrival_date`, `source_base_domain`, `source_name`, `source_type`, `source_asn`, `source_as_name`, `source_as_domain` | `additional.fields` |
 
 ### SMTP TLS → `GENERIC_EVENT`
 
@@ -120,16 +140,19 @@ and [SecurityResult reference](https://cloud.google.com/chronicle/docs/reference
 |---|---|
 | `begin_date` | `metadata.event_timestamp` (ISO 8601, via `date{}`) |
 | `report_id` | `metadata.product_log_id` |
-| `policy_domain` | `target.hostname` (always present → the noun) |
+| `policy_domain` | `target.hostname` (the noun; falls back to `receiving_mx_hostname` when absent) |
 | `receiving_ip` | `target.ip` (failure rows only) |
 | `sending_mta_ip` | `principal.ip` (failure rows only) |
-| `result_type` | `security_result` (`action=FAIL`, `category=POLICY_VIOLATION`) |
+| `result_type` | `security_result` (`action=FAIL`, `category=POLICY_VIOLATION`) — failure rows only |
 | `organization_name`, `policy_type`, `policy_strings`, `mx_host_patterns`, `successful_session_count`, `failed_session_count`, `failure_reason_code`, `receiving_mx_hostname`, `receiving_mx_helo`, `additional_info_uri` | `additional.fields` |
 
 > parsedmarc emits SMTP TLS reports as separate rows: one **success** row per
 > policy (counts, no MTA IPs) and one **failure** row per failure detail (which
-> may also lack MTA IPs, e.g. `sts-policy-fetch-error`). The noun therefore comes
-> from `policy_domain`, which is present on every row.
+> may also lack MTA IPs, e.g. `sts-policy-fetch-error`). As of the parsedmarc
+> release that ships this parser, every row — success and failure — carries
+> `policy_domain` and `policy_type`, so the noun is always available. Rows from
+> older parsedmarc versions omit both on failure details; those are detected
+> via `result_type` and fall back to `receiving_mx_hostname` for the noun.
 
 ## Installation
 
@@ -146,7 +169,7 @@ parsedmarc writes each report record as a single-line JSON message.
 ### 2. Collect the syslog stream into SecOps
 
 Syslog is ingested by a **collector**, not a Feed. Run the
-[Bindplane agent](https://cloud.google.com/chronicle/docs/install/install-forwarder)
+[Bindplane agent](https://docs.cloud.google.com/chronicle/docs/install/install-forwarder)
 (Google's recommended on-premises collector; the legacy Chronicle forwarder is
 end-of-life) with a **Syslog** collector pointed at the port above, and assign it
 a custom log type (for example `PARSEDMARC`).
@@ -155,59 +178,78 @@ a custom log type (for example `PARSEDMARC`).
 
 Associate `parsedmarc.conf` with the custom log type via the SecOps parser
 management UI or API (see
-[Manage parsers](https://cloud.google.com/chronicle/docs/event-processing/manage-parser-updates)).
+[Manage parsers](https://docs.cloud.google.com/chronicle/docs/event-processing/manage-parser-updates)).
 Validate against the sample events below before activating.
 
 ## Sample events for validation
 
-These are **real** single-line outputs from parsedmarc's `[syslog]` serializers
-(generated from the project's sample reports). Use them in the parser-validation
-tool. A live syslog line will also carry a `<PRI>` prefix; the parser strips any
-leading framing before the first `{`.
+These are **real** single-line outputs from parsedmarc's `[syslog]` serializers,
+generated from this repository's sample reports (the source file is named on
+each event). Use them in the parser-validation tool. A live syslog line will
+also carry a `<PRI>` prefix; the parser strips any leading framing before the
+first `{`.
 
-### DMARC Aggregate — fail (`dmarc_aligned=false`)
+Suggested order:
 
-```json
-{"xml_schema": "draft", "org_name": "accurateplastics.com", "org_email": "administrator@accurateplastics.com", "org_extra_contact_info": "", "report_id": "example.com:1538463741", "begin_date": "2018-10-01 17:07:12", "end_date": "2018-10-01 17:07:12", "normalized_timespan": false, "errors": "", "domain": "example.com", "adkim": "r", "aspf": "r", "p": "none", "sp": "reject", "np": "", "pct": "100", "fo": "", "testing": "", "discovery_method": "", "source_ip_address": "12.20.127.122", "source_country": "US", "source_reverse_dns": "", "source_base_domain": "", "source_name": "AT&T", "source_type": "ISP", "source_asn": 7018, "source_as_name": "AT&T Enterprises, LLC", "source_as_domain": "att.com", "count": 1, "spf_aligned": false, "dkim_aligned": false, "dmarc_aligned": false, "disposition": "none", "policy_override_reasons": "", "policy_override_comments": "", "envelope_from": "", "header_from": "example.com", "envelope_to": "", "dkim_domains": "", "dkim_selectors": "", "dkim_results": "", "spf_domains": "", "spf_scopes": "", "spf_results": ""}
-```
+1. **DMARC failure first** — it is the only shape containing JSON `null`
+   values (caveat 3), the least-documented `json{}` behaviour this parser
+   depends on.
+2. The two aggregates — confirm the `dmarc_aligned=false` one yields
+   `security_result.category = AUTH_VIOLATION`, that `count` and `source_asn`
+   land as `number_value`, and that the boolean→string handling holds.
+3. The SMTP TLS rows — confirm the success row produces a `GENERIC_EVENT` with
+   `target.hostname` and no `security_result`, and the failure rows produce
+   `security_result.action = FAIL`.
 
-### DMARC Aggregate — pass (`dmarc_aligned=true`)
-
-```json
-{"xml_schema": "1.0", "org_name": "example.org", "org_email": "noreply-dmarc-support@example.org", "org_extra_contact_info": "https://support.example.org/dmarc", "report_id": "20240125141224705995", "begin_date": "2024-01-25 05:12:24", "end_date": "2024-01-25 12:28:53", "normalized_timespan": false, "errors": "", "domain": "example.com", "adkim": "r", "aspf": "r", "p": "quarantine", "sp": "quarantine", "np": "", "pct": "100", "fo": "1", "testing": "", "discovery_method": "", "source_ip_address": "198.51.100.123", "source_country": "", "source_reverse_dns": "", "source_base_domain": "", "source_name": "", "source_type": "", "source_asn": "", "source_as_name": "", "source_as_domain": "", "count": 2, "spf_aligned": false, "dkim_aligned": true, "dmarc_aligned": true, "disposition": "none", "policy_override_reasons": "none", "policy_override_comments": "none", "envelope_from": "example.edu", "header_from": "example.com", "envelope_to": "example.net", "dkim_domains": "example.com", "dkim_selectors": "example", "dkim_results": "pass", "spf_domains": "example.edu", "spf_scopes": "mfrom", "spf_results": "pass"}
-```
-
-### DMARC Failure report
+### DMARC failure report — from `samples/failure/DMARC Failure Report for domain.de (…).eml` (note the JSON nulls)
 
 ```json
 {"feedback_type": "auth-failure", "user_agent": "Lua/1.0", "version": "1.0", "original_mail_from": "sharepoint@domain.de", "original_rcpt_to": "peter.pan@domain.de", "arrival_date": "Mon, 01 Oct 2018 11:20:27 +0200", "message_id": "<38.E7.30937.BD6E1BB5@ mailrelay.de>", "authentication_results": "dmarc=fail (p=none, dis=none) header.from=domain.de", "delivery_result": "policy", "auth_failure": "dmarc", "reported_domain": "domain.de", "arrival_date_utc": "2018-10-01 09:20:27", "authentication_mechanisms": "", "original_envelope_id": null, "dkim_domain": null, "sample_headers_only": false, "source_ip_address": "10.10.10.10", "source_reverse_dns": null, "source_base_domain": null, "source_name": null, "source_type": null, "source_asn": null, "source_as_name": null, "source_as_domain": null, "source_country": null, "subject": "Subject"}
 ```
 
-### SMTP TLS — success row (counts only)
+### DMARC Aggregate — fail (`dmarc_aligned=false`) — from `samples/aggregate/!example.com!1538204542!1538463818.xml`
 
 ```json
-{"organization_name": "Synametrics Technologies, Inc.", "begin_date": "2025-12-07T19:00:00Z", "end_date": "2025-12-08T18:59:59Z", "report_id": "1765256572301+dmarc-reports.dengage.com", "policy_strings": "version: STSv1|mode: enforce|mx: mta1.inboxsys.net|mx: mta2.inboxsys.net|max_age: 86400", "policy_domain": "dmarc-reports.dengage.com", "policy_type": "sts", "successful_session_count": 2, "failed_session_count": 0}
+{"xml_schema": "draft", "org_name": "accurateplastics.com", "org_email": "administrator@accurateplastics.com", "org_extra_contact_info": "", "report_id": "example.com:1538463741", "begin_date": "2018-10-01 17:07:12", "end_date": "2018-10-01 17:07:12", "normalized_timespan": false, "errors": "", "domain": "example.com", "adkim": "r", "aspf": "r", "p": "none", "sp": "reject", "np": "", "pct": "100", "fo": "", "testing": "", "discovery_method": "", "source_ip_address": "12.20.127.122", "source_country": "US", "source_reverse_dns": "", "source_base_domain": "", "source_name": "AT&T", "source_type": "ISP", "source_asn": 7018, "source_as_name": "AT&T Enterprises, LLC", "source_as_domain": "att.com", "count": 1, "spf_aligned": false, "dkim_aligned": false, "dmarc_aligned": false, "disposition": "none", "policy_override_reasons": "", "policy_override_comments": "", "envelope_from": "", "header_from": "example.com", "envelope_to": "", "dkim_domains": "", "dkim_selectors": "", "dkim_results": "", "spf_domains": "", "spf_scopes": "", "spf_results": ""}
 ```
 
-### SMTP TLS — failure-detail row
+### DMARC Aggregate — pass (`dmarc_aligned=true`) — from `samples/aggregate/empty_reason.xml`
 
 ```json
-{"organization_name": "Mail.ru", "begin_date": "2024-02-22T00:00:00Z", "end_date": "2024-02-23T00:00:00Z", "report_id": "b28254de-7b2e-be36-bb5c-4c3b92da8b25@mail.ru", "result_type": "sts-policy-fetch-error", "failed_session_count": 1, "failure_reason_code": "bad https response code: 404"}
+{"xml_schema": "1.0", "org_name": "example.org", "org_email": "noreply-dmarc-support@example.org", "org_extra_contact_info": "https://support.example.org/dmarc", "report_id": "20240125141224705995", "begin_date": "2024-01-25 05:12:24", "end_date": "2024-01-25 12:28:53", "normalized_timespan": false, "errors": "", "domain": "example.com", "adkim": "r", "aspf": "r", "p": "quarantine", "sp": "quarantine", "np": "", "pct": "100", "fo": "1", "testing": "", "discovery_method": "", "source_ip_address": "198.51.100.123", "source_country": "", "source_reverse_dns": "", "source_base_domain": "", "source_name": "", "source_type": "", "source_asn": "", "source_as_name": "", "source_as_domain": "", "count": 2, "spf_aligned": false, "dkim_aligned": true, "dmarc_aligned": true, "disposition": "none", "policy_override_reasons": "none", "policy_override_comments": "none", "envelope_from": "example.edu", "header_from": "example.com", "envelope_to": "example.net", "dkim_domains": "example.com", "dkim_selectors": "example", "dkim_results": "pass", "spf_domains": "example.edu", "spf_scopes": "mfrom", "spf_results": "pass"}
+```
+
+### SMTP TLS — success row (counts only) — from `samples/smtp_tls/rfc8460.json`
+
+```json
+{"organization_name": "Company-X", "begin_date": "2016-04-01T00:00:00Z", "end_date": "2016-04-01T23:59:59Z", "report_id": "5065427c-23d3-47ca-b6e0-946ea0e8c4be", "policy_domain": "company-y.example", "policy_type": "sts", "policy_strings": "version: STSv1|mode: testing|mx: *.mail.company-y.example|max_age: 86400", "successful_session_count": 5326, "failed_session_count": 303}
+```
+
+### SMTP TLS — failure-detail row with MTA IPs — from `samples/smtp_tls/rfc8460.json`
+
+```json
+{"organization_name": "Company-X", "begin_date": "2016-04-01T00:00:00Z", "end_date": "2016-04-01T23:59:59Z", "report_id": "5065427c-23d3-47ca-b6e0-946ea0e8c4be", "policy_domain": "company-y.example", "policy_type": "sts", "policy_strings": "version: STSv1|mode: testing|mx: *.mail.company-y.example|max_age: 86400", "result_type": "starttls-not-supported", "failed_session_count": 200, "sending_mta_ip": "2001:db8:abcd:0013::1", "receiving_ip": "203.0.113.56", "receiving_mx_hostname": "mx2.mail.company-y.example"}
+```
+
+### SMTP TLS — failure-detail row without MTA IPs — from `samples/smtp_tls/mail.ru.json`
+
+```json
+{"organization_name": "Mail.ru", "begin_date": "2024-02-22T00:00:00Z", "end_date": "2024-02-23T00:00:00Z", "report_id": "b28254de-7b2e-be36-bb5c-4c3b92da8b25@mail.ru", "policy_domain": "example.com", "policy_type": "sts", "result_type": "sts-policy-fetch-error", "failed_session_count": 1, "failure_reason_code": "bad https response code: 404"}
 ```
 
 ## Official references
 
-- [Overview of the UDM](https://cloud.google.com/chronicle/docs/event-processing/udm-overview)
-- [UDM field list](https://cloud.google.com/chronicle/docs/reference/udm-field-list)
-- [SecurityResult reference](https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/SecurityResult)
-- [Overview of log parsing](https://cloud.google.com/chronicle/docs/event-processing/parsing-overview)
-- [Parser syntax reference](https://cloud.google.com/chronicle/docs/reference/parser-syntax)
-- [Tips and troubleshooting when writing parsers](https://cloud.google.com/chronicle/docs/event-processing/parser-tips-troubleshooting) — intermediate fields are discarded unless mapped to `@output`; one parser is active per log type.
-- [Manage prebuilt and custom parsers](https://cloud.google.com/chronicle/docs/event-processing/manage-parser-updates)
+- [Overview of the UDM](https://docs.cloud.google.com/chronicle/docs/event-processing/udm-overview)
+- [UDM field list](https://docs.cloud.google.com/chronicle/docs/reference/udm-field-list)
+- [SecurityResult reference](https://docs.cloud.google.com/chronicle/docs/reference/rest/v1alpha/SecurityResult)
+- [Overview of log parsing](https://docs.cloud.google.com/chronicle/docs/event-processing/parsing-overview)
+- [Parser syntax reference](https://docs.cloud.google.com/chronicle/docs/reference/parser-syntax)
+- [Tips and troubleshooting when writing parsers](https://docs.cloud.google.com/chronicle/docs/event-processing/parser-tips-troubleshooting) — intermediate fields are discarded unless mapped to `@output`; one parser is active per log type.
+- [Manage prebuilt and custom parsers](https://docs.cloud.google.com/chronicle/docs/event-processing/manage-parser-updates)
 - [Chronicle content-hub](https://github.com/chronicle/content-hub) — Google's official repository of third-party SecOps parsers. Its CBN parsers (e.g. [`CLOUDFLARE_PAGESHIELD`](https://github.com/chronicle/content-hub/tree/main/content/parsers/third_party/community/CLOUDFLARE_PAGESHIELD)) use the same conventions this one does: initialize fields before `json`, `convert` JSON types to strings, map to `event.idm.read_only_udm.*`, and finalize with `@output`.
-- [UDM search](https://cloud.google.com/chronicle/docs/investigation/udm-search) — `GENERIC_EVENT` events only surface in raw-log / UDM search, not curated views.
-- [Install the Bindplane agent (collector)](https://cloud.google.com/chronicle/docs/install/install-forwarder)
-- [Feed management](https://cloud.google.com/chronicle/docs/administration/feed-management-overview)
+- [UDM search](https://docs.cloud.google.com/chronicle/docs/investigation/udm-search) — `GENERIC_EVENT` events only surface in raw-log / UDM search, not curated views.
+- [Install the Bindplane agent (collector)](https://docs.cloud.google.com/chronicle/docs/install/install-forwarder)
+- [Feed management](https://docs.cloud.google.com/chronicle/docs/administration/feed-management-overview)
 
 ## Additional sources and tooling
 

google_secops_parser/parsedmarc.conf

@@ -5,14 +5,19 @@ filter {
   #
   # Consumes the single-line JSON events that parsedmarc emits through its
   # built-in [syslog] output and maps them to the Unified Data Model (UDM).
-  # No changes to parsedmarc are required: this is a SecOps-side custom parser
-  # (configuration-based normalizer / CBN).
+  # This is a SecOps-side custom parser (configuration-based normalizer / CBN).
+  # SMTP TLS failure-detail rows carry policy_domain / policy_type as of the
+  # parsedmarc release this parser ships with; older parsedmarc versions omit
+  # them on failure rows (the result_type detection fallback still catches
+  # those rows, but ones without an MTA IP or MX hostname have no UDM noun).
   #
   # parsedmarc emits three flat JSON shapes, one object per syslog line, via the
   # CSV-row serializers (parsed_aggregate/failure/smtp_tls_reports_to_csv_rows):
   #   * DMARC aggregate report record  -> detected by "xml_schema"
-  #   * DMARC failure  report record   -> detected by "feedback_type"
-  #   * SMTP TLS        report record   -> detected by "policy_type"
+  #   * DMARC failure  report record   -> detected by "feedback_type" or
+  #     "arrival_date_utc" (text-format failure reports have no Feedback-Type)
+  #   * SMTP TLS        report record   -> detected by "policy_type" or
+  #     "result_type" (failure-detail rows from older parsedmarc versions)
   #
   # ---------------------------------------------------------------------------
   # UDM facts this parser relies on (all from official Google documentation):
@@ -20,28 +25,29 @@ filter {
   #   * Output object & finalize:  build fields under
   #       event.idm.read_only_udm.* and emit with
   #       mutate { merge => { "@output" => "event" } }
-  #       https://cloud.google.com/chronicle/docs/event-processing/parser-syntax
+  #       https://docs.cloud.google.com/chronicle/docs/event-processing/parser-syntax
   #   * Every event MUST set metadata.event_type (predefined enum) and
   #     metadata.event_timestamp; parsing fails if a required field is missing.
   #     A date{} filter with no target sets metadata.event_timestamp; if no
   #     date{} runs, event_timestamp defaults to ingest time.
-  #       https://cloud.google.com/chronicle/docs/event-processing/parsing-overview
+  #       https://docs.cloud.google.com/chronicle/docs/event-processing/parsing-overview
   #   * Every event MUST contain at least one noun
   #     (principal/target/src/intermediary/observer). principal must carry a
   #     machine or user detail and must NOT carry email.
-  #       https://cloud.google.com/chronicle/docs/event-processing/udm-overview
+  #       https://docs.cloud.google.com/chronicle/docs/event-processing/udm-overview
   #   * EMAIL_TRANSACTION and GENERIC_EVENT are valid metadata.event_type values.
   #     GENERIC_EVENT only surfaces in raw-log / UDM search, not curated views.
-  #       https://cloud.google.com/chronicle/docs/investigation/udm-search
+  #       https://docs.cloud.google.com/chronicle/docs/investigation/udm-search
   #   * security_result.action enum: UNKNOWN_ACTION, ALLOW, BLOCK,
   #     ALLOW_WITH_MODIFICATION, QUARANTINE, FAIL, CHALLENGE.
   #     security_result.category enum includes AUTH_VIOLATION
   #     ("Authentication failed"), POLICY_VIOLATION, MAIL_SPOOFING. There is no
   #     "passed" category, so we set a category only on failures.
-  #       https://cloud.google.com/chronicle/docs/reference/rest/v1alpha/SecurityResult
+  #       https://docs.cloud.google.com/chronicle/docs/reference/rest/v1alpha/SecurityResult
   #   * network.email fields: from, reply_to, to, cc, bcc, mail_id, subject.
-  #     (There is no network.email.mail_from.)
-  #       https://cloud.google.com/chronicle/docs/reference/udm-field-list
+  #     (There is no network.email.mail_from.) to, cc, bcc, and subject are
+  #     REPEATED fields, so they take merge (like principal.ip), not replace.
+  #       https://docs.cloud.google.com/chronicle/docs/reference/udm-field-list
   #
   # ---------------------------------------------------------------------------
   # Caveats (read before trusting output):
@@ -51,12 +57,14 @@ filter {
   #      Validate with the sample events in README.md before production use.
   #   2. JSON TYPES ARE PRESERVED. The CBN json{} filter keeps the original JSON
   #      type (a JSON boolean stays a boolean, a number stays a number).
-  #      Booleans (*_aligned / testing / normalized_timespan) are converted to
-  #      strings (step 1b) so `== "true"` / `== "false"` tests work, and stored
-  #      as string_value (content-hub never uses bool_value). Numbers (count /
-  #      *_session_count / source_asn) are stored as number_value -- built as a
-  #      string, convert-ed to uinteger, then renamed to number_value -- so
-  #      SecOps can range-query and sort them. Matches the content-hub parsers.
+  #      Booleans (*_aligned / normalized_timespan) are converted to strings
+  #      (step 1b) so `== "true"` / `== "false"` tests work, and stored as
+  #      string_value (content-hub never uses bool_value). "testing" is NOT a
+  #      boolean: parsedmarc emits the RFC 9990 t= flag as the string "y"/"n".
+  #      Numbers (count / *_session_count / source_asn) are stored as
+  #      number_value -- built as a string, convert-ed to uinteger, then
+  #      renamed to number_value -- so SecOps can range-query and sort them.
+  #      Matches the content-hub parsers.
   #   2b. CONDITIONALS. Every field tested in an `if` is initialized to "" in
   #      step 1a, and guards use `if [field] != ""` (not bare `if [field]`,
   #      which is true for an initialized-but-empty field). Matches content-hub.
@@ -94,6 +102,8 @@ filter {
       "report_id" => ""
       "org_name" => ""
       "org_email" => ""
+      "org_extra_contact_info" => ""
+      "errors" => ""
       "begin_date" => ""
       "end_date" => ""
       "count" => ""
@@ -177,7 +187,7 @@ filter {
     on_error => "no_json_payload"
   }
   if [no_json_payload] {
-    drop { tag => "TAG_MALFORMED_ENCODING" }
+    drop { tag => "TAG_MALFORMED_MESSAGE" }
   }
 
   json {
@@ -185,7 +195,7 @@ filter {
     on_error => "not_json"
   }
   if [not_json] {
-    drop { tag => "TAG_MALFORMED_ENCODING" }
+    drop { tag => "TAG_MALFORMED_MESSAGE" }
   }
 
   # ---------------------------------------------------------------------------
@@ -203,18 +213,23 @@ filter {
       "dmarc_aligned" => "string"
       "spf_aligned" => "string"
       "dkim_aligned" => "string"
-      "testing" => "string"
       "normalized_timespan" => "string"
     }
     on_error => "convert_error"
   }
 
   # ---------------------------------------------------------------------------
-  # 2. Detect the report type from a field that is always present, non-empty,
-  #    and unique to each shape:
-  #      feedback_type -> failure
-  #      policy_type   -> smtp_tls
-  #      xml_schema    -> aggregate
+  # 2. Detect the report type from fields unique to each shape:
+  #      feedback_type or arrival_date_utc -> failure
+  #      policy_type   or result_type      -> smtp_tls
+  #      xml_schema                        -> aggregate
+  #    feedback_type alone is not enough: parse_failure_report never defaults
+  #    it, and the text-format failure path (parsedmarc/__init__.py) synthesizes
+  #    a feedback report with no Feedback-Type field. arrival_date_utc is
+  #    computed for every failure report and appears in no other shape.
+  #    policy_type alone is not enough for streams from parsedmarc versions
+  #    that omitted policy_domain/policy_type on SMTP TLS failure-detail rows;
+  #    result_type is required in every RFC 8460 failure detail.
   #    xml_schema is aggregate-only and parsedmarc defaults it to "draft" when
   #    the report omits <version> (parsedmarc/__init__.py), so it survives a
   #    missing version. It is preferred over: header_from (can be empty when a
@@ -222,9 +237,9 @@ filter {
   #    (a generic name), and dmarc_aligned (a boolean that only becomes testable
   #    after the convert in step 1b -- detection should not depend on that).
   # ---------------------------------------------------------------------------
-  if [feedback_type] != "" {
+  if [feedback_type] != "" or [arrival_date_utc] != "" {
     mutate { replace => { "report_type" => "failure" } }
-  } else if [policy_type] != "" {
+  } else if [policy_type] != "" or [result_type] != "" {
     mutate { replace => { "report_type" => "smtp_tls" } }
   } else if [xml_schema] != "" {
     mutate { replace => { "report_type" => "aggregate" } }
@@ -317,13 +332,9 @@ filter {
       mutate { replace => { "sr.category" => "AUTH_VIOLATION" } }
     }
 
-    mutate {
-      replace => {
-        "sr.description" => "dmarc_aligned=%{dmarc_aligned} spf_aligned=%{spf_aligned} dkim_aligned=%{dkim_aligned} disposition=%{disposition}"
-      }
-      on_error => "agg_sr_desc_error"
-    }
-
+    # No concatenated description: the official guidance is not to pack
+    # multiple values into one UDM field, and every alignment/disposition
+    # value already lands individually in additional.fields below.
     mutate {
       merge => { "event.idm.read_only_udm.security_result" => "sr" }
     }
@@ -341,6 +352,18 @@ filter {
         merge   => { "event.idm.read_only_udm.additional.fields" => "f_org_email" }
       }
     }
+    if [org_extra_contact_info] != "" {
+      mutate {
+        replace => { "f_org_extra.key" => "org_extra_contact_info" "f_org_extra.value.string_value" => "%{org_extra_contact_info}" }
+        merge   => { "event.idm.read_only_udm.additional.fields" => "f_org_extra" }
+      }
+    }
+    if [errors] != "" {
+      mutate {
+        replace => { "f_errors.key" => "errors" "f_errors.value.string_value" => "%{errors}" }
+        merge   => { "event.idm.read_only_udm.additional.fields" => "f_errors" }
+      }
+    }
     if [begin_date] != "" {
       mutate {
         replace => { "f_begin.key" => "begin_date" "f_begin.value.string_value" => "%{begin_date}" }
@@ -353,8 +376,12 @@ filter {
         merge   => { "event.idm.read_only_udm.additional.fields" => "f_end" }
       }
     }
+    # count is a raw JSON number here; if %{} interpolation of a non-string
+    # errors on this tenant, on_error keeps the event alive (count_nan then
+    # skips the merge below).
     mutate {
       replace => { "f_count.key" => "count" "f_count.value.string_value" => "%{count}" }
+      on_error => "count_nan"
     }
     mutate {
       convert => { "f_count.value.string_value" => "uinteger" }
@@ -406,7 +433,8 @@ filter {
         merge   => { "event.idm.read_only_udm.additional.fields" => "f_aspf" }
       }
     }
-    if [testing] == "true" or [testing] == "false" {
+    # RFC 9990 t= flag; parsedmarc emits "y"/"n" (a string, not a boolean).
+    if [testing] != "" {
       mutate {
         replace => { "f_testing.key" => "dmarc_testing" "f_testing.value.string_value" => "%{testing}" }
         merge   => { "event.idm.read_only_udm.additional.fields" => "f_testing" }
@@ -524,6 +552,7 @@ filter {
     }
     mutate {
       replace => { "f_asn.key" => "source_asn" "f_asn.value.string_value" => "%{source_asn}" }
+      on_error => "asn_nan"
     }
     mutate {
       convert => { "f_asn.value.string_value" => "uinteger" }
@@ -624,19 +653,19 @@ filter {
         }
       }
     }
+    # network.email.to and network.email.subject are REPEATED UDM fields
+    # (see the UDM field list), so merge the source field like principal.ip;
+    # a scalar replace is a type mismatch.
     if [original_rcpt_to] != "" {
       mutate {
-        replace => {
-          "event.idm.read_only_udm.network.email.to" => "%{original_rcpt_to}"
-        }
+        merge    => { "event.idm.read_only_udm.network.email.to" => "original_rcpt_to" }
         on_error => "fail_rcpt_error"
       }
     }
     if [subject] != "" {
       mutate {
-        replace => {
-          "event.idm.read_only_udm.network.email.subject" => "%{subject}"
-        }
+        merge    => { "event.idm.read_only_udm.network.email.subject" => "subject" }
+        on_error => "fail_subject_error"
       }
     }
     if [message_id] != "" {
@@ -663,14 +692,9 @@ filter {
     } else {
       mutate { replace => { "srf.action" => "UNKNOWN_ACTION" } }
     }
-    if [auth_failure] != "" {
-      mutate {
-        replace => {
-          "srf.description" => "auth_failure=%{auth_failure} delivery_result=%{delivery_result}"
-        }
-        on_error => "fail_sr_desc_error"
-      }
-    }
+    # No concatenated description: auth_failure and delivery_result each land
+    # individually in additional.fields below (official guidance is not to
+    # pack multiple values into one UDM field).
     mutate {
       merge => { "event.idm.read_only_udm.security_result" => "srf" }
     }
@@ -724,12 +748,59 @@ filter {
         merge   => { "event.idm.read_only_udm.additional.fields" => "g_ad" }
       }
     }
+
+    # Source enrichment from parsedmarc's reverse-DNS / MMDB maps (same
+    # treatment as the aggregate branch; null in offline mode).
+    if [source_base_domain] != "" {
+      mutate {
+        replace => { "g_sbd.key" => "source_base_domain" "g_sbd.value.string_value" => "%{source_base_domain}" }
+        merge   => { "event.idm.read_only_udm.additional.fields" => "g_sbd" }
+      }
+    }
+    if [source_name] != "" {
+      mutate {
+        replace => { "g_sn.key" => "source_name" "g_sn.value.string_value" => "%{source_name}" }
+        merge   => { "event.idm.read_only_udm.additional.fields" => "g_sn" }
+      }
+    }
+    if [source_type] != "" {
+      mutate {
+        replace => { "g_st.key" => "source_type" "g_st.value.string_value" => "%{source_type}" }
+        merge   => { "event.idm.read_only_udm.additional.fields" => "g_st" }
+      }
+    }
+    mutate {
+      replace => { "g_asn.key" => "source_asn" "g_asn.value.string_value" => "%{source_asn}" }
+      on_error => "fail_asn_nan"
+    }
+    mutate {
+      convert => { "g_asn.value.string_value" => "uinteger" }
+      on_error => "fail_asn_nan"
+    }
+    if ![fail_asn_nan] {
+      mutate { rename => { "g_asn.value.string_value" => "g_asn.value.number_value" } }
+      mutate { merge => { "event.idm.read_only_udm.additional.fields" => "g_asn" } }
+    }
+    if [source_as_name] != "" {
+      mutate {
+        replace => { "g_asnm.key" => "source_as_name" "g_asnm.value.string_value" => "%{source_as_name}" }
+        merge   => { "event.idm.read_only_udm.additional.fields" => "g_asnm" }
+      }
+    }
+    if [source_as_domain] != "" {
+      mutate {
+        replace => { "g_asd.key" => "source_as_domain" "g_asd.value.string_value" => "%{source_as_domain}" }
+        merge   => { "event.idm.read_only_udm.additional.fields" => "g_asd" }
+      }
+    }
   }
 
   # ===========================================================================
   # SMTP TLS  ->  GENERIC_EVENT
   #   No per-message semantics; modeled as a generic event whose noun is the
-  #   reported policy domain (target.hostname), which is always present. Per the
+  #   reported policy domain (target.hostname), present on every row as of the
+  #   paired parsedmarc serializer fix (receiving_mx_hostname fallback for
+  #   failure rows from older parsedmarc versions). Per the
   #   docs, GENERIC_EVENT only appears in raw-log / UDM search, not curated views.
   # ===========================================================================
   if [report_type] == "smtp_tls" {
@@ -754,13 +825,22 @@ filter {
       }
     }
 
-    # -- target: the reported policy domain (always present -> the noun). --
+    # -- target: the reported policy domain (present on every row as of the
+    #    paired parsedmarc serializer fix -> the noun). Fall back to the
+    #    receiving MX hostname for failure rows from older parsedmarc
+    #    versions, which omitted policy_domain on failure-detail rows.
     if [policy_domain] != "" {
       mutate {
         replace => {
           "event.idm.read_only_udm.target.hostname" => "%{policy_domain}"
         }
       }
+    } else if [receiving_mx_hostname] != "" {
+      mutate {
+        replace => {
+          "event.idm.read_only_udm.target.hostname" => "%{receiving_mx_hostname}"
+        }
+      }
     }
     # receiving MTA IP, when a per-session failure row carries it.
     if [receiving_ip] != "" {
@@ -838,6 +918,7 @@ filter {
     }
     mutate {
       replace => { "t_ssc.key" => "successful_session_count" "t_ssc.value.string_value" => "%{successful_session_count}" }
+      on_error => "ssc_nan"
     }
     mutate {
       convert => { "t_ssc.value.string_value" => "uinteger" }
@@ -849,6 +930,7 @@ filter {
     }
     mutate {
       replace => { "t_fsc.key" => "failed_session_count" "t_fsc.value.string_value" => "%{failed_session_count}" }
+      on_error => "fsc_nan"
     }
     mutate {
       convert => { "t_fsc.value.string_value" => "uinteger" }
@@ -908,6 +990,9 @@ filter {
     }
   }
 
+  # "event" appears in the tips doc's reserved-variable list, but Google's own
+  # parser-syntax examples and the content-hub parsers all use it as the
+  # @output intermediate; keeping the production convention here.
   mutate {
     merge => { "@output" => "event" }
   }