From 9be85409d03ca4713a70aad123ba7a8f45bd0b62 Mon Sep 17 00:00:00 2001 From: Sean Whalen <44679+seanthegeek@users.noreply.github.com> Date: Fri, 24 Jul 2026 23:04:26 -0400 Subject: [PATCH] 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 --- CHANGELOG.md | 3 +- docs/source/elasticsearch.md | 60 +++++++++++ parsedmarc/cli.py | 2 + parsedmarc/elastic.py | 204 +++++++++++++++++++++++++++++++---- parsedmarc/opensearch.py | 187 ++++++++++++++++++++++++++++++-- tests/test_elastic.py | 121 +++++++++++++++++++++ tests/test_opensearch.py | 135 +++++++++++++++++++++++ 7 files changed, 686 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f9ecd287..bbcdf768 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,8 +12,9 @@ - **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 have the same class of defect and are tracked separately. +- **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. diff --git a/docs/source/elasticsearch.md b/docs/source/elasticsearch.md index 50bc984f..2d0a1a43 100644 --- a/docs/source/elasticsearch.md +++ b/docs/source/elasticsearch.md @@ -318,6 +318,66 @@ curl -X POST "http://localhost:9200/dmarc_aggregate*/_update_by_query?conflicts= 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 diff --git a/parsedmarc/cli.py b/parsedmarc/cli.py index 426be8e7..4e78b1d8 100644 --- a/parsedmarc/cli.py +++ b/parsedmarc/cli.py @@ -1446,6 +1446,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: @@ -1492,6 +1493,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: diff --git a/parsedmarc/elastic.py b/parsedmarc/elastic.py index d098ad19..ca15f321 100644 --- a/parsedmarc/elastic.py +++ b/parsedmarc/elastic.py @@ -124,6 +124,98 @@ _COMBINED_BACKFILL_SCRIPT = ( "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 @@ -407,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() @@ -448,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) @@ -471,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): @@ -580,43 +686,47 @@ 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, ): """ Backfills the ``dkim_results_combined``/``spf_results_combined`` fields - (added for issue #169) on aggregate report documents that were saved - before those fields existed. + (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. - 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 + 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 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. + 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 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: + 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 " - f"backfill: could not get an Elasticsearch connection: {e}. " - "This will be retried at the next startup." + "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: + for name in aggregate_indexes or []: pattern = f"{name}*" try: count_response = client.count( @@ -651,6 +761,44 @@ def migrate_indexes( "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( aggregate_report: dict[str, Any], @@ -1118,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"], @@ -1163,6 +1321,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) diff --git a/parsedmarc/opensearch.py b/parsedmarc/opensearch.py index de0bfd16..432d0bc0 100644 --- a/parsedmarc/opensearch.py +++ b/parsedmarc/opensearch.py @@ -117,6 +117,98 @@ _COMBINED_BACKFILL_SCRIPT = ( "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() @@ -335,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() @@ -370,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) @@ -387,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): @@ -500,6 +606,7 @@ 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, ): """ Runs index migrations and backfills. @@ -524,17 +631,22 @@ def migrate_indexes( 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 """ - if not aggregate_indexes: + if not aggregate_indexes and not smtp_tls_indexes: return version = 2 - for aggregate_index_name in aggregate_indexes: + for aggregate_index_name in aggregate_indexes or []: try: if not Index(aggregate_index_name).exists(): continue @@ -578,12 +690,13 @@ def migrate_indexes( client = connections.get_connection() except Exception as e: logger.warning( - "Skipping the dkim_results_combined/spf_results_combined " - f"backfill: could not get an OpenSearch connection: {e}. " - "This will be retried at the next startup." + "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: + for name in aggregate_indexes or []: pattern = f"{name}*" try: count_response = client.count( @@ -623,6 +736,46 @@ def migrate_indexes( "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( aggregate_report: dict[str, Any], @@ -1085,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"], @@ -1130,6 +1293,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) diff --git a/tests/test_elastic.py b/tests/test_elastic.py index 702653e6..b2dab2a2 100644 --- a/tests/test_elastic.py +++ b/tests/test_elastic.py @@ -474,6 +474,64 @@ class TestMigrateIndexes(unittest.TestCase): ) 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 @@ -1109,6 +1167,69 @@ 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", + }, + { + "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"] + ) + self.assertEqual( + list(doc.failure_details_combined), + [ + "example.com / sts / certificate-expired / 192.0.2.1 / " + "203.0.113.1 / mx1.example.com", + "example.com / sts / starttls-not-supported / 192.0.2.2 / " + "203.0.113.2 / mx2.example.com", + ], + ) + class TestBackwardCompatAlias(unittest.TestCase): def test_save_forensic_alias_points_to_save_failure(self): diff --git a/tests/test_opensearch.py b/tests/test_opensearch.py index b973c96f..df57f9d4 100644 --- a/tests/test_opensearch.py +++ b/tests/test_opensearch.py @@ -470,6 +470,78 @@ class TestMigrateIndexes(unittest.TestCase): ) 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 @@ -1219,6 +1291,69 @@ 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", + }, + { + "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"] + ) + self.assertEqual( + list(doc.failure_details_combined), + [ + "example.com / sts / certificate-expired / 192.0.2.1 / " + "203.0.113.1 / mx1.example.com", + "example.com / sts / starttls-not-supported / 192.0.2.2 / " + "203.0.113.2 / mx2.example.com", + ], + ) + class TestBackwardCompatAlias(unittest.TestCase): def test_save_forensic_alias_points_to_save_failure(self):