Verify SPF, DKIM, and DMARC Records

id: email-deliverability-dns-check
namespace: company.team
description: |
  Verify SPF, DKIM selector, and DMARC records for a sending domain, and alert
  Slack when any of them is missing or broken - before the quarterly outage.

triggers:
  - id: weekly_mail_dns_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly check so a DNS edit never silently kills deliverability.
    cron: "0 6 * * 2"
    disabled: true

inputs:
  - id: domain
    type: STRING
    defaults: "example.com"
    description: Sending domain to check.
  - id: dkim_selector
    type: STRING
    defaults: "default"
    description: Selector to probe for the DKIM TXT record.

tasks:
  - id: check_records
    type: io.kestra.plugin.scripts.shell.Commands
    description: dig TXT at the apex (SPF, DMARC) and at the selector (DKIM), then
      emit booleans via the stdout outputs protocol. Lookup failure reports
      missing so it cannot read as healthy.
    containerImage: alpine:3.20
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    commands:
      - |
        apk add --no-cache bind-tools python3 >/dev/null 2>&1
        cat > check.py <<'PYEOF'
        import os, json, subprocess
        domain = os.environ.get("DOMAIN", "example.com")
        selector = os.environ.get("SELECTOR", "default")
        def txt(name):
            try:
                out = subprocess.run(["dig", "+short", "TXT", name], capture_output=True, text=True, timeout=20).stdout.lower()
                return out
            except Exception:
                return ""
        apex = txt(domain)
        dmarc = txt(f"_dmarc.{domain}")
        dkim = txt(f"{selector}._domainkey.{domain}")
        spf = "v=spf1" in apex
        dmarc_ok = "v=dmarc1" in dmarc
        dkim_ok = "v=dkim1" in dkim or "p=" in dkim
        missing = [n for n, ok in [("SPF", spf), ("DMARC", dmarc_ok), ("DKIM", dkim_ok)] if not ok]
        print(f"SPF {spf}, DMARC {dmarc_ok}, DKIM {dkim_ok}")
        print("::" + json.dumps({"outputs": {"spf_ok": spf, "dmarc_ok": dmarc_ok, "dkim_ok": dkim_ok, "missing_count": len(missing), "missing": ",".join(missing)}}) + "::")
        PYEOF
        python3 check.py
    env:
      DOMAIN: "{{ inputs.domain }}"
      SELECTOR: "{{ inputs.dkim_selector }}"

  - id: gaps_found
    type: io.kestra.plugin.core.flow.If
    description: Any missing record is an alert; a full set is just a log line.
    condition: "{{ outputs.check_records.vars.missing_count > 0 }}"
    then:
      - id: alert_gaps
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Say exactly which records are missing so the DNS fix starts there.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":email: Deliverability check on {{ inputs.domain }}: missing {{ outputs.check_records.vars.missing }}. Outgoing mail may land in spam or be rejected. Execution {{ execution.id }}."
          }
    else:
      - id: log_ok
        type: io.kestra.plugin.core.log.Log
        description: Record the passing check for the trend.
        message: "SPF, DMARC, and DKIM all present for {{ inputs.domain }}."

errors:
  - id: alert_check_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the check itself fails - a broken resolver must not read
      as compliant.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Deliverability DNS check FAILED in flow {{ flow.id }} (execution {{ execution.id }}). Check dig availability and network egress."
      }

outputs:
  - id: records_status
    type: JSON
    description: 'Record health, e.g. {"spf_ok": true, "dmarc_ok": true, "dkim_ok": false}'
    value: '{{ {"spf_ok": outputs.check_records.vars.spf_ok, "dmarc_ok":
      outputs.check_records.vars.dmarc_ok, "dkim_ok":
      outputs.check_records.vars.dkim_ok} | toJson }}'