#!/usr/bin/env python3 """Independent, dependency-free Pulse liveness check for an external monitor.""" from __future__ import annotations import argparse import sys from urllib.error import HTTPError, URLError from urllib.parse import urlparse from urllib.request import HTTPRedirectHandler, Request, build_opener class NoRedirectHandler(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def check(url: str, timeout: float = 5.0) -> tuple[bool, str]: parsed = urlparse(url) if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password: return False, "endpoint must be an http(s) URL without embedded credentials" if timeout < 1 or timeout > 30: return False, "timeout must be between 1 and 30 seconds" request = Request(url, headers={"Accept": "text/plain", "User-Agent": "pulse-deadman/1"}, method="GET") try: with build_opener(NoRedirectHandler).open(request, timeout=timeout) as response: body = response.read(64).decode("utf-8", errors="replace").strip() if response.status != 200 or body != "ok": return False, f"unexpected liveness response ({response.status})" except HTTPError as error: return False, f"liveness returned HTTP {error.code}" except (URLError, TimeoutError, OSError) as error: return False, f"liveness request failed ({error.__class__.__name__})" return True, "ok" def main() -> int: parser = argparse.ArgumentParser(description="Check Pulse /healthz from an independent monitor") parser.add_argument("url", help="HTTPS URL to Pulse /healthz") parser.add_argument("--timeout", type=float, default=5.0) args = parser.parse_args() ok, message = check(args.url, args.timeout) print(message) return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())