Fake thread with suspicious indicators

type.inbound
// fake thread check
and (length(headers.references) == 0 or headers.in_reply_to is null)
and (
  subject.is_reply
  or subject.is_forward
  // fake thread, but no indication in the subject line
  // current_thread pulls the recent thread, but the full body contains the fake "original" email
  or (
    not (subject.is_reply or subject.is_forward)
    and any([body.current_thread.text, body.html.display_text, body.plain.raw],
            3 of (
              strings.icontains(., "from:"),
              strings.icontains(., "to:"),
              strings.icontains(., "sent:"),
              strings.icontains(., "date:"),
              strings.icontains(., "cc:"),
              strings.icontains(., "subject:")
            )
    )
    and length(body.current_thread.text) + 100 < length(coalesce(body.html.display_text,
                                                                 body.plain.raw
                                                        )
    )
  )
)

// negating bouncebacks
and not any(attachments,
            .content_type in ("message/delivery-status", "message/rfc822")
)
// negating Google Calendar invites
and (
  (
    headers.return_path.domain.domain is not null
    and headers.return_path.domain.domain != 'calendar-server.bounces.google.com'
  )
  or headers.return_path.domain.domain is null
)
// not mimecast secure message from internal source
and not (
  strings.istarts_with(headers.message_id, '<Mimecast.')
  and strings.iends_with(headers.message_id, '.mimecast.lan>')
  and headers.hops[0].received.server.raw == "relay.mimecast.com"
  and strings.icontains(headers.hops[0].received.source.raw, 'mimecast.lan')
)

// and not solicited
and not profile.by_sender().solicited
and 4 of (
  // language attempting to engage
  (
    any(ml.nlu_classifier(body.current_thread.text).entities,
        .name == "request"
    )
    and any(ml.nlu_classifier(body.current_thread.text).entities,
            .name == "financial"
    )
  ),

  // invoicing language
  (
    any(ml.nlu_classifier(body.current_thread.text).tags, .name == "invoice")
    or any(ml.nlu_classifier(body.current_thread.text).entities,
           .text == "invoice"
    )
  ),

  // urgency request
  any(ml.nlu_classifier(body.current_thread.text).entities, .name == "urgency"),

  // cred_theft detection
  any(ml.nlu_classifier(body.current_thread.text).intents,
      .name == "cred_theft" and .confidence in~ ("medium", "high")
  ),

  // commonly abused sender TLD
  strings.ilike(sender.email.domain.tld, "*.jp"),

  // headers traverse abused TLD
  any(headers.domains, strings.ilike(.tld, "*.jp")),

  // known suspicious pattern in the URL path
  any(body.links, regex.match(.href_url.path, '\/[a-z]{3}\d[a-z]')),

  // link display text is in all caps
  any(body.links, regex.match(.display_text, '[A-Z ]+')),

  // link display text contains invisible characters (U+200F)
  any(body.links, strings.contains(.display_text, "\u{200F}")),

  // Low reputation link with display text ending in a document extension
  any(body.links,
      .href_url.domain.root_domain not in $tranco_1m
      and .href_url.domain.valid
      and .href_url.domain.root_domain not in $org_domains
      and .href_url.domain.root_domain not in $high_trust_sender_root_domains
      and (
        any($file_extensions_macros, strings.ends_with(..display_text, .))
        or strings.ends_with(.display_text, 'pdf')
      )
  ),

  // display name contains an email
  regex.contains(sender.display_name, '[a-z0-9]+@[a-z]+'),

  // Sender domain is empty
  sender.email.domain.domain == "",

  // sender domain matches no body domains
  all(body.links,
      .href_url.domain.root_domain != sender.email.domain.root_domain
  ),

  // body contains name of VIP
  (
    any($org_vips, strings.icontains(body.html.inner_text, .display_name))
    or any($org_vips, strings.icontains(body.plain.raw, .display_name))
  ),

  // new body domain
  any(body.links, network.whois(.href_url.domain).days_old < 30),

  // new sender domain
  network.whois(sender.email.domain).days_old < 30,

  // new sender
  profile.by_sender().days_known < 7,

  // excessive whitespace
  (
    regex.icontains(body.html.raw, '((<br\s*/?>\s*){20,}|\n{20,})')
    or regex.icontains(body.html.raw, '(<p[^>]*>\s*<br\s*/?>\s*</p>\s*){30,}')
    or regex.icontains(body.html.raw,
                       '(<p class=".*?"><span style=".*?"><o:p>&nbsp;</o:p></span></p>\s*){30,}'
    )
    or regex.icontains(body.html.raw, '(<p>&nbsp;</p>\s*){7,}')
    or regex.icontains(body.html.raw, '(<p>&nbsp;</p><br>\s*){7,}')
    or regex.icontains(body.html.raw, '(<p[^>]*>\s*&nbsp;<br>\s*</p>\s*){5,}')
    or regex.icontains(body.html.raw, '(<p[^>]*>&nbsp;</p>\s*){7,}')
  ),

  // body contains recipient SLD
  any(recipients.to,
      strings.icontains(body.current_thread.text, .email.domain.sld)
  ),
  (
    // bec
    any(ml.nlu_classifier(body.current_thread.text).intents,
        .name == "bec" and .confidence != "low"
    )
    // previous thread contains matching domain but mismatch local part in current thread
    and any(body.previous_threads,
            .sender.email.domain.root_domain == recipients.to[0].email.domain.root_domain
            and .sender.email.email != recipients.to[0].email.email
    )
  )
)

// negate highly trusted sender domains unless they fail DMARC authentication
and (
  (
    sender.email.domain.root_domain in $high_trust_sender_root_domains
    and not headers.auth_summary.dmarc.pass
  )
  or sender.email.domain.root_domain not in $high_trust_sender_root_domains
)
and not profile.by_sender().any_messages_benign