feat: add BIMI posture readiness checks

Closes #141

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",

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

backend/app/templates/domain_details.html

@@ -243,6 +243,30 @@
                             </div>
                         </div>
                     </div>
+                    <div id="bimi-posture">
+                        <h3 class="font-semibold mb-1 flex items-center">
+                            <span class="mr-2">BIMI</span>
+                            <span x-show="bimi.status === 'pass'" class="inline-flex h-2 w-2 rounded-full bg-green-500"></span>
+                            <span x-show="bimi.status === 'fail'" class="inline-flex h-2 w-2 rounded-full bg-red-500"></span>
+                            <span x-show="!bimi.status" class="inline-flex h-2 w-2 rounded-full bg-base-300"></span>
+                        </h3>
+                        <div class="space-y-2">
+                            <div class="bg-muted p-2 rounded text-sm overflow-x-auto font-mono" x-text="bimi.dns_record || 'No BIMI TXT record found at default._bimi'">-</div>
+                            <div class="rounded border border-base-300 p-3 text-sm">
+                                <div class="grid gap-1">
+                                    <div><span class="font-medium">Selector:</span> <span class="font-mono" x-text="bimi.selector || 'default'"></span></div>
+                                    <div><span class="font-medium">Logo:</span> <span class="font-mono break-all" x-text="bimi.logo_url || 'none'"></span></div>
+                                    <div><span class="font-medium">Certificate:</span> <span class="font-mono break-all" x-text="bimi.certificate_url || 'none'"></span></div>
+                                </div>
+                                <template x-if="bimi.errors && bimi.errors.length">
+                                    <p class="mt-2 text-xs text-red-600" x-text="bimi.errors[0]"></p>
+                                </template>
+                                <template x-if="bimi.warnings && bimi.warnings.length">
+                                    <p class="mt-2 text-xs text-yellow-700" x-text="bimi.warnings[0]"></p>
+                                </template>
+                            </div>
+                        </div>
+                    </div>
                     <!-- DKIM Selectors — live check result -->
                     <div>
                         <h3 class="font-semibold mb-1 flex items-center">
@@ -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`);

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

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"),
     [

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:

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
 
 ```

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.<domain>`.
+
+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: