From 33f6beedaf31ecf4270e18b3e39a139d6ec1b909 Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Mon, 6 Jul 2026 05:46:07 -0500 Subject: [PATCH] fix: address webhook review followups --- coven_github_adapter.py | 25 ++++++-- tests/test_webhook_adapter.py | 111 +++++++++++++++++++++++++++++++++- 2 files changed, 130 insertions(+), 6 deletions(-) diff --git a/coven_github_adapter.py b/coven_github_adapter.py index ce8fec9..5eb521c 100644 --- a/coven_github_adapter.py +++ b/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,12 +203,24 @@ 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): @@ -215,6 +228,7 @@ def verify_webhook_signature(secret, body, signature_header): 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, diff --git a/tests/test_webhook_adapter.py b/tests/test_webhook_adapter.py index 1eff746..bc26872 100644 --- a/tests/test_webhook_adapter.py +++ b/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