Include policy identity on SMTP TLS failure-detail rows

RFC 8460 §4.3 nests each failure detail inside a policy, but
parsed_smtp_tls_reports_to_csv_rows only attached policy_domain and
policy_type to the per-policy summary row. Failure-detail rows therefore
had empty policy_domain/policy_type CSV columns, and flat-JSON consumers
(syslog, GELF) could not attribute a failure detail to its policy — the
Google SecOps parser in this PR could not even detect those rows as SMTP
TLS reports, since they carried none of the shape-identifying fields.

Also rebuild the row template per policy: policy_strings and
mx_host_patterns from an earlier policy leaked into a later policy that
did not define them, because the template dict was created once per
report and mutated inside the policies loop.

Both regression tests fail on the previous serializer with
KeyError: 'policy_domain'.

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

CHANGELOG.md

@@ -1,5 +1,31 @@
 # Changelog
 
+## Unreleased
+
+### Features
+
+- **Google SecOps (Chronicle) UDM parser** (`google_secops_parser/`). A
+  configuration-based normalizer (CBN) that maps the JSON events parsedmarc
+  emits through the `[syslog]` output to the Unified Data Model: DMARC
+  aggregate and failure reports become `EMAIL_TRANSACTION` events, SMTP TLS
+  reports become `GENERIC_EVENT` events. Ships with real sample events for the
+  SecOps parser-validation tool; see `google_secops_parser/README.md` for
+  installation, field mappings, and caveats (not yet validated against a live
+  tenant).
+
+### Bug fixes
+
+- **SMTP TLS failure-detail rows now carry `policy_domain` and `policy_type`**
+  in the flat row output (`parsed_smtp_tls_reports_to_csv_rows`). RFC 8460 §4.3
+  nests each failure detail inside a policy, but the serializer only attached
+  the policy identity to the per-policy summary row, so the `policy_domain` and
+  `policy_type` CSV columns were always empty on failure-detail rows and JSON
+  consumers (syslog, GELF) could not attribute a failure detail to its policy.
+- **Per-policy fields no longer leak across policies** in the same serializer:
+  `policy_strings` / `mx_host_patterns` from an earlier policy were reused for
+  a later policy that did not define them, because the row template dict was
+  built once per report instead of once per policy.
+
 ## 10.2.1
 
 ### Changes

parsedmarc/__init__.py

@@ -685,15 +685,20 @@ def parsed_smtp_tls_reports_to_csv_rows(
             "end_date": report["end_date"],
             "report_id": report["report_id"],
         }
-        record: dict[str, Any] = common_fields.copy()
         for policy in report["policies"]:
+            # Rebuild the shared base per policy so per-policy fields cannot
+            # leak into a later policy that omits them, and so failure-detail
+            # rows carry the policy identity (policy_domain / policy_type) —
+            # RFC 8460 §4.3 nests failure-details inside a policy, so every
+            # detail row belongs to exactly one policy.
+            record: dict[str, Any] = common_fields.copy()
+            record["policy_domain"] = policy["policy_domain"]
+            record["policy_type"] = policy["policy_type"]
             if "policy_strings" in policy:
                 record["policy_strings"] = "|".join(policy["policy_strings"])
             if "mx_host_patterns" in policy:
                 record["mx_host_patterns"] = "|".join(policy["mx_host_patterns"])
             successful_record = record.copy()
-            successful_record["policy_domain"] = policy["policy_domain"]
-            successful_record["policy_type"] = policy["policy_type"]
             successful_record["successful_session_count"] = policy[
                 "successful_session_count"
             ]

tests/test_init.py

@@ -1540,6 +1540,75 @@ class Test(unittest.TestCase):
         self.assertTrue(len(rows) >= 2)
         self.assertEqual(rows[0]["organization_name"], "Org")
         self.assertEqual(rows[0]["policy_domain"], "example.com")
+        # Failure-detail rows must carry the policy identity too: RFC 8460
+        # §4.3 nests failure-details inside a policy, so each detail row
+        # belongs to exactly one policy. Without these keys, downstream
+        # consumers (CSV policy_domain/policy_type columns, the syslog JSON
+        # shape) cannot attribute a failure row to its policy.
+        failure_row = rows[1]
+        self.assertEqual(failure_row["result_type"], "cert-expired")
+        self.assertEqual(failure_row["policy_domain"], "example.com")
+        self.assertEqual(failure_row["policy_type"], "sts")
+
+    def testSmtpTlsCsvRowsNoCrossPolicyLeak(self):
+        """Per-policy fields must not leak into a later policy's rows
+
+        RFC 8460 §4.3: policy-string and mx-host-pattern are properties of a
+        single policy object. The serializer previously reused one shared
+        record dict across the policies loop, so a policy without
+        policy-string inherited the previous policy's value.
+        """
+        report_json = json.dumps(
+            {
+                "organization-name": "Org",
+                "date-range": {
+                    "start-datetime": "2024-01-01T00:00:00Z",
+                    "end-datetime": "2024-01-02T00:00:00Z",
+                },
+                "contact-info": "a@b.com",
+                "report-id": "r1",
+                "policies": [
+                    {
+                        "policy": {
+                            "policy-type": "sts",
+                            "policy-domain": "first.example",
+                            "policy-string": ["v: STSv1"],
+                            "mx-host-pattern": ["*.first.example"],
+                        },
+                        "summary": {
+                            "total-successful-session-count": 10,
+                            "total-failure-session-count": 0,
+                        },
+                    },
+                    {
+                        "policy": {
+                            "policy-type": "tlsa",
+                            "policy-domain": "second.example",
+                        },
+                        "summary": {
+                            "total-successful-session-count": 5,
+                            "total-failure-session-count": 1,
+                        },
+                        "failure-details": [
+                            {
+                                "result-type": "validation-failure",
+                                "failed-session-count": 1,
+                            }
+                        ],
+                    },
+                ],
+            }
+        )
+        parsed = parsedmarc.parse_smtp_tls_report_json(report_json)
+        rows = parsedmarc.parsed_smtp_tls_reports_to_csv_rows(parsed)
+        second_policy_rows = [
+            row for row in rows if row["policy_domain"] == "second.example"
+        ]
+        self.assertEqual(len(second_policy_rows), 2)
+        for row in second_policy_rows:
+            self.assertEqual(row["policy_type"], "tlsa")
+            self.assertNotIn("policy_strings", row)
+            self.assertNotIn("mx_host_patterns", row)
 
     def testParsedAggregateReportsToCsvRowsList(self):
         """parsed_aggregate_reports_to_csv_rows handles list of reports"""