# -*- coding: utf-8 -*-
from __future__ import annotations
from typing import Any
import boto3
from opensearchpy import (
AWSV4SignerAuth,
Boolean,
Date,
Document,
Index,
InnerDoc,
Integer,
Ip,
Keyword,
Nested,
Object,
Q,
RequestsHttpConnection,
Search,
Text,
connections,
)
from opensearchpy.helpers import reindex
from parsedmarc import InvalidFailureReport
from parsedmarc.log import logger
from parsedmarc.utils import human_timestamp_to_datetime
[docs]
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()
class _PublishedPolicy(InnerDoc):
domain = Text()
adkim = Text()
aspf = Text()
p = Text()
sp = Text()
pct = Integer()
fo = Text()
np = Keyword()
testing = Keyword()
discovery_method = Keyword()
class _DKIMResult(InnerDoc):
domain = Text()
selector = Text()
result = Text()
human_result = Text()
class _SPFResult(InnerDoc):
domain = Text()
scope = Text()
result = Text()
human_result = Text()
class _AggregateReportDoc(Document):
class Index:
name = "dmarc_aggregate"
xml_schema = Text()
xml_namespace = Keyword()
org_name = Text()
org_email = Text()
org_extra_contact_info = Text()
report_id = Text()
date_range = Date()
date_begin = Date()
date_end = Date()
normalized_timespan = Boolean()
original_timespan_seconds = Integer
errors = Text()
published_policy = Object(_PublishedPolicy)
source_ip_address = Ip()
source_country = Text()
source_reverse_dns = Text()
source_base_domain = Text()
source_type = Text()
source_name = Text()
source_asn = Integer()
source_as_name = Text()
source_as_domain = Text()
message_count = Integer
disposition = Text()
dkim_aligned = Boolean()
spf_aligned = Boolean()
passed_dmarc = Boolean()
policy_overrides = Nested(_PolicyOverride)
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()
generator = Text()
def add_policy_override(self, type_: str, comment: str):
self.policy_overrides.append(_PolicyOverride(type=type_, comment=comment))
def add_dkim_result(
self,
domain: str,
selector: str,
result: str,
human_result: str | None = None,
):
self.dkim_results.append(
_DKIMResult(
domain=domain,
selector=selector,
result=result,
human_result=human_result,
)
)
self.dkim_results_combined.append(f"{selector} / {domain} / {result}")
def add_spf_result(
self,
domain: str,
scope: str,
result: str,
human_result: str | None = None,
):
self.spf_results.append(
_SPFResult(
domain=domain,
scope=scope,
result=result,
human_result=human_result,
)
)
self.spf_results_combined.append(f"{scope} / {domain} / {result}")
def save(self, **kwargs): # pyright: ignore[reportIncompatibleMethodOverride]
self.passed_dmarc = False
self.passed_dmarc = self.spf_aligned or self.dkim_aligned
return super().save(**kwargs)
class _EmailAddressDoc(InnerDoc):
display_name = Text()
address = Text()
class _EmailAttachmentDoc(Document):
filename = Text()
content_type = Text()
sha256 = Text()
class _FailureSampleDoc(InnerDoc):
raw = Text()
headers = Object()
headers_only = Boolean()
to = Nested(_EmailAddressDoc)
subject = Text()
filename_safe_subject = Text()
_from = Object(_EmailAddressDoc)
date = Date()
reply_to = Nested(_EmailAddressDoc)
cc = Nested(_EmailAddressDoc)
bcc = Nested(_EmailAddressDoc)
body = Text()
attachments = Nested(_EmailAttachmentDoc)
def add_to(self, display_name: str, address: str):
self.to.append(_EmailAddressDoc(display_name=display_name, address=address))
def add_reply_to(self, display_name: str, address: str):
self.reply_to.append(
_EmailAddressDoc(display_name=display_name, address=address)
)
def add_cc(self, display_name: str, address: str):
self.cc.append(_EmailAddressDoc(display_name=display_name, address=address))
def add_bcc(self, display_name: str, address: str):
self.bcc.append(_EmailAddressDoc(display_name=display_name, address=address))
def add_attachment(self, filename: str, content_type: str, sha256: str):
self.attachments.append(
_EmailAttachmentDoc(
filename=filename, content_type=content_type, sha256=sha256
)
)
class _FailureReportDoc(Document):
class Index:
name = "dmarc_failure"
feedback_type = Text()
user_agent = Text()
version = Text()
original_mail_from = Text()
arrival_date = Date()
domain = Text()
original_envelope_id = Text()
authentication_results = Text()
delivery_results = Text()
source_ip_address = Ip()
source_country = Text()
source_reverse_dns = Text()
source_asn = Integer()
source_as_name = Text()
source_as_domain = Text()
source_authentication_mechanisms = Text()
source_auth_failures = Text()
dkim_domain = Text()
original_rcpt_to = Text()
sample = Object(_FailureSampleDoc)
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()
failure_reason_code = Text()
class _SMTPTLSPolicyDoc(InnerDoc):
policy_domain = Text()
policy_type = Text()
policy_strings = Text()
mx_host_patterns = Text()
successful_session_count = Integer()
failed_session_count = Integer()
failure_details = Nested(_SMTPTLSFailureDetailsDoc)
def add_failure_details(
self,
result_type: str | None = None,
ip_address: str | None = None,
receiving_ip: str | None = None,
receiving_mx_helo: str | None = None,
failed_session_count: int | None = None,
sending_mta_ip: str | None = None,
receiving_mx_hostname: str | None = None,
additional_information_uri: str | None = None,
failure_reason_code: str | int | None = None,
):
_details = _SMTPTLSFailureDetailsDoc(
result_type=result_type,
ip_address=ip_address,
sending_mta_ip=sending_mta_ip,
receiving_mx_hostname=receiving_mx_hostname,
receiving_mx_helo=receiving_mx_helo,
receiving_ip=receiving_ip,
failed_session_count=failed_session_count,
additional_information_uri=additional_information_uri,
failure_reason_code=failure_reason_code,
)
self.failure_details.append(_details)
class _SMTPTLSReportDoc(Document):
class Index:
name = "smtp_tls"
organization_name = Text()
date_range = Date()
date_begin = Date()
date_end = Date()
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)}
)
[docs]
class AlreadySaved(ValueError):
"""Raised when a report to be saved matches an existing report"""
[docs]
def set_hosts(
hosts: str | list[str],
*,
use_ssl: bool | None = False,
ssl_cert_path: str | None = None,
skip_certificate_verification: bool = False,
username: str | None = None,
password: str | None = None,
api_key: str | None = None,
timeout: float | None = 60.0,
auth_type: str = "basic",
aws_region: str | None = None,
aws_service: str = "es",
):
"""
Sets the OpenSearch hosts to use
Args:
hosts (str|list[str]): A single hostname or URL, or list of hostnames or URLs
use_ssl (bool): Use an HTTPS connection to the server
ssl_cert_path (str): Path to the certificate chain
skip_certificate_verification (bool): Skip certificate verification
username (str): The username to use for authentication
password (str): The password to use for authentication
api_key (str): The Base64 encoded API key to use for authentication
timeout (float): Timeout in seconds
auth_type (str): OpenSearch auth mode: basic (default) or awssigv4
aws_region (str): AWS region for SigV4 auth (required for awssigv4)
aws_service (str): AWS service for SigV4 signing (default: es)
"""
if not isinstance(hosts, list):
hosts = [hosts]
logger.debug("Connecting to OpenSearch: hosts=%s, use_ssl=%s", hosts, use_ssl)
conn_params = {"hosts": hosts, "timeout": timeout}
if use_ssl:
conn_params["use_ssl"] = True
if ssl_cert_path:
conn_params["ca_certs"] = ssl_cert_path
if skip_certificate_verification:
conn_params["verify_certs"] = False
else:
conn_params["verify_certs"] = True
normalized_auth_type = (auth_type or "basic").strip().lower()
if normalized_auth_type == "awssigv4":
if not aws_region:
raise OpenSearchError(
"OpenSearch AWS SigV4 auth requires 'aws_region' to be set"
)
session = boto3.Session()
credentials = session.get_credentials()
if credentials is None:
raise OpenSearchError(
"Unable to load AWS credentials for OpenSearch SigV4 authentication"
)
conn_params["http_auth"] = AWSV4SignerAuth(credentials, aws_region, aws_service)
conn_params["connection_class"] = RequestsHttpConnection
elif normalized_auth_type == "basic":
if username and password:
conn_params["http_auth"] = (username, password)
if api_key:
conn_params["api_key"] = api_key
else:
raise OpenSearchError(
f"Unsupported OpenSearch auth_type '{auth_type}'. "
"Expected 'basic' or 'awssigv4'."
)
connections.create_connection(**conn_params)
[docs]
def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
"""
Create OpenSearch indexes
Args:
names (list): A list of index names
settings (dict): Index settings
"""
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(f"Creating OpenSearch index: {name}")
if settings is None:
index.settings(number_of_shards=1, number_of_replicas=0)
else:
index.settings(**settings)
index.create()
except Exception as e:
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
_LEGACY_FO_FIELD = "published_policy.fo"
# The same field split into the object/leaf names a mapping body nests it
# under. Derived from the dotted name so the mapping written below cannot
# drift from the field _legacy_fo_field_type() reads.
_LEGACY_FO_OBJECT, _LEGACY_FO_LEAF = _LEGACY_FO_FIELD.split(".")
def _legacy_fo_field_type(index: Index) -> str | None:
"""Return the mapped type of ``published_policy.fo`` in *index*.
Returns ``None`` when the index does not map the field at all.
Elasticsearch 6-era clusters keyed field mappings by the mapping type
name (``doc``); mapping types are gone from both OpenSearch and
Elasticsearch 8, whose responses put the field directly under
``mappings``. The type-keyed shape is only descended into when such a
key is actually present, so this reads either shape.
Args:
index (Index): The index to inspect.
Returns:
str | None: The mapped field type, e.g. ``"long"`` or ``"text"``.
"""
response = index.get_field_mapping(fields=[_LEGACY_FO_FIELD])
mappings = response[list(response.keys())[0]]["mappings"]
if _LEGACY_FO_FIELD not in mappings:
for type_name in ("doc", "_doc"):
if type_name in mappings:
mappings = mappings[type_name]
break
field_mapping = mappings.get(_LEGACY_FO_FIELD)
if not field_mapping:
return None
return field_mapping.get("mapping", {}).get(_LEGACY_FO_LEAF, {}).get("type")
[docs]
def migrate_indexes(
aggregate_indexes: list[str] | None = None,
failure_indexes: list[str] | None = None,
smtp_tls_indexes: list[str] | None = None,
legacy_fo_indexes: list[str] | None = None,
):
"""
Runs index migrations and backfills.
First, the legacy ``published_policy.fo`` migration, for each name in
``legacy_fo_indexes``: parsedmarc releases before 5.0.0 declared that
field as an integer, so those indexes mapped it as ``long``, which
cannot hold the multi-value ``fo`` settings reports carry (``0:1``,
``d:s``). Such an index is rebuilt as a ``-v2`` index with the
text/keyword shape, the documents are reindexed into it, and the
original is deleted.
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
legacy_fo_indexes (list): A list of index names to check for the
pre-5.0.0 ``published_policy.fo`` ``long`` mapping. Unlike the
backfill arguments these are exact names, not patterns:
5.0.0 introduced date-suffixed index names in the same release
that fixed the mapping, so an affected index has no date
component. It may still be prefixed or suffixed -- both
options date back to 4.1.0 -- so callers should pass the
names their own ``index_prefix``/``index_suffix``
configuration produces.
"""
if not aggregate_indexes and not smtp_tls_indexes and not legacy_fo_indexes:
return
version = 2
for legacy_index_name in legacy_fo_indexes or []:
try:
legacy_index = Index(legacy_index_name)
if not legacy_index.exists():
continue
if _legacy_fo_field_type(legacy_index) != "long":
continue
new_index_name = f"{legacy_index_name}-v{version}"
# Nested object form rather than the dotted key this used to
# send. Both are accepted and produce an identical mapping
# (verified against Elasticsearch 8.19 and OpenSearch 3), but
# dot expansion is conditional on the object field's
# `subobjects` setting, and this shape never is.
body = {
"properties": {
_LEGACY_FO_OBJECT: {
"properties": {
_LEGACY_FO_LEAF: {
"type": "text",
"fields": {
"keyword": {"type": "keyword", "ignore_above": 256}
},
}
}
}
}
}
logger.info(
f"Migrating {legacy_index_name} to {new_index_name}: "
f"{_LEGACY_FO_FIELD} is mapped as long, as parsedmarc "
"releases before 5.0.0 declared it"
)
# Reaching here means the original index still holds the data:
# it is deleted only after the reindex below succeeds. So a
# leftover target index is the debris of an earlier attempt
# that died between create() and delete(), and keeping it would
# fail every later attempt on "resource already exists".
if Index(new_index_name).exists():
logger.warning(
f"Discarding {new_index_name} left behind by an earlier "
f"interrupted migration of {legacy_index_name}"
)
Index(new_index_name).delete()
Index(new_index_name).create()
Index(new_index_name).put_mapping(body=body)
reindex(connections.get_connection(), legacy_index_name, new_index_name)
Index(legacy_index_name).delete()
except Exception as e:
logger.warning(
"Failed the legacy published_policy.fo migration for "
f"{legacy_index_name}: {e}. This will be retried at the "
"next startup."
)
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 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."
)
[docs]
def save_aggregate_report_to_opensearch(
aggregate_report: dict[str, Any],
index_suffix: str | None = None,
index_prefix: str | None = None,
monthly_indexes: bool = False,
number_of_shards: int = 1,
number_of_replicas: int = 0,
):
"""
Saves a parsed DMARC aggregate report to OpenSearch
Args:
aggregate_report (dict): A parsed aggregate report
index_suffix (str): The suffix of the name of the index to save to
index_prefix (str): The prefix of the name of the index to save to
monthly_indexes (bool): Use monthly indexes instead of daily indexes
number_of_shards (int): The number of shards to use in the index
number_of_replicas (int): The number of replicas to use in the index
Raises:
AlreadySaved
"""
logger.info("Saving aggregate report to OpenSearch")
aggregate_report = aggregate_report.copy()
metadata = aggregate_report["report_metadata"]
org_name = metadata["org_name"]
report_id = metadata["report_id"]
domain = aggregate_report["policy_published"]["domain"]
begin_date = human_timestamp_to_datetime(metadata["begin_date"], to_utc=True)
end_date = human_timestamp_to_datetime(metadata["end_date"], to_utc=True)
if monthly_indexes:
index_date = begin_date.strftime("%Y-%m")
else:
index_date = begin_date.strftime("%Y-%m-%d")
org_name_query = Q(dict(match_phrase=dict(org_name=org_name)))
report_id_query = Q(dict(match_phrase=dict(report_id=report_id)))
domain_query = Q(dict(match_phrase={"published_policy.domain": domain}))
begin_date_query = Q(dict(range=dict(date_begin=dict(gte=begin_date))))
end_date_query = Q(dict(range=dict(date_end=dict(lte=end_date))))
if index_suffix is not None:
search_index = f"dmarc_aggregate_{index_suffix}*"
else:
search_index = "dmarc_aggregate*"
if index_prefix is not None:
search_index = f"{index_prefix}{search_index}"
search = Search(index=search_index)
query = org_name_query & report_id_query & domain_query
query = query & begin_date_query & end_date_query
search.query = query
begin_date_human = begin_date.strftime("%Y-%m-%d %H:%M:%SZ")
end_date_human = end_date.strftime("%Y-%m-%d %H:%M:%SZ")
try:
existing = search.execute()
except Exception as error_:
raise OpenSearchError(
f"OpenSearch's search for existing report error: {error_.__str__()}"
)
if len(existing) > 0:
raise AlreadySaved(
f"An aggregate report ID {report_id} from {org_name} about {domain} "
f"with a date range of {begin_date_human} UTC to {end_date_human} UTC already "
"exists in "
"OpenSearch"
)
published_policy = _PublishedPolicy(
domain=aggregate_report["policy_published"]["domain"],
adkim=aggregate_report["policy_published"]["adkim"],
aspf=aggregate_report["policy_published"]["aspf"],
p=aggregate_report["policy_published"]["p"],
sp=aggregate_report["policy_published"]["sp"],
pct=aggregate_report["policy_published"]["pct"],
fo=aggregate_report["policy_published"]["fo"],
np=aggregate_report["policy_published"].get("np"),
testing=aggregate_report["policy_published"].get("testing"),
discovery_method=aggregate_report["policy_published"].get("discovery_method"),
)
for record in aggregate_report["records"]:
begin_date = human_timestamp_to_datetime(
record["interval_begin"], to_utc=True, assume_utc=True
)
end_date = human_timestamp_to_datetime(
record["interval_end"], to_utc=True, assume_utc=True
)
normalized_timespan = record["normalized_timespan"]
if monthly_indexes:
index_date = begin_date.strftime("%Y-%m")
else:
index_date = begin_date.strftime("%Y-%m-%d")
aggregate_report["begin_date"] = begin_date
aggregate_report["end_date"] = end_date
date_range = [aggregate_report["begin_date"], aggregate_report["end_date"]]
agg_doc = _AggregateReportDoc(
xml_schema=aggregate_report["xml_schema"],
xml_namespace=aggregate_report.get("xml_namespace"),
org_name=metadata["org_name"],
org_email=metadata["org_email"],
org_extra_contact_info=metadata["org_extra_contact_info"],
report_id=metadata["report_id"],
date_range=date_range,
date_begin=begin_date,
date_end=end_date,
normalized_timespan=normalized_timespan,
errors=metadata["errors"],
published_policy=published_policy,
source_ip_address=record["source"]["ip_address"],
source_country=record["source"]["country"],
source_reverse_dns=record["source"]["reverse_dns"],
source_base_domain=record["source"]["base_domain"],
source_type=record["source"]["type"],
source_name=record["source"]["name"],
source_asn=record["source"]["asn"],
source_as_name=record["source"]["as_name"],
source_as_domain=record["source"]["as_domain"],
message_count=record["count"],
disposition=record["policy_evaluated"]["disposition"],
dkim_aligned=record["policy_evaluated"]["dkim"] is not None
and record["policy_evaluated"]["dkim"].lower() == "pass",
spf_aligned=record["policy_evaluated"]["spf"] is not None
and record["policy_evaluated"]["spf"].lower() == "pass",
header_from=record["identifiers"]["header_from"],
envelope_from=record["identifiers"]["envelope_from"],
envelope_to=record["identifiers"]["envelope_to"],
np=aggregate_report["policy_published"].get("np"),
testing=aggregate_report["policy_published"].get("testing"),
discovery_method=aggregate_report["policy_published"].get(
"discovery_method"
),
generator=metadata.get("generator"),
)
for override in record["policy_evaluated"]["policy_override_reasons"]:
agg_doc.add_policy_override(
type_=override["type"], comment=override["comment"]
)
for dkim_result in record["auth_results"]["dkim"]:
agg_doc.add_dkim_result(
domain=dkim_result["domain"],
selector=dkim_result["selector"],
result=dkim_result["result"],
human_result=dkim_result.get("human_result"),
)
for spf_result in record["auth_results"]["spf"]:
agg_doc.add_spf_result(
domain=spf_result["domain"],
scope=spf_result["scope"],
result=spf_result["result"],
human_result=spf_result.get("human_result"),
)
index = "dmarc_aggregate"
if index_suffix:
index = f"{index}_{index_suffix}"
if index_prefix:
index = f"{index_prefix}{index}"
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
create_indexes([index], index_settings)
agg_doc.meta.index = index
try:
agg_doc.save()
except Exception as e:
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
[docs]
def save_failure_report_to_opensearch(
failure_report: dict[str, Any],
index_suffix: str | None = None,
index_prefix: str | None = None,
monthly_indexes: bool = False,
number_of_shards: int = 1,
number_of_replicas: int = 0,
):
"""
Saves a parsed DMARC failure report to OpenSearch
Args:
failure_report (dict): A parsed failure report
index_suffix (str): The suffix of the name of the index to save to
index_prefix (str): The prefix of the name of the index to save to
monthly_indexes (bool): Use monthly indexes instead of daily
indexes
number_of_shards (int): The number of shards to use in the index
number_of_replicas (int): The number of replicas to use in the
index
Raises:
AlreadySaved
"""
logger.info("Saving failure report to OpenSearch")
failure_report = failure_report.copy()
sample_date = None
if failure_report["parsed_sample"]["date"] is not None:
sample_date = failure_report["parsed_sample"]["date"]
sample_date = human_timestamp_to_datetime(sample_date)
original_headers = failure_report["parsed_sample"]["headers"]
headers: dict[str, Any] = {}
for original_header in original_headers:
headers[original_header.lower()] = original_headers[original_header]
# arrival_date_utc is a UTC wall-clock string; without assume_utc the
# naive .timestamp() below would interpret it as local time and skew
# the epoch by the host's UTC offset.
arrival_date = human_timestamp_to_datetime(
failure_report["arrival_date_utc"], assume_utc=True
)
arrival_date_epoch_milliseconds = int(arrival_date.timestamp() * 1000)
if index_suffix is not None:
search_index = f"dmarc_failure_{index_suffix}*,dmarc_forensic_{index_suffix}*"
else:
search_index = "dmarc_failure*,dmarc_forensic*"
if index_prefix is not None:
search_index = ",".join(
f"{index_prefix}{part}" for part in search_index.split(",")
)
search = Search(index=search_index)
q = Q(dict(match=dict(arrival_date=arrival_date_epoch_milliseconds)))
from_ = None
to_ = None
subject = None
if "from" in headers:
# We convert the FROM header from a string list to a flat string.
headers["from"] = headers["from"][0]
if headers["from"][0] == "":
headers["from"] = headers["from"][1]
else:
headers["from"] = " <".join(headers["from"]) + ">"
from_ = dict()
from_["sample.headers.from"] = headers["from"]
from_query = Q(dict(match_phrase=from_))
q = q & from_query
if "to" in headers:
# We convert the TO header from a string list to a flat string.
headers["to"] = headers["to"][0]
if headers["to"][0] == "":
headers["to"] = headers["to"][1]
else:
headers["to"] = " <".join(headers["to"]) + ">"
to_ = dict()
to_["sample.headers.to"] = headers["to"]
to_query = Q(dict(match_phrase=to_))
q = q & to_query
if "reply-to" in headers:
# Flatten the Reply-To header to a string so it can be displayed
# and aggregated like From/To. Only the first address is used,
# matching the From/To handling above. Not part of the dedup
# query.
headers["reply-to"] = headers["reply-to"][0]
if headers["reply-to"][0] == "":
headers["reply-to"] = headers["reply-to"][1]
else:
headers["reply-to"] = " <".join(headers["reply-to"]) + ">"
if "subject" in headers:
subject = headers["subject"]
subject_query = {"match_phrase": {"sample.headers.subject": subject}}
q = q & Q(subject_query)
search.query = q
existing = search.execute()
if len(existing) > 0:
raise AlreadySaved(
"A failure sample to {} from {} "
"with a subject of {} and arrival date of {} "
"already exists in "
"OpenSearch".format(to_, from_, subject, failure_report["arrival_date_utc"])
)
parsed_sample = failure_report["parsed_sample"]
sample = _FailureSampleDoc(
raw=failure_report["sample"],
headers=headers,
headers_only=failure_report["sample_headers_only"],
date=sample_date,
subject=failure_report["parsed_sample"]["subject"],
filename_safe_subject=parsed_sample["filename_safe_subject"],
body=failure_report["parsed_sample"]["body"],
)
for address in failure_report["parsed_sample"]["to"]:
sample.add_to(display_name=address["display_name"], address=address["address"])
for address in failure_report["parsed_sample"]["reply_to"]:
sample.add_reply_to(
display_name=address["display_name"], address=address["address"]
)
for address in failure_report["parsed_sample"]["cc"]:
sample.add_cc(display_name=address["display_name"], address=address["address"])
for address in failure_report["parsed_sample"]["bcc"]:
sample.add_bcc(display_name=address["display_name"], address=address["address"])
for attachment in failure_report["parsed_sample"]["attachments"]:
sample.add_attachment(
filename=attachment["filename"],
content_type=attachment["mail_content_type"],
sha256=attachment["sha256"],
)
try:
failure_doc = _FailureReportDoc(
feedback_type=failure_report["feedback_type"],
user_agent=failure_report["user_agent"],
version=failure_report["version"],
original_mail_from=failure_report["original_mail_from"],
arrival_date=arrival_date_epoch_milliseconds,
domain=failure_report["reported_domain"],
original_envelope_id=failure_report["original_envelope_id"],
authentication_results=failure_report["authentication_results"],
delivery_results=failure_report["delivery_result"],
source_ip_address=failure_report["source"]["ip_address"],
source_country=failure_report["source"]["country"],
source_reverse_dns=failure_report["source"]["reverse_dns"],
source_base_domain=failure_report["source"]["base_domain"],
source_asn=failure_report["source"]["asn"],
source_as_name=failure_report["source"]["as_name"],
source_as_domain=failure_report["source"]["as_domain"],
authentication_mechanisms=failure_report["authentication_mechanisms"],
auth_failure=failure_report["auth_failure"],
dkim_domain=failure_report["dkim_domain"],
original_rcpt_to=failure_report["original_rcpt_to"],
sample=sample,
)
index = "dmarc_failure"
if index_suffix:
index = f"{index}_{index_suffix}"
if index_prefix:
index = f"{index_prefix}{index}"
if monthly_indexes:
index_date = arrival_date.strftime("%Y-%m")
else:
index_date = arrival_date.strftime("%Y-%m-%d")
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
create_indexes([index], index_settings)
failure_doc.meta.index = index
try:
failure_doc.save()
except Exception as e:
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
except KeyError as e:
raise InvalidFailureReport(
f"Failure report missing required field: {e.__str__()}"
)
[docs]
def save_smtp_tls_report_to_opensearch(
report: dict[str, Any],
index_suffix: str | None = None,
index_prefix: str | None = None,
monthly_indexes: bool = False,
number_of_shards: int = 1,
number_of_replicas: int = 0,
):
"""
Saves a parsed SMTP TLS report to OpenSearch
Args:
report (dict): A parsed SMTP TLS report
index_suffix (str): The suffix of the name of the index to save to
index_prefix (str): The prefix of the name of the index to save to
monthly_indexes (bool): Use monthly indexes instead of daily indexes
number_of_shards (int): The number of shards to use in the index
number_of_replicas (int): The number of replicas to use in the index
Raises:
AlreadySaved
"""
logger.info("Saving SMTP TLS report to OpenSearch")
org_name = report["organization_name"]
report_id = report["report_id"]
begin_date = human_timestamp_to_datetime(report["begin_date"], to_utc=True)
end_date = human_timestamp_to_datetime(report["end_date"], to_utc=True)
begin_date_human = begin_date.strftime("%Y-%m-%d %H:%M:%SZ")
end_date_human = end_date.strftime("%Y-%m-%d %H:%M:%SZ")
if monthly_indexes:
index_date = begin_date.strftime("%Y-%m")
else:
index_date = begin_date.strftime("%Y-%m-%d")
report = report.copy()
report["begin_date"] = begin_date
report["end_date"] = end_date
org_name_query = Q(dict(match_phrase=dict(org_name=org_name)))
report_id_query = Q(dict(match_phrase=dict(report_id=report_id)))
begin_date_query = Q(dict(match=dict(date_begin=begin_date)))
end_date_query = Q(dict(match=dict(date_end=end_date)))
if index_suffix is not None:
search_index = f"smtp_tls_{index_suffix}*"
else:
search_index = "smtp_tls*"
if index_prefix is not None:
search_index = f"{index_prefix}{search_index}"
search = Search(index=search_index)
query = org_name_query & report_id_query
query = query & begin_date_query & end_date_query
search.query = query
try:
existing = search.execute()
except Exception as error_:
raise OpenSearchError(
f"OpenSearch's search for existing report error: {error_.__str__()}"
)
if len(existing) > 0:
raise AlreadySaved(
f"An SMTP TLS report ID {report_id} from "
f" {org_name} with a date range of "
f"{begin_date_human} UTC to "
f"{end_date_human} UTC already "
"exists in OpenSearch"
)
index = "smtp_tls"
if index_suffix:
index = f"{index}_{index_suffix}"
if index_prefix:
index = f"{index_prefix}{index}"
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
smtp_tls_doc = _SMTPTLSReportDoc(
org_name=report["organization_name"],
date_range=[report["begin_date"], report["end_date"]],
date_begin=report["begin_date"],
date_end=report["end_date"],
contact_info=report["contact_info"],
report_id=report["report_id"],
)
for policy in report["policies"]:
policy_strings = None
mx_host_patterns = None
if "policy_strings" in policy:
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(
f"{policy_domain_combined} / {policy_type_combined}"
)
policy_doc = _SMTPTLSPolicyDoc(
policy_domain=policy["policy_domain"],
policy_type=policy["policy_type"],
successful_session_count=policy["successful_session_count"],
failed_session_count=policy["failed_session_count"],
policy_string=policy_strings,
mx_host_patterns=mx_host_patterns,
)
if "failure_details" in policy:
for failure_detail in policy["failure_details"]:
receiving_mx_hostname = None
additional_information_uri = None
failure_reason_code = None
ip_address = None
receiving_ip = None
receiving_mx_helo = None
sending_mta_ip = None
if "receiving_mx_hostname" in failure_detail:
receiving_mx_hostname = failure_detail["receiving_mx_hostname"]
# 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"
]
if "failure_reason_code" in failure_detail:
failure_reason_code = failure_detail["failure_reason_code"]
if "ip_address" in failure_detail:
ip_address = failure_detail["ip_address"]
if "receiving_ip" in failure_detail:
receiving_ip = failure_detail["receiving_ip"]
if "receiving_mx_helo" in failure_detail:
receiving_mx_helo = failure_detail["receiving_mx_helo"]
if "sending_mta_ip" in failure_detail:
sending_mta_ip = failure_detail["sending_mta_ip"]
policy_doc.add_failure_details(
result_type=failure_detail["result_type"],
ip_address=ip_address,
receiving_ip=receiving_ip,
receiving_mx_helo=receiving_mx_helo,
failed_session_count=failure_detail["failed_session_count"],
sending_mta_ip=sending_mta_ip,
receiving_mx_hostname=receiving_mx_hostname,
additional_information_uri=additional_information_uri,
failure_reason_code=failure_reason_code,
)
smtp_tls_doc.failure_details_combined.append(
"{} / {} / {} / {} / {} / {}".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)
smtp_tls_doc.meta.index = index
try:
smtp_tls_doc.save()
except Exception as e:
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
# Backward-compatible aliases
_ForensicSampleDoc = _FailureSampleDoc
_ForensicReportDoc = _FailureReportDoc
save_forensic_report_to_opensearch = save_failure_report_to_opensearch