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>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-29 22:46:49 +00:00
parent 30007d21d6
commit 58b5755f72
2 changed files with 30 additions and 23 deletions
+29 -22
View File
@@ -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),
)
)
+1 -1
View File
@@ -338,7 +338,7 @@
>
<p class="font-semibold mb-1">SPF Fix Suggestion</p>
<p class="text-xs text-muted-foreground mb-2">
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:
</p>
<div class="flex items-center gap-2 bg-base-200 rounded px-2 py-1">
<code class="flex-1 font-mono text-xs" x-text="source.spf_fix_hint"></code>