"""Deployment topology modes for Limristem eMail. Mode names (product + config): standalone — single-node classic install (default) routed — Routed Mesh: split-domain multi-backend (mailboxes on different hosts) cluster — Storage Cluster: shared Maildir (NFS/Ceph/S3) multi-frontend Routed DB strategies: shared — one MariaDB master used by all nodes synced — each node has local DB; cluster_sync_queue replicates users Cluster storage backends: nfs | ceph | s3 """ from __future__ import annotations import hashlib import hmac import json import logging import os import re import secrets from pathlib import Path from typing import Any, Literal from urllib import error as urllib_error from urllib import parse as urllib_parse from urllib import request as urllib_request from .crypto import CryptoUnavailableError, decrypt_token, encrypt_secret_if_needed, encrypt_token from .settings import get_settings logger = logging.getLogger(__name__) TopologyMode = Literal["standalone", "routed", "cluster"] RoutedDbStrategy = Literal["shared", "synced"] ClusterStorageBackend = Literal["nfs", "ceph", "s3"] MODE_LABELS = { "standalone": "Standalone", "routed": "Routed Mesh", "cluster": "Storage Cluster", } MODE_DESCRIPTIONS = { "standalone": "Single server owns all mailboxes, DNS and storage (default).", "routed": "Same domain across multiple hosts; each mailbox has a home server and mail is relayed there.", "cluster": "Multiple frontends share the same mailbox storage (NFS, Ceph or S3 with local cache).", } SERVER_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,62}$") SAFE_URL_RE = re.compile(r"^https://[A-Za-z0-9._:-]+(?::\d+)?(?:/.*)?$") def _env_or_file(key: str, default: str = "") -> str: if key in os.environ: return os.environ[key] try: from .limristem_mail_backend import _read_env_file, _resolve_main_env_file env_map = _read_env_file(_resolve_main_env_file()) if key in env_map: return str(env_map[key]) except Exception as exc: logger.debug("_env_or_file(%s) fallback failed: %s", key, exc) return default def topology_mode() -> TopologyMode: raw = (_env_or_file("LIMRISTEM_MAIL_TOPOLOGY_MODE", "standalone") or "standalone").strip().lower() if raw in {"standalone", "routed", "cluster"}: return raw # type: ignore[return-value] return "standalone" def routed_db_strategy() -> RoutedDbStrategy: raw = (_env_or_file("LIMRISTEM_MAIL_ROUTED_DB_STRATEGY", "shared") or "shared").strip().lower() return "synced" if raw == "synced" else "shared" def cluster_storage_backend() -> ClusterStorageBackend | None: if topology_mode() != "cluster": return None raw = (_env_or_file("LIMRISTEM_MAIL_CLUSTER_STORAGE", "nfs") or "nfs").strip().lower() if raw in {"nfs", "ceph", "s3"}: return raw # type: ignore[return-value] return "nfs" def local_server_id() -> str: value = (_env_or_file("LIMRISTEM_MAIL_SERVER_ID", "") or "").strip().lower() if value and SERVER_ID_RE.fullmatch(value): return value hostname = get_settings().hostname.split(".")[0] candidate = re.sub(r"[^a-z0-9._-]", "-", hostname.lower())[:48] or "node1" if not candidate[0].isalnum(): candidate = f"n{candidate}" return candidate def internode_token() -> str | None: """Shared secret for node-to-node API calls (never logged).""" return _env_or_file("LIMRISTEM_MAIL_INTERNODE_TOKEN", "") or None def internode_token_hash() -> str | None: token = internode_token() if not token: return None return hashlib.sha256(token.encode("utf-8")).hexdigest() def verify_internode_token(presented: str | None) -> bool: expected = internode_token() if not expected or not presented: return False return secrets.compare_digest(presented.encode("utf-8"), expected.encode("utf-8")) def topology_summary() -> dict[str, Any]: mode = topology_mode() payload: dict[str, Any] = { "mode": mode, "label": MODE_LABELS[mode], "description": MODE_DESCRIPTIONS[mode], "local_server_id": local_server_id(), "internode_configured": bool(internode_token()), } if mode == "routed": payload["routed_db_strategy"] = routed_db_strategy() if mode == "cluster": payload["cluster_storage"] = cluster_storage_backend() payload["cluster_storage_path"] = _env_or_file("LIMRISTEM_MAIL_CLUSTER_STORAGE_PATH", "") payload["s3_cache_ttl_seconds"] = int(_env_or_file("LIMRISTEM_MAIL_S3_CACHE_TTL_SECONDS", "3600") or 3600) payload["s3_cache_max_mb"] = int(_env_or_file("LIMRISTEM_MAIL_S3_CACHE_MAX_MB", "10240") or 10240) return payload def validate_server_id(value: str) -> str: candidate = value.strip().lower() if not SERVER_ID_RE.fullmatch(candidate): raise ValueError("Invalid server_id (use [a-z0-9._-], max 63 chars)") return candidate def validate_api_base_url(value: str) -> str: import ipaddress candidate = value.strip().rstrip("/") if any(ord(char) < 32 or ord(char) == 127 for char in candidate) or not SAFE_URL_RE.fullmatch(candidate): raise ValueError("api_base_url must be an https:// URL without credentials") parsed = urllib_parse.urlparse(candidate) if parsed.username or parsed.password or parsed.query or parsed.fragment: raise ValueError("api_base_url must not contain credentials, query or fragment") hostname = (parsed.hostname or "").strip().lower().rstrip(".") if not hostname or (parsed.port is not None and not 1 <= parsed.port <= 65535): raise ValueError("Invalid API host or port") if hostname in {"localhost", "localhost.localdomain", "metadata", "metadata.google.internal"}: raise ValueError("api_base_url host is not allowed") try: ip_obj = ipaddress.ip_address(hostname) except ValueError: ip_obj = None if ip_obj is not None and ( ip_obj.is_loopback or ip_obj.is_link_local or ip_obj.is_unspecified or ip_obj.is_multicast ): raise ValueError("api_base_url IP address is not allowed") return candidate def cluster_config_path() -> Path: # State directory, not config/: the topology wizard writes this in-process under # the service account, and config/ must stay root-owned so that account cannot # replace the env files the root helpers read. from .limristem_mail_backend import _state_dir return _state_dir() / "topology.json" def _legacy_cluster_config_path() -> Path: return get_settings().base_dir / "config" / "topology.json" def load_topology_file() -> dict[str, Any]: path = cluster_config_path() if not path.is_file(): # Installs made before the state directory split kept this under config/. legacy = _legacy_cluster_config_path() if not legacy.is_file(): return {} path = legacy try: data = json.loads(path.read_text(encoding="utf-8")) return data if isinstance(data, dict) else {} except (OSError, ValueError, TypeError): return {} def save_topology_file(payload: dict[str, Any]) -> None: path = cluster_config_path() path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".json.tmp") # Never persist raw internode token in world-readable form; only hash/meta. safe = dict(payload) if "internode_token" in safe: safe.pop("internode_token", None) safe["internode_token_configured"] = True # Encrypt storage secrets if present (idempotent if already encrypted) for secret_key in ("s3_secret_access_key", "storage_password", "ceph_key"): if safe.get(secret_key): safe[secret_key] = encrypt_secret_if_needed(str(safe[secret_key])) tmp.write_text(json.dumps(safe, indent=2, sort_keys=True) + "\n", encoding="utf-8") os.replace(tmp, path) os.chmod(path, 0o600) def decrypt_topology_secrets(payload: dict[str, Any]) -> dict[str, Any]: out = dict(payload) for secret_key in ("s3_secret_access_key", "storage_password", "ceph_key"): if out.get(secret_key): try: out[secret_key] = decrypt_token(str(out[secret_key])) or "" except CryptoUnavailableError: out[secret_key] = "" return out def sign_sync_payload(body: bytes) -> str: token = internode_token() if not token: raise RuntimeError("Internode token is not configured") return hmac.new(token.encode("utf-8"), body, hashlib.sha256).hexdigest() def verify_sync_signature(body: bytes, signature: str | None) -> bool: if not signature: return False try: expected = sign_sync_payload(body) except RuntimeError: return False return secrets.compare_digest(expected.encode("utf-8"), signature.encode("utf-8")) class _NoPeerRedirect(urllib_request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # urllib otherwise forwards the internode bearer secret to a different # host, or even to plaintext HTTP, when a peer sends a redirect. raise urllib_error.HTTPError(req.full_url, code, "Peer redirects are not allowed", headers, fp) def internode_request( api_base_url: str, path: str, *, method: str = "GET", payload: dict[str, Any] | None = None, timeout: int = 20, ) -> dict[str, Any]: """Authenticated HTTPS call to a peer node. Refuses non-HTTPS.""" base = validate_api_base_url(api_base_url) token = internode_token() if not token: raise RuntimeError("Internode token is not configured") url = f"{base}{path if path.startswith('/') else '/' + path}" body = json.dumps(payload).encode("utf-8") if payload is not None else None headers = { "Authorization": f"Bearer {token}", "X-Limristem-Internode": token, "Accept": "application/json", "Content-Type": "application/json", "User-Agent": "limristem-mail-internode/1.0", } if body is not None: headers["X-Limristem-Signature"] = sign_sync_payload(body) req = urllib_request.Request(url, data=body, method=method, headers=headers) try: with urllib_request.build_opener(_NoPeerRedirect()).open(req, timeout=timeout) as resp: raw = resp.read(1024 * 1024) except urllib_error.HTTPError as exc: detail = exc.read(4096).decode("utf-8", errors="replace") raise RuntimeError(f"Peer HTTP {exc.code}: {detail[:300]}") from exc except urllib_error.URLError as exc: raise RuntimeError(f"Peer connection failed: {exc.reason}") from exc try: data = json.loads(raw.decode("utf-8")) except json.JSONDecodeError as exc: raise RuntimeError("Peer returned invalid JSON") from exc if not isinstance(data, dict): raise RuntimeError("Peer returned unexpected payload") return data def transport_map_line(username: str, smtp_host: str, smtp_port: int = 25) -> str: """Postfix transport map entry for a remote mailbox.""" user = username.strip().lower() from .utils import normalize_domain, validate_local_part local, separator, domain = user.partition("@") if not separator: raise ValueError("Invalid transport mailbox") user = f"{validate_local_part(local)}@{normalize_domain(domain)}" host = smtp_host.strip().rstrip(".") if not user or not host: raise ValueError("Invalid transport map components") # smtp: inet only — never shell metacharacters if not re.fullmatch(r"[A-Za-z0-9._-]+", host.replace(":", "")): # allow IPv6 in brackets later; basic hostnames only for MVP if not re.fullmatch(r"[A-Za-z0-9.:\[\]-]+", host): raise ValueError("Invalid smtp_host") port = int(smtp_port) if port < 1 or port > 65535: raise ValueError("Invalid smtp_port") return f"{user} smtp:[{host}]:{port}"