runv_mail_sync.py (11669B)
1 #!/usr/bin/env python3 2 """ 3 Sincroniza aliases aprovados (email-aliases.json) com o MTA local. 4 5 Backends: 6 - postfix-hash: ficheiro + postmap (só se virtual_alias_maps usar hash) 7 - postfix-mysql: tabela MySQL já usada por mysql-virtual-alias-maps.cf 8 9 Separado do Mailgun transacional (/etc/runv-email.json). 10 """ 11 12 from __future__ import annotations 13 14 import json 15 import os 16 import re 17 import subprocess 18 import sys 19 import tempfile 20 from pathlib import Path 21 from typing import Any 22 23 import runv_community as rc 24 25 DEFAULT_CONFIG_PATH = Path("/etc/runv-member-mail.json") 26 QUERY_HINT = re.compile( 27 r"SELECT\s+[`']?(\w+)[`']?\s+FROM\s+[`']?(\w+)[`']?\s+WHERE\s+[`']?(\w+)[`']?\s*=", 28 re.IGNORECASE, 29 ) 30 31 try: 32 import runv_email_aliases as ea 33 except ImportError: # pragma: no cover 34 ea = None # type: ignore[assignment] 35 36 37 def config_path() -> Path: 38 raw = os.environ.get("RUNV_MEMBER_MAIL_CONFIG", "").strip() 39 return Path(raw) if raw else DEFAULT_CONFIG_PATH 40 41 42 def load_config(path: Path | None = None) -> dict[str, Any]: 43 cfg_path = path or config_path() 44 if not cfg_path.is_file(): 45 return {} 46 try: 47 data = json.loads(cfg_path.read_text(encoding="utf-8")) 48 except (OSError, json.JSONDecodeError) as e: 49 rc.friendly_exit(f"config inválida em {cfg_path}: {e}") 50 if not isinstance(data, dict): 51 rc.friendly_exit(f"config inválida em {cfg_path}: esperado objecto JSON.") 52 return data 53 54 55 def is_sync_enabled(cfg: dict[str, Any] | None = None) -> bool: 56 data = cfg if cfg is not None else load_config() 57 return bool(data.get("enabled")) 58 59 60 def active_forwarding_rows() -> list[tuple[str, str]]: 61 if ea is None: 62 rc.friendly_exit("módulo runv_email_aliases indisponível.") 63 rows: list[tuple[str, str]] = [] 64 for username, alias, dest in ea.list_active_aliases(): 65 _ = username 66 rows.append((alias.lower(), dest.lower())) 67 rows.sort(key=lambda r: r[0]) 68 return rows 69 70 71 def sql_literal(value: str) -> str: 72 return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'" 73 74 75 def parse_postfix_mysql_cf(path: Path) -> dict[str, str]: 76 if not path.is_file(): 77 rc.friendly_exit(f"ficheiro MySQL Postfix ausente: {path}") 78 data: dict[str, str] = {} 79 for line in path.read_text(encoding="utf-8").splitlines(): 80 line = line.strip() 81 if not line or line.startswith("#"): 82 continue 83 if "=" not in line: 84 continue 85 key, val = line.split("=", 1) 86 data[key.strip().lower()] = val.strip() 87 for req in ("user", "password", "hosts", "dbname"): 88 if not data.get(req): 89 rc.friendly_exit(f"{path} sem campo obrigatório: {req}") 90 return data 91 92 93 def mysql_exec(mysql_cfg: dict[str, str], sql: str, *, dry_run: bool) -> str: 94 if dry_run: 95 print(f"[dry-run] mysql -e {sql}") 96 return "" 97 cmd = [ 98 "mysql", 99 "-N", 100 "-B", 101 "-h", 102 mysql_cfg["hosts"], 103 "-u", 104 mysql_cfg["user"], 105 mysql_cfg["dbname"], 106 "-e", 107 sql, 108 ] 109 env = os.environ.copy() 110 env["MYSQL_PWD"] = mysql_cfg["password"] 111 proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120, env=env) 112 if proc.returncode != 0: 113 err = (proc.stderr or proc.stdout or "").strip() 114 rc.friendly_exit(f"mysql falhou: {err}") 115 return (proc.stdout or "").strip() 116 117 118 def infer_mysql_table_from_query(query: str) -> tuple[str, str, str]: 119 m = QUERY_HINT.search(query.replace("\n", " ")) 120 if not m: 121 rc.friendly_exit( 122 "não foi possível inferir tabela/colunas da query em mysql_map_file; " 123 "defina mysql.table, mysql.address_column e mysql.goto_column em runv-member-mail.json" 124 ) 125 dest_col, table, addr_col = m.group(1), m.group(2), m.group(3) 126 return table, addr_col, dest_col 127 128 129 def mysql_sync_options(cfg: dict[str, Any]) -> dict[str, str]: 130 block = cfg.get("mysql") 131 if not isinstance(block, dict): 132 block = {} 133 map_file = str( 134 block.get("map_file") 135 or cfg.get("mysql_map_file") 136 or "/etc/postfix/mysql-virtual-alias-maps.cf" 137 ) 138 parsed = parse_postfix_mysql_cf(Path(map_file)) 139 table = str(block.get("table", "")).strip() 140 addr_col = str(block.get("address_column", "")).strip() 141 goto_col = str(block.get("goto_column", "")).strip() 142 if not table or not addr_col or not goto_col: 143 inferred_table, inferred_addr, inferred_goto = infer_mysql_table_from_query( 144 parsed.get("query", "") 145 ) 146 table = table or inferred_table 147 addr_col = addr_col or inferred_addr 148 goto_col = goto_col or inferred_goto 149 managed_col = str(block.get("managed_column", "")).strip() 150 managed_val = str(block.get("managed_value", "runv-email-alias")).strip() 151 active_col = str(block.get("active_column", "")).strip() 152 active_val = str(block.get("active_value", "1")).strip() 153 return { 154 "map_file": map_file, 155 "parsed": parsed, # type: ignore[dict-item] 156 "table": table, 157 "address_column": addr_col, 158 "goto_column": goto_col, 159 "managed_column": managed_col, 160 "managed_value": managed_val, 161 "active_column": active_col, 162 "active_value": active_val, 163 } 164 165 166 def render_postfix_virtual(rows: list[tuple[str, str]]) -> str: 167 lines = [ 168 "# Gerado por runv — não editar à mão; use runv-admin-email-alias sync", 169 "# Formato: alias@dominio destino@externo", 170 ] 171 for alias, dest in rows: 172 lines.append(f"{alias}\t{dest}") 173 lines.append("") 174 return "\n".join(lines) 175 176 177 def _run_cmd(cmd: list[str], *, dry_run: bool) -> None: 178 if dry_run: 179 print(f"[dry-run] {' '.join(cmd)}") 180 return 181 proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120) 182 if proc.returncode != 0: 183 err = (proc.stderr or proc.stdout or "").strip() 184 rc.friendly_exit(f"comando falhou ({' '.join(cmd)}): {err}") 185 186 187 def _atomic_write(path: Path, content: str, *, mode: int, dry_run: bool) -> None: 188 if dry_run: 189 print(f"[dry-run] escrever {path} ({len(content)} bytes, mode {oct(mode)})") 190 return 191 path.parent.mkdir(parents=True, exist_ok=True) 192 fd, tmp_name = tempfile.mkstemp( 193 prefix=f".{path.name}.", 194 suffix=".tmp", 195 dir=str(path.parent), 196 ) 197 tmp_path = Path(tmp_name) 198 try: 199 with os.fdopen(fd, "w", encoding="utf-8") as out: 200 out.write(content) 201 out.flush() 202 os.fsync(out.fileno()) 203 os.chmod(tmp_path, mode) 204 os.replace(tmp_path, path) 205 except Exception: 206 tmp_path.unlink(missing_ok=True) 207 raise 208 209 210 def check_postfix_maps_include(target: Path, *, dry_run: bool) -> None: 211 proc = subprocess.run( 212 ["postconf", "-h", "virtual_alias_maps"], 213 capture_output=True, 214 text=True, 215 timeout=30, 216 ) 217 if proc.returncode != 0: 218 print( 219 "aviso: postconf falhou; confirme manualmente que virtual_alias_maps inclui " 220 f"hash:{target}", 221 file=sys.stderr, 222 ) 223 return 224 maps = (proc.stdout or "").strip() 225 needle = f"hash:{target}" 226 if needle in maps.replace(" ", ""): 227 return 228 print( 229 f"aviso: virtual_alias_maps usa {maps!r} e não hash:{target}.\n" 230 " Para o vosso servidor use backend postfix-mysql, não postfix-hash.", 231 file=sys.stderr, 232 ) 233 234 235 def sync_postfix_hash(*, dry_run: bool = False, cfg: dict[str, Any] | None = None) -> int: 236 data = cfg if cfg is not None else load_config() 237 target = Path(str(data.get("virtual_alias_file", "/etc/postfix/runv-member-aliases"))) 238 file_mode = int(str(data.get("file_mode", "0o644")), 8) 239 rows = active_forwarding_rows() 240 body = render_postfix_virtual(rows) 241 242 _atomic_write(target, body, mode=file_mode, dry_run=dry_run) 243 print(f"mapa Postfix: {target} ({len(rows)} alias(es) activo(s))") 244 245 if data.get("check_maps", True): 246 check_postfix_maps_include(target, dry_run=dry_run) 247 248 if data.get("run_postmap", True): 249 postmap = data.get("postmap_command") 250 if isinstance(postmap, list) and postmap: 251 cmd = [str(x) for x in postmap] 252 else: 253 cmd = ["postmap", str(target)] 254 _run_cmd(cmd, dry_run=dry_run) 255 256 if data.get("reload_postfix", True): 257 reload_cmd = data.get("reload_command") 258 if isinstance(reload_cmd, list) and reload_cmd: 259 cmd = [str(x) for x in reload_cmd] 260 else: 261 cmd = ["systemctl", "reload", "postfix"] 262 _run_cmd(cmd, dry_run=dry_run) 263 264 return 0 265 266 267 def sync_postfix_mysql(*, dry_run: bool = False, cfg: dict[str, Any] | None = None) -> int: 268 data = cfg if cfg is not None else load_config() 269 opts = mysql_sync_options(data) 270 parsed: dict[str, str] = opts["parsed"] # type: ignore[assignment] 271 table = opts["table"] 272 addr_col = opts["address_column"] 273 goto_col = opts["goto_column"] 274 managed_col = opts["managed_column"] 275 managed_val = opts["managed_value"] 276 active_col = opts["active_column"] 277 active_val = opts["active_value"] 278 279 rows = active_forwarding_rows() 280 active_addresses = {alias for alias, _ in rows} 281 282 if managed_col: 283 in_list = ", ".join(sql_literal(a) for a in sorted(active_addresses)) or "''" 284 delete_sql = ( 285 f"DELETE FROM `{table}` WHERE `{managed_col}` = {sql_literal(managed_val)} " 286 f"AND `{addr_col}` NOT IN ({in_list});" 287 ) 288 mysql_exec(parsed, delete_sql, dry_run=dry_run) 289 290 for alias, dest in rows: 291 cols = [f"`{addr_col}`", f"`{goto_col}`"] 292 vals = [sql_literal(alias), sql_literal(dest)] 293 updates = [f"`{goto_col}` = {sql_literal(dest)}"] 294 if managed_col: 295 cols.append(f"`{managed_col}`") 296 vals.append(sql_literal(managed_val)) 297 updates.append(f"`{managed_col}` = {sql_literal(managed_val)}") 298 if active_col: 299 cols.append(f"`{active_col}`") 300 vals.append(active_val) 301 updates.append(f"`{active_col}` = {active_val}") 302 upsert = ( 303 f"INSERT INTO `{table}` ({', '.join(cols)}) VALUES ({', '.join(vals)}) " 304 f"ON DUPLICATE KEY UPDATE {', '.join(updates)};" 305 ) 306 mysql_exec(parsed, upsert, dry_run=dry_run) 307 print(f" {alias} -> {dest}") 308 309 print(f"MySQL {table}: {len(rows)} alias(es) activo(s) sincronizado(s)") 310 311 if data.get("reload_postfix", True): 312 reload_cmd = data.get("reload_command") 313 if isinstance(reload_cmd, list) and reload_cmd: 314 cmd = [str(x) for x in reload_cmd] 315 else: 316 cmd = ["systemctl", "reload", "postfix"] 317 _run_cmd(cmd, dry_run=dry_run) 318 319 return 0 320 321 322 def sync_mail(*, dry_run: bool = False, cfg: dict[str, Any] | None = None) -> int: 323 data = cfg if cfg is not None else load_config() 324 if not data.get("enabled"): 325 rc.friendly_exit( 326 f"sincronização desactivada; defina enabled=true em {config_path()}" 327 ) 328 backend = str(data.get("backend", "postfix-mysql")).strip().lower() 329 if backend == "postfix-hash": 330 return sync_postfix_hash(dry_run=dry_run, cfg=data) 331 if backend == "postfix-mysql": 332 return sync_postfix_mysql(dry_run=dry_run, cfg=data) 333 rc.friendly_exit(f"backend não suportado: {backend!r}") 334 return 1 335 336 337 def maybe_sync_after_approve(*, dry_run: bool = False) -> None: 338 cfg = load_config() 339 if not cfg.get("enabled") or not cfg.get("auto_sync_on_approve"): 340 return 341 sync_mail(dry_run=dry_run, cfg=cfg)