Dis2Hook/app.py

#!/usr/bin/env python3
"""
Dis2Hook - Webhook Relay to Discord Bot
=======================================
Receives webhooks from any source, formats them as Markdown, and relays
them to Discord channels through the Discord Bot API.

Runs inside an Alpine Linux LXC. Configuration lives in two files next
to this script:

  token.json   - secrets (bot token, admin key). Generated at install
                 time, never stored in the repo. chmod 600.
  config.json  - sources, heartbeat and status settings. Managed through
                 the web UI, safe to back up.

Repo: https://gogs.av2x.dev/av2x/Dis2Hook
"""

import atexit
import json
import os
import re
import secrets as pysecrets
import signal
import sys
import threading
import time
from collections import deque
from datetime import datetime, timezone

import requests
from flask import Flask, jsonify, request, send_from_directory

DIS2HOOK_VERSION = "1.0.0"

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
TOKEN_FILE = os.path.join(BASE_DIR, "token.json")
CONFIG_FILE = os.path.join(BASE_DIR, "config.json")

DISCORD_API = "https://discord.com/api/v10"
DISCORD_MSG_LIMIT = 2000

app = Flask(__name__, static_folder=None)

# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------

_lock = threading.RLock()
_started_at = time.time()
_stats = {"received": 0, "relayed": 0, "filtered": 0, "failed": 0}
_activity = deque(maxlen=50)  # ring buffer of recent events for the UI
_bot_identity = {"checked": 0, "ok": False, "name": None, "id": None}


def _log(msg):
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{ts}] {msg}", flush=True)


def _record(source_name, outcome, detail=""):
    with _lock:
        _activity.appendleft({
            "time": datetime.now(timezone.utc).isoformat(),
            "source": source_name,
            "outcome": outcome,   # relayed | filtered | failed | rejected
            "detail": detail[:200],
        })


# ---------------------------------------------------------------------------
# Config / secrets
# ---------------------------------------------------------------------------

DEFAULT_CONFIG = {
    "listen_host": "0.0.0.0",
    "listen_port": 8823,
    "heartbeat": {
        "enabled": False,
        "channel_id": "",
        "interval_minutes": 60,
    },
    "status": {
        "enabled": False,
        "channel_id": "",
    },
    "sources": [],
}


def load_tokens():
    if not os.path.exists(TOKEN_FILE):
        _log(f"FATAL: {TOKEN_FILE} not found. Run the installer to generate it.")
        sys.exit(1)
    with open(TOKEN_FILE, "r", encoding="utf-8") as f:
        data = json.load(f)
    if not data.get("bot_token") or not data.get("admin_key"):
        _log("FATAL: token.json must contain 'bot_token' and 'admin_key'.")
        sys.exit(1)
    return data


def load_config():
    if not os.path.exists(CONFIG_FILE):
        save_config(DEFAULT_CONFIG)
        return json.loads(json.dumps(DEFAULT_CONFIG))
    with open(CONFIG_FILE, "r", encoding="utf-8") as f:
        cfg = json.load(f)
    # Fill any missing keys so older configs keep working after updates.
    merged = json.loads(json.dumps(DEFAULT_CONFIG))
    merged.update({k: v for k, v in cfg.items() if k in ("listen_host", "listen_port", "sources")})
    for section in ("heartbeat", "status"):
        merged[section].update(cfg.get(section, {}))
    return merged


def save_config(cfg):
    tmp = CONFIG_FILE + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(cfg, f, indent=2)
    os.replace(tmp, CONFIG_FILE)


TOKENS = load_tokens()
CONFIG = load_config()


def get_source(source_id):
    with _lock:
        for s in CONFIG.get("sources", []):
            if s.get("id") == source_id:
                return s
    return None


# ---------------------------------------------------------------------------
# Discord Bot API
# ---------------------------------------------------------------------------

def discord_send(channel_id, content):
    """Send a message to a channel via the bot API. Returns (ok, detail)."""
    if not channel_id:
        return False, "no channel configured"
    if len(content) > DISCORD_MSG_LIMIT:
        content = content[: DISCORD_MSG_LIMIT - 25] + "\n*(message truncated)*"
    url = f"{DISCORD_API}/channels/{channel_id}/messages"
    headers = {
        "Authorization": f"Bot {TOKENS['bot_token']}",
        "User-Agent": f"Dis2Hook/{DIS2HOOK_VERSION}",
    }
    body = {"content": content, "allowed_mentions": {"parse": []}}
    for attempt in (1, 2):
        try:
            r = requests.post(url, json=body, headers=headers, timeout=15)
        except requests.RequestException as e:
            return False, f"network error: {e}"
        if r.status_code == 429 and attempt == 1:
            try:
                wait = float(r.json().get("retry_after", 1.0))
            except Exception:
                wait = 1.0
            time.sleep(min(wait, 5.0))
            continue
        if 200 <= r.status_code < 300:
            return True, "sent"
        return False, f"discord api {r.status_code}: {r.text[:150]}"
    return False, "rate limited"


def check_bot_identity(force=False):
    """Verify the bot token by asking Discord who we are. Cached 5 minutes."""
    with _lock:
        fresh = (time.time() - _bot_identity["checked"]) < 300
        if fresh and not force:
            return dict(_bot_identity)
    try:
        r = requests.get(
            f"{DISCORD_API}/users/@me",
            headers={"Authorization": f"Bot {TOKENS['bot_token']}",
                     "User-Agent": f"Dis2Hook/{DIS2HOOK_VERSION}"},
            timeout=10,
        )
        ok = r.status_code == 200
        data = r.json() if ok else {}
    except requests.RequestException:
        ok, data = False, {}
    with _lock:
        _bot_identity.update({
            "checked": time.time(),
            "ok": ok,
            "name": data.get("username"),
            "id": data.get("id"),
        })
        return dict(_bot_identity)


# ---------------------------------------------------------------------------
# Payload formatting
# ---------------------------------------------------------------------------

_PLACEHOLDER = re.compile(r"\{([a-zA-Z0-9_.\[\]-]+)\}")


def dig(payload, path):
    """Resolve a dotted path like 'commits[0].message' inside a payload."""
    cur = payload
    for part in re.split(r"\.", path):
        m = re.match(r"^([a-zA-Z0-9_-]*)((\[\d+\])*)$", part)
        if not m:
            return None
        key, indexes = m.group(1), m.group(2)
        if key:
            if not isinstance(cur, dict) or key not in cur:
                return None
            cur = cur[key]
        for idx in re.findall(r"\[(\d+)\]", indexes or ""):
            i = int(idx)
            if not isinstance(cur, list) or i >= len(cur):
                return None
            cur = cur[i]
    return cur


def render_template(template, payload):
    def sub(m):
        val = dig(payload, m.group(1))
        if val is None:
            return ""
        if isinstance(val, (dict, list)):
            return json.dumps(val, indent=2)[:500]
        return str(val)
    return _PLACEHOLDER.sub(sub, template)


def default_format(source_name, payload):
    pretty = json.dumps(payload, indent=2, ensure_ascii=False)
    if len(pretty) > 1700:
        pretty = pretty[:1700] + "\n…"
    return f"**{source_name}** received an event:\n```json\n{pretty}\n```"


def passes_filters(source, payload):
    """All filters must match (AND). No filters = everything passes."""
    for flt in source.get("filters", []):
        path = (flt.get("path") or "").strip()
        if not path:
            continue
        val = dig(payload, path)
        val_str = "" if val is None else str(val)
        mode = flt.get("mode", "equals")
        want = str(flt.get("value", ""))
        if mode == "equals" and val_str != want:
            return False
        if mode == "contains" and want not in val_str:
            return False
        if mode == "exists" and val is None:
            return False
    return True


# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------

def admin_authorized():
    supplied = request.headers.get("X-Admin-Key", "")
    return pysecrets.compare_digest(supplied, TOKENS["admin_key"])


def require_admin():
    if not admin_authorized():
        return jsonify({"error": "invalid admin key"}), 401
    return None


# ---------------------------------------------------------------------------
# Routes: web UI
# ---------------------------------------------------------------------------

@app.route("/")
@app.route("/index.html")
def ui():
    return send_from_directory(BASE_DIR, "index.html")


# ---------------------------------------------------------------------------
# Routes: admin API
# ---------------------------------------------------------------------------

@app.route("/api/login", methods=["POST"])
def api_login():
    """Lets the UI validate the admin key without exposing anything."""
    if admin_authorized():
        return jsonify({"ok": True})
    return jsonify({"ok": False}), 401


@app.route("/api/status")
def api_status():
    err = require_admin()
    if err:
        return err
    ident = check_bot_identity()
    with _lock:
        return jsonify({
            "version": DIS2HOOK_VERSION,
            "uptime_seconds": int(time.time() - _started_at),
            "stats": dict(_stats),
            "activity": list(_activity),
            "bot": {"ok": ident["ok"], "name": ident["name"], "id": ident["id"]},
        })


@app.route("/api/config", methods=["GET"])
def api_get_config():
    err = require_admin()
    if err:
        return err
    with _lock:
        return jsonify(CONFIG)


@app.route("/api/config", methods=["PUT"])
def api_put_config():
    err = require_admin()
    if err:
        return err
    data = request.get_json(silent=True)
    if not isinstance(data, dict):
        return jsonify({"error": "body must be a JSON object"}), 400

    problems = validate_config(data)
    if problems:
        return jsonify({"error": "; ".join(problems)}), 400

    with _lock:
        # listen_host / listen_port are only changeable by editing config.json
        # directly, so a UI save can never lock you out of the UI.
        data["listen_host"] = CONFIG["listen_host"]
        data["listen_port"] = CONFIG["listen_port"]
        CONFIG.clear()
        CONFIG.update(data)
        save_config(CONFIG)
    _log("Configuration saved via web UI.")
    return jsonify({"ok": True})


def validate_config(data):
    problems = []
    sources = data.get("sources")
    if not isinstance(sources, list):
        return ["'sources' must be a list"]
    seen_ids = set()
    for i, s in enumerate(sources):
        label = s.get("name") or f"source {i + 1}"
        sid = (s.get("id") or "").strip()
        if not re.match(r"^[a-z0-9][a-z0-9-]{1,63}$", sid):
            problems.append(f"{label}: endpoint id must be 2-64 chars of a-z, 0-9, '-'")
        if sid in seen_ids:
            problems.append(f"{label}: duplicate endpoint id '{sid}'")
        seen_ids.add(sid)
        if not (s.get("secret") or "").strip():
            problems.append(f"{label}: secret must not be empty")
        if not re.match(r"^\d{5,25}$", str(s.get("channel_id", ""))):
            problems.append(f"{label}: channel ID must be numeric")
    for section in ("heartbeat", "status"):
        sec = data.get(section, {})
        if sec.get("enabled") and not re.match(r"^\d{5,25}$", str(sec.get("channel_id", ""))):
            problems.append(f"{section}: channel ID must be numeric when enabled")
    hb = data.get("heartbeat", {})
    try:
        if hb.get("enabled") and not (1 <= int(hb.get("interval_minutes", 0)) <= 10080):
            problems.append("heartbeat: interval must be 1-10080 minutes")
    except (TypeError, ValueError):
        problems.append("heartbeat: interval must be a number")
    return problems


@app.route("/api/test/<source_id>", methods=["POST"])
def api_test(source_id):
    err = require_admin()
    if err:
        return err
    source = get_source(source_id)
    if not source:
        return jsonify({"error": "unknown source"}), 404
    sample = {
        "event": "dis2hook.test",
        "message": "Test relay from the Dis2Hook web UI",
        "time": datetime.now(timezone.utc).isoformat(),
    }
    template = (source.get("template") or "").strip()
    content = render_template(template, sample) if template else default_format(source.get("name", source_id), sample)
    content = f"🧪 **Test** — {source.get('name', source_id)}\n{content}"
    ok, detail = discord_send(source.get("channel_id", ""), content)
    _record(source.get("name", source_id), "relayed" if ok else "failed", f"test: {detail}")
    return (jsonify({"ok": True}) if ok
            else (jsonify({"ok": False, "error": detail}), 502))


# ---------------------------------------------------------------------------
# Routes: webhook receiver
# ---------------------------------------------------------------------------

def extract_payload(req):
    payload = req.get_json(silent=True)
    if payload is None and "payload" in req.form:
        # Some services (e.g. legacy GitHub/Slack style) post form-encoded JSON.
        try:
            payload = json.loads(req.form["payload"])
        except (ValueError, TypeError):
            payload = None
    if payload is None:
        raw = req.get_data(as_text=True)[:2000]
        payload = {"raw": raw}
    if not isinstance(payload, dict):
        payload = {"payload": payload}
    return payload


def supplied_secret(req):
    return (
        req.headers.get("X-Hook-Secret")
        or req.headers.get("Authorization", "").removeprefix("Bearer ").strip()
        or req.args.get("secret", "")
    )


@app.route("/hook/<source_id>", methods=["POST"])
def receive_hook(source_id):
    with _lock:
        _stats["received"] += 1
    source = get_source(source_id)
    if not source or not source.get("enabled", True):
        _record(source_id, "rejected", "unknown or disabled source")
        return jsonify({"error": "unknown source"}), 404

    if not pysecrets.compare_digest(supplied_secret(request), source.get("secret", "")):
        _record(source.get("name", source_id), "rejected", "bad secret")
        return jsonify({"error": "invalid secret"}), 401

    payload = extract_payload(request)

    if not passes_filters(source, payload):
        with _lock:
            _stats["filtered"] += 1
        _record(source.get("name", source_id), "filtered", "did not match filters")
        return jsonify({"ok": True, "relayed": False, "reason": "filtered"})

    template = (source.get("template") or "").strip()
    content = render_template(template, payload) if template else ""
    if not content.strip():
        content = default_format(source.get("name", source_id), payload)

    ok, detail = discord_send(source.get("channel_id", ""), content)
    with _lock:
        _stats["relayed" if ok else "failed"] += 1
    _record(source.get("name", source_id), "relayed" if ok else "failed", detail)
    if not ok:
        _log(f"Relay failed for '{source_id}': {detail}")
        return jsonify({"ok": False, "error": detail}), 502
    return jsonify({"ok": True, "relayed": True})


# ---------------------------------------------------------------------------
# Heartbeat + status updates
# ---------------------------------------------------------------------------

def _uptime_text():
    secs = int(time.time() - _started_at)
    d, rem = divmod(secs, 86400)
    h, rem = divmod(rem, 3600)
    m = rem // 60
    parts = ([f"{d}d"] if d else []) + ([f"{h}h"] if h or d else []) + [f"{m}m"]
    return " ".join(parts)


def heartbeat_loop():
    while True:
        with _lock:
            hb = dict(CONFIG.get("heartbeat", {}))
        interval = max(1, int(hb.get("interval_minutes", 60) or 60)) * 60
        time.sleep(interval)
        with _lock:
            hb = dict(CONFIG.get("heartbeat", {}))
            stats = dict(_stats)
        if hb.get("enabled") and hb.get("channel_id"):
            msg = (f"💓 **Dis2Hook heartbeat** — up {_uptime_text()} · "
                   f"{stats['relayed']} relayed · {stats['failed']} failed")
            ok, detail = discord_send(hb["channel_id"], msg)
            if not ok:
                _log(f"Heartbeat send failed: {detail}")


def send_status(text):
    with _lock:
        st = dict(CONFIG.get("status", {}))
    if st.get("enabled") and st.get("channel_id"):
        discord_send(st["channel_id"], text)


_shutdown_sent = False


def on_shutdown(*_args):
    global _shutdown_sent
    if not _shutdown_sent:
        _shutdown_sent = True
        send_status(f"🔴 **Dis2Hook** v{DIS2HOOK_VERSION} going offline.")
    if _args:  # invoked as a signal handler, not via atexit
        sys.exit(0)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    _log(f"Dis2Hook v{DIS2HOOK_VERSION} starting on "
         f"{CONFIG['listen_host']}:{CONFIG['listen_port']}")
    ident = check_bot_identity(force=True)
    if ident["ok"]:
        _log(f"Bot token OK — connected as {ident['name']} ({ident['id']})")
    else:
        _log("WARNING: bot token could not be verified with Discord. "
             "Relays will fail until token.json contains a valid token.")

    threading.Thread(target=heartbeat_loop, daemon=True).start()
    signal.signal(signal.SIGTERM, on_shutdown)
    signal.signal(signal.SIGINT, on_shutdown)
    atexit.register(on_shutdown)
    send_status(f"🟢 **Dis2Hook** v{DIS2HOOK_VERSION} online.")

    app.run(host=CONFIG["listen_host"], port=int(CONFIG["listen_port"]), threaded=True)