FabledGrimoire/src/scribe/routes/webhooks.py

"""Forge push webhook — drift flagging at the moment the repo moves (#2691).

The forge POSTs here on every push; changed paths are matched against recorded
snippet locations (through repo bindings) and matched verdicts get an
``invalidated_by`` marker that surfaces in the ``verification="attention"``
listing. This is what makes verification scale past tens of records: sessions
recheck what pushes flagged instead of sweeping everything.

Registering the webhook on the forge is per-instance setup (Settings → Config
→ Git Forge shows the endpoint and holds the secret) — the server never
self-registers on the forge.

Contract with the forge's delivery loop:
  - No secret configured → 404: the endpoint doesn't exist until an operator
    creates it. Bad signature → 401: that's a caller problem worth signaling.
  - A PROCESSING failure returns 200 with ``{"ok": false}`` and drops a
    WARNING + AppLog row (the #2663 canary pattern): repeated 5xx responses
    make forges mark deliveries failed and operators disable the hook, which
    would silently turn the feature off — the exact failure mode this
    milestone exists to prevent.
"""
from __future__ import annotations

import hashlib
import hmac
import logging
import traceback

from quart import Blueprint, jsonify, request

from scribe.config import Config
from scribe.services.repo_bindings import normalize_repo_key
from scribe.services.settings import get_admin_setting
from scribe.services.snippets import invalidate_for_push

logger = logging.getLogger(__name__)

webhooks_bp = Blueprint("webhooks", __name__, url_prefix="/api/webhooks")

FORGE_WEBHOOK_SECRET_KEY = "forge_webhook_secret"


def signature_ok(secret: str, body: bytes, signature: str) -> bool:
    """Validate Gitea's push signature: X-Gitea-Signature is the hex HMAC-SHA256
    of the raw body under the webhook secret. Constant-time compare."""
    if not secret or not signature:
        return False
    expected = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature.strip().lower())


def push_facts(payload: dict) -> tuple[str, list[str], list[str], str]:
    """(repo identity, changed paths, removed paths, head commit) from a Gitea
    push payload. Tolerant: absent fields read as empty, never raise."""
    repo = payload.get("repository") or {}
    raw_repo = repo.get("clone_url") or repo.get("html_url") or repo.get("full_name") or ""
    changed: list[str] = []
    removed: list[str] = []
    for commit in payload.get("commits") or []:
        changed.extend(commit.get("added") or [])
        changed.extend(commit.get("modified") or [])
        removed.extend(commit.get("removed") or [])
    # De-dup while keeping order stable for logs.
    changed = list(dict.fromkeys(changed))
    removed = list(dict.fromkeys(removed))
    return raw_repo, changed, removed, str(payload.get("after") or "")


@webhooks_bp.route("/forge", methods=["POST"])
async def forge_push():
    secret = await get_admin_setting(FORGE_WEBHOOK_SECRET_KEY, "") or Config.FORGE_WEBHOOK_SECRET
    if not secret:
        # Not "forbidden" — the endpoint is not a thing on this instance.
        return jsonify({"error": "Not found"}), 404

    body = await request.get_data()
    if not signature_ok(secret, body, request.headers.get("X-Gitea-Signature", "")):
        return jsonify({"error": "Invalid signature"}), 401

    try:
        payload = await request.get_json(force=True) or {}
        raw_repo, changed, removed, head = push_facts(payload)
        repo_key = normalize_repo_key(raw_repo)
        if not repo_key:
            return jsonify({"ok": True, "flagged": 0, "reason": "no repository in payload"})
        flagged = await invalidate_for_push(repo_key, changed, removed, head)
        if flagged:
            logger.info(
                "forge push %s flagged %d snippet(s) for recheck", head[:12], flagged
            )
        return jsonify({"ok": True, "flagged": flagged})
    except Exception:
        logger.warning("forge webhook processing failed", exc_info=True)
        try:
            from scribe.services.logging import log_error

            await log_error(
                endpoint="webhooks/forge",
                error_type="forge_webhook_failed",
                error_message="push received but drift flagging failed — "
                "snippets touched by this push were not marked for recheck",
                traceback=traceback.format_exc(),
            )
        except Exception:
            logger.debug("forge webhook canary write failed", exc_info=True)
        # 200 on purpose — see the module docstring's delivery contract.
        return jsonify({"ok": False})