From 57e164d282e2ee2a45900f0da323e460c8ee532a Mon Sep 17 00:00:00 2001
From: Christian Krakau-Louis <christian@Christians-Mac-mini-7.local>
Date: Sat, 23 May 2026 16:16:55 +0200
Subject: [PATCH] feat: add mta-sts posture checks

---
 backend/app/api/api_v1/endpoints/domains.py | 112 +++++++-
 backend/app/services/dns_cache.py           |  27 +-
 backend/app/services/mta_sts.py             | 216 ++++++++++++++
 backend/app/templates/domain_details.html   |  49 +++-
 backend/app/tests/test_dns_endpoints.py     | 161 ++++++++++-
 backend/app/tests/test_mta_sts.py           | 303 ++++++++++++++++++++
 docs/milestones.md                          |   4 +-
 docs/user_guide/domains.md                  |  11 +-
 8 files changed, 866 insertions(+), 17 deletions(-)
 create mode 100644 backend/app/services/mta_sts.py
 create mode 100644 backend/app/tests/test_mta_sts.py

diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py
index f4bf63b..2c9274d 100644
--- a/backend/app/api/api_v1/endpoints/domains.py
+++ b/backend/app/api/api_v1/endpoints/domains.py
@@ -29,6 +29,7 @@ from app.services.dns_resolver import (
     extract_dmarc_policy,
     get_default_provider,
 )
+from app.services.mta_sts import MTAStsResult, check_mta_sts_cached
 from app.services.report_persistence import (
     delete_persisted_domain,
     hydrate_report_store_from_db,
@@ -129,6 +130,22 @@ class DNSHealthResponse(BaseModel):
     recommendations: List[DNSHealthRecommendation]
 
 
+class MTAStsResponse(BaseModel):
+    """MTA-STS posture result for a domain."""
+
+    status: str
+    dns_record: Optional[str] = None
+    policy_url: Optional[str] = None
+    policy_text: Optional[str] = None
+    mode: Optional[str] = None
+    max_age: Optional[int] = None
+    mx: List[str] = Field(default_factory=list)
+    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."""
 
@@ -445,6 +462,54 @@ def _enforcement_recommendation(
     )
 
 
+def _mta_sts_check(result: MTAStsResult) -> DNSHealthCheck:
+    evidence = [
+        _record_evidence("MTA-STS TXT", result.dns_record, "#dns-records"),
+        _record_evidence("Policy URL", result.policy_url, "#mta-sts-posture"),
+    ]
+    if result.mode:
+        evidence.append(_record_evidence("Mode", result.mode, "#mta-sts-posture"))
+    if result.mx:
+        evidence.append(_record_evidence("MX patterns", ", ".join(result.mx), "#mta-sts-posture"))
+    message = (
+        "MTA-STS DNS and HTTPS policy are valid."
+        if result.status == "pass"
+        else (result.errors[0] if result.errors else "MTA-STS posture needs attention.")
+    )
+    return DNSHealthCheck(
+        key="mta_sts",
+        label="MTA-STS",
+        status=result.status,
+        message=message,
+        evidence=evidence,
+    )
+
+
+def _mta_sts_recommendation(result: MTAStsResult) -> Optional[DNSHealthRecommendation]:
+    if result.status == "pass" and not result.warnings:
+        return None
+    severity = "warning" if result.status == "pass" else "error"
+    title = "MTA-STS policy needs review" if result.status == "pass" else "Publish MTA-STS"
+    detail = (
+        "; ".join(result.warnings)
+        if result.status == "pass"
+        else "; ".join(result.errors or ["MTA-STS is not configured or is not valid."])
+    )
+    action = (
+        "Move the policy to mode: enforce once MX coverage is confirmed."
+        if result.status == "pass"
+        else "Publish _mta-sts TXT and a valid HTTPS policy at the well-known URL."
+    )
+    return DNSHealthRecommendation(
+        type="mta_sts_review" if result.status == "pass" else "missing_mta_sts",
+        severity=severity,
+        title=title,
+        detail=detail,
+        action=action,
+        evidence=_mta_sts_check(result).evidence,
+    )
+
+
 @router.get("/summary", response_model=DomainSummaryResponse)
 async def get_domains_summary(db: Session = Depends(get_db)):
     """
@@ -769,6 +834,12 @@ async def get_domain_dns_health(
         selectors=combined_selectors,
         refresh=refresh,
     )
+    mta_sts_result, _, _ = await check_mta_sts_cached(
+        db,
+        provider,
+        domain_id,
+        refresh=refresh,
+    )
     summary = store.get_domain_summary(domain_id)
     policy = extract_dmarc_policy(result.dmarc_record) or "none"
     checks = [
@@ -802,10 +873,11 @@ async def get_domain_dns_health(
                 _record_evidence("DKIM TXT", result.dkim_record),
             ],
         ),
+        _mta_sts_check(mta_sts_result),
     ]
     recommendations: List[DNSHealthRecommendation] = []
     for check in checks:
-        if check.status == "fail":
+        if check.status == "fail" and check.key != "mta_sts":
             recommendations.append(
                 DNSHealthRecommendation(
                     type=f"missing_{check.key}",
@@ -817,6 +889,9 @@ async def get_domain_dns_health(
                 )
             )
     recommendations.append(_enforcement_recommendation(policy, summary))
+    mta_sts_recommendation = _mta_sts_recommendation(mta_sts_result)
+    if mta_sts_recommendation:
+        recommendations.append(mta_sts_recommendation)
 
     failed_checks = sum(1 for check in checks if check.status == "fail")
     health_status = (
@@ -833,6 +908,41 @@ async def get_domain_dns_health(
     )
 
 
+@router.get("/{domain_id}/dns/mta-sts", response_model=MTAStsResponse)
+async def get_domain_mta_sts(
+    domain_id: str = Path(..., title="The domain ID or name"),
+    refresh: bool = Query(False, title="Refresh cached MTA-STS result"),
+    db: Session = Depends(get_db),
+):
+    """Return cached MTA-STS DNS and HTTPS policy 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_mta_sts_cached(
+        db,
+        get_default_provider(db),
+        domain_id,
+        refresh=refresh,
+    )
+    return MTAStsResponse(
+        status=result.status,
+        dns_record=result.dns_record,
+        policy_url=result.policy_url,
+        policy_text=result.policy_text,
+        mode=result.mode,
+        max_age=result.max_age,
+        mx=result.mx,
+        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."""
diff --git a/backend/app/services/dns_cache.py b/backend/app/services/dns_cache.py
index 042f045..0f7ead7 100644
--- a/backend/app/services/dns_cache.py
+++ b/backend/app/services/dns_cache.py
@@ -8,6 +8,7 @@ from dataclasses import asdict
 from datetime import datetime, timedelta, timezone
 from typing import List, Tuple
 
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.models.dns_cache import DNSCache
@@ -74,18 +75,38 @@ async def resolve_domain_dns_cached(
         return _result_from_json(row.result_json), True, row.checked_at
 
     result = await provider.check_domain(domain, selectors=selectors)
+    payload = _result_to_json(result)
     if row is None:
         row = DNSCache(
             domain=domain,
             provider=provider_name,
             selectors_key=selectors_key,
-            result_json=_result_to_json(result),
+            result_json=payload,
             checked_at=now,
         )
         db.add(row)
     else:
-        row.result_json = _result_to_json(result)
+        row.result_json = payload
         row.checked_at = now
-    db.commit()
+
+    try:
+        db.commit()
+    except IntegrityError:
+        db.rollback()
+        row = (
+            db.query(DNSCache)
+            .filter(
+                DNSCache.domain == domain,
+                DNSCache.provider == provider_name,
+                DNSCache.selectors_key == selectors_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/services/mta_sts.py b/backend/app/services/mta_sts.py
new file mode 100644
index 0000000..cdb2c91
--- /dev/null
+++ b/backend/app/services/mta_sts.py
@@ -0,0 +1,216 @@
+"""MTA-STS 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 Any, Dict, List, Optional, Tuple
+
+import httpx
+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 = "mta-sts-v1"
+_POLICY_TIMEOUT_SECONDS = 5.0
+_VALID_MODES = {"enforce", "testing", "none"}
+
+
+@dataclass
+class MTAStsResult:
+    """Operator-facing MTA-STS posture evidence."""
+
+    status: str = "fail"
+    dns_record: Optional[str] = None
+    policy_url: Optional[str] = None
+    policy_text: Optional[str] = None
+    mode: Optional[str] = None
+    max_age: Optional[int] = None
+    mx: List[str] = field(default_factory=list)
+    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) -> MTAStsResult:
+    data = json.loads(value)
+    return MTAStsResult(
+        status=str(data.get("status") or "fail"),
+        dns_record=data.get("dns_record"),
+        policy_url=data.get("policy_url"),
+        policy_text=data.get("policy_text"),
+        mode=data.get("mode"),
+        max_age=data.get("max_age"),
+        mx=list(data.get("mx") or []),
+        errors=list(data.get("errors") or []),
+        warnings=list(data.get("warnings") or []),
+    )
+
+
+def parse_mta_sts_record(records: List[str]) -> Tuple[Optional[str], List[str], List[str]]:
+    """Return the selected MTA-STS TXT record, warnings, and errors."""
+    sts_records = [record for record in records if record.lower().startswith("v=stsv1")]
+    if not sts_records:
+        return None, [], ["No _mta-sts TXT record was found."]
+    warnings = []
+    if len(sts_records) > 1:
+        warnings.append("Multiple _mta-sts TXT records were found; publish exactly one.")
+    record = sts_records[0]
+    tags = {
+        part.split("=", 1)[0].strip().lower(): part.split("=", 1)[1].strip()
+        for part in record.split(";")
+        if "=" in part
+    }
+    errors = []
+    if tags.get("v", "").lower() != "stsv1":
+        errors.append("The _mta-sts TXT record must start with v=STSv1.")
+    if not tags.get("id"):
+        errors.append("The _mta-sts TXT record must include a non-empty id tag.")
+    return record, warnings, errors
+
+
+def parse_mta_sts_policy(  # noqa: C901
+    policy_text: str,
+) -> Tuple[Dict[str, Any], List[str], List[str]]:
+    """Parse and validate an MTA-STS policy file."""
+    data: Dict[str, Any] = {"mx": []}
+    for raw_line in policy_text.splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or ":" not in line:
+            continue
+        key, value = line.split(":", 1)
+        key = key.strip().lower()
+        value = value.strip()
+        if key == "mx":
+            data.setdefault("mx", []).append(value)
+        else:
+            data[key] = value
+
+    errors = []
+    warnings = []
+    if str(data.get("version", "")).upper() != "STSV1":
+        errors.append("The policy file must contain version: STSv1.")
+    mode = str(data.get("mode", "")).lower()
+    if mode not in _VALID_MODES:
+        errors.append("The policy file must contain mode: enforce, testing, or none.")
+    elif mode in {"testing", "none"}:
+        warnings.append(f"MTA-STS policy is valid but not enforcing mail delivery ({mode}).")
+    try:
+        max_age = int(str(data.get("max_age", "")))
+        if max_age <= 0:
+            errors.append("The policy max_age must be greater than zero.")
+        data["max_age"] = max_age
+    except ValueError:
+        errors.append("The policy file must contain an integer max_age value.")
+    if not data.get("mx"):
+        errors.append("The policy file must contain at least one mx entry.")
+    return data, warnings, errors
+
+
+async def check_mta_sts(domain: str, provider: BaseDNSProvider) -> MTAStsResult:
+    """Resolve the MTA-STS TXT record and validate the HTTPS policy file."""
+    result = MTAStsResult(policy_url=f"https://mta-sts.{domain}/.well-known/mta-sts.txt")
+    try:
+        records = await provider.lookup_txt(f"_mta-sts.{domain}")
+    except LookupError as exc:
+        result.errors.append(f"MTA-STS DNS lookup failed: {exc}")
+        return result
+
+    record, warnings, errors = parse_mta_sts_record(records)
+    result.dns_record = record
+    result.warnings.extend(warnings)
+    result.errors.extend(errors)
+    if record is None:
+        return result
+
+    try:
+        async with httpx.AsyncClient(
+            timeout=_POLICY_TIMEOUT_SECONDS, follow_redirects=False
+        ) as client:
+            response = await client.get(result.policy_url)
+            response.raise_for_status()
+            result.policy_text = response.text
+    except (httpx.RequestError, httpx.HTTPStatusError, httpx.TimeoutException) as exc:
+        result.errors.append(f"MTA-STS policy fetch failed: {exc}")
+        return result
+
+    policy, policy_warnings, policy_errors = parse_mta_sts_policy(result.policy_text or "")
+    result.warnings.extend(policy_warnings)
+    result.errors.extend(policy_errors)
+    result.mode = policy.get("mode")
+    result.max_age = policy.get("max_age")
+    result.mx = list(policy.get("mx") or [])
+    result.status = "pass" if not result.errors else "fail"
+    return result
+
+
+async def check_mta_sts_cached(
+    db: Session,
+    provider: BaseDNSProvider,
+    domain: str,
+    *,
+    ttl_seconds: int = DEFAULT_DNS_CACHE_TTL_SECONDS,
+    refresh: bool = False,
+) -> Tuple[MTAStsResult, bool, datetime]:
+    """Resolve MTA-STS posture, reusing the shared DNS cache semantics."""
+    now = _utcnow_naive()
+    provider_name = f"{provider.__class__.__name__}:mta-sts"
+    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_mta_sts(domain, provider)
+    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 38e0921..f1468c8 100644
--- a/backend/app/templates/domain_details.html
+++ b/backend/app/templates/domain_details.html
@@ -143,7 +143,7 @@
                 {% endcall %}
             {% endcall %}
             {% call card_content() %}
-                <div class="grid gap-4 lg:grid-cols-3">
+                <div class="grid gap-4 md:grid-cols-2 xl:grid-cols-4">
                     <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">
@@ -219,6 +219,30 @@
                         </h3>
                         <div class="bg-muted p-2 rounded text-sm overflow-x-auto font-mono" x-text="dns.spfRecord || 'No SPF record found'">-</div>
                     </div>
+                    <div id="mta-sts-posture">
+                        <h3 class="font-semibold mb-1 flex items-center">
+                            <span class="mr-2">MTA-STS</span>
+                            <span x-show="mtaSts.status === 'pass'" class="inline-flex h-2 w-2 rounded-full bg-green-500"></span>
+                            <span x-show="mtaSts.status === 'fail'" class="inline-flex h-2 w-2 rounded-full bg-red-500"></span>
+                            <span x-show="!mtaSts.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="mtaSts.dns_record || 'No _mta-sts TXT record found'">-</div>
+                            <div class="rounded border border-base-300 p-3 text-sm">
+                                <div class="grid gap-1">
+                                    <div><span class="font-medium">Mode:</span> <span x-text="mtaSts.mode || 'unknown'"></span></div>
+                                    <div><span class="font-medium">Max age:</span> <span x-text="mtaSts.max_age || 'unknown'"></span></div>
+                                    <div><span class="font-medium">MX:</span> <span class="font-mono" x-text="mtaSts.mx && mtaSts.mx.length ? mtaSts.mx.join(', ') : 'none'"></span></div>
+                                </div>
+                                <template x-if="mtaSts.errors && mtaSts.errors.length">
+                                    <p class="mt-2 text-xs text-red-600" x-text="mtaSts.errors[0]"></p>
+                                </template>
+                                <template x-if="mtaSts.warnings && mtaSts.warnings.length">
+                                    <p class="mt-2 text-xs text-yellow-700" x-text="mtaSts.warnings[0]"></p>
+                                </template>
+                            </div>
+                        </div>
+                    </div>
                     <!-- DKIM Selectors — live check result -->
                     <div>
                         <h3 class="font-semibold mb-1 flex items-center">
@@ -586,6 +610,15 @@ function domainDetailsApp(domainId) {
             checks: [],
             recommendations: []
         },
+        mtaSts: {
+            status: '',
+            dns_record: '',
+            mode: '',
+            max_age: null,
+            mx: [],
+            errors: [],
+            warnings: []
+        },
         selectors: [],
         reportSelectors: [],
         newSelector: '',
@@ -604,6 +637,7 @@ function domainDetailsApp(domainId) {
             this.fetchDomainStats();
             this.fetchDNSRecords();
             this.fetchDNSHealth();
+            this.fetchMtaSts();
             this.fetchSelectors();
             this.fetchReports();
             this.fetchSources();
@@ -687,6 +721,17 @@ function domainDetailsApp(domainId) {
             }
         },
 
+        async fetchMtaSts() {
+            try {
+                const response = await fetch(`/api/v1/domains/${this.domainId}/dns/mta-sts`);
+                if (response.ok) {
+                    this.mtaSts = await response.json();
+                }
+            } catch (error) {
+                console.error('Error fetching MTA-STS posture:', error);
+            }
+        },
+
         async fetchSelectors() {
             try {
                 const response = await fetch(`/api/v1/domains/${this.domainId}/selectors`);
@@ -716,6 +761,7 @@ function domainDetailsApp(domainId) {
                     this.fetchSelectors();
                     this.fetchDNSRecords();
                     this.fetchDNSHealth();
+                    this.fetchMtaSts();
                 } else {
                     const err = await response.json();
                     this.selectorError = err.detail || 'Failed to add selector';
@@ -737,6 +783,7 @@ function domainDetailsApp(domainId) {
                     this.fetchSelectors();
                     this.fetchDNSRecords();
                     this.fetchDNSHealth();
+                    this.fetchMtaSts();
                 } else {
                     console.error('Error deleting selector:', response.status);
                 }
diff --git a/backend/app/tests/test_dns_endpoints.py b/backend/app/tests/test_dns_endpoints.py
index f86d3df..4e7a87c 100644
--- a/backend/app/tests/test_dns_endpoints.py
+++ b/backend/app/tests/test_dns_endpoints.py
@@ -8,16 +8,20 @@ that the endpoints can find the test domain.
 DNS lookups are mocked so no real network calls are made.
 """
 
+from datetime import datetime
 from unittest.mock import AsyncMock, patch
 
 import pytest
 from fastapi.testclient import TestClient
+from sqlalchemy.exc import IntegrityError
 
 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.dns_cache import _selectors_key, resolve_domain_dns_cached
 from app.services.dns_resolver import DomainDNSResult
+from app.services.mta_sts import MTAStsResult
 from app.services.report_store import ReportStore
 
 # ---------------------------------------------------------------------------
@@ -68,9 +72,12 @@ def _seed_report_store():
 
 def _mock_dns(result: DomainDNSResult = MOCK_DNS_RESULT):
     """Return a context manager that patches the DNS provider's check_domain."""
+    provider = AsyncMock()
+    provider.check_domain = AsyncMock(return_value=result)
+    provider.lookup_txt = AsyncMock(side_effect=LookupError("MTA-STS not configured"))
     return patch(
         "app.api.api_v1.endpoints.domains.get_default_provider",
-        return_value=AsyncMock(check_domain=AsyncMock(return_value=result)),
+        return_value=provider,
     )
 
 
@@ -278,6 +285,72 @@ def test_dns_endpoint_uses_cached_result(client: TestClient, db_session):
     assert db_session.query(DNSCache).count() == 1
 
 
+@pytest.mark.asyncio
+async def test_dns_cache_recovers_from_concurrent_insert(db_session, monkeypatch):
+    """Concurrent DNS widgets should not fail on a duplicate cache insert."""
+    mock_provider = AsyncMock(check_domain=AsyncMock(return_value=MOCK_DNS_RESULT))
+    selectors = ["google"]
+    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=DOMAIN,
+                provider=mock_provider.__class__.__name__,
+                selectors_key=_selectors_key(selectors),
+                result_json=(
+                    '{"dmarc":false,"spf":false,"dkim":false,'
+                    '"dkim_selectors":[],"selectors_checked":[]}'
+                ),
+                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 resolve_domain_dns_cached(
+        db_session,
+        mock_provider,
+        DOMAIN,
+        selectors=selectors,
+    )
+
+    assert result == MOCK_DNS_RESULT
+    assert cached is False
+    assert db_session.query(DNSCache).count() == 1
+
+
+@pytest.mark.asyncio
+async def test_dns_cache_reraises_when_conflict_row_missing(db_session, monkeypatch):
+    """Unexpected cache collisions should still surface when no row can be recovered."""
+    mock_provider = AsyncMock(check_domain=AsyncMock(return_value=MOCK_DNS_RESULT))
+
+    def fake_commit():
+        raise IntegrityError("insert", {}, Exception("duplicate"))
+
+    monkeypatch.setattr(db_session, "commit", fake_commit)
+
+    with pytest.raises(IntegrityError):
+        await resolve_domain_dns_cached(
+            db_session,
+            mock_provider,
+            DOMAIN,
+            selectors=["google"],
+        )
+
+
 def test_dns_endpoint_refresh_bypasses_cache(client: TestClient):
     """The refresh query parameter forces a new DNS lookup."""
     mock_provider = AsyncMock(check_domain=AsyncMock(return_value=MOCK_DNS_RESULT))
@@ -349,15 +422,32 @@ def test_dns_health_links_checks_to_evidence(client: TestClient):
         selectors_checked=["google"],
     )
 
-    with _mock_dns(result=missing_dkim):
+    mta_sts = MTAStsResult(
+        status="pass",
+        dns_record="v=STSv1; id=20260523",
+        policy_url="https://mta-sts.example.com/.well-known/mta-sts.txt",
+        mode="enforce",
+        max_age=86400,
+        mx=["*.example.com"],
+    )
+    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)),
+        ),
+    ):
         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")
+    mta_sts_check = next(check for check in data["checks"] if check["key"] == "mta_sts")
     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 any(item["type"] == "missing_dkim" for item in data["recommendations"])
 
 
@@ -405,21 +495,74 @@ def test_dns_health_marks_all_missing_records_critical(client: TestClient):
     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)
+    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 any(item["type"] == "policy_needs_more_data" for item in data["recommendations"])
 
 
+def test_mta_sts_endpoint_returns_cached_posture(client: TestClient):
+    """The domain detail page can fetch MTA-STS posture with cache metadata."""
+    checked_at = datetime(2026, 5, 23, 12, 0, 0)
+    result = MTAStsResult(
+        status="fail",
+        dns_record=None,
+        policy_url=f"https://mta-sts.{DOMAIN}/.well-known/mta-sts.txt",
+        errors=["No _mta-sts TXT record was found."],
+    )
+
+    with patch(
+        "app.api.api_v1.endpoints.domains.check_mta_sts_cached",
+        new=AsyncMock(return_value=(result, True, checked_at)),
+    ):
+        response = client.get(f"/api/v1/domains/{DOMAIN}/dns/mta-sts")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["status"] == "fail"
+    assert data["cached"] is True
+    assert data["checked_at"] == checked_at.isoformat()
+    assert data["errors"] == ["No _mta-sts TXT record was found."]
+
+
+def test_mta_sts_endpoint_returns_404_for_unknown_domain(client: TestClient):
+    response = client.get("/api/v1/domains/unknown.example.com/dns/mta-sts")
+
+    assert response.status_code == 404
+
+
 @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"),
+        (
+            "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):
+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)
 
diff --git a/backend/app/tests/test_mta_sts.py b/backend/app/tests/test_mta_sts.py
new file mode 100644
index 0000000..2e8fb63
--- /dev/null
+++ b/backend/app/tests/test_mta_sts.py
@@ -0,0 +1,303 @@
+from datetime import datetime
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from app.models.dns_cache import DNSCache
+from app.services.mta_sts import (
+    _CACHE_KEY,
+    MTAStsResult,
+    check_mta_sts,
+    check_mta_sts_cached,
+    parse_mta_sts_policy,
+    parse_mta_sts_record,
+)
+
+
+def test_parse_mta_sts_record_requires_single_record_with_id():
+    record, warnings, errors = parse_mta_sts_record(["v=STSv1; id=20260523"])
+
+    assert record == "v=STSv1; id=20260523"
+    assert warnings == []
+    assert errors == []
+
+
+def test_parse_mta_sts_record_reports_missing_id_and_multiple_records():
+    record, warnings, errors = parse_mta_sts_record(["v=STSv1", "v=STSv1; id=two"])
+
+    assert record == "v=STSv1"
+    assert warnings == ["Multiple _mta-sts TXT records were found; publish exactly one."]
+    assert "id tag" in errors[0]
+
+
+def test_parse_mta_sts_record_reports_missing_and_malformed_versions():
+    missing_record, missing_warnings, missing_errors = parse_mta_sts_record([])
+    malformed_record, malformed_warnings, malformed_errors = parse_mta_sts_record(
+        ["v=STSv1x; id=bad"]
+    )
+
+    assert missing_record is None
+    assert missing_warnings == []
+    assert missing_errors == ["No _mta-sts TXT record was found."]
+    assert malformed_record == "v=STSv1x; id=bad"
+    assert malformed_warnings == []
+    assert malformed_errors == ["The _mta-sts TXT record must start with v=STSv1."]
+
+
+def test_parse_mta_sts_policy_validates_required_fields():
+    policy, warnings, errors = parse_mta_sts_policy(
+        "version: STSv1\nmode: testing\nmx: mail.example.com\nmax_age: 86400\n"
+    )
+
+    assert policy["mode"] == "testing"
+    assert policy["mx"] == ["mail.example.com"]
+    assert policy["max_age"] == 86400
+    assert warnings == ["MTA-STS policy is valid but not enforcing mail delivery (testing)."]
+    assert errors == []
+
+
+def test_parse_mta_sts_policy_reports_invalid_fields():
+    policy, warnings, errors = parse_mta_sts_policy(
+        "# comment\nignored line\nmode: invalid\nmax_age: 0\n"
+    )
+    bad_age_policy, bad_age_warnings, bad_age_errors = parse_mta_sts_policy(
+        "version: STSv1\nmode: enforce\nmax_age: nope\n"
+    )
+
+    assert policy["max_age"] == 0
+    assert warnings == []
+    assert "version: STSv1" in errors[0]
+    assert "mode: enforce, testing, or none" in errors[1]
+    assert "greater than zero" in errors[2]
+    assert "at least one mx" in errors[3]
+    assert bad_age_policy["mode"] == "enforce"
+    assert bad_age_warnings == []
+    assert "integer max_age" in bad_age_errors[0]
+    assert "at least one mx" in bad_age_errors[1]
+
+
+class _FakeAsyncClient:
+    def __init__(self, response):
+        self.response = response
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_args):
+        return False
+
+    async def get(self, _url):
+        return self.response
+
+
+@pytest.mark.asyncio
+async def test_check_mta_sts_validates_dns_and_policy(monkeypatch):
+    provider = AsyncMock()
+    provider.lookup_txt = AsyncMock(return_value=["v=STSv1; id=20260523"])
+    request = httpx.Request("GET", "https://mta-sts.example.com/.well-known/mta-sts.txt")
+    response = httpx.Response(
+        200,
+        text="version: STSv1\nmode: enforce\nmx: *.example.com\nmax_age: 86400\n",
+        request=request,
+    )
+    monkeypatch.setattr(
+        "app.services.mta_sts.httpx.AsyncClient",
+        lambda **_: _FakeAsyncClient(response),
+    )
+
+    result = await check_mta_sts("example.com", provider)
+
+    assert result.status == "pass"
+    assert result.dns_record == "v=STSv1; id=20260523"
+    assert result.mode == "enforce"
+    assert result.mx == ["*.example.com"]
+    assert result.errors == []
+
+
+@pytest.mark.asyncio
+async def test_check_mta_sts_cached_reuses_fresh_result(db_session, monkeypatch):
+    provider = AsyncMock()
+    provider.lookup_txt = AsyncMock(return_value=["v=STSv1; id=20260523"])
+    request = httpx.Request("GET", "https://mta-sts.example.com/.well-known/mta-sts.txt")
+    response = httpx.Response(
+        200,
+        text="version: STSv1\nmode: enforce\nmx: mail.example.com\nmax_age: 86400\n",
+        request=request,
+    )
+    monkeypatch.setattr(
+        "app.services.mta_sts.httpx.AsyncClient",
+        lambda **_: _FakeAsyncClient(response),
+    )
+
+    first, first_cached, first_checked = await check_mta_sts_cached(
+        db_session, provider, "example.com"
+    )
+    second, second_cached, second_checked = await check_mta_sts_cached(
+        db_session, provider, "example.com"
+    )
+
+    assert isinstance(first, MTAStsResult)
+    assert first.status == "pass"
+    assert first_cached is False
+    assert second.status == "pass"
+    assert second_cached is True
+    assert isinstance(first_checked, datetime)
+    assert second_checked == first_checked
+    provider.lookup_txt.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_check_mta_sts_returns_missing_record_without_policy_fetch(monkeypatch):
+    provider = AsyncMock()
+    provider.lookup_txt = AsyncMock(return_value=[])
+    fetch_attempted = False
+
+    class _UnexpectedAsyncClient:
+        async def __aenter__(self):
+            nonlocal fetch_attempted
+            fetch_attempted = True
+            return self
+
+        async def __aexit__(self, *_args):
+            return False
+
+    monkeypatch.setattr(
+        "app.services.mta_sts.httpx.AsyncClient",
+        lambda **_: _UnexpectedAsyncClient(),
+    )
+
+    result = await check_mta_sts("example.com", provider)
+
+    assert result.status == "fail"
+    assert result.errors == ["No _mta-sts TXT record was found."]
+    assert fetch_attempted is False
+
+
+@pytest.mark.asyncio
+async def test_check_mta_sts_reports_policy_fetch_error(monkeypatch):
+    provider = AsyncMock()
+    provider.lookup_txt = AsyncMock(return_value=["v=STSv1; id=20260523"])
+    request = httpx.Request("GET", "https://mta-sts.example.com/.well-known/mta-sts.txt")
+    response = httpx.Response(404, text="missing", request=request)
+    monkeypatch.setattr(
+        "app.services.mta_sts.httpx.AsyncClient",
+        lambda **_: _FakeAsyncClient(response),
+    )
+
+    result = await check_mta_sts("example.com", provider)
+
+    assert result.status == "fail"
+    assert result.policy_text is None
+    assert result.errors[0].startswith("MTA-STS policy fetch failed:")
+
+
+@pytest.mark.asyncio
+async def test_check_mta_sts_cached_refresh_updates_existing_row(db_session, monkeypatch):
+    provider = AsyncMock()
+    provider.lookup_txt = AsyncMock(return_value=["v=STSv1; id=20260523"])
+    request = httpx.Request("GET", "https://mta-sts.example.com/.well-known/mta-sts.txt")
+    responses = [
+        httpx.Response(
+            200,
+            text="version: STSv1\nmode: testing\nmx: mail.example.com\nmax_age: 86400\n",
+            request=request,
+        ),
+        httpx.Response(
+            200,
+            text="version: STSv1\nmode: enforce\nmx: mail.example.com\nmax_age: 86400\n",
+            request=request,
+        ),
+    ]
+    monkeypatch.setattr(
+        "app.services.mta_sts.httpx.AsyncClient",
+        lambda **_: _FakeAsyncClient(responses.pop(0)),
+    )
+
+    first, first_cached, _first_checked = await check_mta_sts_cached(
+        db_session, provider, "example.com"
+    )
+    second, second_cached, _second_checked = await check_mta_sts_cached(
+        db_session, provider, "example.com", refresh=True
+    )
+
+    assert first.mode == "testing"
+    assert first_cached is False
+    assert second.mode == "enforce"
+    assert second_cached is False
+    assert db_session.query(DNSCache).count() == 1
+
+
+@pytest.mark.asyncio
+async def test_check_mta_sts_cached_recovers_from_concurrent_insert(db_session, monkeypatch):
+    """Concurrent page widgets should not fail on a duplicate cache insert."""
+    provider = AsyncMock()
+    provider.lookup_txt = AsyncMock(return_value=["v=STSv1; id=20260523"])
+    request = httpx.Request("GET", "https://mta-sts.example.com/.well-known/mta-sts.txt")
+    response = httpx.Response(
+        200,
+        text="version: STSv1\nmode: enforce\nmx: mail.example.com\nmax_age: 86400\n",
+        request=request,
+    )
+    monkeypatch.setattr(
+        "app.services.mta_sts.httpx.AsyncClient",
+        lambda **_: _FakeAsyncClient(response),
+    )
+
+    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__}:mta-sts",
+                selectors_key=_CACHE_KEY,
+                result_json='{"status":"fail","errors":["stale"],"warnings":[],"mx":[]}',
+                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_mta_sts_cached(db_session, provider, "example.com")
+
+    assert result.status == "pass"
+    assert cached is False
+    assert db_session.query(DNSCache).count() == 1
+
+
+@pytest.mark.asyncio
+async def test_check_mta_sts_cached_reraises_when_conflict_row_missing(db_session, monkeypatch):
+    provider = AsyncMock()
+    provider.lookup_txt = AsyncMock(return_value=["v=STSv1; id=20260523"])
+    request = httpx.Request("GET", "https://mta-sts.example.com/.well-known/mta-sts.txt")
+    response = httpx.Response(
+        200,
+        text="version: STSv1\nmode: enforce\nmx: mail.example.com\nmax_age: 86400\n",
+        request=request,
+    )
+    monkeypatch.setattr(
+        "app.services.mta_sts.httpx.AsyncClient",
+        lambda **_: _FakeAsyncClient(response),
+    )
+
+    def fake_commit():
+        raise IntegrityError("insert", {}, Exception("duplicate"))
+
+    monkeypatch.setattr(db_session, "commit", fake_commit)
+
+    with pytest.raises(IntegrityError):
+        await check_mta_sts_cached(db_session, provider, "example.com")
diff --git a/docs/milestones.md b/docs/milestones.md
index d297102..35de81a 100644
--- a/docs/milestones.md
+++ b/docs/milestones.md
@@ -214,12 +214,12 @@ Exit criteria:
 
 ## Milestone 13: Email Security Posture (Beyond DMARC)
 
-Status: Backlog
+Status: In Progress
 
 Goal: turn DMARQ into a broader email authentication posture console (still privacy-first and self-hostable).
 
 Planned:
-- MTA-STS posture: DNS record evaluation + policy fetch validation (plus optional helper tooling).
+- 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: ingest and summarize TLS report data (where available) with actionable failure grouping.
 - BIMI posture: record validation + readiness checks + operator guidance.
 - Extended DNS checks that support the posture surface (e.g., MX/BIMI; optional DANE/TLSA where relevant).
diff --git a/docs/user_guide/domains.md b/docs/user_guide/domains.md
index ff55a74..d4857c6 100644
--- a/docs/user_guide/domains.md
+++ b/docs/user_guide/domains.md
@@ -59,9 +59,18 @@ DMARQ provides a health check feature for each domain:
    - SPF record validation
    - DKIM selector verification
    - DMARC record syntax check
+   - MTA-STS TXT and HTTPS policy validation
    - MX record confirmation
    - BIMI record validation (if applicable)
 
+### MTA-STS Posture
+
+The domain detail page checks `_mta-sts.<domain>` and fetches the policy from `https://mta-sts.<domain>/.well-known/mta-sts.txt`.
+
+DMARQ marks the check healthy when the TXT record contains `v=STSv1` with an `id`, the HTTPS policy is reachable, and the policy includes `version`, `mode`, `mx`, and `max_age`. Findings include the DNS record, policy URL, mode, MX patterns, and actionable guidance for missing records, fetch failures, invalid policies, or non-enforcing `testing`/`none` modes.
+
+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.
+
 ## Domain Groups
 
 If you manage multiple domains, you can organize them into groups:
@@ -86,4 +95,4 @@ To remove a domain from DMARQ:
 4. Select **Remove Domain**
 5. Confirm the removal
 
-Note that removing a domain will delete all stored DMARC reports for that domain.
\ No newline at end of file
+Note that removing a domain will delete all stored DMARC reports for that domain.