import os import re import subprocess from pathlib import Path from tempfile import NamedTemporaryFile from typing import Iterable from fastapi import HTTPException from passlib.context import CryptContext from .settings import get_settings # Use argon2id for new hashes; keep bcrypt/pbkdf2 for legacy verification. pwd_context = CryptContext( bcrypt__truncate_error=True, schemes=["argon2", "bcrypt", "pbkdf2_sha512"], deprecated="auto", default="argon2", argon2__type="ID", ) DOMAIN_RE = re.compile( r"^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$" ) LOCAL_PART_RE = re.compile(r"^(?=.{1,64}$)[A-Za-z0-9.!#$%&'*+/=?^_`{|}~-]+$") SELECTOR_RE = re.compile(r"^[A-Za-z0-9_-]{1,63}$") def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(password: str, hashed: str) -> bool: # bcrypt raises on >72-byte secrets; truncate for legacy bcrypt hashes. if hashed.startswith("$2"): secret = password.encode("utf-8") if len(secret) > 72: password = secret[:72].decode("utf-8", errors="ignore") return pwd_context.verify(password, hashed) def normalize_domain(value: str) -> str: candidate = value.strip().rstrip(".").lower() try: normalized = candidate.encode("idna").decode("ascii") except UnicodeError as exc: raise ValueError("Invalid domain name") from exc if not DOMAIN_RE.fullmatch(normalized): raise ValueError("Invalid domain name") return normalized def validate_local_part(value: str) -> str: candidate = value.strip() if not LOCAL_PART_RE.fullmatch(candidate) or candidate.startswith(".") or candidate.endswith(".") or ".." in candidate: raise ValueError("Invalid local part") return candidate.lower() def validate_selector(value: str) -> str: candidate = value.strip().lower() if not SELECTOR_RE.fullmatch(candidate): raise ValueError("Invalid DKIM selector") return candidate def build_dkim_key_path(domain: str, selector: str, base_dir: Path) -> Path: try: safe_domain = normalize_domain(domain) safe_selector = validate_selector(selector) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc base_path = base_dir.resolve(strict=False) key_path = (base_path / f"{safe_domain}.{safe_selector}.key").resolve(strict=False) try: key_path.relative_to(base_path) except ValueError as exc: raise HTTPException(status_code=400, detail="Invalid DKIM path") from exc return key_path def remove_managed_dkim_key(raw_path: str | Path | None) -> bool: if not raw_path: return False base_dir = get_settings().dkim_keys_dir.resolve(strict=False) candidate = Path(raw_path).resolve(strict=False) try: candidate.relative_to(base_dir) except ValueError: return False if candidate.suffix != ".key": return False try: candidate.unlink(missing_ok=True) except OSError: return False return True def _atomic_write(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) with NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as tmp_file: tmp_file.write(content) tmp_path = Path(tmp_file.name) os.replace(tmp_path, path) path.chmod(0o640) def sync_dkim_signing_maps(domains: Iterable[object]) -> None: settings = get_settings() base_dir = settings.dkim_keys_dir.resolve(strict=False) selector_lines: list[str] = [] path_lines: list[str] = [] for domain in domains: raw_path = getattr(domain, "dkim_private_path", None) if not raw_path: continue safe_domain = normalize_domain(getattr(domain, "name", "")) safe_selector = validate_selector(getattr(domain, "dkim_selector", "default") or "default") key_path = Path(str(raw_path)).resolve(strict=False) try: key_path.relative_to(base_dir) except ValueError: continue selector_lines.append(f"{safe_domain} {safe_selector}") path_lines.append(f"{safe_domain} {key_path}") if not selector_lines and not path_lines and not base_dir.exists(): return _atomic_write(base_dir / "selectors.map", "\n".join(selector_lines) + ("\n" if selector_lines else "")) _atomic_write(base_dir / "paths.map", "\n".join(path_lines) + ("\n" if path_lines else "")) def extract_dkim_dns_record(output: str) -> str | None: if not output.strip(): return None normalized = output.replace("\r", "\n") squashed = " ".join(normalized.split()) compact = re.sub(r'"\s+"', "", squashed) compact = compact.replace('"', "").replace("(", " ").replace(")", " ") compact = re.sub(r"\s+", " ", compact).strip() match = re.search(r"v=DKIM1\s*;.*?\bp\s*=\s*[A-Za-z0-9+/=]+", compact, flags=re.IGNORECASE) if not match: return None record = match.group(0) record = re.sub(r"\s*;\s*", "; ", record) record = re.sub(r"\s*=\s*", "=", record) return record.strip() def generate_dkim_key(domain: str, selector: str = "default") -> dict: settings = get_settings() try: safe_domain = normalize_domain(domain) safe_selector = validate_selector(selector) key_path = build_dkim_key_path(safe_domain, safe_selector, settings.dkim_keys_dir) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc key_path.parent.mkdir(parents=True, exist_ok=True) cmd = ["rspamadm", "dkim_keygen", "-d", safe_domain, "-s", safe_selector, "-k", str(key_path)] try: result = subprocess.run(cmd, capture_output=True, text=True, check=True) except FileNotFoundError as exc: raise HTTPException(status_code=500, detail="rspamadm is not available on this host") from exc except subprocess.CalledProcessError as exc: detail = (exc.stderr or exc.stdout or "").strip() or f"rspamadm exited with status {exc.returncode}" raise HTTPException(status_code=500, detail=f"DKIM keygen failed: {detail}") from exc if key_path.exists(): key_path.chmod(0o640) dns_record = extract_dkim_dns_record("\n".join(part for part in [result.stdout, result.stderr] if part)) if not dns_record: raise HTTPException(status_code=500, detail="DKIM key generated but DNS record output could not be parsed") return {"path": str(key_path), "dns": dns_record, "selector": safe_selector}