#!/usr/bin/env python3 #checkmk_call # -*- coding: utf-8 -*- """ Checkmk Notification → HTTP POST (whitelist only) This script reads a fixed, whitelisted set of NOTIFY_* environment variables (from Checkmk notifications) and forwards them to a remote HTTP endpoint as JSON. - Only the keys listed in NOTIFY_KEYS are sent. - Supports Bearer token or Basic auth via environment variables. - Includes retries, timeouts, TLS verification control, and debug mode. Exit codes: 0 = success 1 = HTTP/network error 2 = unexpected error """ import json import os import sys import time from typing import Dict, Tuple import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # ------------------------- Configuration ------------------------- # Your API endpoint (override via env: NOTIFY_API_URL) API_URL = os.environ.get("NOTIFY_API_URL", "http://10.201.102.97:9660/webhook") # Auth (choose one) API_BEARER_TOKEN = "Hj4umM4mxRBF9FSeFx3e" # e.g., "eyJhbGciOi..." API_BASIC_USER = os.environ.get("NOTIFY_API_USER") # e.g., "cmk_client" API_BASIC_PASS = os.environ.get("NOTIFY_API_PASS") # e.g., "s3cret" # TLS verification (keep True in prod) VERIFY_TLS = os.environ.get("NOTIFY_VERIFY_TLS", "true").lower() not in ("0", "false", "no") # Networking TIMEOUT: Tuple[float, float] = (5.0, 15.0) # (connect, read) RETRY_TOTAL = int(os.environ.get("NOTIFY_RETRY_TOTAL", "3")) RETRY_BACKOFF = float(os.environ.get("NOTIFY_RETRY_BACKOFF", "0.5")) # Debug logging to stderr (avoid in prod if sensitive) DEBUG = os.environ.get("NOTIFY_DEBUG", "false").lower() in ("1", "true", "yes") # Optional meta tag SOURCE_TAG = os.environ.get("NOTIFY_SOURCE_TAG", "checkmk") # ------------------------- Whitelisted keys ------------------------- NOTIFY_KEYS = [ "NOTIFY_CONTACTNAME", "NOTIFY_CONTACTEMAIL", "NOTIFY_CONTACTPAGER", "NOTIFY_NOTIFICATIONTYPE", "NOTIFY_HOSTNOTIFICATIONNUMBER", "NOTIFY_SERVICENOTIFICATIONNUMBER", "NOTIFY_HOSTPROBLEMID", "NOTIFY_SERVICEPROBLEMID", "NOTIFY_HOSTNAME", "NOTIFY_HOSTALIAS", "NOTIFY_HOSTADDRESS", "NOTIFY_HOSTATTEMPT", "NOTIFY_LASTHOSTSTATE", "NOTIFY_LASTHOSTSTATEID", "NOTIFY_LASTHOSTSTATECHANGE", "NOTIFY_LASTHOSTUP", "NOTIFY_HOSTSTATE", "NOTIFY_HOSTSTATEID", "NOTIFY_HOSTCHECKCOMMAND", "NOTIFY_HOSTOUTPUT", "NOTIFY_HOSTPERFDATA", "NOTIFY_LONGHOSTOUTPUT", "NOTIFY_SERVICEDESC", "NOTIFY_LASTSERVICESTATE", "NOTIFY_LASTSERVICESTATEID", "NOTIFY_LASTSERVICESTATECHANGE", "NOTIFY_LASTSERVICEOK", "NOTIFY_SERVICEATTEMPT", "NOTIFY_SERVICESTATE", "NOTIFY_SERVICESTATEID", "NOTIFY_SERVICEOUTPUT", "NOTIFY_LONGSERVICEOUTPUT", "NOTIFY_SERVICEPERFDATA", "NOTIFY_SERVICECHECKCOMMAND", "NOTIFY_DATE", "NOTIFY_SHORTDATETIME", "NOTIFY_LONGDATETIME", "NOTIFY_HOSTDOWNTIME", "NOTIFY_NOTIFICATIONCOMMENT", "NOTIFY_NOTIFICATIONAUTHOR", "NOTIFY_NOTIFICATIONAUTHORNAME", "NOTIFY_NOTIFICATIONAUTHORALIAS", "NOTIFY_SERVICEACKAUTHOR", "NOTIFY_SERVICEACKCOMMENT", "NOTIFY_SERVICEGROUPNAMES", "NOTIFY_HOSTACKAUTHOR", "NOTIFY_HOSTACKCOMMENT", "NOTIFY_HOSTGROUPNAMES", "NOTIFY_HOSTTAGS", "NOTIFY_HOST_SL", "NOTIFY_SVC_SL", "NOTIFY_SERVICE_SL", "NOTIFY_HOST_EC_CONTACT", "NOTIFY_SERVICE_EC_CONTACT", "NOTIFY_HOST_ADDRESS_4", "NOTIFY_HOST_ADDRESS_6", "NOTIFY_HOST_ADDRESS_FAMILY", "NOTIFY_HOST_ALERTNOTIFICATION", "NOTIFY_CALLTYPE", ] # ------------------------- Helpers ------------------------- def normalize_value(v: str) -> str: """Strip outer quotes if the shell passed them through.""" if v is None: return "" vv = v.strip() if (vv.startswith("'") and vv.endswith("'")) or (vv.startswith('"') and vv.endswith('"')): return vv[1:-1] return vv def collect_whitelisted() -> Dict[str, str]: """Collect only the whitelisted NOTIFY_* keys from the environment.""" data: Dict[str, str] = {} for key in NOTIFY_KEYS: if key in os.environ: data[key] = normalize_value(os.environ.get(key, "")) # If a key is missing, we simply omit it. return data def build_session() -> requests.Session: """Create a session with retry strategy.""" session = requests.Session() retry = Retry( total=RETRY_TOTAL, connect=RETRY_TOTAL, read=RETRY_TOTAL, backoff_factor=RETRY_BACKOFF, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset(["POST"]), raise_on_status=False, respect_retry_after_header=True, ) adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=20) session.mount("http://", adapter) session.mount("https://", adapter) return session def auth_headers_and_tuple(): headers = {"Content-Type": "application/json"} auth = None if API_BEARER_TOKEN: headers["X-API-Key"] = f"{API_BEARER_TOKEN}" elif API_BASIC_USER and API_BASIC_PASS: auth = (API_BASIC_USER, API_BASIC_PASS) return headers, auth def main() -> int: payload = collect_whitelisted() # Optional metadata (non-intrusive) payload["_meta"] = { "source": SOURCE_TAG, "ts_epoch": int(time.time()), "schema": "checkmk.notify.v1", } if DEBUG: print("[DEBUG] URL:", API_URL, file=sys.stderr) print("[DEBUG] TLS verify:", VERIFY_TLS, file=sys.stderr) print("[DEBUG] Keys included:", sorted(k for k in payload.keys() if k != "_meta"), file=sys.stderr) session = build_session() headers, auth = auth_headers_and_tuple() try: resp = session.post( API_URL, data=json.dumps(payload, ensure_ascii=False), headers=headers, timeout=TIMEOUT, verify=VERIFY_TLS, auth=auth, ) except requests.RequestException as e: print(f"[ERROR] HTTP request failed: {e}", file=sys.stderr) return 1 except Exception as e: print(f"[ERROR] Unexpected error: {e}", file=sys.stderr) return 2 if not (200 <= resp.status_code < 300): body = "" try: body = resp.text[:500] except Exception: body = "" print(f"[ERROR] API responded {resp.status_code}. Body (first 500 chars): {body}", file=sys.stderr) return 1 if DEBUG: try: print("[DEBUG] Response JSON:", resp.json(), file=sys.stderr) except Exception: print("[DEBUG] Response Text:", resp.text, file=sys.stderr) return 0 if __name__ == "__main__": sys.exit(main())