Merge PR #168: add Cloudflare DNS integration

Add Cloudflare REST-backed domain discovery, DNS inspection, recommendations, and DNS change tracking. Closes #31.

README.md

@@ -4,7 +4,7 @@
 
 🌐 [Live Demo (soon)](https://app.dmarq.org)  
 🔒 Self-hosted. Secure. Beautifully visual.  
-🛠️ Docker-deployable. DNS posture checks (Cloudflare inspection planned).  
+🛠️ Docker-deployable. DNS posture checks with optional Cloudflare inspection.
 📬 Aggregate report support (failure/forensic reports planned).
 
 ---
@@ -17,7 +17,7 @@ No more guessing. See which services are passing DMARC, which are failing, and h
 
 ---
 
-## 🚀 Current Status (Milestones 1–7 Complete)
+## 🚀 Current Status (Milestones 1–8 Complete)
 
 DMARQ currently supports end-to-end aggregate DMARC monitoring with mailbox ingestion, persistence, reporting, DNS checks, and notifications.
 
@@ -30,10 +30,10 @@ Included:
 - ✅ Import history + backfills for mail sources
 - ✅ Alerts & notifications via Apprise (test send, alert rules, daily/weekly summaries)
 - ✅ DNS checks (DMARC/SPF/DKIM) with DKIM selector discovery from report data
+- ✅ Cloudflare read-only domain discovery, DNS inspection, recommendations, and change tracking
 
 Up next:
 
-- 🔜 DNS health guidance + optional Cloudflare read-only inspection (Milestone 8)
 - 🔜 Setup and operations polish (Milestone 9)
 - 🧊 Failure/forensic report support (RUF) (Milestone 10)
 
@@ -54,8 +54,9 @@ Up next:
 - Show which records are missing, broken, or invalid
 - 🔒 No automatic changes — all DNS updates require explicit confirmation (when remediation workflows are added)
 
-### 🌐 Cloudflare Integration (Planned)
+### 🌐 Cloudflare Integration
 - Optional read-only domain discovery and DNS inspection
+- Import Cloudflare zones as monitored domains from Settings
 - Suggestions for missing or malformed entries
 - Track configuration changes over time
 
@@ -147,8 +148,7 @@ settings.
 
 ## 🧭 Development Roadmap
 
-- ✅ **Milestones 1–7**: Parsing, ingestion (upload/IMAP/Gmail), persistence, reporting, notifications, production hardening
-- 🔜 **Milestone 8**: DNS health guidance + Cloudflare read-only inspection
+- ✅ **Milestones 1–8**: Parsing, ingestion (upload/IMAP/Gmail), persistence, reporting, notifications, production hardening, DNS health, Cloudflare read-only inspection
 - 🔜 **Milestone 9**: Setup and operations polish
 - 🧊 **Milestone 10**: Failure/forensic report support (RUF)
 - 🧠 **Milestones 11–16**: DMARC format compatibility, Microsoft 365 ingestion, broader email posture, APIs/webhooks, workspaces/MSP, AI/MCP (see docs)

backend/alembic/versions/c0d1e2f3a4b5_add_dns_record_change_tracking.py

@@ -0,0 +1,172 @@
+"""add dns record change tracking
+
+Revision ID: c0d1e2f3a4b5
+Revises: b9c0d1e2f3a4
+Create Date: 2026-05-23 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "c0d1e2f3a4b5"
+down_revision: Union[str, Sequence[str], None] = "b9c0d1e2f3a4"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create DNS record snapshot and change history tables."""
+    op.create_table(
+        "dns_record_snapshots",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("domain", sa.String(), nullable=False),
+        sa.Column("provider", sa.String(), nullable=False),
+        sa.Column("zone_id", sa.String(), nullable=True),
+        sa.Column("record_key", sa.String(length=128), nullable=False),
+        sa.Column("record_id", sa.String(), nullable=True),
+        sa.Column("record_type", sa.String(length=20), nullable=False),
+        sa.Column("record_name", sa.String(), nullable=False),
+        sa.Column("content", sa.Text(), nullable=True),
+        sa.Column("proxied", sa.Boolean(), nullable=True),
+        sa.Column("ttl", sa.Integer(), nullable=True),
+        sa.Column("record_hash", sa.String(length=64), nullable=False),
+        sa.Column("active", sa.Boolean(), nullable=False),
+        sa.Column("first_seen_at", sa.DateTime(), nullable=False),
+        sa.Column("last_seen_at", sa.DateTime(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "domain", "provider", "record_key", name="uq_dns_record_snapshot_lookup"
+        ),
+    )
+    op.create_index(op.f("ix_dns_record_snapshots_id"), "dns_record_snapshots", ["id"])
+    op.create_index(op.f("ix_dns_record_snapshots_domain"), "dns_record_snapshots", ["domain"])
+    op.create_index(op.f("ix_dns_record_snapshots_provider"), "dns_record_snapshots", ["provider"])
+    op.create_index(op.f("ix_dns_record_snapshots_zone_id"), "dns_record_snapshots", ["zone_id"])
+    op.create_index(
+        op.f("ix_dns_record_snapshots_record_key"),
+        "dns_record_snapshots",
+        ["record_key"],
+    )
+    op.create_index(
+        op.f("ix_dns_record_snapshots_record_id"),
+        "dns_record_snapshots",
+        ["record_id"],
+    )
+    op.create_index(
+        op.f("ix_dns_record_snapshots_record_type"),
+        "dns_record_snapshots",
+        ["record_type"],
+    )
+    op.create_index(
+        op.f("ix_dns_record_snapshots_record_name"),
+        "dns_record_snapshots",
+        ["record_name"],
+    )
+    op.create_index(
+        op.f("ix_dns_record_snapshots_record_hash"),
+        "dns_record_snapshots",
+        ["record_hash"],
+    )
+    op.create_index(op.f("ix_dns_record_snapshots_active"), "dns_record_snapshots", ["active"])
+    op.create_index(
+        op.f("ix_dns_record_snapshots_first_seen_at"),
+        "dns_record_snapshots",
+        ["first_seen_at"],
+    )
+    op.create_index(
+        op.f("ix_dns_record_snapshots_last_seen_at"),
+        "dns_record_snapshots",
+        ["last_seen_at"],
+    )
+    op.create_index(
+        "ix_dns_record_snapshots_domain_active",
+        "dns_record_snapshots",
+        ["domain", "active"],
+    )
+    op.create_index(
+        "ix_dns_record_snapshots_domain_seen",
+        "dns_record_snapshots",
+        ["domain", "last_seen_at"],
+    )
+
+    op.create_table(
+        "dns_record_changes",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("domain", sa.String(), nullable=False),
+        sa.Column("provider", sa.String(), nullable=False),
+        sa.Column("zone_id", sa.String(), nullable=True),
+        sa.Column("record_key", sa.String(length=128), nullable=False),
+        sa.Column("record_id", sa.String(), nullable=True),
+        sa.Column("record_type", sa.String(length=20), nullable=False),
+        sa.Column("record_name", sa.String(), nullable=False),
+        sa.Column("change_type", sa.String(length=20), nullable=False),
+        sa.Column("previous_content", sa.Text(), nullable=True),
+        sa.Column("current_content", sa.Text(), nullable=True),
+        sa.Column("observed_at", sa.DateTime(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(op.f("ix_dns_record_changes_id"), "dns_record_changes", ["id"])
+    op.create_index(op.f("ix_dns_record_changes_domain"), "dns_record_changes", ["domain"])
+    op.create_index(op.f("ix_dns_record_changes_provider"), "dns_record_changes", ["provider"])
+    op.create_index(op.f("ix_dns_record_changes_zone_id"), "dns_record_changes", ["zone_id"])
+    op.create_index(op.f("ix_dns_record_changes_record_key"), "dns_record_changes", ["record_key"])
+    op.create_index(op.f("ix_dns_record_changes_record_id"), "dns_record_changes", ["record_id"])
+    op.create_index(
+        op.f("ix_dns_record_changes_record_type"), "dns_record_changes", ["record_type"]
+    )
+    op.create_index(
+        op.f("ix_dns_record_changes_record_name"), "dns_record_changes", ["record_name"]
+    )
+    op.create_index(
+        op.f("ix_dns_record_changes_change_type"), "dns_record_changes", ["change_type"]
+    )
+    op.create_index(
+        op.f("ix_dns_record_changes_observed_at"), "dns_record_changes", ["observed_at"]
+    )
+    op.create_index(
+        "ix_dns_record_changes_domain_observed",
+        "dns_record_changes",
+        ["domain", "observed_at"],
+    )
+    op.create_index(
+        "ix_dns_record_changes_record_observed",
+        "dns_record_changes",
+        ["record_key", "observed_at"],
+    )
+
+
+def downgrade() -> None:
+    """Drop DNS record snapshot and change history tables."""
+    op.drop_index("ix_dns_record_changes_record_observed", table_name="dns_record_changes")
+    op.drop_index("ix_dns_record_changes_domain_observed", table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_observed_at"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_change_type"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_record_name"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_record_type"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_record_id"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_record_key"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_zone_id"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_provider"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_domain"), table_name="dns_record_changes")
+    op.drop_index(op.f("ix_dns_record_changes_id"), table_name="dns_record_changes")
+    op.drop_table("dns_record_changes")
+
+    op.drop_index("ix_dns_record_snapshots_domain_seen", table_name="dns_record_snapshots")
+    op.drop_index("ix_dns_record_snapshots_domain_active", table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_last_seen_at"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_first_seen_at"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_active"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_record_hash"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_record_name"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_record_type"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_record_id"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_record_key"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_zone_id"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_provider"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_domain"), table_name="dns_record_snapshots")
+    op.drop_index(op.f("ix_dns_record_snapshots_id"), table_name="dns_record_snapshots")
+    op.drop_table("dns_record_snapshots")

backend/app/api/api_v1/endpoints/domains.py

@@ -13,6 +13,14 @@ from sqlalchemy.orm import Session
 
 from app.core.database import get_db
 from app.models.domain import Domain
+from app.services.cloudflare_dns import (
+    analyze_dns_records,
+    discover_cloudflare_zones,
+    get_zone_for_domain,
+    import_cloudflare_domains,
+    list_dns_record_changes,
+    sync_dns_record_changes,
+)
 from app.services.dns_cache import resolve_domain_dns_cached
 from app.services.dns_resolver import (
     DomainDNSResult,
@@ -69,6 +77,48 @@ class DNSRecordResponse(BaseModel):
     checkedAt: Optional[str] = None
 
 
+class CloudflareZoneResponse(BaseModel):
+    """Cloudflare zone available for import."""
+
+    id: str
+    name: str
+    status: Optional[str] = None
+    account_name: Optional[str] = None
+    imported: bool = False
+
+
+class CloudflareImportRequest(BaseModel):
+    """Optional list of Cloudflare domains to import."""
+
+    domains: Optional[List[str]] = None
+
+
+class CloudflareImportResponse(BaseModel):
+    """Cloudflare domain import summary."""
+
+    imported: List[str]
+    existing: List[str]
+    skipped: List[str]
+    total_discovered: int
+
+
+class CloudflareDNSAnalysisResponse(BaseModel):
+    """Cloudflare-managed DNS analysis and recent change details."""
+
+    zone: Dict[str, Any]
+    records: List[Dict[str, Any]]
+    checks: Dict[str, Any]
+    suggestions: List[Dict[str, str]]
+    changes: List[Dict[str, Any]]
+    history: List[Dict[str, Any]]
+
+
+class DNSChangeHistoryResponse(BaseModel):
+    """Recent DNS record changes for a domain."""
+
+    history: List[Dict[str, Any]]
+
+
 class TimelinePoint(BaseModel):
     """Data point for compliance timeline"""
 
@@ -199,6 +249,40 @@ def _get_domain_selectors_map_from_db(db: Session, domain_names: List[str]) -> D
     return selectors_by_domain
 
 
+def _policy_enforcement_suggestions(
+    dmarc_policy: Optional[str],
+    summary: Dict[str, Any],
+) -> List[Dict[str, str]]:
+    """Suggest policy enforcement when report history supports moving beyond monitoring."""
+    if dmarc_policy != "none":
+        return []
+    total_count = int(summary.get("total_count", 0) or 0)
+    compliance_rate = float(summary.get("compliance_rate", 0.0) or 0.0)
+    if total_count >= 100 and compliance_rate >= 98.0:
+        return [
+            {
+                "type": "policy_enforcement_ready",
+                "severity": "info",
+                "message": (
+                    "Recent reports show very high DMARC compliance. Consider moving from "
+                    "p=none to p=quarantine with a limited pct value."
+                ),
+            }
+        ]
+    if total_count >= 100 and compliance_rate >= 90.0:
+        return [
+            {
+                "type": "policy_enforcement_review",
+                "severity": "info",
+                "message": (
+                    "DMARC compliance is trending high. Review remaining failures before "
+                    "moving the domain policy beyond p=none."
+                ),
+            }
+        ]
+    return []
+
+
 @router.get("/summary", response_model=DomainSummaryResponse)
 async def get_domains_summary(db: Session = Depends(get_db)):
     """
@@ -215,7 +299,7 @@ async def get_domains_summary(db: Session = Depends(get_db)):
     summaries = store.get_all_domain_summaries()
 
     # Perform DNS checks for all domains, reusing fresh cached results.
-    provider = get_default_provider()
+    provider = get_default_provider(db)
     manual_selectors_by_domain = _get_domain_selectors_map_from_db(db, domains)
 
     async def _dns_for_domain(domain_name: str) -> DomainDNSResult:
@@ -413,7 +497,7 @@ async def get_domain_dns_records(
     report_selectors = _get_selectors_from_reports(store, domain_id)
     combined_selectors = list(dict.fromkeys(manual_selectors + report_selectors))
 
-    provider = get_default_provider()
+    provider = get_default_provider(db)
     result, cached, checked_at = await resolve_domain_dns_cached(
         db,
         provider,
@@ -434,6 +518,84 @@ async def get_domain_dns_records(
     )
 
 
+@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."""
+    try:
+        return await discover_cloudflare_zones(db)
+    except LookupError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(exc),
+        ) from exc
+
+
+@router.post("/cloudflare/import", response_model=CloudflareImportResponse)
+async def import_cloudflare_domain_zones(
+    payload: CloudflareImportRequest,
+    db: Session = Depends(get_db),
+):
+    """Import selected, or all, Cloudflare zones as monitored domains."""
+    try:
+        return await import_cloudflare_domains(db, requested_domains=payload.domains)
+    except LookupError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(exc),
+        ) from exc
+
+
+@router.get("/{domain_id}/dns/cloudflare", response_model=CloudflareDNSAnalysisResponse)
+async def get_cloudflare_domain_dns_analysis(
+    domain_id: str = Path(..., title="The domain ID or name"),
+    db: Session = Depends(get_db),
+):
+    """Analyze Cloudflare-managed DNS records and persist detected changes."""
+    try:
+        zone_data = await get_zone_for_domain(db, domain_id)
+    except LookupError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=str(exc),
+        ) from exc
+
+    records = zone_data["records"]
+    changes = sync_dns_record_changes(
+        db,
+        domain=domain_id,
+        zone_id=zone_data["id"],
+        records=records,
+    )
+    analysis = analyze_dns_records(domain_id, records)
+    store = ReportStore.get_instance()
+    hydrate_report_store_from_db(db, store)
+    analysis["suggestions"].extend(
+        _policy_enforcement_suggestions(
+            analysis["checks"].get("dmarc_policy"),
+            store.get_domain_summary(domain_id),
+        )
+    )
+    history = list_dns_record_changes(db, domain_id)
+    return CloudflareDNSAnalysisResponse(
+        zone={"id": zone_data["id"], "name": zone_data["name"]},
+        records=analysis["records"],
+        checks=analysis["checks"],
+        suggestions=analysis["suggestions"],
+        changes=changes,
+        history=history,
+    )
+
+
+@router.get("/{domain_id}/dns/history", response_model=DNSChangeHistoryResponse)
+async def get_domain_dns_change_history(
+    domain_id: str = Path(..., title="The domain ID or name"),
+    limit: int = Query(50, title="Maximum number of change events to return"),
+    db: Session = Depends(get_db),
+):
+    """Return recent provider-backed DNS record changes for a domain."""
+    return DNSChangeHistoryResponse(history=list_dns_record_changes(db, domain_id, limit=limit))
+
+
 @router.get("/{domain_id}/reports", response_model=DomainReportsResponse)
 async def get_domain_reports(
     domain_id: str = Path(..., title="The domain ID or name"),
@@ -800,7 +962,7 @@ async def get_domain_sources(
         )
 
     sources = store.get_domain_sources(domain_id, days=days)
-    provider = get_default_provider()
+    provider = get_default_provider(db)
 
     ips = [s.get("source_ip", "unknown") for s in sources]
     hostnames = await asyncio.gather(*[_safe_ptr_lookup(provider, ip) for ip in ips])

backend/app/models/dns_cache.py

@@ -1,6 +1,6 @@
 from datetime import UTC, datetime
 
-from sqlalchemy import Column, DateTime, Index, Integer, String, Text, UniqueConstraint
+from sqlalchemy import Boolean, Column, DateTime, Index, Integer, String, Text, UniqueConstraint
 
 from app.core.database import Base
 
@@ -28,3 +28,66 @@ class DNSCache(Base):
 
     def __repr__(self):
         return f"<DNSCache {self.domain} provider={self.provider}>"
+
+
+class DNSRecordSnapshot(Base):
+    """Last observed DNS record state for provider-backed DNS integrations."""
+
+    __tablename__ = "dns_record_snapshots"
+
+    id = Column(Integer, primary_key=True, index=True)
+    domain = Column(String, nullable=False, index=True)
+    provider = Column(String, nullable=False, index=True)
+    zone_id = Column(String, nullable=True, index=True)
+    record_key = Column(String(128), nullable=False, index=True)
+    record_id = Column(String, nullable=True, index=True)
+    record_type = Column(String(20), nullable=False, index=True)
+    record_name = Column(String, nullable=False, index=True)
+    content = Column(Text, nullable=True)
+    proxied = Column(Boolean, nullable=True)
+    ttl = Column(Integer, nullable=True)
+    record_hash = Column(String(64), nullable=False, index=True)
+    active = Column(Boolean, default=True, nullable=False, index=True)
+    first_seen_at = Column(DateTime, default=_utcnow_naive, nullable=False, index=True)
+    last_seen_at = Column(DateTime, default=_utcnow_naive, nullable=False, index=True)
+
+    __table_args__ = (
+        UniqueConstraint(
+            "domain",
+            "provider",
+            "record_key",
+            name="uq_dns_record_snapshot_lookup",
+        ),
+        Index("ix_dns_record_snapshots_domain_active", "domain", "active"),
+        Index("ix_dns_record_snapshots_domain_seen", "domain", "last_seen_at"),
+    )
+
+    def __repr__(self):
+        return f"<DNSRecordSnapshot {self.domain} {self.record_type} {self.record_name}>"
+
+
+class DNSRecordChange(Base):
+    """Append-only DNS record change event detected during provider sync."""
+
+    __tablename__ = "dns_record_changes"
+
+    id = Column(Integer, primary_key=True, index=True)
+    domain = Column(String, nullable=False, index=True)
+    provider = Column(String, nullable=False, index=True)
+    zone_id = Column(String, nullable=True, index=True)
+    record_key = Column(String(128), nullable=False, index=True)
+    record_id = Column(String, nullable=True, index=True)
+    record_type = Column(String(20), nullable=False, index=True)
+    record_name = Column(String, nullable=False, index=True)
+    change_type = Column(String(20), nullable=False, index=True)
+    previous_content = Column(Text, nullable=True)
+    current_content = Column(Text, nullable=True)
+    observed_at = Column(DateTime, default=_utcnow_naive, nullable=False, index=True)
+
+    __table_args__ = (
+        Index("ix_dns_record_changes_domain_observed", "domain", "observed_at"),
+        Index("ix_dns_record_changes_record_observed", "record_key", "observed_at"),
+    )
+
+    def __repr__(self):
+        return f"<DNSRecordChange {self.domain} {self.change_type} {self.record_name}>"

backend/app/services/cloudflare_dns.py

@@ -0,0 +1,427 @@
+"""Cloudflare DNS discovery, analysis, and change tracking."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.core.config import get_settings
+from app.core.credential_encryption import decrypt_secret
+from app.models.dns_cache import DNSRecordChange, DNSRecordSnapshot
+from app.models.domain import Domain
+from app.models.setting import Setting
+from app.services.dns_resolver import CloudflareDNSProvider, extract_dmarc_policy
+
+PROVIDER_NAME = "cloudflare"
+
+
+@dataclass
+class CloudflareCredentials:
+    """Resolved Cloudflare credentials from persisted settings or environment."""
+
+    api_token: Optional[str] = None
+    zone_id: Optional[str] = None
+
+    @property
+    def configured(self) -> bool:
+        return bool(self.api_token)
+
+
+def _utcnow_naive() -> datetime:
+    return datetime.now(UTC).replace(tzinfo=None)
+
+
+def _plain_setting_value(db: Session, key: str) -> Optional[str]:
+    row = db.query(Setting).filter(Setting.key == key).first()
+    if row is None or not row.value:
+        return None
+    if key == "cloudflare.api_token":
+        return decrypt_secret(row.value)
+    return row.value
+
+
+def get_cloudflare_credentials(db: Session) -> CloudflareCredentials:
+    """Resolve Cloudflare credentials from app settings, falling back to env vars."""
+    settings = get_settings()
+    return CloudflareCredentials(
+        api_token=_plain_setting_value(db, "cloudflare.api_token") or settings.CLOUDFLARE_API_TOKEN,
+        zone_id=_plain_setting_value(db, "cloudflare.zone_id") or settings.CLOUDFLARE_ZONE_ID,
+    )
+
+
+def build_cloudflare_provider(db: Session) -> CloudflareDNSProvider:
+    """Return a Cloudflare provider configured from settings and environment."""
+    credentials = get_cloudflare_credentials(db)
+    if not credentials.configured:
+        raise LookupError("Cloudflare API token is not configured")
+    return CloudflareDNSProvider(
+        api_token=credentials.api_token,
+        zone_id=credentials.zone_id,
+    )
+
+
+async def discover_cloudflare_zones(db: Session) -> List[Dict[str, Any]]:
+    """Return zones visible to the configured Cloudflare token with import state."""
+    provider = build_cloudflare_provider(db)
+    known_domains = {name for (name,) in db.query(Domain.name).all()}
+    zones = await provider.list_zones()
+    return [
+        {
+            "id": zone.get("id"),
+            "name": zone.get("name"),
+            "status": zone.get("status"),
+            "account_name": (zone.get("account") or {}).get("name"),
+            "imported": zone.get("name") in known_domains,
+        }
+        for zone in zones
+        if zone.get("id") and zone.get("name")
+    ]
+
+
+async def import_cloudflare_domains(
+    db: Session,
+    *,
+    requested_domains: Optional[List[str]] = None,
+) -> Dict[str, Any]:
+    """Create Domain rows for Cloudflare zones, returning imported and existing names."""
+    zones = await discover_cloudflare_zones(db)
+    requested = {domain.strip().lower() for domain in requested_domains or [] if domain.strip()}
+    imported: List[str] = []
+    existing: List[str] = []
+    skipped: List[str] = []
+
+    for zone in zones:
+        name = str(zone["name"]).lower()
+        if requested and name not in requested:
+            skipped.append(name)
+            continue
+        domain = db.query(Domain).filter(Domain.name == name).first()
+        if domain is None:
+            db.add(Domain(name=name, active=True, verified=True))
+            imported.append(name)
+        else:
+            existing.append(name)
+
+    db.commit()
+    return {
+        "imported": imported,
+        "existing": existing,
+        "skipped": skipped,
+        "total_discovered": len(zones),
+    }
+
+
+async def get_zone_for_domain(db: Session, domain: str) -> Dict[str, Any]:
+    """Resolve the Cloudflare zone for a domain name."""
+    provider = build_cloudflare_provider(db)
+    credentials = get_cloudflare_credentials(db)
+    if credentials.zone_id:
+        records = await provider.list_dns_records(zone_id=credentials.zone_id)
+        try:
+            zone = await provider.find_zone_for_domain(domain)
+        except LookupError:
+            zone = None
+        return {
+            "id": credentials.zone_id,
+            "name": zone.get("name") if zone else domain,
+            "records": records,
+        }
+    zone = await provider.find_zone_for_domain(domain)
+    if not zone:
+        raise LookupError(f"No Cloudflare zone found for {domain}")
+    records = await provider.list_dns_records(zone_id=zone["id"])
+    return {"id": zone["id"], "name": zone["name"], "records": records}
+
+
+def _record_key(record: Dict[str, Any]) -> str:
+    record_id = record.get("id")
+    if record_id:
+        return str(record_id)
+    payload = json.dumps(
+        {
+            "type": record.get("type"),
+            "name": record.get("name"),
+            "content": record.get("content"),
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _record_hash(record: Dict[str, Any]) -> str:
+    payload = json.dumps(
+        {
+            "type": record.get("type"),
+            "name": record.get("name"),
+            "content": record.get("content"),
+            "proxied": record.get("proxied"),
+            "ttl": record.get("ttl"),
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _change_to_dict(change: DNSRecordChange) -> Dict[str, Any]:
+    return {
+        "id": change.id,
+        "domain": change.domain,
+        "provider": change.provider,
+        "zone_id": change.zone_id,
+        "record_type": change.record_type,
+        "record_name": change.record_name,
+        "change_type": change.change_type,
+        "previous_content": change.previous_content,
+        "current_content": change.current_content,
+        "observed_at": change.observed_at.isoformat() if change.observed_at else None,
+    }
+
+
+def sync_dns_record_changes(
+    db: Session,
+    *,
+    domain: str,
+    zone_id: str,
+    records: List[Dict[str, Any]],
+) -> List[Dict[str, Any]]:
+    """Track additions, modifications, and removals for a Cloudflare DNS snapshot."""
+    now = _utcnow_naive()
+    existing = {
+        snapshot.record_key: snapshot
+        for snapshot in db.query(DNSRecordSnapshot)
+        .filter(
+            DNSRecordSnapshot.domain == domain,
+            DNSRecordSnapshot.provider == PROVIDER_NAME,
+            DNSRecordSnapshot.zone_id == zone_id,
+            DNSRecordSnapshot.active == True,  # noqa: E712
+        )
+        .all()
+    }
+    seen: set[str] = set()
+    changes: List[DNSRecordChange] = []
+
+    for record in records:
+        record_type = str(record.get("type") or "").upper()
+        record_name = str(record.get("name") or "")
+        if not record_type or not record_name:
+            continue
+        key = _record_key(record)
+        seen.add(key)
+        content = record.get("content")
+        current_hash = _record_hash(record)
+        snapshot = existing.get(key)
+        if snapshot is None:
+            snapshot = DNSRecordSnapshot(
+                domain=domain,
+                provider=PROVIDER_NAME,
+                zone_id=zone_id,
+                record_key=key,
+                record_id=record.get("id"),
+                record_type=record_type,
+                record_name=record_name,
+                content=content,
+                proxied=record.get("proxied"),
+                ttl=record.get("ttl"),
+                record_hash=current_hash,
+                active=True,
+                first_seen_at=now,
+                last_seen_at=now,
+            )
+            db.add(snapshot)
+            changes.append(
+                DNSRecordChange(
+                    domain=domain,
+                    provider=PROVIDER_NAME,
+                    zone_id=zone_id,
+                    record_key=key,
+                    record_id=record.get("id"),
+                    record_type=record_type,
+                    record_name=record_name,
+                    change_type="added",
+                    current_content=content,
+                    observed_at=now,
+                )
+            )
+            continue
+
+        if snapshot.record_hash != current_hash:
+            changes.append(
+                DNSRecordChange(
+                    domain=domain,
+                    provider=PROVIDER_NAME,
+                    zone_id=zone_id,
+                    record_key=key,
+                    record_id=record.get("id"),
+                    record_type=record_type,
+                    record_name=record_name,
+                    change_type="modified",
+                    previous_content=snapshot.content,
+                    current_content=content,
+                    observed_at=now,
+                )
+            )
+            snapshot.content = content
+            snapshot.proxied = record.get("proxied")
+            snapshot.ttl = record.get("ttl")
+            snapshot.record_hash = current_hash
+        snapshot.record_id = record.get("id")
+        snapshot.record_type = record_type
+        snapshot.record_name = record_name
+        snapshot.active = True
+        snapshot.last_seen_at = now
+
+    for key, snapshot in existing.items():
+        if key in seen:
+            continue
+        snapshot.active = False
+        snapshot.last_seen_at = now
+        changes.append(
+            DNSRecordChange(
+                domain=domain,
+                provider=PROVIDER_NAME,
+                zone_id=zone_id,
+                record_key=key,
+                record_id=snapshot.record_id,
+                record_type=snapshot.record_type,
+                record_name=snapshot.record_name,
+                change_type="removed",
+                previous_content=snapshot.content,
+                observed_at=now,
+            )
+        )
+
+    for change in changes:
+        db.add(change)
+    db.commit()
+    for change in changes:
+        db.refresh(change)
+    return [_change_to_dict(change) for change in changes]
+
+
+def list_dns_record_changes(db: Session, domain: str, *, limit: int = 50) -> List[Dict[str, Any]]:
+    """Return recent DNS record change events for a domain."""
+    rows = (
+        db.query(DNSRecordChange)
+        .filter(DNSRecordChange.domain == domain)
+        .order_by(DNSRecordChange.observed_at.desc(), DNSRecordChange.id.desc())
+        .limit(max(1, min(limit, 200)))
+        .all()
+    )
+    return [_change_to_dict(row) for row in rows]
+
+
+def _txt_contents(records: List[Dict[str, Any]], name: str) -> List[str]:
+    target = name.rstrip(".").lower()
+    return [
+        str(record.get("content") or "")
+        for record in records
+        if str(record.get("type") or "").upper() == "TXT"
+        and str(record.get("name") or "").rstrip(".").lower() == target
+    ]
+
+
+def _cloudflare_record_to_dict(record: Dict[str, Any]) -> Dict[str, Any]:
+    return {
+        "id": record.get("id"),
+        "type": record.get("type"),
+        "name": record.get("name"),
+        "content": record.get("content"),
+        "ttl": record.get("ttl"),
+        "proxied": record.get("proxied"),
+        "modified_on": record.get("modified_on"),
+    }
+
+
+def analyze_dns_records(domain: str, records: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Analyze Cloudflare DNS records and return checks plus actionable suggestions."""
+    root_txt = _txt_contents(records, domain)
+    dmarc_records = _txt_contents(records, f"_dmarc.{domain}")
+    spf_records = [record for record in root_txt if record.lower().startswith("v=spf1")]
+    dmarc_auth_records = [
+        record for record in dmarc_records if record.lower().startswith("v=dmarc1")
+    ]
+    dkim_records = [
+        record
+        for record in records
+        if str(record.get("type") or "").upper() == "TXT"
+        and "._domainkey." in str(record.get("name") or "").lower()
+        and ("v=dkim1" in str(record.get("content") or "").lower())
+    ]
+
+    suggestions: List[Dict[str, str]] = []
+    if not dmarc_auth_records:
+        suggestions.append(
+            {
+                "type": "missing_dmarc",
+                "severity": "error",
+                "message": "Add a TXT record at _dmarc with a v=DMARC1 policy.",
+            }
+        )
+    elif len(dmarc_auth_records) > 1:
+        suggestions.append(
+            {
+                "type": "duplicate_dmarc",
+                "severity": "error",
+                "message": "Keep exactly one DMARC TXT record at _dmarc.",
+            }
+        )
+    elif extract_dmarc_policy(dmarc_auth_records[0]) is None:
+        suggestions.append(
+            {
+                "type": "malformed_dmarc",
+                "severity": "error",
+                "message": "Add a p=none, p=quarantine, or p=reject tag to the DMARC record.",
+            }
+        )
+
+    if not spf_records:
+        suggestions.append(
+            {
+                "type": "missing_spf",
+                "severity": "warning",
+                "message": "Add an SPF TXT record at the root domain for authorized senders.",
+            }
+        )
+    elif len(spf_records) > 1:
+        suggestions.append(
+            {
+                "type": "duplicate_spf",
+                "severity": "error",
+                "message": "Merge multiple SPF records into a single v=spf1 TXT record.",
+            }
+        )
+
+    if not dkim_records:
+        suggestions.append(
+            {
+                "type": "missing_dkim",
+                "severity": "warning",
+                "message": "No DKIM TXT records were found; configure DKIM for active mail providers.",
+            }
+        )
+
+    return {
+        "records": [_cloudflare_record_to_dict(record) for record in records],
+        "checks": {
+            "dmarc": bool(dmarc_auth_records),
+            "dmarc_record": dmarc_auth_records[0] if dmarc_auth_records else None,
+            "dmarc_policy": (
+                extract_dmarc_policy(dmarc_auth_records[0]) if dmarc_auth_records else None
+            ),
+            "spf": len(spf_records) == 1,
+            "spf_record": spf_records[0] if spf_records else None,
+            "dkim": bool(dkim_records),
+            "dkim_records": [
+                _cloudflare_record_to_dict(record)
+                for record in sorted(dkim_records, key=lambda item: str(item.get("name") or ""))
+            ],
+        },
+        "suggestions": suggestions,
+    }

backend/app/services/dns_resolver.py

@@ -11,7 +11,7 @@ import ipaddress
 import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 logger = logging.getLogger(__name__)
 
@@ -239,24 +239,16 @@ class SystemDNSProvider(BaseDNSProvider):
 
 
 class CloudflareDNSProvider(BaseDNSProvider):
-    """DNS provider using Cloudflare's DNS-over-HTTPS (DoH) endpoint.
+    """DNS provider using Cloudflare DoH and, when configured, the REST API.
 
-    This provider resolves DNS queries via Cloudflare's public DoH API
-    (``1.1.1.1`` / ``cloudflare-dns.com``).  When *api_token* and *zone_id*
-    are supplied, future versions will also support reading and writing DNS
-    records directly through the Cloudflare REST API, enabling automated DNS
-    synchronisation.
-
-    Current status
-    --------------
-    * DoH-based lookups are fully functional.
-    * Direct Cloudflare API integration (zone management, record sync) is
-      reserved for a future release.
+    Public DNS lookups continue to use Cloudflare's DNS-over-HTTPS endpoint.
+    If an API token is supplied, the provider can also discover account zones
+    and read managed DNS records directly from the Cloudflare REST API.
     """
 
     #: Cloudflare DNS-over-HTTPS endpoint (JSON wire format)
     CLOUDFLARE_DOH_URL: str = "https://cloudflare-dns.com/dns-query"
-    #: Cloudflare REST API base URL (for future zone-management support)
+    #: Cloudflare REST API base URL
     CLOUDFLARE_API_BASE: str = "https://api.cloudflare.com/client/v4"
 
     def __init__(
@@ -268,15 +260,121 @@ class CloudflareDNSProvider(BaseDNSProvider):
         Parameters
         ----------
         api_token:
-            Cloudflare API token.  Required for future DNS record management;
-            not needed for read-only DoH lookups.
+            Cloudflare API token. Required for zone discovery and managed
+            DNS record reads; not needed for read-only DoH lookups.
         zone_id:
-            Cloudflare zone identifier.  Required for future DNS record
-            management.
+            Optional Cloudflare zone identifier used as a preferred zone.
         """
         self.api_token = api_token
         self.zone_id = zone_id
 
+    def _auth_headers(self) -> Dict[str, str]:
+        if not self.api_token:
+            raise LookupError("Cloudflare API token is not configured")
+        return {
+            "Authorization": f"Bearer {self.api_token}",
+            "Accept": "application/json",
+        }
+
+    async def _api_get(
+        self,
+        path: str,
+        *,
+        params: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """Call Cloudflare's REST API and return the decoded response."""
+        import httpx  # type: ignore[import]
+
+        url = f"{self.CLOUDFLARE_API_BASE}{path}"
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.get(
+                    url,
+                    params=params,
+                    headers=self._auth_headers(),
+                    timeout=DNS_TIMEOUT,
+                )
+                response.raise_for_status()
+                data = response.json()
+        except (httpx.RequestError, httpx.HTTPStatusError, httpx.TimeoutException) as exc:
+            raise LookupError(f"Cloudflare API request failed for {path}: {exc}") from exc
+
+        if not data.get("success", False):
+            errors = data.get("errors") or []
+            message = "; ".join(str(error.get("message", error)) for error in errors[:3])
+            raise LookupError(message or f"Cloudflare API request failed for {path}")
+        return data
+
+    async def list_zones(self) -> List[Dict[str, Any]]:
+        """Return all zones visible to the configured Cloudflare API token."""
+        zones: List[Dict[str, Any]] = []
+        page = 1
+        while True:
+            data = await self._api_get(
+                "/zones",
+                params={"page": page, "per_page": 50, "status": "active"},
+            )
+            result = data.get("result") or []
+            if not isinstance(result, list):
+                return zones
+            zones.extend(result)
+            info = data.get("result_info") or {}
+            total_pages = int(info.get("total_pages") or 1)
+            if page >= total_pages:
+                return zones
+            page += 1
+
+    async def find_zone_for_domain(self, domain: str) -> Optional[Dict[str, Any]]:
+        """Return the best matching Cloudflare zone for *domain*."""
+        zones = await self.list_zones()
+        domain_lc = domain.rstrip(".").lower()
+        matches = [
+            zone
+            for zone in zones
+            if isinstance(zone.get("name"), str)
+            and (
+                domain_lc == zone["name"].lower() or domain_lc.endswith(f".{zone['name'].lower()}")
+            )
+        ]
+        if not matches:
+            return None
+        return sorted(matches, key=lambda zone: len(zone.get("name", "")), reverse=True)[0]
+
+    async def list_dns_records(
+        self,
+        *,
+        zone_id: Optional[str] = None,
+        name: Optional[str] = None,
+        record_type: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        """Return DNS records for a Cloudflare zone."""
+        resolved_zone_id = zone_id or self.zone_id
+        if not resolved_zone_id:
+            raise LookupError("Cloudflare zone ID is not configured")
+
+        records: List[Dict[str, Any]] = []
+        page = 1
+        while True:
+            params: Dict[str, Any] = {"page": page, "per_page": 100}
+            if name:
+                params["name"] = name
+            if record_type:
+                params["type"] = record_type
+
+            data = await self._api_get(
+                f"/zones/{resolved_zone_id}/dns_records",
+                params=params,
+            )
+            result = data.get("result") or []
+            if not isinstance(result, list):
+                return records
+            records.extend(result)
+            info = data.get("result_info") or {}
+            total_pages = int(info.get("total_pages") or 1)
+            if page >= total_pages:
+                return records
+            page += 1
+
     async def lookup_txt(self, name: str) -> List[str]:
         """Resolve TXT records via Cloudflare's DoH endpoint (JSON format)."""
         import httpx  # type: ignore[import]
@@ -332,13 +430,42 @@ class CloudflareDNSProvider(BaseDNSProvider):
         return None
 
 
-def get_default_provider() -> BaseDNSProvider:
-    """Return the default DNS provider (system resolver).
+def _decrypt_setting_value(value: Optional[str]) -> Optional[str]:
+    if not value:
+        return value
+    try:
+        from app.core.credential_encryption import decrypt_secret
 
-    In a future release this function will inspect application settings and
-    return a ``CloudflareDNSProvider`` when Cloudflare credentials are
-    configured.
-    """
+        return decrypt_secret(value)
+    except Exception:
+        return value
+
+
+def _setting_value(db: Any, key: str) -> Optional[str]:
+    if db is None:
+        return None
+    try:
+        from app.models.setting import Setting
+
+        row = db.query(Setting).filter(Setting.key == key).first()
+        return row.value if row is not None else None
+    except Exception:
+        return None
+
+
+def get_default_provider(db: Any = None) -> BaseDNSProvider:
+    """Return the configured default DNS provider."""
+    resolver = (_setting_value(db, "dns.resolver") or "").strip().lower()
+    if resolver == "cloudflare":
+        from app.core.config import get_settings
+
+        settings = get_settings()
+        api_token = _decrypt_setting_value(_setting_value(db, "cloudflare.api_token"))
+        zone_id = _setting_value(db, "cloudflare.zone_id")
+        return CloudflareDNSProvider(
+            api_token=api_token or settings.CLOUDFLARE_API_TOKEN,
+            zone_id=zone_id or settings.CLOUDFLARE_ZONE_ID,
+        )
     return SystemDNSProvider()
 
 

backend/app/templates/settings.html

@@ -189,7 +189,65 @@
                     <input type="text" x-model="s['cloudflare.zone_id']"
                         class="input input-bordered w-full"
                         placeholder="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" />
-                    <label class="label"><span class="label-text-alt text-muted-foreground">Found on the Cloudflare dashboard overview for your domain</span></label>
+                    <label class="label"><span class="label-text-alt text-muted-foreground">Optional. Leave blank to discover all zones available to the token.</span></label>
+                </div>
+                <div class="border-t border-border pt-4 space-y-4">
+                    <div class="flex flex-col gap-3 sm:flex-row sm:items-center sm:justify-between">
+                        <div>
+                            <h3 class="text-sm font-semibold">Domain Discovery</h3>
+                            <p class="text-sm text-muted-foreground">Find Cloudflare zones and import them into DMARQ.</p>
+                        </div>
+                        <div class="flex flex-col gap-2 sm:flex-row">
+                            <button type="button" class="btn btn-outline btn-sm" :disabled="saving || loadingCfZones" @click="discoverCloudflareZones()">
+                                <template x-if="!loadingCfZones">
+                                    <svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" viewBox="0 0 24 24" fill="none"
+                                        stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" class="mr-2">
+                                        <path d="M3 12a9 9 0 1 0 9-9 9.75 9.75 0 0 0-6.74 2.74L3 8"></path>
+                                        <path d="M3 3v5h5"></path>
+                                    </svg>
+                                </template>
+                                <template x-if="loadingCfZones"><span class="loading loading-spinner loading-xs mr-2"></span></template>
+                                Discover
+                            </button>
+                            <button type="button" class="btn btn-default btn-sm" :disabled="saving || importingCfZones || cfZones.length === 0" @click="importCloudflareZones()">
+                                <template x-if="!importingCfZones">
+                                    <svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" viewBox="0 0 24 24" fill="none"
+                                        stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" class="mr-2">
+                                        <path d="M12 5v14"></path>
+                                        <path d="m19 12-7 7-7-7"></path>
+                                    </svg>
+                                </template>
+                                <template x-if="importingCfZones"><span class="loading loading-spinner loading-xs mr-2"></span></template>
+                                Import New
+                            </button>
+                        </div>
+                    </div>
+                    <template x-if="cfZones.length > 0">
+                        <div class="overflow-x-auto rounded-md border border-border">
+                            <table class="table table-sm">
+                                <thead>
+                                    <tr>
+                                        <th>Domain</th>
+                                        <th>Status</th>
+                                        <th>Account</th>
+                                        <th>Imported</th>
+                                    </tr>
+                                </thead>
+                                <tbody>
+                                    <template x-for="zone in cfZones" :key="zone.id">
+                                        <tr>
+                                            <td class="font-medium" x-text="zone.name"></td>
+                                            <td x-text="zone.status || 'unknown'"></td>
+                                            <td x-text="zone.account_name || ''"></td>
+                                            <td>
+                                                <span class="badge" :class="zone.imported ? 'badge-success' : 'badge-outline'" x-text="zone.imported ? 'Yes' : 'No'"></span>
+                                            </td>
+                                        </tr>
+                                    </template>
+                                </tbody>
+                            </table>
+                        </div>
+                    </template>
                 </div>
                 <div class="flex justify-end">
                     <button type="submit" class="btn btn-default btn-md" :disabled="saving">
@@ -577,6 +635,9 @@ function settingsApp() {
         alertHistory: [],
         loadingConfigAudit: false,
         configAudit: [],
+        loadingCfZones: false,
+        importingCfZones: false,
+        cfZones: [],
         showCfToken: false,
 
         // Session cookie is sent automatically by the browser (httpOnly, same-origin).
@@ -804,6 +865,50 @@ function settingsApp() {
             }
         },
 
+        async discoverCloudflareZones() {
+            this.loadingCfZones = true;
+            try {
+                await this.saveCategory('cloudflare');
+                const res = await fetch('/api/v1/domains/cloudflare/discover', {
+                    headers: this.apiHeaders(),
+                });
+                const data = await res.json().catch(() => ({}));
+                if (!res.ok) {
+                    this.showFlash('Cloudflare discovery failed: ' + (data.detail || res.statusText), false);
+                } else {
+                    this.cfZones = data || [];
+                    this.showFlash(`${this.cfZones.length} Cloudflare zone${this.cfZones.length === 1 ? '' : 's'} found.`, true);
+                }
+            } catch (err) {
+                this.showFlash('Error discovering Cloudflare zones: ' + err.message, false);
+            } finally {
+                this.loadingCfZones = false;
+            }
+        },
+
+        async importCloudflareZones() {
+            this.importingCfZones = true;
+            try {
+                const domains = this.cfZones.filter(z => !z.imported).map(z => z.name);
+                const res = await fetch('/api/v1/domains/cloudflare/import', {
+                    method: 'POST',
+                    headers: this.apiHeaders(),
+                    body: JSON.stringify({ domains }),
+                });
+                const data = await res.json().catch(() => ({}));
+                if (!res.ok) {
+                    this.showFlash('Cloudflare import failed: ' + (data.detail || res.statusText), false);
+                } else {
+                    await this.discoverCloudflareZones();
+                    this.showFlash(`${data.imported.length} domain${data.imported.length === 1 ? '' : 's'} imported.`, true);
+                }
+            } catch (err) {
+                this.showFlash('Error importing Cloudflare zones: ' + err.message, false);
+            } finally {
+                this.importingCfZones = false;
+            }
+        },
+
         showFlash(msg, ok) {
             this.flashMsg = msg;
             this.flashOk = ok;

backend/app/tests/test_cloudflare_dns.py

@@ -0,0 +1,426 @@
+"""Tests for Cloudflare DNS discovery, analysis, and change tracking."""
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+from fastapi.testclient import TestClient
+
+from app.api.api_v1.endpoints.domains import _policy_enforcement_suggestions
+from app.core.credential_encryption import encrypt_secret
+from app.models.dns_cache import DNSRecordChange, DNSRecordSnapshot
+from app.models.domain import Domain
+from app.models.setting import Setting
+from app.services import cloudflare_dns
+from app.services.cloudflare_dns import analyze_dns_records, sync_dns_record_changes
+
+DOMAIN = "example.com"
+
+
+def _record(record_id: str, record_type: str, name: str, content: str, ttl: int = 1):
+    return {
+        "id": record_id,
+        "type": record_type,
+        "name": name,
+        "content": content,
+        "ttl": ttl,
+        "proxied": False,
+        "modified_on": "2026-05-23T00:00:00Z",
+    }
+
+
+class FakeCloudflareProvider:
+    def __init__(self, *, zones=None, records=None, fail_zone_lookup=False):
+        self.zones = zones or []
+        self.records = records or []
+        self.fail_zone_lookup = fail_zone_lookup
+
+    async def list_zones(self):
+        return self.zones
+
+    async def list_dns_records(self, *, zone_id=None, name=None, record_type=None):
+        return self.records
+
+    async def find_zone_for_domain(self, domain):
+        if self.fail_zone_lookup:
+            raise LookupError("zone list forbidden")
+        for zone in self.zones:
+            zone_name = zone["name"]
+            if domain == zone_name or domain.endswith(f".{zone_name}"):
+                return zone
+        return None
+
+
+def test_analyze_dns_records_reports_healthy_auth_records():
+    records = [
+        _record("spf", "TXT", DOMAIN, "v=spf1 include:_spf.google.com ~all"),
+        _record("dmarc", "TXT", f"_dmarc.{DOMAIN}", "v=DMARC1; p=quarantine"),
+        _record("dkim", "TXT", f"google._domainkey.{DOMAIN}", "v=DKIM1; p=abc"),
+    ]
+
+    result = analyze_dns_records(DOMAIN, records)
+
+    assert result["checks"]["dmarc"] is True
+    assert result["checks"]["dmarc_policy"] == "quarantine"
+    assert result["checks"]["spf"] is True
+    assert result["checks"]["dkim"] is True
+    assert result["suggestions"] == []
+
+
+def test_analyze_dns_records_suggests_missing_and_duplicate_fixes():
+    records = [
+        _record("spf-1", "TXT", DOMAIN, "v=spf1 include:_spf.google.com ~all"),
+        _record("spf-2", "TXT", DOMAIN, "v=spf1 ip4:192.0.2.10 ~all"),
+    ]
+
+    result = analyze_dns_records(DOMAIN, records)
+    suggestion_types = {item["type"] for item in result["suggestions"]}
+
+    assert "missing_dmarc" in suggestion_types
+    assert "duplicate_spf" in suggestion_types
+    assert "missing_dkim" in suggestion_types
+
+
+def test_analyze_dns_records_suggests_duplicate_dmarc_fix():
+    records = [
+        _record("dmarc-1", "TXT", f"_dmarc.{DOMAIN}", "v=DMARC1; p=none"),
+        _record("dmarc-2", "TXT", f"_dmarc.{DOMAIN}", "v=DMARC1; p=reject"),
+        _record("spf", "TXT", DOMAIN, "v=spf1 include:_spf.google.com ~all"),
+    ]
+
+    result = analyze_dns_records(DOMAIN, records)
+
+    assert {item["type"] for item in result["suggestions"]} == {
+        "duplicate_dmarc",
+        "missing_dkim",
+    }
+
+
+def test_analyze_dns_records_suggests_malformed_dmarc_fix():
+    records = [
+        _record("dmarc", "TXT", f"_dmarc.{DOMAIN}", "v=DMARC1; rua=mailto:dmarc@example.com"),
+        _record("spf", "TXT", DOMAIN, "v=spf1 include:_spf.google.com ~all"),
+    ]
+
+    result = analyze_dns_records(DOMAIN, records)
+
+    assert "malformed_dmarc" in {item["type"] for item in result["suggestions"]}
+
+
+def test_sync_dns_record_changes_tracks_add_modify_and_remove(db_session):
+    first_records = [
+        _record("spf", "TXT", DOMAIN, "v=spf1 include:_spf.google.com ~all"),
+        _record("dmarc", "TXT", f"_dmarc.{DOMAIN}", "v=DMARC1; p=none"),
+    ]
+    initial_changes = sync_dns_record_changes(
+        db_session,
+        domain=DOMAIN,
+        zone_id="zone-1",
+        records=first_records,
+    )
+
+    assert [change["change_type"] for change in initial_changes] == ["added", "added"]
+    assert db_session.query(DNSRecordSnapshot).count() == 2
+
+    no_changes = sync_dns_record_changes(
+        db_session,
+        domain=DOMAIN,
+        zone_id="zone-1",
+        records=first_records,
+    )
+
+    assert no_changes == []
+
+    second_records = [
+        _record("spf", "TXT", DOMAIN, "v=spf1 include:_spf.google.com -all"),
+    ]
+    later_changes = sync_dns_record_changes(
+        db_session,
+        domain=DOMAIN,
+        zone_id="zone-1",
+        records=second_records,
+    )
+    change_types = {change["change_type"] for change in later_changes}
+
+    assert change_types == {"modified", "removed"}
+    assert db_session.query(DNSRecordChange).count() == 4
+    removed_snapshot = (
+        db_session.query(DNSRecordSnapshot)
+        .filter(DNSRecordSnapshot.record_name == f"_dmarc.{DOMAIN}")
+        .first()
+    )
+    assert removed_snapshot.active is False
+
+
+def test_sync_dns_record_changes_ignores_incomplete_records(db_session):
+    changes = sync_dns_record_changes(
+        db_session,
+        domain=DOMAIN,
+        zone_id="zone-1",
+        records=[{"id": "bad", "type": "TXT", "content": "v=spf1 -all"}],
+    )
+
+    assert changes == []
+    assert db_session.query(DNSRecordSnapshot).count() == 0
+
+
+def test_list_dns_record_changes_clamps_limit(db_session):
+    sync_dns_record_changes(
+        db_session,
+        domain=DOMAIN,
+        zone_id="zone-1",
+        records=[_record("spf", "TXT", DOMAIN, "v=spf1 ~all")],
+    )
+
+    history = cloudflare_dns.list_dns_record_changes(db_session, DOMAIN, limit=0)
+
+    assert len(history) == 1
+    assert history[0]["change_type"] == "added"
+
+
+def test_cloudflare_credentials_read_encrypted_settings(db_session):
+    db_session.add_all(
+        [
+            Setting(
+                key="cloudflare.api_token",
+                value=encrypt_secret("cf-token"),
+                category="cloudflare",
+            ),
+            Setting(key="cloudflare.zone_id", value="zone-1", category="cloudflare"),
+        ]
+    )
+    db_session.commit()
+
+    credentials = cloudflare_dns.get_cloudflare_credentials(db_session)
+
+    assert credentials.configured is True
+    assert credentials.api_token == "cf-token"
+    assert credentials.zone_id == "zone-1"
+
+
+def test_build_cloudflare_provider_requires_token(db_session):
+    try:
+        cloudflare_dns.build_cloudflare_provider(db_session)
+    except LookupError as exc:
+        assert "Cloudflare API token" in str(exc)
+    else:
+        raise AssertionError("Expected LookupError")
+
+
+def test_discover_cloudflare_zones_marks_imported_and_filters_invalid(db_session):
+    db_session.add(Domain(name=DOMAIN))
+    db_session.commit()
+    provider = FakeCloudflareProvider(
+        zones=[
+            {
+                "id": "zone-1",
+                "name": DOMAIN,
+                "status": "active",
+                "account": {"name": "Example"},
+            },
+            {"id": None, "name": "invalid.example"},
+        ]
+    )
+
+    with patch("app.services.cloudflare_dns.build_cloudflare_provider", return_value=provider):
+        zones = asyncio.run(cloudflare_dns.discover_cloudflare_zones(db_session))
+
+    assert zones == [
+        {
+            "id": "zone-1",
+            "name": DOMAIN,
+            "status": "active",
+            "account_name": "Example",
+            "imported": True,
+        }
+    ]
+
+
+def test_import_cloudflare_domains_imports_requested_and_skips_others(db_session):
+    db_session.add(Domain(name=DOMAIN))
+    db_session.commit()
+
+    async def fake_discover(_db):
+        return [
+            {"id": "zone-1", "name": DOMAIN, "imported": True},
+            {"id": "zone-2", "name": "new.example", "imported": False},
+            {"id": "zone-3", "name": "skip.example", "imported": False},
+        ]
+
+    with patch("app.services.cloudflare_dns.discover_cloudflare_zones", new=fake_discover):
+        result = asyncio.run(
+            cloudflare_dns.import_cloudflare_domains(
+                db_session,
+                requested_domains=["new.example"],
+            )
+        )
+
+    assert result["imported"] == ["new.example"]
+    assert result["existing"] == []
+    assert sorted(result["skipped"]) == [DOMAIN, "skip.example"]
+    assert db_session.query(Domain).filter(Domain.name == "new.example").first() is not None
+
+
+def test_get_zone_for_domain_uses_configured_zone_id_even_if_zone_lookup_fails(db_session):
+    db_session.add_all(
+        [
+            Setting(
+                key="cloudflare.api_token",
+                value=encrypt_secret("cf-token"),
+                category="cloudflare",
+            ),
+            Setting(key="cloudflare.zone_id", value="zone-1", category="cloudflare"),
+        ]
+    )
+    db_session.commit()
+    provider = FakeCloudflareProvider(
+        records=[_record("spf", "TXT", DOMAIN, "v=spf1 ~all")],
+        fail_zone_lookup=True,
+    )
+
+    with patch("app.services.cloudflare_dns.build_cloudflare_provider", return_value=provider):
+        result = asyncio.run(cloudflare_dns.get_zone_for_domain(db_session, DOMAIN))
+
+    assert result["id"] == "zone-1"
+    assert result["name"] == DOMAIN
+    assert result["records"][0]["id"] == "spf"
+
+
+def test_get_zone_for_domain_finds_best_matching_zone(db_session):
+    db_session.add(
+        Setting(
+            key="cloudflare.api_token",
+            value=encrypt_secret("cf-token"),
+            category="cloudflare",
+        )
+    )
+    db_session.commit()
+    provider = FakeCloudflareProvider(
+        zones=[
+            {"id": "zone-root", "name": "example.com"},
+            {"id": "zone-sub", "name": "mail.example.com"},
+        ],
+        records=[_record("spf", "TXT", "mail.example.com", "v=spf1 ~all")],
+    )
+
+    with patch("app.services.cloudflare_dns.build_cloudflare_provider", return_value=provider):
+        result = asyncio.run(cloudflare_dns.get_zone_for_domain(db_session, "mail.example.com"))
+
+    assert result["id"] == "zone-root"
+    assert result["name"] == "example.com"
+
+
+def test_get_zone_for_domain_raises_when_no_zone_matches(db_session):
+    db_session.add(
+        Setting(
+            key="cloudflare.api_token",
+            value=encrypt_secret("cf-token"),
+            category="cloudflare",
+        )
+    )
+    db_session.commit()
+    provider = FakeCloudflareProvider(zones=[])
+
+    with patch("app.services.cloudflare_dns.build_cloudflare_provider", return_value=provider):
+        try:
+            asyncio.run(cloudflare_dns.get_zone_for_domain(db_session, DOMAIN))
+        except LookupError as exc:
+            assert DOMAIN in str(exc)
+        else:
+            raise AssertionError("Expected LookupError")
+
+
+def test_policy_enforcement_suggestion_requires_high_compliance():
+    suggestions = _policy_enforcement_suggestions(
+        "none",
+        {"total_count": 250, "compliance_rate": 99.2},
+    )
+
+    assert suggestions[0]["type"] == "policy_enforcement_ready"
+
+
+def test_policy_enforcement_suggestion_ignores_enforced_policy():
+    suggestions = _policy_enforcement_suggestions(
+        "reject",
+        {"total_count": 250, "compliance_rate": 99.2},
+    )
+
+    assert suggestions == []
+
+
+def test_cloudflare_discover_endpoint_returns_zones(client: TestClient):
+    with patch(
+        "app.api.api_v1.endpoints.domains.discover_cloudflare_zones",
+        new=AsyncMock(
+            return_value=[
+                {
+                    "id": "zone-1",
+                    "name": DOMAIN,
+                    "status": "active",
+                    "account_name": "Example",
+                    "imported": False,
+                }
+            ]
+        ),
+    ):
+        response = client.get("/api/v1/domains/cloudflare/discover")
+
+    assert response.status_code == 200
+    assert response.json()[0]["name"] == DOMAIN
+
+
+def test_cloudflare_import_endpoint_returns_import_summary(client: TestClient):
+    with patch(
+        "app.api.api_v1.endpoints.domains.import_cloudflare_domains",
+        new=AsyncMock(
+            return_value={
+                "imported": [DOMAIN],
+                "existing": [],
+                "skipped": [],
+                "total_discovered": 1,
+            }
+        ),
+    ):
+        response = client.post(
+            "/api/v1/domains/cloudflare/import",
+            json={"domains": [DOMAIN]},
+        )
+
+    assert response.status_code == 200
+    assert response.json()["imported"] == [DOMAIN]
+
+
+def test_cloudflare_dns_analysis_endpoint_persists_history(client: TestClient, db_session):
+    db_session.add(Domain(name=DOMAIN))
+    db_session.commit()
+    records = [
+        _record("spf", "TXT", DOMAIN, "v=spf1 include:_spf.google.com ~all"),
+        _record("dmarc", "TXT", f"_dmarc.{DOMAIN}", "v=DMARC1; p=reject"),
+    ]
+
+    with patch(
+        "app.api.api_v1.endpoints.domains.get_zone_for_domain",
+        new=AsyncMock(return_value={"id": "zone-1", "name": DOMAIN, "records": records}),
+    ):
+        response = client.get(f"/api/v1/domains/{DOMAIN}/dns/cloudflare")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["zone"]["id"] == "zone-1"
+    assert data["checks"]["dmarc_policy"] == "reject"
+    assert len(data["changes"]) == 2
+    assert len(data["history"]) == 2
+
+    history_response = client.get(f"/api/v1/domains/{DOMAIN}/dns/history")
+    assert history_response.status_code == 200
+    assert len(history_response.json()["history"]) == 2
+
+
+def test_cloudflare_dns_analysis_endpoint_returns_configuration_errors(client: TestClient):
+    with patch(
+        "app.api.api_v1.endpoints.domains.get_zone_for_domain",
+        new=AsyncMock(side_effect=LookupError("Cloudflare API token is not configured")),
+    ):
+        response = client.get(f"/api/v1/domains/{DOMAIN}/dns/cloudflare")
+
+    assert response.status_code == 400
+    assert "Cloudflare API token" in response.json()["detail"]

backend/app/tests/test_dns_resolver.py

@@ -9,6 +9,8 @@ from unittest.mock import AsyncMock, patch
 
 import pytest
 
+from app.core.credential_encryption import encrypt_secret
+from app.models.setting import Setting
 from app.services.dns_resolver import (
     BaseDNSProvider,
     CloudflareDNSProvider,
@@ -276,6 +278,78 @@ async def test_cloudflare_provider_raises_on_http_error():
             await provider.lookup_txt("_dmarc.example.com")
 
 
+@pytest.mark.asyncio
+async def test_cloudflare_provider_lists_zones_from_rest_api():
+    from unittest.mock import MagicMock
+
+    responses = [
+        {
+            "success": True,
+            "result": [{"id": "zone-1", "name": "example.com"}],
+            "result_info": {"total_pages": 2},
+        },
+        {
+            "success": True,
+            "result": [{"id": "zone-2", "name": "example.net"}],
+            "result_info": {"total_pages": 2},
+        },
+    ]
+
+    mock_response = AsyncMock()
+    mock_response.raise_for_status = MagicMock()
+    mock_response.json = lambda: responses.pop(0)
+
+    with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)) as mock_get:
+        provider = CloudflareDNSProvider(api_token="token")
+        zones = await provider.list_zones()
+
+    assert [zone["name"] for zone in zones] == ["example.com", "example.net"]
+    assert mock_get.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_cloudflare_provider_lists_dns_records_from_rest_api():
+    from unittest.mock import MagicMock
+
+    fake_response_data = {
+        "success": True,
+        "result": [{"id": "record-1", "type": "TXT", "name": "example.com"}],
+        "result_info": {"total_pages": 1},
+    }
+
+    mock_response = AsyncMock()
+    mock_response.raise_for_status = MagicMock()
+    mock_response.json = lambda: fake_response_data
+
+    with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)) as mock_get:
+        provider = CloudflareDNSProvider(api_token="token", zone_id="zone-1")
+        records = await provider.list_dns_records(record_type="TXT")
+
+    assert records == fake_response_data["result"]
+    _, kwargs = mock_get.await_args
+    assert "/zones/zone-1/dns_records" in str(mock_get.await_args.args[0])
+    assert kwargs["params"]["type"] == "TXT"
+
+
+@pytest.mark.asyncio
+async def test_cloudflare_provider_raises_when_rest_api_reports_error():
+    from unittest.mock import MagicMock
+
+    fake_response_data = {
+        "success": False,
+        "errors": [{"message": "invalid token"}],
+    }
+
+    mock_response = AsyncMock()
+    mock_response.raise_for_status = MagicMock()
+    mock_response.json = lambda: fake_response_data
+
+    with patch("httpx.AsyncClient.get", new=AsyncMock(return_value=mock_response)):
+        provider = CloudflareDNSProvider(api_token="token")
+        with pytest.raises(LookupError, match="invalid token"):
+            await provider.list_zones()
+
+
 # ---------------------------------------------------------------------------
 # get_default_provider
 # ---------------------------------------------------------------------------
@@ -286,6 +360,27 @@ def test_get_default_provider_returns_system():
     assert isinstance(provider, SystemDNSProvider)
 
 
+def test_get_default_provider_uses_cloudflare_settings(db_session):
+    db_session.add_all(
+        [
+            Setting(key="dns.resolver", value="cloudflare", category="dns"),
+            Setting(
+                key="cloudflare.api_token",
+                value=encrypt_secret("cf-token"),
+                category="cloudflare",
+            ),
+            Setting(key="cloudflare.zone_id", value="zone-1", category="cloudflare"),
+        ]
+    )
+    db_session.commit()
+
+    provider = get_default_provider(db_session)
+
+    assert isinstance(provider, CloudflareDNSProvider)
+    assert provider.api_token == "cf-token"
+    assert provider.zone_id == "zone-1"
+
+
 # ---------------------------------------------------------------------------
 # _ip_to_arpa_name helper
 # ---------------------------------------------------------------------------

docs/deployment/configuration.md

@@ -110,11 +110,21 @@ operational policy if long-term storage size matters.
 
 | Variable | Description | Default | Example |
 |----------|-------------|---------|---------|
-| `CF_ENABLED` | Enable Cloudflare integration | `false` | `true`, `false` |
-| `CF_API_TOKEN` | Cloudflare API token | - | `your_cloudflare_api_token` |
-| `CF_ZONE_ID` | Cloudflare Zone ID | - | `your_cloudflare_zone_id` |
+| `CLOUDFLARE_API_TOKEN` | Cloudflare API token for read-only zone discovery and DNS inspection | - | `your_cloudflare_api_token` |
+| `CLOUDFLARE_ZONE_ID` | Optional default Cloudflare Zone ID | - | `your_cloudflare_zone_id` |
 | `WEBHOOK_SECRET` | Required secret for inbound email worker webhooks | - | `openssl rand -hex 32` |
 
+Cloudflare credentials can also be stored from **Settings**. The API token is
+encrypted in the settings table and redacted when settings are read back. Leave
+the Zone ID blank to discover every active zone visible to the token.
+
+The read-only integration exposes:
+
+- `GET /api/v1/domains/cloudflare/discover` to list available zones.
+- `POST /api/v1/domains/cloudflare/import` to create monitored domain rows from zones.
+- `GET /api/v1/domains/{domain}/dns/cloudflare` to inspect managed DNS records, return DMARC/SPF/DKIM suggestions, and record detected DNS changes.
+- `GET /api/v1/domains/{domain}/dns/history` to review DNS record additions, modifications, and removals.
+
 ### DNS Result Cache
 
 DMARC, SPF, and DKIM DNS checks are cached in the database-backed `dns_cache`
@@ -123,6 +133,10 @@ API responses include whether the result came from cache and when it was
 checked. Use `?refresh=true` on the domain DNS endpoint to bypass a fresh cache
 entry for operational rechecks.
 
+Cloudflare-managed DNS record snapshots and change events are stored in
+`dns_record_snapshots` and `dns_record_changes`. They are updated whenever the
+Cloudflare DNS analysis endpoint is called.
+
 ### Advanced Configuration
 
 | Variable | Description | Default | Example |

docs/development/roadmap.md

@@ -84,7 +84,7 @@ Follow-up:
 ## Later Milestones
 
 - Notifications and alert rules. Apprise delivery, test notifications, alert-rule evaluation, scheduled daily/weekly summaries, and alert history are in place.
-- DNS health and Cloudflare read-only inspection.
+- DNS health and Cloudflare read-only inspection are in place, including zone import, record recommendations, and DNS change tracking.
 - Guided setup and operator health screens.
 - Forensic/RUF report support.
 

docs/milestones.md

@@ -125,18 +125,19 @@ Exit criteria:
 
 ## Milestone 8: DNS Health and Guidance
 
-Status: Planned
+Status: Complete
 
 Goal: connect report findings with DNS configuration guidance.
 
 Delivered:
 - DMARC/SPF/DKIM DNS checks with database-backed cached results.
 - DKIM selector discovery from report data.
-
-Planned:
 - Per-domain DNS health summary.
+- Cloudflare read-only integration for automatic domain discovery and DNS record inspection.
+- Import Cloudflare zones as monitored domains from Settings.
+- Suggestions for missing, duplicate, or malformed DMARC/SPF/DKIM records.
+- DNS record snapshots and change history for Cloudflare-managed records, including additions, modifications, and removals.
 - Suggestions for moving from `p=none` to enforcement when compliance supports it.
-- Optional Cloudflare read-only integration for DNS record inspection.
 
 Exit criteria:
 - A user can see whether DNS records match the actual senders observed in DMARC reports.

docs/user_guide/settings.md

@@ -103,11 +103,17 @@ DMARQ provides an API for integration with other systems:
 
 If you use Cloudflare for DNS management:
 
-1. Navigate to **Settings** > **Integrations** > **Cloudflare**
+1. Navigate to **Settings** > **Cloudflare Integration**
 2. Configure:
    - **API Token**: Your Cloudflare API token
-   - **Zone ID**: The Cloudflare Zone ID for your domain
-   - **Permissions**: What actions DMARQ can take on your DNS records
+   - **Zone ID**: Optional Cloudflare Zone ID for a single domain
+3. Use **Discover** to list zones visible to the token.
+4. Use **Import New** to create monitored domain rows for discovered zones that are not already tracked.
+
+The token only needs read access for zone and DNS record inspection. DMARQ uses
+it to fetch managed DNS records, detect missing or malformed DMARC/SPF/DKIM
+entries, and record DNS additions, modifications, or removals over time. DMARQ
+does not automatically change Cloudflare DNS records.
 
 ### Other Integrations