Merge pull request #72 from christianlouis/copilot/make-domain-report-more-actionable
feat: enhance domain sources with rDNS hostname, SPF fix hints, and auth tooltips
This commit is contained in:
Christian Krakau-Louis
@@ -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):
|
||||
@@ -434,13 +437,41 @@ 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"),
|
||||
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()
|
||||
@@ -451,23 +482,27 @@ 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()
|
||||
|
||||
ips = [s.get("source_ip", "unknown") for s in sources]
|
||||
hostnames = await asyncio.gather(*[_safe_ptr_lookup(provider, 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")
|
||||
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(ip, spf_result),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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() %}
|
||||
<template x-if="sources.length === 0">
|
||||
<tr>
|
||||
<td colspan="6" class="text-center py-4">
|
||||
<td colspan="7" class="text-center py-4">
|
||||
<div class="text-muted-foreground">No data available for this time period</div>
|
||||
</td>
|
||||
</tr>
|
||||
@@ -253,39 +254,60 @@
|
||||
<template x-for="source in filteredSources" :key="source.ip">
|
||||
{% call tr() %}
|
||||
{% call td() %}
|
||||
<span x-text="source.ip"></span>
|
||||
<div class="flex flex-col">
|
||||
<span class="font-mono text-sm" x-text="source.ip"></span>
|
||||
<template x-if="source.hostname">
|
||||
<span class="text-xs text-muted-foreground" x-text="source.hostname"></span>
|
||||
</template>
|
||||
</div>
|
||||
{% endcall %}
|
||||
{% call td() %}
|
||||
<span x-text="source.count"></span>
|
||||
{% endcall %}
|
||||
{% call td() %}
|
||||
<template x-if="source.spf === 'pass'">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-green-100 text-green-800">Pass</span>
|
||||
<div class="tooltip" data-tip="IP is authorized in the SPF record">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-green-100 text-green-800">Pass</span>
|
||||
</div>
|
||||
</template>
|
||||
<template x-if="source.spf === 'fail'">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-red-100 text-red-800">Fail</span>
|
||||
<div class="tooltip" data-tip="IP is not listed in the SPF record">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-red-100 text-red-800">Fail</span>
|
||||
</div>
|
||||
</template>
|
||||
<template x-if="source.spf === 'neutral' || source.spf === 'none'">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-gray-100 text-gray-800" x-text="source.spf"></span>
|
||||
<div class="tooltip" :data-tip="'SPF returned: ' + source.spf">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-gray-100 text-gray-800" x-text="source.spf"></span>
|
||||
</div>
|
||||
</template>
|
||||
{% endcall %}
|
||||
{% call td() %}
|
||||
<template x-if="source.dkim === 'pass'">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-green-100 text-green-800">Pass</span>
|
||||
<div class="tooltip" data-tip="Valid DKIM signature found">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-green-100 text-green-800">Pass</span>
|
||||
</div>
|
||||
</template>
|
||||
<template x-if="source.dkim === 'fail'">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-red-100 text-red-800">Fail</span>
|
||||
<div class="tooltip" data-tip="No valid DKIM signature for this domain">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-red-100 text-red-800">Fail</span>
|
||||
</div>
|
||||
</template>
|
||||
<template x-if="source.dkim === 'neutral' || source.dkim === 'none'">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-gray-100 text-gray-800" x-text="source.dkim"></span>
|
||||
<div class="tooltip" :data-tip="'DKIM returned: ' + source.dkim">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-gray-100 text-gray-800" x-text="source.dkim"></span>
|
||||
</div>
|
||||
</template>
|
||||
{% endcall %}
|
||||
{% call td() %}
|
||||
<template x-if="source.dmarc === 'pass'">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-green-100 text-green-800">Pass</span>
|
||||
<div class="tooltip" data-tip="DMARC passed: SPF or DKIM alignment succeeded">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-green-100 text-green-800">Pass</span>
|
||||
</div>
|
||||
</template>
|
||||
<template x-if="source.dmarc === 'fail'">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-red-100 text-red-800">Fail</span>
|
||||
<div class="tooltip" data-tip="DMARC failed: neither SPF nor DKIM alignment passed">
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-red-100 text-red-800">Fail</span>
|
||||
</div>
|
||||
</template>
|
||||
{% endcall %}
|
||||
{% call td() %}
|
||||
@@ -299,6 +321,42 @@
|
||||
<span class="inline-flex items-center px-2 py-1 rounded-full text-xs bg-red-100 text-red-800">Reject</span>
|
||||
</template>
|
||||
{% endcall %}
|
||||
{% call td() %}
|
||||
<template x-if="source.spf_fix_hint">
|
||||
<div x-data="{ open: false }" class="relative">
|
||||
<button
|
||||
@click="open = !open"
|
||||
class="btn btn-xs btn-warning"
|
||||
title="Show SPF fix suggestion"
|
||||
>
|
||||
Fix SPF
|
||||
</button>
|
||||
<div
|
||||
x-show="open"
|
||||
@click.outside="open = false"
|
||||
class="absolute right-0 z-20 mt-1 w-72 bg-base-100 border border-base-300 rounded-lg shadow-lg p-3 text-sm"
|
||||
>
|
||||
<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 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>
|
||||
<button
|
||||
@click="navigator.clipboard.writeText(source.spf_fix_hint)"
|
||||
class="btn btn-xs btn-ghost"
|
||||
title="Copy to clipboard"
|
||||
>
|
||||
<svg xmlns="http://www.w3.org/2000/svg" width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><rect width="14" height="14" x="8" y="8" rx="2" ry="2"/><path d="M4 16c-1.1 0-2-.9-2-2V4c0-1.1.9-2 2-2h10c1.1 0 2 .9 2 2"/></svg>
|
||||
</button>
|
||||
</div>
|
||||
<p class="text-xs text-muted-foreground mt-2">
|
||||
Example: <code class="font-mono" x-text="'v=spf1 ' + source.spf_fix_hint + ' ~all'"></code>
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
</template>
|
||||
{% endcall %}
|
||||
{% endcall %}
|
||||
</template>
|
||||
{% endcall %}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user