emailsec

127async def authenticate_message(
128    smtp_context: SMTPContext,
129    raw_email: bytes,
130    configuration: AuthenticationConfiguration | None = None,
131) -> AuthenticationResult:
132    """
133    Authenticate an incoming email using SPF, DKIM, and DMARC.
134
135    Authentication flow:
136    1. SPF (RFC 7208): Verify the sending IP is authorized for the envelope sender domain
137    2. DKIM (RFC 6376): Verify cryptographic signatures on the email
138    3. ARC (RFC 8617): Check authentication chain if present (for forwarded mail)
139    4. DMARC (RFC 7489): Evaluate if SPF/DKIM align with the From header domain
140    5. Make delivery decision based on combined results
141
142    Delivery decision logic following RFC 7489.
143
144    1. DMARC policy is enforced first (if the sender has one)
145    2. If DMARC passes, then accept
146    3. If no DMARC or policy is "none", then check individual SPF/DKIM
147    4. Default to quarantine for unauthenticated mail
148    """
149    body_and_headers = emailsec._utils.body_and_headers_for_canonicalization(raw_email)
150    header_from_raw = emailsec._utils.header_value(body_and_headers[1], "from")
151
152    # Extract domain from From header for DMARC evaluation (RFC 7489 Section 3.1)
153    # From header can be "Name <email@domain.com>" or just "email@domain.com"
154    _, email_address = parseaddr(header_from_raw)
155    if not email_address or "@" not in email_address:
156        # RFC 5322: From header must contain a valid email address
157        # Malformed From header should result in rejection
158        return AuthenticationResult(
159            delivery_action=DeliveryAction.REJECT,
160            spf_check=SPFCheck(SPFResult.NONE, "", "", "Malformed From header"),
161            dkim_check=DKIMCheck(DKIMResult.PERMFAIL, None, None),
162            dmarc_check=None,
163            arc_check=ARCCheck(ARCChainStatus.NONE, "Malformed From header"),
164        )
165
166    header_from = email_address.partition("@")[2]
167
168    # Step 1: SPF Check (RFC 7208)
169    # RFC 7489: SPF authenticates the envelope sender domain
170    spf_check = await check_spf(
171        smtp_context.sender_ip_address,
172        smtp_context.mail_from,
173    )
174
175    # Step 2: DKIM Verification (RFC 6376)
176    # Performed independently of SPF per RFC 7489 Section 4.3
177    dkim_check = await check_dkim(raw_email)
178
179    # Step 3: ARC Processing (RFC 8617) - if ARC headers present
180    # RFC 8617 Section 7.2: "allows Internet Mail Handler to potentially base
181    # decisions of message disposition on authentication assessments"
182    arc_check = await check_arc(raw_email, body_and_headers)
183
184    # Step 4: DMARC Evaluation (RFC 7489)
185    # RFC 7489: "A message satisfies the DMARC checks if at least one of the
186    # supported authentication mechanisms produces a 'pass' result"
187    dmarc_check = await check_dmarc(
188        header_from=header_from,
189        envelope_from=smtp_context.mail_from,
190        spf_check=spf_check,
191        dkim_check=dkim_check,
192        arc_check=arc_check,
193        configuration=configuration,
194    )
195
196    # Handle DMARC temp errors, defer for temporary DNS issues
197    # (treat perm errors as "no DMARC policy" and continue processing)
198    if dmarc_check.result == DMARCResult.TEMPERROR:
199        return AuthenticationResult(
200            delivery_action=DeliveryAction.DEFER,
201            spf_check=spf_check,
202            dkim_check=dkim_check,
203            dmarc_check=dmarc_check,
204            arc_check=arc_check,
205        )
206
207    # Step 5: Make delivery decision
208    # RFC 7489: "Final disposition of a message is always a matter of local policy"
209    return AuthenticationResult(
210        delivery_action=_make_delivery_decision(
211            spf_check, dkim_check, arc_check, dmarc_check
212        ),
213        spf_check=spf_check,
214        dkim_check=dkim_check,
215        dmarc_check=dmarc_check,
216        arc_check=arc_check,
217    )

118@dataclass
119class AuthenticationResult:
120    delivery_action: DeliveryAction
121    spf_check: SPFCheck
122    dkim_check: DKIMCheck
123    dmarc_check: DMARCCheck | None
124    arc_check: ARCCheck

 5@dataclass
 6class AuthenticationConfiguration:
 7    """Configuration items to tweak the authentication behavior."""
 8
 9    trusted_arc_signers: list[str] | None = None
10    """Trusted ARC signers that will enable overriding DMARC results with the
11    authentication results from ARC.
12    """

27class DeliveryAction(Enum):
28    """Final delivery results based on the authentication checks."""
29
30    ACCEPT = "accept"
31    QUARANTINE = "quarantine"
32    REJECT = "reject"
33    DEFER = "defer"  # SMTP server should return 451 4.3.0 Temporary lookup failure

36@dataclass
37class SMTPContext:
38    """Context from the SMTP server processing the incoming email."""
39
40    # Connection info
41    sender_ip_address: str
42    client_hostname: str | None  # EHLO/HELO hostname
43
44    # Envelope data
45    mail_from: str  # MAIL FROM address (envelope sender)
46
47    # TODO: timestamp to check for expired signature?

18def body_and_headers_for_canonicalization(message: bytes) -> BodyAndHeaders:
19    """
20    Parse a raw email message into its body and headers for DKIM/ARC canonicalization.
21
22    This function splits the message at the first empty line and parses headers,
23    handling folded header values according to RFC 5322.
24
25    Args:
26        message: The raw email message as bytes.
27
28    Returns:
29        A tuple of (body, headers) where body is the raw body bytes and headers
30        is a dictionary mapping lowercase header names to lists of (name, value) tuples.
31    """
32    lines = re.split(b"\r?\n", message)
33
34    headers_idx = collections.defaultdict(list)
35    headers = []
36    for header_line in lines[: lines.index(b"")]:
37        if (m := re.match(rb"([\x21-\x7e]+?):", header_line)) is not None:
38            header_name = m.group(1)
39            header_value = header_line[m.end() :] + b"\r\n"
40            headers.append([header_name, header_value])
41        elif header_line.startswith(b" ") or header_line.startswith(b"\t"):
42            # Unfold header values
43            headers[-1][1] += header_line + b"\r\n"
44        else:
45            raise ValueError(f"Invalid line {header_line!r}")
46
47    for header_name, header_value in headers:
48        headers_idx[header_name.decode().lower()].append((header_name, header_value))
49
50    try:
51        # Split on the first empty line and join the remaining ones with CRLF
52        can_body = b"\r\n".join(lines[lines.index(b"") + 1 :])
53    except ValueError:
54        # No body defaults to CRLF
55        can_body = b"\r\n"
56
57    return can_body, dict(headers_idx)