mirror of
https://github.com/domainaware/parsedmarc.git
synced 2026-08-01 13:12:18 +00:00
* Fix DKIM/SPF alignment detail cross-product in dashboards (#169) Elasticsearch and OpenSearch dynamic-map the dkim_results/spf_results object arrays as `object` (create_indexes never registers the DSL document mappings), so Lucene flattens each array into independent multi-valued fields and stacked terms aggregations on dkim_results.selector/.domain/.result return every combination of values across a report's signatures — each phantom row repeating the full message count. Aggregate documents now also carry dkim_results_combined and spf_results_combined: one "selector / domain / result" ("scope / domain / result") string per auth result, composed in add_dkim_result/add_spf_result. The Kibana/OpenSearch Dashboards and Grafana (Elasticsearch) alignment-detail tables aggregate those instead, and the Splunk detail panels pair the values with mvzip/mvexpand. A documented idempotent _update_by_query backfills documents saved by older versions; the query matches only documents that have auth results and lack the combined fields, because an `exists` query cannot see an empty array. Also corrects the dead _SPFResult.results (plural) declaration to `result` (the save path always wrote the singular key), fixes the result parameter annotations on add_dkim_result/add_spf_result, and removes the Grafana dmarcian.com DKIM-checker data link, which required the separate domain/selector columns. The SMTP TLS visualizations have the same class of defect and are tracked separately. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Address Copilot review findings on #839 Reword the combined-field regression test docstrings: the DKIM/SPF auth results are dynamic-mapped as plain `object`, not the `nested` mapping type the previous wording implied — the distinction is the crux of the fix. Also drop the inert renameByName entries Copilot flagged on the Grafana Overview and DKIM Alignment Details panels, which referenced fields those panels' queries no longer (or never) produced. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Add PR #839 review lessons to AGENTS.md Extend the "Review passes cover prose" section with two rules from the #839 Copilot findings: docstrings/comments get the same text-level review pass as docs and dashboard labels (with suspicion for dual-use terms like "nested" near Elasticsearch code), and inert config entries inside hunks a PR already rewrites should be cleaned rather than preserved to minimize the diff. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Reflow create_indexes comments flagged by Copilot The line wrap placed "#169" directly after the comment marker, so the raw source read "# #169;". Reword so the issue reference stays on one line. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Extend the hunk-proofreading rule with rendered-text wraps Fold the PR #839 second-round Copilot lesson into the existing rule: proofread how wrapped lines render (comment markers, punctuation at wrap points), not just the wording itself. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Rename Overview combined-result column labels (Copilot round 3) The Overview table's "DKIM Auth Result" / "SPF Auth Result" labels were kept when the columns switched to the combined "selector / domain / result" values, leaving the headers misleading. Rename them to match the detail panels' convention and retarget the byName width overrides that matched the old labels, widening them for the longer values. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Document per-signature row semantics in the alignment tables A message carrying multiple DKIM signatures appears once per signature in the details tables, so summing the messages column across rows can exceed the total message count. State that explicitly rather than leaving readers to infer it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Address Copilot round-4 findings on dashboards Fix three pre-existing saved-object title typos in the OpenSearch ndjson (leading space on "Aggregate DMARC passed DMARC", trailing space on "Aggregate DMARC reporting organizations", double space in "map of message sources by country"), in both the top-level title and the embedded visState title. Normalize the Splunk DKIM details placeholders: the base search's fillnull renders wholly-missing DKIM fields as the literal string "null", so unsigned mail showed "null / null / null" while the SPF panel shows "none". Rewrite the values to "none" after the signature split, where the fields are single-valued and the mvzip pairing cannot be disturbed. Verified against the dev Splunk that no truncation or mis-pairing occurs either way, since fillnull guarantees the fields are never actually null. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Backfill combined DKIM/SPF fields automatically at startup migrate_indexes() now backfills dkim_results_combined and spf_results_combined on aggregate documents saved by older versions, so ES/OS users get historical data in the reworked alignment tables without running the documented _update_by_query by hand. The backfill is submitted as a non-blocking background task (wait_for_completion=false, conflicts=proceed) guarded by a cheap count query, making repeated startups a fast no-op once an index is backfilled; any cluster error is logged as a warning and retried at the next startup rather than raised. The manual command remains documented for users who upgrade dashboards without pointing the new parsedmarc at the cluster or who want to control write-load timing. The legacy published_policy.fo long-to-text reindex migration in the OpenSearch module is kept ahead of the new backfill, for clusters upgraded from very old data. Verified end-to-end against the live dev environment: a real CLI startup backfilled 9 stripped OpenSearch documents (logged with task ID) while the already-backfilled Elasticsearch side stayed silent, and a second startup was silent on both engines. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Match backfill guard on either domain or result subfield End-to-end upgrade testing (real parsedmarc 10.2.4 ingest, then a branch startup) surfaced that an exists query cannot see an empty string: a text field with no tokens is invisible to exists. The parsers we audited never store an auth result with an empty or missing domain — they drop such entries entirely, so the previous domain-only guard was sufficient for their data — but the storage shape of every historical parsedmarc version can't be audited, so the guard (and the documented manual command) now matches either the domain or the result subfield per protocol. Matching either costs nothing and cannot skip a document that has something to backfill. Verified by recomputing expected combined values from _source for all 2,299 documents on both engines: every document with stored auth results has exactly the recomputed pairs, and the legacy fo migration correctly did not fire on typeless indexes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Add auth-result filter controls to the Kibana/OSD aggregate dashboard The combined per-signature columns fixed the #169 cross-product but left no way to click-filter by an individual selector, domain, or result. Add an "Aggregate DMARC auth result filters" input_control_vis panel above the SPF/DKIM details tables with six option-list dropdowns (DKIM selector/domain/result, SPF scope/domain/result) that emit ordinary dashboard-wide filter pills. Works on both Kibana 8.19 and OpenSearch Dashboards 3, verified by driving the controls in both UIs against the issue's two-signature repro report. Documented in kibana.md, including the flat-mapping caveat: combining two component filters matches documents where any signature satisfies each condition individually. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Scale Grafana source-country map markers with message volume The "Map of Message Source Countries" panel drew fixed 5 px dark-green markers at 50% opacity — nearly invisible on the dark basemap, so the panel read as empty even when data was flowing (verified via the query API). Markers now scale with Sum(message_count) (min 4, max 30 px) at 0.8 opacity in a higher-contrast green. Pre-existing issue; the identically-styled failure-dashboard map panel is intentionally left untouched. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Correct the nested-mapping rationale in the create_indexes comments The comments claimed Kibana/OSD/Grafana "cannot terms-aggregate fields inside a nested mapping" — too absolute. Fact-checked empirically and against primary docs: Kibana/OSD visual editors (Lens and classic Visualize) do not support nested fields, but Vega panels can run nested aggregations (they just cannot render tables, per Elastic's docs), and Grafana >= 9.4 has a nested bucket aggregation (grafana/grafana#62301) but no reverse_nested, so parent-level metrics like Sum(message_count) return 0 inside per-signature buckets (reproduced live). Conclusion unchanged: the dynamic object mapping stays load-bearing for the shipped dashboards. PR #839's body was updated with the same correction. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Pin ruff exactly, matching the existing pyright pin rationale CI installs the [build] extra fresh on every run, and ruff was the one lint tool left unpinned. ruff 0.16.0 (released this week) began flagging this codebase's str.format() house style, so every PR started failing lint on lines it never touched. Pin to 0.15.21 — the version the codebase is clean under — with the same bump-deliberately comment pyright carries. Upgrading to 0.16 and converting to f-strings can be its own PR. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Address Copilot findings: harden migrate_indexes, normalize panel titles Three unresolved review threads, all verified against cli.py's re-raising init handler before fixing: - elastic.py/opensearch.py: connections.get_connection() sat outside migrate_indexes()'s try/except, so a connection-registration failure would abort startup despite the docstring's promise that migration errors are caught and logged. Now caught, logged, and skipped until the next startup. - opensearch.py: the legacy published_policy.fo migration loop did unguarded network I/O (exists/get_field_mapping/reindex/delete), so a transient cluster error aborted startup on the OpenSearch path while the identical situation on the Elasticsearch path was logged and survived. Each index's migration attempt is now wrapped, warns, and moves on. - opensearch_dashboards.ndjson: normalized two pre-existing panel titles in the aggregate dashboard's panelsJSON ("Reporting organizations " trailing space, "Map of message sources by country" double space). Regression tests assert migrate_indexes never propagates connection or per-index cluster errors on either backend. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Clarify that Nested() on auth-result fields is in-memory shape only Copilot flagged that _AggregateReportDoc declares dkim_results and spf_results with Nested(...) while the create_indexes comment insists the stored mapping must stay dynamic `object`. Both are true: the Nested declaration only shapes the DSL's in-memory document building and is never installed as a mapping, because create_indexes skips Index.document() registration. Say so at both sites, in both backends, so nobody "fixes" the mismatch by registering the mapping — which would install real nested mappings and blank the dashboards. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Refer to the filter panel by its displayed title in docs and CHANGELOG The dashboard convention is a short panel display title backed by a long-form saved-object name ("SPF details" / "Aggregate DMARC SPF details"), and the new controls panel follows it. The docs and CHANGELOG named the panel by its saved-object title, which is not what a user sees on the dashboard; use the displayed "Auth result filters" instead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Fix misspelled column label in the failure email samples table The "DMARC failure email samples" visualization labeled its authentication_results column "autentication_results". The underlying field reference was already correct; only the user-facing customLabel was misspelled. A sweep of every title and customLabel in the ndjson found no other misspellings. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Extend the combined-field fix to SMTP TLS documents SMTP TLS reports have the same cross-product defect as the DKIM/SPF alignment tables (issue #169), one level deeper: policies is an object array and each policy's failure_details is an object array inside it, so stacked terms aggregations on their subfields fabricate rows. Documents now also carry policies_combined ("domain / type" per policy) and failure_details_combined ("domain / type / result / sending mta / receiving ip / mx" per failure detail), composed at save time with the same "none" fallbacks as the aggregate fields. migrate_indexes() gains smtp_tls_indexes and backfills old documents with the same guarded, non-blocking update_by_query pattern; cli.py wires the index name in on both backends, and the manual _update_by_query command is documented. Also fixes two adjacent dead fields: add_failure_details stored additional_information_uri under the wrong constructor kwarg (additional_information), and receiving_mx_hostname had no declaration despite always being stored. Verified live on ES 8.19 and OpenSearch 3: a two-policy repro report yields exactly 2 policy rows and 2 failure-detail rows via the combined fields where the old stacked aggregations return 4 of each; the startup backfill converted the 4 pre-existing sample documents on both engines with zero recompute mismatches. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Rework the SMTP TLS dashboards onto the combined fields Kibana/OSD: "SMTP TLS domains" replaces its stacked policy_domain × policy_type terms with one terms agg on policies_combined.keyword; "SMTP TLS failure details" replaces six stacked terms spanning both array levels with one on failure_details_combined.keyword; the smtp_tls* index-pattern field cache gains the new fields. The "reporting organizations" table only buckets on doc-level org_name and needed no change. Splunk: the base search now expands policies at the JSON level (spath + mvexpand) so policy fields are scalars per event, and the failure details panel expands the second level the same way — sums are the detail's own failed_session_count, correctly paired. Verified via the search REST API: a two-policy repro returns exactly one row per real failure detail with per-detail counts. kibana.md documents the per-policy/per-detail row semantics and the honest caveat that session-count sums remain per report document. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Persist additional_info_uri from parsed SMTP TLS failure details Copilot caught that the savers read additional_information_uri from the parsed failure-detail dict, but the parser's key is additional_info_uri (SMTPTLSFailureDetailsOptional in types.py, set in parse_smtp_tls_report_json), so the URI was never persisted — the read-side half of the dead-field bug whose write-side half (wrong constructor kwarg) was fixed earlier. Read the parser's key first, keeping the long-form key as a fallback for dicts built by other callers. Regression test proven to fail on the unfixed savers. Also restructured the expected combined-string test values into named locals so no implicit string concatenation sits inside a list literal. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Say "inner doc", not "nested doc", in the singular-key test docstrings Final review sweep: in this codebase "nested" is reserved for the Elasticsearch mapping type, and these InnerDoc-serialization docstrings used it colloquially — the same dual-use-term trap documented in AGENTS.md. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Sean Whalen
co-authored by
Claude Fable 5
Claude Fable 5
parent
14d881ef74
commit
23c5ea9ad6
@@ -179,9 +179,14 @@ Before rewriting a tracked list/data file from freshly-generated content (anythi
|
||||
A review that only verifies functional/numeric correctness (queries return the right values, files import cleanly, types check) will sail past exactly the defects a text-first reviewer catches. On PR #834, four such misses survived a thorough functional review: two long-standing typos inside the OSD ndjson ("SMPT TLS", "filed DMARC"), a typo on an *unchanged* line adjacent to a docs edit, and hand-written bootstrap glue that duplicated the script's existing `wait_for()` helper. Rules drawn from that:
|
||||
|
||||
- **Whole-file canonical exports put every line in the diff — review them as text, too.** Re-exporting `dashboards/opensearch/opensearch_dashboards.ndjson` or a Grafana JSON from a running instance rewrites the entire file, so pre-existing user-facing strings (saved-object titles, markdown panels, column labels) are formally part of the change. A semantic before/after comparison ("attributes identical") proves no unintended changes but deliberately looks through pre-existing content problems; add one text-level pass over titles and markdown before committing.
|
||||
- **Proofread the whole hunk around prose edits, not just the `+`/`-` lines.** Typos one line away from an edit are in the reviewer's context window and fair game; they should be in yours.
|
||||
- **Proofread the whole hunk around prose edits, not just the `+`/`-` lines.** Typos one line away from an edit are in the reviewer's context window and fair game; they should be in yours. Proofread the *rendered* text, not just the wording: on PR #839 a comment wrapped so `#169` landed right after the `#` comment marker, making the raw source read `# #169;` — no line was ever wrong, but the wrap point was. Watch how wraps interact with markers and punctuation (`#` before an issue number, a trailing `-`, a code span split across lines) and reflow rather than argue the text is technically correct.
|
||||
- **Code written mid-incident gets the same review bar as planned code.** Before writing new shell/infra glue while firefighting, check the file for an existing helper that already does it (e.g. `wait_for()` in `dashboard-dev-bootstrap.sh`), and give your own inline code the same scrutiny you'd give a subagent's.
|
||||
|
||||
Two more rules, drawn from the PR #839 review (Copilot caught both after a thorough Fable pass missed them):
|
||||
|
||||
- **Docstrings and comments are prose surface too — and beware dual-use terms.** A regression-test docstring described DKIM/SPF results as "stored as nested object arrays"; in Elasticsearch/OpenSearch "nested" is a specific mapping type, and the fix under review hinged on the fields being dynamic-mapped as plain `object`, *not* `nested`. The reviewer had held both facts all session, so the blended sentence pattern-matched as true — author's-context blindness that a fresh reader doesn't share. Give docstrings/comments the same text-level pass as docs and dashboard labels, with extra suspicion for words that are both colloquial English and load-bearing technical terms near the code in question ("nested", "index", "keyword" in anything Elasticsearch-adjacent).
|
||||
- **Clean inert config inside hunks the diff already rewrites.** Stale entries (e.g. orphaned `renameByName` keys in a Grafana panel) sitting inside a block the PR is editing anyway cost nothing to remove and confuse every later reader if kept; "minimize the diff" is the wrong tiebreaker there. It remains the right tiebreaker for untouched panels/files — don't expand a PR's blast radius to chase pre-existing cruft elsewhere.
|
||||
|
||||
## Releases
|
||||
|
||||
A release isn't done until built artifacts are attached to the GitHub release page. Full sequence:
|
||||
|
||||
@@ -13,6 +13,11 @@
|
||||
- **The Elasticsearch/OpenSearch aggregate dashboards' over-time charts (and the Grafana ES dashboard's summary pies and time series) bucketed on the multi-valued `date_range` field**; a date histogram counts a report once per value, double-counting any report whose begin and end dates fall in different buckets. All date histograms and time-range filters now use the single-valued `date_begin`, matching the report-begin semantics of the PostgreSQL (`begin_date`) and Splunk (`_time` = interval begin) dashboards.
|
||||
- **Aggregate-report policy and authentication result words are now normalized to lowercase** ([#288](https://github.com/domainaware/parsedmarc/issues/288)): reporters that emit mixed-case values such as `Pass` no longer create duplicate result categories in outputs and dashboards.
|
||||
- **The results email (SMTP and Microsoft Graph) is no longer sent when no reports were parsed** ([#200](https://github.com/domainaware/parsedmarc/issues/200)): previously an empty run — an empty inbox, or one where every message was invalid — still emailed a zip of headers-only CSVs. The email step is now skipped with an INFO log when the run produced no aggregate, failure, or SMTP TLS reports.
|
||||
- **The DKIM/SPF alignment-detail tables on the Kibana/OpenSearch Dashboards, Grafana (Elasticsearch), and Splunk aggregate dashboards no longer show a selector × domain × result cross-product** ([#169](https://github.com/domainaware/parsedmarc/issues/169)). Elasticsearch/OpenSearch flatten the `dkim_results`/`spf_results` object arrays (they are dynamic-mapped as `object`, not `nested`), so stacking terms aggregations on their subfields produced every combination of selector/domain/result across a report's signatures, each repeating the full message count. Aggregate documents now also carry `dkim_results_combined` and `spf_results_combined` — one `"selector / domain / result"` (`"scope / domain / result"`) string per auth result — and the dashboards aggregate those instead; the Splunk detail panels now pair the values with `mvzip`/`mvexpand`. Documents saved by older versions are now backfilled automatically at startup (a non-blocking, idempotent background task); the documented `_update_by_query` command (see the Elasticsearch docs page) remains available for running the backfill manually. The Grafana "DKIM Alignment Details" panel's dmarcian.com DKIM-checker data link was removed because it required the separate domain/selector columns. The SMTP TLS visualizations had the same class of defect; that fix is described in its own entry below.
|
||||
- **The Kibana/OpenSearch Dashboards aggregate dashboard now includes an "Auth result filters" control panel** above the SPF/DKIM details tables, with dropdowns for DKIM selector/domain/result and SPF scope/domain/result, so results can still be filtered by individual components alongside the combined per-signature columns.
|
||||
- **SMTP TLS visualizations had the same cross-product defect as the DKIM/SPF alignment tables** ([#169](https://github.com/domainaware/parsedmarc/issues/169)): `policies` is an object array, and each policy's `failure_details` is itself an object array inside it, so stacking terms aggregations on their subfields cross-products the same way. SMTP TLS documents now also carry `policies_combined` — one `"policy_domain / policy_type"` string per policy — and `failure_details_combined` — one `"policy_domain / policy_type / result_type / sending_mta_ip / receiving_ip / receiving_mx_hostname"` string per failure detail — and documents saved by older versions are backfilled automatically at startup the same non-blocking, idempotent way as the DKIM/SPF backfill; the equivalent manual `_update_by_query` command is documented on the Elasticsearch docs page. Also fixed two adjacent dead fields found while making this change: `_SMTPTLSFailureDetailsDoc` declared `additional_information_uri`, but `add_failure_details` passed it to the constructor as `additional_information`, so it was never actually populated on the declared field; and `receiving_mx_hostname`, which `add_failure_details` has always stored, had no field declaration at all. Both are now correctly wired in the Elasticsearch and OpenSearch outputs.
|
||||
- **The Grafana "Map of Message Source Countries" panel's markers now scale with message volume and use a higher-contrast style.** Previously the markers were fixed-size 5 px dots at 50% opacity in dark green, which were nearly invisible on the dark basemap.
|
||||
- **Corrected the dead `_SPFResult.results` (plural) field declaration to `result`**, matching what was always written to it.
|
||||
|
||||
## 10.2.4
|
||||
|
||||
|
||||
@@ -2132,12 +2132,13 @@
|
||||
"showLegend": true,
|
||||
"style": {
|
||||
"color": {
|
||||
"fixed": "dark-green"
|
||||
"fixed": "green"
|
||||
},
|
||||
"opacity": 0.5,
|
||||
"opacity": 0.8,
|
||||
"size": {
|
||||
"field": "Sum",
|
||||
"fixed": 5,
|
||||
"min": 2,
|
||||
"min": 4,
|
||||
"max": 30
|
||||
},
|
||||
"symbol": {
|
||||
@@ -2426,24 +2427,24 @@
|
||||
{
|
||||
"matcher": {
|
||||
"id": "byName",
|
||||
"options": "SPF Auth Result"
|
||||
"options": "SPF Scope / Domain / Result"
|
||||
},
|
||||
"properties": [
|
||||
{
|
||||
"id": "custom.width",
|
||||
"value": 115
|
||||
"value": 200
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"matcher": {
|
||||
"id": "byName",
|
||||
"options": "DKIM Auth Result"
|
||||
"options": "DKIM Selector / Domain / Result"
|
||||
},
|
||||
"properties": [
|
||||
{
|
||||
"id": "custom.width",
|
||||
"value": 123
|
||||
"value": 260
|
||||
}
|
||||
]
|
||||
},
|
||||
@@ -2598,7 +2599,7 @@
|
||||
{
|
||||
"$$hashKey": "object:412",
|
||||
"fake": true,
|
||||
"field": "spf_results.result.keyword",
|
||||
"field": "spf_results_combined.keyword",
|
||||
"id": "16",
|
||||
"settings": {
|
||||
"min_doc_count": "1",
|
||||
@@ -2626,7 +2627,7 @@
|
||||
{
|
||||
"$$hashKey": "object:461",
|
||||
"fake": true,
|
||||
"field": "dkim_results.result.keyword",
|
||||
"field": "dkim_results_combined.keyword",
|
||||
"id": "10",
|
||||
"settings": {
|
||||
"min_doc_count": "1",
|
||||
@@ -2666,7 +2667,7 @@
|
||||
"Sum": 11,
|
||||
"disposition.keyword": 4,
|
||||
"dkim_aligned": 6,
|
||||
"dkim_results.result.keyword": 9,
|
||||
"dkim_results_combined.keyword": 9,
|
||||
"header_from.keyword": 10,
|
||||
"org_name.keyword": 7,
|
||||
"source_base_domain.keyword": 0,
|
||||
@@ -2674,16 +2675,14 @@
|
||||
"source_ip_address.keyword": 2,
|
||||
"source_reverse_dns.keyword": 1,
|
||||
"spf_aligned": 5,
|
||||
"spf_results.result.keyword": 8
|
||||
"spf_results_combined.keyword": 8
|
||||
},
|
||||
"renameByName": {
|
||||
"Sum": "Messages",
|
||||
"date_range": "Date",
|
||||
"disposition.keyword": "Disposition",
|
||||
"dkim_aligned": "DKIM",
|
||||
"dkim_results.domain.keyword": "DKIM Domain",
|
||||
"dkim_results.result.keyword": "DKIM Auth Result",
|
||||
"dkim_results.selector.keyword": "DKIM Selector",
|
||||
"dkim_results_combined.keyword": "DKIM Selector / Domain / Result",
|
||||
"envelope_from.keyword": "Envelope From",
|
||||
"header_from.keyword": "Email Domain",
|
||||
"org_name.keyword": "Reporter",
|
||||
@@ -2692,7 +2691,7 @@
|
||||
"source_ip_address.keyword": "Source IP",
|
||||
"source_reverse_dns.keyword": "PTR",
|
||||
"spf_aligned": "SPF",
|
||||
"spf_results.result.keyword": "SPF Auth Result"
|
||||
"spf_results_combined.keyword": "SPF Scope / Domain / Result"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3431,7 +3430,7 @@
|
||||
{
|
||||
"$$hashKey": "object:459",
|
||||
"fake": true,
|
||||
"field": "spf_results.result.keyword",
|
||||
"field": "spf_results_combined.keyword",
|
||||
"id": "8",
|
||||
"settings": {
|
||||
"min_doc_count": 1,
|
||||
@@ -3502,7 +3501,7 @@
|
||||
"header_from.keyword": "Header From",
|
||||
"source_base_domain.keyword": "Reverse DNS Base",
|
||||
"spf_aligned": "SPF Aligned",
|
||||
"spf_results.result.keyword": "SPF Result"
|
||||
"spf_results_combined.keyword": "SPF Scope / Domain / Result"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3590,22 +3589,12 @@
|
||||
{
|
||||
"matcher": {
|
||||
"id": "byName",
|
||||
"options": "DKIM Selector"
|
||||
"options": "DKIM Selector / Domain / Result"
|
||||
},
|
||||
"properties": [
|
||||
{
|
||||
"id": "custom.width",
|
||||
"value": 320
|
||||
},
|
||||
{
|
||||
"id": "links",
|
||||
"value": [
|
||||
{
|
||||
"targetBlank": true,
|
||||
"title": "Open dmarcian.com DKIM Record Checker",
|
||||
"url": "https://dmarcian.com/dkim-inspector/?domain=${__data.fields[\"dkim_results.domain.keyword\"]}&selector=${__data.fields[\"dkim_results.selector.keyword\"]}"
|
||||
}
|
||||
]
|
||||
"value": 400
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -3649,7 +3638,7 @@
|
||||
{
|
||||
"$$hashKey": "object:458",
|
||||
"fake": true,
|
||||
"field": "dkim_results.selector.keyword",
|
||||
"field": "dkim_results_combined.keyword",
|
||||
"id": "7",
|
||||
"settings": {
|
||||
"min_doc_count": "1",
|
||||
@@ -3660,34 +3649,6 @@
|
||||
},
|
||||
"type": "terms"
|
||||
},
|
||||
{
|
||||
"$$hashKey": "object:459",
|
||||
"fake": true,
|
||||
"field": "dkim_results.domain.keyword",
|
||||
"id": "8",
|
||||
"settings": {
|
||||
"min_doc_count": 1,
|
||||
"missing": "-",
|
||||
"order": "desc",
|
||||
"orderBy": "4",
|
||||
"size": "0"
|
||||
},
|
||||
"type": "terms"
|
||||
},
|
||||
{
|
||||
"$$hashKey": "object:460",
|
||||
"fake": true,
|
||||
"field": "dkim_results.result.keyword",
|
||||
"id": "9",
|
||||
"settings": {
|
||||
"min_doc_count": 1,
|
||||
"missing": null,
|
||||
"order": "desc",
|
||||
"orderBy": "4",
|
||||
"size": "0"
|
||||
},
|
||||
"type": "terms"
|
||||
},
|
||||
{
|
||||
"$$hashKey": "object:798",
|
||||
"fake": true,
|
||||
@@ -3744,14 +3705,11 @@
|
||||
"renameByName": {
|
||||
"Sum": "Messages",
|
||||
"dkim_aligned": "DKIM Aligned",
|
||||
"dkim_results.domain.keyword": "DKIM Domain",
|
||||
"dkim_results.result.keyword": "DKIM Result",
|
||||
"dkim_results.selector.keyword": "DKIM Selector",
|
||||
"dkim_results_combined.keyword": "DKIM Selector / Domain / Result",
|
||||
"envelope_from.keyword": "Envelope From",
|
||||
"header_from.keyword": "Header From",
|
||||
"source_base_domain.keyword": "Reverse DNS Base",
|
||||
"spf_aligned": "SPF Aligned",
|
||||
"spf_results.result.keyword": "SPF Result"
|
||||
"spf_aligned": "SPF Aligned"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -323,7 +323,12 @@
|
||||
<title>SPF details</title>
|
||||
<table>
|
||||
<search base="base_search">
|
||||
<query>| fillnull value="none" source_base_domain | stats sum(message_count) as message_count by header_from,envelope_from,spf_result,source_base_domain,spf_aligned
|
||||
<query>| fillnull value="none" source_base_domain envelope_domain spf_scope spf_result
|
||||
| eval spf_signature=mvzip(mvzip(spf_scope, envelope_domain, " / "), spf_result, " / ")
|
||||
| mvexpand spf_signature
|
||||
| stats sum(message_count) as message_count by header_from, envelope_from, spf_signature, source_base_domain, spf_aligned
|
||||
| eval parts=split(spf_signature, " / "), spf_scope=mvindex(parts, 0), spf_domain=mvindex(parts, 1), spf_result=mvindex(parts, 2)
|
||||
| table header_from, envelope_from, spf_scope, spf_domain, spf_result, spf_aligned, source_base_domain, message_count
|
||||
| sort -message_count</query>
|
||||
</search>
|
||||
<option name="drilldown">none</option>
|
||||
@@ -339,7 +344,13 @@
|
||||
<title>DKIM details</title>
|
||||
<table>
|
||||
<search base="base_search">
|
||||
<query>| fillnull value="none" source_base_domain | stats sum(message_count) as message_count by header_from,dkim_selector,dkim_domain,dkim_result,dkim_aligned,source_base_domain
|
||||
<query>| fillnull value="none" source_base_domain
|
||||
| eval dkim_signature=mvzip(mvzip(dkim_selector, dkim_domain, " / "), dkim_result, " / ")
|
||||
| mvexpand dkim_signature
|
||||
| stats sum(message_count) as message_count by header_from, dkim_signature, dkim_aligned, source_base_domain
|
||||
| eval parts=split(dkim_signature, " / "), dkim_selector=mvindex(parts, 0), dkim_domain=mvindex(parts, 1), dkim_result=mvindex(parts, 2)
|
||||
| eval dkim_selector=if(dkim_selector=="null", "none", dkim_selector), dkim_domain=if(dkim_domain=="null", "none", dkim_domain), dkim_result=if(dkim_result=="null", "none", dkim_result)
|
||||
| table header_from, dkim_selector, dkim_domain, dkim_result, dkim_aligned, source_base_domain, message_count
|
||||
| sort -message_count</query>
|
||||
</search>
|
||||
<option name="drilldown">none</option>
|
||||
|
||||
@@ -2,18 +2,14 @@
|
||||
<label>SMTP TLS Reporting</label>
|
||||
<search id="base_search">
|
||||
<query>
|
||||
index=email sourcetype=smtp:tls organization_name=$organization_name$ policies{}.policy_domain=$policy_domain$ policies{}.policy_type=$policy_type$
|
||||
| rename policies{}.policy_domain as policy_domain
|
||||
| rename policies{}.policy_type as policy_type
|
||||
| rename policies{}.failed_session_count as failed_sessions
|
||||
| rename policies{}.successful_session_count as successful_sessions
|
||||
| rename policies{}.failure_details{}.receiving_mx_hostname as receiving_mx_hostname
|
||||
| rename policies{}.failure_details{}.result_type as failure_type
|
||||
| rename policies{}.failure_details{}.sending_mta_ip as sending_mta_ip
|
||||
| rename policies{}.failure_details{}.receiving_ip as receiving_mta_ip
|
||||
index=email sourcetype=smtp:tls organization_name=$organization_name$
|
||||
| spath policies{} output=policy
|
||||
| mvexpand policy
|
||||
| spath input=policy
|
||||
| search policy_domain=$policy_domain$ policy_type=$policy_type$
|
||||
| rename successful_session_count as successful_sessions
|
||||
| rename failed_session_count as failed_sessions
|
||||
| fillnull value=0 failed_sessions successful_sessions
|
||||
| table *
|
||||
| table *
|
||||
</query>
|
||||
<earliest>$time_range.earliest$</earliest>
|
||||
<latest>$time_range.latest$</latest>
|
||||
@@ -78,8 +74,14 @@ index=email sourcetype=smtp:tls organization_name=$organization_name$ policies{}
|
||||
<table>
|
||||
<search base="base_search">
|
||||
<query>
|
||||
where failed_sessions > 0
|
||||
| stats sum(failed_sessions) as failed_sessions by organization_name, policy_domain, policy_type, failure_type, sending_mta_ip, receiving_mta_ip, receiving_mx_hostname
|
||||
| spath input=policy path=failure_details{} output=detail
|
||||
| mvexpand detail
|
||||
| spath input=detail
|
||||
| where failed_session_count > 0
|
||||
| rename result_type as failure_type
|
||||
| rename receiving_ip as receiving_mta_ip
|
||||
| fillnull value="none" failure_type sending_mta_ip receiving_mta_ip receiving_mx_hostname
|
||||
| stats sum(failed_session_count) as failed_sessions by organization_name, policy_domain, policy_type, failure_type, sending_mta_ip, receiving_mta_ip, receiving_mx_hostname
|
||||
</query>
|
||||
</search>
|
||||
<option name="drilldown">none</option>
|
||||
|
||||
@@ -226,6 +226,158 @@ Kibana index patterns with versions that match the upgraded indexes:
|
||||
7. Import `export.ndjson` by clicking Import from the Kibana
|
||||
Saved Objects page
|
||||
|
||||
## Backfilling the combined DKIM/SPF result fields
|
||||
|
||||
As of the version fixing [#169](https://github.com/domainaware/parsedmarc/issues/169),
|
||||
aggregate documents include `dkim_results_combined` and `spf_results_combined` —
|
||||
scalar string arrays that keep each auth result's selector/scope, domain, and
|
||||
result paired, which the dashboards' alignment-detail tables aggregate on.
|
||||
Reports saved by older versions lack these fields and will not appear in
|
||||
those tables.
|
||||
|
||||
parsedmarc now backfills this automatically. On startup, it runs a cheap
|
||||
count query against each configured aggregate index pattern to check for
|
||||
documents that have DKIM or SPF results but are missing the corresponding
|
||||
combined field. If any are found, it submits the backfill as a background
|
||||
`_update_by_query` task (`wait_for_completion=false`), so startup is never
|
||||
blocked on it; progress is logged, including the task ID. The check itself
|
||||
is idempotent — once an index is fully backfilled, later startups see a
|
||||
count of 0 and log nothing further — and it works the same way on
|
||||
OpenSearch. Any error talking to the cluster (for example, no indexes yet
|
||||
on a fresh install) is logged as a warning and retried on the next startup,
|
||||
rather than aborting parsedmarc.
|
||||
|
||||
If you upgrade the dashboards without pointing the new parsedmarc version
|
||||
at the cluster, or you'd rather control when the write load happens, you
|
||||
can still run the backfill manually. It is idempotent (documents that
|
||||
already have the fields are skipped), so it is safe to re-run. It works
|
||||
identically on OpenSearch; just adjust the URL and credentials. The query
|
||||
matches only documents that have at least one DKIM or SPF auth result and
|
||||
lack the corresponding combined field; documents with no auth results are
|
||||
skipped, because an `exists` query cannot see an empty array, and for
|
||||
search purposes an empty `dkim_results_combined` is identical to an
|
||||
absent one. Each result is matched on either its `domain` or its `result`
|
||||
subfield as defense in depth: an empty string indexes no text tokens and
|
||||
is invisible to `exists`, and the storage shape of every historical
|
||||
parsedmarc version can't be audited, so matching either subfield ensures
|
||||
no backfillable document is skipped.
|
||||
|
||||
```bash
|
||||
curl -X POST "http://localhost:9200/dmarc_aggregate*/_update_by_query?conflicts=proceed&wait_for_completion=false" \
|
||||
-H "Content-Type: application/json" -d '
|
||||
{
|
||||
"query": {
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "dkim_results.domain"}},
|
||||
{"exists": {"field": "dkim_results.result"}}
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "dkim_results_combined"}}]
|
||||
}
|
||||
},
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "spf_results.domain"}},
|
||||
{"exists": {"field": "spf_results.result"}}
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "spf_results_combined"}}]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"script": {
|
||||
"lang": "painless",
|
||||
"source": "List dk = new ArrayList(); def dr = ctx._source.dkim_results; if (dr != null) { if (!(dr instanceof List)) { dr = [dr]; } for (e in dr) { if (e == null) { continue; } def sel = e.selector != null ? e.selector : \"none\"; def dom = e.domain != null ? e.domain : \"none\"; def res = e.result != null ? e.result : \"none\"; dk.add(sel + \" / \" + dom + \" / \" + res); } } ctx._source.dkim_results_combined = dk; List sp = new ArrayList(); def sr = ctx._source.spf_results; if (sr != null) { if (!(sr instanceof List)) { sr = [sr]; } for (e in sr) { if (e == null) { continue; } def sc = e.scope != null ? e.scope : \"mfrom\"; def dom = e.domain != null ? e.domain : \"none\"; def res = e.result != null ? e.result : (e.results != null ? e.results : \"none\"); sp.add(sc + \" / \" + dom + \" / \" + res); } } ctx._source.spf_results_combined = sp;"
|
||||
}
|
||||
}'
|
||||
```
|
||||
|
||||
`wait_for_completion=false` returns a task ID — check progress with
|
||||
`GET _tasks/<task id>`. Adjust the index pattern if you use a custom
|
||||
`index_prefix`/`index_suffix`. After backfilling, re-import the updated
|
||||
dashboards ndjson (the index pattern saved object changed too) per the
|
||||
import instructions above.
|
||||
|
||||
SMTP TLS documents have the same class of defect one level deeper:
|
||||
`policies` is an object array, and each policy's `failure_details` is an
|
||||
object array inside it. SMTP TLS documents now also carry
|
||||
`policies_combined` and `failure_details_combined`, backfilled
|
||||
automatically at startup the same way, and the equivalent manual command
|
||||
is:
|
||||
|
||||
```bash
|
||||
curl -X POST "http://localhost:9200/smtp_tls*/_update_by_query?conflicts=proceed&wait_for_completion=false" \
|
||||
-H "Content-Type: application/json" -d '
|
||||
{
|
||||
"query": {
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "policies.policy_domain"}},
|
||||
{"exists": {"field": "policies.policy_type"}}
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "policies_combined"}}]
|
||||
}
|
||||
},
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "policies.failure_details.result_type"}},
|
||||
{"exists": {"field": "policies.failure_details.sending_mta_ip"}}
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "failure_details_combined"}}]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"script": {
|
||||
"lang": "painless",
|
||||
"source": "List pols = new ArrayList(); List dets = new ArrayList(); def ps = ctx._source.policies; if (ps != null) { if (!(ps instanceof List)) { ps = [ps]; } for (p in ps) { if (p == null) { continue; } def dom = p.policy_domain != null ? p.policy_domain : \"none\"; def typ = p.policy_type != null ? p.policy_type : \"none\"; pols.add(dom + \" / \" + typ); def fds = p.failure_details; if (fds != null) { if (!(fds instanceof List)) { fds = [fds]; } for (f in fds) { if (f == null) { continue; } def rt = f.result_type != null ? f.result_type : \"none\"; def smi = f.sending_mta_ip != null ? f.sending_mta_ip : \"none\"; def ri = f.receiving_ip != null ? f.receiving_ip : \"none\"; def rmh = f.receiving_mx_hostname != null ? f.receiving_mx_hostname : \"none\"; dets.add(dom + \" / \" + typ + \" / \" + rt + \" / \" + smi + \" / \" + ri + \" / \" + rmh); } } } } ctx._source.policies_combined = pols; ctx._source.failure_details_combined = dets;"
|
||||
}
|
||||
}'
|
||||
```
|
||||
|
||||
It works identically on OpenSearch; just adjust the URL and credentials, same
|
||||
as the aggregate command above.
|
||||
|
||||
## Records retention
|
||||
|
||||
Starting in version 5.0.0, `parsedmarc` stores data in a separate
|
||||
|
||||
+29
-1
@@ -82,7 +82,23 @@ Further down the dashboard, you can filter by source country or source IP
|
||||
address.
|
||||
|
||||
Tables showing SPF and DKIM alignment details are located under the IP address
|
||||
table.
|
||||
table. Each row of the DKIM details table is one real DKIM signature, shown
|
||||
as a combined `selector / domain / result` value; the SPF details table
|
||||
shows `scope / domain / result` the same way. Combining the values into one
|
||||
column keeps each signature's selector, domain, and result paired together,
|
||||
rather than aggregating them as separate columns. Because a message that
|
||||
carries multiple DKIM signatures appears once per signature, summing the
|
||||
messages column across rows can exceed the total number of messages.
|
||||
|
||||
The "Auth result filters" panel above the details tables
|
||||
provides dropdowns for the individual auth-result components — DKIM
|
||||
selector, DKIM domain, DKIM result, SPF scope, SPF domain, and SPF
|
||||
result — and filters the whole dashboard by them. Because components from
|
||||
different signatures of the same message are indexed together, combining
|
||||
two of these component filters matches documents where any signature
|
||||
satisfies each condition individually, not necessarily the same signature;
|
||||
the combined `selector / domain / result` (`scope / domain / result`)
|
||||
column remains the per-signature source of truth.
|
||||
|
||||
:::{note}
|
||||
The alignment tables (SPF details, DKIM details) and the per-IP source
|
||||
@@ -112,3 +128,15 @@ reporting organizations, the policy domains they report on, and the
|
||||
specific failure types — certificate expiry, STARTTLS not supported,
|
||||
STS policy fetch errors, validation failures, and similar — together with
|
||||
the sending and receiving MTA addresses involved.
|
||||
|
||||
Like the DKIM and SPF details tables above, the "SMTP TLS domains" and
|
||||
"SMTP TLS failure details" tables show one row per policy and one row per
|
||||
failure detail, respectively, using combined `policy (domain / type)` and
|
||||
`failure detail (domain / type / result / sending mta / receiving ip / mx)`
|
||||
columns so that each policy's or failure detail's fields stay paired
|
||||
together, rather than aggregating them as separate columns. The
|
||||
`successful_sessions` and `failed_sessions` columns are summed per report
|
||||
document, though, not per policy: when a single report carries multiple
|
||||
policies, a row's session sums include the sibling policies from that
|
||||
report as well as its own. Fully attributing session counts to a single
|
||||
policy would require restructuring the stored documents.
|
||||
|
||||
@@ -1472,6 +1472,7 @@ def _init_output_clients(opts):
|
||||
elastic.migrate_indexes(
|
||||
aggregate_indexes=[es_aggregate_index],
|
||||
failure_indexes=[es_failure_index],
|
||||
smtp_tls_indexes=[es_smtp_tls_index],
|
||||
)
|
||||
clients["elasticsearch"] = _ElasticsearchHandle()
|
||||
except Exception as e:
|
||||
@@ -1518,6 +1519,7 @@ def _init_output_clients(opts):
|
||||
opensearch.migrate_indexes(
|
||||
aggregate_indexes=[os_aggregate_index],
|
||||
failure_indexes=[os_failure_index],
|
||||
smtp_tls_indexes=[os_smtp_tls_index],
|
||||
)
|
||||
clients["opensearch"] = _OpenSearchHandle()
|
||||
except Exception as e:
|
||||
|
||||
+357
-15
@@ -41,6 +41,181 @@ _SERVERLESS = False
|
||||
# settings (e.g. ``refresh_interval``) are accepted and pass through.
|
||||
_SERVERLESS_REJECTED_SETTINGS = frozenset({"number_of_shards", "number_of_replicas"})
|
||||
|
||||
# Guard query for the dkim_results_combined/spf_results_combined backfill
|
||||
# (see ``migrate_indexes``). Matches only documents that have at least one
|
||||
# DKIM or SPF auth result and are missing the corresponding combined field.
|
||||
# Empty arrays are invisible to ``exists``, so documents with zero
|
||||
# DKIM/SPF results are correctly skipped (verified against real data;
|
||||
# this also makes the query idempotent — a backfilled document no longer
|
||||
# matches). Each result is matched on an OR of its ``domain``/``result``
|
||||
# subfields as defense in depth: the parsers we audited never store a
|
||||
# result without both, but an empty string indexes no text tokens and is
|
||||
# invisible to ``exists``, and the storage shape of every historical
|
||||
# parsedmarc version can't be audited — matching either subfield costs
|
||||
# nothing and cannot skip a document that has something to backfill.
|
||||
_COMBINED_BACKFILL_QUERY: dict[str, Any] = {
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "dkim_results.domain"}},
|
||||
{"exists": {"field": "dkim_results.result"}},
|
||||
],
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "dkim_results_combined"}}],
|
||||
}
|
||||
},
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "spf_results.domain"}},
|
||||
{"exists": {"field": "spf_results.result"}},
|
||||
],
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "spf_results_combined"}}],
|
||||
}
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
# Painless script that (re)derives dkim_results_combined/spf_results_combined
|
||||
# from dkim_results/spf_results, matching the format written by
|
||||
# save_aggregate_report_to_elasticsearch(): "{selector} / {domain} / {result}"
|
||||
# per DKIM result and "{scope} / {domain} / {result}" per SPF result.
|
||||
_COMBINED_BACKFILL_SCRIPT = (
|
||||
"List dk = new ArrayList(); "
|
||||
"def dr = ctx._source.dkim_results; "
|
||||
"if (dr != null) { "
|
||||
"if (!(dr instanceof List)) { dr = [dr]; } "
|
||||
"for (e in dr) { "
|
||||
"if (e == null) { continue; } "
|
||||
'def sel = e.selector != null ? e.selector : "none"; '
|
||||
'def dom = e.domain != null ? e.domain : "none"; '
|
||||
'def res = e.result != null ? e.result : "none"; '
|
||||
'dk.add(sel + " / " + dom + " / " + res); '
|
||||
"} } "
|
||||
"ctx._source.dkim_results_combined = dk; "
|
||||
"List sp = new ArrayList(); "
|
||||
"def sr = ctx._source.spf_results; "
|
||||
"if (sr != null) { "
|
||||
"if (!(sr instanceof List)) { sr = [sr]; } "
|
||||
"for (e in sr) { "
|
||||
"if (e == null) { continue; } "
|
||||
'def sc = e.scope != null ? e.scope : "mfrom"; '
|
||||
'def dom = e.domain != null ? e.domain : "none"; '
|
||||
'def res = e.result != null ? e.result : (e.results != null ? e.results : "none"); '
|
||||
'sp.add(sc + " / " + dom + " / " + res); '
|
||||
"} } "
|
||||
"ctx._source.spf_results_combined = sp;"
|
||||
)
|
||||
|
||||
# Guard query for the policies_combined/failure_details_combined backfill
|
||||
# (see ``migrate_indexes``). Matches only SMTP TLS documents that have at
|
||||
# least one policy or failure detail and are missing the corresponding
|
||||
# combined field. Empty arrays are invisible to ``exists``, so documents
|
||||
# with zero policies/failure details are correctly skipped (this also
|
||||
# makes the query idempotent — a backfilled document no longer matches).
|
||||
# Each result is matched on an OR of its relevant subfields as defense in
|
||||
# depth: the parsers we audited never store a policy/failure detail
|
||||
# without these fields, but an empty string indexes no text tokens and is
|
||||
# invisible to ``exists``, and the storage shape of every historical
|
||||
# parsedmarc version can't be audited — matching either subfield costs
|
||||
# nothing and cannot skip a document that has something to backfill.
|
||||
_SMTP_TLS_COMBINED_BACKFILL_QUERY: dict[str, Any] = {
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "policies.policy_domain"}},
|
||||
{"exists": {"field": "policies.policy_type"}},
|
||||
],
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "policies_combined"}}],
|
||||
}
|
||||
},
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{
|
||||
"exists": {
|
||||
"field": "policies.failure_details.result_type"
|
||||
}
|
||||
},
|
||||
{
|
||||
"exists": {
|
||||
"field": "policies.failure_details.sending_mta_ip"
|
||||
}
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "failure_details_combined"}}],
|
||||
}
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
# Painless script that (re)derives policies_combined/failure_details_combined
|
||||
# from policies/policies.failure_details, matching the format written by
|
||||
# save_smtp_tls_report_to_elasticsearch(): "{policy_domain} / {policy_type}"
|
||||
# per policy and "{policy_domain} / {policy_type} / {result_type} /
|
||||
# {sending_mta_ip} / {receiving_ip} / {receiving_mx_hostname}" per failure
|
||||
# detail.
|
||||
_SMTP_TLS_COMBINED_BACKFILL_SCRIPT = (
|
||||
"List pols = new ArrayList(); "
|
||||
"List dets = new ArrayList(); "
|
||||
"def ps = ctx._source.policies; "
|
||||
"if (ps != null) { "
|
||||
"if (!(ps instanceof List)) { ps = [ps]; } "
|
||||
"for (p in ps) { "
|
||||
"if (p == null) { continue; } "
|
||||
'def dom = p.policy_domain != null ? p.policy_domain : "none"; '
|
||||
'def typ = p.policy_type != null ? p.policy_type : "none"; '
|
||||
'pols.add(dom + " / " + typ); '
|
||||
"def fds = p.failure_details; "
|
||||
"if (fds != null) { "
|
||||
"if (!(fds instanceof List)) { fds = [fds]; } "
|
||||
"for (f in fds) { "
|
||||
"if (f == null) { continue; } "
|
||||
'def rt = f.result_type != null ? f.result_type : "none"; '
|
||||
'def smi = f.sending_mta_ip != null ? f.sending_mta_ip : "none"; '
|
||||
'def ri = f.receiving_ip != null ? f.receiving_ip : "none"; '
|
||||
'def rmh = f.receiving_mx_hostname != null ? f.receiving_mx_hostname : "none"; '
|
||||
'dets.add(dom + " / " + typ + " / " + rt + " / " + smi + " / " + ri + " / " + rmh); '
|
||||
"} } } } "
|
||||
"ctx._source.policies_combined = pols; "
|
||||
"ctx._source.failure_details_combined = dets;"
|
||||
)
|
||||
|
||||
|
||||
class _PolicyOverride(InnerDoc):
|
||||
# The elasticsearch.dsl 8.x type stubs use dataclass_transform and only
|
||||
@@ -99,7 +274,7 @@ class _SPFResult(InnerDoc):
|
||||
|
||||
domain = Text()
|
||||
scope = Text()
|
||||
results = Text()
|
||||
result = Text()
|
||||
human_result = Text()
|
||||
|
||||
|
||||
@@ -144,8 +319,25 @@ class _AggregateReportDoc(Document):
|
||||
header_from = Text()
|
||||
envelope_from = Text()
|
||||
envelope_to = Text()
|
||||
# Nested(...) on the two auth-result fields below is only the DSL's
|
||||
# in-memory document shape; it is never installed as a mapping.
|
||||
# create_indexes() deliberately skips Index.document() registration so
|
||||
# these fields stay dynamic-mapped as plain `object` in the cluster
|
||||
# (see the comment there and issue #169).
|
||||
dkim_results = Nested(_DKIMResult)
|
||||
spf_results = Nested(_SPFResult)
|
||||
# One "{selector} / {domain} / {result}" (DKIM) or "{scope} / {domain} /
|
||||
# {result}" (SPF) string per auth result. Kibana/Grafana tables cannot
|
||||
# terms-aggregate the subfields of an object array without producing a
|
||||
# cross-product of values (issue #169), so dashboards aggregate these
|
||||
# composed keywords instead. Declared to match what dynamic mapping
|
||||
# produces for a string array (text + .keyword).
|
||||
dkim_results_combined = Text(
|
||||
multi=True, fields={"keyword": Keyword(ignore_above=256)}
|
||||
)
|
||||
spf_results_combined = Text(
|
||||
multi=True, fields={"keyword": Keyword(ignore_above=256)}
|
||||
)
|
||||
np = Keyword()
|
||||
testing = Keyword()
|
||||
discovery_method = Keyword()
|
||||
@@ -158,7 +350,7 @@ class _AggregateReportDoc(Document):
|
||||
self,
|
||||
domain: str,
|
||||
selector: str,
|
||||
result: _DKIMResult,
|
||||
result: str,
|
||||
human_result: str | None = None,
|
||||
):
|
||||
self.dkim_results.append(
|
||||
@@ -169,12 +361,15 @@ class _AggregateReportDoc(Document):
|
||||
human_result=human_result,
|
||||
)
|
||||
)
|
||||
self.dkim_results_combined.append(
|
||||
"{0} / {1} / {2}".format(selector, domain, result)
|
||||
)
|
||||
|
||||
def add_spf_result(
|
||||
self,
|
||||
domain: str,
|
||||
scope: str,
|
||||
result: _SPFResult,
|
||||
result: str,
|
||||
human_result: str | None = None,
|
||||
):
|
||||
self.spf_results.append(
|
||||
@@ -185,6 +380,9 @@ class _AggregateReportDoc(Document):
|
||||
human_result=human_result,
|
||||
)
|
||||
)
|
||||
self.spf_results_combined.append(
|
||||
"{0} / {1} / {2}".format(scope, domain, result)
|
||||
)
|
||||
|
||||
def save(self, **kwargs): # pyright: ignore[reportIncompatibleMethodOverride]
|
||||
self.passed_dmarc = False
|
||||
@@ -301,6 +499,7 @@ class _SMTPTLSFailureDetailsDoc(InnerDoc):
|
||||
result_type = Text()
|
||||
sending_mta_ip = Ip()
|
||||
receiving_mx_helo = Text()
|
||||
receiving_mx_hostname = Text()
|
||||
receiving_ip = Ip()
|
||||
failed_session_count = Integer()
|
||||
additional_information_uri = Text()
|
||||
@@ -342,7 +541,7 @@ class _SMTPTLSPolicyDoc(InnerDoc):
|
||||
receiving_mx_helo=receiving_mx_helo,
|
||||
receiving_ip=receiving_ip,
|
||||
failed_session_count=failed_session_count,
|
||||
additional_information=additional_information_uri,
|
||||
additional_information_uri=additional_information_uri,
|
||||
failure_reason_code=failure_reason_code,
|
||||
)
|
||||
self.failure_details.append(_details)
|
||||
@@ -365,6 +564,19 @@ class _SMTPTLSReportDoc(Document):
|
||||
contact_info = Text()
|
||||
report_id = Text()
|
||||
policies = Nested(_SMTPTLSPolicyDoc)
|
||||
# One "{policy_domain} / {policy_type}" string per policy. Kibana/
|
||||
# Grafana tables cannot terms-aggregate the subfields of an object
|
||||
# array without producing a cross-product of values (issue #169), so
|
||||
# dashboards aggregate these composed keywords instead. Declared to
|
||||
# match what dynamic mapping produces for a string array (text +
|
||||
# .keyword).
|
||||
policies_combined = Text(multi=True, fields={"keyword": Keyword(ignore_above=256)})
|
||||
# One "{policy_domain} / {policy_type} / {result_type} /
|
||||
# {sending_mta_ip} / {receiving_ip} / {receiving_mx_hostname}" string
|
||||
# per failure detail, across all policies.
|
||||
failure_details_combined = Text(
|
||||
multi=True, fields={"keyword": Keyword(ignore_above=256)}
|
||||
)
|
||||
|
||||
|
||||
class AlreadySaved(ValueError):
|
||||
@@ -450,6 +662,18 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
|
||||
for name in names:
|
||||
index = Index(name)
|
||||
try:
|
||||
# Deliberately no Index.document() registration: the shipped
|
||||
# dashboards cannot rebuild their detail tables on a `nested`
|
||||
# mapping — Kibana/OSD visual editors do not support nested
|
||||
# fields, Vega can run nested aggregations but does not render
|
||||
# tables, and Grafana's nested bucket aggregation (9.4+) lacks
|
||||
# reverse_nested for parent-level metrics like message_count —
|
||||
# so the dynamic `object` mapping produced by a bare create is
|
||||
# load-bearing for the shipped dashboards. _AggregateReportDoc
|
||||
# still declares dkim_results/spf_results with Nested(...), but
|
||||
# that is only the DSL's in-memory shape for building documents
|
||||
# — it is never installed as a mapping. See issue #169 and the
|
||||
# *_combined fields on _AggregateReportDoc.
|
||||
if not index.exists():
|
||||
logger.debug("Creating Elasticsearch index: {0}".format(name))
|
||||
if effective_settings:
|
||||
@@ -462,25 +686,118 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
|
||||
def migrate_indexes(
|
||||
aggregate_indexes: list[str] | None = None,
|
||||
failure_indexes: list[str] | None = None,
|
||||
smtp_tls_indexes: list[str] | None = None,
|
||||
):
|
||||
"""
|
||||
Updates index mappings
|
||||
Backfills the ``dkim_results_combined``/``spf_results_combined`` fields
|
||||
(added for issue #169) on aggregate report documents, and the
|
||||
``policies_combined``/``failure_details_combined`` fields on SMTP TLS
|
||||
report documents, that were saved before those fields existed.
|
||||
|
||||
This is a no-op kept for API compatibility (``cli.py`` calls it on
|
||||
startup). The only migration this function ever performed was
|
||||
re-typing ``published_policy.fo`` from ``long`` to ``text``, which
|
||||
applied exclusively to indices still carrying the legacy
|
||||
Elasticsearch 6-era ``"doc"`` mapping type. The 8.x client can only
|
||||
reach servers (Elasticsearch 8.x/9.x) whose indices were created on
|
||||
Elasticsearch 7.x or later and are therefore typeless, so that
|
||||
migration path is unreachable and has been removed.
|
||||
For each name in ``aggregate_indexes``/``smtp_tls_indexes``, this
|
||||
submits an ``update_by_query`` against the ``f"{name}*"`` index pattern
|
||||
(the real indexes are date-suffixed) as a non-blocking background task
|
||||
(``wait_for_completion=False``), so it never delays parsedmarc startup.
|
||||
Submission is guarded by a cheap ``count`` query that only matches
|
||||
documents with DKIM/SPF results (or policies/failure details) but no
|
||||
combined field, so once an index is fully backfilled, later calls are a
|
||||
fast no-op. Any error talking to the cluster (e.g. no indexes yet on a
|
||||
fresh install, or a transient connection issue) is caught and logged as
|
||||
a warning rather than raised; the backfill is simply retried on the
|
||||
next startup, and the manual ``_update_by_query`` commands documented in
|
||||
``docs/source/elasticsearch.md`` remain available in the meantime.
|
||||
|
||||
Args:
|
||||
aggregate_indexes (list): A list of aggregate index names
|
||||
(accepted for API compatibility; unused)
|
||||
failure_indexes (list): A list of failure index names
|
||||
(accepted for API compatibility; unused)
|
||||
smtp_tls_indexes (list): A list of SMTP TLS index names
|
||||
"""
|
||||
if not aggregate_indexes and not smtp_tls_indexes:
|
||||
return
|
||||
|
||||
try:
|
||||
client = connections.get_connection()
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Skipping the dkim_results_combined/spf_results_combined/"
|
||||
"policies_combined/failure_details_combined backfill: could "
|
||||
f"not get an Elasticsearch connection: {e}. This will be "
|
||||
"retried at the next startup."
|
||||
)
|
||||
return
|
||||
for name in aggregate_indexes or []:
|
||||
pattern = f"{name}*"
|
||||
try:
|
||||
count_response = client.count(
|
||||
index=pattern,
|
||||
query=_COMBINED_BACKFILL_QUERY,
|
||||
ignore_unavailable=True,
|
||||
allow_no_indices=True,
|
||||
)
|
||||
count = count_response["count"]
|
||||
if not count:
|
||||
continue
|
||||
update_response = client.update_by_query(
|
||||
index=pattern,
|
||||
query=_COMBINED_BACKFILL_QUERY,
|
||||
script={"source": _COMBINED_BACKFILL_SCRIPT, "lang": "painless"},
|
||||
conflicts="proceed",
|
||||
wait_for_completion=False,
|
||||
ignore_unavailable=True,
|
||||
allow_no_indices=True,
|
||||
)
|
||||
task_id = update_response.get("task")
|
||||
logger.info(
|
||||
"Backfilling dkim_results_combined/spf_results_combined on "
|
||||
f"{count} existing documents in {pattern} (task {task_id})"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Failed to check/submit the dkim_results_combined/"
|
||||
f"spf_results_combined backfill for {pattern}: {e}. This "
|
||||
"will be retried at the next startup; the manual "
|
||||
"_update_by_query command in the documentation remains "
|
||||
"available in the meantime."
|
||||
)
|
||||
|
||||
for name in smtp_tls_indexes or []:
|
||||
pattern = f"{name}*"
|
||||
try:
|
||||
count_response = client.count(
|
||||
index=pattern,
|
||||
query=_SMTP_TLS_COMBINED_BACKFILL_QUERY,
|
||||
ignore_unavailable=True,
|
||||
allow_no_indices=True,
|
||||
)
|
||||
count = count_response["count"]
|
||||
if not count:
|
||||
continue
|
||||
update_response = client.update_by_query(
|
||||
index=pattern,
|
||||
query=_SMTP_TLS_COMBINED_BACKFILL_QUERY,
|
||||
script={
|
||||
"source": _SMTP_TLS_COMBINED_BACKFILL_SCRIPT,
|
||||
"lang": "painless",
|
||||
},
|
||||
conflicts="proceed",
|
||||
wait_for_completion=False,
|
||||
ignore_unavailable=True,
|
||||
allow_no_indices=True,
|
||||
)
|
||||
task_id = update_response.get("task")
|
||||
logger.info(
|
||||
"Backfilling policies_combined/failure_details_combined on "
|
||||
f"{count} existing documents in {pattern} (task {task_id})"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Failed to check/submit the policies_combined/"
|
||||
f"failure_details_combined backfill for {pattern}: {e}. "
|
||||
"This will be retried at the next startup; the manual "
|
||||
"_update_by_query command in the documentation remains "
|
||||
"available in the meantime."
|
||||
)
|
||||
|
||||
|
||||
def save_aggregate_report_to_elasticsearch(
|
||||
@@ -949,6 +1266,16 @@ def save_smtp_tls_report_to_elasticsearch(
|
||||
policy_strings = policy["policy_strings"]
|
||||
if "mx_host_patterns" in policy:
|
||||
mx_host_patterns = policy["mx_host_patterns"]
|
||||
# policies_combined/failure_details_combined: see the field
|
||||
# declarations on _SMTPTLSReportDoc and issue #169. policies and
|
||||
# their failure_details are object arrays with the same
|
||||
# cross-product problem as dkim_results/spf_results, so dashboards
|
||||
# aggregate these composed strings instead of the raw subfields.
|
||||
policy_domain_combined = policy.get("policy_domain") or "none"
|
||||
policy_type_combined = policy.get("policy_type") or "none"
|
||||
smtp_tls_doc.policies_combined.append(
|
||||
"{0} / {1}".format(policy_domain_combined, policy_type_combined)
|
||||
)
|
||||
policy_doc = _SMTPTLSPolicyDoc(
|
||||
policy_domain=policy["policy_domain"],
|
||||
policy_type=policy["policy_type"],
|
||||
@@ -969,7 +1296,12 @@ def save_smtp_tls_report_to_elasticsearch(
|
||||
|
||||
if "receiving_mx_hostname" in failure_detail:
|
||||
receiving_mx_hostname = failure_detail["receiving_mx_hostname"]
|
||||
if "additional_information_uri" in failure_detail:
|
||||
# The parser's key is additional_info_uri (see
|
||||
# SMTPTLSFailureDetailsOptional in types.py); accept the
|
||||
# long-form key too for dicts built by other callers.
|
||||
if "additional_info_uri" in failure_detail:
|
||||
additional_information_uri = failure_detail["additional_info_uri"]
|
||||
elif "additional_information_uri" in failure_detail:
|
||||
additional_information_uri = failure_detail[
|
||||
"additional_information_uri"
|
||||
]
|
||||
@@ -994,6 +1326,16 @@ def save_smtp_tls_report_to_elasticsearch(
|
||||
additional_information_uri=additional_information_uri,
|
||||
failure_reason_code=failure_reason_code,
|
||||
)
|
||||
smtp_tls_doc.failure_details_combined.append(
|
||||
"{0} / {1} / {2} / {3} / {4} / {5}".format(
|
||||
policy_domain_combined,
|
||||
policy_type_combined,
|
||||
failure_detail.get("result_type") or "none",
|
||||
sending_mta_ip or "none",
|
||||
receiving_ip or "none",
|
||||
receiving_mx_hostname or "none",
|
||||
)
|
||||
)
|
||||
smtp_tls_doc.policies.append(policy_doc)
|
||||
|
||||
create_indexes([index], index_settings)
|
||||
|
||||
+412
-34
@@ -34,6 +34,182 @@ class OpenSearchError(Exception):
|
||||
"""Raised when an OpenSearch error occurs"""
|
||||
|
||||
|
||||
# Guard query for the dkim_results_combined/spf_results_combined backfill
|
||||
# (see ``migrate_indexes``). Matches only documents that have at least one
|
||||
# DKIM or SPF auth result and are missing the corresponding combined field.
|
||||
# Empty arrays are invisible to ``exists``, so documents with zero
|
||||
# DKIM/SPF results are correctly skipped (verified against real data;
|
||||
# this also makes the query idempotent — a backfilled document no longer
|
||||
# matches). Each result is matched on an OR of its ``domain``/``result``
|
||||
# subfields as defense in depth: the parsers we audited never store a
|
||||
# result without both, but an empty string indexes no text tokens and is
|
||||
# invisible to ``exists``, and the storage shape of every historical
|
||||
# parsedmarc version can't be audited — matching either subfield costs
|
||||
# nothing and cannot skip a document that has something to backfill.
|
||||
_COMBINED_BACKFILL_QUERY: dict[str, Any] = {
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "dkim_results.domain"}},
|
||||
{"exists": {"field": "dkim_results.result"}},
|
||||
],
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "dkim_results_combined"}}],
|
||||
}
|
||||
},
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "spf_results.domain"}},
|
||||
{"exists": {"field": "spf_results.result"}},
|
||||
],
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "spf_results_combined"}}],
|
||||
}
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
# Painless script that (re)derives dkim_results_combined/spf_results_combined
|
||||
# from dkim_results/spf_results, matching the format written by
|
||||
# save_aggregate_report_to_opensearch(): "{selector} / {domain} / {result}"
|
||||
# per DKIM result and "{scope} / {domain} / {result}" per SPF result.
|
||||
_COMBINED_BACKFILL_SCRIPT = (
|
||||
"List dk = new ArrayList(); "
|
||||
"def dr = ctx._source.dkim_results; "
|
||||
"if (dr != null) { "
|
||||
"if (!(dr instanceof List)) { dr = [dr]; } "
|
||||
"for (e in dr) { "
|
||||
"if (e == null) { continue; } "
|
||||
'def sel = e.selector != null ? e.selector : "none"; '
|
||||
'def dom = e.domain != null ? e.domain : "none"; '
|
||||
'def res = e.result != null ? e.result : "none"; '
|
||||
'dk.add(sel + " / " + dom + " / " + res); '
|
||||
"} } "
|
||||
"ctx._source.dkim_results_combined = dk; "
|
||||
"List sp = new ArrayList(); "
|
||||
"def sr = ctx._source.spf_results; "
|
||||
"if (sr != null) { "
|
||||
"if (!(sr instanceof List)) { sr = [sr]; } "
|
||||
"for (e in sr) { "
|
||||
"if (e == null) { continue; } "
|
||||
'def sc = e.scope != null ? e.scope : "mfrom"; '
|
||||
'def dom = e.domain != null ? e.domain : "none"; '
|
||||
'def res = e.result != null ? e.result : (e.results != null ? e.results : "none"); '
|
||||
'sp.add(sc + " / " + dom + " / " + res); '
|
||||
"} } "
|
||||
"ctx._source.spf_results_combined = sp;"
|
||||
)
|
||||
|
||||
# Guard query for the policies_combined/failure_details_combined backfill
|
||||
# (see ``migrate_indexes``). Matches only SMTP TLS documents that have at
|
||||
# least one policy or failure detail and are missing the corresponding
|
||||
# combined field. Empty arrays are invisible to ``exists``, so documents
|
||||
# with zero policies/failure details are correctly skipped (this also
|
||||
# makes the query idempotent — a backfilled document no longer matches).
|
||||
# Each result is matched on an OR of its relevant subfields as defense in
|
||||
# depth: the parsers we audited never store a policy/failure detail
|
||||
# without these fields, but an empty string indexes no text tokens and is
|
||||
# invisible to ``exists``, and the storage shape of every historical
|
||||
# parsedmarc version can't be audited — matching either subfield costs
|
||||
# nothing and cannot skip a document that has something to backfill.
|
||||
_SMTP_TLS_COMBINED_BACKFILL_QUERY: dict[str, Any] = {
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{"exists": {"field": "policies.policy_domain"}},
|
||||
{"exists": {"field": "policies.policy_type"}},
|
||||
],
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "policies_combined"}}],
|
||||
}
|
||||
},
|
||||
{
|
||||
"bool": {
|
||||
"must": [
|
||||
{
|
||||
"bool": {
|
||||
"minimum_should_match": 1,
|
||||
"should": [
|
||||
{
|
||||
"exists": {
|
||||
"field": "policies.failure_details.result_type"
|
||||
}
|
||||
},
|
||||
{
|
||||
"exists": {
|
||||
"field": "policies.failure_details.sending_mta_ip"
|
||||
}
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
],
|
||||
"must_not": [{"exists": {"field": "failure_details_combined"}}],
|
||||
}
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
# Painless script that (re)derives policies_combined/failure_details_combined
|
||||
# from policies/policies.failure_details, matching the format written by
|
||||
# save_smtp_tls_report_to_opensearch(): "{policy_domain} / {policy_type}"
|
||||
# per policy and "{policy_domain} / {policy_type} / {result_type} /
|
||||
# {sending_mta_ip} / {receiving_ip} / {receiving_mx_hostname}" per failure
|
||||
# detail.
|
||||
_SMTP_TLS_COMBINED_BACKFILL_SCRIPT = (
|
||||
"List pols = new ArrayList(); "
|
||||
"List dets = new ArrayList(); "
|
||||
"def ps = ctx._source.policies; "
|
||||
"if (ps != null) { "
|
||||
"if (!(ps instanceof List)) { ps = [ps]; } "
|
||||
"for (p in ps) { "
|
||||
"if (p == null) { continue; } "
|
||||
'def dom = p.policy_domain != null ? p.policy_domain : "none"; '
|
||||
'def typ = p.policy_type != null ? p.policy_type : "none"; '
|
||||
'pols.add(dom + " / " + typ); '
|
||||
"def fds = p.failure_details; "
|
||||
"if (fds != null) { "
|
||||
"if (!(fds instanceof List)) { fds = [fds]; } "
|
||||
"for (f in fds) { "
|
||||
"if (f == null) { continue; } "
|
||||
'def rt = f.result_type != null ? f.result_type : "none"; '
|
||||
'def smi = f.sending_mta_ip != null ? f.sending_mta_ip : "none"; '
|
||||
'def ri = f.receiving_ip != null ? f.receiving_ip : "none"; '
|
||||
'def rmh = f.receiving_mx_hostname != null ? f.receiving_mx_hostname : "none"; '
|
||||
'dets.add(dom + " / " + typ + " / " + rt + " / " + smi + " / " + ri + " / " + rmh); '
|
||||
"} } } } "
|
||||
"ctx._source.policies_combined = pols; "
|
||||
"ctx._source.failure_details_combined = dets;"
|
||||
)
|
||||
|
||||
|
||||
class _PolicyOverride(InnerDoc):
|
||||
type = Text()
|
||||
comment = Text()
|
||||
@@ -62,7 +238,7 @@ class _DKIMResult(InnerDoc):
|
||||
class _SPFResult(InnerDoc):
|
||||
domain = Text()
|
||||
scope = Text()
|
||||
results = Text()
|
||||
result = Text()
|
||||
human_result = Text()
|
||||
|
||||
|
||||
@@ -101,8 +277,25 @@ class _AggregateReportDoc(Document):
|
||||
header_from = Text()
|
||||
envelope_from = Text()
|
||||
envelope_to = Text()
|
||||
# Nested(...) on the two auth-result fields below is only the DSL's
|
||||
# in-memory document shape; it is never installed as a mapping.
|
||||
# create_indexes() deliberately skips Index.document() registration so
|
||||
# these fields stay dynamic-mapped as plain `object` in the cluster
|
||||
# (see the comment there and issue #169).
|
||||
dkim_results = Nested(_DKIMResult)
|
||||
spf_results = Nested(_SPFResult)
|
||||
# One "{selector} / {domain} / {result}" (DKIM) or "{scope} / {domain} /
|
||||
# {result}" (SPF) string per auth result. Kibana/Grafana tables cannot
|
||||
# terms-aggregate the subfields of an object array without producing a
|
||||
# cross-product of values (issue #169), so dashboards aggregate these
|
||||
# composed keywords instead. Declared to match what dynamic mapping
|
||||
# produces for a string array (text + .keyword).
|
||||
dkim_results_combined = Text(
|
||||
multi=True, fields={"keyword": Keyword(ignore_above=256)}
|
||||
)
|
||||
spf_results_combined = Text(
|
||||
multi=True, fields={"keyword": Keyword(ignore_above=256)}
|
||||
)
|
||||
np = Keyword()
|
||||
testing = Keyword()
|
||||
discovery_method = Keyword()
|
||||
@@ -115,7 +308,7 @@ class _AggregateReportDoc(Document):
|
||||
self,
|
||||
domain: str,
|
||||
selector: str,
|
||||
result: _DKIMResult,
|
||||
result: str,
|
||||
human_result: str | None = None,
|
||||
):
|
||||
self.dkim_results.append(
|
||||
@@ -126,12 +319,15 @@ class _AggregateReportDoc(Document):
|
||||
human_result=human_result,
|
||||
)
|
||||
)
|
||||
self.dkim_results_combined.append(
|
||||
"{0} / {1} / {2}".format(selector, domain, result)
|
||||
)
|
||||
|
||||
def add_spf_result(
|
||||
self,
|
||||
domain: str,
|
||||
scope: str,
|
||||
result: _SPFResult,
|
||||
result: str,
|
||||
human_result: str | None = None,
|
||||
):
|
||||
self.spf_results.append(
|
||||
@@ -142,6 +338,9 @@ class _AggregateReportDoc(Document):
|
||||
human_result=human_result,
|
||||
)
|
||||
)
|
||||
self.spf_results_combined.append(
|
||||
"{0} / {1} / {2}".format(scope, domain, result)
|
||||
)
|
||||
|
||||
def save(self, **kwargs): # pyright: ignore[reportIncompatibleMethodOverride]
|
||||
self.passed_dmarc = False
|
||||
@@ -228,6 +427,7 @@ class _SMTPTLSFailureDetailsDoc(InnerDoc):
|
||||
result_type = Text()
|
||||
sending_mta_ip = Ip()
|
||||
receiving_mx_helo = Text()
|
||||
receiving_mx_hostname = Text()
|
||||
receiving_ip = Ip()
|
||||
failed_session_count = Integer()
|
||||
additional_information_uri = Text()
|
||||
@@ -263,7 +463,7 @@ class _SMTPTLSPolicyDoc(InnerDoc):
|
||||
receiving_mx_helo=receiving_mx_helo,
|
||||
receiving_ip=receiving_ip,
|
||||
failed_session_count=failed_session_count,
|
||||
additional_information=additional_information_uri,
|
||||
additional_information_uri=additional_information_uri,
|
||||
failure_reason_code=failure_reason_code,
|
||||
)
|
||||
self.failure_details.append(_details)
|
||||
@@ -280,6 +480,19 @@ class _SMTPTLSReportDoc(Document):
|
||||
contact_info = Text()
|
||||
report_id = Text()
|
||||
policies = Nested(_SMTPTLSPolicyDoc)
|
||||
# One "{policy_domain} / {policy_type}" string per policy. Kibana/
|
||||
# Grafana tables cannot terms-aggregate the subfields of an object
|
||||
# array without producing a cross-product of values (issue #169), so
|
||||
# dashboards aggregate these composed keywords instead. Declared to
|
||||
# match what dynamic mapping produces for a string array (text +
|
||||
# .keyword).
|
||||
policies_combined = Text(multi=True, fields={"keyword": Keyword(ignore_above=256)})
|
||||
# One "{policy_domain} / {policy_type} / {result_type} /
|
||||
# {sending_mta_ip} / {receiving_ip} / {receiving_mx_hostname}" string
|
||||
# per failure detail, across all policies.
|
||||
failure_details_combined = Text(
|
||||
multi=True, fields={"keyword": Keyword(ignore_above=256)}
|
||||
)
|
||||
|
||||
|
||||
class AlreadySaved(ValueError):
|
||||
@@ -367,6 +580,18 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
|
||||
for name in names:
|
||||
index = Index(name)
|
||||
try:
|
||||
# Deliberately no Index.document() registration: the shipped
|
||||
# dashboards cannot rebuild their detail tables on a `nested`
|
||||
# mapping — Kibana/OSD visual editors do not support nested
|
||||
# fields, Vega can run nested aggregations but does not render
|
||||
# tables, and Grafana's nested bucket aggregation (9.4+) lacks
|
||||
# reverse_nested for parent-level metrics like message_count —
|
||||
# so the dynamic `object` mapping produced by a bare create is
|
||||
# load-bearing for the shipped dashboards. _AggregateReportDoc
|
||||
# still declares dkim_results/spf_results with Nested(...), but
|
||||
# that is only the DSL's in-memory shape for building documents
|
||||
# — it is never installed as a mapping. See issue #169 and the
|
||||
# *_combined fields on _AggregateReportDoc.
|
||||
if not index.exists():
|
||||
logger.debug("Creating OpenSearch index: {0}".format(name))
|
||||
if settings is None:
|
||||
@@ -381,47 +606,175 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
|
||||
def migrate_indexes(
|
||||
aggregate_indexes: list[str] | None = None,
|
||||
failure_indexes: list[str] | None = None,
|
||||
smtp_tls_indexes: list[str] | None = None,
|
||||
):
|
||||
"""
|
||||
Updates index mappings
|
||||
Runs index migrations and backfills.
|
||||
|
||||
First, the legacy ``published_policy.fo`` migration: indexes where that
|
||||
field was mapped as ``long`` (data indexed by very old parsedmarc
|
||||
releases under the Elasticsearch 6-era ``doc`` mapping type) are rebuilt
|
||||
as a ``-v2`` index with the text/keyword shape.
|
||||
|
||||
Second, the ``dkim_results_combined``/``spf_results_combined`` backfill
|
||||
(added for issue #169) for aggregate report documents that were saved
|
||||
before those fields existed. For each name in ``aggregate_indexes``,
|
||||
this submits an ``update_by_query`` against the ``f"{name}*"`` index
|
||||
pattern (the real indexes are date-suffixed) as a non-blocking
|
||||
background task (``wait_for_completion=False``), so it never delays
|
||||
parsedmarc startup. Submission is guarded by a cheap ``count`` query
|
||||
that only matches documents with DKIM/SPF results but no combined
|
||||
field, so once an index is fully backfilled, later calls are a fast
|
||||
no-op. Any error talking to the cluster (e.g. no indexes yet on a
|
||||
fresh install, or a transient connection issue) is caught and logged
|
||||
as a warning rather than raised; the backfill is simply retried on the
|
||||
next startup, and the manual ``_update_by_query`` command documented
|
||||
in ``docs/source/elasticsearch.md`` remains available in the meantime.
|
||||
|
||||
Third, the same treatment for the ``policies_combined``/
|
||||
``failure_details_combined`` fields (same issue #169) on SMTP TLS
|
||||
report documents, for each name in ``smtp_tls_indexes``.
|
||||
|
||||
Args:
|
||||
aggregate_indexes (list): A list of aggregate index names
|
||||
failure_indexes (list): A list of failure index names
|
||||
(accepted for API compatibility; no migrations are
|
||||
currently needed for failure indexes)
|
||||
smtp_tls_indexes (list): A list of SMTP TLS index names
|
||||
"""
|
||||
version = 2
|
||||
if aggregate_indexes is None:
|
||||
aggregate_indexes = []
|
||||
for aggregate_index_name in aggregate_indexes:
|
||||
if not Index(aggregate_index_name).exists():
|
||||
continue
|
||||
aggregate_index = Index(aggregate_index_name)
|
||||
doc = "doc"
|
||||
fo_field = "published_policy.fo"
|
||||
fo = "fo"
|
||||
fo_mapping = aggregate_index.get_field_mapping(fields=[fo_field])
|
||||
fo_mapping = fo_mapping[list(fo_mapping.keys())[0]]["mappings"]
|
||||
if doc not in fo_mapping:
|
||||
continue
|
||||
if not aggregate_indexes and not smtp_tls_indexes:
|
||||
return
|
||||
|
||||
fo_mapping = fo_mapping[doc][fo_field]["mapping"][fo]
|
||||
fo_type = fo_mapping["type"]
|
||||
if fo_type == "long":
|
||||
new_index_name = "{0}-v{1}".format(aggregate_index_name, version)
|
||||
body = {
|
||||
"properties": {
|
||||
"published_policy.fo": {
|
||||
"type": "text",
|
||||
"fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
|
||||
version = 2
|
||||
for aggregate_index_name in aggregate_indexes or []:
|
||||
try:
|
||||
if not Index(aggregate_index_name).exists():
|
||||
continue
|
||||
aggregate_index = Index(aggregate_index_name)
|
||||
doc = "doc"
|
||||
fo_field = "published_policy.fo"
|
||||
fo = "fo"
|
||||
fo_mapping = aggregate_index.get_field_mapping(fields=[fo_field])
|
||||
fo_mapping = fo_mapping[list(fo_mapping.keys())[0]]["mappings"]
|
||||
if doc not in fo_mapping:
|
||||
continue
|
||||
|
||||
fo_mapping = fo_mapping[doc][fo_field]["mapping"][fo]
|
||||
fo_type = fo_mapping["type"]
|
||||
if fo_type == "long":
|
||||
new_index_name = "{0}-v{1}".format(aggregate_index_name, version)
|
||||
body = {
|
||||
"properties": {
|
||||
"published_policy.fo": {
|
||||
"type": "text",
|
||||
"fields": {
|
||||
"keyword": {"type": "keyword", "ignore_above": 256}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Index(new_index_name).create()
|
||||
Index(new_index_name).put_mapping(doc_type=doc, body=body)
|
||||
reindex(connections.get_connection(), aggregate_index_name, new_index_name)
|
||||
Index(aggregate_index_name).delete()
|
||||
Index(new_index_name).create()
|
||||
Index(new_index_name).put_mapping(doc_type=doc, body=body)
|
||||
reindex(
|
||||
connections.get_connection(), aggregate_index_name, new_index_name
|
||||
)
|
||||
Index(aggregate_index_name).delete()
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Failed the legacy published_policy.fo migration for "
|
||||
f"{aggregate_index_name}: {e}. This will be retried at the "
|
||||
"next startup."
|
||||
)
|
||||
|
||||
try:
|
||||
client = connections.get_connection()
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Skipping the dkim_results_combined/spf_results_combined/"
|
||||
"policies_combined/failure_details_combined backfill: could "
|
||||
f"not get an OpenSearch connection: {e}. This will be retried "
|
||||
"at the next startup."
|
||||
)
|
||||
return
|
||||
for name in aggregate_indexes or []:
|
||||
pattern = f"{name}*"
|
||||
try:
|
||||
count_response = client.count(
|
||||
index=pattern,
|
||||
body={"query": _COMBINED_BACKFILL_QUERY},
|
||||
ignore_unavailable=True,
|
||||
allow_no_indices=True,
|
||||
)
|
||||
count = count_response["count"]
|
||||
if not count:
|
||||
continue
|
||||
update_response = client.update_by_query(
|
||||
index=pattern,
|
||||
body={
|
||||
"query": _COMBINED_BACKFILL_QUERY,
|
||||
"script": {
|
||||
"source": _COMBINED_BACKFILL_SCRIPT,
|
||||
"lang": "painless",
|
||||
},
|
||||
},
|
||||
conflicts="proceed",
|
||||
wait_for_completion=False,
|
||||
ignore_unavailable=True,
|
||||
allow_no_indices=True,
|
||||
)
|
||||
task_id = update_response.get("task")
|
||||
logger.info(
|
||||
"Backfilling dkim_results_combined/spf_results_combined on "
|
||||
f"{count} existing documents in {pattern} (task {task_id})"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Failed to check/submit the dkim_results_combined/"
|
||||
f"spf_results_combined backfill for {pattern}: {e}. This "
|
||||
"will be retried at the next startup; the manual "
|
||||
"_update_by_query command in the documentation remains "
|
||||
"available in the meantime."
|
||||
)
|
||||
|
||||
for name in smtp_tls_indexes or []:
|
||||
pattern = f"{name}*"
|
||||
try:
|
||||
count_response = client.count(
|
||||
index=pattern,
|
||||
body={"query": _SMTP_TLS_COMBINED_BACKFILL_QUERY},
|
||||
ignore_unavailable=True,
|
||||
allow_no_indices=True,
|
||||
)
|
||||
count = count_response["count"]
|
||||
if not count:
|
||||
continue
|
||||
update_response = client.update_by_query(
|
||||
index=pattern,
|
||||
body={
|
||||
"query": _SMTP_TLS_COMBINED_BACKFILL_QUERY,
|
||||
"script": {
|
||||
"source": _SMTP_TLS_COMBINED_BACKFILL_SCRIPT,
|
||||
"lang": "painless",
|
||||
},
|
||||
},
|
||||
conflicts="proceed",
|
||||
wait_for_completion=False,
|
||||
ignore_unavailable=True,
|
||||
allow_no_indices=True,
|
||||
)
|
||||
task_id = update_response.get("task")
|
||||
logger.info(
|
||||
"Backfilling policies_combined/failure_details_combined on "
|
||||
f"{count} existing documents in {pattern} (task {task_id})"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"Failed to check/submit the policies_combined/"
|
||||
f"failure_details_combined backfill for {pattern}: {e}. "
|
||||
"This will be retried at the next startup; the manual "
|
||||
"_update_by_query command in the documentation remains "
|
||||
"available in the meantime."
|
||||
)
|
||||
|
||||
|
||||
def save_aggregate_report_to_opensearch(
|
||||
@@ -885,6 +1238,16 @@ def save_smtp_tls_report_to_opensearch(
|
||||
policy_strings = policy["policy_strings"]
|
||||
if "mx_host_patterns" in policy:
|
||||
mx_host_patterns = policy["mx_host_patterns"]
|
||||
# policies_combined/failure_details_combined: see the field
|
||||
# declarations on _SMTPTLSReportDoc and issue #169. policies and
|
||||
# their failure_details are object arrays with the same
|
||||
# cross-product problem as dkim_results/spf_results, so dashboards
|
||||
# aggregate these composed strings instead of the raw subfields.
|
||||
policy_domain_combined = policy.get("policy_domain") or "none"
|
||||
policy_type_combined = policy.get("policy_type") or "none"
|
||||
smtp_tls_doc.policies_combined.append(
|
||||
"{0} / {1}".format(policy_domain_combined, policy_type_combined)
|
||||
)
|
||||
policy_doc = _SMTPTLSPolicyDoc(
|
||||
policy_domain=policy["policy_domain"],
|
||||
policy_type=policy["policy_type"],
|
||||
@@ -905,7 +1268,12 @@ def save_smtp_tls_report_to_opensearch(
|
||||
|
||||
if "receiving_mx_hostname" in failure_detail:
|
||||
receiving_mx_hostname = failure_detail["receiving_mx_hostname"]
|
||||
if "additional_information_uri" in failure_detail:
|
||||
# The parser's key is additional_info_uri (see
|
||||
# SMTPTLSFailureDetailsOptional in types.py); accept the
|
||||
# long-form key too for dicts built by other callers.
|
||||
if "additional_info_uri" in failure_detail:
|
||||
additional_information_uri = failure_detail["additional_info_uri"]
|
||||
elif "additional_information_uri" in failure_detail:
|
||||
additional_information_uri = failure_detail[
|
||||
"additional_information_uri"
|
||||
]
|
||||
@@ -930,6 +1298,16 @@ def save_smtp_tls_report_to_opensearch(
|
||||
additional_information_uri=additional_information_uri,
|
||||
failure_reason_code=failure_reason_code,
|
||||
)
|
||||
smtp_tls_doc.failure_details_combined.append(
|
||||
"{0} / {1} / {2} / {3} / {4} / {5}".format(
|
||||
policy_domain_combined,
|
||||
policy_type_combined,
|
||||
failure_detail.get("result_type") or "none",
|
||||
sending_mta_ip or "none",
|
||||
receiving_ip or "none",
|
||||
receiving_mx_hostname or "none",
|
||||
)
|
||||
)
|
||||
smtp_tls_doc.policies.append(policy_doc)
|
||||
|
||||
create_indexes([index], index_settings)
|
||||
|
||||
@@ -16,6 +16,7 @@ from parsedmarc.elastic import (
|
||||
AlreadySaved,
|
||||
ElasticsearchError,
|
||||
create_indexes,
|
||||
migrate_indexes,
|
||||
save_aggregate_report_to_elasticsearch,
|
||||
save_failure_report_to_elasticsearch,
|
||||
save_smtp_tls_report_to_elasticsearch,
|
||||
@@ -395,6 +396,143 @@ class TestCreateIndexesServerless(unittest.TestCase):
|
||||
mock_index.create.assert_called_once()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# migrate_indexes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMigrateIndexes(unittest.TestCase):
|
||||
"""migrate_indexes backfills dkim_results_combined/spf_results_combined
|
||||
(issue #169) on pre-existing aggregate documents as a non-blocking
|
||||
background task. It is guarded by a cheap count() query so repeated
|
||||
startups against an already-backfilled index are a no-op, and any SDK
|
||||
error is caught and logged rather than raised, so it never blocks
|
||||
parsedmarc startup."""
|
||||
|
||||
def test_backfill_submitted_when_old_docs_exist(self):
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.return_value = {"count": 42}
|
||||
mock_get_conn.return_value = mock_client
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
mock_client.update_by_query.assert_called_once()
|
||||
kwargs = mock_client.update_by_query.call_args.kwargs
|
||||
self.assertEqual(kwargs["index"], "dmarc_aggregate*")
|
||||
self.assertEqual(kwargs["conflicts"], "proceed")
|
||||
self.assertFalse(kwargs["wait_for_completion"])
|
||||
self.assertEqual(kwargs["query"], elastic_module._COMBINED_BACKFILL_QUERY)
|
||||
script_source = kwargs["script"]["source"]
|
||||
self.assertIn("ctx._source.dkim_results_combined", script_source)
|
||||
self.assertIn("ctx._source.spf_results_combined", script_source)
|
||||
|
||||
# The count() guard query also targets the date-suffixed pattern.
|
||||
count_kwargs = mock_client.count.call_args.kwargs
|
||||
self.assertEqual(count_kwargs["index"], "dmarc_aggregate*")
|
||||
self.assertEqual(count_kwargs["query"], elastic_module._COMBINED_BACKFILL_QUERY)
|
||||
|
||||
def test_backfill_skipped_when_no_old_docs(self):
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.return_value = {"count": 0}
|
||||
mock_get_conn.return_value = mock_client
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
mock_client.update_by_query.assert_not_called()
|
||||
|
||||
def test_backfill_skipped_when_no_aggregate_indexes(self):
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
migrate_indexes()
|
||||
migrate_indexes(aggregate_indexes=None)
|
||||
|
||||
mock_get_conn.assert_not_called()
|
||||
|
||||
def test_backfill_failure_does_not_raise(self):
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.side_effect = RuntimeError("cluster unreachable")
|
||||
mock_get_conn.return_value = mock_client
|
||||
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
self.assertTrue(any("cluster unreachable" in msg for msg in cm.output))
|
||||
mock_client.update_by_query.assert_not_called()
|
||||
|
||||
def test_get_connection_failure_does_not_raise(self):
|
||||
"""connections.get_connection() itself sits outside the per-index
|
||||
try/except; if it raises (e.g. no Elasticsearch connection has been
|
||||
configured yet), migrate_indexes must still not propagate the
|
||||
exception, per its docstring's promise that any cluster error is
|
||||
caught and logged."""
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
mock_get_conn.side_effect = RuntimeError("no connection")
|
||||
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
self.assertTrue(
|
||||
any("Skipping the dkim_results_combined" in msg for msg in cm.output)
|
||||
)
|
||||
self.assertTrue(any("no connection" in msg for msg in cm.output))
|
||||
|
||||
def test_smtp_tls_backfill_submitted_when_old_docs_exist(self):
|
||||
"""SMTP TLS analogue of test_backfill_submitted_when_old_docs_exist:
|
||||
policies_combined/failure_details_combined backfill (also issue
|
||||
#169) is submitted with its own guard query and painless script."""
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.return_value = {"count": 7}
|
||||
mock_get_conn.return_value = mock_client
|
||||
migrate_indexes(smtp_tls_indexes=["smtp_tls"])
|
||||
|
||||
mock_client.update_by_query.assert_called_once()
|
||||
kwargs = mock_client.update_by_query.call_args.kwargs
|
||||
self.assertEqual(kwargs["index"], "smtp_tls*")
|
||||
self.assertEqual(kwargs["conflicts"], "proceed")
|
||||
self.assertFalse(kwargs["wait_for_completion"])
|
||||
self.assertEqual(
|
||||
kwargs["query"], elastic_module._SMTP_TLS_COMBINED_BACKFILL_QUERY
|
||||
)
|
||||
script_source = kwargs["script"]["source"]
|
||||
self.assertIn("ctx._source.policies_combined", script_source)
|
||||
self.assertIn("ctx._source.failure_details_combined", script_source)
|
||||
|
||||
count_kwargs = mock_client.count.call_args.kwargs
|
||||
self.assertEqual(count_kwargs["index"], "smtp_tls*")
|
||||
self.assertEqual(
|
||||
count_kwargs["query"], elastic_module._SMTP_TLS_COMBINED_BACKFILL_QUERY
|
||||
)
|
||||
|
||||
def test_smtp_tls_backfill_skipped_when_no_old_docs(self):
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.return_value = {"count": 0}
|
||||
mock_get_conn.return_value = mock_client
|
||||
migrate_indexes(smtp_tls_indexes=["smtp_tls"])
|
||||
|
||||
mock_client.update_by_query.assert_not_called()
|
||||
|
||||
def test_smtp_tls_backfill_skipped_when_no_smtp_tls_indexes_or_aggregate(self):
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
migrate_indexes()
|
||||
migrate_indexes(smtp_tls_indexes=None)
|
||||
|
||||
mock_get_conn.assert_not_called()
|
||||
|
||||
def test_smtp_tls_backfill_failure_does_not_raise(self):
|
||||
"""SMTP TLS analogue of test_backfill_failure_does_not_raise: an
|
||||
error from the cluster during the smtp_tls_indexes loop is caught
|
||||
and logged rather than raised."""
|
||||
with patch("parsedmarc.elastic.connections.get_connection") as mock_get_conn:
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.side_effect = RuntimeError("cluster unreachable")
|
||||
mock_get_conn.return_value = mock_client
|
||||
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
|
||||
migrate_indexes(smtp_tls_indexes=["smtp_tls"])
|
||||
|
||||
self.assertTrue(any("cluster unreachable" in msg for msg in cm.output))
|
||||
mock_client.update_by_query.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# save_aggregate_report_to_elasticsearch
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -549,6 +687,54 @@ class TestSaveAggregateReport(unittest.TestCase):
|
||||
mock_doc_cls.call_args.kwargs["date_begin"].timestamp(), 1705276800
|
||||
)
|
||||
|
||||
def test_save_populates_combined_dkim_and_spf_fields(self):
|
||||
"""Regression guard for issue #169: two DKIM signatures on one
|
||||
record must yield exactly two combined entries, not a 4-way
|
||||
cross-product. autospec=True is required on the save patch so
|
||||
mock_save.call_args captures the doc instance as ``self``."""
|
||||
report = _aggregate_report()
|
||||
report["records"][0]["auth_results"] = {
|
||||
"dkim": [
|
||||
{
|
||||
"domain": "example.net",
|
||||
"selector": "net1",
|
||||
"result": "fail",
|
||||
"human_result": None,
|
||||
},
|
||||
{
|
||||
"domain": "example.org",
|
||||
"selector": "org1",
|
||||
"result": "pass",
|
||||
"human_result": None,
|
||||
},
|
||||
],
|
||||
"spf": [
|
||||
{
|
||||
"domain": "example.org",
|
||||
"scope": "mfrom",
|
||||
"result": "pass",
|
||||
"human_result": None,
|
||||
},
|
||||
],
|
||||
}
|
||||
with (
|
||||
patch("parsedmarc.elastic.Search", return_value=_empty_search()),
|
||||
patch(
|
||||
"parsedmarc.elastic.Index",
|
||||
return_value=MagicMock(exists=MagicMock(return_value=True)),
|
||||
),
|
||||
patch.object(
|
||||
elastic_module._AggregateReportDoc, "save", autospec=True
|
||||
) as mock_save,
|
||||
):
|
||||
save_aggregate_report_to_elasticsearch(report)
|
||||
doc = mock_save.call_args[0][0]
|
||||
self.assertEqual(
|
||||
list(doc.dkim_results_combined),
|
||||
["net1 / example.net / fail", "org1 / example.org / pass"],
|
||||
)
|
||||
self.assertEqual(list(doc.spf_results_combined), ["mfrom / example.org / pass"])
|
||||
|
||||
|
||||
class TestAggregateDocPassedDmarc(unittest.TestCase):
|
||||
"""The _AggregateReportDoc.save() override derives passed_dmarc — the
|
||||
@@ -576,6 +762,57 @@ class TestAggregateDocPassedDmarc(unittest.TestCase):
|
||||
self.assertEqual(bool(doc.passed_dmarc), expected)
|
||||
|
||||
|
||||
class TestAggregateDocCombinedResults(unittest.TestCase):
|
||||
"""add_dkim_result/add_spf_result never touch the network, so these
|
||||
construct _AggregateReportDoc directly rather than going through the
|
||||
save_* entry point."""
|
||||
|
||||
def test_add_dkim_result_appends_combined_string(self):
|
||||
"""Regression guard for issue #169: dkim_results/spf_results are
|
||||
arrays of objects that the engine dynamic-maps as plain ``object``
|
||||
(not ``nested``) and flattens, so Kibana/Grafana tables cannot
|
||||
terms-aggregate their subfields without producing a cross-product
|
||||
of selector/domain/result values. The composed
|
||||
"selector / domain / result" string preserves the per-signature
|
||||
pairing that the flattened array loses."""
|
||||
doc = elastic_module._AggregateReportDoc()
|
||||
doc.add_dkim_result(
|
||||
domain="example.net", selector="net1", result="fail", human_result=None
|
||||
)
|
||||
doc.add_dkim_result(
|
||||
domain="example.org", selector="org1", result="pass", human_result=None
|
||||
)
|
||||
expected = ["net1 / example.net / fail", "org1 / example.org / pass"]
|
||||
# dkim_results_combined is declared as Text(multi=True, ...); the SDK
|
||||
# stub types the class attribute as Text (no Iterable protocol),
|
||||
# even though the runtime value is an AttrList once multi=True is
|
||||
# set. Same category of stub gap as the Q()/meta.index ignores in
|
||||
# elastic.py.
|
||||
self.assertEqual(list(doc.dkim_results_combined), expected) # pyright: ignore[reportArgumentType]
|
||||
self.assertEqual(doc.to_dict()["dkim_results_combined"], expected)
|
||||
|
||||
def test_add_spf_result_appends_combined_string(self):
|
||||
doc = elastic_module._AggregateReportDoc()
|
||||
doc.add_spf_result(
|
||||
domain="example.org", scope="mfrom", result="pass", human_result=None
|
||||
)
|
||||
expected = ["mfrom / example.org / pass"]
|
||||
self.assertEqual(list(doc.spf_results_combined), expected) # pyright: ignore[reportArgumentType]
|
||||
self.assertEqual(doc.to_dict()["spf_results_combined"], expected)
|
||||
|
||||
def test_spf_result_serializes_under_singular_result_key(self):
|
||||
"""The _SPFResult class previously declared a dead ``results``
|
||||
(plural) field while the save path wrote ``result``; verify the
|
||||
serialized inner doc actually uses the singular key."""
|
||||
doc = elastic_module._AggregateReportDoc()
|
||||
doc.add_spf_result(
|
||||
domain="example.org", scope="mfrom", result="pass", human_result=None
|
||||
)
|
||||
d = doc.to_dict()["spf_results"][0]
|
||||
self.assertEqual(d["result"], "pass")
|
||||
self.assertNotIn("results", d)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# save_failure_report_to_elasticsearch
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -930,6 +1167,82 @@ class TestSaveSmtpTlsReport(unittest.TestCase):
|
||||
save_smtp_tls_report_to_elasticsearch(report)
|
||||
mock_save.assert_called_once()
|
||||
|
||||
def test_save_populates_combined_policy_and_failure_detail_fields(self):
|
||||
"""Regression guard for the SMTP TLS analogue of issue #169:
|
||||
policies and their failure_details are object arrays, so stacked
|
||||
terms aggregations on their subfields cross-product just like
|
||||
dkim_results/spf_results did. Two policies (one with two failure
|
||||
details, one with none) must yield exactly two policies_combined
|
||||
entries and two failure_details_combined entries, not a
|
||||
cross-product. autospec=True is required on the save patch so
|
||||
mock_save.call_args captures the doc instance as ``self``."""
|
||||
report = _smtp_tls_report(
|
||||
policies=[
|
||||
{
|
||||
"policy_domain": "example.com",
|
||||
"policy_type": "sts",
|
||||
"successful_session_count": 100,
|
||||
"failed_session_count": 2,
|
||||
"failure_details": [
|
||||
{
|
||||
"result_type": "certificate-expired",
|
||||
"failed_session_count": 1,
|
||||
"sending_mta_ip": "192.0.2.1",
|
||||
"receiving_ip": "203.0.113.1",
|
||||
"receiving_mx_hostname": "mx1.example.com",
|
||||
"additional_info_uri": (
|
||||
"https://reports.example.com/tls-help"
|
||||
),
|
||||
},
|
||||
{
|
||||
"result_type": "starttls-not-supported",
|
||||
"failed_session_count": 1,
|
||||
"sending_mta_ip": "192.0.2.2",
|
||||
"receiving_ip": "203.0.113.2",
|
||||
"receiving_mx_hostname": "mx2.example.com",
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
"policy_domain": "example.net",
|
||||
"policy_type": "tlsa",
|
||||
"successful_session_count": 50,
|
||||
"failed_session_count": 0,
|
||||
},
|
||||
]
|
||||
)
|
||||
with (
|
||||
patch("parsedmarc.elastic.Search", return_value=_empty_search()),
|
||||
patch("parsedmarc.elastic.Index"),
|
||||
patch.object(
|
||||
elastic_module._SMTPTLSReportDoc, "save", autospec=True
|
||||
) as mock_save,
|
||||
):
|
||||
save_smtp_tls_report_to_elasticsearch(report)
|
||||
doc = mock_save.call_args[0][0]
|
||||
self.assertEqual(
|
||||
list(doc.policies_combined), ["example.com / sts", "example.net / tlsa"]
|
||||
)
|
||||
expected_detail_expired = (
|
||||
"example.com / sts / certificate-expired / 192.0.2.1 / "
|
||||
"203.0.113.1 / mx1.example.com"
|
||||
)
|
||||
expected_detail_starttls = (
|
||||
"example.com / sts / starttls-not-supported / 192.0.2.2 / "
|
||||
"203.0.113.2 / mx2.example.com"
|
||||
)
|
||||
self.assertEqual(
|
||||
list(doc.failure_details_combined),
|
||||
[expected_detail_expired, expected_detail_starttls],
|
||||
)
|
||||
# The parser emits additional_info_uri (SMTPTLSFailureDetailsOptional
|
||||
# in types.py); the saver must persist it on the declared
|
||||
# additional_information_uri field rather than dropping it.
|
||||
self.assertEqual(
|
||||
doc.policies[0].failure_details[0].additional_information_uri,
|
||||
"https://reports.example.com/tls-help",
|
||||
)
|
||||
|
||||
|
||||
class TestBackwardCompatAlias(unittest.TestCase):
|
||||
def test_save_forensic_alias_points_to_save_failure(self):
|
||||
|
||||
+398
-10
@@ -376,16 +376,193 @@ class TestCreateIndexes(unittest.TestCase):
|
||||
|
||||
|
||||
class TestMigrateIndexes(unittest.TestCase):
|
||||
"""migrate_indexes backfills dkim_results_combined/spf_results_combined
|
||||
(issue #169) on pre-existing aggregate documents as a non-blocking
|
||||
background task. It is guarded by a cheap count() query so repeated
|
||||
startups against an already-backfilled index are a no-op, and any SDK
|
||||
error is caught and logged rather than raised, so it never blocks
|
||||
parsedmarc startup."""
|
||||
|
||||
def test_backfill_submitted_when_old_docs_exist(self):
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
# The legacy fo migration that runs first sees no base index.
|
||||
mock_index_cls.return_value.exists.return_value = False
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.return_value = {"count": 42}
|
||||
mock_get_conn.return_value = mock_client
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
mock_client.update_by_query.assert_called_once()
|
||||
kwargs = mock_client.update_by_query.call_args.kwargs
|
||||
self.assertEqual(kwargs["index"], "dmarc_aggregate*")
|
||||
self.assertEqual(kwargs["conflicts"], "proceed")
|
||||
self.assertFalse(kwargs["wait_for_completion"])
|
||||
self.assertEqual(
|
||||
kwargs["body"]["query"], opensearch_module._COMBINED_BACKFILL_QUERY
|
||||
)
|
||||
script_source = kwargs["body"]["script"]["source"]
|
||||
self.assertIn("ctx._source.dkim_results_combined", script_source)
|
||||
self.assertIn("ctx._source.spf_results_combined", script_source)
|
||||
|
||||
# The count() guard query also targets the date-suffixed pattern.
|
||||
count_kwargs = mock_client.count.call_args.kwargs
|
||||
self.assertEqual(count_kwargs["index"], "dmarc_aggregate*")
|
||||
self.assertEqual(
|
||||
count_kwargs["body"]["query"], opensearch_module._COMBINED_BACKFILL_QUERY
|
||||
)
|
||||
|
||||
def test_backfill_skipped_when_no_old_docs(self):
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
mock_index_cls.return_value.exists.return_value = False
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.return_value = {"count": 0}
|
||||
mock_get_conn.return_value = mock_client
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
mock_client.update_by_query.assert_not_called()
|
||||
|
||||
def test_backfill_skipped_when_no_aggregate_indexes(self):
|
||||
with patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn:
|
||||
migrate_indexes()
|
||||
migrate_indexes(aggregate_indexes=None)
|
||||
|
||||
mock_get_conn.assert_not_called()
|
||||
|
||||
def test_backfill_failure_does_not_raise(self):
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
mock_index_cls.return_value.exists.return_value = False
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.side_effect = RuntimeError("cluster unreachable")
|
||||
mock_get_conn.return_value = mock_client
|
||||
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
self.assertTrue(any("cluster unreachable" in msg for msg in cm.output))
|
||||
mock_client.update_by_query.assert_not_called()
|
||||
|
||||
def test_get_connection_failure_does_not_raise(self):
|
||||
"""connections.get_connection() itself sits outside the per-index
|
||||
try/except for the combined-field backfill; if it raises (e.g. no
|
||||
OpenSearch connection has been configured yet), migrate_indexes must
|
||||
still not propagate the exception, per its docstring's promise that
|
||||
any cluster error is caught and logged."""
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
# The legacy fo migration that runs first sees no base index.
|
||||
mock_index_cls.return_value.exists.return_value = False
|
||||
mock_get_conn.side_effect = RuntimeError("no connection")
|
||||
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
self.assertTrue(
|
||||
any("Skipping the dkim_results_combined" in msg for msg in cm.output)
|
||||
)
|
||||
self.assertTrue(any("no connection" in msg for msg in cm.output))
|
||||
|
||||
def test_smtp_tls_backfill_submitted_when_old_docs_exist(self):
|
||||
"""SMTP TLS analogue of test_backfill_submitted_when_old_docs_exist:
|
||||
policies_combined/failure_details_combined backfill (also issue
|
||||
#169) is submitted with its own guard query and painless script."""
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
# The legacy fo migration that runs first sees no base index.
|
||||
mock_index_cls.return_value.exists.return_value = False
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.return_value = {"count": 7}
|
||||
mock_get_conn.return_value = mock_client
|
||||
migrate_indexes(smtp_tls_indexes=["smtp_tls"])
|
||||
|
||||
mock_client.update_by_query.assert_called_once()
|
||||
kwargs = mock_client.update_by_query.call_args.kwargs
|
||||
self.assertEqual(kwargs["index"], "smtp_tls*")
|
||||
self.assertEqual(kwargs["conflicts"], "proceed")
|
||||
self.assertFalse(kwargs["wait_for_completion"])
|
||||
self.assertEqual(
|
||||
kwargs["body"]["query"], opensearch_module._SMTP_TLS_COMBINED_BACKFILL_QUERY
|
||||
)
|
||||
script_source = kwargs["body"]["script"]["source"]
|
||||
self.assertIn("ctx._source.policies_combined", script_source)
|
||||
self.assertIn("ctx._source.failure_details_combined", script_source)
|
||||
|
||||
count_kwargs = mock_client.count.call_args.kwargs
|
||||
self.assertEqual(count_kwargs["index"], "smtp_tls*")
|
||||
self.assertEqual(
|
||||
count_kwargs["body"]["query"],
|
||||
opensearch_module._SMTP_TLS_COMBINED_BACKFILL_QUERY,
|
||||
)
|
||||
|
||||
def test_smtp_tls_backfill_skipped_when_no_old_docs(self):
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
mock_index_cls.return_value.exists.return_value = False
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.return_value = {"count": 0}
|
||||
mock_get_conn.return_value = mock_client
|
||||
migrate_indexes(smtp_tls_indexes=["smtp_tls"])
|
||||
|
||||
mock_client.update_by_query.assert_not_called()
|
||||
|
||||
def test_smtp_tls_backfill_skipped_when_no_smtp_tls_indexes_or_aggregate(self):
|
||||
with patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn:
|
||||
migrate_indexes()
|
||||
migrate_indexes(smtp_tls_indexes=None)
|
||||
|
||||
mock_get_conn.assert_not_called()
|
||||
|
||||
def test_smtp_tls_backfill_failure_does_not_raise(self):
|
||||
"""SMTP TLS analogue of test_backfill_failure_does_not_raise: an
|
||||
error from the cluster during the smtp_tls_indexes loop is caught
|
||||
and logged rather than raised."""
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
mock_index_cls.return_value.exists.return_value = False
|
||||
mock_client = MagicMock()
|
||||
mock_client.count.side_effect = RuntimeError("cluster unreachable")
|
||||
mock_get_conn.return_value = mock_client
|
||||
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
|
||||
migrate_indexes(smtp_tls_indexes=["smtp_tls"])
|
||||
|
||||
self.assertTrue(any("cluster unreachable" in msg for msg in cm.output))
|
||||
mock_client.update_by_query.assert_not_called()
|
||||
|
||||
|
||||
class TestMigrateIndexesFoMigration(unittest.TestCase):
|
||||
"""The legacy `published_policy.fo` field was mapped as `long` in
|
||||
older indexes. migrate_indexes detects that and rebuilds the index
|
||||
with the text/keyword shape. The branch is gnarly; a regression
|
||||
would silently leave old data un-migrated."""
|
||||
would silently leave old data un-migrated. Each test stubs the
|
||||
combined-field backfill that now runs afterwards in the same call
|
||||
(count 0 → no-op)."""
|
||||
|
||||
def test_no_indexes_is_noop(self):
|
||||
migrate_indexes() # Should not raise
|
||||
@staticmethod
|
||||
def _noop_backfill_client():
|
||||
client = MagicMock()
|
||||
client.count.return_value = {"count": 0}
|
||||
return client
|
||||
|
||||
def test_skips_non_existent_index(self):
|
||||
with patch("parsedmarc.opensearch.Index") as mock_index_cls:
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
mock_get_conn.return_value = self._noop_backfill_client()
|
||||
mock_index_cls.return_value.exists.return_value = False
|
||||
migrate_indexes(aggregate_indexes=["missing"])
|
||||
# exists() returned False — no field_mapping fetch.
|
||||
@@ -394,12 +571,16 @@ class TestMigrateIndexes(unittest.TestCase):
|
||||
def test_skips_when_doc_mapping_absent(self):
|
||||
"""An index that has 'fo' but not under the 'doc' type
|
||||
(e.g., empty index with default mapping) is left alone."""
|
||||
with patch("parsedmarc.opensearch.Index") as mock_index_cls:
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
patch("parsedmarc.opensearch.reindex") as mock_reindex,
|
||||
):
|
||||
mock_get_conn.return_value = self._noop_backfill_client()
|
||||
idx = mock_index_cls.return_value
|
||||
idx.exists.return_value = True
|
||||
idx.get_field_mapping.return_value = {"some_key": {"mappings": {}}}
|
||||
with patch("parsedmarc.opensearch.reindex") as mock_reindex:
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate-2023-01-01"])
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate-2023-01-01"])
|
||||
mock_reindex.assert_not_called()
|
||||
|
||||
def test_migrates_when_fo_is_long(self):
|
||||
@@ -411,6 +592,8 @@ class TestMigrateIndexes(unittest.TestCase):
|
||||
patch("parsedmarc.opensearch.reindex") as mock_reindex,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
mock_client = self._noop_backfill_client()
|
||||
mock_get_conn.return_value = mock_client
|
||||
idx = mock_index_cls.return_value
|
||||
idx.exists.return_value = True
|
||||
idx.get_field_mapping.return_value = {
|
||||
@@ -423,16 +606,18 @@ class TestMigrateIndexes(unittest.TestCase):
|
||||
}
|
||||
}
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate-2023-01-01"])
|
||||
# reindex called from old → new (v2) index.
|
||||
# reindex called from old → new (v2) index, with the client from
|
||||
# connections.get_connection().
|
||||
mock_reindex.assert_called_once()
|
||||
# connections.get_connection consulted to get the ES client.
|
||||
mock_get_conn.assert_called_once()
|
||||
self.assertIs(mock_reindex.call_args.args[0], mock_client)
|
||||
|
||||
def test_skips_when_fo_already_text(self):
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
patch("parsedmarc.opensearch.reindex") as mock_reindex,
|
||||
):
|
||||
mock_get_conn.return_value = self._noop_backfill_client()
|
||||
idx = mock_index_cls.return_value
|
||||
idx.exists.return_value = True
|
||||
idx.get_field_mapping.return_value = {
|
||||
@@ -447,6 +632,35 @@ class TestMigrateIndexes(unittest.TestCase):
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate-2024-01-01"])
|
||||
mock_reindex.assert_not_called()
|
||||
|
||||
def test_index_exists_failure_does_not_raise(self):
|
||||
"""A cluster error inside the per-index fo-migration loop (e.g.
|
||||
Index(...).exists() raising because the cluster is unreachable)
|
||||
must not abort startup: it is caught, logged, and the loop moves
|
||||
on to the combined-field backfill, which is exercised here with
|
||||
its own connection failure so both warnings are asserted."""
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Index") as mock_index_cls,
|
||||
patch("parsedmarc.opensearch.connections.get_connection") as mock_get_conn,
|
||||
):
|
||||
mock_index_cls.return_value.exists.side_effect = ConnectionError(
|
||||
"cluster unreachable"
|
||||
)
|
||||
mock_get_conn.side_effect = RuntimeError("no connection")
|
||||
with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
|
||||
migrate_indexes(aggregate_indexes=["dmarc_aggregate"])
|
||||
|
||||
self.assertTrue(
|
||||
any(
|
||||
"legacy published_policy.fo migration" in msg
|
||||
and "cluster unreachable" in msg
|
||||
for msg in cm.output
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
any("Skipping the dkim_results_combined" in msg for msg in cm.output)
|
||||
)
|
||||
self.assertTrue(any("no connection" in msg for msg in cm.output))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# save_aggregate_report_to_opensearch
|
||||
@@ -602,6 +816,54 @@ class TestSaveAggregateReport(unittest.TestCase):
|
||||
mock_doc_cls.call_args.kwargs["date_begin"].timestamp(), 1705276800
|
||||
)
|
||||
|
||||
def test_save_populates_combined_dkim_and_spf_fields(self):
|
||||
"""Regression guard for issue #169: two DKIM signatures on one
|
||||
record must yield exactly two combined entries, not a 4-way
|
||||
cross-product. autospec=True is required on the save patch so
|
||||
mock_save.call_args captures the doc instance as ``self``."""
|
||||
report = _aggregate_report()
|
||||
report["records"][0]["auth_results"] = {
|
||||
"dkim": [
|
||||
{
|
||||
"domain": "example.net",
|
||||
"selector": "net1",
|
||||
"result": "fail",
|
||||
"human_result": None,
|
||||
},
|
||||
{
|
||||
"domain": "example.org",
|
||||
"selector": "org1",
|
||||
"result": "pass",
|
||||
"human_result": None,
|
||||
},
|
||||
],
|
||||
"spf": [
|
||||
{
|
||||
"domain": "example.org",
|
||||
"scope": "mfrom",
|
||||
"result": "pass",
|
||||
"human_result": None,
|
||||
},
|
||||
],
|
||||
}
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Search", return_value=_empty_search()),
|
||||
patch(
|
||||
"parsedmarc.opensearch.Index",
|
||||
return_value=MagicMock(exists=MagicMock(return_value=True)),
|
||||
),
|
||||
patch.object(
|
||||
opensearch_module._AggregateReportDoc, "save", autospec=True
|
||||
) as mock_save,
|
||||
):
|
||||
save_aggregate_report_to_opensearch(report)
|
||||
doc = mock_save.call_args[0][0]
|
||||
self.assertEqual(
|
||||
list(doc.dkim_results_combined),
|
||||
["net1 / example.net / fail", "org1 / example.org / pass"],
|
||||
)
|
||||
self.assertEqual(list(doc.spf_results_combined), ["mfrom / example.org / pass"])
|
||||
|
||||
|
||||
class TestAggregateDocPassedDmarc(unittest.TestCase):
|
||||
"""The _AggregateReportDoc.save() override derives passed_dmarc — the
|
||||
@@ -629,6 +891,56 @@ class TestAggregateDocPassedDmarc(unittest.TestCase):
|
||||
self.assertEqual(bool(doc.passed_dmarc), expected)
|
||||
|
||||
|
||||
class TestAggregateDocCombinedResults(unittest.TestCase):
|
||||
"""add_dkim_result/add_spf_result never touch the network, so these
|
||||
construct _AggregateReportDoc directly rather than going through the
|
||||
save_* entry point."""
|
||||
|
||||
def test_add_dkim_result_appends_combined_string(self):
|
||||
"""Regression guard for issue #169: dkim_results/spf_results are
|
||||
arrays of objects that the engine dynamic-maps as plain ``object``
|
||||
(not ``nested``) and flattens, so Kibana/Grafana tables cannot
|
||||
terms-aggregate their subfields without producing a cross-product
|
||||
of selector/domain/result values. The composed
|
||||
"selector / domain / result" string preserves the per-signature
|
||||
pairing that the flattened array loses."""
|
||||
doc = opensearch_module._AggregateReportDoc()
|
||||
doc.add_dkim_result(
|
||||
domain="example.net", selector="net1", result="fail", human_result=None
|
||||
)
|
||||
doc.add_dkim_result(
|
||||
domain="example.org", selector="org1", result="pass", human_result=None
|
||||
)
|
||||
expected = ["net1 / example.net / fail", "org1 / example.org / pass"]
|
||||
# dkim_results_combined is declared as Text(multi=True, ...); the SDK
|
||||
# stub types the class attribute as Text (no Iterable protocol),
|
||||
# even though the runtime value is an AttrList once multi=True is
|
||||
# set.
|
||||
self.assertEqual(list(doc.dkim_results_combined), expected) # pyright: ignore[reportArgumentType]
|
||||
self.assertEqual(doc.to_dict()["dkim_results_combined"], expected)
|
||||
|
||||
def test_add_spf_result_appends_combined_string(self):
|
||||
doc = opensearch_module._AggregateReportDoc()
|
||||
doc.add_spf_result(
|
||||
domain="example.org", scope="mfrom", result="pass", human_result=None
|
||||
)
|
||||
expected = ["mfrom / example.org / pass"]
|
||||
self.assertEqual(list(doc.spf_results_combined), expected) # pyright: ignore[reportArgumentType]
|
||||
self.assertEqual(doc.to_dict()["spf_results_combined"], expected)
|
||||
|
||||
def test_spf_result_serializes_under_singular_result_key(self):
|
||||
"""The _SPFResult class previously declared a dead ``results``
|
||||
(plural) field while the save path wrote ``result``; verify the
|
||||
serialized inner doc actually uses the singular key."""
|
||||
doc = opensearch_module._AggregateReportDoc()
|
||||
doc.add_spf_result(
|
||||
domain="example.org", scope="mfrom", result="pass", human_result=None
|
||||
)
|
||||
d = doc.to_dict()["spf_results"][0]
|
||||
self.assertEqual(d["result"], "pass")
|
||||
self.assertNotIn("results", d)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# save_failure_report_to_opensearch
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -979,6 +1291,82 @@ class TestSaveSmtpTlsReport(unittest.TestCase):
|
||||
save_smtp_tls_report_to_opensearch(report)
|
||||
mock_save.assert_called_once()
|
||||
|
||||
def test_save_populates_combined_policy_and_failure_detail_fields(self):
|
||||
"""Regression guard for the SMTP TLS analogue of issue #169:
|
||||
policies and their failure_details are object arrays, so stacked
|
||||
terms aggregations on their subfields cross-product just like
|
||||
dkim_results/spf_results did. Two policies (one with two failure
|
||||
details, one with none) must yield exactly two policies_combined
|
||||
entries and two failure_details_combined entries, not a
|
||||
cross-product. autospec=True is required on the save patch so
|
||||
mock_save.call_args captures the doc instance as ``self``."""
|
||||
report = _smtp_tls_report(
|
||||
policies=[
|
||||
{
|
||||
"policy_domain": "example.com",
|
||||
"policy_type": "sts",
|
||||
"successful_session_count": 100,
|
||||
"failed_session_count": 2,
|
||||
"failure_details": [
|
||||
{
|
||||
"result_type": "certificate-expired",
|
||||
"failed_session_count": 1,
|
||||
"sending_mta_ip": "192.0.2.1",
|
||||
"receiving_ip": "203.0.113.1",
|
||||
"receiving_mx_hostname": "mx1.example.com",
|
||||
"additional_info_uri": (
|
||||
"https://reports.example.com/tls-help"
|
||||
),
|
||||
},
|
||||
{
|
||||
"result_type": "starttls-not-supported",
|
||||
"failed_session_count": 1,
|
||||
"sending_mta_ip": "192.0.2.2",
|
||||
"receiving_ip": "203.0.113.2",
|
||||
"receiving_mx_hostname": "mx2.example.com",
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
"policy_domain": "example.net",
|
||||
"policy_type": "tlsa",
|
||||
"successful_session_count": 50,
|
||||
"failed_session_count": 0,
|
||||
},
|
||||
]
|
||||
)
|
||||
with (
|
||||
patch("parsedmarc.opensearch.Search", return_value=_empty_search()),
|
||||
patch("parsedmarc.opensearch.Index"),
|
||||
patch.object(
|
||||
opensearch_module._SMTPTLSReportDoc, "save", autospec=True
|
||||
) as mock_save,
|
||||
):
|
||||
save_smtp_tls_report_to_opensearch(report)
|
||||
doc = mock_save.call_args[0][0]
|
||||
self.assertEqual(
|
||||
list(doc.policies_combined), ["example.com / sts", "example.net / tlsa"]
|
||||
)
|
||||
expected_detail_expired = (
|
||||
"example.com / sts / certificate-expired / 192.0.2.1 / "
|
||||
"203.0.113.1 / mx1.example.com"
|
||||
)
|
||||
expected_detail_starttls = (
|
||||
"example.com / sts / starttls-not-supported / 192.0.2.2 / "
|
||||
"203.0.113.2 / mx2.example.com"
|
||||
)
|
||||
self.assertEqual(
|
||||
list(doc.failure_details_combined),
|
||||
[expected_detail_expired, expected_detail_starttls],
|
||||
)
|
||||
# The parser emits additional_info_uri (SMTPTLSFailureDetailsOptional
|
||||
# in types.py); the saver must persist it on the declared
|
||||
# additional_information_uri field rather than dropping it.
|
||||
self.assertEqual(
|
||||
doc.policies[0].failure_details[0].additional_information_uri,
|
||||
"https://reports.example.com/tls-help",
|
||||
)
|
||||
|
||||
|
||||
class TestBackwardCompatAlias(unittest.TestCase):
|
||||
def test_save_forensic_alias_points_to_save_failure(self):
|
||||
|
||||
Reference in New Issue
Block a user