from __future__ import annotations import re import secrets import socket import ssl import subprocess from dataclasses import dataclass from datetime import datetime, timezone from email.utils import parsedate_to_datetime from sqlalchemy import Select, func, select from sqlalchemy.orm import Session from app.modules.consultants.models import ConsultantProfile from app.modules.core.tenancy.models import Branch, Tenant from app.modules.domain_management.models import DomainMapping DOMAIN_TYPES: tuple[tuple[str, str], ...] = ( ("marketplace", "Marketplace / SaaS platform"), ("audit_firm_domain", "Audit firm custom domain"), ("audit_firm_subdomain", "Audit firm platform subdomain"), ("consultant_custom_domain", "Consultant custom domain"), ("consultant_firm_domain", "Consultant under firm domain"), ("consultant_marketplace_subdomain", "Consultant platform subdomain"), ) PLATFORM_SUBDOMAIN_BASE_DOMAINS: tuple[str, ...] = ("filingabc.com", "filingabc.local") RESERVED_SUBDOMAIN_LABELS: set[str] = {"www", "mail", "smtp", "imap", "pop", "api", "admin", "app", "portal", "client", "clients", "staff", "system", "marketplace", "support", "billing"} DOMAIN_STATUSES: tuple[tuple[str, str], ...] = ( ("draft", "Draft"), ("pending_verification", "Pending verification"), ("active", "Active"), ("suspended", "Suspended"), ("inactive", "Inactive"), ) SSL_MODES: tuple[tuple[str, str], ...] = ( ("manual", "Manual / deployment managed"), ("coolify", "Coolify proxy"), ("caddy", "Caddy"), ("traefik", "Traefik"), ("cloudflare", "Cloudflare"), ) SSL_STATUSES: tuple[tuple[str, str], ...] = ( ("not_checked", "Not checked"), ("pending_dns", "Pending DNS/proxy"), ("pending_ssl", "Pending SSL"), ("active", "Active"), ("failed", "Failed"), ("manual", "Manual / external"), ) _HOST_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+$", re.IGNORECASE) @dataclass(frozen=True) class DomainSslCheckResult: ok: bool status: str message: str subject: str | None = None issuer: str | None = None not_after_utc: datetime | None = None @dataclass(frozen=True) class DomainSslProxySnippet: provider: str title: str body: str @dataclass(frozen=True) class DomainMappingPayload: domain_name: str domain_type: str tenant_id: int | None = None branch_id: int | None = None consultant_id: int | None = None parent_tenant_id: int | None = None is_primary: bool = False is_verified: bool = False status: str = "draft" ssl_mode: str = "manual" notes: str | None = None def normalize_domain(domain_name: str) -> str: value = (domain_name or "").strip().lower() value = value.removeprefix("http://").removeprefix("https://") value = value.split("/", 1)[0] value = value.split(":", 1)[0] return value.rstrip(".") def validate_domain_payload(db: Session, payload: DomainMappingPayload, mapping_id: int | None = None) -> list[str]: errors: list[str] = [] domain_name = normalize_domain(payload.domain_name) if not domain_name: errors.append("Domain name is required.") elif not _HOST_RE.match(domain_name) and domain_name not in {"localhost"}: errors.append("Enter a valid domain name, for example filingabc.com or arrr.filingabc.com.") if payload.domain_type not in {code for code, _label in DOMAIN_TYPES}: errors.append("Invalid domain type selected.") if payload.status not in {code for code, _label in DOMAIN_STATUSES}: errors.append("Invalid status selected.") if payload.ssl_mode not in {code for code, _label in SSL_MODES}: errors.append("Invalid SSL mode selected.") existing_stmt = select(DomainMapping).where(DomainMapping.domain_name == domain_name) if mapping_id: existing_stmt = existing_stmt.where(DomainMapping.id != mapping_id) if db.execute(existing_stmt).scalar_one_or_none(): errors.append("This domain is already mapped.") if payload.tenant_id and not db.get(Tenant, payload.tenant_id): errors.append("Selected audit firm was not found.") if payload.parent_tenant_id and not db.get(Tenant, payload.parent_tenant_id): errors.append("Selected parent audit firm was not found.") if payload.branch_id and not db.get(Branch, payload.branch_id): errors.append("Selected branch was not found.") if payload.consultant_id and not db.get(ConsultantProfile, payload.consultant_id): errors.append("Selected consultant was not found.") if payload.domain_type in {"audit_firm_domain", "audit_firm_subdomain"} and not payload.tenant_id: errors.append("Audit firm domains must be linked to an audit firm.") if payload.domain_type in {"consultant_custom_domain", "consultant_firm_domain", "consultant_marketplace_subdomain"} and not payload.consultant_id: errors.append("Consultant domains must be linked to a consultant profile.") if payload.domain_type == "consultant_firm_domain" and not payload.parent_tenant_id: errors.append("Consultant firm-domain mappings must have a parent audit firm.") if payload.domain_type == "audit_firm_subdomain" and domain_name: parts = domain_name.split(".") if len(parts) < 3: errors.append("Audit firm subdomain must be like auditfirm.filingabc.com.") subdomain_label = parts[0] if parts else "" if subdomain_label in RESERVED_SUBDOMAIN_LABELS: errors.append(f"'{subdomain_label}' is a reserved subdomain label. Please use another audit firm code.") if subdomain_label and not re.match(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$", subdomain_label): errors.append("Subdomain label may contain only lowercase letters, numbers and hyphen, and cannot start/end with hyphen.") return errors def make_verification_token() -> str: return "af-domain-" + secrets.token_urlsafe(24).replace("-", "").replace("_", "")[:36] def dns_txt_name_for(domain_name: str) -> str: return f"_audit-firm-verify.{normalize_domain(domain_name)}" def list_domain_mappings(db: Session, *, q: str = "", status: str = "", domain_type: str = "") -> list[DomainMapping]: stmt: Select[tuple[DomainMapping]] = select(DomainMapping).order_by(DomainMapping.created_at_utc.desc(), DomainMapping.id.desc()) q = (q or "").strip().lower() if q: like = f"%{q}%" stmt = stmt.where(func.lower(DomainMapping.domain_name).like(like)) if status: stmt = stmt.where(DomainMapping.status == status) if domain_type: stmt = stmt.where(DomainMapping.domain_type == domain_type) return list(db.execute(stmt).scalars().all()) def create_domain_mapping(db: Session, payload: DomainMappingPayload, *, user_id: int | None) -> DomainMapping: domain_name = normalize_domain(payload.domain_name) token = make_verification_token() mapping = DomainMapping( domain_name=domain_name, domain_type=payload.domain_type, tenant_id=payload.tenant_id, branch_id=payload.branch_id, consultant_id=payload.consultant_id, parent_tenant_id=payload.parent_tenant_id, is_primary=payload.is_primary, is_verified=payload.is_verified, status="active" if payload.is_verified and payload.status == "active" else payload.status, verification_token=token, dns_txt_name=dns_txt_name_for(domain_name), dns_txt_value=token, ssl_mode=payload.ssl_mode or "manual", notes=(payload.notes or "").strip() or None, created_by_user_id=user_id, updated_by_user_id=user_id, ) db.add(mapping) db.commit() db.refresh(mapping) return mapping def update_domain_mapping(db: Session, mapping: DomainMapping, payload: DomainMappingPayload, *, user_id: int | None) -> DomainMapping: old_domain = mapping.domain_name mapping.domain_name = normalize_domain(payload.domain_name) mapping.domain_type = payload.domain_type mapping.tenant_id = payload.tenant_id mapping.branch_id = payload.branch_id mapping.consultant_id = payload.consultant_id mapping.parent_tenant_id = payload.parent_tenant_id mapping.is_primary = payload.is_primary mapping.is_verified = payload.is_verified mapping.status = payload.status mapping.ssl_mode = payload.ssl_mode or "manual" mapping.notes = (payload.notes or "").strip() or None mapping.updated_by_user_id = user_id mapping.updated_at_utc = datetime.now(timezone.utc) if old_domain != mapping.domain_name or not mapping.verification_token: mapping.verification_token = make_verification_token() mapping.dns_txt_name = dns_txt_name_for(mapping.domain_name) mapping.dns_txt_value = mapping.verification_token mapping.is_verified = False if mapping.status == "active": mapping.status = "pending_verification" db.commit() db.refresh(mapping) return mapping def regenerate_verification_token(db: Session, mapping: DomainMapping, *, user_id: int | None) -> DomainMapping: mapping.verification_token = make_verification_token() mapping.dns_txt_name = dns_txt_name_for(mapping.domain_name) mapping.dns_txt_value = mapping.verification_token mapping.is_verified = False if mapping.status == "active": mapping.status = "pending_verification" mapping.updated_by_user_id = user_id mapping.updated_at_utc = datetime.now(timezone.utc) db.commit() db.refresh(mapping) return mapping def mark_verified(db: Session, mapping: DomainMapping, *, user_id: int | None) -> DomainMapping: mapping.is_verified = True mapping.status = "active" mapping.updated_by_user_id = user_id mapping.updated_at_utc = datetime.now(timezone.utc) db.commit() db.refresh(mapping) return mapping def slugify_subdomain_label(value: str) -> str: """Create a safe subdomain label from a tenant code/name.""" raw = (value or "").strip().lower() raw = re.sub(r"[^a-z0-9-]+", "-", raw) raw = re.sub(r"-+", "-", raw).strip("-") if not raw: raw = "firm" if raw in RESERVED_SUBDOMAIN_LABELS: raw = f"{raw}-firm" return raw[:63].strip("-") or "firm" def tenant_subdomain_label(tenant: Tenant) -> str: code = getattr(tenant, "code", None) or getattr(tenant, "name", None) or f"firm-{tenant.id}" return slugify_subdomain_label(str(code)) def build_tenant_subdomain(tenant: Tenant, base_domain: str = "filingabc.com") -> str: base = normalize_domain(base_domain or "filingabc.com") return f"{tenant_subdomain_label(tenant)}.{base}" def list_tenant_subdomain_candidates(db: Session, *, base_domain: str = "filingabc.com") -> list[dict]: """Return all tenants with their suggested platform subdomain and existing mapping, if any.""" base = normalize_domain(base_domain or "filingabc.com") tenants = db.execute(select(Tenant).order_by(Tenant.name.asc())).scalars().all() out: list[dict] = [] for tenant in tenants: domain_name = build_tenant_subdomain(tenant, base) mapping = db.execute(select(DomainMapping).where(DomainMapping.domain_name == domain_name)).scalar_one_or_none() out.append({"tenant": tenant, "domain_name": domain_name, "mapping": mapping}) return out def create_or_get_tenant_subdomain_mapping( db: Session, *, tenant_id: int, base_domain: str = "filingabc.com", branch_id: int | None = None, mark_verified_active: bool = True, user_id: int | None = None, ) -> tuple[DomainMapping, bool]: """Create auditfirm.filingabc.com mapping for a tenant without duplicating existing rows. Returns (mapping, created). The platform owner normally verifies wildcard DNS/SSL once, so mark_verified_active=True is safe for the owned *.filingabc.com style domain. """ tenant = db.get(Tenant, int(tenant_id)) if not tenant: raise ValueError("Selected audit firm was not found.") domain_name = build_tenant_subdomain(tenant, base_domain) existing = db.execute(select(DomainMapping).where(DomainMapping.domain_name == domain_name)).scalar_one_or_none() if existing: return existing, False payload = DomainMappingPayload( domain_name=domain_name, domain_type="audit_firm_subdomain", tenant_id=tenant.id, branch_id=branch_id, is_primary=False, is_verified=bool(mark_verified_active), status="active" if mark_verified_active else "pending_verification", ssl_mode="coolify", notes="Auto-created platform tenant subdomain. Ensure wildcard DNS/SSL for *.filingabc.com is configured in deployment.", ) errors = validate_domain_payload(db, payload) if errors: raise ValueError(" ".join(errors)) mapping = create_domain_mapping(db, payload, user_id=user_id) return mapping, True def create_or_get_firm_custom_domain_mapping( db: Session, *, tenant_id: int, domain_name: str = "arrr.associates", branch_id: int | None = None, mark_verified_active: bool = False, user_id: int | None = None, ) -> tuple[DomainMapping, bool]: """Create an audit-firm custom domain mapping such as arrr.associates. This is intentionally separate from tenant subdomains. Custom domains normally need DNS verification before activation, so mark_verified_active defaults to False. For a domain owned by the platform owner and already configured in DNS, an admin may tick mark_verified_active to make it active immediately. Returns (mapping, created). """ tenant = db.get(Tenant, int(tenant_id)) if not tenant: raise ValueError("Selected audit firm was not found.") normalized_domain = normalize_domain(domain_name or "arrr.associates") existing = db.execute(select(DomainMapping).where(DomainMapping.domain_name == normalized_domain)).scalar_one_or_none() if existing: return existing, False payload = DomainMappingPayload( domain_name=normalized_domain, domain_type="audit_firm_domain", tenant_id=tenant.id, branch_id=branch_id, is_primary=True, is_verified=bool(mark_verified_active), status="active" if mark_verified_active else "pending_verification", ssl_mode="coolify", notes=( "Custom audit firm domain created from Phase 7T.6. " "Ensure DNS points to the ERP server and SSL/proxy configuration is completed before live use." ), ) errors = validate_domain_payload(db, payload) if errors: raise ValueError(" ".join(errors)) mapping = create_domain_mapping(db, payload, user_id=user_id) return mapping, True def consultant_subdomain_label(consultant: ConsultantProfile) -> str: """Create a safe domain label for consultant profile domains.""" raw = (getattr(consultant, "firm_name", None) or getattr(consultant, "contact_person", None) or f"consultant-{consultant.id}") return slugify_subdomain_label(str(raw)) def build_consultant_platform_subdomain(consultant: ConsultantProfile, base_domain: str = "filingabc.com") -> str: base = normalize_domain(base_domain or "filingabc.com") return f"{consultant_subdomain_label(consultant)}.{base}" def build_consultant_firm_domain(consultant: ConsultantProfile, firm_base_domain: str = "arrr.accountant") -> str: base = normalize_domain(firm_base_domain or "arrr.accountant") return f"{consultant_subdomain_label(consultant)}.{base}" def list_consultant_domain_candidates( db: Session, *, base_domain: str = "filingabc.com", firm_base_domain: str = "arrr.accountant", ) -> list[dict]: """Return consultants with suggested marketplace and firm-linked profile domains.""" consultants = db.execute(select(ConsultantProfile).order_by(ConsultantProfile.contact_person.asc())).scalars().all() out: list[dict] = [] for consultant in consultants: platform_domain = build_consultant_platform_subdomain(consultant, base_domain) firm_domain = build_consultant_firm_domain(consultant, firm_base_domain) platform_mapping = db.execute(select(DomainMapping).where(DomainMapping.domain_name == platform_domain)).scalar_one_or_none() firm_mapping = db.execute(select(DomainMapping).where(DomainMapping.domain_name == firm_domain)).scalar_one_or_none() out.append({ "consultant": consultant, "platform_domain": platform_domain, "platform_mapping": platform_mapping, "firm_domain": firm_domain, "firm_mapping": firm_mapping, }) return out def create_or_get_consultant_domain_mapping( db: Session, *, consultant_id: int, domain_type: str = "consultant_marketplace_subdomain", domain_name: str | None = None, base_domain: str = "filingabc.com", firm_base_domain: str = "arrr.accountant", parent_tenant_id: int | None = None, mark_verified_active: bool = False, user_id: int | None = None, ) -> tuple[DomainMapping, bool]: """Create consultant profile domain mapping without duplicating existing rows. Supported examples: consultant_marketplace_subdomain -> consultant.filingabc.com consultant_firm_domain -> consultant.arrr.accountant consultant_custom_domain -> abc.accountants """ consultant = db.get(ConsultantProfile, int(consultant_id)) if not consultant: raise ValueError("Selected consultant profile was not found.") if domain_type == "consultant_marketplace_subdomain": normalized_domain = normalize_domain(domain_name or build_consultant_platform_subdomain(consultant, base_domain)) resolved_parent_tenant_id = parent_tenant_id ssl_mode = "coolify" default_notes = "Auto-created consultant marketplace profile subdomain. Ensure wildcard DNS/SSL for the platform base domain is configured." default_verified = True elif domain_type == "consultant_firm_domain": normalized_domain = normalize_domain(domain_name or build_consultant_firm_domain(consultant, firm_base_domain)) resolved_parent_tenant_id = parent_tenant_id or consultant.tenant_id ssl_mode = "coolify" default_notes = "Consultant domain under an audit firm brand. Ensure DNS/SSL for the firm consultant domain is configured." default_verified = False elif domain_type == "consultant_custom_domain": if not domain_name: raise ValueError("Custom consultant domain is required.") normalized_domain = normalize_domain(domain_name) resolved_parent_tenant_id = parent_tenant_id ssl_mode = "coolify" default_notes = "Consultant custom domain. Verify DNS ownership before activation." default_verified = False else: raise ValueError("Invalid consultant domain type selected.") existing = db.execute(select(DomainMapping).where(DomainMapping.domain_name == normalized_domain)).scalar_one_or_none() if existing: return existing, False verified = bool(mark_verified_active or default_verified) payload = DomainMappingPayload( domain_name=normalized_domain, domain_type=domain_type, tenant_id=consultant.tenant_id, branch_id=consultant.branch_id, consultant_id=consultant.id, parent_tenant_id=resolved_parent_tenant_id, is_primary=False, is_verified=verified, status="active" if verified else "pending_verification", ssl_mode=ssl_mode, notes=default_notes, ) errors = validate_domain_payload(db, payload) if errors: raise ValueError(" ".join(errors)) mapping = create_domain_mapping(db, payload, user_id=user_id) return mapping, True def reference_data(db: Session) -> dict: return { "tenants": db.execute(select(Tenant).order_by(Tenant.name.asc())).scalars().all(), "branches": db.execute(select(Branch).order_by(Branch.name.asc())).scalars().all(), "consultants": db.execute(select(ConsultantProfile).order_by(ConsultantProfile.contact_person.asc())).scalars().all(), "domain_types": DOMAIN_TYPES, "statuses": DOMAIN_STATUSES, "ssl_modes": SSL_MODES, "ssl_statuses": SSL_STATUSES, } @dataclass(frozen=True) class DomainResolution: """Resolved runtime domain context used by Phase 7T.2 middleware.""" is_resolved: bool host: str mapping_id: int | None = None domain_name: str | None = None domain_type: str | None = None tenant_id: int | None = None tenant_code: str | None = None branch_id: int | None = None branch_code: str | None = None consultant_id: int | None = None parent_tenant_id: int | None = None is_verified: bool = False status: str | None = None def normalize_request_host(host_header: str | None) -> str: """Normalise the HTTP Host / X-Forwarded-Host value for lookup. Handles values such as: localhost:8000 arrr.associates arrr.associates:443 arrr.associates, proxy-host """ value = (host_header or "").strip().lower() if not value: return "" value = value.split(",", 1)[0].strip() value = value.removeprefix("http://").removeprefix("https://") value = value.split("/", 1)[0] if value.startswith("[") and "]" in value: # IPv6 literal; keep without brackets/port for safety. value = value.split("]", 1)[0].lstrip("[") else: value = value.split(":", 1)[0] return value.rstrip(".") def resolve_domain_context(db: Session, host: str) -> DomainResolution: """Resolve an incoming host to an active domain mapping. Phase 7T.2 intentionally resolves only exact domain mappings. Wildcard/platform subdomain inference comes later in Phase 7T.5, and custom-domain DNS checks come in Phase 7T.8. """ normalized = normalize_domain(host) if not normalized: return DomainResolution(is_resolved=False, host="") stmt = ( select(DomainMapping, Tenant.code, Branch.code) .outerjoin(Tenant, Tenant.id == DomainMapping.tenant_id) .outerjoin(Branch, Branch.id == DomainMapping.branch_id) .where(DomainMapping.domain_name == normalized) .where(DomainMapping.status == "active") .where(DomainMapping.is_verified.is_(True)) .limit(1) ) row = db.execute(stmt).first() if not row: return DomainResolution(is_resolved=False, host=normalized) mapping, tenant_code, branch_code = row return DomainResolution( is_resolved=True, host=normalized, mapping_id=int(mapping.id), domain_name=mapping.domain_name, domain_type=mapping.domain_type, tenant_id=mapping.tenant_id, tenant_code=tenant_code, branch_id=mapping.branch_id, branch_code=branch_code, consultant_id=mapping.consultant_id, parent_tenant_id=mapping.parent_tenant_id, is_verified=bool(mapping.is_verified), status=mapping.status, ) @dataclass(frozen=True) class DomainDnsVerificationResult: ok: bool message: str txt_name: str expected_value: str found_values: tuple[str, ...] = () def _normalise_txt_value(value: str) -> str: value = (value or "").strip() # TXT records may be returned with quotes and split chunks. if value.startswith('"') and value.endswith('"') and len(value) >= 2: value = value[1:-1] value = value.replace('" "', '').replace('"', '').strip() return value def _resolve_txt_with_dnspython(txt_name: str) -> tuple[list[str], str | None]: try: import dns.resolver # type: ignore except Exception: return [], "dnspython is not installed" try: answers = dns.resolver.resolve(txt_name, "TXT") values: list[str] = [] for answer in answers: try: chunks = [part.decode("utf-8", errors="ignore") if isinstance(part, bytes) else str(part) for part in answer.strings] values.append("".join(chunks)) except Exception: values.append(str(answer).strip()) return values, None except Exception as exc: return [], str(exc) def _resolve_txt_with_nslookup(txt_name: str) -> tuple[list[str], str | None]: try: completed = subprocess.run( ["nslookup", "-type=TXT", txt_name], capture_output=True, text=True, timeout=12, check=False, ) except FileNotFoundError: return [], "nslookup command is not available on this machine" except Exception as exc: return [], str(exc) output = "\n".join([completed.stdout or "", completed.stderr or ""]) if completed.returncode != 0 and not output.strip(): return [], "DNS lookup failed" values: list[str] = [] for line in output.splitlines(): line = line.strip() if not line: continue # Windows/Linux nslookup generally prints TXT values inside quotes. quoted = re.findall(r'"([^"]+)"', line) if quoted: values.append("".join(quoted)) continue if "text =" in line.lower(): values.append(line.split("=", 1)[1].strip()) if not values and output.strip(): # Keep a short diagnostic without storing full command noise. return [], output.strip().splitlines()[-1][:240] return values, None def lookup_dns_txt_values(txt_name: str) -> tuple[list[str], str | None]: """Return TXT records for a DNS name using dnspython when available, else nslookup. No new dependency is required. On Windows, nslookup is normally available by default. """ txt_name = normalize_domain(txt_name) if not txt_name: return [], "TXT name is empty" values, error = _resolve_txt_with_dnspython(txt_name) if values: return [_normalise_txt_value(v) for v in values], None ns_values, ns_error = _resolve_txt_with_nslookup(txt_name) if ns_values: return [_normalise_txt_value(v) for v in ns_values], None return [], ns_error or error or "No TXT record found" def verify_domain_dns_txt(db: Session, mapping: DomainMapping, *, user_id: int | None) -> DomainDnsVerificationResult: """Verify a custom/platform domain using its DNS TXT token. If the expected TXT value is found, the domain is marked verified and active. If not, the mapping remains unverified and is moved to pending_verification unless suspended. """ txt_name = mapping.dns_txt_name or dns_txt_name_for(mapping.domain_name) expected = mapping.dns_txt_value or mapping.verification_token or "" if not expected: mapping.verification_token = make_verification_token() mapping.dns_txt_name = txt_name mapping.dns_txt_value = mapping.verification_token expected = mapping.dns_txt_value db.commit() db.refresh(mapping) found_values, error = lookup_dns_txt_values(txt_name) found_normalised = tuple(_normalise_txt_value(v) for v in found_values if _normalise_txt_value(v)) expected_normalised = _normalise_txt_value(expected) matched = any(v == expected_normalised or expected_normalised in v for v in found_normalised) now = datetime.now(timezone.utc) if matched: mapping.is_verified = True mapping.status = "active" mapping.updated_by_user_id = user_id mapping.updated_at_utc = now db.commit() db.refresh(mapping) return DomainDnsVerificationResult( ok=True, message="DNS TXT verification successful. Domain is now active.", txt_name=txt_name, expected_value=expected_normalised, found_values=found_normalised, ) if mapping.status not in {"suspended", "inactive"}: mapping.status = "pending_verification" mapping.is_verified = False mapping.updated_by_user_id = user_id mapping.updated_at_utc = now db.commit() db.refresh(mapping) msg = "DNS TXT record not found or value does not match." if error: msg = f"{msg} DNS response: {error}" return DomainDnsVerificationResult( ok=False, message=msg, txt_name=txt_name, expected_value=expected_normalised, found_values=found_normalised, ) def list_domains_requiring_verification(db: Session) -> list[DomainMapping]: stmt = ( select(DomainMapping) .where(DomainMapping.is_verified.is_(False)) .where(DomainMapping.status.in_(["draft", "pending_verification"])) .order_by(DomainMapping.updated_at_utc.desc(), DomainMapping.id.desc()) ) return list(db.execute(stmt).scalars().all()) def verify_all_pending_domains(db: Session, *, user_id: int | None, limit: int = 25) -> list[tuple[DomainMapping, DomainDnsVerificationResult]]: results: list[tuple[DomainMapping, DomainDnsVerificationResult]] = [] for mapping in list_domains_requiring_verification(db)[: max(1, min(limit, 100))]: result = verify_domain_dns_txt(db, mapping, user_id=user_id) results.append((mapping, result)) return results def _name_tuple_to_text(name_tuple) -> str | None: try: parts: list[str] = [] for section in name_tuple or []: for key, value in section: if value: parts.append(f"{key}={value}") return ", ".join(parts) if parts else None except Exception: return None def _parse_ssl_not_after(value: str | None) -> datetime | None: if not value: return None try: dt = parsedate_to_datetime(value) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) except Exception: return None def check_domain_ssl_certificate(db: Session, mapping: DomainMapping, *, user_id: int | None = None, timeout: float = 5.0) -> DomainSslCheckResult: """Check whether the domain currently serves a valid TLS certificate on 443. This does not issue certificates. Certificate issuance is handled by Coolify/Caddy/ Traefik/Cloudflare/Certbot. This function records readiness/status inside ERP. """ domain = normalize_domain(mapping.domain_name) now = datetime.now(timezone.utc) if not domain or domain == "localhost": mapping.ssl_status = "manual" mapping.ssl_last_checked_at_utc = now mapping.ssl_last_error = "Localhost does not require public SSL certificate checking." db.commit() return DomainSslCheckResult(False, "manual", mapping.ssl_last_error) if not mapping.is_verified or mapping.status != "active": mapping.ssl_status = "pending_dns" mapping.ssl_last_checked_at_utc = now mapping.ssl_last_error = "Domain must be verified and active before SSL check." db.commit() return DomainSslCheckResult(False, "pending_dns", mapping.ssl_last_error) try: context = ssl.create_default_context() with socket.create_connection((domain, 443), timeout=timeout) as sock: with context.wrap_socket(sock, server_hostname=domain) as ssock: cert = ssock.getpeercert() subject = _name_tuple_to_text(cert.get("subject")) issuer = _name_tuple_to_text(cert.get("issuer")) not_after = _parse_ssl_not_after(cert.get("notAfter")) mapping.ssl_status = "active" mapping.ssl_last_checked_at_utc = now mapping.ssl_not_after_utc = not_after mapping.ssl_subject = subject mapping.ssl_issuer = issuer mapping.ssl_last_error = None db.commit() db.refresh(mapping) return DomainSslCheckResult(True, "active", "SSL certificate is active and trusted.", subject, issuer, not_after) except Exception as exc: mapping.ssl_status = "failed" mapping.ssl_last_checked_at_utc = now mapping.ssl_last_error = str(exc)[:1000] db.commit() db.refresh(mapping) return DomainSslCheckResult(False, "failed", f"SSL check failed: {mapping.ssl_last_error}") def mark_ssl_managed(db: Session, mapping: DomainMapping, *, provider: str, user_id: int | None = None) -> DomainMapping: provider = (provider or mapping.ssl_mode or "manual").strip().lower() valid = {code for code, _label in SSL_MODES} if provider not in valid: provider = "manual" mapping.ssl_provider = provider mapping.ssl_mode = provider mapping.ssl_status = "manual" if provider in {"manual", "cloudflare"} else "pending_ssl" mapping.updated_by_user_id = user_id mapping.updated_at_utc = datetime.now(timezone.utc) db.commit() db.refresh(mapping) return mapping def list_ssl_domains(db: Session) -> list[DomainMapping]: stmt = ( select(DomainMapping) .where(DomainMapping.status.in_(["active", "pending_verification", "draft"])) .order_by(DomainMapping.domain_type.asc(), DomainMapping.domain_name.asc()) ) return list(db.execute(stmt).scalars().all()) def build_ssl_proxy_snippets(mapping: DomainMapping, *, app_upstream: str = "http://127.0.0.1:8000") -> list[DomainSslProxySnippet]: domain = normalize_domain(mapping.domain_name) if not domain or domain == "localhost": return [] caddy = f"""{domain} {{ encode gzip reverse_proxy {app_upstream} }}""" nginx = f"""server {{ listen 80; server_name {domain}; location / {{ proxy_pass {app_upstream}; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; }} }} # After DNS points to this server: # certbot --nginx -d {domain}""" coolify = f"""Coolify setup checklist for {domain} 1. Open the Audit ERP application in Coolify. 2. Add domain: https://{domain} 3. Ensure DNS A/CNAME points to the Coolify server. 4. Enable Force HTTPS after certificate is issued. 5. Keep Host header forwarding enabled. 6. In ERP, verify DNS TXT and then run SSL check.""" return [ DomainSslProxySnippet("coolify", "Coolify domain setup", coolify), DomainSslProxySnippet("caddy", "Caddy automatic HTTPS snippet", caddy), DomainSslProxySnippet("nginx", "Nginx + Certbot snippet", nginx), ]