gitea-admin/gh-christianlouis-dmarq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add evidence-linked DNS health guidance

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-05-23 11:56:51 +02:00
parent 0512f8cfa9
commit a6c795ee16
3 changed files with 426 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/api_v1/endpoints/domains.py
+222 -4
View File
@@ -88,6 +88,47 @@ class DNSRecordResponse(BaseModel):
checkedAt: Optional[str] = None
class DNSHealthEvidence(BaseModel):
"""Evidence backing a DNS health recommendation."""
label: str
value: str
href: str
class DNSHealthCheck(BaseModel):
"""Single DNS health check result."""
key: str
label: str
status: str
message: str
evidence: List[DNSHealthEvidence] = Field(default_factory=list)
class DNSHealthRecommendation(BaseModel):
"""Actionable DNS or enforcement recommendation."""
type: str
severity: str
title: str
detail: str
action: str
evidence: List[DNSHealthEvidence] = Field(default_factory=list)
class DNSHealthResponse(BaseModel):
"""Evidence-linked DNS health summary for a domain."""
status: str
policy: str
compliance_rate: float
total_emails: int
failed_emails: int
checks: List[DNSHealthCheck]
recommendations: List[DNSHealthRecommendation]
class CloudflareZoneResponse(BaseModel):
"""Cloudflare zone available for import."""
@@ -310,6 +351,96 @@ def _domain_names_for_summary(db: Session, store: ReportStore) -> List[str]:
return list(dict.fromkeys(stored_domains + report_domains))
def _domain_exists(db: Session, store: ReportStore, domain_name: str) -> bool:
return domain_name in store.get_domains() or bool(
db.query(Domain.id).filter(Domain.name == domain_name).first()
)
def _record_evidence(label: str, value: Optional[str], href: str = "#dns-records") -> DNSHealthEvidence:
return DNSHealthEvidence(label=label, value=value or "Not found", href=href)
def _summary_evidence(label: str, value: object, href: str = "#compliance-chart") -> DNSHealthEvidence:
return DNSHealthEvidence(label=label, value=str(value), href=href)
def _dns_check(
key: str,
label: str,
present: bool,
present_message: str,
missing_message: str,
evidence: List[DNSHealthEvidence],
) -> DNSHealthCheck:
return DNSHealthCheck(
key=key,
label=label,
status="pass" if present else "fail",
message=present_message if present else missing_message,
evidence=evidence,
)
def _enforcement_recommendation(
policy: str,
summary: Dict[str, Any],
) -> DNSHealthRecommendation:
total = int(summary.get("total_count", 0) or 0)
failed = int(summary.get("failed_count", 0) or 0)
compliance = float(summary.get("compliance_rate", 0.0) or 0.0)
evidence = [
_summary_evidence("Policy", f"p={policy}", "#dns-records"),
_summary_evidence("Total messages", total),
_summary_evidence("Compliance", f"{compliance}%"),
_summary_evidence("Failed messages", failed, "#sending-sources"),
]
if policy != "none":
return DNSHealthRecommendation(
type="policy_already_enforced",
severity="info",
title="DMARC policy is already enforced",
detail="This domain is already beyond monitoring mode.",
action="Continue watching failure trends before tightening further.",
evidence=evidence,
)
if total < 100:
return DNSHealthRecommendation(
type="policy_needs_more_data",
severity="warning",
title="Collect more report volume before enforcement",
detail="DMARQ needs at least 100 observed messages before recommending quarantine.",
action="Keep p=none until more aggregate reports arrive.",
evidence=evidence,
)
if compliance >= 98.0 and failed <= max(2, int(total * 0.02)):
return DNSHealthRecommendation(
type="policy_enforcement_ready",
severity="info",
title="Ready to plan quarantine",
detail="Recent report volume is high and failures are low enough to plan enforcement.",
action="Move gradually: set p=quarantine with a low pct value, then watch failures.",
evidence=evidence,
)
if compliance >= 90.0:
return DNSHealthRecommendation(
type="policy_enforcement_review",
severity="warning",
title="Close remaining failures before enforcement",
detail="Compliance is improving, but failures still need review before policy changes.",
action="Review failing sources and SPF/DKIM alignment before changing p=none.",
evidence=evidence,
)
return DNSHealthRecommendation(
type="policy_not_ready",
severity="error",
title="Not ready for enforcement",
detail="Current DMARC compliance is too low for a safe policy change.",
action="Fix unauthenticated or unknown senders before moving beyond p=none.",
evidence=evidence,
)
@router.get("/summary", response_model=DomainSummaryResponse)
async def get_domains_summary(db: Session = Depends(get_db)):
"""
@@ -540,9 +671,8 @@ async def get_domain_stats(
"""
store = ReportStore.get_instance()
hydrate_report_store_from_db(db, store)
domains = store.get_domains()
domains = _domain_names_for_summary(db, store)
# For Milestone 1, domain_id is simply the domain name
if domain_id not in domains:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
@@ -579,9 +709,8 @@ async def get_domain_dns_records(
"""
store = ReportStore.get_instance()
hydrate_report_store_from_db(db, store)
domains = store.get_domains()
if domain_id not in domains:
if not _domain_exists(db, store, domain_id):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Domain not found",
@@ -612,6 +741,95 @@ async def get_domain_dns_records(
)
@router.get("/{domain_id}/dns/health", response_model=DNSHealthResponse)
async def get_domain_dns_health(
domain_id: str = Path(..., title="The domain ID or name"),
refresh: bool = Query(False, title="Refresh cached DNS result"),
db: Session = Depends(get_db),
):
"""Return evidence-linked DNS health and enforcement readiness guidance."""
store = ReportStore.get_instance()
hydrate_report_store_from_db(db, store)
if not _domain_exists(db, store, domain_id):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Domain not found",
)
manual_selectors = _get_domain_selectors_from_db(db, domain_id)
report_selectors = _get_selectors_from_reports(store, domain_id)
combined_selectors = list(dict.fromkeys(manual_selectors + report_selectors))
provider = get_default_provider(db)
result, _, _ = await resolve_domain_dns_cached(
db,
provider,
domain_id,
selectors=combined_selectors,
refresh=refresh,
)
summary = store.get_domain_summary(domain_id)
policy = extract_dmarc_policy(result.dmarc_record) or "none"
checks = [
_dns_check(
"dmarc",
"DMARC",
result.dmarc,
"DMARC record is published.",
"No DMARC record was found.",
[_record_evidence("DMARC TXT", result.dmarc_record)],
),
_dns_check(
"spf",
"SPF",
result.spf,
"SPF record is published.",
"No SPF record was found at the domain root.",
[_record_evidence("SPF TXT", result.spf_record)],
),
_dns_check(
"dkim",
"DKIM",
result.dkim,
"At least one DKIM selector resolved.",
"No DKIM record was found for configured or observed selectors.",
[
_record_evidence(
"Selectors checked",
", ".join(combined_selectors or result.selectors_checked or []),
),
_record_evidence("DKIM TXT", result.dkim_record),
],
),
]
recommendations: List[DNSHealthRecommendation] = []
for check in checks:
if check.status == "fail":
recommendations.append(
DNSHealthRecommendation(
type=f"missing_{check.key}",
severity="error" if check.key == "dmarc" else "warning",
title=f"{check.label} needs attention",
detail=check.message,
action=f"Publish or repair the {check.label} DNS record, then refresh DNS health.",
evidence=check.evidence,
)
)
recommendations.append(_enforcement_recommendation(policy, summary))
failed_checks = sum(1 for check in checks if check.status == "fail")
health_status = "healthy" if failed_checks == 0 else "degraded" if failed_checks < 3 else "critical"
return DNSHealthResponse(
status=health_status,
policy=policy,
compliance_rate=float(summary.get("compliance_rate", 0.0) or 0.0),
total_emails=int(summary.get("total_count", 0) or 0),
failed_emails=int(summary.get("failed_count", 0) or 0),
checks=checks,
recommendations=recommendations,
)
@router.get("/cloudflare/discover", response_model=List[CloudflareZoneResponse])
async def discover_cloudflare_domains(db: Session = Depends(get_db)):
"""Discover active Cloudflare zones visible to the configured API token."""
backend/app/templates/domain_details.html
+96
View File
@@ -112,6 +112,7 @@
</div>
<!-- Compliance Chart -->
<section id="compliance-chart-section">
{% call card() %}
{% call card_header() %}
{% call card_title() %}Compliance Over Time{% endcall %}
@@ -125,8 +126,74 @@
</div>
{% endcall %}
{% endcall %}
</section>
<!-- DNS Health Summary -->
<section id="dns-health-summary">
{% call card() %}
{% call card_header() %}
<div class="flex items-center justify-between">
{% call card_title() %}DNS Health Summary{% endcall %}
<span class="inline-flex items-center rounded px-2 py-1 text-xs font-semibold capitalize"
:class="dnsHealthStatusClass"
x-text="dnsHealth.status || 'checking'"></span>
</div>
{% call card_description() %}
Evidence-linked authentication posture and enforcement readiness
{% endcall %}
{% endcall %}
{% call card_content() %}
<div class="grid gap-4 lg:grid-cols-3">
<template x-for="check in dnsHealth.checks" :key="check.key">
<div class="rounded border border-base-300 p-4">
<div class="flex items-center justify-between">
<h3 class="font-semibold" x-text="check.label"></h3>
<span class="rounded px-2 py-1 text-xs font-semibold capitalize"
:class="check.status === 'pass' ? 'bg-green-100 text-green-700' : 'bg-red-100 text-red-700'"
x-text="check.status"></span>
</div>
<p class="mt-2 text-sm text-base-content/70" x-text="check.message"></p>
<div class="mt-3 space-y-1">
<template x-for="item in check.evidence" :key="item.label + item.value">
<a :href="item.href" class="block text-xs link link-primary">
<span x-text="item.label"></span>:
<span class="font-mono break-all" x-text="item.value"></span>
</a>
</template>
</div>
</div>
</template>
</div>
<div class="mt-5 space-y-3">
<template x-for="recommendation in dnsHealth.recommendations" :key="recommendation.type">
<div class="rounded border border-base-300 p-4">
<div class="flex items-start gap-3">
<span class="mt-1 h-2.5 w-2.5 rounded-full"
:class="recommendationSeverityClass(recommendation.severity)"></span>
<div class="min-w-0 flex-1">
<h3 class="font-semibold" x-text="recommendation.title"></h3>
<p class="mt-1 text-sm text-base-content/70" x-text="recommendation.detail"></p>
<p class="mt-2 text-sm font-medium" x-text="recommendation.action"></p>
<div class="mt-3 flex flex-wrap gap-2">
<template x-for="item in recommendation.evidence" :key="item.label + item.value">
<a :href="item.href" class="badge badge-outline gap-1">
<span x-text="item.label"></span>
<span x-text="item.value"></span>
</a>
</template>
</div>
</div>
</div>
</div>
</template>
</div>
{% endcall %}
{% endcall %}
</section>
<!-- DNS Records -->
<section id="dns-records">
{% call card() %}
{% call card_header() %}
{% call card_title() %}DNS Records{% endcall %}
@@ -217,8 +284,10 @@
</div>
{% endcall %}
{% endcall %}
</section>
<!-- Source Report Table -->
<section id="sending-sources">
{% call card() %}
{% call card_header() %}
<div class="flex items-center justify-between">
@@ -405,6 +474,7 @@
{% endcall %}
{% endcall %}
{% endcall %}
</section>
<!-- Recent Reports -->
{% call card() %}
@@ -511,6 +581,11 @@ function domainDetailsApp(domainId) {
dkim: false,
dkimSelectors: []
},
dnsHealth: {
status: '',
checks: [],
recommendations: []
},
selectors: [],
reportSelectors: [],
newSelector: '',
@@ -528,6 +603,7 @@ function domainDetailsApp(domainId) {
init() {
this.fetchDomainStats();
this.fetchDNSRecords();
this.fetchDNSHealth();
this.fetchSelectors();
this.fetchReports();
this.fetchSources();
@@ -563,6 +639,13 @@ function domainDetailsApp(domainId) {
return 'Verified';
},
get dnsHealthStatusClass() {
if (this.dnsHealth.status === 'healthy') return 'bg-green-100 text-green-700';
if (this.dnsHealth.status === 'degraded') return 'bg-yellow-100 text-yellow-800';
if (this.dnsHealth.status === 'critical') return 'bg-red-100 text-red-700';
return 'bg-base-200 text-base-content/70';
},
recommendationSeverityClass(severity) {
if (severity === 'error') return 'bg-red-500';
if (severity === 'warning') return 'bg-yellow-500';
@@ -593,6 +676,17 @@ function domainDetailsApp(domainId) {
}
},
async fetchDNSHealth() {
try {
const response = await fetch(`/api/v1/domains/${this.domainId}/dns/health`);
if (response.ok) {
this.dnsHealth = await response.json();
}
} catch (error) {
console.error('Error fetching DNS health:', error);
}
},
async fetchSelectors() {
try {
const response = await fetch(`/api/v1/domains/${this.domainId}/selectors`);
@@ -621,6 +715,7 @@ function domainDetailsApp(domainId) {
// Refresh selectors (both manual and report) and DNS check
this.fetchSelectors();
this.fetchDNSRecords();
this.fetchDNSHealth();
} else {
const err = await response.json();
this.selectorError = err.detail || 'Failed to add selector';
@@ -641,6 +736,7 @@ function domainDetailsApp(domainId) {
// Refresh selectors (both manual and report) and DNS check
this.fetchSelectors();
this.fetchDNSRecords();
this.fetchDNSHealth();
} else {
console.error('Error deleting selector:', response.status);
}
backend/app/tests/test_dns_endpoints.py
+108
View File
@@ -320,6 +320,114 @@ def test_dns_endpoint_404_for_unknown_domain(client: TestClient):
assert response.status_code == 404
def test_dns_endpoint_supports_manually_configured_domain(client: TestClient, db_session):
"""A domain created before reports arrive can still run DNS checks."""
db_session.add(Domain(name="manual.example", active=True))
db_session.commit()
with _mock_dns():
response = client.get("/api/v1/domains/manual.example/dns")
assert response.status_code == 200
assert response.json()["dmarc"] is True
def test_dns_health_404_for_unknown_domain(client: TestClient):
with _mock_dns():
response = client.get("/api/v1/domains/unknown.example.com/dns/health")
assert response.status_code == 404
def test_dns_health_links_checks_to_evidence(client: TestClient):
"""DNS health returns provider-neutral checks, recommendations, and evidence links."""
missing_dkim = DomainDNSResult(
dmarc=True,
dmarc_record="v=DMARC1; p=none; rua=mailto:dmarc@example.com",
spf=True,
spf_record="v=spf1 include:_spf.google.com ~all",
dkim=False,
selectors_checked=["google"],
)
with _mock_dns(result=missing_dkim):
response = client.get(f"/api/v1/domains/{DOMAIN}/dns/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "degraded"
dkim_check = next(check for check in data["checks"] if check["key"] == "dkim")
assert dkim_check["status"] == "fail"
assert dkim_check["evidence"][0]["href"] == "#dns-records"
assert any(item["type"] == "missing_dkim" for item in data["recommendations"])
def test_dns_health_recommends_enforcement_when_evidence_supports_it(client: TestClient):
"""High-volume p=none domains with strong compliance get plan-only guidance."""
store = ReportStore.get_instance()
store.clear()
store.add_report(
{
**MINIMAL_REPORT,
"summary": {"total_count": 500, "passed_count": 495, "failed_count": 5},
"records": [
{
**MINIMAL_REPORT["records"][0],
"count": 500,
"dkim_result": "pass",
"spf_result": "pass",
}
],
}
)
with _mock_dns():
response = client.get(f"/api/v1/domains/{DOMAIN}/dns/health")
assert response.status_code == 200
recommendations = response.json()["recommendations"]
readiness = next(item for item in recommendations if item["type"] == "policy_enforcement_ready")
assert "low pct" in readiness["action"]
assert any(item["label"] == "Compliance" for item in readiness["evidence"])
def test_dns_health_marks_all_missing_records_critical(client: TestClient):
"""Missing DMARC, SPF, and DKIM produce specific repair recommendations."""
missing_all = DomainDNSResult(
dmarc=False,
spf=False,
dkim=False,
selectors_checked=["google"],
)
with _mock_dns(result=missing_all):
response = client.get(f"/api/v1/domains/{DOMAIN}/dns/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "critical"
recommendation_types = {item["type"] for item in data["recommendations"]}
assert {"missing_dmarc", "missing_spf", "missing_dkim"}.issubset(recommendation_types)
assert any(item["type"] == "policy_needs_more_data" for item in data["recommendations"])
@pytest.mark.parametrize(
("policy", "summary", "expected_type", "expected_severity"),
[
("quarantine", {"total_count": 1000, "failed_count": 50, "compliance_rate": 95.0}, "policy_already_enforced", "info"),
("none", {"total_count": 50, "failed_count": 0, "compliance_rate": 100.0}, "policy_needs_more_data", "warning"),
("none", {"total_count": 200, "failed_count": 15, "compliance_rate": 92.5}, "policy_enforcement_review", "warning"),
("none", {"total_count": 200, "failed_count": 80, "compliance_rate": 60.0}, "policy_not_ready", "error"),
],
)
def test_enforcement_recommendation_common_states(policy, summary, expected_type, expected_severity):
"""Policy guidance covers enforced, low-volume, review, and not-ready states."""
recommendation = domains_endpoint._enforcement_recommendation(policy, summary)
assert recommendation.type == expected_type
assert recommendation.severity == expected_severity
assert recommendation.evidence[0].value == f"p={policy}"
# ---------------------------------------------------------------------------
# GET /api/v1/domains/summary (DNS fields included)
# ---------------------------------------------------------------------------