okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauffeat(api)!: migrate daemon and clients to protocol 361c43ca

main
branchesmainall refs →
10.7 KiB265 linesraw
1"""Signed webhook ingress deduplicates deliveries and wakes a durable session."""
2
3import hashlib
4import hmac
5import json
6import sqlite3
7import time
8import unittest
9import urllib.error
10import urllib.request
11
12from daemon_test import delayed_request, refused_request
13from harness import Albedo, Provider, python, text
14
15
16class WebhookTests(unittest.TestCase):
17    def setUp(self):
18        def script(request):
19            latest = request["input"][-1]
20            if latest.get("role") == "user" and "probe webhook binding" in str(
21                latest.get("content", "")
22            ):
23                code = (
24                    "hooks = await webhooks.list()\n"
25                    f"assert any(h.name == {self.hook_name!r} for h in hooks)\n"
26                    f"payload = await webhooks.delivery('{self.delivery}')\n"
27                    "assert 'down' in payload.body\n"
28                    "print('WEBHOOK_BINDING_OK')"
29                )
30                return python(code)
31            return text("acknowledged")
32
33        self.provider = Provider(script)
34        self.addCleanup(self.provider.close)
35        self.app = Albedo(self.provider, protocol="responses")
36        self.app.__enter__()
37        self.addCleanup(self.app.__exit__, None, None, None)
38        self.session = self.app.session()
39        self.hook_name = f"outage-{self.provider.route}"
40        with self.app.api(
41            "/extensions/webhooks/hooks",
42            {"name": self.hook_name, "session_id": self.session},
43        ) as response:
44            created = json.load(response)
45        self.hook, self.secret = created["resource"]["value"], created["secret"]
46        self.hook_route = f"/extensions/webhooks/hooks/{self.hook['id']}"
47        self.addCleanup(self.delete_hook)
48        self.payload = b'{"status":"down"}'
49        self.delivery_path = self.hook_route + "/deliveries"
50        self.url = self.app.base + self.delivery_path
51
52    def delete_hook(self):
53        with self.app.api(self.hook_route + "?view=configuration") as response:
54            json.load(response)
55            revision = response.headers["ETag"]
56        self.app.api(
57            self.hook_route + "?view=configuration",
58            method="DELETE",
59            headers={"If-Match": revision},
60        ).close()
61
62    def allow_agent(self):
63        route = f"/extensions/webhooks/permissions/{self.session}"
64        with self.app.api(route) as response:
65            json.load(response)
66            revision = response.headers["ETag"]
67        self.app.api(
68            route,
69            {"agent_manage": True},
70            method="PATCH",
71            headers={"If-Match": revision},
72        ).close()
73
74    def delivery_count(self):
75        with sqlite3.connect(self.app.home / "albedo.sqlite", timeout=10) as database:
76            return database.execute(
77                "SELECT count(*) FROM webhook_deliveries WHERE hook = ?",
78                (self.hook["id"],),
79            ).fetchone()[0]
80
81    def send(
82        self, body=None, *, signature=None, event_id="alert-1", origin=None, token=None
83    ):
84        body = self.payload if body is None else body
85        signature = (
86            signature
87            or "sha256="
88            + hmac.new(self.secret.encode(), body, hashlib.sha256).hexdigest()
89        )
90        headers = {"X-Albedo-Signature": signature, "X-Albedo-Event-Id": event_id}
91        if token:
92            headers["Authorization"] = "Bearer " + token
93        if origin:
94            headers["Origin"] = origin
95        request = urllib.request.Request(self.url, data=body, headers=headers)
96        return urllib.request.urlopen(request, timeout=25)
97
98    def test_hook_configuration_is_conditional_and_never_discloses_the_secret(self):
99        resource = self.hook_route + "?view=configuration"
100        with self.app.api(resource) as response:
101            original = json.load(response)
102            revision = response.headers["ETag"]
103        self.assertNotIn(self.secret, json.dumps(original))
104        patch = {
105            "name": self.hook_name + "-edited",
106            "enabled": False,
107            "signature_header": "x-observed-signature",
108            "signature_prefix": "hmac=",
109        }
110        with self.app.api(
111            resource, patch, method="PATCH", headers={"If-Match": revision}
112        ) as response:
113            changed = json.load(response)
114        self.assertNotIn(self.secret, json.dumps(changed))
115        current = changed["resource"]["value"]
116        for key, value in patch.items():
117            self.assertEqual(current[key], value)
118        with self.assertRaises(urllib.error.HTTPError) as rejected:
119            self.app.api(
120                resource,
121                {"name": "stale-must-not-win", "enabled": True},
122                method="PATCH",
123                headers={"If-Match": revision},
124            )
125        self.assertEqual(rejected.exception.code, 412)
126        with self.app.api(resource) as response:
127            self.assertEqual(json.load(response), current)
128        with self.app.api(self.hook_route) as response:
129            self.assertNotIn(self.secret, response.read().decode("utf-8"))
130        with self.assertRaises(urllib.error.HTTPError) as rejected:
131            self.app.api(
132                "/extensions/webhooks/hooks",
133                {"name": current["name"], "session_id": self.session},
134            )
135        self.assertEqual(rejected.exception.code, 409)
136        self.assertFalse(self.provider.requests)
137
138    def test_signature_origin_and_idempotency_are_enforced(self):
139        with self.assertRaises(urllib.error.HTTPError) as rejected:
140            self.send(signature="sha256=wrong", token=self.app.connection["token"])
141        self.assertEqual(rejected.exception.code, 403)
142        with self.assertRaises(urllib.error.HTTPError) as rejected:
143            self.send(origin="https://attacker.example")
144        self.assertEqual(rejected.exception.code, 403)
145        self.assertFalse(self.provider.requests)
146        self.assertEqual(self.delivery_count(), 0)
147        with self.send() as response:
148            self.assertEqual(response.status, 202)
149            delivery = json.load(response)["delivery_id"]
150        with self.send() as response:
151            self.assertEqual(json.load(response)["delivery_id"], delivery)
152        with self.assertRaises(urllib.error.HTTPError) as rejected:
153            self.send(body=b"different")
154        self.assertEqual(rejected.exception.code, 409)
155        deadline = time.monotonic() + 40
156        while time.monotonic() < deadline and not self.provider.requests:
157            time.sleep(0.25)
158        self.assertEqual(
159            len(self.provider.requests), 1, "retry should wake the session once"
160        )
161        self.assertIn(
162            f"[webhook {self.hook_name} #", json.dumps(self.provider.requests[0])
163        )
164        with self.app.api(
165            f"/extensions/webhooks/hooks?session_id={self.session}"
166        ) as response:
167            hooks = json.load(response)["items"]
168        self.assertIn(
169            self.hook["id"],
170            [item["configuration_resource"]["value"]["id"] for item in hooks],
171        )
172
173    def test_ingress_framing_refusals_do_not_admit_a_delivery(self):
174        path = self.delivery_path
175        signature = (
176            "sha256="
177            + hmac.new(self.secret.encode(), self.payload, hashlib.sha256).hexdigest()
178        )
179        for framing, expected_status, code in (
180            ({"Transfer-Encoding": "identity"}, 400, "unsupported_transfer_encoding"),
181            ({"Content-Length": "invalid"}, 400, "invalid_request"),
182            ({"Content-Length": "65537"}, 413, "request_body_too_large"),
183        ):
184            with self.subTest(framing=framing):
185                status, headers, body = refused_request(
186                    self.app,
187                    path,
188                    {**framing, "X-Albedo-Signature": signature},
189                    method="POST",
190                )
191                self.assertEqual(status, expected_status)
192                self.assertEqual(json.loads(body)["code"], code)
193                self.assertEqual(json.loads(body)["status"], expected_status)
194                status, _, _ = refused_request(
195                    self.app,
196                    path,
197                    {**framing, "Origin": "https://attacker.example"},
198                    method="POST",
199                )
200                self.assertEqual(status, 403)
201        self.assertFalse(self.provider.requests)
202        self.assertEqual(self.delivery_count(), 0)
203        with self.send() as response:
204            self.assertEqual(response.status, 202)
205
206    def test_signed_delayed_body_and_unknown_bodies_keep_connection_usable(self):
207        signature = (
208            "sha256="
209            + hmac.new(self.secret.encode(), self.payload, hashlib.sha256).hexdigest()
210        )
211        self.assertEqual(
212            delayed_request(
213                self.app,
214                "POST",
215                self.delivery_path,
216                self.payload,
217                {
218                    "X-Albedo-Signature": signature,
219                    "X-Albedo-Event-Id": "split-delivery",
220                },
221            ),
222            (202, 200),
223        )
224        for method, path in (
225            ("GET", self.delivery_path + "/missing"),
226            ("POST", self.delivery_path + "/missing"),
227        ):
228            with self.subTest(method=method, path=path):
229                status, headers, _ = refused_request(
230                    self.app, path, {"Content-Length": "2"}, method=method
231                )
232                self.assertEqual(status, 401)
233                self.assertEqual(headers["www-authenticate"], "Bearer")
234                self.assertEqual(
235                    delayed_request(
236                        self.app,
237                        method,
238                        path,
239                        b"{}",
240                        {"Authorization": "Bearer " + self.app.connection["token"]},
241                    ),
242                    (404, 200),
243                )
244
245    def test_agent_binding_reads_hook_and_delivery(self):
246        with self.send() as response:
247            self.delivery = json.load(response)["delivery_id"]
248        deadline = time.monotonic() + 40
249        while time.monotonic() < deadline and not self.provider.requests:
250            time.sleep(0.25)
251        self.assertTrue(self.provider.requests)
252        self.app.idle(self.session, timeout=40)
253        self.allow_agent()
254        self.app.prompt(self.session, "probe webhook binding").close()
255        self.app.idle(self.session)
256        requests = [entry["request"] for entry in self.provider.requests]
257        self.assertEqual(len(requests), 3)
258        outputs = [
259            item.get("output", "")
260            for item in requests[-1]["input"]
261            if item.get("type") == "function_call_output"
262        ]
263        self.assertTrue(
264            any("WEBHOOK_BINDING_OK" in str(output) for output in outputs), outputs
265        )