homelab/services/dev/gitea-hermes-webhook-relay.py

#!/usr/bin/env python3
"""Normalize authenticated Gitea PR webhooks for Hermes Agent."""
from __future__ import annotations

import hashlib
import hmac
import json
import logging
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

LOG = logging.getLogger("gitea-hermes-webhook-relay")

LISTEN_HOST = os.environ.get("LISTEN_HOST", "0.0.0.0")
LISTEN_PORT = int(os.environ.get("LISTEN_PORT", "8645"))
HERMES_URL = os.environ.get(
    "HERMES_WEBHOOK_URL",
    "http://127.0.0.1:8644/webhooks/gitea-events",
)
MAX_BODY_BYTES = int(os.environ.get("MAX_BODY_BYTES", str(1024 * 1024)))
CREDENTIAL_NAME = os.environ.get("WEBHOOK_CREDENTIAL_NAME", "webhook_secret")


def load_secret() -> bytes:
    credentials_dir = os.environ.get("CREDENTIALS_DIRECTORY")
    if credentials_dir:
        path = Path(credentials_dir) / CREDENTIAL_NAME
        if path.is_file():
            return path.read_bytes().strip()
    value = os.environ.get("GITEA_HERMES_WEBHOOK_SECRET", "")
    if value:
        return value.encode()
    raise RuntimeError("webhook secret is not available")


def json_bytes(payload: dict) -> bytes:
    return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode()


def normalized_event(headers, payload: dict) -> str:
    event = headers.get("X-Gitea-Event-Type", "") or headers.get("X-Gitea-Event", "")
    if event == "issue_comment" and payload.get("is_pull") is True:
        return "pull_request_comment"
    return event or "unknown"


class Handler(BaseHTTPRequestHandler):
    server_version = "gitea-hermes-relay/1.0"

    def log_message(self, format: str, *args) -> None:
        LOG.info("%s - %s", self.address_string(), format % args)

    def send_json(self, status: int, payload: dict) -> None:
        body = json_bytes(payload)
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        if self.path == "/health":
            self.send_json(200, {"status": "ok", "service": "gitea-hermes-webhook-relay"})
        else:
            self.send_json(404, {"status": "not_found"})

    def do_POST(self) -> None:
        if self.path not in {"/gitea", "/"}:
            self.send_json(404, {"status": "not_found"})
            return

        try:
            content_length = int(self.headers.get("Content-Length", "-1"))
        except ValueError:
            self.send_json(400, {"status": "invalid_content_length"})
            return
        if content_length < 0 or content_length > MAX_BODY_BYTES:
            self.send_json(413, {"status": "payload_too_large"})
            return

        body = self.rfile.read(content_length)
        try:
            payload = json.loads(body)
        except json.JSONDecodeError:
            self.send_json(400, {"status": "invalid_json"})
            return
        if not isinstance(payload, dict):
            self.send_json(400, {"status": "invalid_payload"})
            return

        try:
            secret = load_secret()
        except RuntimeError as exc:
            LOG.error("%s", exc)
            self.send_json(503, {"status": "relay_not_ready"})
            return

        provided = self.headers.get("X-Gitea-Signature", "").strip()
        if provided.startswith("sha256="):
            provided = provided.removeprefix("sha256=")
        expected = hmac.new(secret, body, hashlib.sha256).hexdigest()
        if not provided or not hmac.compare_digest(provided, expected):
            LOG.warning("rejected webhook with invalid signature")
            self.send_json(401, {"status": "invalid_signature"})
            return

        event = normalized_event(self.headers, payload)
        normalized = dict(payload)
        normalized["event_type"] = event
        normalized["relay_source"] = "gitea"
        forwarded_body = json_bytes(normalized)
        forwarded_signature = hmac.new(
            secret, forwarded_body, hashlib.sha256
        ).hexdigest()
        delivery_id = self.headers.get("X-Gitea-Delivery", "")
        forwarded_headers = {
            "Content-Type": "application/json",
            "X-GitHub-Event": event,
            "X-Webhook-Signature": forwarded_signature,
        }
        if delivery_id:
            forwarded_headers["X-Request-ID"] = delivery_id

        request = Request(
            HERMES_URL,
            data=forwarded_body,
            headers=forwarded_headers,
            method="POST",
        )
        try:
            with urlopen(request, timeout=15) as response:
                response.read()
                status = response.status
        except HTTPError as exc:
            LOG.error("Hermes returned HTTP %s", exc.code)
            self.send_json(502, {"status": "hermes_error"})
            return
        except (URLError, TimeoutError, OSError) as exc:
            LOG.error("failed to forward webhook to Hermes: %s", exc)
            self.send_json(502, {"status": "hermes_unreachable"})
            return

        if status < 200 or status >= 300:
            self.send_json(502, {"status": "hermes_error", "http_status": status})
            return

        LOG.info(
            "forwarded %s action=%s delivery=%s",
            event,
            payload.get("action", ""),
            delivery_id or "none",
        )
        self.send_json(200, {"status": "forwarded", "event": event})


def main() -> None:
    logging.basicConfig(
        level=os.environ.get("LOG_LEVEL", "INFO"),
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    server = ThreadingHTTPServer((LISTEN_HOST, LISTEN_PORT), Handler)
    LOG.info("listening on %s:%s; forwarding to %s", LISTEN_HOST, LISTEN_PORT, HERMES_URL)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    main()