#!/usr/bin/env python3 """Send a redacted, deduplicated OA operations alert to a WeCom group robot.""" from __future__ import annotations import argparse import json import os import re import stat import sys import tempfile import urllib.parse import urllib.request from pathlib import Path SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$") ALLOWED_STATUSES = { "FAIL_APP_HEALTH", "FAIL_AUTOMATIC_BACKUP", "FAIL_OFFHOST_DEADMAN", "PASS_APP_HEALTH_READY", "PASS_AUTOMATIC_BACKUP_COMPLETE", "PASS_AUTOMATIC_BACKUP_FRESH", "PASS_OFFHOST_DEADMAN_FRESH", "PASS_WITH_FINDINGS", } def fail(message: str) -> "NoReturn": print(f"[wecom-group-alert][ERROR] {message}", file=sys.stderr) raise SystemExit(1) def validate_private_file(path: Path, label: str) -> None: try: info = path.lstat() except FileNotFoundError: fail(f"{label} is missing") if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode): fail(f"{label} must be a regular non-symbolic file") if stat.S_IMODE(info.st_mode) != 0o600: fail(f"{label} mode must be 0600") if info.st_uid != os.geteuid(): fail(f"{label} must be owned by the executing account") def read_webhook(path: Path) -> str: validate_private_file(path, "webhook file") try: value = path.read_text(encoding="utf-8").strip() except OSError: fail("webhook file cannot be read") if "\n" in value or "\r" in value: fail("webhook file must contain one line") parsed = urllib.parse.urlparse(value) query = urllib.parse.parse_qs(parsed.query, strict_parsing=True) if ( parsed.scheme != "https" or parsed.hostname != "qyapi.weixin.qq.com" or parsed.netloc != "qyapi.weixin.qq.com" or parsed.path != "/cgi-bin/webhook/send" or set(query) != {"key"} or len(query["key"]) != 1 or not re.fullmatch(r"[A-Za-z0-9-]{16,128}", query["key"][0]) or parsed.fragment ): fail("webhook file format is invalid") return value def load_summary(path: Path) -> tuple[str, str, int]: validate_private_file(path, "summary file") try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): fail("summary file is invalid") status = payload.get("status") findings = payload.get("findings") or [] finding_count = payload.get("findingCount") if status not in ALLOWED_STATUSES or not isinstance(finding_count, int): fail("summary status is invalid") if finding_count != len(findings) or finding_count not in {0, 1}: fail("summary finding count is invalid") category = "healthy" if findings: category = findings[0].get("detail", "") if not isinstance(category, str) or not SAFE_TOKEN.fullmatch(category): fail("summary category is invalid") return status, category, finding_count def load_previous(path: Path) -> dict[str, object]: if not path.exists(): return {} validate_private_file(path, "alert state file") try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): fail("alert state file is invalid") return value if isinstance(value, dict) else {} def atomic_write(path: Path, payload: dict[str, object]) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(path.parent, 0o700) fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=True, indent=2) handle.write("\n") os.replace(temp_name, path) except BaseException: try: os.unlink(temp_name) except FileNotFoundError: pass raise def should_send(previous: dict[str, object], status: str, category: str) -> bool: previous_status = previous.get("status") previous_category = previous.get("category") if status.startswith("FAIL_") or status == "PASS_WITH_FINDINGS": return previous_status != status or previous_category != category return isinstance(previous_status, str) and ( previous_status.startswith("FAIL_") or previous_status == "PASS_WITH_FINDINGS" ) def safe_text(status: str, category: str) -> str: if status.startswith("FAIL_"): headline = "内部 OA 运维告警" elif status == "PASS_WITH_FINDINGS": headline = "内部 OA 运维提醒" else: headline = "内部 OA 运维恢复通知" return f"{headline}\n状态：{status}\n类别：{category}\n请按恢复卡检查，不要在群内粘贴原始日志。" def send(webhook: str, text: str) -> None: body = json.dumps({"msgtype": "text", "text": {"content": text}}).encode("utf-8") capture = os.environ.get("HYB_WECOM_ALERT_TEST_CAPTURE") if capture: if os.environ.get("HYB_WECOM_ALERT_TEST_MODE") != "1": fail("test capture is disabled") capture_path = Path(capture) with capture_path.open("ab") as handle: handle.write(body + b"\n") os.chmod(capture_path, 0o600) return request = urllib.request.Request( webhook, data=body, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(request, timeout=10) as response: result = json.loads(response.read().decode("utf-8")) except Exception: fail("group robot request failed") if result.get("errcode") != 0: fail("group robot rejected the alert") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--webhook-file", required=True, type=Path) parser.add_argument("--summary-file", required=True, type=Path) parser.add_argument("--state-file", required=True, type=Path) args = parser.parse_args() try: status, category, _ = load_summary(args.summary_file) previous = load_previous(args.state_file) send_now = should_send(previous, status, category) if send_now: webhook = read_webhook(args.webhook_file) send(webhook, safe_text(status, category)) atomic_write(args.state_file, {"schemaVersion": 1, "status": status, "category": category}) except OSError: fail("local alert file operation failed") print(f"[wecom-group-alert][PASS] evaluated sent={str(send_now).lower()}") if __name__ == "__main__": main()