mirror of
https://github.com/suitenumerique/messages.git
synced 2026-09-25 19:24:59 +02:00
🐛(dnscheck) relax DKIM whitespace check for DNS configs
Whitespace is ignored in DKIM keys but that wasn't implemented in our validator, which showed "Invalid" in the admin UI for some keys
This commit is contained in:
@@ -18,6 +18,10 @@ logger = logging.getLogger(__name__)
|
||||
SPF_CHECK_CACHE_KEY_PREFIX = "dns:spf_check:"
|
||||
SPF_CHECK_CACHE_TIMEOUT = 600 # 10 minutes
|
||||
|
||||
# DKIM tags whose values are base64: internal whitespace is not significant.
|
||||
DKIM_BASE64_TAGS = frozenset({"p", "b", "bh"})
|
||||
DKIM_VERSION = "DKIM1"
|
||||
|
||||
|
||||
def normalize_txt_value(value: str) -> str:
|
||||
"""
|
||||
@@ -27,25 +31,27 @@ def normalize_txt_value(value: str) -> str:
|
||||
|
||||
|
||||
def parse_dkim_tags(value: str) -> Optional[Dict[str, str]]:
|
||||
"""Parse a DKIM record into a dict of tag=value pairs.
|
||||
"""Parse a DKIM key record into a dict of tag=value pairs.
|
||||
|
||||
Per RFC 6376, tags are separated by semicolons, with tag=value format.
|
||||
The v= tag MUST be first and equal to DKIM1.
|
||||
Returns None if the record is not a valid DKIM record.
|
||||
Per RFC 6376 3.2, tags are separated by semicolons and folding whitespace
|
||||
is allowed on both sides of the "=". Per 3.6.1, v= is optional and defaults
|
||||
to DKIM1, but MUST be first and equal to DKIM1 when present.
|
||||
Returns None if the record is not a valid DKIM key record.
|
||||
"""
|
||||
parts = [p.strip() for p in value.split(";") if p.strip()]
|
||||
if not parts:
|
||||
return None
|
||||
# v= must be first
|
||||
first = parts[0]
|
||||
if not first.startswith("v=") or first.split("=", 1)[1].strip() != "DKIM1":
|
||||
return None
|
||||
tags = {}
|
||||
for part in parts:
|
||||
if "=" not in part:
|
||||
continue
|
||||
key, val = part.split("=", 1)
|
||||
tags[key.strip()] = val.strip()
|
||||
if not tags:
|
||||
return None
|
||||
if "v" in tags:
|
||||
if tags["v"] != DKIM_VERSION or parts[0].split("=", 1)[0].strip() != "v":
|
||||
return None
|
||||
else:
|
||||
tags["v"] = DKIM_VERSION
|
||||
return tags
|
||||
|
||||
|
||||
@@ -71,6 +77,17 @@ def parse_spf_terms(value: str) -> Optional[Tuple[str, set]]:
|
||||
return (all_mechanism, other_terms)
|
||||
|
||||
|
||||
def _dkim_tag_equal(tag: str, expected: str, found: str) -> bool:
|
||||
"""Compare a single DKIM tag value.
|
||||
|
||||
Per RFC 6376, whitespace inside a base64 value must be ignored. It is
|
||||
significant for other tags, so this cannot be applied globally.
|
||||
"""
|
||||
if tag in DKIM_BASE64_TAGS:
|
||||
return "".join(expected.split()) == "".join(found.split())
|
||||
return expected == found
|
||||
|
||||
|
||||
def _check_dkim_semantic(
|
||||
expected_value: str, found_values: List[str]
|
||||
) -> Optional[Dict[str, any]]:
|
||||
@@ -82,7 +99,10 @@ def _check_dkim_semantic(
|
||||
found_tags = parse_dkim_tags(found_value)
|
||||
if not found_tags:
|
||||
continue
|
||||
if not all(found_tags.get(k) == v for k, v in expected_tags.items()):
|
||||
if not all(
|
||||
k in found_tags and _dkim_tag_equal(k, v, found_tags[k])
|
||||
for k, v in expected_tags.items()
|
||||
):
|
||||
continue
|
||||
# Check for t=y (testing mode) → insecure
|
||||
if found_tags.get("t") and "y" in found_tags["t"].split(":"):
|
||||
|
||||
@@ -833,6 +833,20 @@ class TestParseDkimTags:
|
||||
result = parse_dkim_tags("v=DKIM1; k=rsa; p=MIGfMA0; t=y:s")
|
||||
assert result == {"v": "DKIM1", "k": "rsa", "p": "MIGfMA0", "t": "y:s"}
|
||||
|
||||
def test_whitespace_around_equals(self):
|
||||
"""RFC 6376 3.2 allows folding whitespace on both sides of the '='."""
|
||||
result = parse_dkim_tags("v = DKIM1; k = rsa; p = MIGfMA0")
|
||||
assert result == {"v": "DKIM1", "k": "rsa", "p": "MIGfMA0"}
|
||||
|
||||
def test_missing_v_defaults_to_dkim1(self):
|
||||
"""RFC 6376 3.6.1: v= is optional in a key record and defaults to DKIM1."""
|
||||
result = parse_dkim_tags("k=rsa; p=MIGfMA0")
|
||||
assert result == {"v": "DKIM1", "k": "rsa", "p": "MIGfMA0"}
|
||||
|
||||
def test_record_without_any_tag_returns_none(self):
|
||||
"""A TXT record with no tag=value pair is not a DKIM record."""
|
||||
assert parse_dkim_tags("not a dkim record") is None
|
||||
|
||||
def test_v_not_first_returns_none(self):
|
||||
"""Test that v= not being first tag returns None."""
|
||||
assert parse_dkim_tags("k=rsa; v=DKIM1; p=MIGfMA0") is None
|
||||
@@ -958,6 +972,83 @@ class TestDKIMSemanticComparison:
|
||||
result = check_single_record(maildomain, expected_record)
|
||||
assert result["status"] == "incorrect"
|
||||
|
||||
def test_dkim_key_with_internal_whitespace_is_correct(self, maildomain_factory):
|
||||
"""Whitespace inside the base64 key is not significant (RFC 6376)."""
|
||||
maildomain = maildomain_factory(name="example.com")
|
||||
expected_record = {
|
||||
"type": "TXT",
|
||||
"target": "selector._domainkey",
|
||||
"value": "v=DKIM1; k=rsa; p=MIGfMA0GCSqGSIb3",
|
||||
}
|
||||
|
||||
with patch("core.services.dns.check.dns.resolver.resolve") as mock_resolve:
|
||||
mock_resolve.return_value = _txt_answer(
|
||||
"v=DKIM1; k=rsa; p=MIGfMA0 GCSqGSIb3"
|
||||
)
|
||||
|
||||
result = check_single_record(maildomain, expected_record)
|
||||
assert result["status"] == "correct"
|
||||
|
||||
def test_dkim_whitespace_in_other_tags_is_significant(self, maildomain_factory):
|
||||
"""Internal whitespace outside base64 tags still marks a mismatch."""
|
||||
maildomain = maildomain_factory(name="example.com")
|
||||
expected_record = {
|
||||
"type": "TXT",
|
||||
"target": "selector._domainkey",
|
||||
"value": "v=DKIM1; k=rsa; p=MIGfMA0",
|
||||
}
|
||||
|
||||
with patch("core.services.dns.check.dns.resolver.resolve") as mock_resolve:
|
||||
mock_resolve.return_value = _txt_answer("v=DKIM1; k=r sa; p=MIGfMA0")
|
||||
|
||||
result = check_single_record(maildomain, expected_record)
|
||||
assert result["status"] == "incorrect"
|
||||
|
||||
def test_dkim_whitespace_before_equals_is_correct(self, maildomain_factory):
|
||||
"""Folding whitespace before the '=' is legal, including on v= (RFC 6376)."""
|
||||
maildomain = maildomain_factory(name="example.com")
|
||||
expected_record = {
|
||||
"type": "TXT",
|
||||
"target": "selector._domainkey",
|
||||
"value": "v=DKIM1; k=rsa; p=MIGfMA0",
|
||||
}
|
||||
|
||||
with patch("core.services.dns.check.dns.resolver.resolve") as mock_resolve:
|
||||
mock_resolve.return_value = _txt_answer("v = DKIM1; k=rsa; p = MIGfMA0")
|
||||
|
||||
result = check_single_record(maildomain, expected_record)
|
||||
assert result["status"] == "correct"
|
||||
|
||||
def test_dkim_without_v_tag_is_correct(self, maildomain_factory):
|
||||
"""A key record omitting the optional v= tag is still valid (RFC 6376)."""
|
||||
maildomain = maildomain_factory(name="example.com")
|
||||
expected_record = {
|
||||
"type": "TXT",
|
||||
"target": "selector._domainkey",
|
||||
"value": "v=DKIM1; k=rsa; p=MIGfMA0",
|
||||
}
|
||||
|
||||
with patch("core.services.dns.check.dns.resolver.resolve") as mock_resolve:
|
||||
mock_resolve.return_value = _txt_answer("k=rsa; p=MIGfMA0")
|
||||
|
||||
result = check_single_record(maildomain, expected_record)
|
||||
assert result["status"] == "correct"
|
||||
|
||||
def test_dkim_missing_expected_tag_is_incorrect(self, maildomain_factory):
|
||||
"""A tag present in the expected record but absent in DNS is a mismatch."""
|
||||
maildomain = maildomain_factory(name="example.com")
|
||||
expected_record = {
|
||||
"type": "TXT",
|
||||
"target": "selector._domainkey",
|
||||
"value": "v=DKIM1; k=rsa; p=MIGfMA0",
|
||||
}
|
||||
|
||||
with patch("core.services.dns.check.dns.resolver.resolve") as mock_resolve:
|
||||
mock_resolve.return_value = _txt_answer("v=DKIM1; k=rsa")
|
||||
|
||||
result = check_single_record(maildomain, expected_record)
|
||||
assert result["status"] == "incorrect"
|
||||
|
||||
def test_dkim_multiline_txt_record_with_t_s(self, maildomain_factory):
|
||||
"""Multiline DKIM TXT record (split across quoted strings) with t=s."""
|
||||
maildomain = maildomain_factory(name="example.com")
|
||||
|
||||
Reference in New Issue
Block a user