#!/usr/bin/env python3 """Stateful external production health and operations notifications.""" from __future__ import annotations import argparse import json import os import smtplib import ssl import sys import tempfile import urllib.error import urllib.request from datetime import datetime, timezone from email.message import EmailMessage from pathlib import Path from typing import Any def utc_now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def read_json(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as handle: value = json.load(handle) if not isinstance(value, dict): raise ValueError(f"Expected a JSON object in {path}") return value def write_json_atomic(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) descriptor, temporary_name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.") temporary = Path(temporary_name) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(value, 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 required_text(mapping: dict[str, Any], name: str) -> str: value = mapping.get(name) if not isinstance(value, str) or not value.strip(): raise ValueError(f"Missing non-empty configuration value: {name}") return value.strip() def send_message(config: dict[str, Any], subject: str, body: str) -> None: smtp = config.get("smtp") if not isinstance(smtp, dict): raise ValueError("Missing SMTP configuration") relay = required_text(smtp, "relay") port = int(smtp.get("port")) username = required_text(smtp, "username") password = required_text(smtp, "password") from_address = required_text(smtp, "from_address") from_name = required_text(smtp, "from_name") recipient = required_text(smtp, "recipient") implicit_ssl = bool(smtp.get("implicit_ssl", False)) starttls = bool(smtp.get("starttls", True)) message = EmailMessage() message["From"] = f"{from_name} <{from_address}>" message["To"] = recipient message["Subject"] = subject message.set_content(body) context = ssl.create_default_context() if implicit_ssl: client_context = smtplib.SMTP_SSL(relay, port, timeout=30, context=context) else: client_context = smtplib.SMTP(relay, port, timeout=30) with client_context as client: if not implicit_ssl: client.ehlo() if starttls: client.starttls(context=context) client.ehlo() client.login(username, password) client.send_message(message) def check_health(config: dict[str, Any]) -> tuple[str, str]: url = required_text(config, "health_url") timeout = float(config.get("health_timeout_seconds")) request = urllib.request.Request( url, headers={"User-Agent": "WhoNeedHelp-External-Monitor/1.0"}, ) try: with urllib.request.urlopen(request, timeout=timeout) as response: status = response.status payload = response.read(4096) parsed = json.loads(payload.decode("utf-8")) if status == 200 and parsed == {"status": "ready"}: return "up", f"HTTP {status}; ready payload matched" return "down", f"HTTP {status}; unexpected readiness payload" except (OSError, ValueError, urllib.error.URLError) as error: return "down", f"{type(error).__name__}: {str(error)[:500]}" def monitor(config_path: Path, state_path: Path) -> int: config = read_json(config_path) previous: dict[str, Any] = {} if state_path.exists(): previous = read_json(state_path) status, detail = check_health(config) previous_status = previous.get("status") if status != previous_status: if status == "down": send_message( config, "[Who Need Help] Production readiness is DOWN", "\n".join( [ "The independent production readiness check failed.", f"URL: {required_text(config, 'health_url')}", f"Observed at: {utc_now()}", f"Result: {detail}", "This alert is sent once per state transition.", ] ), ) elif previous_status == "down": send_message( config, "[Who Need Help] Production readiness recovered", "\n".join( [ "The independent production readiness check recovered.", f"URL: {required_text(config, 'health_url')}", f"Observed at: {utc_now()}", f"Result: {detail}", ] ), ) write_json_atomic( state_path, { "checked_at": utc_now(), "detail": detail, "status": status, }, ) print(json.dumps({"status": status, "detail": detail}, sort_keys=True)) return 0 if status == "up" else 1 def notify_backup_failure(config_path: Path, unit: str) -> int: config = read_json(config_path) send_message( config, "[Who Need Help] Production backup or restore verification failed", "\n".join( [ "The scheduled encrypted production backup did not finish successfully.", f"Unit: {unit}", f"Observed at: {utc_now()}", "Inspect the local user-systemd journal and do not treat the newest snapshot as verified until a restore drill passes.", ] ), ) print("Backup failure notification sent.") return 0 def send_test_notification(config_path: Path) -> int: config = read_json(config_path) send_message( config, "[Who Need Help] Operations monitoring test", "\n".join( [ "This is a one-time delivery verification for the independent production monitor.", f"Health URL: {required_text(config, 'health_url')}", f"Sent at: {utc_now()}", "No production incident was detected and no application data was changed.", ] ), ) print("Operations monitoring test notification sent.") return 0 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument( "--config", type=Path, default=Path.home() / ".config/who-need-help/monitor.json", ) parser.add_argument( "--state", type=Path, default=Path.home() / ".local/state/who-need-help/monitor.json", ) subparsers = parser.add_subparsers(dest="action", required=True) subparsers.add_parser("check") subparsers.add_parser("send-test-notification") backup = subparsers.add_parser("notify-backup-failure") backup.add_argument("--unit", required=True) return parser.parse_args() def main() -> int: args = parse_args() try: if args.action == "check": return monitor(args.config, args.state) if args.action == "send-test-notification": return send_test_notification(args.config) return notify_backup_failure(args.config, args.unit) except (OSError, ValueError, smtplib.SMTPException, json.JSONDecodeError) as error: print(f"Operations monitor failed: {type(error).__name__}: {error}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())