flaxo/ghost

public

Source code of Ghost - On Ghost

delivery: follow the code standards883ec1c13 days ago415 lines · 12.7 KB

1	package main
2
3	import (
4	"bytes"
5	"context"
6	"crypto/aes"
7	"crypto/cipher"
8	"crypto/rand"
9	"encoding/base64"
10	"encoding/json"
11	"errors"
12	"io"
13	"net"
14	"net/http"
15	"net/http/httptest"
16	"net/netip"
17	"strconv"
18	"strings"
19	"testing"
20	"time"
21
22	"github.com/jackc/pgx/v5/pgxpool"
23	)
24
25	func TestBlocked(t *testing.T) {
26	tests := []struct {
27	addr string
28	blockedStrict, blockedLAN bool
29	}{
30	{"93.184.216.34", false, false},
31	{"2606:2800:220:1::1", false, false},
32	{"127.0.0.1", true, false},
33	{"::1", true, false},
34	{"10.1.2.3", true, false},
35	{"172.16.0.1", true, false},
36	{"192.168.1.4", true, false},
37	{"100.64.0.1", true, false},
38	{"fd00::1", true, false},
39	{"::ffff:127.0.0.1", true, false}, // IPv4-mapped loopback
40	{"169.254.169.254", true, true}, // cloud metadata, never allowed
41	{"::ffff:169.254.169.254", true, true},
42	{"fe80::1", true, true},
43	{"0.0.0.0", true, true},
44	{"224.0.0.1", true, true},
45	}
46	for _, tt := range tests {
47	a := netip.MustParseAddr(tt.addr)
48	if got := blocked(a, false); got != tt.blockedStrict {
49	t.Errorf("blocked(%s, strict) = %v, want %v", tt.addr, got, tt.blockedStrict)
50	}
51	if got := blocked(a, true); got != tt.blockedLAN {
52	t.Errorf("blocked(%s, allowPrivate) = %v, want %v", tt.addr, got, tt.blockedLAN)
53	}
54	}
55	}
56
57	func TestWebhookClientRefusesLoopbackOnTheDialedIP(t *testing.T) {
58	hits := make(chan struct{}, 1)
59	srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { hits <- struct{}{} }))
60	defer srv.Close()
61
62	// "localhost" resolves to 127.0.0.1: the check runs after DNS, on the real address.
63	url := "http://localhost:" + strconv.Itoa(srv.Listener.Addr().(*net.TCPAddr).Port)
64	_, err := newWebhookClient(false).Get(url)
65	if !errors.Is(err, errBlockedAddress) {
66	t.Fatalf("err = %v, want errBlockedAddress", err)
67	}
68	if len(hits) > 0 {
69	t.Error("request reached the loopback server")
70	}
71	}
72
73	func TestSignedHeaders(t *testing.T) {
74	secret, body := []byte("s3cret"), []byte(`{"action":"opened"}`)
75	now := time.Unix(1_700_000_000, 0)
76	h := signedHeaders(secret, "job-1", "issue.opened", body, now)
77
78	if h["X-Ghost-Timestamp"] != "1700000000" \|\| h["X-Ghost-Delivery"] != "job-1" \|\| h["X-Ghost-Event"] != "issue.opened" {
79	t.Errorf("headers = %v", h)
80	}
81	// what a receiver computes
82	if want := "sha256=" + hmacHex(secret, []byte("1700000000."+string(body))); h["X-Ghost-Signature-256"] != want {
83	t.Errorf("X-Ghost-Signature-256 = %s, want %s", h["X-Ghost-Signature-256"], want)
84	}
85	if want := "sha256=" + hmacHex(secret, body); h["X-Hub-Signature-256"] != want {
86	t.Errorf("X-Hub-Signature-256 = %s, want %s", h["X-Hub-Signature-256"], want)
87	}
88	}
89
90	func TestOpenSecret(t *testing.T) {
91	key := make([]byte, 32)
92	rand.Read(key)
93	got, err := openSecret(key, sealSecret(t, key, "whsec_abc"))
94	if err != nil \|\| string(got) != "whsec_abc" {
95	t.Fatalf("openSecret = %q, %v", got, err)
96	}
97	tests := []struct {
98	name string
99	key []byte
100	sealed string
101	}{
102	{"wrong key", make([]byte, 32), sealSecret(t, key, "whsec_abc")},
103	{"no key", nil, sealSecret(t, key, "whsec_abc")},
104	{"key of the wrong size", make([]byte, 7), sealSecret(t, key, "whsec_abc")},
105	{"unknown format", key, "whsec_abc"},
106	{"not base64", key, "v1:%%%"},
107	{"shorter than a nonce", key, "v1:" + base64.StdEncoding.EncodeToString([]byte("short"))},
108	}
109	for _, tt := range tests {
110	if _, err := openSecret(tt.key, tt.sealed); err == nil {
111	t.Errorf("%s: opened the secret", tt.name)
112	}
113	}
114	}
115
116	func TestParseRetryAfter(t *testing.T) {
117	now := time.Date(2026, 9, 29, 12, 0, 0, 0, time.UTC)
118	tests := []struct {
119	v string
120	want time.Duration
121	ok bool
122	}{
123	{"120", 2 * time.Minute, true},
124	{"Tue, 29 Sep 2026 12:05:00 GMT", 5 * time.Minute, true},
125	{"", 0, false},
126	{"soon", 0, false},
127	{"Tue, 29 Sep 2026 11:00:00 GMT", 0, true}, // already past: retry now
128	}
129	for _, tt := range tests {
130	got, ok := parseRetryAfter(tt.v, now)
131	if got != tt.want \|\| ok != tt.ok {
132	t.Errorf("parseRetryAfter(%q) = %v, %v; want %v, %v", tt.v, got, ok, tt.want, tt.ok)
133	}
134	}
135	}
136
137	// sealSecret encrypts the way the API does.
138	func sealSecret(t *testing.T, key []byte, secret string) string {
139	t.Helper()
140	block, err := aes.NewCipher(key)
141	if err != nil {
142	t.Fatal(err)
143	}
144	gcm, err := cipher.NewGCM(block)
145	if err != nil {
146	t.Fatal(err)
147	}
148	nonce := make([]byte, gcm.NonceSize())
149	rand.Read(nonce)
150	return "v1:" + base64.StdEncoding.EncodeToString(gcm.Seal(nonce, nonce, []byte(secret), nil))
151	}
152
153	// Deleting the organization cascades to the endpoint and its job.
154	func webhookFixture(t testing.T, db pgxpool.Pool, url string) (webhookSender, Job, []byte) {
155	t.Helper()
156	key := make([]byte, 32)
157	rand.Read(key)
158	suffix := strconv.FormatInt(time.Now().UnixNano(), 36)
159	org := "org_test_delivery_" + suffix
160
161	_, err := db.Exec(t.Context(),
162	`INSERT INTO organization (id, name, slug, created_at) VALUES ($1, 'Delivery test', $2, now())`,
163	org, "test-delivery-"+suffix)
164	if err != nil {
165	t.Fatal(err)
166	}
167	t.Cleanup(func() {
168	if _, err := db.Exec(context.Background(), `DELETE FROM organization WHERE id = $1`, org); err != nil {
169	t.Errorf("cleanup: %v", err)
170	}
171	})
172
173	var endpointID string
174	err = db.QueryRow(t.Context(),
175	`INSERT INTO webhook_endpoint (id, organization_id, url, secret, events)
176	VALUES ($1, $2, $3, $4, '{issue.opened}') RETURNING id`,
177	"whk_test_"+suffix, org, url, sealSecret(t, key, "whsec_test")).Scan(&endpointID)
178	if err != nil {
179	t.Fatal(err)
180	}
181
182	payload, _ := json.Marshal(webhookPayload{Event: "issue.opened", Body: `{"action":"opened"}`})
183	j := Job{Kind: "webhook", EndpointID: &endpointID, Payload: payload}
184	err = db.QueryRow(t.Context(),
185	`INSERT INTO delivery_job (kind, idempotency_key, payload, endpoint_id)
186	VALUES ('webhook', $1 \|\| gen_random_uuid()::text, $2, $3) RETURNING id`,
187	testKeyPrefix, payload, endpointID).Scan(&j.ID)
188	if err != nil {
189	t.Fatal(err)
190	}
191	return webhookSender{db: db, client: newWebhookClient(true), key: key}, j, []byte("whsec_test")
192	}
193
194	func TestWebhookSendSignsAndRecordsTheAttempt(t *testing.T) {
195	db := testDB(t)
196	type received struct {
197	header http.Header
198	body []byte
199	}
200	reqs := make(chan received, 1)
201	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
202	body, _ := io.ReadAll(r.Body)
203	reqs <- received{r.Header, body}
204	w.Write([]byte("thanks"))
205	}))
206	defer srv.Close()
207	s, j, secret := webhookFixture(t, db, srv.URL)
208
209	if err := s.send(t.Context(), j); err != nil {
210	t.Fatal(err)
211	}
212	r := <-reqs
213	gotBody := r.body
214	got := struct{ Header http.Header }{r.header}
215
216	if string(gotBody) != `{"action":"opened"}` {
217	t.Errorf("body = %s", gotBody)
218	}
219	ts := got.Header.Get("X-Ghost-Timestamp")
220	if want := "sha256=" + hmacHex(secret, []byte(ts+"."), gotBody); got.Header.Get("X-Ghost-Signature-256") != want {
221	t.Error("signature does not verify with the endpoint secret")
222	}
223	if got.Header.Get("X-Ghost-Delivery") != j.ID {
224	t.Errorf("X-Ghost-Delivery = %s, want job id %s", got.Header.Get("X-Ghost-Delivery"), j.ID)
225	}
226
227	var (
228	status int
229	body string
230	)
231	err := db.QueryRow(t.Context(),
232	`SELECT status_code, response_body FROM delivery_attempt WHERE job_id = $1`, j.ID).Scan(&status, &body)
233	if err != nil {
234	t.Fatal(err)
235	}
236	if status != 200 \|\| body != "thanks" {
237	t.Errorf("attempt = %d %q, want 200 thanks", status, body)
238	}
239	}
240
241	func TestWebhookResponseClassification(t *testing.T) {
242	db := testDB(t)
243	tests := []struct {
244	name string
245	status int
246	header map[string]string
247	check func(t *testing.T, err error, endpointActive bool)
248	}{
249	{"500 retries", 500, nil, func(t *testing.T, err error, active bool) {
250	var p permanentError
251	if err == nil \|\| errors.As(err, &p) {
252	t.Errorf("err = %v, want a retryable error", err)
253	}
254	}},
255	{"410 disables the endpoint", 410, nil, func(t *testing.T, err error, active bool) {
256	var p permanentError
257	if !errors.As(err, &p) {
258	t.Errorf("err = %v, want permanentError", err)
259	}
260	if active {
261	t.Error("endpoint still active after 410")
262	}
263	}},
264	{"429 honors Retry-After", 429, map[string]string{"Retry-After": "120"}, func(t *testing.T, err error, active bool) {
265	var r retryAfterError
266	if !errors.As(err, &r) \|\| r.after != 2*time.Minute {
267	t.Errorf("err = %v, want retryAfterError of 2m", err)
268	}
269	}},
270	{"429 without Retry-After backs off as usual", 429, nil, func(t *testing.T, err error, active bool) {
271	var r retryAfterError
272	if err == nil \|\| errors.As(err, &r) {
273	t.Errorf("err = %v, want a plain retryable error", err)
274	}
275	}},
276	{"redirect is not followed", 302, map[string]string{"Location": "http://169.254.169.254/"}, func(t *testing.T, err error, active bool) {
277	if err == nil \|\| err.Error() != "endpoint responded 302" {
278	t.Errorf("err = %v, want endpoint responded 302", err)
279	}
280	}},
281	}
282	for _, tt := range tests {
283	t.Run(tt.name, func(t *testing.T) {
284	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
285	for k, v := range tt.header {
286	w.Header().Set(k, v)
287	}
288	w.WriteHeader(tt.status)
289	}))
290	defer srv.Close()
291	s, j, _ := webhookFixture(t, db, srv.URL)
292
293	err := s.send(t.Context(), j)
294
295	var active bool
296	if err := db.QueryRow(t.Context(), `SELECT active FROM webhook_endpoint WHERE id = $1`, *j.EndpointID).Scan(&active); err != nil {
297	t.Fatal(err)
298	}
299	tt.check(t, err, active)
300	})
301	}
302	}
303
304	func TestWebhookSendDropsJobForDisabledEndpoint(t *testing.T) {
305	db := testDB(t)
306	hits := make(chan struct{}, 1)
307	srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { hits <- struct{}{} }))
308	defer srv.Close()
309	s, j, _ := webhookFixture(t, db, srv.URL)
310	if _, err := db.Exec(t.Context(), `UPDATE webhook_endpoint SET active = false WHERE id = $1`, *j.EndpointID); err != nil {
311	t.Fatal(err)
312	}
313
314	var p permanentError
315	if err := s.send(t.Context(), j); !errors.As(err, &p) {
316	t.Errorf("err = %v, want permanentError", err)
317	}
318	if len(hits) > 0 {
319	t.Error("sent to a disabled endpoint")
320	}
321	}
322
323	// Sealed by sealWebhookSecret in apps/api/src/lib/webhooks/webhooks.ts with 32 bytes of 7 as the key; if this breaks, the API and delivery disagree on the format.
324	func TestOpenSecretSealedByTheAPI(t *testing.T) {
325	key := bytes.Repeat([]byte{7}, 32)
326	got, err := openSecret(key, "v1:imP+jJazaD8GgSnv6BdCuGKPLoZM2DO5Psxx9jR7ursCPfH6JAzhj4jGDlPn0w==")
327	if err != nil \|\| string(got) != "whsec_from_the_api" {
328	t.Fatalf("openSecret = %q, %v", got, err)
329	}
330	}
331
332	func TestHostLimiter(t *testing.T) {
333	h := newHostLimiter(2)
334	if !h.acquire("a:443") \|\| !h.acquire("a:443") {
335	t.Fatal("could not take the first two slots")
336	}
337	if h.acquire("a:443") {
338	t.Error("took a third slot for a:443")
339	}
340	if !h.acquire("b:443") {
341	t.Error("a busy a:443 blocked b:443")
342	}
343	h.release("a:443")
344	if !h.acquire("a:443") {
345	t.Error("released slot was not reusable")
346	}
347	var none *hostLimiter
348	if !none.acquire("a:443") {
349	t.Error("a nil limiter refused")
350	}
351	}
352
353	func TestWebhookSendWaitsWhenItsHostIsBusy(t *testing.T) {
354	db := testDB(t)
355	hits := make(chan struct{}, 1)
356	srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { hits <- struct{}{} }))
357	defer srv.Close()
358	s, j, _ := webhookFixture(t, db, srv.URL)
359	s.hosts = newHostLimiter(1)
360	s.hosts.acquire(srv.Listener.Addr().String()) // another send holds the only slot
361
362	var r retryAfterError
363	if err := s.send(t.Context(), j); !errors.As(err, &r) \|\| r.after != hostBusyRetry {
364	t.Errorf("err = %v, want retryAfterError of %v", err, hostBusyRetry)
365	}
366	if len(hits) > 0 {
367	t.Error("sent while the host was at its limit")
368	}
369	}
370
371	func TestWebhookSendRefusesJobsItCannotSend(t *testing.T) {
372	db := testDB(t)
373	srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
374	defer srv.Close()
375	s, j, _ := webhookFixture(t, db, srv.URL)
376
377	noEndpoint := j
378	noEndpoint.EndpointID = nil
379	badPayload := j
380	badPayload.Payload = json.RawMessage(`[]`)
381	deleted := j
382	deleted.EndpointID = ptr("whk_does_not_exist")
383
384	for name, job := range map[string]Job{"no endpoint": noEndpoint, "bad payload": badPayload, "deleted endpoint": deleted} {
385	var p permanentError
386	if err := s.send(t.Context(), job); !errors.As(err, &p) {
387	t.Errorf("%s: err = %v, want permanentError", name, err)
388	}
389	}
390
391	noKey := s
392	noKey.key = nil
393	if err := noKey.send(t.Context(), j); err == nil \|\| !strings.Contains(err.Error(), "WEBHOOK_SECRET_KEY") {
394	t.Errorf("without a key: err = %v", err)
395	}
396	}
397
398	func TestWebhookSendRecordsAnAttemptWithNoResponse(t *testing.T) {
399	db := testDB(t)
400	s, j, _ := webhookFixture(t, db, "http://127.0.0.1:1")
401
402	if err := s.send(t.Context(), j); err == nil {
403	t.Fatal("sent to a closed port")
404	}
405	var (
406	status *int
407	msg string
408	)
409	if err := db.QueryRow(t.Context(), `SELECT status_code, error FROM delivery_attempt WHERE job_id = $1`, j.ID).Scan(&status, &msg); err != nil {
410	t.Fatal(err)
411	}
412	if status != nil \|\| msg == "" {
413	t.Errorf("attempt = %v %q, want no status and the error", status, msg)
414	}
415	}