"""Runtime configuration for the mail API. Loads environment variables once at import time, validates required values, and exposes a frozen :class:`Settings` dataclass plus module-level constants for compatibility with code that still imports the legacy names. """ from __future__ import annotations import logging import logging.handlers import os import re import sys from dataclasses import dataclass from pathlib import Path import resend # ── Logging ────────────────────────────────────────────────────────────────── def setup_logging() -> logging.Logger: log_dir = Path(os.environ.get("LOG_DIR", "logs")) log_dir.mkdir(parents=True, exist_ok=True) log_file = log_dir / "mail-api.log" fmt = logging.Formatter( "%(asctime)s %(levelname)-8s %(name)s: %(message)s", datefmt="%d/%m/%Y %H:%M:%S %Z", ) root = logging.getLogger() root.setLevel(logging.DEBUG) for handler in list(root.handlers): root.removeHandler(handler) console = logging.StreamHandler(sys.stdout) console.setLevel(logging.INFO) console.setFormatter(fmt) root.addHandler(console) rotating = logging.handlers.RotatingFileHandler( log_file, maxBytes=2_000_000, backupCount=5, encoding="utf-8" ) rotating.setLevel(logging.DEBUG) rotating.setFormatter(fmt) root.addHandler(rotating) log = logging.getLogger("mail-api") log.info("Logging initialised → console=INFO, file=%s (DEBUG, rotating)", log_file) return log logger = setup_logging() # ── Settings ───────────────────────────────────────────────────────────────── DEV_MODE = os.environ.get("DEV_MODE", "").strip().lower() in {"1", "true", "yes"} REQUIRED_ENV = { "RESEND_API_KEY": "API key from https://resend.com/api-keys", "OWNER_EMAIL": "Email address that receives new lead notifications", } def _parse_email_env_list(*values: str | None) -> list[str]: emails: list[str] = [] seen: set[str] = set() for raw in values: if not raw: continue for part in re.split(r"[;,\s]+", raw.strip()): normalized = part.strip().lower() if normalized and normalized not in seen: seen.add(normalized) emails.append(normalized) return emails @dataclass(frozen=True) class Settings: resend_api_key: str owner_email: str cp_admin_emails: tuple[str, ...] from_email: str reply_to: str owner_bcc: str client_bcc: str enable_general_enquiries: bool max_attempts: int form_min_seconds: int form_max_seconds: int rate_limit_window_seconds: int rate_limit_max_per_ip: int rate_limit_max_per_email: int rate_limit_min_interval_seconds: int email_send_timeout_seconds: int def _load_settings() -> Settings: if DEV_MODE: owner_email = os.environ.get("OWNER_EMAIL", "dev@localhost").strip().lower() return Settings( resend_api_key=os.environ.get("RESEND_API_KEY", "dev"), owner_email=owner_email, cp_admin_emails=tuple(_parse_email_env_list( owner_email, os.environ.get("SECONDARY_CP_EMAIL"), os.environ.get("SECONDARY_CP_EMAILS"), )), from_email=os.environ.get("FROM_EMAIL", "GoodWalk "), reply_to=os.environ.get("REPLY_TO", "info@goodwalk.co.nz"), owner_bcc="", client_bcc="", enable_general_enquiries=False, max_attempts=3, form_min_seconds=1, form_max_seconds=7200, rate_limit_window_seconds=900, rate_limit_max_per_ip=50, rate_limit_max_per_email=50, rate_limit_min_interval_seconds=1, email_send_timeout_seconds=20, ) missing = [(name, hint) for name, hint in REQUIRED_ENV.items() if not os.environ.get(name)] if missing: lines = [ "", "Mail API cannot start — required environment variables are not set:", "", ] for name, hint in missing: lines.append(f" • {name} ({hint})") lines += [ "", "Set them in your shell and try again. For example, in PowerShell:", "", ] for name, _ in missing: lines.append(f' $env:{name} = "..."') lines.append("") message = "\n".join(lines) logger.critical("Startup aborted: missing env vars: %s", [n for n, _ in missing]) print(message, file=sys.stderr) sys.exit(1) owner_email = os.environ["OWNER_EMAIL"].strip().lower() return Settings( resend_api_key=os.environ["RESEND_API_KEY"], owner_email=owner_email, cp_admin_emails=tuple(_parse_email_env_list( owner_email, os.environ.get("SECONDARY_CP_EMAIL"), os.environ.get("SECONDARY_CP_EMAILS"), )), from_email=os.environ.get("FROM_EMAIL", "GoodWalk "), reply_to=os.environ.get("REPLY_TO", "info@goodwalk.co.nz"), owner_bcc=os.environ.get("OWNER_BCC", "example@example.com").strip(), client_bcc=os.environ.get("CLIENT_BCC", "").strip(), enable_general_enquiries=os.environ.get("ENABLE_GENERAL_ENQUIRIES", "false").strip().lower() in {"1", "true", "yes", "on", "enabled"}, max_attempts=max(1, int(os.environ.get("MAIL_MAX_ATTEMPTS", "3"))), form_min_seconds=max(1, int(os.environ.get("FORM_MIN_SECONDS", "4"))), form_max_seconds=max(60, int(os.environ.get("FORM_MAX_SECONDS", "7200"))), rate_limit_window_seconds=max(60, int(os.environ.get("RATE_LIMIT_WINDOW_SECONDS", "900"))), rate_limit_max_per_ip=max(1, int(os.environ.get("RATE_LIMIT_MAX_PER_IP", "5"))), rate_limit_max_per_email=max(1, int(os.environ.get("RATE_LIMIT_MAX_PER_EMAIL", "3"))), rate_limit_min_interval_seconds=max(1, int(os.environ.get("RATE_LIMIT_MIN_INTERVAL_SECONDS", "20"))), email_send_timeout_seconds=max(5, int(os.environ.get("EMAIL_SEND_TIMEOUT_SECONDS", "20"))), ) settings = _load_settings() resend.api_key = settings.resend_api_key APP_VERSION = os.environ.get("APP_VERSION", "unknown") AUTH_CODE_TTL_SECONDS = max(60, int(os.environ.get("AUTH_CODE_TTL_SECONDS", "600"))) AUTH_SESSION_TTL_SECONDS = max(3600, int(os.environ.get("AUTH_SESSION_TTL_SECONDS", str(7 * 24 * 3600)))) AUTH_CODE_MAX_ATTEMPTS = 5 AUTH_CODE_REQUESTS_PER_HOUR = 5 AUTH_IP_MAX_FAILURES = max(3, int(os.environ.get("AUTH_IP_MAX_FAILURES", "10"))) AUTH_IP_FAILURE_WINDOW = max(60, int(os.environ.get("AUTH_IP_FAILURE_WINDOW", "600"))) AUTH_IP_BLOCK_DURATION = max(60, int(os.environ.get("AUTH_IP_BLOCK_DURATION", "3600"))) BIRTHDAY_CHECK_INTERVAL_SECONDS = max(3600, int(os.environ.get("BIRTHDAY_CHECK_INTERVAL_SECONDS", str(12 * 3600)))) def _split_csv_env(name: str, default: str) -> tuple[str, ...]: # Treat empty-string env (e.g. compose `${VAR:-}`) the same as "unset" so # an unset compose key falls back to the default, not to an empty allowlist. raw = (os.environ.get(name) or "").strip() or default parts = [p.strip() for p in raw.split(",")] return tuple(p for p in parts if p) CORS_ALLOWED_ORIGINS = _split_csv_env( "CORS_ALLOWED_ORIGINS", "https://goodwalk.co.nz,https://www.goodwalk.co.nz,https://clients.goodwalk.co.nz,https://cp.goodwalk.co.nz", ) TRUSTED_HOSTS = _split_csv_env( "TRUSTED_HOSTS", "goodwalk.co.nz,www.goodwalk.co.nz,clients.goodwalk.co.nz,cp.goodwalk.co.nz,localhost,127.0.0.1", ) # Hard cap on request body size. Signed contracts include a base64 PNG of the # signature (~30 KB) plus form fields, so 2 MB is a generous ceiling. MAX_REQUEST_BODY_BYTES = max(64 * 1024, int((os.environ.get("MAX_REQUEST_BODY_BYTES") or str(2 * 1024 * 1024)).strip() or str(2 * 1024 * 1024))) _DATA_DIR = Path(os.environ.get("DATA_DIR", "data")) ALLOWED_EMAILS_FILE = _DATA_DIR / "allowed_emails.json" CLIENT_PROFILES_FILE = _DATA_DIR / "client_profiles.json" DRAFTS_FILE = _DATA_DIR / "drafts.json" # Legacy seed lives OUTSIDE the data volume — it's shipped in the image so a # fresh deploy always carries the same baked-in copy. On boot the mail-api # merges this dict into _client_profiles, adding any emails that aren't # already present. It never overwrites a live entry. LEGACY_SEED_FILE = Path( os.environ.get("LEGACY_SEED_FILE", "/app/legacy-clients-seed.json") ) LOGO_URL = "https://www.goodwalk.co.nz/images/goodwalk-auckland-dog-walking-logo.png" # ── Legacy module constants (kept for compatibility with the existing main.py) ── OWNER_EMAIL = settings.owner_email CP_ADMIN_EMAILS = set(settings.cp_admin_emails) OWNER_BCC = settings.owner_bcc CLIENT_BCC = settings.client_bcc FROM_EMAIL = settings.from_email REPLY_TO = settings.reply_to ENABLE_GENERAL_ENQUIRIES = settings.enable_general_enquiries MAX_SEND_ATTEMPTS = settings.max_attempts FORM_MIN_SECONDS = settings.form_min_seconds FORM_MAX_SECONDS = settings.form_max_seconds RATE_LIMIT_WINDOW_SECONDS = settings.rate_limit_window_seconds RATE_LIMIT_MAX_PER_IP = settings.rate_limit_max_per_ip RATE_LIMIT_MAX_PER_EMAIL = settings.rate_limit_max_per_email RATE_LIMIT_MIN_INTERVAL_SECONDS = settings.rate_limit_min_interval_seconds EMAIL_SEND_TIMEOUT_SECONDS = settings.email_send_timeout_seconds # Owner-BCC placeholder used by deploy.env.template; treat it as "unset" for the smoke email. STARTUP_TEST_RECIPIENT = OWNER_BCC if OWNER_BCC and OWNER_BCC.lower() != "example@example.com" else "" # Shared secret presented by the deploy script's post-deploy form smoke tests. # When set, requests carrying a matching X-Deploy-Smoke header short-circuit the # form handlers before any email/db side effects so production submissions stay # clean. When unset, the bypass is fully disabled. DEPLOY_SMOKE_SECRET = (os.environ.get("DEPLOY_SMOKE_SECRET") or "").strip() logger.info( "Mail API config: version=%r timezone=%r from=%r reply_to=%r owner=%r cp_admins=%r owner_bcc=%r client_bcc=%r general_enquiries=%r max_attempts=%d form_min=%ss form_max=%ss rate_window=%ss per_ip=%d per_email=%d min_interval=%ss send_timeout=%ss", APP_VERSION, os.environ.get("TZ", "system-default"), FROM_EMAIL, REPLY_TO, OWNER_EMAIL, sorted(CP_ADMIN_EMAILS), OWNER_BCC, CLIENT_BCC, ENABLE_GENERAL_ENQUIRIES, MAX_SEND_ATTEMPTS, FORM_MIN_SECONDS, FORM_MAX_SECONDS, RATE_LIMIT_WINDOW_SECONDS, RATE_LIMIT_MAX_PER_IP, RATE_LIMIT_MAX_PER_EMAIL, RATE_LIMIT_MIN_INTERVAL_SECONDS, EMAIL_SEND_TIMEOUT_SECONDS, )