amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-08-15 03:43:27 +00:00
Code Issues Projects Releases Wiki Activity

Extend the combined-field fix to SMTP TLS documents

Browse Source 
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>
This commit is contained in:
Sean Whalen
2026-07-24 23:04:26 -04:00
co-authored by Claude Fable 5
parent c3bdaab508
commit 9be85409d0
7 changed files with 686 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+2 -1
View File
@@ -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.
docs/source/elasticsearch.md
+60
View File
@@ -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
parsedmarc/cli.py
+2
View File
@@ -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:
parsedmarc/elastic.py
+186 -18
View File
@@ -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)
parsedmarc/opensearch.py
+180 -7
View File
@@ -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)
tests/test_elastic.py
+121
View File
@@ -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):
tests/test_opensearch.py
+135
View File
@@ -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):