from __future__ import annotations from typing import Dict, Optional import newrelic.agent from app.email import headers from app.models import EnumE, Phase from email.message import Message class DmarcCheckResult(EnumE): allow = 0 soft_fail = 1 quarantine = 2 reject = 3 not_available = 4 bad_policy = 5 @staticmethod def get_string_dict(): return { "DMARC_POLICY_ALLOW": DmarcCheckResult.allow, "DMARC_POLICY_SOFTFAIL": DmarcCheckResult.soft_fail, "DMARC_POLICY_QUARANTINE": DmarcCheckResult.quarantine, "DMARC_POLICY_REJECT": DmarcCheckResult.reject, "DMARC_NA": DmarcCheckResult.not_available, "DMARC_BAD_POLICY": DmarcCheckResult.bad_policy, } class SPFCheckResult(EnumE): allow = 0 fail = 1 soft_fail = 1 neutral = 2 temp_error = 3 not_available = 4 perm_error = 5 @staticmethod def get_string_dict(): return { "R_SPF_ALLOW": SPFCheckResult.allow, "R_SPF_FAIL": SPFCheckResult.fail, "R_SPF_SOFTFAIL": SPFCheckResult.soft_fail, "R_SPF_NEUTRAL": SPFCheckResult.neutral, "R_SPF_DNSFAIL": SPFCheckResult.temp_error, "R_SPF_NA": SPFCheckResult.not_available, "R_SPF_PERMFAIL": SPFCheckResult.perm_error, } class SpamdResult: def __init__(self, phase: Phase = Phase.unknown): self.phase: Phase = phase self.dmarc: DmarcCheckResult = DmarcCheckResult.not_available self.spf: SPFCheckResult = SPFCheckResult.not_available def set_dmarc_result(self, dmarc_result: DmarcCheckResult): self.dmarc = dmarc_result def set_spf_result(self, spf_result: SPFCheckResult): self.spf = spf_result def event_data(self) -> Dict: return { "header": "present", "dmarc": self.dmarc.name, "spf": self.spf.name, "phase": self.phase.name, } @classmethod def extract_from_headers( cls, msg: Message, phase: Phase = Phase.unknown ) -> Optional[SpamdResult]: cached = cls._get_from_message(msg) if cached: return cached spam_result_header = msg.get_all(headers.SPAMD_RESULT) if not spam_result_header: return None spam_entries = [ entry.strip() for entry in str(spam_result_header[-1]).split("\n") ] for entry_pos in range(len(spam_entries)): sep = spam_entries[entry_pos].find("(") if sep > -1: spam_entries[entry_pos] = spam_entries[entry_pos][:sep] spamd_result = SpamdResult(phase) for header_value, dmarc_result in DmarcCheckResult.get_string_dict().items(): if header_value in spam_entries: spamd_result.set_dmarc_result(dmarc_result) break for header_value, spf_result in SPFCheckResult.get_string_dict().items(): if header_value in spam_entries: spamd_result.set_spf_result(spf_result) break cls._store_in_message(spamd_result, msg) return spamd_result @classmethod def _store_in_message(cls, check: SpamdResult, msg: Message): msg.spamd_check = check @classmethod def _get_from_message(cls, msg: Message) -> Optional[SpamdResult]: return getattr(msg, "spamd_check", None) @classmethod def send_to_new_relic(cls, msg: Message): check = cls._get_from_message(msg) if check: newrelic.agent.record_custom_event("SpamdCheck", check.event_data()) else: newrelic.agent.record_custom_event("SpamdCheck", {"header": "missing"})