From 6681aa4da06c46f382140b7b43234cddec79047c Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Sat, 23 May 2026 17:05:05 +0200 Subject: [PATCH] feat: add BIMI posture readiness checks --- backend/app/api/api_v1/endpoints/domains.py | 178 ++++++++++++++++- backend/app/services/bimi.py | 203 ++++++++++++++++++++ backend/app/templates/domain_details.html | 46 +++++ backend/app/tests/test_bimi.py | 118 ++++++++++++ backend/app/tests/test_dns_endpoints.py | 68 +++++++ docs/milestones.md | 2 +- docs/reference/api.md | 9 + docs/user_guide/domains.md | 15 ++ 8 files changed, 631 insertions(+), 8 deletions(-) create mode 100644 backend/app/services/bimi.py create mode 100644 backend/app/tests/test_bimi.py diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py index 2c9274d..a852094 100644 --- a/backend/app/api/api_v1/endpoints/domains.py +++ b/backend/app/api/api_v1/endpoints/domains.py @@ -15,6 +15,7 @@ from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import require_admin_auth from app.models.domain import Domain +from app.services.bimi import BIMIResult, check_bimi_cached from app.services.cloudflare_dns import ( analyze_dns_records, discover_cloudflare_zones, @@ -146,6 +147,22 @@ class MTAStsResponse(BaseModel): checked_at: Optional[str] = None +class BIMIResponse(BaseModel): + """BIMI posture result for a domain.""" + + status: str + selector: str = "default" + query_name: str + dns_record: Optional[str] = None + logo_url: Optional[str] = None + certificate_url: Optional[str] = None + evidence_url: Optional[str] = None + errors: List[str] = Field(default_factory=list) + warnings: List[str] = Field(default_factory=list) + cached: bool = False + checked_at: Optional[str] = None + + class CloudflareZoneResponse(BaseModel): """Cloudflare zone available for import.""" @@ -462,6 +479,100 @@ def _enforcement_recommendation( ) +def _dmarc_tags(record: Optional[str]) -> Dict[str, str]: + if not record: + return {} + return { + part.split("=", 1)[0].strip().lower(): part.split("=", 1)[1].strip().lower() + for part in record.split(";") + if "=" in part + } + + +def _bimi_dmarc_readiness(record: Optional[str]) -> tuple[bool, List[str], List[DNSHealthEvidence]]: + tags = _dmarc_tags(record) + policy = tags.get("p", "none") + subdomain_policy = tags.get("sp") + pct = tags.get("pct", "100") + issues = [] + if policy not in {"quarantine", "reject"}: + issues.append("DMARC policy must be p=quarantine or p=reject for BIMI.") + if pct != "100": + issues.append("DMARC pct must be 100 or omitted for BIMI.") + if subdomain_policy and subdomain_policy not in {"quarantine", "reject"}: + issues.append("DMARC subdomain policy must also be enforced when sp= is present.") + evidence = [ + _record_evidence("DMARC TXT", record), + _summary_evidence("Policy", f"p={policy}", "#dns-records"), + _summary_evidence( + "Subdomain policy", f"sp={subdomain_policy or 'inherit'}", "#dns-records" + ), + _summary_evidence("Percentage", f"pct={pct}", "#dns-records"), + ] + return not issues, issues, evidence + + +def _bimi_check(result: BIMIResult, dmarc_ready: bool) -> DNSHealthCheck: + evidence = [ + _record_evidence("BIMI TXT", result.dns_record, "#bimi-posture"), + _record_evidence("Logo URL", result.logo_url, "#bimi-posture"), + ] + if result.certificate_url: + evidence.append( + _record_evidence("Certificate URL", result.certificate_url, "#bimi-posture") + ) + if result.status == "pass" and dmarc_ready: + message = "BIMI record is published and DMARC is enforcement-ready." + elif result.status == "pass": + message = "BIMI record is published, but DMARC enforcement is not ready." + else: + message = result.errors[0] if result.errors else "BIMI posture needs attention." + return DNSHealthCheck( + key="bimi", + label="BIMI", + status="pass" if result.status == "pass" and dmarc_ready else "fail", + message=message, + evidence=evidence, + ) + + +def _bimi_recommendation( + result: BIMIResult, + dmarc_ready: bool, + dmarc_issues: List[str], + dmarc_evidence: List[DNSHealthEvidence], +) -> Optional[DNSHealthRecommendation]: + bimi_evidence = _bimi_check(result, dmarc_ready).evidence + if result.status == "pass" and dmarc_ready and not result.warnings: + return None + if result.status == "pass" and not dmarc_ready: + return DNSHealthRecommendation( + type="bimi_dmarc_not_ready", + severity="warning", + title="DMARC enforcement is blocking BIMI readiness", + detail="; ".join(dmarc_issues), + action="Move DMARC to quarantine or reject at pct=100 before relying on BIMI.", + evidence=dmarc_evidence + bimi_evidence, + ) + if result.status == "pass": + return DNSHealthRecommendation( + type="bimi_review", + severity="info", + title="BIMI record needs provider-readiness review", + detail="; ".join(result.warnings), + action="Confirm the SVG logo profile and add a certificate URL if mailbox providers require one.", + evidence=bimi_evidence, + ) + return DNSHealthRecommendation( + type="missing_bimi", + severity="info", + title="Publish BIMI after DMARC enforcement", + detail="; ".join(result.errors or ["No BIMI record is published."]), + action="Publish a BIMI TXT record at default._bimi with an HTTPS SVG logo URL.", + evidence=dmarc_evidence + bimi_evidence, + ) + + def _mta_sts_check(result: MTAStsResult) -> DNSHealthCheck: evidence = [ _record_evidence("MTA-STS TXT", result.dns_record, "#dns-records"), @@ -840,8 +951,17 @@ async def get_domain_dns_health( domain_id, refresh=refresh, ) + bimi_result, _, _ = await check_bimi_cached( + db, + provider, + domain_id, + refresh=refresh, + ) summary = store.get_domain_summary(domain_id) policy = extract_dmarc_policy(result.dmarc_record) or "none" + bimi_dmarc_ready, bimi_dmarc_issues, bimi_dmarc_evidence = _bimi_dmarc_readiness( + result.dmarc_record + ) checks = [ _dns_check( "dmarc", @@ -874,10 +994,11 @@ async def get_domain_dns_health( ], ), _mta_sts_check(mta_sts_result), + _bimi_check(bimi_result, bimi_dmarc_ready), ] recommendations: List[DNSHealthRecommendation] = [] for check in checks: - if check.status == "fail" and check.key != "mta_sts": + if check.status == "fail" and check.key not in {"mta_sts", "bimi"}: recommendations.append( DNSHealthRecommendation( type=f"missing_{check.key}", @@ -892,6 +1013,14 @@ async def get_domain_dns_health( mta_sts_recommendation = _mta_sts_recommendation(mta_sts_result) if mta_sts_recommendation: recommendations.append(mta_sts_recommendation) + bimi_recommendation = _bimi_recommendation( + bimi_result, + bimi_dmarc_ready, + bimi_dmarc_issues, + bimi_dmarc_evidence, + ) + if bimi_recommendation: + recommendations.append(bimi_recommendation) failed_checks = sum(1 for check in checks if check.status == "fail") health_status = ( @@ -943,6 +1072,43 @@ async def get_domain_mta_sts( ) +@router.get("/{domain_id}/dns/bimi", response_model=BIMIResponse) +async def get_domain_bimi( + domain_id: str = Path(..., title="The domain ID or name"), + selector: str = Query("default", title="BIMI selector"), + refresh: bool = Query(False, title="Refresh cached BIMI result"), + db: Session = Depends(get_db), +): + """Return cached BIMI DNS posture for a domain.""" + 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", + ) + result, cached, checked_at = await check_bimi_cached( + db, + get_default_provider(db), + domain_id, + selector=selector, + refresh=refresh, + ) + return BIMIResponse( + status=result.status, + selector=result.selector, + query_name=result.query_name, + dns_record=result.dns_record, + logo_url=result.logo_url, + certificate_url=result.certificate_url, + evidence_url=result.evidence_url, + errors=result.errors, + warnings=result.warnings, + cached=cached, + checked_at=checked_at.isoformat(), + ) + + @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.""" @@ -1032,9 +1198,8 @@ async def get_domain_reports( """ 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", @@ -1086,7 +1251,7 @@ async def export_domain_reports( store = ReportStore.get_instance() hydrate_report_store_from_db(db, store) - if domain_id not in store.get_domains(): + if not _domain_exists(db, store, domain_id): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Domain not found", @@ -1399,9 +1564,8 @@ async def get_domain_sources( """ 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", @@ -1457,7 +1621,7 @@ async def get_domain_selectors( """ store = ReportStore.get_instance() hydrate_report_store_from_db(db, store) - if domain_id not in store.get_domains(): + if not _domain_exists(db, store, domain_id): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Domain not found", diff --git a/backend/app/services/bimi.py b/backend/app/services/bimi.py new file mode 100644 index 0000000..6bb07ea --- /dev/null +++ b/backend/app/services/bimi.py @@ -0,0 +1,203 @@ +"""BIMI DNS posture checks for monitored domains.""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass, field +from datetime import datetime, timedelta, timezone +from typing import List, Optional, Tuple +from urllib.parse import urlparse + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.models.dns_cache import DNSCache +from app.services.dns_cache import DEFAULT_DNS_CACHE_TTL_SECONDS +from app.services.dns_resolver import BaseDNSProvider + +_CACHE_KEY_PREFIX = "bimi-v1" + + +@dataclass +class BIMIResult: + """Operator-facing BIMI posture evidence.""" + + status: str = "fail" + selector: str = "default" + query_name: str = "" + dns_record: Optional[str] = None + logo_url: Optional[str] = None + certificate_url: Optional[str] = None + evidence_url: Optional[str] = None + errors: List[str] = field(default_factory=list) + warnings: List[str] = field(default_factory=list) + + +def _utcnow_naive() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def _is_fresh(row: DNSCache, ttl_seconds: int, now: datetime) -> bool: + return row.checked_at >= now - timedelta(seconds=ttl_seconds) + + +def _result_from_json(value: str) -> BIMIResult: + data = json.loads(value) + return BIMIResult( + status=str(data.get("status") or "fail"), + selector=str(data.get("selector") or "default"), + query_name=str(data.get("query_name") or ""), + dns_record=data.get("dns_record"), + logo_url=data.get("logo_url"), + certificate_url=data.get("certificate_url"), + evidence_url=data.get("evidence_url"), + errors=list(data.get("errors") or []), + warnings=list(data.get("warnings") or []), + ) + + +def _https_url(value: Optional[str]) -> bool: + if not value: + return False + parsed = urlparse(value) + return parsed.scheme == "https" and bool(parsed.netloc) + + +def _tags(record: str) -> dict[str, str]: + return { + part.split("=", 1)[0].strip().lower(): part.split("=", 1)[1].strip() + for part in record.split(";") + if "=" in part + } + + +def parse_bimi_record(records: List[str]) -> Tuple[Optional[BIMIResult], List[str], List[str]]: + """Parse BIMI TXT records into a normalized result, warnings, and errors.""" + bimi_records = [record for record in records if record.lower().startswith("v=bimi1")] + if not bimi_records: + return None, [], ["No BIMI TXT record was found at the selector."] + + warnings: List[str] = [] + errors: List[str] = [] + if len(bimi_records) > 1: + warnings.append("Multiple BIMI TXT records were found; publish exactly one.") + + record = bimi_records[0] + tags = _tags(record) + logo_url = tags.get("l") + certificate_url = tags.get("a") + if tags.get("v", "").lower() != "bimi1": + errors.append("The BIMI TXT record must start with v=BIMI1.") + if not logo_url: + errors.append("The BIMI TXT record must include an l= HTTPS SVG logo URL.") + elif not _https_url(logo_url): + errors.append("The BIMI logo URL must use HTTPS.") + elif not urlparse(logo_url).path.lower().endswith(".svg"): + warnings.append("The BIMI logo URL should point to an SVG file.") + + if certificate_url and not _https_url(certificate_url): + errors.append("The BIMI certificate URL must use HTTPS when present.") + elif not certificate_url: + warnings.append("No BIMI certificate URL is published; some mailbox providers require one.") + + result = BIMIResult( + status="pass" if not errors else "fail", + dns_record=record, + logo_url=logo_url, + certificate_url=certificate_url, + evidence_url=logo_url, + errors=errors, + warnings=warnings, + ) + return result, warnings, errors + + +async def check_bimi( + domain: str, + provider: BaseDNSProvider, + *, + selector: str = "default", +) -> BIMIResult: + """Resolve and validate the BIMI TXT record for a domain selector.""" + normalized_selector = (selector or "default").strip().lower() + query_name = f"{normalized_selector}._bimi.{domain}" + result = BIMIResult(selector=normalized_selector, query_name=query_name) + try: + records = await provider.lookup_txt(query_name) + except LookupError as exc: + result.errors.append(f"BIMI DNS lookup failed: {exc}") + return result + + parsed, warnings, errors = parse_bimi_record(records) + if parsed is None: + result.errors.extend(errors) + return result + parsed.selector = normalized_selector + parsed.query_name = query_name + parsed.warnings = warnings + parsed.errors = errors + return parsed + + +async def check_bimi_cached( + db: Session, + provider: BaseDNSProvider, + domain: str, + *, + selector: str = "default", + ttl_seconds: int = DEFAULT_DNS_CACHE_TTL_SECONDS, + refresh: bool = False, +) -> Tuple[BIMIResult, bool, datetime]: + """Resolve BIMI posture, reusing the shared DNS cache semantics.""" + normalized_selector = (selector or "default").strip().lower() + cache_key = f"{_CACHE_KEY_PREFIX}:{normalized_selector}" + now = _utcnow_naive() + provider_name = f"{provider.__class__.__name__}:bimi" + row = ( + db.query(DNSCache) + .filter( + DNSCache.domain == domain, + DNSCache.provider == provider_name, + DNSCache.selectors_key == cache_key, + ) + .first() + ) + if row and not refresh and _is_fresh(row, ttl_seconds, now): + return _result_from_json(row.result_json), True, row.checked_at + + result = await check_bimi(domain, provider, selector=normalized_selector) + payload = json.dumps(asdict(result), sort_keys=True, separators=(",", ":")) + if row is None: + row = DNSCache( + domain=domain, + provider=provider_name, + selectors_key=cache_key, + result_json=payload, + checked_at=now, + ) + db.add(row) + else: + row.result_json = payload + row.checked_at = now + + try: + db.commit() + except IntegrityError: + db.rollback() + row = ( + db.query(DNSCache) + .filter( + DNSCache.domain == domain, + DNSCache.provider == provider_name, + DNSCache.selectors_key == cache_key, + ) + .first() + ) + if row is None: + raise + row.result_json = payload + row.checked_at = now + db.commit() + + db.refresh(row) + return result, False, row.checked_at diff --git a/backend/app/templates/domain_details.html b/backend/app/templates/domain_details.html index f1468c8..cfd7ac0 100644 --- a/backend/app/templates/domain_details.html +++ b/backend/app/templates/domain_details.html @@ -243,6 +243,30 @@ +
+

+ BIMI + + + +

+
+
-
+
+
+
Selector:
+
Logo:
+
Certificate:
+
+ + +
+
+

@@ -619,6 +643,16 @@ function domainDetailsApp(domainId) { errors: [], warnings: [] }, + bimi: { + status: '', + selector: 'default', + query_name: '', + dns_record: '', + logo_url: '', + certificate_url: '', + errors: [], + warnings: [] + }, selectors: [], reportSelectors: [], newSelector: '', @@ -638,6 +672,7 @@ function domainDetailsApp(domainId) { this.fetchDNSRecords(); this.fetchDNSHealth(); this.fetchMtaSts(); + this.fetchBimi(); this.fetchSelectors(); this.fetchReports(); this.fetchSources(); @@ -732,6 +767,17 @@ function domainDetailsApp(domainId) { } }, + async fetchBimi() { + try { + const response = await fetch(`/api/v1/domains/${this.domainId}/dns/bimi`); + if (response.ok) { + this.bimi = await response.json(); + } + } catch (error) { + console.error('Error fetching BIMI posture:', error); + } + }, + async fetchSelectors() { try { const response = await fetch(`/api/v1/domains/${this.domainId}/selectors`); diff --git a/backend/app/tests/test_bimi.py b/backend/app/tests/test_bimi.py new file mode 100644 index 0000000..cc6e077 --- /dev/null +++ b/backend/app/tests/test_bimi.py @@ -0,0 +1,118 @@ +from datetime import datetime +from unittest.mock import AsyncMock + +import pytest +from sqlalchemy.exc import IntegrityError + +from app.models.dns_cache import DNSCache +from app.services.bimi import check_bimi, check_bimi_cached, parse_bimi_record + + +def test_parse_bimi_record_requires_single_bimi_record(): + result, warnings, errors = parse_bimi_record( + ["v=BIMI1; l=https://example.com/logo.svg; a=https://example.com/vmc.pem"] + ) + + assert result is not None + assert result.status == "pass" + assert result.logo_url == "https://example.com/logo.svg" + assert result.certificate_url == "https://example.com/vmc.pem" + assert warnings == [] + assert errors == [] + + +def test_parse_bimi_record_reports_missing_and_malformed_records(): + missing, missing_warnings, missing_errors = parse_bimi_record([]) + malformed, _, malformed_errors = parse_bimi_record(["v=BIMI1; l=http://example.com/logo.png"]) + + assert missing is None + assert missing_warnings == [] + assert missing_errors == ["No BIMI TXT record was found at the selector."] + assert malformed is not None + assert malformed.status == "fail" + assert "HTTPS" in malformed_errors[0] + + +def test_parse_bimi_record_warns_without_certificate(): + result, warnings, errors = parse_bimi_record(["v=BIMI1; l=https://example.com/logo.svg"]) + + assert result is not None + assert result.status == "pass" + assert errors == [] + assert warnings == ["No BIMI certificate URL is published; some mailbox providers require one."] + + +@pytest.mark.asyncio +async def test_check_bimi_looks_up_default_selector(): + provider = AsyncMock() + provider.lookup_txt = AsyncMock( + return_value=["v=BIMI1; l=https://example.com/logo.svg; a=https://example.com/vmc.pem"] + ) + + result = await check_bimi("example.com", provider) + + assert result.status == "pass" + assert result.query_name == "default._bimi.example.com" + provider.lookup_txt.assert_awaited_once_with("default._bimi.example.com") + + +@pytest.mark.asyncio +async def test_check_bimi_cached_reuses_fresh_result(db_session): + provider = AsyncMock() + provider.lookup_txt = AsyncMock(return_value=["v=BIMI1; l=https://example.com/logo.svg"]) + + first, first_cached, first_checked = await check_bimi_cached( + db_session, provider, "example.com" + ) + second, second_cached, second_checked = await check_bimi_cached( + db_session, provider, "example.com" + ) + + assert first.status == "pass" + assert first_cached is False + assert second_cached is True + assert second.logo_url == "https://example.com/logo.svg" + assert second_checked == first_checked + assert provider.lookup_txt.await_count == 1 + + +@pytest.mark.asyncio +async def test_check_bimi_cached_recovers_from_concurrent_insert(db_session, monkeypatch): + provider = AsyncMock() + provider.lookup_txt = AsyncMock(return_value=["v=BIMI1; l=https://example.com/logo.svg"]) + original_commit = db_session.commit + original_rollback = db_session.rollback + commit_calls = 0 + + def fake_commit(): + nonlocal commit_calls + commit_calls += 1 + if commit_calls == 1: + raise IntegrityError("insert", {}, Exception("duplicate")) + original_commit() + + def fake_rollback(): + original_rollback() + db_session.add( + DNSCache( + domain="example.com", + provider=f"{provider.__class__.__name__}:bimi", + selectors_key="bimi-v1:default", + result_json=( + '{"certificate_url":null,"dns_record":null,"errors":["missing"],' + '"evidence_url":null,"logo_url":null,"query_name":"default._bimi.example.com",' + '"selector":"default","status":"fail","warnings":[]}' + ), + checked_at=datetime(2026, 5, 23, 12, 0, 0), + ) + ) + original_commit() + + monkeypatch.setattr(db_session, "commit", fake_commit) + monkeypatch.setattr(db_session, "rollback", fake_rollback) + + result, cached, _checked = await check_bimi_cached(db_session, provider, "example.com") + + assert result.status == "pass" + assert cached is False + assert db_session.query(DNSCache).count() == 1 diff --git a/backend/app/tests/test_dns_endpoints.py b/backend/app/tests/test_dns_endpoints.py index 4e7a87c..25d671a 100644 --- a/backend/app/tests/test_dns_endpoints.py +++ b/backend/app/tests/test_dns_endpoints.py @@ -19,6 +19,7 @@ from app.api.api_v1.endpoints import domains as domains_endpoint from app.api.api_v1.endpoints.domains import _spf_fix_hint from app.models.dns_cache import DNSCache from app.models.domain import Domain +from app.services.bimi import BIMIResult from app.services.dns_cache import _selectors_key, resolve_domain_dns_cached from app.services.dns_resolver import DomainDNSResult from app.services.mta_sts import MTAStsResult @@ -430,12 +431,22 @@ def test_dns_health_links_checks_to_evidence(client: TestClient): max_age=86400, mx=["*.example.com"], ) + bimi = BIMIResult( + status="pass", + dns_record="v=BIMI1; l=https://example.com/logo.svg; a=https://example.com/vmc.pem", + logo_url="https://example.com/logo.svg", + certificate_url="https://example.com/vmc.pem", + ) with ( _mock_dns(result=missing_dkim), patch( "app.api.api_v1.endpoints.domains.check_mta_sts_cached", new=AsyncMock(return_value=(mta_sts, False, None)), ), + patch( + "app.api.api_v1.endpoints.domains.check_bimi_cached", + new=AsyncMock(return_value=(bimi, False, None)), + ), ): response = client.get(f"/api/v1/domains/{DOMAIN}/dns/health") @@ -444,10 +455,14 @@ def test_dns_health_links_checks_to_evidence(client: TestClient): assert data["status"] == "degraded" dkim_check = next(check for check in data["checks"] if check["key"] == "dkim") mta_sts_check = next(check for check in data["checks"] if check["key"] == "mta_sts") + bimi_check = next(check for check in data["checks"] if check["key"] == "bimi") assert dkim_check["status"] == "fail" assert dkim_check["evidence"][0]["href"] == "#dns-records" assert mta_sts_check["status"] == "pass" assert mta_sts_check["evidence"][1]["href"] == "#mta-sts-posture" + assert bimi_check["status"] == "fail" + assert bimi_check["evidence"][0]["href"] == "#bimi-posture" + assert any(item["type"] == "bimi_dmarc_not_ready" for item in data["recommendations"]) assert any(item["type"] == "missing_dkim" for item in data["recommendations"]) @@ -498,6 +513,7 @@ def test_dns_health_marks_all_missing_records_critical(client: TestClient): recommendation_types = [item["type"] for item in data["recommendations"]] assert {"missing_dmarc", "missing_spf", "missing_dkim"}.issubset(set(recommendation_types)) assert recommendation_types.count("missing_mta_sts") == 1 + assert recommendation_types.count("missing_bimi") == 1 assert any(item["type"] == "policy_needs_more_data" for item in data["recommendations"]) @@ -531,6 +547,58 @@ def test_mta_sts_endpoint_returns_404_for_unknown_domain(client: TestClient): assert response.status_code == 404 +def test_bimi_endpoint_returns_cached_posture(client: TestClient): + """The domain detail page can fetch BIMI posture with cache metadata.""" + checked_at = datetime(2026, 5, 23, 12, 0, 0) + result = BIMIResult( + status="pass", + selector="default", + query_name=f"default._bimi.{DOMAIN}", + dns_record="v=BIMI1; l=https://example.com/logo.svg; a=https://example.com/vmc.pem", + logo_url="https://example.com/logo.svg", + certificate_url="https://example.com/vmc.pem", + ) + + with patch( + "app.api.api_v1.endpoints.domains.check_bimi_cached", + new=AsyncMock(return_value=(result, True, checked_at)), + ): + response = client.get(f"/api/v1/domains/{DOMAIN}/dns/bimi") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "pass" + assert data["query_name"] == f"default._bimi.{DOMAIN}" + assert data["logo_url"] == "https://example.com/logo.svg" + assert data["cached"] is True + assert data["checked_at"] == checked_at.isoformat() + + +def test_bimi_endpoint_returns_404_for_unknown_domain(client: TestClient): + response = client.get("/api/v1/domains/unknown.example.com/dns/bimi") + + assert response.status_code == 404 + + +def test_domain_detail_data_endpoints_support_manually_configured_domain( + client: TestClient, db_session +): + """Manually monitored domains should render empty detail data instead of 404s.""" + db_session.add(Domain(name="manual.example", active=True)) + db_session.commit() + + reports = client.get("/api/v1/domains/manual.example/reports") + sources = client.get("/api/v1/domains/manual.example/sources") + selectors = client.get("/api/v1/domains/manual.example/selectors") + + assert reports.status_code == 200 + assert reports.json()["reports"] == [] + assert sources.status_code == 200 + assert sources.json()["sources"] == [] + assert selectors.status_code == 200 + assert selectors.json() == {"selectors": [], "report_selectors": []} + + @pytest.mark.parametrize( ("policy", "summary", "expected_type", "expected_severity"), [ diff --git a/docs/milestones.md b/docs/milestones.md index bf9c688..407363f 100644 --- a/docs/milestones.md +++ b/docs/milestones.md @@ -221,7 +221,7 @@ Goal: turn DMARQ into a broader email authentication posture console (still priv Planned: - MTA-STS posture: delivered cached `_mta-sts` TXT checks, HTTPS policy validation, domain-detail evidence, and operator guidance for missing, invalid, or non-enforcing policies. Optional helper tooling remains a future enhancement. - TLS reporting posture: delivered authenticated TLS-RPT upload for `.json`, `.json.gz`, and `.zip` attachments; duplicate-safe persistence by report ID and policy domain; daily session trends; top failure-cause grouping; affected-domain summaries; and explicit privacy controls that avoid storing message content or recipient data. -- BIMI posture: record validation + readiness checks + operator guidance. +- BIMI posture: delivered default-selector BIMI TXT validation, HTTPS logo/certificate URL checks, DMARC enforcement readiness checks, domain-detail evidence, and operator guidance for missing or blocked BIMI prerequisites. - Extended DNS checks that support the posture surface (e.g., MX/BIMI; optional DANE/TLSA where relevant). Exit criteria: diff --git a/docs/reference/api.md b/docs/reference/api.md index 4c43002..f76625f 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -96,6 +96,15 @@ Returns details for a specific domain. } ``` +#### Get BIMI Posture + +``` +GET /domains/{domain_id}/dns/bimi +``` + +Returns the cached BIMI TXT posture for the default selector, including the +queried DNS name, record text, logo URL, certificate URL, warnings, and errors. + #### Add Domain ``` diff --git a/docs/user_guide/domains.md b/docs/user_guide/domains.md index d4857c6..a1e4bd7 100644 --- a/docs/user_guide/domains.md +++ b/docs/user_guide/domains.md @@ -71,6 +71,21 @@ DMARQ marks the check healthy when the TXT record contains `v=STSv1` with an `id MTA-STS posture uses the same cached DNS refresh behavior as the existing DNS health checks. Use the DNS refresh action when you publish or update a policy and need DMARQ to re-check immediately. +### BIMI Readiness + +The domain detail page checks the default BIMI selector at +`default._bimi.`. + +DMARQ validates that the BIMI TXT record starts with `v=BIMI1`, includes an +HTTPS `l=` SVG logo URL, and uses HTTPS for the optional `a=` certificate URL. +The readiness guidance also checks whether DMARC is ready for BIMI: the domain +must use `p=quarantine` or `p=reject`, `pct` must be `100` or omitted, and any +published `sp=` subdomain policy must also enforce. + +BIMI posture is read-only. Findings link back to the BIMI TXT record, logo URL, +certificate URL, and DMARC policy evidence so operators can see which +prerequisite is blocking readiness. + ## Domain Groups If you manage multiple domains, you can organize them into groups: