#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# 40-probe — 42i-hc Drop-in: externes Blackbox-Probing (Ersatz fuer Uptime Kuma).
# hc-tags: checks, probe
# hc-ttl: 30
#
# Probt eine konfigurierte Target-Liste (http/ping/dns/smtp/port) und meldet je
# Target einen check (ok/fail bzw. warn je severity). Damit bildet hc die
# aktiven Kuma-Monitore nach — von einem Collector aus, ohne Historie/Alerting
# (Eskalation laeuft ueber Buzz' hc-Report + Ticket). OHNE Config ein No-op,
# also ueberall mitlieferbar; aktiv nur, wo eine probes.yaml liegt.
#
# Config (YAML), Default /etc/42i-hc/probes.yaml (Override HC_PROBES):
#   defaults: { timeout: 5, severity: fail }
#   probes:
#     - { name: app,    type: http, target: https://app.inkasso.live, expect: 200 }
#     - { name: KNX,    type: ping, target: 192.168.1.30 }
#     - { name: DNS,    type: dns,  target: 42i.org, resolver: 192.168.1.2 }
#     - { name: mail,   type: smtp, target: mx2.42i.org, port: 25 }
#     - { name: s99p1,  type: port, target: 192.168.1.112, port: 8007 }
#
# Cache macht der hc-Daemon (`# hc-ttl: 30`). Schema: 42i/intern#236. Von 42i-hc.

import concurrent.futures
import json
import os
import socket
import ssl
import subprocess
import sys
import urllib.request

CFG = os.environ.get("HC_PROBES") or "/etc/42i-hc/probes.yaml"

try:
    import yaml
except Exception:
    yaml = None


def load():
    """Laedt defaults + probes aus der YAML-Config; ([], {}) wenn keine da."""
    if not (yaml and os.path.isfile(CFG)):
        return [], {}
    try:
        with open(CFG, encoding="utf-8") as fh:
            d = yaml.safe_load(fh) or {}
        return d.get("probes") or [], d.get("defaults") or {}
    except Exception:
        return [], {}


def probe_http(p, timeout):
    """HTTP(S)-Probe: erreichbar + Statuscode wie erwartet (Default < 400)."""
    target = p["target"]
    expect = p.get("expect")
    ctx = ssl.create_default_context()
    if p.get("insecure"):
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
    req = urllib.request.Request(target, method="GET")
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
            code = r.status
    except urllib.error.HTTPError as e:
        code = e.code
    except Exception as e:
        return False, str(e)[:120]
    # ohne expliziten expect: erreichbar = HTTP-Antwort < 400 ODER 401/403
    # (Dienst laeuft, ist nur auth-geschuetzt — haeufig bei Geraeten/WebDAV).
    ok = (code == expect) if expect else (code < 400 or code in (401, 403))
    return ok, "HTTP %s" % code


def probe_ping(p, timeout):
    """ICMP-Ping (ein Paket)."""
    try:
        r = subprocess.run(["ping", "-c", "1", "-W", str(int(max(1, timeout))),
                           p["target"]], capture_output=True, timeout=timeout + 2)
        return r.returncode == 0, "ping" if r.returncode == 0 else "kein Echo"
    except Exception as e:
        return False, str(e)[:120]


def probe_port(p, timeout, default_port=None):
    """TCP-Connect auf target:port."""
    port = int(p.get("port") or default_port or 0)
    try:
        with socket.create_connection((p["target"], port), timeout=timeout):
            return True, "tcp/%d offen" % port
    except Exception as e:
        return False, "tcp/%d: %s" % (port, str(e)[:80])


def probe_smtp(p, timeout):
    """SMTP: connect + 220-Banner."""
    port = int(p.get("port") or 25)
    try:
        with socket.create_connection((p["target"], port), timeout=timeout) as s:
            s.settimeout(timeout)
            banner = s.recv(128).decode("utf-8", "replace")
        return banner.startswith("220"), banner.strip()[:80]
    except Exception as e:
        return False, str(e)[:120]


def _dns_query(name, resolver, timeout):
    """Minimaler A-Record-Query (stdlib, kein dig) gegen resolver:53 (UDP).

    Liefert (ok, antwortzahl-oder-fehler). ok = Antwort mit >=1 Answer-Record.
    """
    import struct
    tid = int.from_bytes(os.urandom(2), "big")
    header = struct.pack(">HHHHHH", tid, 0x0100, 1, 0, 0, 0)  # RD=1, 1 Frage
    q = b"".join(bytes([len(lbl)]) + lbl.encode() for lbl in name.split(".")) + b"\x00"
    q += struct.pack(">HH", 1, 1)  # QTYPE=A, QCLASS=IN
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.settimeout(timeout)
    try:
        s.sendto(header + q, (resolver, 53))
        data, _ = s.recvfrom(512)
    finally:
        s.close()
    rcode = data[3] & 0x0F
    ancount = struct.unpack(">H", data[6:8])[0]
    if rcode != 0:
        return False, "RCODE %d von %s" % (rcode, resolver)
    return ancount > 0, "%d Antwort(en)" % ancount


def probe_dns(p, timeout):
    """DNS: loest der (optional spezifische) Resolver den Namen auf?"""
    target = p["target"]
    resolver = p.get("resolver")
    if resolver:
        try:
            return _dns_query(target, resolver, timeout)
        except Exception as e:
            return False, "%s: %s" % (resolver, str(e)[:90])
    try:
        socket.getaddrinfo(target, None)
        return True, "aufgeloest"
    except Exception as e:
        return False, str(e)[:120]


PROBERS = {"http": probe_http, "ping": probe_ping, "dns": probe_dns,
           "smtp": probe_smtp, "port": probe_port}


def run_one(p, defaults):
    """Fuehrt eine Probe aus und liefert den check-Eintrag."""
    name = p.get("name", p.get("target", "?"))
    typ = p.get("type", "http")
    timeout = float(p.get("timeout") or defaults.get("timeout") or 5)
    sev = p.get("severity") or defaults.get("severity") or "fail"
    fn = PROBERS.get(typ)
    if not fn:
        return {"name": name, "status": "no-check", "message": "Typ %s unbekannt" % typ}
    try:
        ok, msg = fn(p, timeout)
    except Exception as e:
        ok, msg = False, str(e)[:120]
    out = {"name": name, "status": "ok" if ok else sev, "type": typ, "message": msg}
    tags = p.get("tags")
    if tags:
        out["tags"] = tags if isinstance(tags, list) else [tags]
    return out


def build():
    """Probt alle Targets parallel und gibt {"checks": [...]} zurueck."""
    probes, defaults = load()
    if not probes:
        return {"checks": []}
    with concurrent.futures.ThreadPoolExecutor(max_workers=min(20, len(probes))) as ex:
        checks = list(ex.map(lambda p: run_one(p, defaults), probes))
    return {"checks": checks}


def main():
    """Gibt {"checks": ...} als JSON aus (Cache macht der hc-Daemon via hc-ttl)."""
    sys.stdout.write(json.dumps(build(), ensure_ascii=False) + "\n")


if __name__ == "__main__":
    main()
