mars: add generic Gitea webhook relay

2026-08-23 01:45:18 +00:00
parent 3c1f3e5fc3
commit 806cec77e8
7 changed files with 333 additions and 0 deletions
@@ -0,0 +1,175 @@
+#!/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-pr-comments",
+)
+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()