# -*- 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,
NotFoundError,
Object,
OpenSearch,
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, redact_url_credentials
[docs]
class OpenSearchError(Exception):
"""Raised when an OpenSearch error occurs"""
# Names of the data streams whose index template and stream
# ``create_data_streams`` has already set up in this process, so each one
# costs its API calls once per process rather than once per saved document.
# ``set_hosts`` clears it, so a new connection (including a SIGHUP reload,
# which may point at another cluster or change the index settings)
# re-applies the template. Forgetting an entry is always safe: it only
# repeats idempotent calls.
_READY_DATA_STREAMS: set[str] = set()
# Priority of the index template ``create_data_streams`` installs. OpenSearch
# itself installs no index templates, but plugins may; 501 stays above the
# usual priority-100 convention and matches the Elasticsearch module. The
# template's ``index_patterns`` is the exact stream name, so an overlap with
# another template is unlikely in the first place.
_DATA_STREAM_TEMPLATE_PRIORITY = 501
# 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",
) -> OpenSearch:
"""
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)
Returns:
OpenSearch: The client registered under the ``default`` connection
alias. Callers that need to close this exact client later (rather than
whatever holds the alias at that point) should hold on to it.
"""
_READY_DATA_STREAMS.clear()
if not isinstance(hosts, list):
hosts = [hosts]
logger.debug(
"Connecting to OpenSearch: hosts=%s, use_ssl=%s",
redact_url_credentials(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'."
)
return 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__()}")
[docs]
def create_data_streams(names: list[str], settings: dict[str, Any] | None = None):
"""
Create OpenSearch data streams and their index templates
For each name, puts an index template named after the stream whose
``index_patterns`` is exactly that name -- no wildcard, so it never
applies to the date-suffixed indexes parsedmarc writes without data
streams -- and then creates the data stream if it does not exist yet.
parsedmarc puts its template again in every process that writes to the
stream, so edits made to it directly do not last; an ISM policy attaches
itself to the stream's backing indices through its own
``ism_template`` instead. Each name is set up once per process;
``set_hosts`` resets that.
The template maps only ``@timestamp``. Like ``create_indexes``, it
deliberately installs no mappings for the report fields themselves (see
issue #169), so they keep their dynamic ``object`` mapping.
Args:
names (list): A list of data stream names
settings (dict): Index settings for the stream's backing indices;
one shard and no replicas when ``None``, as in
``create_indexes``.
Raises:
OpenSearchError: A concrete index or alias already holds one of the
names, or the cluster returned an error. A name that fails is
not remembered, so the next save retries it.
"""
if settings is None:
effective_settings: dict[str, Any] = {
"number_of_shards": 1,
"number_of_replicas": 0,
}
else:
effective_settings = dict(settings)
for name in names:
if name in _READY_DATA_STREAMS:
continue
try:
client = connections.get_connection()
# OpenSearch 3 answers both a missing name and a name held by a
# concrete index with a 404. Elasticsearch answers the latter
# with 200 and an empty list, so, as in the Elasticsearch module,
# look for the stream in the response rather than trust the
# status alone.
try:
response = client.indices.get_data_stream(name=name)
stream_exists = any(
stream.get("name") == name
for stream in response.get("data_streams", [])
)
except NotFoundError:
stream_exists = False
# Checked before anything is written, so a name collision leaves
# no template behind.
collision = not stream_exists and Index(name).exists()
if not collision:
client.indices.put_index_template(
name=name,
body={
"index_patterns": [name],
"data_stream": {},
"priority": _DATA_STREAM_TEMPLATE_PRIORITY,
"template": {
"settings": effective_settings,
"mappings": {
"properties": {"@timestamp": {"type": "date"}}
},
},
"_meta": {
"description": (
"Managed by parsedmarc, which overwrites it. "
"Attach ISM policies with an ism_template."
)
},
},
)
if not stream_exists:
logger.debug(f"Creating OpenSearch data stream: {name}")
client.indices.create_data_stream(name=name)
except Exception as e:
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
if collision:
raise OpenSearchError(
f"OpenSearch error: cannot create the data stream {name}: "
"an index or alias with that name already exists"
)
_READY_DATA_STREAMS.add(name)
_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``.
All three work unchanged with ``use_data_streams``: a data stream
carries the undated name itself, so the backfill patterns match it as
well as any dated indexes, and a stream that holds that exact name is
skipped by the ``published_policy.fo`` migration, because the backing
index it inspects maps the field as ``text``, not ``long``.
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,
use_data_streams: bool = False,
):
"""
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
(ignored when ``use_data_streams`` is True)
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
use_data_streams (bool): Save to a data stream named like the index,
without the date suffix, instead of to a dated index
Raises:
AlreadySaved
OpenSearchError
"""
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)
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}"
# The trailing wildcard matches the data stream as well as the dated
# indexes, so duplicates are caught across a switch between the two.
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"]
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"),
**{"@timestamp": begin_date},
)
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_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
if use_data_streams:
create_data_streams([index], index_settings)
else:
if monthly_indexes:
index_date = begin_date.strftime("%Y-%m")
else:
index_date = begin_date.strftime("%Y-%m-%d")
index = f"{index}-{index_date}"
create_indexes([index], index_settings)
agg_doc.meta.index = index
try:
# Data streams accept only op_type=create. No _id is set, so
# for a dated index it means the same as the default.
agg_doc.save(op_type="create")
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,
use_data_streams: bool = False,
):
"""
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 (ignored when ``use_data_streams``
is True)
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
use_data_streams (bool): Save to a data stream named like the
index, without the date suffix, instead
of to a dated index
Raises:
AlreadySaved
OpenSearchError
InvalidFailureReport
"""
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(",")
)
# The trailing wildcards match the data stream as well as the dated
# indexes, so duplicates are caught across a switch between the two.
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,
**{"@timestamp": arrival_date},
)
index = "dmarc_failure"
if index_suffix:
index = f"{index}_{index_suffix}"
if index_prefix:
index = f"{index_prefix}{index}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
if use_data_streams:
create_data_streams([index], index_settings)
else:
if monthly_indexes:
index_date = arrival_date.strftime("%Y-%m")
else:
index_date = arrival_date.strftime("%Y-%m-%d")
index = f"{index}-{index_date}"
create_indexes([index], index_settings)
failure_doc.meta.index = index
try:
# Data streams accept only op_type=create. No _id is set, so
# for a dated index it means the same as the default.
failure_doc.save(op_type="create")
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,
use_data_streams: bool = False,
):
"""
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
(ignored when ``use_data_streams`` is True)
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
use_data_streams (bool): Save to a data stream named like the index,
without the date suffix, instead of to a dated index
Raises:
AlreadySaved
OpenSearchError
"""
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")
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}"
# The trailing wildcard matches the data stream as well as the dated
# indexes, so duplicates are caught across a switch between the two.
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}"
if not use_data_streams:
if monthly_indexes:
index_date = begin_date.strftime("%Y-%m")
else:
index_date = begin_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
)
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"],
**{"@timestamp": report["begin_date"]},
)
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)
if use_data_streams:
create_data_streams([index], index_settings)
else:
create_indexes([index], index_settings)
smtp_tls_doc.meta.index = index
try:
# Data streams accept only op_type=create. No _id is set, so for a
# dated index it means the same as the default.
smtp_tls_doc.save(op_type="create")
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