284 行
9.9 KiB
Python
可执行文件
284 行
9.9 KiB
Python
可执行文件
#!/usr/bin/env python3
|
|
"""Probe the public OA readiness endpoint and drive a deduplicated alert."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import http.client
|
|
import json
|
|
import os
|
|
import re
|
|
import socket
|
|
import ssl
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$")
|
|
SAFE_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
|
|
MAX_RESPONSE_BYTES = 64 * 1024
|
|
HEALTHY_STATUS = "PASS_APP_HEALTH_READY"
|
|
FAILURE_STATUS = "FAIL_APP_HEALTH"
|
|
|
|
|
|
class ProbeFailure(Exception):
|
|
def __init__(self, category: str):
|
|
super().__init__(category)
|
|
self.category = category
|
|
|
|
|
|
def fail(category: str) -> "NoReturn":
|
|
print(f"[app-health][ERROR] {category}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def validate_regular_file(path: Path, label: str, private: bool = False) -> None:
|
|
try:
|
|
info = path.lstat()
|
|
except FileNotFoundError:
|
|
fail(f"{label}_missing")
|
|
if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
|
|
fail(f"{label}_invalid")
|
|
if info.st_uid != os.geteuid():
|
|
fail(f"{label}_owner_invalid")
|
|
if private and stat.S_IMODE(info.st_mode) != 0o600:
|
|
fail(f"{label}_mode_invalid")
|
|
if not private and stat.S_IMODE(info.st_mode) & 0o022:
|
|
fail(f"{label}_mode_invalid")
|
|
|
|
|
|
def prepare_state_dir(path: Path) -> None:
|
|
try:
|
|
path.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
info = path.lstat()
|
|
except OSError:
|
|
fail("local_state_invalid")
|
|
if (
|
|
stat.S_ISLNK(info.st_mode)
|
|
or not stat.S_ISDIR(info.st_mode)
|
|
or info.st_uid != os.geteuid()
|
|
or stat.S_IMODE(info.st_mode) != 0o700
|
|
):
|
|
fail("local_state_invalid")
|
|
|
|
|
|
def validate_ready_url(value: str) -> str:
|
|
try:
|
|
parsed = urllib.parse.urlsplit(value)
|
|
port = parsed.port
|
|
except ValueError:
|
|
fail("ready_url_invalid")
|
|
if (
|
|
parsed.scheme != "https"
|
|
or not parsed.hostname
|
|
or not SAFE_HOST.fullmatch(parsed.hostname)
|
|
or parsed.username is not None
|
|
or parsed.password is not None
|
|
or parsed.path != "/ready"
|
|
or parsed.query
|
|
or parsed.fragment
|
|
or (port is not None and not 1 <= port <= 65535)
|
|
):
|
|
fail("ready_url_invalid")
|
|
return value
|
|
|
|
|
|
def load_probe_state(path: Path) -> dict[str, object]:
|
|
if not path.exists():
|
|
return {"schemaVersion": 1, "consecutiveFailures": 0, "lastCategory": None}
|
|
validate_regular_file(path, "local_state", private=True)
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
fail("local_state_invalid")
|
|
if not isinstance(payload, dict):
|
|
fail("local_state_invalid")
|
|
count = payload.get("consecutiveFailures")
|
|
category = payload.get("lastCategory")
|
|
if (
|
|
payload.get("schemaVersion") != 1
|
|
or not isinstance(count, int)
|
|
or isinstance(count, bool)
|
|
or count < 0
|
|
or (category is not None and (not isinstance(category, str) or not SAFE_TOKEN.fullmatch(category)))
|
|
):
|
|
fail("local_state_invalid")
|
|
return payload
|
|
|
|
|
|
def atomic_write(path: Path, payload: dict[str, object]) -> None:
|
|
fd, temporary = 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(temporary, path)
|
|
except BaseException:
|
|
try:
|
|
os.unlink(temporary)
|
|
except FileNotFoundError:
|
|
pass
|
|
raise
|
|
|
|
|
|
def classify_url_error(error: urllib.error.URLError) -> str:
|
|
reason = error.reason
|
|
if isinstance(reason, (ssl.SSLError, ssl.CertificateError)):
|
|
return "tls_validation_failed"
|
|
if isinstance(reason, (socket.timeout, TimeoutError)):
|
|
return "request_timeout"
|
|
return "dns_or_connect_failed"
|
|
|
|
|
|
def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
|
|
if ca_file is not None:
|
|
validate_regular_file(ca_file, "ca_file")
|
|
try:
|
|
context = ssl.create_default_context(cafile=str(ca_file) if ca_file else None)
|
|
request = urllib.request.Request(
|
|
ready_url,
|
|
headers={"Accept": "application/json", "User-Agent": "hyb-oa-app-health/1"},
|
|
method="GET",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=timeout_seconds, context=context) as response:
|
|
if not 200 <= response.status < 300:
|
|
raise ProbeFailure("http_unavailable")
|
|
content_length = response.headers.get("Content-Length")
|
|
if content_length:
|
|
try:
|
|
if int(content_length) > MAX_RESPONSE_BYTES:
|
|
raise ProbeFailure("response_too_large")
|
|
except ValueError:
|
|
raise ProbeFailure("response_contract_invalid") from None
|
|
body = response.read(MAX_RESPONSE_BYTES + 1)
|
|
except urllib.error.HTTPError:
|
|
raise ProbeFailure("http_unavailable") from None
|
|
except urllib.error.URLError as error:
|
|
raise ProbeFailure(classify_url_error(error)) from None
|
|
except (socket.timeout, TimeoutError):
|
|
raise ProbeFailure("request_timeout") from None
|
|
except (ssl.SSLError, ssl.CertificateError):
|
|
raise ProbeFailure("tls_validation_failed") from None
|
|
except (http.client.HTTPException, ConnectionError, OSError):
|
|
raise ProbeFailure("dns_or_connect_failed") from None
|
|
if len(body) > MAX_RESPONSE_BYTES:
|
|
raise ProbeFailure("response_too_large")
|
|
try:
|
|
payload = json.loads(body.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
raise ProbeFailure("response_contract_invalid") from None
|
|
if not isinstance(payload, dict) or not isinstance(payload.get("data"), dict) or payload["data"].get("ok") is not True:
|
|
raise ProbeFailure("response_contract_invalid")
|
|
|
|
|
|
def write_summary(path: Path, status: str, category: str) -> None:
|
|
finding = status == FAILURE_STATUS
|
|
payload = {
|
|
"issueId": "APP_HEALTH",
|
|
"status": status,
|
|
"conclusion": status,
|
|
"findingCount": 1 if finding else 0,
|
|
"findings": (
|
|
[{"source": "public_ready", "status": status, "severity": "P0", "detail": category}]
|
|
if finding
|
|
else []
|
|
),
|
|
"privacy": "status/count/category only; no credentials, addresses, paths, or payloads",
|
|
}
|
|
atomic_write(path, payload)
|
|
|
|
|
|
def evaluate_alert(alert_helper: Path, webhook_file: Path, summary_file: Path, alert_state_file: Path) -> None:
|
|
validate_regular_file(alert_helper, "alert_helper")
|
|
try:
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(alert_helper),
|
|
"--webhook-file",
|
|
str(webhook_file),
|
|
"--summary-file",
|
|
str(summary_file),
|
|
"--state-file",
|
|
str(alert_state_file),
|
|
],
|
|
check=False,
|
|
)
|
|
except OSError:
|
|
fail("alert_evaluation_failed")
|
|
if result.returncode != 0:
|
|
fail("alert_evaluation_failed")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--ready-url", required=True)
|
|
parser.add_argument("--state-dir", required=True, type=Path)
|
|
parser.add_argument("--webhook-file", required=True, type=Path)
|
|
parser.add_argument("--alert-helper", required=True, type=Path)
|
|
parser.add_argument("--failure-threshold", type=int, default=2)
|
|
parser.add_argument("--timeout-seconds", type=float, default=10.0)
|
|
parser.add_argument("--ca-file", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
ready_url = validate_ready_url(args.ready_url)
|
|
if not 1 <= args.failure_threshold <= 10:
|
|
fail("failure_threshold_invalid")
|
|
if not 0.1 <= args.timeout_seconds <= 30:
|
|
fail("timeout_invalid")
|
|
prepare_state_dir(args.state_dir)
|
|
probe_state_file = args.state_dir / "app-health-probe-state.json"
|
|
summary_file = args.state_dir / "last-app-health.json"
|
|
alert_state_file = args.state_dir / "app-health-alert-state.json"
|
|
state = load_probe_state(probe_state_file)
|
|
|
|
try:
|
|
probe(ready_url, args.timeout_seconds, args.ca_file)
|
|
except ProbeFailure as error:
|
|
failures = min(int(state["consecutiveFailures"]) + 1, args.failure_threshold)
|
|
try:
|
|
atomic_write(
|
|
probe_state_file,
|
|
{"schemaVersion": 1, "consecutiveFailures": failures, "lastCategory": error.category},
|
|
)
|
|
except OSError:
|
|
fail("local_state_invalid")
|
|
if failures >= args.failure_threshold:
|
|
try:
|
|
write_summary(summary_file, FAILURE_STATUS, error.category)
|
|
except OSError:
|
|
fail("local_state_invalid")
|
|
evaluate_alert(args.alert_helper, args.webhook_file, summary_file, alert_state_file)
|
|
print(
|
|
f"[app-health][ERROR] health_check_failed category={error.category} consecutive={failures} alert=evaluated",
|
|
file=sys.stderr,
|
|
)
|
|
else:
|
|
print(
|
|
f"[app-health][ERROR] health_check_failed category={error.category} consecutive={failures} alert=pending",
|
|
file=sys.stderr,
|
|
)
|
|
raise SystemExit(1)
|
|
|
|
try:
|
|
atomic_write(
|
|
probe_state_file,
|
|
{"schemaVersion": 1, "consecutiveFailures": 0, "lastCategory": None},
|
|
)
|
|
write_summary(summary_file, HEALTHY_STATUS, "public_ready_healthy")
|
|
except OSError:
|
|
fail("local_state_invalid")
|
|
evaluate_alert(args.alert_helper, args.webhook_file, summary_file, alert_state_file)
|
|
print("[app-health][PASS] public readiness is healthy")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|