From daa38b8337b8d0392794ea523feac6a27ab8ca2b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 29 Mar 2026 22:37:47 +0000 Subject: [PATCH 1/3] Initial plan From 30007d21d6f78f80b5bf5ea9793d313b40ca7a60 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 29 Mar 2026 22:44:18 +0000 Subject: [PATCH 2/3] feat: add rDNS hostname, SPF fix hints and auth tooltips to domain sources view - dns_resolver.py: add _ip_to_arpa_name() helper and lookup_ptr() to BaseDNSProvider (no-op default), SystemDNSProvider (dnspython PTR), and CloudflareDNSProvider (DoH PTR type=12) - domains.py: extend SourceEntry with hostname + spf_fix_hint; update get_domain_sources to run async PTR lookups and generate ip4:/ip6: SPF mechanism hints for failing IPs - domain_details.html: show rDNS hostname below IP in sources table; add DaisyUI tooltip explaining each auth result; add "Fix SPF" popover with copy-paste mechanism for IPs that fail SPF - tests: 15 new tests covering _ip_to_arpa_name, SystemDNSProvider/ CloudflareDNSProvider PTR lookup, and sources endpoint hostname + fix-hint fields Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/9eaa7749-047c-46bd-8bc0-2851ea02ffe4 Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- backend/app/api/api_v1/endpoints/domains.py | 48 +++++-- backend/app/services/dns_resolver.py | 75 +++++++++- backend/app/templates/domain_details.html | 80 +++++++++-- backend/app/tests/test_dns_endpoints.py | 91 +++++++++++++ backend/app/tests/test_dns_resolver.py | 144 ++++++++++++++++++++ 5 files changed, 416 insertions(+), 22 deletions(-) diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py index f65d43d..9a1dc76 100644 --- a/backend/app/api/api_v1/endpoints/domains.py +++ b/backend/app/api/api_v1/endpoints/domains.py @@ -1,4 +1,5 @@ import asyncio +import ipaddress import logging from datetime import datetime, timezone from typing import Any, Dict, List, Optional @@ -85,6 +86,8 @@ class SourceEntry(BaseModel): dkim: str dmarc: str disposition: str + hostname: Optional[str] = None + spf_fix_hint: Optional[str] = None class DomainReportsResponse(BaseModel): @@ -440,7 +443,8 @@ async def get_domain_sources( days: int = Query(30, title="Number of days to look back"), ): """ - Get sending sources for a specific domain + Get sending sources for a specific domain, including reverse-DNS hostnames + and SPF fix hints for sources that fail authentication. """ store = ReportStore.get_instance() domains = store.get_domains() @@ -454,20 +458,44 @@ async def get_domain_sources( # Get sending sources for this domain sources = store.get_domain_sources(domain_id, days=days) + provider = get_default_provider() + + async def _safe_ptr(ip: str) -> Optional[str]: + """Perform a PTR lookup with a short timeout; return None on any failure.""" + try: + return await asyncio.wait_for(provider.lookup_ptr(ip), timeout=3.0) + except Exception: + return None + + ips = [s.get("source_ip", "unknown") for s in sources] + hostnames = await asyncio.gather(*[_safe_ptr(ip) for ip in ips]) + source_entries = [] - for source in sources: + for source, hostname in zip(sources, hostnames): + ip = source.get("source_ip", "unknown") + spf_result = source.get("spf_result", "unknown") + dkim_result = source.get("dkim_result", "unknown") + + # Build a copy-paste SPF mechanism for IPs that fail SPF + spf_fix_hint: Optional[str] = None + if spf_result == "fail": + try: + addr = ipaddress.ip_address(ip) + prefix = "ip6" if isinstance(addr, ipaddress.IPv6Address) else "ip4" + spf_fix_hint = f"{prefix}:{ip}" + except ValueError: + pass + source_entries.append( SourceEntry( - ip=source.get("source_ip", "unknown"), + ip=ip, count=source.get("count", 0), - spf=source.get("spf_result", "unknown"), - dkim=source.get("dkim_result", "unknown"), - dmarc=( - "pass" - if source.get("spf_result") == "pass" or source.get("dkim_result") == "pass" - else "fail" - ), + spf=spf_result, + dkim=dkim_result, + dmarc=("pass" if spf_result == "pass" or dkim_result == "pass" else "fail"), disposition=source.get("disposition", "none"), + hostname=hostname, + spf_fix_hint=spf_fix_hint, ) ) diff --git a/backend/app/services/dns_resolver.py b/backend/app/services/dns_resolver.py index c290b0d..0dd1d84 100644 --- a/backend/app/services/dns_resolver.py +++ b/backend/app/services/dns_resolver.py @@ -1,5 +1,5 @@ """ -DNS resolver service for DMARC, SPF, and DKIM record lookups. +DNS resolver service for DMARC, SPF, DKIM, and PTR record lookups. Provides an extensible provider architecture so that DNS data can be fetched either via the system resolver (dnspython) or via the Cloudflare DNS API for @@ -7,6 +7,7 @@ future Cloudflare integration. """ import asyncio +import ipaddress import logging from abc import ABC, abstractmethod from dataclasses import dataclass, field @@ -20,6 +21,23 @@ def _sanitize_for_log(value: str) -> str: return value.replace("\r", "").replace("\n", "") +def _ip_to_arpa_name(ip: str) -> str: + """Convert an IP address string to its reverse-DNS ARPA lookup name. + + E.g. ``"1.2.3.4"`` → ``"4.3.2.1.in-addr.arpa"`` + ``"2001:db8::1"`` → ``"...ip6.arpa"`` + + Raises ``ValueError`` for invalid IP address strings. + """ + addr = ipaddress.ip_address(ip) + if isinstance(addr, ipaddress.IPv4Address): + parts = ip.split(".") + return ".".join(reversed(parts)) + ".in-addr.arpa" + # IPv6: expand, strip colons, reverse nibbles + expanded = addr.exploded.replace(":", "") + return ".".join(reversed(expanded)) + ".ip6.arpa" + + # Well-known DKIM selectors tried when no selectors are configured COMMON_DKIM_SELECTORS: List[str] = [ "default", @@ -103,6 +121,16 @@ class BaseDNSProvider(ABC): logger.debug("SPF lookup failed for %s: %s", _sanitize_for_log(domain), exc) return False, None + async def lookup_ptr(self, ip: str) -> Optional[str]: + """Return the PTR (reverse DNS) hostname for *ip*, or ``None`` if unavailable. + + The base implementation always returns ``None``. Concrete providers + override this to perform an actual DNS PTR lookup so that existing + test doubles (which only implement ``lookup_txt``) keep working without + modification. + """ + return None + async def check_dkim( self, domain: str, selectors: List[str] ) -> Tuple[bool, Optional[str], Optional[str]]: @@ -180,6 +208,23 @@ class SystemDNSProvider(BaseDNSProvider): except dns.exception.DNSException as exc: raise LookupError(f"TXT lookup failed for {name}: {exc}") from exc + async def lookup_ptr(self, ip: str) -> Optional[str]: + """Resolve a PTR record for *ip* via the system resolver.""" + import dns.asyncresolver # type: ignore[import] + import dns.exception # type: ignore[import] + + try: + ptr_name = _ip_to_arpa_name(ip) + answers = await dns.asyncresolver.resolve( + ptr_name, "PTR", lifetime=DNS_TIMEOUT, raise_on_no_answer=False + ) + if answers: + for rdata in answers: + return str(rdata).rstrip(".") + except (dns.exception.DNSException, ValueError): + pass + return None + class CloudflareDNSProvider(BaseDNSProvider): """DNS provider using Cloudflare's DNS-over-HTTPS (DoH) endpoint. @@ -246,6 +291,34 @@ class CloudflareDNSProvider(BaseDNSProvider): except (httpx.RequestError, httpx.HTTPStatusError, httpx.TimeoutException) as exc: raise LookupError(f"Cloudflare DoH lookup failed for {name}: {exc}") from exc + async def lookup_ptr(self, ip: str) -> Optional[str]: + """Resolve a PTR record for *ip* via Cloudflare's DoH endpoint.""" + import httpx # type: ignore[import] + + try: + ptr_name = _ip_to_arpa_name(ip) + except ValueError: + return None + + params = {"name": ptr_name, "type": "PTR"} + headers = {"Accept": "application/dns-json"} + try: + async with httpx.AsyncClient() as client: + response = await client.get( + self.CLOUDFLARE_DOH_URL, + params=params, + headers=headers, + timeout=DNS_TIMEOUT, + ) + response.raise_for_status() + data = response.json() + for answer in data.get("Answer", []): + if answer.get("type") == 12: # PTR record type + return answer.get("data", "").rstrip(".") + except (httpx.RequestError, httpx.HTTPStatusError, httpx.TimeoutException): + pass + return None + def get_default_provider() -> BaseDNSProvider: """Return the default DNS provider (system resolver). diff --git a/backend/app/templates/domain_details.html b/backend/app/templates/domain_details.html index 12d39f9..7bd23c5 100644 --- a/backend/app/templates/domain_details.html +++ b/backend/app/templates/domain_details.html @@ -234,18 +234,19 @@ {% call table() %} {% call thead() %} {% call tr() %} - {% call th() %}Source IP{% endcall %} + {% call th() %}Source IP / Hostname{% endcall %} {% call th() %}Total Emails{% endcall %} {% call th() %}SPF{% endcall %} {% call th() %}DKIM{% endcall %} {% call th() %}DMARC{% endcall %} {% call th() %}Disposition{% endcall %} + {% call th() %}Fix{% endcall %} {% endcall %} {% endcall %} {% call tbody() %} {% endcall %} diff --git a/backend/app/tests/test_dns_endpoints.py b/backend/app/tests/test_dns_endpoints.py index 8326372..8ae9bbd 100644 --- a/backend/app/tests/test_dns_endpoints.py +++ b/backend/app/tests/test_dns_endpoints.py @@ -250,3 +250,94 @@ def test_summary_dns_failure_defaults_false(client: TestClient): assert domain["dmarc_status"] is False assert domain["spf_status"] is False assert domain["dkim_status"] is False + + +# --------------------------------------------------------------------------- +# GET /api/v1/domains/{domain_id}/sources (PTR + fix hints) +# --------------------------------------------------------------------------- + +# A failing-source report used for sources tests +FAILING_SOURCE_REPORT = { + "domain": DOMAIN, + "report_id": "fail-src-001", + "org_name": "Fail Org", + "policy": {"p": "reject", "sp": "", "pct": "100"}, + "records": [ + { + "source_ip": "10.0.0.1", + "count": 3, + "disposition": "reject", + "dkim_result": "fail", + "spf_result": "fail", + "dkim": [], + "spf": [], + } + ], + "summary": {"total_count": 3, "passed_count": 0, "failed_count": 3, "pass_rate": 0.0}, +} + + +def _mock_provider(hostname=None): + """Return a context manager that patches get_default_provider with a PTR mock.""" + mock_prov = AsyncMock() + mock_prov.check_domain = AsyncMock(return_value=MOCK_DNS_RESULT) + mock_prov.lookup_ptr = AsyncMock(return_value=hostname) + return patch( + "app.api.api_v1.endpoints.domains.get_default_provider", + return_value=mock_prov, + ) + + +def test_sources_endpoint_includes_hostname(client: TestClient): + """The /sources endpoint should return the rDNS hostname when available.""" + store = ReportStore.get_instance() + store.add_report(FAILING_SOURCE_REPORT) + + with _mock_provider(hostname="mail.example.com"): + response = client.get(f"/api/v1/domains/{DOMAIN}/sources") + + assert response.status_code == 200 + sources = response.json()["sources"] + # Find the failing source + failing = next((s for s in sources if s["ip"] == "10.0.0.1"), None) + assert failing is not None + assert failing["hostname"] == "mail.example.com" + + +def test_sources_endpoint_hostname_none_when_no_ptr(client: TestClient): + """The /sources endpoint should return null hostname when no PTR record exists.""" + with _mock_provider(hostname=None): + response = client.get(f"/api/v1/domains/{DOMAIN}/sources") + + assert response.status_code == 200 + sources = response.json()["sources"] + for source in sources: + # hostname may be null; it must not crash + assert "hostname" in source + + +def test_sources_endpoint_spf_fix_hint_for_failing_ip(client: TestClient): + """A source with spf=fail should receive an spf_fix_hint containing its IP.""" + store = ReportStore.get_instance() + store.add_report(FAILING_SOURCE_REPORT) + + with _mock_provider(): + response = client.get(f"/api/v1/domains/{DOMAIN}/sources") + + assert response.status_code == 200 + sources = response.json()["sources"] + failing = next((s for s in sources if s["ip"] == "10.0.0.1"), None) + assert failing is not None + assert failing["spf_fix_hint"] == "ip4:10.0.0.1" + + +def test_sources_endpoint_no_fix_hint_when_spf_passes(client: TestClient): + """A source with spf=pass should not receive an spf_fix_hint.""" + with _mock_provider(): + response = client.get(f"/api/v1/domains/{DOMAIN}/sources") + + assert response.status_code == 200 + sources = response.json()["sources"] + passing = next((s for s in sources if s["ip"] == "1.2.3.4"), None) + if passing is not None: + assert passing["spf_fix_hint"] is None diff --git a/backend/app/tests/test_dns_resolver.py b/backend/app/tests/test_dns_resolver.py index 127332a..3216d45 100644 --- a/backend/app/tests/test_dns_resolver.py +++ b/backend/app/tests/test_dns_resolver.py @@ -267,3 +267,147 @@ async def test_cloudflare_provider_raises_on_http_error(): def test_get_default_provider_returns_system(): provider = get_default_provider() assert isinstance(provider, SystemDNSProvider) + + +# --------------------------------------------------------------------------- +# _ip_to_arpa_name helper +# --------------------------------------------------------------------------- + + +def test_ip_to_arpa_name_ipv4(): + from app.services.dns_resolver import _ip_to_arpa_name + + assert _ip_to_arpa_name("1.2.3.4") == "4.3.2.1.in-addr.arpa" + + +def test_ip_to_arpa_name_ipv4_leading_zero_safe(): + from app.services.dns_resolver import _ip_to_arpa_name + + assert _ip_to_arpa_name("192.168.1.100") == "100.1.168.192.in-addr.arpa" + + +def test_ip_to_arpa_name_ipv6(): + from app.services.dns_resolver import _ip_to_arpa_name + + # 2001:db8::1 expanded → 20010db8000000000000000000000001 + name = _ip_to_arpa_name("2001:db8::1") + assert name.endswith(".ip6.arpa") + + +def test_ip_to_arpa_name_invalid_raises(): + from app.services.dns_resolver import _ip_to_arpa_name + + with pytest.raises(ValueError): + _ip_to_arpa_name("not-an-ip") + + +# --------------------------------------------------------------------------- +# BaseDNSProvider.lookup_ptr (default returns None) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_base_provider_lookup_ptr_returns_none(): + """FakeDNSProvider only implements lookup_txt; lookup_ptr must return None.""" + provider = FakeDNSProvider({}) + result = await provider.lookup_ptr("1.2.3.4") + assert result is None + + +# --------------------------------------------------------------------------- +# SystemDNSProvider.lookup_ptr +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_system_provider_lookup_ptr_returns_hostname(): + """SystemDNSProvider.lookup_ptr should decode the first PTR rdata.""" + + class FakePTRRdata: + def __str__(self): + return "mail.example.com." + + class FakePTRAnswers: + def __iter__(self): + return iter([FakePTRRdata()]) + + with patch("dns.asyncresolver.resolve", new=AsyncMock(return_value=FakePTRAnswers())): + provider = SystemDNSProvider() + hostname = await provider.lookup_ptr("1.2.3.4") + + # Trailing dot should be stripped + assert hostname == "mail.example.com" + + +@pytest.mark.asyncio +async def test_system_provider_lookup_ptr_returns_none_on_nxdomain(): + import dns.exception # type: ignore[import] + + with patch( + "dns.asyncresolver.resolve", + new=AsyncMock(side_effect=dns.exception.DNSException("NXDOMAIN")), + ): + provider = SystemDNSProvider() + hostname = await provider.lookup_ptr("1.2.3.4") + + assert hostname is None + + +@pytest.mark.asyncio +async def test_system_provider_lookup_ptr_returns_none_for_invalid_ip(): + provider = SystemDNSProvider() + result = await provider.lookup_ptr("not-an-ip") + assert result is None + + +# --------------------------------------------------------------------------- +# CloudflareDNSProvider.lookup_ptr +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_cloudflare_provider_lookup_ptr_returns_hostname(): + """CloudflareDNSProvider.lookup_ptr should extract the PTR name from DoH JSON.""" + from unittest.mock import MagicMock + + fake_response_data = { + "Answer": [ + {"type": 12, "data": "mail.example.com."}, + {"type": 1, "data": "93.184.216.34"}, # A record — should be ignored + ] + } + + mock_response = AsyncMock() + mock_response.raise_for_status = MagicMock() + mock_response.json = lambda: fake_response_data + + with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)): + provider = CloudflareDNSProvider() + hostname = await provider.lookup_ptr("1.2.3.4") + + assert hostname == "mail.example.com" + + +@pytest.mark.asyncio +async def test_cloudflare_provider_lookup_ptr_returns_none_when_no_ptr(): + """CloudflareDNSProvider.lookup_ptr returns None when no PTR answer exists.""" + from unittest.mock import MagicMock + + fake_response_data = {"Answer": []} + + mock_response = AsyncMock() + mock_response.raise_for_status = MagicMock() + mock_response.json = lambda: fake_response_data + + with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)): + provider = CloudflareDNSProvider() + hostname = await provider.lookup_ptr("1.2.3.4") + + assert hostname is None + + +@pytest.mark.asyncio +async def test_cloudflare_provider_lookup_ptr_returns_none_for_invalid_ip(): + provider = CloudflareDNSProvider() + result = await provider.lookup_ptr("not-an-ip") + assert result is None From 58b5755f72f18f2b63c04cb5d47f0ffcd17dd45d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 29 Mar 2026 22:46:49 +0000 Subject: [PATCH 3/3] refactor: extract _spf_fix_hint and _safe_ptr_lookup helpers; fix spelling Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/9eaa7749-047c-46bd-8bc0-2851ea02ffe4 Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- backend/app/api/api_v1/endpoints/domains.py | 51 ++++++++++++--------- backend/app/templates/domain_details.html | 2 +- 2 files changed, 30 insertions(+), 23 deletions(-) diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py index 9a1dc76..595b75e 100644 --- a/backend/app/api/api_v1/endpoints/domains.py +++ b/backend/app/api/api_v1/endpoints/domains.py @@ -437,6 +437,33 @@ def _build_compliance_timeline(store: ReportStore, domain: str) -> List[Timeline return timeline +def _spf_fix_hint(ip: str, spf_result: str) -> Optional[str]: + """Return a copy-paste SPF mechanism (e.g. ``ip4:1.2.3.4``) for a failing IP. + + Returns ``None`` when SPF did not fail or when *ip* is not a valid address. + """ + if spf_result != "fail": + return None + try: + addr = ipaddress.ip_address(ip) + prefix = "ip6" if isinstance(addr, ipaddress.IPv6Address) else "ip4" + return f"{prefix}:{ip}" + except ValueError: + return None + + +async def _safe_ptr_lookup(provider: Any, ip: str, timeout: float = 3.0) -> Optional[str]: + """Perform a PTR lookup for *ip*, returning ``None`` on any error or timeout.""" + try: + ipaddress.ip_address(ip) # validate before making a DNS query + except ValueError: + return None + try: + return await asyncio.wait_for(provider.lookup_ptr(ip), timeout=timeout) + except Exception: + return None + + @router.get("/{domain_id}/sources", response_model=DomainSourcesResponse) async def get_domain_sources( domain_id: str = Path(..., title="The domain ID or name"), @@ -455,37 +482,17 @@ async def get_domain_sources( detail="Domain not found", ) - # Get sending sources for this domain sources = store.get_domain_sources(domain_id, days=days) - provider = get_default_provider() - async def _safe_ptr(ip: str) -> Optional[str]: - """Perform a PTR lookup with a short timeout; return None on any failure.""" - try: - return await asyncio.wait_for(provider.lookup_ptr(ip), timeout=3.0) - except Exception: - return None - ips = [s.get("source_ip", "unknown") for s in sources] - hostnames = await asyncio.gather(*[_safe_ptr(ip) for ip in ips]) + hostnames = await asyncio.gather(*[_safe_ptr_lookup(provider, ip) for ip in ips]) source_entries = [] for source, hostname in zip(sources, hostnames): ip = source.get("source_ip", "unknown") spf_result = source.get("spf_result", "unknown") dkim_result = source.get("dkim_result", "unknown") - - # Build a copy-paste SPF mechanism for IPs that fail SPF - spf_fix_hint: Optional[str] = None - if spf_result == "fail": - try: - addr = ipaddress.ip_address(ip) - prefix = "ip6" if isinstance(addr, ipaddress.IPv6Address) else "ip4" - spf_fix_hint = f"{prefix}:{ip}" - except ValueError: - pass - source_entries.append( SourceEntry( ip=ip, @@ -495,7 +502,7 @@ async def get_domain_sources( dmarc=("pass" if spf_result == "pass" or dkim_result == "pass" else "fail"), disposition=source.get("disposition", "none"), hostname=hostname, - spf_fix_hint=spf_fix_hint, + spf_fix_hint=_spf_fix_hint(ip, spf_result), ) ) diff --git a/backend/app/templates/domain_details.html b/backend/app/templates/domain_details.html index 7bd23c5..3f6d43f 100644 --- a/backend/app/templates/domain_details.html +++ b/backend/app/templates/domain_details.html @@ -338,7 +338,7 @@ >

SPF Fix Suggestion

- Add the following mechanism to your SPF TXT record to authorise this server: + Add the following mechanism to your SPF TXT record to authorize this server: