fix: harden webhook entrypoint

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:

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 {}

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()