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: