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

"""Integration test for gitea-hermes-webhook-relay.py.

Spawns the real relay as a subprocess against a stub Hermes and drives it over
real HTTP. Run it directly:  python3 services/dev/gitea-hermes-webhook-relay-test.py

The assertion that matters is `X-GitHub-Event injected`: Hermes derives the
event name it matches `--events` against from X-GitHub-Event, and Gitea only
ever sends X-Gitea-Event. If that copy regresses, every delivery silently
becomes event "unknown" and no Hermes-side event selection can work.
"""
import hashlib, hmac, json, os, pathlib, subprocess, sys, threading, time, urllib.request, urllib.error
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = b"s3cr3t-test-value"
RELAY_PORT, HERMES_PORT = 18645, 18644
received = []

class Hermes(BaseHTTPRequestHandler):
    def do_POST(self):
        n = int(self.headers.get("Content-Length", 0))
        # lower-cased keys: HTTP headers are case-insensitive and urllib
        # normalises them with .title() on the wire ("X-GitHub-Event" leaves
        # as "X-Github-Event"). aiohttp reads them into a case-insensitive
        # CIMultiDict, so matching case-insensitively here is the correct
        # assertion, not a workaround.
        received.append({"path": self.path, "body": self.rfile.read(n),
                         "headers": {k.lower(): v for k, v in self.headers.items()}})
        self.send_response(200); self.send_header("Content-Length", "2")
        self.end_headers(); self.wfile.write(b"ok")
    def log_message(self, *a): pass

hermes = HTTPServer(("127.0.0.1", HERMES_PORT), Hermes)
threading.Thread(target=hermes.serve_forever, daemon=True).start()

import tempfile
creds = os.path.join(tempfile.mkdtemp(), "creds"); os.makedirs(creds, exist_ok=True)
# trailing newline on purpose: mimics a sops secret file
open(os.path.join(creds, "webhook_secret"), "wb").write(SECRET + b"\n")

env = {**os.environ, "CREDENTIALS_DIRECTORY": creds, "LISTEN_HOST": "127.0.0.1",
       "LISTEN_PORT": str(RELAY_PORT),
       "HERMES_WEBHOOK_BASE": f"http://127.0.0.1:{HERMES_PORT}/webhooks",
       "DEFAULT_ROUTE": "gitea-pr-comments"}
RELAY = str(pathlib.Path(__file__).with_name('gitea-hermes-webhook-relay.py'))
relay = subprocess.Popen([sys.executable, RELAY],
                         env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

for _ in range(50):
    try:
        urllib.request.urlopen(f"http://127.0.0.1:{RELAY_PORT}/health", timeout=1); break
    except Exception: time.sleep(0.1)

def post(body, headers, path="/gitea"):
    req = urllib.request.Request(f"http://127.0.0.1:{RELAY_PORT}{path}", data=body,
                                 headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=5) as r: return r.status, json.load(r)
    except urllib.error.HTTPError as e: return e.code, json.load(e)

fails = []
def check(name, cond, detail=""):
    print(("PASS " if cond else "FAIL ") + name + ("" if cond else f"  <- {detail}"))
    if not cond: fails.append(name)

payload = json.dumps({"action": "opened", "number": 7}).encode()
sig = hmac.new(SECRET, payload, hashlib.sha256).hexdigest()

# 1. health
with urllib.request.urlopen(f"http://127.0.0.1:{RELAY_PORT}/health") as r:
    check("health endpoint", json.load(r)["status"] == "ok")

# 2. happy path with BOTH gitea headers (what Gitea really sends)
received.clear()
st, resp = post(payload, {"Content-Type": "application/json",
                          "X-Gitea-Event": "pull_request_comment",
                          "X-Gitea-Event-Type": "pull_request_comment",
                          "X-Gitea-Delivery": "abc-123",
                          "X-Gitea-Signature": sig,
                          "X-Hub-Signature-256": "sha256=" + sig})
check("valid delivery accepted", st == 200, f"got {st} {resp}")
check("forwarded to hermes", len(received) == 1)
fwd = received[0]
check("body forwarded byte-identical", fwd["body"] == payload)
check("X-GitHub-Event injected (THE fix)",
      fwd["headers"].get("x-github-event") == "pull_request_comment",
      f"got {fwd['headers'].get('X-GitHub-Event')!r}")
check("X-Hub-Signature-256 forwarded unchanged",
      fwd["headers"].get("x-hub-signature-256") == "sha256=" + sig)
check("signature still valid over forwarded body",
      hmac.compare_digest(
          fwd["headers"]["x-hub-signature-256"].removeprefix("sha256="),
          hmac.new(SECRET, fwd["body"], hashlib.sha256).hexdigest()))
check("delivery id propagated", fwd["headers"].get("x-request-id") == "abc-123")
check("bare /gitea uses DEFAULT_ROUTE", fwd["path"] == "/webhooks/gitea-pr-comments",
      f"got {fwd['path']}")

# 3. gitea-only signature header (no X-Hub-Signature-256)
received.clear()
st, _ = post(payload, {"Content-Type": "application/json", "X-Gitea-Event": "push",
                       "X-Gitea-Signature": sig})
check("bare X-Gitea-Signature accepted", st == 200)
check("relay signs when hub header absent",
      received and received[0]["headers"].get("x-hub-signature-256") == "sha256=" + sig)

# 4. rejections
received.clear()
st, _ = post(payload, {"Content-Type": "application/json", "X-Hub-Signature-256": "sha256=" + "0"*64})
check("bad signature -> 401", st == 401)
st, _ = post(payload, {"Content-Type": "application/json"})
check("missing signature -> 401", st == 401)
st, _ = post(payload + b"x", {"Content-Type": "application/json", "X-Hub-Signature-256": "sha256=" + sig})
check("tampered body -> 401", st == 401)
check("nothing leaked to hermes on rejection", len(received) == 0, f"{len(received)} forwarded")

# 5. oversize
big = b"x" * 200
st, _ = post(big, {"Content-Type": "application/json", "MAX": "1",
                   "X-Hub-Signature-256": "sha256=" + hmac.new(SECRET, big, hashlib.sha256).hexdigest()})
check("normal-size body still ok", st == 200)

# 6. unknown path
st, _ = post(payload, {"X-Hub-Signature-256": "sha256=" + sig})
check("POST /gitea ok baseline", st == 200)

# 7. route travels in the path: /gitea/<route> -> /webhooks/<route>
hdrs = {"Content-Type": "application/json", "X-Gitea-Event": "push",
        "X-Hub-Signature-256": "sha256=" + sig}
for route in ("gitea-pr-comments", "some-other_route.v2", "a"):
    received.clear()
    st, resp = post(payload, hdrs, path=f"/gitea/{route}")
    check(f"route {route!r} forwarded to /webhooks/{route}",
          st == 200 and received and received[0]["path"] == f"/webhooks/{route}",
          f"status={st} path={received[0]['path'] if received else None}")
    check(f"route {route!r} echoed in response", resp.get("route") == route, f"got {resp}")

# 8. route validation — these must never reach Hermes at all
for bad, label in [
    ("../admin",            "parent-dir traversal"),
    ("..%2fadmin",          "encoded traversal"),
    ("..",                  "bare .."),
    (".",                   "bare ."),
    (".hidden",             "leading dot"),
    ("-dash",               "leading dash"),
    ("route%20name",        "percent-encoded space"),
    ("route/extra",         "embedded slash"),
    ("x" * 65,              "over length limit"),
]:
    received.clear()
    st, _ = post(payload, hdrs, path=f"/gitea/{bad}")
    check(f"rejects {label}", st == 404 and not received,
          f"status={st} forwarded={len(received)}")

received.clear()
st, _ = post(payload, hdrs, path="/webhooks/gitea-pr-comments")
check("rejects non-/gitea prefix", st == 404 and not received, f"status={st}")

relay.terminate(); relay.wait(timeout=5); hermes.shutdown()
print()
print(f"{'ALL PASSED' if not fails else 'FAILURES: ' + ', '.join(fails)}")
sys.exit(1 if fails else 0)