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.

The target Hermes route travels in the request path (POST /gitea/<route> ->
POST <base>/webhooks/<route>) rather than being configured here, so one relay
serves every subscription and adding a Hermes route means adding a Gitea hook
URL, nothing more.
"""
from __future__ import annotations

import hashlib
import hmac
import json
import logging
import os
import re
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"))
# The Hermes route is taken from the request path (POST /gitea/<route>), not
# baked in here, so one relay serves every subscription: a new Hermes route
# needs a new Gitea hook URL and nothing else. HERMES_WEBHOOK_BASE is the
# prefix the route name is appended to; DEFAULT_ROUTE serves the legacy bare
# /gitea and / paths.
HERMES_WEBHOOK_BASE = os.environ.get(
    "HERMES_WEBHOOK_BASE",
    "http://127.0.0.1:8644/webhooks",
).rstrip("/")
DEFAULT_ROUTE = os.environ.get("DEFAULT_ROUTE", "gitea-pr-comments")

# The route name is interpolated into an outbound URL, so it is validated
# strictly rather than sanitised: anything outside this charset is refused
# instead of being cleaned up. This is what stops POST /gitea/..%2fadmin (or
# any other traversal) from steering the relay at a different Hermes endpoint.
# Leading character must be alphanumeric, which also rejects "." and "..".
ROUTE_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}")
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


def route_from_path(path: str) -> str | None:
    """Map a request path to a Hermes route name, or None if it is not ours.

    /gitea/<route> -> <route>;  /gitea and / -> DEFAULT_ROUTE.
    The path is matched raw, never URL-decoded, so percent-encoded separators
    fail the charset check rather than surviving it.
    """
    path = path.split("?", 1)[0].split("#", 1)[0]
    if path in ("/", "/gitea"):
        return DEFAULT_ROUTE
    prefix = "/gitea/"
    if not path.startswith(prefix):
        return None
    route = path[len(prefix):].rstrip("/")
    if not ROUTE_RE.fullmatch(route):
        return None
    return route


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:
        route = route_from_path(self.path)
        if route is None:
            LOG.warning("rejected POST to unroutable path %r", self.path)
            self.send_json(404, {"status": "not_found"})
            return
        hermes_url = f"{HERMES_WEBHOOK_BASE}/{route}"

        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 to route=%s",
            gitea_event or gitea_event_type or "unknown",
            delivery_id or "none",
            route,
        )
        self.send_json(200, {"status": "forwarded", "route": route})


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/<route> (default route %s)",
        LISTEN_HOST, LISTEN_PORT, HERMES_WEBHOOK_BASE, DEFAULT_ROUTE,
    )
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    main()