fix: address webhook review followups

coven_github_adapter.py

@@ -30,6 +30,7 @@
 ).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()
+MAX_WEBHOOK_BODY_BYTES = 10 * 1024 * 1024
 
 
 def env_int(name, default, minimum=0, maximum=10):
@@ -202,19 +203,32 @@ def json_response(start_response, status, body):
     return [payload]
 
 
+class PayloadTooLarge(Exception):
+    pass
+
+
 def read_request_body(environ):
+    raw_length = (environ.get("CONTENT_LENGTH") or "").strip()
     try:
-        length = int(environ.get("CONTENT_LENGTH") or "0")
+        length = int(raw_length) if raw_length else -1
     except ValueError:
-        length = 0
-    return environ["wsgi.input"].read(length)
+        length = -1
+    if length > MAX_WEBHOOK_BODY_BYTES:
+        raise PayloadTooLarge
+    if length > 0:
+        return environ["wsgi.input"].read(length)
+    body = environ["wsgi.input"].read(MAX_WEBHOOK_BODY_BYTES + 1)
+    if len(body) > MAX_WEBHOOK_BODY_BYTES:
+        raise PayloadTooLarge
+    return body
 
 
 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"
+    signature_header = signature_header.strip()
     prefix = "sha256="
     if not signature_header.startswith(prefix):
         return False, "invalid signature"
@@ -232,7 +246,10 @@ def application(environ, start_response):
     if method != "POST" or path not in ("/", "/webhook"):
         return json_response(start_response, "404 Not Found", {"error": "not found"})
 
-    body = read_request_body(environ)
+    try:
+        body = read_request_body(environ)
+    except PayloadTooLarge:
+        return json_response(start_response, "413 Payload Too Large", {"error": "payload too large"})
     ok, error = verify_webhook_signature(
         WEBHOOK_SECRET,
         body,

tests/test_webhook_adapter.py

@@ -33,15 +33,18 @@ def signature(secret, body):
 
 
 class WebhookAdapterTests(unittest.TestCase):
-    def call_app(self, adapter, body, headers=None):
+    def call_app(self, adapter, body, headers=None, content_length="auto"):
         headers = headers or {}
         status_headers = []
         environ = {
             "REQUEST_METHOD": "POST",
             "PATH_INFO": "/webhook",
-            "CONTENT_LENGTH": str(len(body)),
             "wsgi.input": io.BytesIO(body),
         }
+        if content_length == "auto":
+            environ["CONTENT_LENGTH"] = str(len(body))
+        elif content_length is not None:
+            environ["CONTENT_LENGTH"] = str(content_length)
         for name, value in headers.items():
             environ["HTTP_" + name.upper().replace("-", "_")] = value
 
@@ -102,6 +105,110 @@ def test_webhook_accepts_valid_signed_ping_without_runtime(self):
             self.assertEqual(payload["action"], "ignored")
             self.assertEqual(payload["reason"], "no_policy_for_installation_repo")
 
+    def test_webhook_reads_body_when_content_length_is_missing(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            from pathlib import Path
+
+            secret = "missing-length-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-missing-length",
+                    "X-Hub-Signature-256": signature(secret, body),
+                },
+                content_length=None,
+            )
+
+            self.assertEqual(status, "200 OK")
+            self.assertTrue(payload["ok"])
+
+    def test_webhook_reads_body_when_content_length_is_unparsable(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            from pathlib import Path
+
+            secret = "bad-length-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-bad-length",
+                    "X-Hub-Signature-256": signature(secret, body),
+                },
+                content_length="not-a-number",
+            )
+
+            self.assertEqual(status, "200 OK")
+            self.assertTrue(payload["ok"])
+
+    def test_webhook_rejects_oversized_content_length_before_signature_check(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            from pathlib import Path
+
+            adapter = import_adapter(Path(tmp))
+            status, payload = self.call_app(
+                adapter,
+                b"",
+                {
+                    "X-GitHub-Event": "ping",
+                    "X-GitHub-Delivery": "delivery-large-body",
+                    "X-Hub-Signature-256": "sha256=deadbeef",
+                },
+                content_length=10 * 1024 * 1024 + 1,
+            )
+
+            self.assertEqual(status, "413 Payload Too Large")
+            self.assertEqual(payload["error"], "payload too large")
+
+    def test_webhook_signature_allows_surrounding_whitespace(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            from pathlib import Path
+
+            secret = "whitespace-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-whitespace-signature",
+                    "X-Hub-Signature-256": " " + signature(secret, body) + " ",
+                },
+            )
+
+            self.assertEqual(status, "200 OK")
+            self.assertTrue(payload["ok"])
+
+    def test_webhook_reports_missing_secret_as_server_misconfiguration(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            from pathlib import Path
+
+            adapter = import_adapter(Path(tmp), "")
+            body = b'{"zen":"Keep it logically awesome."}'
+
+            status, payload = self.call_app(
+                adapter,
+                body,
+                {
+                    "X-GitHub-Event": "ping",
+                    "X-GitHub-Delivery": "delivery-missing-secret",
+                    "X-Hub-Signature-256": signature("ignored", body),
+                },
+            )
+
+            self.assertEqual(status, "500 Internal Server Error")
+            self.assertEqual(payload["error"], "webhook secret not configured")
+
     def test_webhook_secret_supports_smoke_script_environment_name(self):
         with tempfile.TemporaryDirectory() as tmp:
             from pathlib import Path