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

#!/usr/bin/env python3
"""Relay authenticated Gitea webhook requests to Hermes Agent.

This service exists for exactly one reason: Hermes derives the event name it
matches a subscription's `events` filter against from `X-GitHub-Event` /
`X-GitLab-Event`, falling back to the payload's `event_type`/`type` keys and
then to the literal string "unknown" (gateway/platforms/webhook.py). Gitea
never sends any of those — its event name rides `X-Gitea-Event`, and its
payloads carry no `event_type`/`type` key — so a Gitea webhook pointed
straight at Hermes authenticates fine but arrives as "unknown" forever, which
makes `hermes webhook subscribe --events ...` unable to select anything.

Everything else about a Gitea delivery already speaks Hermes natively:
Gitea's addDefaultHeaders() signs EVERY webhook type with
`X-Hub-Signature-256: sha256=<hmac-sha256(body)>`, byte-identical to GitHub's
scheme, and Hermes accepts that header on any route with no per-route
provider gating. So the body and the signature are forwarded untouched — this
process re-signs nothing and rewrites no payload. It copies one header.

It still verifies the signature itself rather than forwarding blindly, so an
unauthenticated caller that reaches this port never reaches the 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:
    """Read the shared secret, preferring systemd's credential store.

    Both sources are stripped: the sops secret file usually ends in a newline,
    while the value Gitea signs with comes from `$(cat ...)` in the
    provisioning unit, which drops trailing newlines. Stripping here is what
    keeps those two in agreement.
    """
    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.strip().encode()
    raise RuntimeError("webhook secret is not available")


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


def signature_matches(secret: bytes, body: bytes, headers) -> bool:
    """Check the body against whichever signature header Gitea supplied.

    Gitea sends both on every delivery: `X-Hub-Signature-256` (GitHub format,
    `sha256=` prefixed) and `X-Gitea-Signature` (bare lowercase hex). Either is
    accepted so the relay keeps working if one is ever dropped upstream.
    """
    expected = hmac.new(secret, body, hashlib.sha256).hexdigest()
    for header in ("X-Hub-Signature-256", "X-Gitea-Signature"):
        provided = headers.get(header, "").strip()
        if not provided:
            continue
        if provided.startswith("sha256="):
            provided = provided.removeprefix("sha256=")
        if hmac.compare_digest(provided, expected):
            return True
    return False


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

        raw_length = self.headers.get("Content-Length")
        if raw_length is None:
            self.send_json(411, {"status": "length_required"})
            return
        try:
            content_length = int(raw_length)
        except ValueError:
            self.send_json(400, {"status": "invalid_content_length"})
            return
        if content_length < 0:
            self.send_json(400, {"status": "invalid_content_length"})
            return
        if 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

        if not signature_matches(secret, body, self.headers):
            LOG.warning("rejected webhook with invalid signature")
            self.send_json(401, {"status": "invalid_signature"})
            return

        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", "")
        hub_signature = self.headers.get("X-Hub-Signature-256", "")

        # The body is forwarded byte-for-byte, so Gitea's own signature stays
        # valid — nothing is re-signed here. If Gitea ever stops sending the
        # GitHub-format header, sign the unchanged body ourselves so Hermes
        # still has something its GitHub branch can verify.
        if not hub_signature:
            hub_signature = "sha256=" + hmac.new(secret, body, hashlib.sha256).hexdigest()

        forwarded_headers = {
            "Content-Type": "application/json",
            "X-Hub-Signature-256": hub_signature,
        }
        # The one transformation this service performs. Gitea's event names are
        # passed through verbatim rather than mapped onto GitHub's vocabulary:
        # Hermes only string-matches them against the subscription's `events`
        # list, and Gitea has events (pull_request_comment, pull_request_sync,
        # pull_request_review_approved, ...) with no GitHub equivalent to map to.
        if gitea_event:
            forwarded_headers["X-GitHub-Event"] = 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=body,
            headers=forwarded_headers,
            method="POST",
        )
        try:
            with urlopen(request, timeout=15) as response:
                response.read()
        except HTTPError as exc:
            LOG.error("Hermes returned HTTP %s", exc.code)
            self.send_json(502, {"status": "hermes_error", "http_status": exc.code})
            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

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