parsedmarc/docs/source/elasticsearch.md

Elasticsearch and Kibana

To set up visual dashboards of DMARC data, install Elasticsearch and Kibana.

:::{note} Elasticsearch and Kibana 8 or later are required (parsedmarc's 8.x Python client also supports Elasticsearch 9). OpenSearch users must use the [opensearch] configuration section instead — the Elasticsearch 8.x client refuses to connect to non-Elasticsearch clusters. :::

Installation

On Debian/Ubuntu based systems, run:

sudo apt-get install -y apt-transport-https
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --dearmor -o /usr/share/keyrings/elasticsearch-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/elasticsearch-keyring.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-8.x.list
sudo apt-get update
sudo apt-get install -y elasticsearch kibana

For CentOS, RHEL, and other RPM systems, follow the Elastic RPM guides for Elasticsearch and Kibana.

:::{note} Previously, the default JVM heap size for Elasticsearch was very small (1g), which will cause it to crash under a heavy load. To fix this, increase the minimum and maximum JVM heap sizes in /etc/elasticsearch/jvm.options to more reasonable levels, depending on your server's resources.

Make sure the system has at least 2 GB more RAM than the assigned JVM heap size.

Always set the minimum and maximum JVM heap sizes to the same value.

For example, to set a 4 GB heap size, set

-Xms4g
-Xmx4g

See https://www.elastic.co/guide/en/elasticsearch/reference/current/important-settings.html#heap-size-settings for more information. :::

sudo systemctl daemon-reload
sudo systemctl enable elasticsearch.service
sudo systemctl enable kibana.service
sudo systemctl start elasticsearch.service
sudo systemctl start kibana.service

Since Elasticsearch 8.0, security is enabled and auto-configured on first startup: TLS certificates are generated, the xpack.security.* settings below are written to elasticsearch.yml, and a password is generated for the elastic user. Verify the settings are present — and add them only if your install skipped auto-configuration:

sudo vim /etc/elasticsearch/elasticsearch.yml

The security configuration looks like this:

# Enable security features
xpack.security.enabled: true
xpack.security.enrollment.enabled: true
# Enable encryption for HTTP API client connections, such as Kibana, Logstash, and Agents
xpack.security.http.ssl:
  enabled: true
  keystore.path: certs/http.p12
# Enable encryption and mutual authentication between cluster nodes
xpack.security.transport.ssl:
  enabled: true
  verification_mode: certificate
  keystore.path: certs/transport.p12
  truststore.path: certs/transport.p12

sudo systemctl restart elasticsearch

To create a self-signed certificate, run:

openssl req -x509 -nodes -days 365 -newkey rsa:4096 -keyout kibana.key -out kibana.crt

Or, to create a Certificate Signing Request (CSR) for a CA, run:

openssl req -newkey rsa:4096 -nodes -keyout kibana.key -out kibana.csr

Fill in the prompts. Watch out for Common Name (e.g. server FQDN or YOUR domain name), which is the IP address or domain name that you will use to access Kibana. It is the most important field.

If you generated a CSR, remove the CSR after you have your certs

rm -f kibana.csr

Move the keys into place and secure them:

sudo mv kibana.* /etc/kibana
sudo chmod 660 /etc/kibana/kibana.key

Activate the HTTPS server in Kibana

sudo vim /etc/kibana/kibana.yml

Add the following configuration

server.host: "SERVER_IP"
server.publicBaseUrl: "https://SERVER_IP"
server.ssl.enabled: true
server.ssl.certificate: /etc/kibana/kibana.crt
server.ssl.key: /etc/kibana/kibana.key

:::{note} For more security, you can configure Kibana to use a local network connection to Elasticsearch:

elasticsearch.hosts: ['https://SERVER_IP:9200']

=>

elasticsearch.hosts: ['https://127.0.0.1:9200']

:::

sudo systemctl restart kibana

Enroll Kibana in Elasticsearch

sudo /usr/share/elasticsearch/bin/elasticsearch-create-enrollment-token -s kibana

Then access your web server at https://SERVER_IP:5601, accept the self-signed certificate, and paste the token in the "Enrollment token" field.

sudo /usr/share/kibana/bin/kibana-verification-code

Then enter the verification code in your web browser.

End Kibana configuration

sudo /usr/share/elasticsearch/bin/elasticsearch-setup-passwords interactive
sudo /usr/share/kibana/bin/kibana-encryption-keys generate
sudo vim /etc/kibana/kibana.yml

Add previously generated encryption keys

xpack.encryptedSavedObjects.encryptionKey: xxxx...xxxx
xpack.reporting.encryptionKey: xxxx...xxxx
xpack.security.encryptionKey: xxxx...xxxx

sudo systemctl restart kibana
sudo systemctl restart elasticsearch

Now that Elasticsearch is up and running, use parsedmarc to send data to it.

Download (right-click the link and click save as) opensearch_dashboards.ndjson.

Connect to Kibana using the "elastic" user and the password you previously provided on the console ("End Kibana configuration" part).

Import opensearch_dashboards.ndjson in the Saved Objects tab of the Stack Management page of Kibana. (Hamburger menu -> "Management" -> "Stack Management" -> "Kibana" -> "Saved Objects")

It will give you the option to overwrite existing saved dashboards or visualizations, which could be used to restore them if you or someone else breaks them, as there are no permissions/access controls in Kibana without the commercial X-Pack.

:align: center
:alt: A screenshot of the Saved Objects Stack Management UI in Kibana
:target: _static/screenshots/saved-objects.png

:align: center
:alt: A screenshot of the overwrite confirmation prompt
:target: _static/screenshots/confirm-overwrite.png

Upgrading Kibana index patterns

parsedmarc 5.0.0 makes some changes to the way data is indexed in Elasticsearch. If you are upgrading from a previous release of parsedmarc, you need to complete the following steps to replace the Kibana index patterns with versions that match the upgraded indexes:

1. Login in to Kibana, and click on Management
2. Under Kibana, click on Saved Objects
3. Check the checkboxes for the dmarc_aggregate and dmarc_failure index patterns
4. Click Delete
5. Click Delete on the confirmation message
6. Download (right-click the link and click save as) the latest version of opensearch_dashboards.ndjson
7. Import opensearch_dashboards.ndjson by clicking Import from the Kibana Saved Objects page

Backfilling the combined DKIM/SPF result fields

As of the version fixing #169, aggregate documents include dkim_results_combined and spf_results_combined — scalar string arrays that keep each auth result's selector/scope, domain, and result paired, which the dashboards' alignment-detail tables aggregate on. Reports saved by older versions lack these fields and will not appear in those tables.

parsedmarc now backfills this automatically. On startup, it runs a cheap count query against each configured aggregate index pattern to check for documents that have DKIM or SPF results but are missing the corresponding combined field. If any are found, it submits the backfill as a background _update_by_query task (wait_for_completion=false), so startup is never blocked on it; progress is logged, including the task ID. The check itself is idempotent — once an index is fully backfilled, later startups see a count of 0 and log nothing further — and it works the same way on OpenSearch. Any error talking to the cluster (for example, no indexes yet on a fresh install) is logged as a warning and retried on the next startup, rather than aborting parsedmarc.

Which index patterns it targets follows the ones parsedmarc writes to, and is logged at debug level on startup:

• With index_prefix_domain_map configured in [general] and no index_prefix set, every tenant prefix in the map gets its own index pattern, alongside the unprefixed one — aggregate and failure reports for a domain that is not in the map are still saved without a prefix. A configuration reload (SIGHUP) re-reads the map, so a newly onboarded tenant is covered without a restart.
• With an index_prefix set in [elasticsearch]/[opensearch], only that prefix is targeted, and the map is not consulted. That is deliberate: such a deployment writes only under its own prefix, and an _update_by_query against a pattern it does not write to could reach another deployment's data on a shared cluster.
• With an index_suffix set, both the suffixed and the unsuffixed pattern are targeted, so documents indexed before the suffix was configured are backfilled too. Note that the unsuffixed pattern also matches any other suffix on the same cluster.

If you upgrade the dashboards without pointing the new parsedmarc version at the cluster, or you'd rather control when the write load happens, you can still run the backfill manually. It is idempotent (documents that already have the fields are skipped), so it is safe to re-run. It works identically on OpenSearch; just adjust the URL and credentials. The query matches only documents that have at least one DKIM or SPF auth result and lack the corresponding combined field; documents with no auth results are skipped, because an exists query cannot see an empty array, and for search purposes an empty dkim_results_combined is identical to an absent one. Each result is matched on either its domain or its result subfield as defense in depth: an empty string indexes no text tokens and is invisible to exists, and the storage shape of every historical parsedmarc version can't be audited, so matching either subfield ensures no backfillable document is skipped.

curl -X POST "http://localhost:9200/dmarc_aggregate*/_update_by_query?conflicts=proceed&wait_for_completion=false" \
  -H "Content-Type: application/json" -d '
{
  "query": {
    "bool": {
      "minimum_should_match": 1,
      "should": [
        {
          "bool": {
            "must": [
              {
                "bool": {
                  "minimum_should_match": 1,
                  "should": [
                    {"exists": {"field": "dkim_results.domain"}},
                    {"exists": {"field": "dkim_results.result"}}
                  ]
                }
              }
            ],
            "must_not": [{"exists": {"field": "dkim_results_combined"}}]
          }
        },
        {
          "bool": {
            "must": [
              {
                "bool": {
                  "minimum_should_match": 1,
                  "should": [
                    {"exists": {"field": "spf_results.domain"}},
                    {"exists": {"field": "spf_results.result"}}
                  ]
                }
              }
            ],
            "must_not": [{"exists": {"field": "spf_results_combined"}}]
          }
        }
      ]
    }
  },
  "script": {
    "lang": "painless",
    "source": "List dk = new ArrayList(); def dr = ctx._source.dkim_results; if (dr != null) { if (!(dr instanceof List)) { dr = [dr]; } for (e in dr) { if (e == null) { continue; } def sel = e.selector != null ? e.selector : \"none\"; def dom = e.domain != null ? e.domain : \"none\"; def res = e.result != null ? e.result : \"none\"; dk.add(sel + \" / \" + dom + \" / \" + res); } } ctx._source.dkim_results_combined = dk; List sp = new ArrayList(); def sr = ctx._source.spf_results; if (sr != null) { if (!(sr instanceof List)) { sr = [sr]; } for (e in sr) { if (e == null) { continue; } def sc = e.scope != null ? e.scope : \"mfrom\"; def dom = e.domain != null ? e.domain : \"none\"; def res = e.result != null ? e.result : (e.results != null ? e.results : \"none\"); sp.add(sc + \" / \" + dom + \" / \" + res); } } ctx._source.spf_results_combined = sp;"
  }
}'

wait_for_completion=false returns a task ID — check progress with GET _tasks/<task id>. Adjust the index pattern if you use a custom index_prefix/index_suffix; with index_prefix_domain_map, run the command once per tenant prefix (acme_corp_dmarc_aggregate*) plus once for the unprefixed pattern, or widen it to *dmarc_aggregate* to cover every tenant in one pass. After backfilling, re-import the updated dashboards ndjson (the index pattern saved object changed too) per the import instructions above.

SMTP TLS documents have the same class of defect one level deeper: policies is an object array, and each policy's failure_details is an object array inside it. SMTP TLS documents now also carry policies_combined and failure_details_combined, backfilled automatically at startup the same way, and the equivalent manual command is:

curl -X POST "http://localhost:9200/smtp_tls*/_update_by_query?conflicts=proceed&wait_for_completion=false" \
  -H "Content-Type: application/json" -d '
{
  "query": {
    "bool": {
      "minimum_should_match": 1,
      "should": [
        {
          "bool": {
            "must": [
              {
                "bool": {
                  "minimum_should_match": 1,
                  "should": [
                    {"exists": {"field": "policies.policy_domain"}},
                    {"exists": {"field": "policies.policy_type"}}
                  ]
                }
              }
            ],
            "must_not": [{"exists": {"field": "policies_combined"}}]
          }
        },
        {
          "bool": {
            "must": [
              {
                "bool": {
                  "minimum_should_match": 1,
                  "should": [
                    {"exists": {"field": "policies.failure_details.result_type"}},
                    {"exists": {"field": "policies.failure_details.sending_mta_ip"}}
                  ]
                }
              }
            ],
            "must_not": [{"exists": {"field": "failure_details_combined"}}]
          }
        }
      ]
    }
  },
  "script": {
    "lang": "painless",
    "source": "List pols = new ArrayList(); List dets = new ArrayList(); def ps = ctx._source.policies; if (ps != null) { if (!(ps instanceof List)) { ps = [ps]; } for (p in ps) { if (p == null) { continue; } def dom = p.policy_domain != null ? p.policy_domain : \"none\"; def typ = p.policy_type != null ? p.policy_type : \"none\"; pols.add(dom + \" / \" + typ); def fds = p.failure_details; if (fds != null) { if (!(fds instanceof List)) { fds = [fds]; } for (f in fds) { if (f == null) { continue; } def rt = f.result_type != null ? f.result_type : \"none\"; def smi = f.sending_mta_ip != null ? f.sending_mta_ip : \"none\"; def ri = f.receiving_ip != null ? f.receiving_ip : \"none\"; def rmh = f.receiving_mx_hostname != null ? f.receiving_mx_hostname : \"none\"; dets.add(dom + \" / \" + typ + \" / \" + rt + \" / \" + smi + \" / \" + ri + \" / \" + rmh); } } } } ctx._source.policies_combined = pols; ctx._source.failure_details_combined = dets;"
  }
}'

It works identically on OpenSearch; just adjust the URL and credentials, same as the aggregate command above.

Data streams and records retention

Starting in version 5.0.0, parsedmarc stores data in a separate index for each day (or each month, with monthly_indexes), named after the report's date, such as dmarc_aggregate-2024-03-31. That makes it easy to comply with records retention regulations such as GDPR: delete the indexes for the dates you no longer keep. For more information, check out the Elastic guide to managing time-based indexes efficiently. A single aggregate report creates at most 31 dated indexes (MAX_INDEXES_PER_REPORT): when its records would use more than 31 dated indexes, the later records are saved to the report's first dated index with their own dates, and a warning is logged. Dated index names still come from each report's own dates, so reports with many different dates still create one index per date; data streams (use_data_streams) are affected by neither. A report whose begin or end date (aggregate), begin date (SMTP TLS) or Arrival-Date (failure) is more than two days in the future is rejected as invalid, because those dates name the dated indexes; past dates are not bounded.

Set use_data_streams = True in the [elasticsearch] section to write to data streams instead. Each report type then goes to one data stream named like the indexes without the date: dmarc_aggregate, dmarc_failure, and smtp_tls, with index_prefix, index_suffix, or the index_prefix_domain_map prefix applied as usual. The cluster creates the stream's backing indices (.ds-dmarc_aggregate-…), and an ILM policy you attach rolls them over and deletes old ones, so there is one stable name per report type and far fewer shards.

Before a parsedmarc process first saves to a stream, it puts an index template named after the stream and creates the stream if it does not exist yet. The template:

• matches exactly the stream's name, with no wildcard, so it never applies to the dated indexes;
• has priority 501 and sets number_of_shards and number_of_replicas from the configuration, for each new backing index;
• maps @timestamp as a date. Every document parsedmarc saves carries @timestamp: each record's begin date (date_begin) for aggregate reports, which is the report's begin date unless parsedmarc split a report spanning more than a day into daily records; the arrival date for failure reports; and the report's begin date for SMTP TLS reports. The other fields get the same field mappings as the dated indexes;
• includes a component template named <stream>@custom, such as dmarc_aggregate@custom, if one exists.

Data streams need Elasticsearch 8.7 or later, because the template uses ignore_missing_component_templates to make the @custom component template optional.

parsedmarc puts the template again in every process, so changes made to it directly do not last. Put your own settings, such as the ILM policy, in the @custom component template instead.

parsedmarc does not create or change any ILM policy. Here is an example that rolls each stream over every 30 days, or sooner when a primary shard reaches 50 GB, and deletes a backing index 180 days after its rollover:

PUT _ilm/policy/parsedmarc
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_age": "30d",
            "max_primary_shard_size": "50gb"
          }
        }
      },
      "delete": {
        "min_age": "180d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

Attach it to each stream through its @custom component template:

PUT _component_template/dmarc_aggregate@custom
{
  "template": {
    "settings": {
      "index.lifecycle.name": "parsedmarc"
    }
  }
}

Repeat that for dmarc_failure@custom and smtp_tls@custom, and for the prefixed or suffixed names if you use them. The component template applies only to backing indices created after it. A stream that already exists has a write index with no policy, so nothing would ever roll it over; attach the policy to its existing backing indices as well:

PUT dmarc_aggregate/_settings
{
  "index.lifecycle.name": "parsedmarc"
}

The delete phase's min_age counts from the rollover, not from the report dates, so a backing index is deleted 180 days after it stopped receiving documents, whatever dates the reports in it cover. If you need retention by report date, keep using dated indexes and delete them by name.

On Elastic Cloud Serverless, which has no ILM, use a data stream lifecycle instead:

PUT _data_stream/dmarc_aggregate/_lifecycle
{
  "data_retention": "180d"
}

Field mappings

Indexes and data stream templates created by parsedmarc 11.2.2 or later turn date detection off and declare the date fields (date_range, date_begin, date_end, arrival_date, @timestamp, and sample.date), so a report cannot make a text field such as report_id a date and block every later report for that index. In failure reports, only these sample headers are indexed: authentication-results, auto-submitted, cc, content-transfer-encoding, content-type, date, from, in-reply-to, message-id, mime-version, received, references, reply-to, return-path, subject, thread-index, thread-topic, to, x-auto-response-suppress, x-exclaimer-md-config, x-linkedin-fe, x-mailer, x-ms-exchange-generated-message-source, x-ms-exchange-inbox-rules-loop, x-ms-exchange-parent-message-id, x-ms-exchange-transport-fromentityheader, x-ms-has-attach, x-ms-tnef-correlator, and x-onpremexternalip. Every header is still kept in _source under its lower-cased name, with any . replaced by _, except a From, To or Reply-To header that holds no address, which is left out of the document. To make another header searchable, declare it under sample.headers.properties as text with a keyword sub-field: in the dmarc_failure@custom component template for a data stream, or in an index template whose index_patterns match the dated indexes (dmarc_failure-*). Either applies only to indexes created after it. Dated indexes created by earlier releases keep their mappings unchanged. parsedmarc puts the data stream template again in every process before its first save to the stream, so after upgrading, an existing stream's next backing index after a rollover gets these mappings too.

Switching an existing installation to data streams

Dated indexes that already exist are left alone. They keep working next to the streams: the shipped dashboards use the dmarc_aggregate*, dmarc_f*, and smtp_tls* index patterns, which match both, and the check for duplicate reports searches both, so a report that is already in a dated index is not saved again into a stream. You can delete the dated indexes when they age out of your retention period, or move their documents into the streams. Documents saved by earlier releases have no @timestamp, which a data stream requires, so add it while reindexing. Run parsedmarc once with use_data_streams = True first, so the stream and its template exist; a reindex into a name that is not a data stream yet creates an ordinary index of that name instead:

POST _reindex
{
  "source": {"index": "dmarc_aggregate-*"},
  "dest": {"index": "dmarc_aggregate", "op_type": "create"},
  "script": {
    "lang": "painless",
    "source": "ctx._source['@timestamp'] = ctx._source.date_begin"
  }
}

Use date_begin for smtp_tls-* too, and arrival_date for dmarc_failure-*. Delete the dated indexes once the reindex has finished; until then, the dashboards count every document twice.

If saving fails with cannot create the data stream dmarc_aggregate: an index or alias with that name already exists, an index or alias already uses the stream's name. parsedmarc releases before 5.0.0 wrote to an undated dmarc_aggregate index, for example. parsedmarc refuses to write into such an index rather than mixing new documents into it. Move its documents to a dated name, which the dashboards still match, and delete it:

POST _reindex
{
  "source": {"index": "dmarc_aggregate"},
  "dest": {"index": "dmarc_aggregate-legacy"}
}

DELETE dmarc_aggregate