svn-git-server/svn_mirror/webhook.py

"""Gitea webhook receiver – HTTP server that triggers Git→SVN sync on push."""

import hashlib
import hmac
import json
import logging
from http.server import HTTPServer, BaseHTTPRequestHandler
from typing import Dict, Optional

from .config import MirrorConfig
from .git_to_svn import sync_git_to_svn
from .gitea import fetch_from_gitea
from .mirror import Mirror

logger = logging.getLogger(__name__)


class _Handler(BaseHTTPRequestHandler):
    """HTTP request handler for Gitea webhooks."""

    # Set by the caller before starting the server
    mirrors: Dict[str, Mirror] = {}
    lookup: Dict[str, str] = {}  # "owner/repo" → mirror_id
    secrets: Dict[str, Optional[str]] = {}  # mirror_id → secret (or None)

    def log_request(self, code="-", size="-"):
        logger.debug("Webhook: %s %s → %s", self.command, self.path, code)

    def _send_json(self, code: int, body: dict):
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(body).encode())

    def _verify_signature(self, body: bytes, secret: Optional[str]) -> bool:
        """Verify X-Gitea-Signature HMAC-SHA256 if a secret is configured."""
        if secret is None:
            return True  # no secret configured – trust all
        sig = self.headers.get("X-Gitea-Signature", "")
        expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
        return hmac.compare_digest(sig, expected)

    def do_GET(self):
        if self.path == "/health":
            self._send_json(200, {"status": "ok"})
        else:
            self._send_json(404, {"error": "not found"})

    def do_POST(self):
        if self.path != "/webhook":
            self._send_json(404, {"error": "not found"})
            return

        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length)

        # Determine which mirror this webhook is for
        payload = json.loads(body)
        owner = (
            payload.get("repository", {}).get("owner", {}).get("login")
            or payload.get("repository", {}).get("owner", {}).get("username")
        )
        repo = payload.get("repository", {}).get("name")
        if not owner or not repo:
            self._send_json(400, {"error": "missing repository owner/name"})
            return

        key = f"{owner}/{repo}"
        mirror_id = self.lookup.get(key)
        if mirror_id is None:
            logger.warning("No mirror configured for %s", key)
            self._send_json(404, {"error": f"no mirror for {key}"})
            return

        # Verify signature
        secret = self.secrets.get(mirror_id)
        if not self._verify_signature(body, secret):
            logger.warning("Invalid webhook signature for %s", mirror_id)
            self._send_json(403, {"error": "invalid signature"})
            return

        ref = payload.get("ref", "")
        logger.info(
            "Webhook: %s pushed to %s (%s)",
            payload.get("pusher", {}).get("login", "?"), key, ref,
        )

        mirror = self.mirrors.get(mirror_id)
        if mirror is None:
            self._send_json(500, {"error": f"mirror {mirror_id} not loaded"})
            return

        try:
            fetch_from_gitea(mirror, ref=ref)
            count = sync_git_to_svn(mirror)
            self._send_json(200, {
                "status": "ok",
                "mirror": mirror_id,
                "git_to_svn_count": count,
            })
        except Exception as e:
            logger.exception("Webhook processing failed for %s", mirror_id)
            self._send_json(500, {"error": str(e)})

    do_PUT = do_POST  # tolerate PUT for convenience


def _build_lookup(mirrors: Dict[str, Mirror]) -> Dict[str, str]:
    """Build ``owner/repo → mirror_id`` lookup dict."""
    lookup: Dict[str, str] = {}
    for mid, m in mirrors.items():
        cfg = m.config.gitea
        key = f"{cfg.owner}/{cfg.repo}"
        lookup[key] = mid
    return lookup


def build_handler_class(mirrors: Dict[str, Mirror]) -> type:
    """Return a handler class pre-configured with mirror references."""
    lookup = _build_lookup(mirrors)
    secrets = {
        mid: m.config.gitea.webhook_secret
        for mid, m in mirrors.items()
    }
    return type("ConfiguredHandler", (_Handler,), {
        "mirrors": mirrors,
        "lookup": lookup,
        "secrets": secrets,
    })


def run_webhook_server(mirrors: Dict[str, Mirror], host: str = "0.0.0.0",
                       port: int = 8080):
    """Start the webhook HTTP server (blocks until KeyboardInterrupt)."""
    handler_cls = build_handler_class(mirrors)
    server = HTTPServer((host, port), handler_cls)
    logger.info("Webhook server listening on %s:%d", host, port)
    logger.info("  POST /webhook  – Gitea push events")
    logger.info("  GET  /health   – health check")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        logger.info("Webhook server stopped")
        server.server_close()