141 lines
4.5 KiB
Python
Executable File
141 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Atomically replace only the SMTP section of a monitor configuration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shlex
|
|
import stat
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
EXPECTED_KEYS = {
|
|
"EMAIL_FROM_ADDRESS",
|
|
"EMAIL_FROM_NAME",
|
|
"SMTP_PASSWORD",
|
|
"SMTP_PORT",
|
|
"SMTP_RELAY",
|
|
"SMTP_SSL",
|
|
"SMTP_TLS",
|
|
"SMTP_USERNAME",
|
|
"SUPPORT_INBOX_ADDRESS",
|
|
}
|
|
|
|
|
|
def parse_values(path: Path) -> dict[str, str]:
|
|
mode = stat.S_IMODE(path.stat().st_mode)
|
|
if mode not in {0o400, 0o600}:
|
|
raise ValueError("SMTP values file must have mode 0400 or 0600")
|
|
|
|
values: dict[str, str] = {}
|
|
with path.open(encoding="utf-8") as handle:
|
|
for line_number, raw_line in enumerate(handle, start=1):
|
|
line = raw_line.rstrip("\r\n")
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if "=" not in line:
|
|
raise ValueError(f"Malformed SMTP value on line {line_number}")
|
|
key, raw_value = line.split("=", 1)
|
|
if key in values:
|
|
raise ValueError(f"Duplicate SMTP value: {key}")
|
|
parsed = shlex.split(raw_value, comments=False, posix=True)
|
|
if len(parsed) != 1 or not parsed[0]:
|
|
raise ValueError(f"SMTP value must contain one non-empty token: {key}")
|
|
values[key] = parsed[0]
|
|
|
|
missing = sorted(EXPECTED_KEYS - values.keys())
|
|
extra = sorted(values.keys() - EXPECTED_KEYS)
|
|
if missing or extra:
|
|
details: list[str] = []
|
|
if missing:
|
|
details.append("missing " + ", ".join(missing))
|
|
if extra:
|
|
details.append("unexpected " + ", ".join(extra))
|
|
raise ValueError("Invalid SMTP values file: " + "; ".join(details))
|
|
return values
|
|
|
|
|
|
def parse_bool(value: str, name: str) -> bool:
|
|
normalized = value.lower()
|
|
if normalized in {"true", "1"}:
|
|
return True
|
|
if normalized in {"false", "0"}:
|
|
return False
|
|
raise ValueError(f"{name} must be true, false, 1, or 0")
|
|
|
|
|
|
def build_smtp(values: dict[str, str]) -> dict[str, Any]:
|
|
try:
|
|
port = int(values["SMTP_PORT"])
|
|
except ValueError as error:
|
|
raise ValueError("SMTP_PORT must be an integer") from error
|
|
if not 1 <= port <= 65535:
|
|
raise ValueError("SMTP_PORT must be between 1 and 65535")
|
|
|
|
tls = values["SMTP_TLS"].lower()
|
|
if tls not in {"always", "never"}:
|
|
raise ValueError("SMTP_TLS must be always or never for the external monitor")
|
|
implicit_ssl = parse_bool(values["SMTP_SSL"], "SMTP_SSL")
|
|
if implicit_ssl and tls != "never":
|
|
raise ValueError("SMTP_TLS must be never when SMTP_SSL enables implicit TLS")
|
|
|
|
return {
|
|
"from_address": values["EMAIL_FROM_ADDRESS"],
|
|
"from_name": values["EMAIL_FROM_NAME"],
|
|
"implicit_ssl": implicit_ssl,
|
|
"password": values["SMTP_PASSWORD"],
|
|
"port": port,
|
|
"recipient": values["SUPPORT_INBOX_ADDRESS"],
|
|
"relay": values["SMTP_RELAY"],
|
|
"starttls": tls == "always",
|
|
"username": values["SMTP_USERNAME"],
|
|
}
|
|
|
|
|
|
def read_config(path: Path) -> dict[str, Any]:
|
|
with path.open(encoding="utf-8") as handle:
|
|
config = json.load(handle)
|
|
if not isinstance(config, dict) or not isinstance(config.get("smtp"), dict):
|
|
raise ValueError("Monitor configuration must contain an SMTP object")
|
|
return config
|
|
|
|
|
|
def write_atomic(path: Path, config: dict[str, Any]) -> None:
|
|
descriptor, temporary_name = tempfile.mkstemp(
|
|
dir=path.parent, prefix=f".{path.name}.smtp."
|
|
)
|
|
temporary = Path(temporary_name)
|
|
try:
|
|
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
|
|
json.dump(config, handle, ensure_ascii=False, indent=2, sort_keys=True)
|
|
handle.write("\n")
|
|
temporary.chmod(0o600)
|
|
temporary.replace(path)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("config", type=Path)
|
|
parser.add_argument("values", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
config = read_config(args.config)
|
|
config["smtp"] = build_smtp(parse_values(args.values))
|
|
write_atomic(args.config, config)
|
|
except (OSError, ValueError, json.JSONDecodeError) as error:
|
|
parser.error(str(error))
|
|
|
|
print("External monitor SMTP configuration updated without printing credentials.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|