From 48e69e75d4447af10f38c11d6784a86034690f1b Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Mon, 6 Jul 2026 03:42:39 -0500 Subject: [PATCH] fix: harden webhook entrypoint --- README.md | 15 ++++ coven_github_adapter.py | 80 ++++++++++++++++- tests/test_webhook_adapter.py | 157 ++++++++++++++++++++++++++++++++++ 3 files changed, 251 insertions(+), 1 deletion(-) create mode 100644 tests/test_webhook_adapter.py diff --git a/README.md b/README.md index 9739a39..d2b0665 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,7 @@ reproduced, and changed through PRs instead of server-only edits. The deployment expects secrets and mutable state to be supplied outside git: - `GITHUB_APP_ID` +- `GITHUB_WEBHOOK_SECRET` or `WEBHOOK_SECRET` - `GITHUB_APP_PRIVATE_KEY_PATH` or `.coven-github-private-key.pem` - `COVEN_GITHUB_STATE_DIR` - `COVEN_GITHUB_POLICY_PATH` @@ -33,6 +34,20 @@ The deployment expects secrets and mutable state to be supplied outside git: Do not commit private keys, webhook secrets, OAuth tokens, generated task state, workspaces, or attempt artifacts. +## Local smoke test + +Run the adapter behind any WSGI server that points at +`coven_github_adapter:application`, then verify signature handling: + +```bash +WEBHOOK_SECRET="replace-with-local-secret" \ + scripts/smoke-webhook.sh http://localhost:3000/webhook +``` + +The smoke test proves that unsigned requests and bad signatures are rejected, +while a correctly HMAC-signed GitHub `ping` delivery is accepted without needing +`coven-code` or a GitHub installation token. + ## Policy The default checked-in policy is empty: diff --git a/coven_github_adapter.py b/coven_github_adapter.py index 785e7c0..ce8fec9 100644 --- a/coven_github_adapter.py +++ b/coven_github_adapter.py @@ -1,11 +1,13 @@ import base64 import hashlib +import hmac import json import os import subprocess import tempfile import time import traceback +import uuid from datetime import datetime, timezone from pathlib import Path from urllib.error import HTTPError @@ -23,6 +25,9 @@ os.environ.get("GITHUB_APP_PRIVATE_KEY_PATH", ROOT_DIR / ".coven-github-private-key.pem") ) APP_ID = os.environ.get("GITHUB_APP_ID", "").strip() +WEBHOOK_SECRET = ( + os.environ.get("GITHUB_WEBHOOK_SECRET") or os.environ.get("WEBHOOK_SECRET", "") +).strip() COVEN_CODE_BIN = os.environ.get("COVEN_CODE_BIN", "coven-code").strip() or "coven-code" COVEN_CODE_MODEL = os.environ.get("COVEN_CODE_MODEL", "gpt-5.5").strip() @@ -180,6 +185,77 @@ def task_path(task_id): return TASKS_DIR / (task_id + ".json") +def header(environ, name, default=""): + key = "HTTP_" + name.upper().replace("-", "_") + return environ.get(key, default) + + +def json_response(start_response, status, body): + payload = json.dumps(body, sort_keys=True).encode("utf-8") + start_response( + status, + [ + ("Content-Type", "application/json"), + ("Content-Length", str(len(payload))), + ], + ) + return [payload] + + +def read_request_body(environ): + try: + length = int(environ.get("CONTENT_LENGTH") or "0") + except ValueError: + length = 0 + return environ["wsgi.input"].read(length) + + +def verify_webhook_signature(secret, body, signature_header): + if not secret: + return False, "webhook secret not configured" + if not signature_header: + return False, "missing signature" + prefix = "sha256=" + if not signature_header.startswith(prefix): + return False, "invalid signature" + expected = prefix + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest() + if not hmac.compare_digest(expected, signature_header): + return False, "invalid signature" + return True, None + + +def application(environ, start_response): + method = environ.get("REQUEST_METHOD", "GET").upper() + path = environ.get("PATH_INFO", "/") + if method == "GET" and path in ("/", "/healthz"): + return json_response(start_response, "200 OK", {"ok": True}) + if method != "POST" or path not in ("/", "/webhook"): + return json_response(start_response, "404 Not Found", {"error": "not found"}) + + body = read_request_body(environ) + ok, error = verify_webhook_signature( + WEBHOOK_SECRET, + body, + header(environ, "X-Hub-Signature-256"), + ) + if not ok: + status = "500 Internal Server Error" if error == "webhook secret not configured" else "401 Unauthorized" + return json_response(start_response, status, {"error": error}) + + event_name = header(environ, "X-GitHub-Event") + if not event_name: + return json_response(start_response, "400 Bad Request", {"error": "missing event"}) + + try: + payload = json.loads(body.decode("utf-8")) + except json.JSONDecodeError: + return json_response(start_response, "400 Bad Request", {"error": "invalid json"}) + + delivery_id = header(environ, "X-GitHub-Delivery") or str(uuid.uuid4()) + result = route_delivery(event_name, delivery_id, payload, lambda message: print(message, flush=True)) + return json_response(start_response, "200 OK", result) + + def payload_hash(payload): raw = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(raw).hexdigest() @@ -222,7 +298,7 @@ def labels_include_trigger(labels, policy): def build_task_from_event(event_name, delivery_id, payload, policy): repository = payload.get("repository") or {} installation = payload.get("installation") or {} - familiar = policy.get("familiar") or DEFAULT_POLICY["installations"]["144638200"]["repositories"]["1290307745"]["familiar"] + familiar = policy.get("familiar") base = { "task_id": delivery_id, "delivery_id": delivery_id, @@ -240,6 +316,8 @@ def build_task_from_event(event_name, delivery_id, payload, policy): "publication": policy.get("publication") or {"mode": "record_only"}, "issue_refs": ["OpenCoven/coven-github#2", "OpenCoven/coven-github#7"], } + if not familiar: + return ignored(base, "missing_familiar_policy") if event_name == "issue_comment": issue = payload.get("issue") or {} diff --git a/tests/test_webhook_adapter.py b/tests/test_webhook_adapter.py new file mode 100644 index 0000000..1eff746 --- /dev/null +++ b/tests/test_webhook_adapter.py @@ -0,0 +1,157 @@ +import hashlib +import hmac +import importlib +import io +import json +import os +import sys +import tempfile +import unittest + + +def import_adapter(state_dir, webhook_secret="test-webhook-secret"): + os.environ["COVEN_GITHUB_STATE_DIR"] = str(state_dir) + os.environ["COVEN_GITHUB_POLICY_PATH"] = str(state_dir / "policy.json") + os.environ.pop("WEBHOOK_SECRET", None) + os.environ["GITHUB_WEBHOOK_SECRET"] = webhook_secret + sys.modules.pop("coven_github_adapter", None) + return importlib.import_module("coven_github_adapter") + + +def import_adapter_with_legacy_secret(state_dir, webhook_secret="legacy-webhook-secret"): + os.environ["COVEN_GITHUB_STATE_DIR"] = str(state_dir) + os.environ["COVEN_GITHUB_POLICY_PATH"] = str(state_dir / "policy.json") + os.environ.pop("GITHUB_WEBHOOK_SECRET", None) + os.environ["WEBHOOK_SECRET"] = webhook_secret + sys.modules.pop("coven_github_adapter", None) + return importlib.import_module("coven_github_adapter") + + +def signature(secret, body): + digest = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest() + return "sha256=" + digest + + +class WebhookAdapterTests(unittest.TestCase): + def call_app(self, adapter, body, headers=None): + headers = headers or {} + status_headers = [] + environ = { + "REQUEST_METHOD": "POST", + "PATH_INFO": "/webhook", + "CONTENT_LENGTH": str(len(body)), + "wsgi.input": io.BytesIO(body), + } + for name, value in headers.items(): + environ["HTTP_" + name.upper().replace("-", "_")] = value + + def start_response(status, response_headers): + status_headers.append((status, response_headers)) + + response = b"".join(adapter.application(environ, start_response)) + status = status_headers[0][0] + return status, json.loads(response.decode("utf-8")) + + def test_webhook_rejects_missing_and_invalid_signatures(self): + with tempfile.TemporaryDirectory() as tmp: + from pathlib import Path + + adapter = import_adapter(Path(tmp)) + body = b'{"zen":"Keep it logically awesome."}' + + missing_status, missing_payload = self.call_app( + adapter, + body, + {"X-GitHub-Event": "ping", "X-GitHub-Delivery": "delivery-1"}, + ) + self.assertEqual(missing_status, "401 Unauthorized") + self.assertEqual(missing_payload["error"], "missing signature") + + invalid_status, invalid_payload = self.call_app( + adapter, + body, + { + "X-GitHub-Event": "ping", + "X-GitHub-Delivery": "delivery-2", + "X-Hub-Signature-256": "sha256=deadbeef", + }, + ) + self.assertEqual(invalid_status, "401 Unauthorized") + self.assertEqual(invalid_payload["error"], "invalid signature") + + def test_webhook_accepts_valid_signed_ping_without_runtime(self): + with tempfile.TemporaryDirectory() as tmp: + from pathlib import Path + + secret = "valid-webhook-secret" + adapter = import_adapter(Path(tmp), secret) + body = b'{"zen":"Keep it logically awesome."}' + + status, payload = self.call_app( + adapter, + body, + { + "X-GitHub-Event": "ping", + "X-GitHub-Delivery": "delivery-3", + "X-Hub-Signature-256": signature(secret, body), + }, + ) + + self.assertEqual(status, "200 OK") + self.assertTrue(payload["ok"]) + self.assertEqual(payload["action"], "ignored") + self.assertEqual(payload["reason"], "no_policy_for_installation_repo") + + def test_webhook_secret_supports_smoke_script_environment_name(self): + with tempfile.TemporaryDirectory() as tmp: + from pathlib import Path + + secret = "legacy-webhook-secret" + adapter = import_adapter_with_legacy_secret(Path(tmp), secret) + body = b'{"zen":"Keep it logically awesome."}' + + status, payload = self.call_app( + adapter, + body, + { + "X-GitHub-Event": "ping", + "X-GitHub-Delivery": "delivery-legacy", + "X-Hub-Signature-256": signature(secret, body), + }, + ) + + self.assertEqual(status, "200 OK") + self.assertTrue(payload["ok"]) + + def test_missing_familiar_policy_does_not_fall_back_to_hardcoded_installation(self): + with tempfile.TemporaryDirectory() as tmp: + from pathlib import Path + + adapter = import_adapter(Path(tmp)) + task = adapter.build_task_from_event( + "issues", + "delivery-4", + { + "action": "opened", + "installation": {"id": 111}, + "repository": { + "id": 222, + "full_name": "OpenCoven/example", + "clone_url": "https://github.com/OpenCoven/example.git", + "default_branch": "main", + }, + "issue": {"number": 7, "title": "Fix it", "body": "Please fix it."}, + }, + { + "trigger_labels": ["coven:fix"], + "bot_usernames": ["coven-github[bot]"], + "publication": {"mode": "record_only"}, + }, + ) + + self.assertEqual(task["state"], "ignored") + self.assertEqual(task["ignored_reason"], "missing_familiar_policy") + + +if __name__ == "__main__": + unittest.main()