"""Signed webhook ingress deduplicates deliveries and wakes a durable session."""
import hashlib
import hmac
import json
import sqlite3
import time
import unittest
import urllib.error
import urllib.request
from daemon_test import delayed_request, refused_request
from harness import Albedo, Provider, python, text
class WebhookTests(unittest.TestCase):
def setUp(self):
def script(request):
latest = request["input"][-1]
if latest.get("role") == "user" and "probe webhook binding" in str(
latest.get("content", "")
):
code = (
"hooks = await webhooks.list()\n"
f"assert any(h.name == {self.hook_name!r} for h in hooks)\n"
f"payload = await webhooks.delivery('{self.delivery}')\n"
"assert 'down' in payload.body\n"
"print('WEBHOOK_BINDING_OK')"
)
return python(code)
return text("acknowledged")
self.provider = Provider(script)
self.addCleanup(self.provider.close)
self.app = Albedo(self.provider, protocol="responses")
self.app.__enter__()
self.addCleanup(self.app.__exit__, None, None, None)
self.session = self.app.session()
self.hook_name = f"outage-{self.provider.route}"
self.command_route = f"/sessions/{self.session}/commands"
created = self.command("create", self.hook_name)
self.hook, self.secret = created["hook"], created["secret"]
self.addCleanup(self.command, "delete", self.hook["id"])
self.payload = b'{"status":"down"}'
self.url = self.app.base + self.hook["url"]
def command(self, action, details=None):
args = {"action": action}
if details is not None:
args["details"] = details
with self.app.api(
self.command_route, {"name": "/webhooks", "args": args}
) as response:
return json.load(response)["result"]
def delivery_count(self):
with sqlite3.connect(self.app.home / "albedo.sqlite", timeout=10) as database:
return database.execute(
"SELECT count(*) FROM webhook_deliveries WHERE hook = ?",
(self.hook["id"],),
).fetchone()[0]
def send(
self, body=None, *, signature=None, event_id="alert-1", origin=None, token=None
):
body = self.payload if body is None else body
signature = (
signature
or "sha256="
+ hmac.new(self.secret.encode(), body, hashlib.sha256).hexdigest()
)
headers = {"X-Albedo-Signature": signature, "X-Albedo-Event-Id": event_id}
if token:
headers["Authorization"] = "Bearer " + token
if origin:
headers["Origin"] = origin
request = urllib.request.Request(self.url, data=body, headers=headers)
return urllib.request.urlopen(request, timeout=25)
def test_signature_origin_and_idempotency_are_enforced(self):
with self.assertRaises(urllib.error.HTTPError) as rejected:
self.send(signature="sha256=wrong", token=self.app.connection["token"])
self.assertEqual(rejected.exception.code, 401)
with self.assertRaises(urllib.error.HTTPError) as rejected:
self.send(origin="https://attacker.example")
self.assertEqual(rejected.exception.code, 403)
self.assertFalse(self.provider.requests)
self.assertEqual(self.delivery_count(), 0)
with self.send() as response:
self.assertEqual(response.status, 202)
delivery = json.load(response)["deliveryId"]
with self.send() as response:
self.assertEqual(json.load(response)["deliveryId"], delivery)
with self.assertRaises(urllib.error.HTTPError) as rejected:
self.send(body=b"different")
self.assertEqual(rejected.exception.code, 409)
deadline = time.monotonic() + 40
while time.monotonic() < deadline and not self.provider.requests:
time.sleep(0.25)
self.assertEqual(
len(self.provider.requests), 1, "retry should wake the session once"
)
self.assertIn(
f"[webhook {self.hook_name} #", json.dumps(self.provider.requests[0])
)
self.assertIn(
self.hook["id"],
[item["hook"]["id"] for item in self.command("list")["hooks"]],
)
def test_ingress_framing_refusals_do_not_admit_a_delivery(self):
path = self.hook["url"]
signature = (
"sha256="
+ hmac.new(self.secret.encode(), self.payload, hashlib.sha256).hexdigest()
)
for framing, expected_status, code in (
({"Transfer-Encoding": "chunked"}, 400, "unsupported_transfer_encoding"),
({"Content-Length": "invalid"}, 400, "invalid_content_length"),
({"Content-Length": "65537"}, 413, "request_body_too_large"),
):
with self.subTest(framing=framing):
status, headers, body = refused_request(
self.app, path, {**framing, "X-Albedo-Signature": signature}
)
self.assertEqual(status, expected_status)
self.assertEqual(json.loads(body)["code"], code)
self.assertEqual(headers["albedo-error-code"], code)
status, _, _ = refused_request(
self.app, path, {**framing, "Origin": "https://attacker.example"}
)
self.assertEqual(status, 403)
self.assertFalse(self.provider.requests)
self.assertEqual(self.delivery_count(), 0)
with self.send() as response:
self.assertEqual(response.status, 202)
def test_signed_delayed_body_and_unknown_bodies_keep_connection_usable(self):
signature = (
"sha256="
+ hmac.new(self.secret.encode(), self.payload, hashlib.sha256).hexdigest()
)
self.assertEqual(
delayed_request(
self.app,
"POST",
self.hook["url"],
self.payload,
{
"X-Albedo-Signature": signature,
"X-Albedo-Event-Id": "split-delivery",
},
),
(202, 200),
)
for method, path in (
("GET", self.hook["url"]),
("POST", self.hook["url"] + "/missing"),
):
with self.subTest(method=method, path=path):
self.assertEqual(
delayed_request(self.app, method, path, b"{}", {}), (404, 200)
)
def test_agent_binding_reads_hook_and_delivery(self):
with self.send() as response:
self.delivery = json.load(response)["deliveryId"]
deadline = time.monotonic() + 40
while time.monotonic() < deadline and not self.provider.requests:
time.sleep(0.25)
self.assertTrue(self.provider.requests)
self.app.idle(self.session, timeout=40)
self.command("agent_on")
self.app.prompt(self.session, "probe webhook binding").close()
self.app.idle(self.session)
requests = [entry["request"] for entry in self.provider.requests]
self.assertEqual(len(requests), 3)
outputs = [
item.get("output", "")
for item in requests[-1]["input"]
if item.get("type") == "function_call_output"
]
self.assertTrue(
any("WEBHOOK_BINDING_OK" in str(output) for output in outputs), outputs
)