mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
Fix quoted-AUID DKIM domain spoof in sender auth
Resolve three confirmed /sh-security-review findings on the fail-closed SES sender-authentication control. HIGH: header.i/header.d domain extraction was not quoted-string aware. An attacker with a valid DKIM key for their own domain could set an RFC 6376-legal AUID such as i="@seahaven.com"@attacker.com; the naive extractor stopped at the closing quote and returned seahaven.com, accepting forged mail. Extraction now tokenises the clause with the same quoted-string discipline already used for clause splitting: header.d (the plain signing domain) is authoritative when present, otherwise the header.i domain is the part after the AUID's LAST top-level "@", so a "@" inside a quoted local-part is treated as signer-controlled label text and yields the true signer (attacker.com), not seahaven.com. LOW: the topmost-header parse ran outside evaluate_sender_authentication's try/except, so an unexpected parser exception on crafted input could propagate into the handler and Lambda async retries/DLQ. The parse now fails CLOSED with an authentication_results_unparseable reason. MEDIUM: the sender_auth_rejected alarm used Sum>=3 over 15 min, blind to a low-volume total-reject outage (a trickle that never sums to 3). Both stacks now alarm on >=1 reject per 5-min period with evaluation_periods=3 / datapoints_to_alarm=2, so a sustained reject condition pages even at one reject per period while a lone stray probe self-clears. Refs: INFRA-107
This commit is contained in:
parent
f9bb731484
commit
25b7fc188d
5 changed files with 412 additions and 59 deletions
|
|
@ -114,14 +114,22 @@ def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic
|
|||
default_value=0,
|
||||
)
|
||||
|
||||
# >=3 rejections in a 15-min window pages: a lone spoofed probe to the
|
||||
# (obscure, internal) ingest address is tolerated, but a genuine
|
||||
# false-reject storm -- where legitimate senders are being dropped -- trips
|
||||
# quickly. Threshold is intentionally conservative and easy to tune.
|
||||
# Fire on a *sustained* reject condition rather than a volume spike. The
|
||||
# earlier Sum>=3-over-15-min threshold had a blind spot that is exactly the
|
||||
# failure this alarm exists to catch: a low-traffic pipeline in total
|
||||
# drift outage (allowlist wrong / signing-domain changed) may only produce
|
||||
# a trickle of rejects -- one every few minutes -- that never sums to 3 in
|
||||
# any window, so the outage never pages. Instead: >=1 reject per 5-min
|
||||
# period, alarming when 2 of the last 3 periods breach (evaluation_periods=3
|
||||
# / datapoints_to_alarm=2, the same idiom as the duration alarm). A single
|
||||
# stray spoof probe (one lone period) is tolerated and self-clears, but a
|
||||
# sustained reject condition trips within ~10-15 min even at one reject per
|
||||
# period. default_value=0 on the metric filter keeps the series continuous
|
||||
# so NOT_BREACHING only applies before the first datapoint ever arrives.
|
||||
cloudwatch.Metric(
|
||||
namespace=_SENDER_AUTH_METRIC_NAMESPACE,
|
||||
metric_name=metric_name,
|
||||
period=Duration.minutes(15),
|
||||
period=Duration.minutes(5),
|
||||
statistic="Sum",
|
||||
).create_alarm(
|
||||
scope,
|
||||
|
|
@ -131,8 +139,9 @@ def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic
|
|||
f"{function_name} rejected inbound mail on sender authentication "
|
||||
"(possible allowlist/DKIM-domain drift silently dropping real mail)"
|
||||
),
|
||||
threshold=3,
|
||||
evaluation_periods=1,
|
||||
threshold=1,
|
||||
evaluation_periods=3,
|
||||
datapoints_to_alarm=2,
|
||||
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,
|
||||
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
|
||||
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
|
||||
|
|
|
|||
|
|
@ -114,14 +114,22 @@ def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic
|
|||
default_value=0,
|
||||
)
|
||||
|
||||
# >=3 rejections in a 15-min window pages: a lone spoofed probe to the
|
||||
# (obscure, internal) ingest address is tolerated, but a genuine
|
||||
# false-reject storm -- where legitimate senders are being dropped -- trips
|
||||
# quickly. Threshold is intentionally conservative and easy to tune.
|
||||
# Fire on a *sustained* reject condition rather than a volume spike. The
|
||||
# earlier Sum>=3-over-15-min threshold had a blind spot that is exactly the
|
||||
# failure this alarm exists to catch: a low-traffic pipeline in total
|
||||
# drift outage (allowlist wrong / signing-domain changed) may only produce
|
||||
# a trickle of rejects -- one every few minutes -- that never sums to 3 in
|
||||
# any window, so the outage never pages. Instead: >=1 reject per 5-min
|
||||
# period, alarming when 2 of the last 3 periods breach (evaluation_periods=3
|
||||
# / datapoints_to_alarm=2, the same idiom as the duration alarm). A single
|
||||
# stray spoof probe (one lone period) is tolerated and self-clears, but a
|
||||
# sustained reject condition trips within ~10-15 min even at one reject per
|
||||
# period. default_value=0 on the metric filter keeps the series continuous
|
||||
# so NOT_BREACHING only applies before the first datapoint ever arrives.
|
||||
cloudwatch.Metric(
|
||||
namespace=_SENDER_AUTH_METRIC_NAMESPACE,
|
||||
metric_name=metric_name,
|
||||
period=Duration.minutes(15),
|
||||
period=Duration.minutes(5),
|
||||
statistic="Sum",
|
||||
).create_alarm(
|
||||
scope,
|
||||
|
|
@ -131,8 +139,9 @@ def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic
|
|||
f"{function_name} rejected inbound mail on sender authentication "
|
||||
"(possible allowlist/DKIM-domain drift silently dropping real mail)"
|
||||
),
|
||||
threshold=3,
|
||||
evaluation_periods=1,
|
||||
threshold=1,
|
||||
evaluation_periods=3,
|
||||
datapoints_to_alarm=2,
|
||||
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,
|
||||
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
|
||||
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
|
||||
|
|
|
|||
|
|
@ -26,7 +26,13 @@ Observed SES format (2026-07-15, both ingest buckets)::
|
|||
dmarc=none header.from=hxgnsmartcloud.com;
|
||||
|
||||
Note SES reports the passing DKIM identity as ``header.i=@<domain>``
|
||||
(RFC 6376 AUID), not ``header.d=``; the parser accepts both.
|
||||
(RFC 6376 AUID), not ``header.d=``; the parser accepts both, and treats
|
||||
``header.d`` (the plain signing domain) as authoritative over ``header.i``
|
||||
when both are present. The ``header.i`` domain is derived per RFC 6376 as
|
||||
the part after the AUID's *last* ``@`` -- a ``@`` inside a quoted
|
||||
local-part (e.g. ``i="@seahaven.com"@attacker.com``) is signer-controlled
|
||||
label text, never the identity domain, and yields the true signer
|
||||
(``attacker.com``).
|
||||
|
||||
**Clause injection defence (INFRA-107 hardening).** SES echoes several
|
||||
attacker-controlled SMTP-session tokens into its own Authentication-Results
|
||||
|
|
@ -68,8 +74,6 @@ SES_AUTHSERV_ID = "amazonses.com"
|
|||
# before matching, so the pre-strip "dkim=pass(comment)" form arrives here
|
||||
# as "dkim=pass ..." and still terminates on whitespace.)
|
||||
_DKIM_RESULT_RE = re.compile(r"^dkim\s*=\s*([a-z0-9]+)(?=$|\s)", re.IGNORECASE)
|
||||
_HEADER_D_RE = re.compile(r"header\.d\s*=\s*\"?([^\s\";]+)", re.IGNORECASE)
|
||||
_HEADER_I_RE = re.compile(r"header\.i\s*=\s*\"?([^\s\";]+)", re.IGNORECASE)
|
||||
|
||||
|
||||
def get_allowed_dkim_domains() -> frozenset:
|
||||
|
|
@ -183,16 +187,146 @@ def _split_clauses(value: str) -> list:
|
|||
return clauses
|
||||
|
||||
|
||||
def _split_properties(clause: str) -> list:
|
||||
"""Split a clause into whitespace-separated ``name=pvalue`` tokens.
|
||||
|
||||
Whitespace *inside* a quoted-string does not split, so an RFC 8601 pvalue
|
||||
that embeds a quoted-string (e.g. a DKIM AUID with a quoted local-part that
|
||||
legally contains spaces) survives as a single token. This is the same
|
||||
quoted-string discipline :func:`_split_clauses` applies at the ``;`` level,
|
||||
reused here at the token level so ``header.d=`` / ``header.i=`` extraction
|
||||
is quoted-string-aware rather than a naive regex grab.
|
||||
"""
|
||||
tokens = []
|
||||
buf = []
|
||||
in_quote = False
|
||||
i = 0
|
||||
n = len(clause)
|
||||
while i < n:
|
||||
c = clause[i]
|
||||
if in_quote:
|
||||
buf.append(c)
|
||||
if c == "\\" and i + 1 < n:
|
||||
buf.append(clause[i + 1])
|
||||
i += 2
|
||||
continue
|
||||
if c == '"':
|
||||
in_quote = False
|
||||
i += 1
|
||||
continue
|
||||
if c == '"':
|
||||
in_quote = True
|
||||
buf.append(c)
|
||||
i += 1
|
||||
continue
|
||||
if c.isspace():
|
||||
if buf:
|
||||
tokens.append("".join(buf))
|
||||
buf = []
|
||||
i += 1
|
||||
continue
|
||||
buf.append(c)
|
||||
i += 1
|
||||
if buf:
|
||||
tokens.append("".join(buf))
|
||||
return tokens
|
||||
|
||||
|
||||
def _auid_domain(pvalue: str) -> str:
|
||||
"""Domain of a DKIM AUID (``header.i``) per RFC 6376.
|
||||
|
||||
The identity domain is the part after the *last* ``@`` of the AUID -- but a
|
||||
``@`` inside a quoted local-part is NOT the identity separator. So
|
||||
``"@seahaven.com"@attacker.com`` yields ``attacker.com`` (the real signer),
|
||||
not ``seahaven.com``: the ``@seahaven.com`` sits inside the quoted
|
||||
local-part and is signer-controlled label text, never the domain. A bare
|
||||
unquoted ``@seahaven.com`` still yields ``seahaven.com``. Returns "" when
|
||||
there is no top-level ``@`` (no valid domain) or the domain looks malformed.
|
||||
"""
|
||||
last_at = -1
|
||||
in_quote = False
|
||||
i = 0
|
||||
n = len(pvalue)
|
||||
while i < n:
|
||||
c = pvalue[i]
|
||||
if in_quote:
|
||||
if c == "\\":
|
||||
i += 2
|
||||
continue
|
||||
if c == '"':
|
||||
in_quote = False
|
||||
i += 1
|
||||
continue
|
||||
if c == '"':
|
||||
in_quote = True
|
||||
elif c == "@":
|
||||
last_at = i
|
||||
i += 1
|
||||
if last_at < 0:
|
||||
return ""
|
||||
domain = pvalue[last_at + 1 :].lower().rstrip(".")
|
||||
# A DKIM domain-name is a plain dot-atom; anything with a residual quote is
|
||||
# malformed (or a smuggling attempt) and must not be trusted.
|
||||
if not domain or '"' in domain:
|
||||
return ""
|
||||
return domain
|
||||
|
||||
|
||||
def _plain_domain(pvalue: str) -> str:
|
||||
"""Domain of ``header.d`` -- the DKIM signing domain, a plain dot-atom.
|
||||
|
||||
SES writes ``header.d`` verbatim from the signature's ``d=`` tag, which is
|
||||
never a quoted-string. A residual quote means malformed/smuggled input and
|
||||
fails closed.
|
||||
"""
|
||||
domain = pvalue.lower().rstrip(".")
|
||||
if not domain or '"' in domain:
|
||||
return ""
|
||||
return domain
|
||||
|
||||
|
||||
def _clause_signer_domain(clause: str) -> str:
|
||||
"""Authoritative DKIM signer domain of a ``dkim=pass`` clause, or "".
|
||||
|
||||
``header.d`` (the signing domain) is authoritative and is preferred when
|
||||
present; only when it is absent does this fall back to ``header.i`` and
|
||||
derive the domain from the AUID's post-final-``@`` part. Both lookups run
|
||||
over :func:`_split_properties` tokens, so a ``header.d=``/``header.i=``
|
||||
literal smuggled *inside* another property's quoted pvalue is confined to
|
||||
that one token and can never be read as a top-level property.
|
||||
"""
|
||||
header_d = None
|
||||
header_i = None
|
||||
for token in _split_properties(clause):
|
||||
name, sep, val = token.partition("=")
|
||||
if not sep:
|
||||
continue
|
||||
key = name.strip().lower()
|
||||
if key == "header.d" and header_d is None:
|
||||
header_d = val
|
||||
elif key == "header.i" and header_i is None:
|
||||
header_i = val
|
||||
if header_d is not None:
|
||||
return _plain_domain(header_d)
|
||||
if header_i is not None:
|
||||
return _auid_domain(header_i)
|
||||
return ""
|
||||
|
||||
|
||||
def parse_authentication_results(value: str) -> tuple:
|
||||
"""Parse one Authentication-Results header value.
|
||||
|
||||
Returns ``(authserv_id, passing_dkim_domains)`` where the domains are
|
||||
the lowercased d=/i= domains of every ``dkim=pass`` clause. Malformed
|
||||
input yields ``("", frozenset())``, which callers treat as a rejection.
|
||||
Returns ``(authserv_id, passing_dkim_domains)`` where the domains are the
|
||||
authoritative signer domains of every ``dkim=pass`` clause -- ``header.d``
|
||||
when present, else the ``header.i`` AUID's post-final-``@`` domain (see
|
||||
:func:`_clause_signer_domain`). Malformed input yields ``("", frozenset())``,
|
||||
which callers treat as a rejection.
|
||||
|
||||
Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322)
|
||||
so attacker-controlled tokens SES echoes into its header (envelope-from,
|
||||
helo, header.from) cannot be split into a forged ``dkim=pass`` clause.
|
||||
Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322) at
|
||||
the ``;`` (clause), whitespace (property) and ``@`` (AUID domain) levels, so
|
||||
attacker-controlled tokens SES echoes into its header (envelope-from, helo,
|
||||
header.from) cannot be split into a forged ``dkim=pass`` clause and a quoted
|
||||
AUID local-part cannot masquerade as the identity domain.
|
||||
"""
|
||||
text, well_formed = _strip_comments(_unfold(value))
|
||||
if not well_formed:
|
||||
|
|
@ -211,18 +345,9 @@ def parse_authentication_results(value: str) -> tuple:
|
|||
match = _DKIM_RESULT_RE.match(clause)
|
||||
if not match or match.group(1).lower() != "pass":
|
||||
continue
|
||||
d_match = _HEADER_D_RE.search(clause)
|
||||
if d_match:
|
||||
passing.add(d_match.group(1).lower().rstrip("."))
|
||||
continue
|
||||
i_match = _HEADER_I_RE.search(clause)
|
||||
if i_match:
|
||||
# AUID (header.i) is "local-part@domain" or "@domain";
|
||||
# the signing domain is everything after the last "@".
|
||||
identity = i_match.group(1).lower().rstrip(".")
|
||||
domain = identity.rsplit("@", 1)[-1]
|
||||
if domain:
|
||||
passing.add(domain)
|
||||
domain = _clause_signer_domain(clause)
|
||||
if domain:
|
||||
passing.add(domain)
|
||||
return authserv_id, frozenset(passing)
|
||||
|
||||
|
||||
|
|
@ -252,7 +377,14 @@ def evaluate_sender_authentication(raw_email: bytes, allowed_domains) -> tuple:
|
|||
# SES prepends its trace headers, so index 0 is the SES-stamped copy.
|
||||
# Any Authentication-Results header further down arrived inside the
|
||||
# message (attacker-suppliable) and is deliberately ignored.
|
||||
authserv_id, passing = parse_authentication_results(str(ar_headers[0]))
|
||||
#
|
||||
# The parse is wrapped so an unexpected parser exception fails CLOSED
|
||||
# (rejected, structured reason) instead of propagating out of the handler
|
||||
# into Lambda's async retries / DLQ on attacker-crafted input.
|
||||
try:
|
||||
authserv_id, passing = parse_authentication_results(str(ar_headers[0]))
|
||||
except Exception:
|
||||
return False, "authentication_results_unparseable", {}
|
||||
detail = {
|
||||
"authserv_id": authserv_id,
|
||||
"passing_dkim_domains": sorted(passing),
|
||||
|
|
|
|||
|
|
@ -26,7 +26,13 @@ Observed SES format (2026-07-15, both ingest buckets)::
|
|||
dmarc=none header.from=hxgnsmartcloud.com;
|
||||
|
||||
Note SES reports the passing DKIM identity as ``header.i=@<domain>``
|
||||
(RFC 6376 AUID), not ``header.d=``; the parser accepts both.
|
||||
(RFC 6376 AUID), not ``header.d=``; the parser accepts both, and treats
|
||||
``header.d`` (the plain signing domain) as authoritative over ``header.i``
|
||||
when both are present. The ``header.i`` domain is derived per RFC 6376 as
|
||||
the part after the AUID's *last* ``@`` -- a ``@`` inside a quoted
|
||||
local-part (e.g. ``i="@seahaven.com"@attacker.com``) is signer-controlled
|
||||
label text, never the identity domain, and yields the true signer
|
||||
(``attacker.com``).
|
||||
|
||||
**Clause injection defence (INFRA-107 hardening).** SES echoes several
|
||||
attacker-controlled SMTP-session tokens into its own Authentication-Results
|
||||
|
|
@ -68,8 +74,6 @@ SES_AUTHSERV_ID = "amazonses.com"
|
|||
# before matching, so the pre-strip "dkim=pass(comment)" form arrives here
|
||||
# as "dkim=pass ..." and still terminates on whitespace.)
|
||||
_DKIM_RESULT_RE = re.compile(r"^dkim\s*=\s*([a-z0-9]+)(?=$|\s)", re.IGNORECASE)
|
||||
_HEADER_D_RE = re.compile(r"header\.d\s*=\s*\"?([^\s\";]+)", re.IGNORECASE)
|
||||
_HEADER_I_RE = re.compile(r"header\.i\s*=\s*\"?([^\s\";]+)", re.IGNORECASE)
|
||||
|
||||
|
||||
def get_allowed_dkim_domains() -> frozenset:
|
||||
|
|
@ -183,16 +187,146 @@ def _split_clauses(value: str) -> list:
|
|||
return clauses
|
||||
|
||||
|
||||
def _split_properties(clause: str) -> list:
|
||||
"""Split a clause into whitespace-separated ``name=pvalue`` tokens.
|
||||
|
||||
Whitespace *inside* a quoted-string does not split, so an RFC 8601 pvalue
|
||||
that embeds a quoted-string (e.g. a DKIM AUID with a quoted local-part that
|
||||
legally contains spaces) survives as a single token. This is the same
|
||||
quoted-string discipline :func:`_split_clauses` applies at the ``;`` level,
|
||||
reused here at the token level so ``header.d=`` / ``header.i=`` extraction
|
||||
is quoted-string-aware rather than a naive regex grab.
|
||||
"""
|
||||
tokens = []
|
||||
buf = []
|
||||
in_quote = False
|
||||
i = 0
|
||||
n = len(clause)
|
||||
while i < n:
|
||||
c = clause[i]
|
||||
if in_quote:
|
||||
buf.append(c)
|
||||
if c == "\\" and i + 1 < n:
|
||||
buf.append(clause[i + 1])
|
||||
i += 2
|
||||
continue
|
||||
if c == '"':
|
||||
in_quote = False
|
||||
i += 1
|
||||
continue
|
||||
if c == '"':
|
||||
in_quote = True
|
||||
buf.append(c)
|
||||
i += 1
|
||||
continue
|
||||
if c.isspace():
|
||||
if buf:
|
||||
tokens.append("".join(buf))
|
||||
buf = []
|
||||
i += 1
|
||||
continue
|
||||
buf.append(c)
|
||||
i += 1
|
||||
if buf:
|
||||
tokens.append("".join(buf))
|
||||
return tokens
|
||||
|
||||
|
||||
def _auid_domain(pvalue: str) -> str:
|
||||
"""Domain of a DKIM AUID (``header.i``) per RFC 6376.
|
||||
|
||||
The identity domain is the part after the *last* ``@`` of the AUID -- but a
|
||||
``@`` inside a quoted local-part is NOT the identity separator. So
|
||||
``"@seahaven.com"@attacker.com`` yields ``attacker.com`` (the real signer),
|
||||
not ``seahaven.com``: the ``@seahaven.com`` sits inside the quoted
|
||||
local-part and is signer-controlled label text, never the domain. A bare
|
||||
unquoted ``@seahaven.com`` still yields ``seahaven.com``. Returns "" when
|
||||
there is no top-level ``@`` (no valid domain) or the domain looks malformed.
|
||||
"""
|
||||
last_at = -1
|
||||
in_quote = False
|
||||
i = 0
|
||||
n = len(pvalue)
|
||||
while i < n:
|
||||
c = pvalue[i]
|
||||
if in_quote:
|
||||
if c == "\\":
|
||||
i += 2
|
||||
continue
|
||||
if c == '"':
|
||||
in_quote = False
|
||||
i += 1
|
||||
continue
|
||||
if c == '"':
|
||||
in_quote = True
|
||||
elif c == "@":
|
||||
last_at = i
|
||||
i += 1
|
||||
if last_at < 0:
|
||||
return ""
|
||||
domain = pvalue[last_at + 1 :].lower().rstrip(".")
|
||||
# A DKIM domain-name is a plain dot-atom; anything with a residual quote is
|
||||
# malformed (or a smuggling attempt) and must not be trusted.
|
||||
if not domain or '"' in domain:
|
||||
return ""
|
||||
return domain
|
||||
|
||||
|
||||
def _plain_domain(pvalue: str) -> str:
|
||||
"""Domain of ``header.d`` -- the DKIM signing domain, a plain dot-atom.
|
||||
|
||||
SES writes ``header.d`` verbatim from the signature's ``d=`` tag, which is
|
||||
never a quoted-string. A residual quote means malformed/smuggled input and
|
||||
fails closed.
|
||||
"""
|
||||
domain = pvalue.lower().rstrip(".")
|
||||
if not domain or '"' in domain:
|
||||
return ""
|
||||
return domain
|
||||
|
||||
|
||||
def _clause_signer_domain(clause: str) -> str:
|
||||
"""Authoritative DKIM signer domain of a ``dkim=pass`` clause, or "".
|
||||
|
||||
``header.d`` (the signing domain) is authoritative and is preferred when
|
||||
present; only when it is absent does this fall back to ``header.i`` and
|
||||
derive the domain from the AUID's post-final-``@`` part. Both lookups run
|
||||
over :func:`_split_properties` tokens, so a ``header.d=``/``header.i=``
|
||||
literal smuggled *inside* another property's quoted pvalue is confined to
|
||||
that one token and can never be read as a top-level property.
|
||||
"""
|
||||
header_d = None
|
||||
header_i = None
|
||||
for token in _split_properties(clause):
|
||||
name, sep, val = token.partition("=")
|
||||
if not sep:
|
||||
continue
|
||||
key = name.strip().lower()
|
||||
if key == "header.d" and header_d is None:
|
||||
header_d = val
|
||||
elif key == "header.i" and header_i is None:
|
||||
header_i = val
|
||||
if header_d is not None:
|
||||
return _plain_domain(header_d)
|
||||
if header_i is not None:
|
||||
return _auid_domain(header_i)
|
||||
return ""
|
||||
|
||||
|
||||
def parse_authentication_results(value: str) -> tuple:
|
||||
"""Parse one Authentication-Results header value.
|
||||
|
||||
Returns ``(authserv_id, passing_dkim_domains)`` where the domains are
|
||||
the lowercased d=/i= domains of every ``dkim=pass`` clause. Malformed
|
||||
input yields ``("", frozenset())``, which callers treat as a rejection.
|
||||
Returns ``(authserv_id, passing_dkim_domains)`` where the domains are the
|
||||
authoritative signer domains of every ``dkim=pass`` clause -- ``header.d``
|
||||
when present, else the ``header.i`` AUID's post-final-``@`` domain (see
|
||||
:func:`_clause_signer_domain`). Malformed input yields ``("", frozenset())``,
|
||||
which callers treat as a rejection.
|
||||
|
||||
Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322)
|
||||
so attacker-controlled tokens SES echoes into its header (envelope-from,
|
||||
helo, header.from) cannot be split into a forged ``dkim=pass`` clause.
|
||||
Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322) at
|
||||
the ``;`` (clause), whitespace (property) and ``@`` (AUID domain) levels, so
|
||||
attacker-controlled tokens SES echoes into its header (envelope-from, helo,
|
||||
header.from) cannot be split into a forged ``dkim=pass`` clause and a quoted
|
||||
AUID local-part cannot masquerade as the identity domain.
|
||||
"""
|
||||
text, well_formed = _strip_comments(_unfold(value))
|
||||
if not well_formed:
|
||||
|
|
@ -211,18 +345,9 @@ def parse_authentication_results(value: str) -> tuple:
|
|||
match = _DKIM_RESULT_RE.match(clause)
|
||||
if not match or match.group(1).lower() != "pass":
|
||||
continue
|
||||
d_match = _HEADER_D_RE.search(clause)
|
||||
if d_match:
|
||||
passing.add(d_match.group(1).lower().rstrip("."))
|
||||
continue
|
||||
i_match = _HEADER_I_RE.search(clause)
|
||||
if i_match:
|
||||
# AUID (header.i) is "local-part@domain" or "@domain";
|
||||
# the signing domain is everything after the last "@".
|
||||
identity = i_match.group(1).lower().rstrip(".")
|
||||
domain = identity.rsplit("@", 1)[-1]
|
||||
if domain:
|
||||
passing.add(domain)
|
||||
domain = _clause_signer_domain(clause)
|
||||
if domain:
|
||||
passing.add(domain)
|
||||
return authserv_id, frozenset(passing)
|
||||
|
||||
|
||||
|
|
@ -252,7 +377,14 @@ def evaluate_sender_authentication(raw_email: bytes, allowed_domains) -> tuple:
|
|||
# SES prepends its trace headers, so index 0 is the SES-stamped copy.
|
||||
# Any Authentication-Results header further down arrived inside the
|
||||
# message (attacker-suppliable) and is deliberately ignored.
|
||||
authserv_id, passing = parse_authentication_results(str(ar_headers[0]))
|
||||
#
|
||||
# The parse is wrapped so an unexpected parser exception fails CLOSED
|
||||
# (rejected, structured reason) instead of propagating out of the handler
|
||||
# into Lambda's async retries / DLQ on attacker-crafted input.
|
||||
try:
|
||||
authserv_id, passing = parse_authentication_results(str(ar_headers[0]))
|
||||
except Exception:
|
||||
return False, "authentication_results_unparseable", {}
|
||||
detail = {
|
||||
"authserv_id": authserv_id,
|
||||
"passing_dkim_domains": sorted(passing),
|
||||
|
|
|
|||
|
|
@ -96,10 +96,66 @@ class TestParseAuthenticationResults:
|
|||
)
|
||||
assert passing == frozenset({"seahaven.com"})
|
||||
|
||||
def test_quoted_domain_value(self, ses_auth):
|
||||
def test_quoted_auid_localpart_is_not_the_domain(self, ses_auth):
|
||||
# header.i="@seahaven.com" is a *quoted local-part* with no top-level
|
||||
# "@domain" -- per RFC 6376 there is no identity domain, so it must
|
||||
# yield nothing rather than mistaking the quoted label for the domain.
|
||||
_, passing = ses_auth.parse_authentication_results(
|
||||
'amazonses.com; dkim=pass header.i="@seahaven.com"'
|
||||
)
|
||||
assert passing == frozenset()
|
||||
|
||||
def test_quoted_auid_localpart_attack_yields_true_signer(self, ses_auth):
|
||||
# The core INFRA-107 defect: a signer with their own valid DKIM key for
|
||||
# attacker.com sets AUID i="@seahaven.com"@attacker.com (RFC 6376-legal:
|
||||
# the quoted local-part contains an "@"). The identity domain is the
|
||||
# part after the LAST top-level "@" -> attacker.com, NOT seahaven.com.
|
||||
_, passing = ses_auth.parse_authentication_results(
|
||||
'amazonses.com; dkim=pass header.i="@seahaven.com"@attacker.com'
|
||||
)
|
||||
assert passing == frozenset({"attacker.com"})
|
||||
assert "seahaven.com" not in passing
|
||||
|
||||
def test_quoted_auid_attack_rejected_end_to_end(self, ses_auth):
|
||||
# Same attack through the full evaluator against a seahaven.com
|
||||
# allowlist: the genuine dkim=pass binds to attacker.com, so the forged
|
||||
# mail must fail closed, not be accepted for seahaven.com.
|
||||
forged = (
|
||||
"Authentication-Results: amazonses.com;\r\n"
|
||||
' dkim=pass header.i="@seahaven.com"@attacker.com;\r\n'
|
||||
)
|
||||
accepted, reason, _ = ses_auth.evaluate_sender_authentication(
|
||||
raw(forged), {"seahaven.com"}
|
||||
)
|
||||
assert not accepted
|
||||
assert reason == "dkim_domain_not_allowlisted"
|
||||
|
||||
def test_header_d_takes_precedence_over_mismatched_header_i(self, ses_auth):
|
||||
# header.d is the authoritative signing domain; when both appear it
|
||||
# wins over header.i, whichever order SES emits them in.
|
||||
for clause in (
|
||||
"dkim=pass header.i=@evil.com header.d=seahaven.com",
|
||||
"dkim=pass header.d=seahaven.com header.i=@evil.com",
|
||||
):
|
||||
_, passing = ses_auth.parse_authentication_results(
|
||||
f"amazonses.com; {clause}"
|
||||
)
|
||||
assert passing == frozenset({"seahaven.com"})
|
||||
|
||||
def test_header_d_smuggled_in_quoted_header_i_is_not_top_level(self, ses_auth):
|
||||
# A "header.d=seahaven.com" literal hidden inside header.i's quoted
|
||||
# local-part must not be read as a top-level header.d property; the real
|
||||
# (top-level) header.d=attacker.com is authoritative.
|
||||
_, passing = ses_auth.parse_authentication_results(
|
||||
'amazonses.com; dkim=pass header.i="header.d=seahaven.com x"@attacker.com'
|
||||
" header.d=attacker.com"
|
||||
)
|
||||
assert passing == frozenset({"attacker.com"})
|
||||
|
||||
def test_legitimate_unquoted_header_i_still_passes(self, ses_auth):
|
||||
_, passing = ses_auth.parse_authentication_results(
|
||||
"amazonses.com; dkim=pass header.i=@seahaven.com"
|
||||
)
|
||||
assert passing == frozenset({"seahaven.com"})
|
||||
|
||||
def test_folding_inside_dkim_clause(self, ses_auth):
|
||||
|
|
@ -273,6 +329,21 @@ class TestEvaluateSenderAuthentication:
|
|||
assert not accepted
|
||||
assert reason == "allowlist_not_configured"
|
||||
|
||||
def test_parser_exception_fails_closed(self, ses_auth, monkeypatch):
|
||||
# The parse of the topmost AR header runs outside the BytesParser
|
||||
# try/except; an unexpected parser exception on crafted input must fail
|
||||
# CLOSED (rejected, structured reason) rather than propagate out of the
|
||||
# handler into Lambda's async retries / DLQ.
|
||||
def boom(_value):
|
||||
raise ValueError("simulated parser blow-up")
|
||||
|
||||
monkeypatch.setattr(ses_auth, "parse_authentication_results", boom)
|
||||
accepted, reason, _ = ses_auth.evaluate_sender_authentication(
|
||||
raw(WO_SES_HEADER), {"seahaven.com"}
|
||||
)
|
||||
assert not accepted
|
||||
assert reason == "authentication_results_unparseable"
|
||||
|
||||
def test_envelope_from_clause_injection_rejected(self, ses_auth):
|
||||
# Full forged email: attacker's quoted MAIL FROM is echoed by SES into
|
||||
# its own (topmost, genuinely SES-stamped) Authentication-Results as
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue