filter {

  # ==========================================================================
  # parsedmarc Google SecOps Parser
  #
  # Parses JSON syslog events produced by parsedmarc
  # (https://domainaware.github.io/parsedmarc/)
  #
  # parsedmarc outputs three types of reports as single-line JSON via syslog:
  #   1. DMARC Aggregate Reports  (contains "header_from" and "dmarc_aligned")
  #   2. DMARC Forensic Reports   (contains "feedback_type")
  #   3. SMTP TLS Reports         (contains "policy_type")
  #
  # UDM Reference:
  #   https://cloud.google.com/chronicle/docs/reference/udm-field-list
  # ==========================================================================

  # Initialize variables
  mutate {
    replace => {
      "event_type" => "EMAIL_UNCATEGORIZED"
      "report_type" => ""
    }
  }

  # Parse the JSON payload from the syslog message
  json {
    source => "message"
    array_function => "split_columns"
    on_error => "not_json"
  }

  # Abort if not valid JSON
  if [not_json] {
    drop {}
  }

  # --------------------------------------------------------------------------
  # Detect report type
  # --------------------------------------------------------------------------

  # Aggregate reports contain "header_from" and "dmarc_aligned"
  if [header_from] {
    mutate {
      replace => {
        "report_type" => "aggregate"
      }
    }
  }

  # Forensic reports contain "feedback_type"
  if [feedback_type] {
    mutate {
      replace => {
        "report_type" => "forensic"
      }
    }
  }

  # SMTP TLS reports contain "policy_type"
  if [policy_type] {
    mutate {
      replace => {
        "report_type" => "smtp_tls"
      }
    }
  }

  # ==========================================================================
  # DMARC Aggregate Report Parsing
  # ==========================================================================
  if [report_type] == "aggregate" {

    mutate {
      replace => {
        "event_type" => "EMAIL_TRANSACTION"
      }
    }

    # -- metadata --
    if [report_id] {
      mutate {
        replace => {
          "event.idm.read_only_udm.metadata.product_log_id" => "%{report_id}"
        }
      }
    }

    if [org_name] {
      mutate {
        replace => {
          "org_name_label.key" => "org_name"
          "org_name_label.value.string_value" => "%{org_name}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "org_name_label"
        }
      }
    }

    if [org_email] {
      mutate {
        replace => {
          "org_email_label.key" => "org_email"
          "org_email_label.value.string_value" => "%{org_email}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "org_email_label"
        }
      }
    }

    # -- timestamps --
    if [begin_date] {
      mutate {
        replace => {
          "begin_date_label.key" => "begin_date"
          "begin_date_label.value.string_value" => "%{begin_date}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "begin_date_label"
        }
      }
    }

    if [end_date] {
      mutate {
        replace => {
          "end_date_label.key" => "end_date"
          "end_date_label.value.string_value" => "%{end_date}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "end_date_label"
        }
      }
    }

    # -- principal (source of emails) --
    if [source_ip_address] {
      mutate {
        merge => {
          "event.idm.read_only_udm.principal.ip" => "source_ip_address"
        }
        on_error => "merge_ip_error"
      }
    }

    if [source_reverse_dns] {
      mutate {
        replace => {
          "event.idm.read_only_udm.principal.hostname" => "%{source_reverse_dns}"
        }
      }
    }

    if [source_country] {
      mutate {
        replace => {
          "event.idm.read_only_udm.principal.location.country_or_region" => "%{source_country}"
        }
      }
    }

    if [source_base_domain] {
      mutate {
        replace => {
          "src_base_domain_label.key" => "source_base_domain"
          "src_base_domain_label.value.string_value" => "%{source_base_domain}"
        }
        merge => {
          "event.idm.read_only_udm.principal.labels" => "src_base_domain_label"
        }
      }
    }

    if [source_name] {
      mutate {
        replace => {
          "src_name_label.key" => "source_name"
          "src_name_label.value.string_value" => "%{source_name}"
        }
        merge => {
          "event.idm.read_only_udm.principal.labels" => "src_name_label"
        }
      }
    }

    if [source_type] {
      mutate {
        replace => {
          "src_type_label.key" => "source_type"
          "src_type_label.value.string_value" => "%{source_type}"
        }
        merge => {
          "event.idm.read_only_udm.principal.labels" => "src_type_label"
        }
      }
    }

    # -- email fields --
    if [header_from] {
      mutate {
        replace => {
          "event.idm.read_only_udm.network.email.from" => "%{header_from}"
        }
      }
    }

    if [envelope_from] {
      mutate {
        replace => {
          "event.idm.read_only_udm.network.email.mail_from" => "%{envelope_from}"
        }
      }
    }

    if [envelope_to] {
      mutate {
        replace => {
          "event.idm.read_only_udm.network.email.to" => "%{envelope_to}"
        }
        on_error => "envelope_to_error"
      }
    }

    # -- target (domain being reported on) --
    if [domain] {
      mutate {
        replace => {
          "event.idm.read_only_udm.target.hostname" => "%{domain}"
        }
      }
    }

    # -- message count --
    if [count] {
      mutate {
        replace => {
          "count_label.key" => "count"
          "count_label.value.string_value" => "%{count}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "count_label"
        }
      }
    }

    # -- security results: DMARC alignment and disposition --
    mutate {
      replace => {
        "security_result_dmarc.category" => "POLICY_VIOLATION"
        "security_result_dmarc.summary" => "DMARC Aggregate Report"
      }
    }

    # Map disposition to security_result.action
    if [disposition] == "none" {
      mutate {
        replace => {
          "security_result_dmarc.action" => "ALLOW"
        }
      }
    } else if [disposition] == "quarantine" {
      mutate {
        replace => {
          "security_result_dmarc.action" => "QUARANTINE"
        }
      }
    } else if [disposition] == "reject" {
      mutate {
        replace => {
          "security_result_dmarc.action" => "BLOCK"
        }
      }
    } else {
      mutate {
        replace => {
          "security_result_dmarc.action" => "UNKNOWN_ACTION"
        }
      }
    }

    # Build description of alignment results
    mutate {
      replace => {
        "security_result_dmarc.description" => "dmarc_aligned=%{dmarc_aligned} spf_aligned=%{spf_aligned} dkim_aligned=%{dkim_aligned} disposition=%{disposition}"
      }
      on_error => "desc_error"
    }

    # DMARC aligned: passed authentication, not a policy violation
    if [dmarc_aligned] == "True" or [dmarc_aligned] == "true" {
      mutate {
        replace => {
          "security_result_dmarc.category" => "NETWORK_CATEGORIZED_CONTENT"
        }
      }
    }

    # Merge security_result
    mutate {
      merge => {
        "event.idm.read_only_udm.security_result" => "security_result_dmarc"
      }
    }

    # -- DMARC policy fields --
    if [p] {
      mutate {
        replace => {
          "p_label.key" => "dmarc_policy"
          "p_label.value.string_value" => "%{p}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "p_label"
        }
      }
    }

    if [sp] {
      mutate {
        replace => {
          "sp_label.key" => "dmarc_sub_policy"
          "sp_label.value.string_value" => "%{sp}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "sp_label"
        }
      }
    }

    if [pct] {
      mutate {
        replace => {
          "pct_label.key" => "dmarc_pct"
          "pct_label.value.string_value" => "%{pct}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "pct_label"
        }
      }
    }

    if [adkim] {
      mutate {
        replace => {
          "adkim_label.key" => "dkim_alignment_mode"
          "adkim_label.value.string_value" => "%{adkim}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "adkim_label"
        }
      }
    }

    if [aspf] {
      mutate {
        replace => {
          "aspf_label.key" => "spf_alignment_mode"
          "aspf_label.value.string_value" => "%{aspf}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "aspf_label"
        }
      }
    }

    # -- alignment detail fields --
    if [dmarc_aligned] {
      mutate {
        replace => {
          "dmarc_aligned_label.key" => "dmarc_aligned"
          "dmarc_aligned_label.value.string_value" => "%{dmarc_aligned}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "dmarc_aligned_label"
        }
      }
    }

    if [spf_aligned] {
      mutate {
        replace => {
          "spf_aligned_label.key" => "spf_aligned"
          "spf_aligned_label.value.string_value" => "%{spf_aligned}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "spf_aligned_label"
        }
      }
    }

    if [dkim_aligned] {
      mutate {
        replace => {
          "dkim_aligned_label.key" => "dkim_aligned"
          "dkim_aligned_label.value.string_value" => "%{dkim_aligned}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "dkim_aligned_label"
        }
      }
    }

    if [disposition] {
      mutate {
        replace => {
          "disposition_label.key" => "disposition"
          "disposition_label.value.string_value" => "%{disposition}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "disposition_label"
        }
      }
    }

    # -- DKIM auth detail fields --
    if [dkim_domains] {
      mutate {
        replace => {
          "dkim_domains_label.key" => "dkim_domains"
          "dkim_domains_label.value.string_value" => "%{dkim_domains}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "dkim_domains_label"
        }
      }
    }

    if [dkim_selectors] {
      mutate {
        replace => {
          "dkim_selectors_label.key" => "dkim_selectors"
          "dkim_selectors_label.value.string_value" => "%{dkim_selectors}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "dkim_selectors_label"
        }
      }
    }

    if [dkim_results] {
      mutate {
        replace => {
          "dkim_results_label.key" => "dkim_results"
          "dkim_results_label.value.string_value" => "%{dkim_results}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "dkim_results_label"
        }
      }
    }

    # -- SPF auth detail fields --
    if [spf_domains] {
      mutate {
        replace => {
          "spf_domains_label.key" => "spf_domains"
          "spf_domains_label.value.string_value" => "%{spf_domains}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "spf_domains_label"
        }
      }
    }

    if [spf_scopes] {
      mutate {
        replace => {
          "spf_scopes_label.key" => "spf_scopes"
          "spf_scopes_label.value.string_value" => "%{spf_scopes}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "spf_scopes_label"
        }
      }
    }

    if [spf_results] {
      mutate {
        replace => {
          "spf_results_label.key" => "spf_results"
          "spf_results_label.value.string_value" => "%{spf_results}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "spf_results_label"
        }
      }
    }

    # -- policy override fields --
    if [policy_override_reasons] and [policy_override_reasons] != "none" {
      mutate {
        replace => {
          "override_label.key" => "policy_override_reasons"
          "override_label.value.string_value" => "%{policy_override_reasons}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "override_label"
        }
      }
    }

    if [policy_override_comments] and [policy_override_comments] != "none" {
      mutate {
        replace => {
          "override_comments_label.key" => "policy_override_comments"
          "override_comments_label.value.string_value" => "%{policy_override_comments}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "override_comments_label"
        }
      }
    }
  }

  # ==========================================================================
  # DMARC Forensic Report Parsing
  # ==========================================================================
  if [report_type] == "forensic" {

    mutate {
      replace => {
        "event_type" => "EMAIL_TRANSACTION"
      }
    }

    # -- metadata --
    if [message_id] {
      mutate {
        replace => {
          "event.idm.read_only_udm.metadata.product_log_id" => "%{message_id}"
        }
      }
    }

    # -- timestamps --
    if [arrival_date_utc] {
      date {
        match => ["arrival_date_utc", "yyyy-MM-dd HH:mm:ss"]
        on_error => "date_parse_error"
      }
    }

    # -- principal (source of the email) --
    if [source_ip_address] {
      mutate {
        merge => {
          "event.idm.read_only_udm.principal.ip" => "source_ip_address"
        }
        on_error => "merge_forensic_ip_error"
      }
    }

    if [source_reverse_dns] {
      mutate {
        replace => {
          "event.idm.read_only_udm.principal.hostname" => "%{source_reverse_dns}"
        }
      }
    }

    if [source_country] {
      mutate {
        replace => {
          "event.idm.read_only_udm.principal.location.country_or_region" => "%{source_country}"
        }
      }
    }

    if [source_base_domain] {
      mutate {
        replace => {
          "fsrc_base_domain_label.key" => "source_base_domain"
          "fsrc_base_domain_label.value.string_value" => "%{source_base_domain}"
        }
        merge => {
          "event.idm.read_only_udm.principal.labels" => "fsrc_base_domain_label"
        }
      }
    }

    if [source_name] {
      mutate {
        replace => {
          "fsrc_name_label.key" => "source_name"
          "fsrc_name_label.value.string_value" => "%{source_name}"
        }
        merge => {
          "event.idm.read_only_udm.principal.labels" => "fsrc_name_label"
        }
      }
    }

    if [source_type] {
      mutate {
        replace => {
          "fsrc_type_label.key" => "source_type"
          "fsrc_type_label.value.string_value" => "%{source_type}"
        }
        merge => {
          "event.idm.read_only_udm.principal.labels" => "fsrc_type_label"
        }
      }
    }

    # -- email fields --
    if [original_mail_from] {
      mutate {
        replace => {
          "event.idm.read_only_udm.network.email.from" => "%{original_mail_from}"
        }
      }
    }

    if [original_rcpt_to] {
      mutate {
        replace => {
          "event.idm.read_only_udm.network.email.to" => "%{original_rcpt_to}"
        }
        on_error => "rcpt_to_error"
      }
    }

    if [subject] {
      mutate {
        replace => {
          "event.idm.read_only_udm.network.email.subject" => "%{subject}"
        }
      }
    }

    # -- target (reported domain) --
    if [reported_domain] {
      mutate {
        replace => {
          "event.idm.read_only_udm.target.hostname" => "%{reported_domain}"
        }
      }
    }

    # -- security result --
    mutate {
      replace => {
        "security_result_forensic.summary" => "DMARC Forensic Report"
      }
    }

    if [auth_failure] {
      mutate {
        replace => {
          "security_result_forensic.description" => "auth_failure=%{auth_failure}"
        }
      }
    }

    if [delivery_result] {
      mutate {
        replace => {
          "delivery_result_label.key" => "delivery_result"
          "delivery_result_label.value.string_value" => "%{delivery_result}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "delivery_result_label"
        }
      }
    }

    # Map auth_failure to security_result.category
    if [auth_failure] == "dmarc" {
      mutate {
        replace => {
          "security_result_forensic.category" => "POLICY_VIOLATION"
          "security_result_forensic.action" => "ALLOW"
        }
      }
    } else if [auth_failure] == "spf" {
      mutate {
        replace => {
          "security_result_forensic.category" => "POLICY_VIOLATION"
          "security_result_forensic.action" => "ALLOW"
        }
      }
    } else if [auth_failure] == "dkim" {
      mutate {
        replace => {
          "security_result_forensic.category" => "POLICY_VIOLATION"
          "security_result_forensic.action" => "ALLOW"
        }
      }
    } else {
      mutate {
        replace => {
          "security_result_forensic.category" => "POLICY_VIOLATION"
          "security_result_forensic.action" => "UNKNOWN_ACTION"
        }
      }
    }

    mutate {
      merge => {
        "event.idm.read_only_udm.security_result" => "security_result_forensic"
      }
    }

    # -- additional forensic fields --
    if [feedback_type] {
      mutate {
        replace => {
          "feedback_type_label.key" => "feedback_type"
          "feedback_type_label.value.string_value" => "%{feedback_type}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "feedback_type_label"
        }
      }
    }

    if [user_agent] {
      mutate {
        replace => {
          "user_agent_label.key" => "user_agent"
          "user_agent_label.value.string_value" => "%{user_agent}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "user_agent_label"
        }
      }
    }

    if [authentication_results] {
      mutate {
        replace => {
          "auth_results_label.key" => "authentication_results"
          "auth_results_label.value.string_value" => "%{authentication_results}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "auth_results_label"
        }
      }
    }

    if [authentication_mechanisms] {
      mutate {
        replace => {
          "auth_mechanisms_label.key" => "authentication_mechanisms"
          "auth_mechanisms_label.value.string_value" => "%{authentication_mechanisms}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "auth_mechanisms_label"
        }
      }
    }

    if [dkim_domain] {
      mutate {
        replace => {
          "f_dkim_domain_label.key" => "dkim_domain"
          "f_dkim_domain_label.value.string_value" => "%{dkim_domain}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "f_dkim_domain_label"
        }
      }
    }

    if [auth_failure] {
      mutate {
        replace => {
          "auth_failure_label.key" => "auth_failure"
          "auth_failure_label.value.string_value" => "%{auth_failure}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "auth_failure_label"
        }
      }
    }

    if [arrival_date] {
      mutate {
        replace => {
          "arrival_date_label.key" => "arrival_date"
          "arrival_date_label.value.string_value" => "%{arrival_date}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "arrival_date_label"
        }
      }
    }
  }

  # ==========================================================================
  # SMTP TLS Report Parsing
  # ==========================================================================
  if [report_type] == "smtp_tls" {

    mutate {
      replace => {
        "event_type" => "GENERIC_EVENT"
      }
    }

    # -- metadata --
    if [report_id] {
      mutate {
        replace => {
          "event.idm.read_only_udm.metadata.product_log_id" => "%{report_id}"
        }
      }
    }

    if [organization_name] {
      mutate {
        replace => {
          "tls_org_label.key" => "organization_name"
          "tls_org_label.value.string_value" => "%{organization_name}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "tls_org_label"
        }
      }
    }

    # -- timestamps --
    if [begin_date] {
      mutate {
        replace => {
          "tls_begin_label.key" => "begin_date"
          "tls_begin_label.value.string_value" => "%{begin_date}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "tls_begin_label"
        }
      }
    }

    if [end_date] {
      mutate {
        replace => {
          "tls_end_label.key" => "end_date"
          "tls_end_label.value.string_value" => "%{end_date}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "tls_end_label"
        }
      }
    }

    # -- principal (sending MTA) --
    if [sending_mta_ip] {
      mutate {
        merge => {
          "event.idm.read_only_udm.principal.ip" => "sending_mta_ip"
        }
        on_error => "merge_tls_sending_ip_error"
      }
    }

    # -- target (receiving MTA) --
    if [receiving_ip] {
      mutate {
        merge => {
          "event.idm.read_only_udm.target.ip" => "receiving_ip"
        }
        on_error => "merge_tls_receiving_ip_error"
      }
    }

    if [receiving_mx_hostname] {
      mutate {
        replace => {
          "event.idm.read_only_udm.target.hostname" => "%{receiving_mx_hostname}"
        }
      }
    }

    # -- policy fields --
    if [policy_domain] {
      mutate {
        replace => {
          "policy_domain_label.key" => "policy_domain"
          "policy_domain_label.value.string_value" => "%{policy_domain}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "policy_domain_label"
        }
      }
    }

    if [policy_type] {
      mutate {
        replace => {
          "policy_type_label.key" => "policy_type"
          "policy_type_label.value.string_value" => "%{policy_type}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "policy_type_label"
        }
      }
    }

    if [policy_strings] {
      mutate {
        replace => {
          "policy_strings_label.key" => "policy_strings"
          "policy_strings_label.value.string_value" => "%{policy_strings}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "policy_strings_label"
        }
      }
    }

    if [mx_host_patterns] {
      mutate {
        replace => {
          "mx_patterns_label.key" => "mx_host_patterns"
          "mx_patterns_label.value.string_value" => "%{mx_host_patterns}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "mx_patterns_label"
        }
      }
    }

    # -- session counts --
    if [successful_session_count] {
      mutate {
        replace => {
          "success_count_label.key" => "successful_session_count"
          "success_count_label.value.string_value" => "%{successful_session_count}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "success_count_label"
        }
      }
    }

    if [failed_session_count] {
      mutate {
        replace => {
          "failed_count_label.key" => "failed_session_count"
          "failed_count_label.value.string_value" => "%{failed_session_count}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "failed_count_label"
        }
      }
    }

    # -- failure details --
    if [result_type] {
      mutate {
        replace => {
          "result_type_label.key" => "result_type"
          "result_type_label.value.string_value" => "%{result_type}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "result_type_label"
        }
      }
    }

    if [receiving_mx_helo] {
      mutate {
        replace => {
          "mx_helo_label.key" => "receiving_mx_helo"
          "mx_helo_label.value.string_value" => "%{receiving_mx_helo}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "mx_helo_label"
        }
      }
    }

    if [additional_info_uri] {
      mutate {
        replace => {
          "info_uri_label.key" => "additional_info_uri"
          "info_uri_label.value.string_value" => "%{additional_info_uri}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "info_uri_label"
        }
      }
    }

    if [failure_reason_code] {
      mutate {
        replace => {
          "failure_code_label.key" => "failure_reason_code"
          "failure_code_label.value.string_value" => "%{failure_reason_code}"
        }
        merge => {
          "event.idm.read_only_udm.additional.fields" => "failure_code_label"
        }
      }
    }

    # -- security result for TLS reports --
    mutate {
      replace => {
        "security_result_tls.summary" => "SMTP TLS Report"
      }
    }

    if [result_type] {
      mutate {
        replace => {
          "security_result_tls.description" => "result_type=%{result_type}"
          "security_result_tls.action" => "UNKNOWN_ACTION"
        }
      }
    }

    mutate {
      merge => {
        "event.idm.read_only_udm.security_result" => "security_result_tls"
      }
    }
  }

  # ==========================================================================
  # Common metadata fields
  # ==========================================================================
  mutate {
    replace => {
      "event.idm.read_only_udm.metadata.event_type" => "%{event_type}"
      "event.idm.read_only_udm.metadata.vendor_name" => "parsedmarc"
      "event.idm.read_only_udm.metadata.product_name" => "parsedmarc"
    }
  }

  if [report_type] {
    mutate {
      replace => {
        "event.idm.read_only_udm.metadata.product_event_type" => "%{report_type}"
      }
    }
  }

  # ==========================================================================
  # Output
  # ==========================================================================
  statedump {}

  mutate {
    merge => {
      "@output" => "event"
    }
  }
}