diff --git a/src/mail_municipalities/provider_classification/probes.py b/src/mail_municipalities/provider_classification/probes.py index 164447a..c220319 100644 --- a/src/mail_municipalities/provider_classification/probes.py +++ b/src/mail_municipalities/provider_classification/probes.py @@ -92,26 +92,53 @@ async def probe_spf(domain: str) -> list[Evidence]: async def probe_dkim(domain: str) -> list[Evidence]: - """Query DKIM selector CNAMEs and match targets.""" + """Query DKIM selectors: match CNAME targets, with a TXT fallback for Google. + + Microsoft-style tenants expose DKIM as a CNAME (selector1/selector2 → + ``*.onmicrosoft.com``). Google Workspace instead publishes the DKIM key + directly as a **TXT** record at ``google._domainkey`` (``v=DKIM1; k=rsa; + p=...``), not a CNAME — so a CNAME-only probe can never fire for standard + Google tenants. When the CNAME query for a Google selector yields nothing, + fall back to a TXT query: the selector names ("google", "google2048") are + Google-distinctive, so a v=DKIM1 key there is provider evidence on its own. + """ results: list[Evidence] = [] for sig in SIGNATURES: for selector in sig.dkim_selectors: qname = f"{selector}._domainkey.{domain}" answer = await resolve_robust(qname, "CNAME") - if answer is None: + if answer is not None: + for rdata in answer: + target = str(rdata.target).rstrip(".").lower() + if match_patterns(target, sig.dkim_cname_patterns): + results.append( + Evidence( + kind=SignalKind.DKIM, + provider=sig.provider, + weight=WEIGHTS[SignalKind.DKIM], + detail=f"DKIM {qname} CNAME → {target}", + raw=target, + ) + ) + continue + if sig.provider is not Provider.GOOGLE: continue - for rdata in answer: - target = str(rdata.target).rstrip(".").lower() - if match_patterns(target, sig.dkim_cname_patterns): + txt_answer = await resolve_robust(qname, "TXT") + if txt_answer is None: + continue + for rdata in txt_answer: + txt = b"".join(rdata.strings).decode("utf-8", errors="ignore") + if "v=dkim1" in txt.lower(): results.append( Evidence( kind=SignalKind.DKIM, provider=sig.provider, weight=WEIGHTS[SignalKind.DKIM], - detail=f"DKIM {qname} CNAME → {target}", - raw=target, + detail=f"DKIM {qname} TXT key present (v=DKIM1)", + raw=txt[:120], ) ) + break return results diff --git a/tests/provider_classification/test_probes.py b/tests/provider_classification/test_probes.py index 8f536f2..1a1422b 100644 --- a/tests/provider_classification/test_probes.py +++ b/tests/provider_classification/test_probes.py @@ -213,6 +213,52 @@ async def _resolve(qname, rdtype): results = await probe_dkim("example.com") assert any(e.provider == Provider.GOOGLE for e in results) + async def test_google_txt_fallback_hit(self): + """Google Workspace publishes DKIM as a TXT key (v=DKIM1), not a CNAME.""" + + async def _resolve(qname, rdtype): + if "google._domainkey" in qname and rdtype == "TXT": + return [_txt_rdata("v=DKIM1; k=rsa; p=MIIBIjANBgkqhkiG9w0BAQEFA")] + return None + + with patch( + "mail_municipalities.provider_classification.probes.resolve_robust", + side_effect=_resolve, + ): + results = await probe_dkim("example.com") + assert any( + e.provider == Provider.GOOGLE and e.kind == SignalKind.DKIM + for e in results + ) + + async def test_google_txt_fallback_ignores_non_dkim_txt(self): + async def _resolve(qname, rdtype): + if rdtype == "TXT": + return [_txt_rdata("some unrelated verification token")] + return None + + with patch( + "mail_municipalities.provider_classification.probes.resolve_robust", + side_effect=_resolve, + ): + results = await probe_dkim("example.com") + assert results == [] + + async def test_txt_fallback_only_for_google_selectors(self): + """A stray v=DKIM1 TXT at a Microsoft-style CNAME selector must not classify.""" + + async def _resolve(qname, rdtype): + if qname.startswith("selector1.") and rdtype == "TXT": + return [_txt_rdata("v=DKIM1; k=rsa; p=abc")] + return None + + with patch( + "mail_municipalities.provider_classification.probes.resolve_robust", + side_effect=_resolve, + ): + results = await probe_dkim("example.com") + assert results == [] + async def test_no_match(self): with patch( "mail_municipalities.provider_classification.probes.resolve_robust",