232 lines
7.7 KiB
Python
Executable File
232 lines
7.7 KiB
Python
Executable File
#!/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())
|