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 base64
import binascii
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.1.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
_last_payloads = {}           # source_id -> last received payload (for the template builder)
_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, image=None):
    """Send a message to a channel via the bot API. Returns (ok, detail).

    image may be None, {"mode": "url", "url": ...} for a Discord-side embed,
    or {"mode": "bytes", "data": ..., "mime": ..., "filename": ...} to upload
    the picture as an attachment (needed for LAN URLs and base64 payloads).
    """
    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": []}}
    files = None
    if image and image.get("mode") == "url":
        body["embeds"] = [{"image": {"url": image["url"]}}]
    elif image and image.get("mode") == "bytes":
        fname = image.get("filename", "image.jpg")
        body["embeds"] = [{"image": {"url": f"attachment://{fname}"}}]
        files = {"files[0]": (fname, image["data"], image.get("mime", "image/jpeg"))}

    for attempt in (1, 2):
        try:
            if files:
                r = requests.post(url, data={"payload_json": json.dumps(body)},
                                  files=files, headers=headers, timeout=30)
            else:
                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 with image" if image else "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):
    def shorten(v):
        if isinstance(v, str) and len(v) > 200:
            return v[:200] + f"… ({len(v)} chars)"
        if isinstance(v, dict):
            return {k: shorten(x) for k, x in v.items()}
        if isinstance(v, list):
            return [shorten(x) for x in v[:20]]
        return v
    pretty = json.dumps(shorten(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


# ---------------------------------------------------------------------------
# Image embedding
# ---------------------------------------------------------------------------

IMAGE_MAX_BYTES = 8 * 1024 * 1024  # Discord's default upload cap

_MAGIC = [
    (b"\xff\xd8\xff", "image/jpeg", "jpg"),
    (b"\x89PNG", "image/png", "png"),
    (b"GIF8", "image/gif", "gif"),
    (b"RIFF", "image/webp", "webp"),
]


def _sniff(data):
    for magic, mime, ext in _MAGIC:
        if data.startswith(magic):
            return mime, ext
    return None, None


def resolve_image(source, payload):
    """Resolve a source's image_path against the payload.

    Returns an image dict for discord_send, or None. Modes:
      embed  - hand Discord the URL (only works for internet-reachable URLs)
      attach - Dis2Hook fetches the URL itself (works on the LAN) or decodes
               base64 image data, then uploads it as an attachment.
    """
    path = (source.get("image_path") or "").strip().strip("{}")
    if not path:
        return None
    val = dig(payload, path)
    if val is None or not isinstance(val, str) or not val.strip():
        return None
    val = val.strip()
    mode = source.get("image_mode", "attach")

    if val.startswith("http://") or val.startswith("https://"):
        if mode == "embed":
            return {"mode": "url", "url": val}
        try:
            r = requests.get(val, timeout=20, stream=True)
            r.raise_for_status()
            data = r.raw.read(IMAGE_MAX_BYTES + 1, decode_content=True)
        except requests.RequestException as e:
            _log(f"Image fetch failed ({val[:80]}): {e}")
            return None
        if len(data) > IMAGE_MAX_BYTES:
            _log("Image skipped: larger than 8 MB.")
            return None
        mime, ext = _sniff(data)
        if not mime:
            _log("Image skipped: fetched data is not a recognized image format.")
            return None
        return {"mode": "bytes", "data": data, "mime": mime, "filename": f"image.{ext}"}

    # Not a URL: try base64 (raw, or a data: URI)
    b64 = val
    if b64.startswith("data:"):
        b64 = b64.split(",", 1)[-1]
    try:
        data = base64.b64decode(b64, validate=False)
    except (ValueError, binascii.Error):
        return None
    if not data or len(data) > IMAGE_MAX_BYTES:
        return None
    mime, ext = _sniff(data)
    if not mime:
        return None
    return {"mode": "bytes", "data": data, "mime": mime, "filename": f"image.{ext}"}


# ---------------------------------------------------------------------------
# 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")
        if s.get("image_mode") not in (None, "", "attach", "embed"):
            problems.append(f"{label}: image mode must be 'attach' or 'embed'")
    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/sample/<source_id>")
def api_sample(source_id):
    """Last received payload for a source — powers the drag-and-drop
    template builder in the web UI."""
    err = require_admin()
    if err:
        return err
    with _lock:
        payload = _last_payloads.get(source_id)
    if payload is None:
        return jsonify({"ok": False, "payload": None})
    return jsonify({"ok": True, "payload": payload})


@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
    with _lock:
        sample = _last_payloads.get(source_id)
    used_real = sample is not None
    if sample is None:
        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)
    label = "last received event" if used_real else "sample payload"
    content = f"🧪 **Test** ({label}) — {source.get('name', source_id)}\n{content}"
    image = resolve_image(source, sample)
    ok, detail = discord_send(source.get("channel_id", ""), content, image=image)
    _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)
    with _lock:
        _last_payloads[source_id] = payload  # feeds the UI template builder

    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)

    image = resolve_image(source, payload)
    ok, detail = discord_send(source.get("channel_id", ""), content, image=image)
    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)