From 521dbc92d0e71a3403fb88a59b9283ab579ac6c6 Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Sat, 23 May 2026 01:13:06 +0200 Subject: [PATCH] feat: add Cloudflare DNS integration Closes #31 --- README.md | 12 +- ...e2f3a4b5_add_dns_record_change_tracking.py | 172 +++++++ backend/app/api/api_v1/endpoints/domains.py | 168 ++++++- backend/app/models/dns_cache.py | 65 ++- backend/app/services/cloudflare_dns.py | 427 ++++++++++++++++++ backend/app/services/dns_resolver.py | 175 ++++++- backend/app/templates/settings.html | 107 ++++- backend/app/tests/test_cloudflare_dns.py | 426 +++++++++++++++++ backend/app/tests/test_dns_resolver.py | 95 ++++ docs/deployment/configuration.md | 20 +- docs/development/roadmap.md | 2 +- docs/milestones.md | 9 +- docs/user_guide/settings.md | 12 +- 13 files changed, 1644 insertions(+), 46 deletions(-) create mode 100644 backend/alembic/versions/c0d1e2f3a4b5_add_dns_record_change_tracking.py create mode 100644 backend/app/services/cloudflare_dns.py create mode 100644 backend/app/tests/test_cloudflare_dns.py diff --git a/README.md b/README.md index f17e8ff..3c8203b 100644 --- a/README.md +++ b/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) diff --git a/backend/alembic/versions/c0d1e2f3a4b5_add_dns_record_change_tracking.py b/backend/alembic/versions/c0d1e2f3a4b5_add_dns_record_change_tracking.py new file mode 100644 index 0000000..b5af4cd --- /dev/null +++ b/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") diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py index cfec2ca..0607251 100644 --- a/backend/app/api/api_v1/endpoints/domains.py +++ b/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]) diff --git a/backend/app/models/dns_cache.py b/backend/app/models/dns_cache.py index f2580d0..fcb992b 100644 --- a/backend/app/models/dns_cache.py +++ b/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"" + + +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"" + + +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"" diff --git a/backend/app/services/cloudflare_dns.py b/backend/app/services/cloudflare_dns.py new file mode 100644 index 0000000..75833fa --- /dev/null +++ b/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, + } diff --git a/backend/app/services/dns_resolver.py b/backend/app/services/dns_resolver.py index b379d08..246d34c 100644 --- a/backend/app/services/dns_resolver.py +++ b/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() diff --git a/backend/app/templates/settings.html b/backend/app/templates/settings.html index f26bb34..654d7ce 100644 --- a/backend/app/templates/settings.html +++ b/backend/app/templates/settings.html @@ -189,7 +189,65 @@ - + + +
+
+
+

Domain Discovery

+

Find Cloudflare zones and import them into DMARQ.

+
+
+ + +
+
+