#!/usr/bin/env python3
"""certwatch: one-file server alerts. TLS certificate expiry, disk space, failed systemd
units and URLs that stop answering. Python 3.8+, standard library only, no Docker.

  python3 certwatch.py --cert example.com --disk / --units --url https://example.com/
  python3 certwatch.py --config /etc/certwatch.ini        # same options, one per line
  python3 certwatch.py --self-test

Alerts go to stdout (cron mails it to you), and optionally to an ntfy topic (phone push)
or a Slack/Discord webhook. It only alerts when something changes: a new problem, or
everything back to OK. Exit code 1 while any problem is open.
From raidandev.org/tools/certwatch/. MIT licence.
"""
import argparse, json, os, shutil, socket, ssl, subprocess, sys, time, urllib.request

VERSION = "1.0.0"


def cert_days_left(host, port=443, timeout=10):
    ctx = ssl.create_default_context()
    with socket.create_connection((host, port), timeout=timeout) as s:
        with ctx.wrap_socket(s, server_hostname=host) as t:
            exp = ssl.cert_time_to_seconds(t.getpeercert()["notAfter"])
    return (exp - time.time()) / 86400


def check_cert(target, warn_days):
    host, _, port = target.partition(":")
    try:
        d = cert_days_left(host, int(port or 443))
    except ssl.SSLCertVerificationError as e:
        return f"cert {target}: invalid ({e.verify_message})"
    except (OSError, ValueError) as e:
        return f"cert {target}: cannot connect ({e.__class__.__name__})"
    return f"cert {target}: expires in {d:.0f} days" if d < warn_days else None


def disk_problem(path, used, total, max_pct):
    pct = 100 * used / total if total else 0
    return f"disk {path}: {pct:.0f}% full ({(total - used) / 2**30:.1f} GB free)" if pct >= max_pct else None


def check_disk(path, max_pct):
    try:
        u = shutil.disk_usage(path)
    except OSError as e:
        return f"disk {path}: cannot read ({e.strerror})"
    return disk_problem(path, u.used, u.total, max_pct)


def check_units():
    try:
        out = subprocess.run(["systemctl", "--failed", "--plain", "--no-legend", "--no-pager"],
                             capture_output=True, text=True, timeout=20).stdout
    except (OSError, subprocess.TimeoutExpired):
        return ["units: systemctl not available"]
    return [f"unit {line.split()[0]}: failed" for line in out.splitlines() if line.strip()]


def check_url(url, timeout=15):
    try:
        req = urllib.request.Request(url, headers={"User-Agent": f"certwatch/{VERSION}"})
        with urllib.request.urlopen(req, timeout=timeout) as r:
            code = r.status
    except urllib.error.HTTPError as e:
        code = e.code
    except (OSError, ValueError) as e:
        return f"url {url}: no answer ({e.__class__.__name__})"
    return f"url {url}: HTTP {code}" if code >= 400 else None


def diff(old, new):
    """What to say this run: new problems, fixed ones. Empty list = stay quiet.
    Matched by what was checked ("disk /"), so 76% -> 77% full is not a new alert."""
    key = lambda p: p.split(":")[0] if not p.startswith("url ") else p.rsplit(":", 1)[0]
    ok, nk = {key(p) for p in old}, {key(p) for p in new}
    msg = [f"PROBLEM {p}" for p in sorted(new) if key(p) not in ok]
    msg += [f"OK again: {k}" for k in sorted(ok - nk)]
    if ok - nk and not new:
        msg.append("All checks OK.")
    return msg


def send(text, ntfy=None, webhook=None):
    if ntfy:
        urllib.request.urlopen(urllib.request.Request(ntfy, data=text.encode(), headers={
            "Title": f"certwatch on {socket.gethostname()}", "Priority": "high"}), timeout=15)
    if webhook:  # "text" is read by Slack, "content" by Discord
        body = json.dumps({"text": text, "content": text[:1900]}).encode()
        urllib.request.urlopen(urllib.request.Request(webhook, data=body, headers={
            "Content-Type": "application/json"}), timeout=15)


def load_config(path):
    """INI-ish: key = value lines; cert/disk/url may repeat. Lines starting with # are skipped."""
    args = []
    for line in open(path, encoding="utf-8"):
        line = line.strip()
        if not line or line.startswith(("#", ";", "[")):
            continue
        k, _, v = line.partition("=")
        k, v = k.strip().replace("_", "-"), v.strip()
        args += [f"--{k}"] if v.lower() in ("yes", "true", "on") else [f"--{k}", v]
    return args


def self_test():
    assert disk_problem("/", 95, 100, 90).startswith("disk /: 95%")
    assert disk_problem("/", 50, 100, 90) is None
    assert diff([], ["a"]) == ["PROBLEM a"]
    assert diff(["a"], ["a"]) == []
    assert diff(["a"], []) == ["OK again: a", "All checks OK."]
    assert diff(["a", "b"], ["b", "c"]) == ["PROBLEM c", "OK again: a"]
    assert diff(["disk /: 90% full"], ["disk /: 91% full"]) == []
    assert diff(["url https://x/a: HTTP 500"], ["url https://x/a: no answer (URLError)"]) == []
    assert load_config.__doc__
    import tempfile
    with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as f:
        f.write("# x\ncert = a.com\ncert = b.com:8443\nunits = yes\nwarn_days = 21\n")
    assert load_config(f.name) == ["--cert", "a.com", "--cert", "b.com:8443", "--units", "--warn-days", "21"]
    os.unlink(f.name)
    print("certwatch self-test: OK")


def main(argv=None):
    p = argparse.ArgumentParser(description="One-file alerts for certs, disk, systemd units and URLs.")
    p.add_argument("--config", help="file with one option per line (key = value)")
    p.add_argument("--cert", action="append", default=[], metavar="HOST[:PORT]")
    p.add_argument("--warn-days", type=int, default=14, help="alert when a cert has fewer days left (14)")
    p.add_argument("--disk", action="append", default=[], metavar="PATH")
    p.add_argument("--disk-max", type=float, default=90, help="alert at this percent full (90)")
    p.add_argument("--units", action="store_true", help="alert on failed systemd units")
    p.add_argument("--url", action="append", default=[], help="alert when it does not answer below 400")
    p.add_argument("--ntfy", help="ntfy topic URL, e.g. https://ntfy.sh/your-secret-topic")
    p.add_argument("--webhook", help="Slack or Discord incoming webhook URL")
    p.add_argument("--state", default=os.path.expanduser("~/.cache/certwatch.json"))
    p.add_argument("--always", action="store_true", help="print the full status every run")
    p.add_argument("--self-test", action="store_true")
    p.add_argument("--version", action="version", version=VERSION)
    a = p.parse_args(argv)
    if a.self_test:
        return self_test()
    if a.config:
        a = p.parse_args(load_config(a.config) + (argv if argv is not None else sys.argv[1:]))
    if not (a.cert or a.disk or a.units or a.url):
        p.error("nothing to check: add --cert, --disk, --units or --url")

    problems = [check_cert(c, a.warn_days) for c in a.cert] + [check_disk(d, a.disk_max) for d in a.disk]
    problems += check_units() if a.units else []
    problems += [check_url(u) for u in a.url]
    problems = [x for x in problems if x]

    try:
        old = json.load(open(a.state)).get("problems", [])
    except (OSError, ValueError):
        old = []
    msg = diff(old, problems)
    if a.always and not msg:
        msg = [f"PROBLEM {x}" for x in problems] or ["All checks OK."]
    if msg:
        text = "\n".join(msg)
        print(text)
        try:
            send(text, a.ntfy, a.webhook)
        except OSError as e:  # keep the old state so the alert is tried again next run
            print(f"certwatch: could not send alert ({e})", file=sys.stderr)
            return 2
    os.makedirs(os.path.dirname(a.state) or ".", exist_ok=True)
    with open(a.state + ".tmp", "w") as f:
        json.dump({"problems": problems, "checked": int(time.time())}, f)
    os.replace(a.state + ".tmp", a.state)
    return 1 if problems else 0


if __name__ == "__main__":
    sys.exit(main())