DKIM probe: TXT fallback for Google Workspace (google._domainkey) — fixes the CNAME-only asymmetry

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
 
 

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",