#!/usr/bin/env python3 """cc-ci comment-bridge (§4.1). Receives Gitea `issue_comment` webhooks; when a *collaborator* comments exactly `!testme` on an open PR, triggers a parameterized Drone build of the cc-ci pipeline for that PR's head commit and posts a PR comment linking the run. Everything else is ignored. Python stdlib only. Config (env): BRIDGE_LISTEN host:port to bind (default 0.0.0.0:8080) GITEA_API e.g. https://git.autonomic.zone/api/v1 DRONE_URL e.g. https://drone.ci.commoninternet.net CI_REPO the pipeline repo, e.g. recipe-maintainers/cc-ci HMAC_FILE file with the webhook HMAC secret DRONE_TOKEN_FILE file with the Drone API token GITEA_TOKEN_FILE file with the Gitea API token """ import hashlib import hmac import json import os import sys import urllib.error import urllib.parse import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer GITEA_API = os.environ.get("GITEA_API", "https://git.autonomic.zone/api/v1") DRONE_URL = os.environ.get("DRONE_URL", "https://drone.ci.commoninternet.net") CI_REPO = os.environ.get("CI_REPO", "recipe-maintainers/cc-ci") TRIGGER = "!testme" def _read(path): with open(path) as fh: return fh.read().strip() HMAC_SECRET = _read(os.environ["HMAC_FILE"]).encode() DRONE_TOKEN = _read(os.environ["DRONE_TOKEN_FILE"]) GITEA_TOKEN = _read(os.environ["GITEA_TOKEN_FILE"]) def log(*a): print(*a, file=sys.stderr, flush=True) def _api(url, token, method="GET", data=None): headers = {"Authorization": "token " + token} if token else {} body = None if data is not None: body = json.dumps(data).encode() headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=30) as resp: raw = resp.read() return resp.status, (json.loads(raw) if raw else None) except urllib.error.HTTPError as e: return e.code, None def is_collaborator(full_name, user): # 204 => the user has push access (collaborator or org member with access). status, _ = _api(f"{GITEA_API}/repos/{full_name}/collaborators/{user}", GITEA_TOKEN) return status == 204 def pr_head(owner, repo, number): status, pr = _api(f"{GITEA_API}/repos/{owner}/{repo}/pulls/{number}", GITEA_TOKEN) if status != 200 or not pr: return None head = pr.get("head", {}) return {"sha": head.get("sha"), "repo": (head.get("repo") or {}).get("full_name")} def trigger_build(recipe, ref, pr, src): # Drone "create build" with custom params -> exposed to the pipeline as env vars. q = urllib.parse.urlencode( {"branch": "main", "RECIPE": recipe, "REF": ref, "PR": str(pr), "SRC": src} ) url = f"{DRONE_URL}/api/repos/{CI_REPO}/builds?{q}" status, build = _api(url, DRONE_TOKEN, method="POST") if status in (200, 201) and build: return build.get("number") log("drone trigger failed", status) return None def post_comment(owner, repo, number, body): _api( f"{GITEA_API}/repos/{owner}/{repo}/issues/{number}/comments", GITEA_TOKEN, method="POST", data={"body": body}, ) class Handler(BaseHTTPRequestHandler): def _send(self, code, msg=""): self.send_response(code) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(msg.encode()) def do_GET(self): # health endpoint if self.path.rstrip("/") in ("/hook/healthz", "/healthz"): return self._send(200, "ok") return self._send(404, "not found") def do_POST(self): length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) # 1) verify HMAC (Gitea sends hex sha256 in X-Gitea-Signature) sig = self.headers.get("X-Gitea-Signature", "") expected = hmac.new(HMAC_SECRET, body, hashlib.sha256).hexdigest() if not hmac.compare_digest(sig, expected): log(f"rejected: bad signature event={self.headers.get('X-Gitea-Event')} " f"got={sig[:12]} want={expected[:12]} bodylen={len(body)} seclen={len(HMAC_SECRET)} " f"hub256={(self.headers.get('X-Hub-Signature-256') or '')[:20]}") return self._send(401, "bad signature") if self.headers.get("X-Gitea-Event") != "issue_comment": return self._send(204, "ignored") try: payload = json.loads(body) except ValueError: return self._send(400, "bad json") action = payload.get("action") comment = (payload.get("comment") or {}).get("body", "") issue = payload.get("issue") or {} repo = payload.get("repository") or {} user = (payload.get("comment") or {}).get("user", {}).get("login", "") full_name = repo.get("full_name", "") owner = (repo.get("owner") or {}).get("login", "") name = repo.get("name", "") number = issue.get("number") # 2) only a created comment, exactly "!testme", on a PR if action != "created" or comment.strip() != TRIGGER: return self._send(204, "ignored") if not issue.get("pull_request"): return self._send(204, "not a PR") # 3) commenter must be a collaborator / org member with access if not is_collaborator(full_name, user): log(f"rejected: {user} not a collaborator on {full_name}") return self._send(403, "not authorized") # 4) resolve PR head (test the code at the PR head commit) head = pr_head(owner, name, number) if not head or not head["sha"]: return self._send(502, "cannot resolve PR head") # 5) trigger the parameterized Drone build num = trigger_build(name, head["sha"], number, head["repo"] or full_name) if not num: post_comment(owner, name, number, "cc-ci: failed to start a CI run (see bridge logs).") return self._send(502, "trigger failed") run_url = f"{DRONE_URL}/{CI_REPO}/{num}" post_comment( owner, name, number, f"cc-ci: started CI run for `{name}` @ `{head['sha'][:8]}` → {run_url}", ) log(f"triggered build {num} for {name}@{head['sha'][:8]} (PR #{number}) by {user}") return self._send(201, run_url) def log_message(self, *a): # quiet default access logging pass def main(): host, _, port = os.environ.get("BRIDGE_LISTEN", "0.0.0.0:8080").rpartition(":") srv = ThreadingHTTPServer((host or "0.0.0.0", int(port)), Handler) log(f"comment-bridge listening on {host or '0.0.0.0'}:{port}") srv.serve_forever() if __name__ == "__main__": main()