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

#!/usr/bin/env python3
"""Relay authenticated Gitea webhook requests to 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()


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:
            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

        # Keep the incoming body unchanged. Event interpretation and policy
        # belong to Hermes, not to this transport service.
        forwarded_body = body
        forwarded_signature = hmac.new(
            secret, forwarded_body, hashlib.sha256
        ).hexdigest()
        gitea_event = self.headers.get("X-Gitea-Event", "")
        gitea_event_type = self.headers.get("X-Gitea-Event-Type", "")
        delivery_id = self.headers.get("X-Gitea-Delivery", "")
        forwarded_headers = {
            "Content-Type": "application/json",
            "X-Webhook-Signature": forwarded_signature,
        }
        if gitea_event:
            forwarded_headers["X-Gitea-Event"] = gitea_event
        if gitea_event_type:
            forwarded_headers["X-Gitea-Event-Type"] = gitea_event_type
        if delivery_id:
            forwarded_headers["X-Request-ID"] = delivery_id
            forwarded_headers["X-Gitea-Delivery"] = 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 Gitea event=%s delivery=%s",
            gitea_event or gitea_event_type or "unknown",
            delivery_id or "none",
        )
        self.send_json(200, {"status": "forwarded"})


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()