krz/gitbay
A CLI-first git forge.
clone: git clone https://gitbay.org/krz/gitbay.git
9cdfb3a89de20278ac7db640fd3c85c46559b91e
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-08-24T00:53:13Z
cmd/gitbay/main.go | 11 + cmd/gitbayd/main.go | 15 ++ e2e/webhook_test.go | 256 +++++++++++++++++++++++ internal/config/config.go | 7 + internal/control/webhook.go | 202 ++++++++++++++++++ internal/hookd/hookd.go | 5 + internal/store/issues.go | 27 ++- internal/store/migrations/0005_webhooks.down.sql | 2 + internal/store/migrations/0005_webhooks.up.sql | 25 +++ internal/store/webhooks.go | 174 +++++++++++++++ internal/webhook/webhook.go | 182 ++++++++++++++++ 11 files changed, 903 insertions(+), 3 deletions(-) @@ -31,6 +31,7 @@ func main() { mrCmd(), webCmd(), orgCmd(), + webhookCmd(), remoteCmd(), initCmd(), pass("register", "create an account on the default instance: gitbay register --username <n> --email <a> | --invite <code>", @@ -294,6 +295,16 @@ func webCmd() *cobra.Command { ) } +func webhookCmd() *cobra.Command { + return group("webhook", "outbound event delivery", + pass("add", "add a webhook: <url> [--secret s] [--events k1,k2|*]", passOpts{server: []string{"webhook", "add"}, needsRepo: true}), + pass("list", "list webhooks", passOpts{server: []string{"webhook", "list"}, needsRepo: true}), + pass("remove", "remove a webhook: <id>", passOpts{server: []string{"webhook", "remove"}, needsRepo: true}), + pass("deliveries", "recent deliveries [--limit n]", passOpts{server: []string{"webhook", "deliveries"}, needsRepo: true}), + pass("redeliver", "requeue a delivery: <delivery-id>", passOpts{server: []string{"webhook", "redeliver"}, needsRepo: true}), + ) +} + func orgCmd() *cobra.Command { return group("org", "organizations", pass("create", "create an organization", passOpts{server: []string{"org", "create"}}), @@ -3,6 +3,7 @@ package main import ( + "context" "fmt" "log/slog" "net" @@ -11,6 +12,7 @@ import ( "path/filepath" "strconv" "strings" + "time" "github.com/spf13/cobra" "golang.org/x/crypto/acme/autocert" @@ -25,6 +27,7 @@ import ( "gitbay.org/gitbay/internal/policy" "gitbay.org/gitbay/internal/sshd" "gitbay.org/gitbay/internal/store" + "gitbay.org/gitbay/internal/webhook" ) func openStore(cfg config.Config) (*store.Store, error) { @@ -119,6 +122,18 @@ func serveCmd() *cobra.Command { } defer stopHookd() + // Outbound webhook deliveries. The retry base is overridable + // for tests via GITBAY_WEBHOOK_RETRY_BASE. + retryBase := 30 * time.Second + if v := os.Getenv("GITBAY_WEBHOOK_RETRY_BASE"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + retryBase = d + } + } + whCtx, whCancel := context.WithCancel(context.Background()) + defer whCancel() + go webhook.New(st, cfg.Webhooks.AllowLocal, retryBase).Run(whCtx) + errCh := make(chan error, 3) if cfg.SSH.Mode == "embedded" { srv, err := sshd.New(cfg, st) new file mode 100644 @@ -0,0 +1,256 @@ +package e2e + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "os" + "os/exec" + "strings" + "sync" + "testing" + "time" +) + +// hookReceiver captures webhook deliveries and can be told to fail. +type hookReceiver struct { + addr string + mu sync.Mutex + got []capturedHook + failNext int // respond 500 to this many requests +} + +type capturedHook struct { + event string + delivery string + signature string + body []byte +} + +func startHookReceiver(t *testing.T) *hookReceiver { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + h := &hookReceiver{addr: ln.Addr().String()} + go http.Serve(ln, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + h.mu.Lock() + defer h.mu.Unlock() + if h.failNext > 0 { + h.failNext-- + w.WriteHeader(500) + return + } + h.got = append(h.got, capturedHook{ + event: r.Header.Get("X-Gitbay-Event"), + delivery: r.Header.Get("X-Gitbay-Delivery"), + signature: r.Header.Get("X-Gitbay-Signature-256"), + body: body, + }) + w.WriteHeader(204) + })) + return h +} + +func (h *hookReceiver) waitN(t *testing.T, n int) []capturedHook { + t.Helper() + deadline := time.Now().Add(15 * time.Second) + for time.Now().Before(deadline) { + h.mu.Lock() + if len(h.got) >= n { + out := append([]capturedHook(nil), h.got...) + h.mu.Unlock() + return out + } + h.mu.Unlock() + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("only %d deliveries arrived, want %d", len(h.got), n) + return nil +} + +func TestWebhooks(t *testing.T) { + inst := startInstanceWith(t, "[webhooks]\nallow_local = true\n") + // Restart the daemon with a fast retry base for the failure tests. + inst.proc.Process.Kill() + inst.proc.Wait() + inst.proc = exec.Command(inst.gitbayd, "--config", inst.config, "serve") + inst.proc.Env = append(os.Environ(), "GITBAY_WEBHOOK_RETRY_BASE=500ms") + inst.proc.Stderr = os.Stderr + if err := inst.proc.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { inst.proc.Process.Kill(); inst.proc.Wait() }) + deadline := time.Now().Add(10 * time.Second) + for { + conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", inst.port), 200*time.Millisecond) + if err == nil { + conn.Close() + break + } + if time.Now().After(deadline) { + t.Fatal("daemon did not restart") + } + time.Sleep(50 * time.Millisecond) + } + + aliceKey := inst.newKey(t, "alice") + inst.admin(t, "admin", "user", "create", "alice", + "--key", aliceKey+".pub", "--email", "alice@example.test", "--verified") + if _, errOut, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/proj"); code != 0 { + t.Fatalf("repo create: %s", errOut) + } + + recv := startHookReceiver(t) + hookURL := "http://" + recv.addr + "/hook" + if _, errOut, code := inst.ssh(t, aliceKey, "", + "webhook", "add", "alice/proj", hookURL, "--secret", "s3cret"); code != 0 { + t.Fatalf("webhook add: %s", errOut) + } + + // An issue event arrives, signed and shaped. + if _, _, code := inst.ssh(t, aliceKey, "", "issue", "create", "alice/proj", "--title", "'hook me'"); code != 0 { + t.Fatal("issue create failed") + } + got := recv.waitN(t, 1) + h := got[0] + if h.event != "issue.created" || h.delivery == "" { + t.Fatalf("delivery headers: %+v", h) + } + mac := hmac.New(sha256.New, []byte("s3cret")) + mac.Write(h.body) + if h.signature != "sha256="+hex.EncodeToString(mac.Sum(nil)) { + t.Fatalf("HMAC mismatch: %s", h.signature) + } + var p struct { + Event string `json:"event"` + Repo string `json:"repo"` + Actor string `json:"actor"` + Data struct { + Number int `json:"number"` + } `json:"data"` + } + if err := json.Unmarshal(h.body, &p); err != nil { + t.Fatalf("payload: %v\n%s", err, h.body) + } + if p.Repo != "alice/proj" || p.Actor != "alice" || p.Data.Number != 1 { + t.Fatalf("payload fields: %+v", p) + } + + // Push events flow through the hook chain. + work := t.TempDir() + env := inst.gitEnv(aliceKey) + mustGit(t, work, env, "clone", inst.sshURL("alice/proj"), "w") + dir := work + "/w" + os.WriteFile(dir+"/f.txt", []byte("x\n"), 0o644) + mustGit(t, dir, env, "checkout", "-q", "-b", "main") + mustGit(t, dir, env, "add", ".") + mustGit(t, dir, env, "commit", "-q", "-m", "push event") + mustGit(t, dir, env, "push", "-q", "origin", "main") + got = recv.waitN(t, 2) + push := got[1] + if push.event != "push" || !strings.Contains(string(push.body), `"ref":"refs/heads/main"`) { + t.Fatalf("push event: %s %s", push.event, push.body) + } + + // Event filters: a hook subscribed to mr.created ignores issues. + recv2 := startHookReceiver(t) + if _, _, code := inst.ssh(t, aliceKey, "", + "webhook", "add", "alice/proj", "http://"+recv2.addr+"/", "--events", "mr.created"); code != 0 { + t.Fatal("filtered webhook add failed") + } + if _, _, code := inst.ssh(t, aliceKey, "", "issue", "create", "alice/proj", "--title", "'no hook'"); code != 0 { + t.Fatal("issue 2 failed") + } + got = recv.waitN(t, 3) // unfiltered hook sees it + if got[2].event != "issue.created" { + t.Fatalf("third delivery: %s", got[2].event) + } + time.Sleep(500 * time.Millisecond) + recv2.mu.Lock() + if len(recv2.got) != 0 { + t.Fatalf("filtered hook received %d deliveries", len(recv2.got)) + } + recv2.mu.Unlock() + + // Retries: fail twice, then succeed; attempts recorded. + recv.mu.Lock() + recv.failNext = 2 + recv.mu.Unlock() + if _, _, code := inst.ssh(t, aliceKey, "", "issue", "close", "alice/proj", "1"); code != 0 { + t.Fatal("close failed") + } + got = recv.waitN(t, 4) + if got[3].event != "issue.closed" { + t.Fatalf("retried event: %s", got[3].event) + } + out, _, _ := inst.ssh(t, aliceKey, "", "webhook", "deliveries", "alice/proj", "--json") + if !strings.Contains(out, `"attempts":3`) { + t.Fatalf("retry attempts not recorded:\n%s", out) + } + + // Dead-letter after max attempts, then manual redelivery revives it. + recv.mu.Lock() + recv.failNext = 99 + recv.mu.Unlock() + if _, _, code := inst.ssh(t, aliceKey, "", "issue", "reopen", "alice/proj", "1"); code != 0 { + t.Fatal("reopen failed") + } + var deadID string + deadlineDL := time.Now().Add(30 * time.Second) + for time.Now().Before(deadlineDL) { + out, _, _ = inst.ssh(t, aliceKey, "", "webhook", "deliveries", "alice/proj", "--json") + var envl struct { + Data []struct { + ID int64 `json:"id"` + Event string `json:"event"` + Status string `json:"status"` + } `json:"data"` + } + json.Unmarshal([]byte(out), &envl) + for _, d := range envl.Data { + if d.Event == "issue.open" && d.Status == "failed" { + deadID = fmt.Sprint(d.ID) + } + } + if deadID != "" { + break + } + time.Sleep(300 * time.Millisecond) + } + if deadID == "" { + t.Fatalf("delivery never dead-lettered:\n%s", out) + } + recv.mu.Lock() + recv.failNext = 0 + prev := len(recv.got) + recv.mu.Unlock() + if _, errOut, code := inst.ssh(t, aliceKey, "", "webhook", "redeliver", "alice/proj", deadID); code != 0 { + t.Fatalf("redeliver: %s", errOut) + } + recv.waitN(t, prev+1) + + // SSRF: on a default instance (allow_local off), local targets are + // rejected at add time. + inst2 := startInstance(t) + k2 := inst2.newKey(t, "a2") + inst2.admin(t, "admin", "user", "create", "a2", "--key", k2+".pub") + if _, _, code := inst2.ssh(t, k2, "", "repo", "create", "a2/r"); code != 0 { + t.Fatal("repo create failed") + } + _, errOut, code := inst2.ssh(t, k2, "", "webhook", "add", "a2/r", "http://127.0.0.1:9/x") + if code != 2 || !strings.Contains(errOut, "SSRF") { + t.Fatalf("local webhook target accepted: exit %d, %s", code, errOut) + } + if _, _, code := inst2.ssh(t, k2, "", "webhook", "add", "a2/r", "ftp://example.com/x"); code != 2 { + t.Fatal("non-http scheme accepted") + } +} @@ -20,6 +20,7 @@ type Config struct { Web Web `toml:"web"` Registration Registration `toml:"registration"` API API `toml:"api"` + Webhooks Webhooks `toml:"webhooks"` Limits Limits `toml:"limits"` Mail Mail `toml:"mail"` } @@ -69,6 +70,12 @@ type API struct { Enabled bool `toml:"enabled"` } +// Webhooks controls outbound delivery. AllowLocal permits endpoints on +// loopback/private addresses (off by default: SSRF). +type Webhooks struct { + AllowLocal bool `toml:"allow_local"` +} + type Limits struct { MaxPackBytes int64 `toml:"max_pack_bytes"` MaxBlobBytes int64 `toml:"max_blob_bytes"` new file mode 100644 @@ -0,0 +1,202 @@ +package control + +import ( + "errors" + "fmt" + "io" + "strconv" + + "gitbay.org/gitbay/internal/policy" + "gitbay.org/gitbay/internal/protocol" + "gitbay.org/gitbay/internal/store" + "gitbay.org/gitbay/internal/webhook" +) + +func init() { + register(Command{Path: []string{"webhook", "add"}, + Summary: "add a webhook: webhook add <owner/name> <url> [--secret <s>] [--events push,issue.created|*]", Run: runWebhookAdd}) + register(Command{Path: []string{"webhook", "list"}, + Summary: "list webhooks: webhook list <owner/name>", ReadOnly: true, Run: runWebhookList}) + register(Command{Path: []string{"webhook", "remove"}, + Summary: "remove a webhook: webhook remove <owner/name> <id>", Run: runWebhookRemove}) + register(Command{Path: []string{"webhook", "deliveries"}, + Summary: "recent deliveries: webhook deliveries <owner/name> [--limit n]", ReadOnly: true, Run: runWebhookDeliveries}) + register(Command{Path: []string{"webhook", "redeliver"}, + Summary: "queue a delivery again: webhook redeliver <owner/name> <delivery-id>", Run: runWebhookRedeliver}) +} + +func runWebhookAdd(c *Ctx, args []string) int { + var path, url, secret string + events := "*" + for i := 0; i < len(args); i++ { + switch args[i] { + case "--secret", "--events": + if i+1 >= len(args) { + return c.fail(protocol.ExitUsage, "%s requires a value", args[i]) + } + if args[i] == "--secret" { + secret = args[i+1] + } else { + events = args[i+1] + } + i++ + default: + if path == "" { + path = args[i] + } else if url == "" { + url = args[i] + } else { + return c.fail(protocol.ExitUsage, "unexpected argument %q", args[i]) + } + } + } + if path == "" || url == "" { + return c.fail(protocol.ExitUsage, "usage: webhook add <owner/name> <url> [--secret <s>] [--events <k1,k2>|*]") + } + repo, code := resolveRepo(c, path, policy.CanAdmin) + if code >= 0 { + return code + } + if err := webhook.ValidateURL(url, c.Cfg.Webhooks.AllowLocal); err != nil { + return c.fail(protocol.ExitUsage, "%v", err) + } + id, err := c.Store.AddWebhook(repo.ID, url, secret, events) + if err != nil { + return c.fail(protocol.ExitFailure, "%v", err) + } + return c.emit(map[string]any{"id": id, "url": url, "events": events}, func(w io.Writer) { + fmt.Fprintf(w, "webhook %d added for %s (%s)\n", id, repo.Path(), events) + }) +} + +func runWebhookList(c *Ctx, args []string) int { + if len(args) != 1 { + return c.fail(protocol.ExitUsage, "usage: webhook list <owner/name>") + } + repo, code := resolveRepo(c, args[0], policy.CanAdmin) + if code >= 0 { + return code + } + hooks, err := c.Store.ListWebhooks(repo.ID) + if err != nil { + return c.fail(protocol.ExitFailure, "%v", err) + } + type out struct { + ID int64 `json:"id"` + URL string `json:"url"` + Events string `json:"events"` + Active bool `json:"active"` + Secret bool `json:"has_secret"` + } + var ds []out + for _, h := range hooks { + ds = append(ds, out{h.ID, h.URL, h.Events, h.Active, h.Secret != ""}) + } + return c.emit(ds, func(w io.Writer) { + for _, d := range ds { + fmt.Fprintf(w, "%d\t%s\t%s\n", d.ID, d.URL, d.Events) + } + }) +} + +func runWebhookRemove(c *Ctx, args []string) int { + if len(args) != 2 { + return c.fail(protocol.ExitUsage, "usage: webhook remove <owner/name> <id>") + } + repo, code := resolveRepo(c, args[0], policy.CanAdmin) + if code >= 0 { + return code + } + id, err := strconv.ParseInt(args[1], 10, 64) + if err != nil { + return c.fail(protocol.ExitUsage, "bad webhook id %q", args[1]) + } + if err := c.Store.RemoveWebhook(repo.ID, id); err != nil { + if errors.Is(err, store.ErrNotFound) { + return c.fail(protocol.ExitNotFound, "no webhook %d on %s", id, repo.Path()) + } + return c.fail(protocol.ExitFailure, "%v", err) + } + return c.emit(map[string]any{"removed": id}, func(w io.Writer) { + fmt.Fprintf(w, "removed webhook %d\n", id) + }) +} + +func runWebhookDeliveries(c *Ctx, args []string) int { + limit := 20 + var path string + for i := 0; i < len(args); i++ { + if args[i] == "--limit" { + if i+1 >= len(args) { + return c.fail(protocol.ExitUsage, "--limit requires a value") + } + n, err := strconv.Atoi(args[i+1]) + if err != nil || n < 1 || n > 200 { + return c.fail(protocol.ExitUsage, "--limit must be 1..200") + } + limit = n + i++ + continue + } + if path != "" { + return c.fail(protocol.ExitUsage, "usage: webhook deliveries <owner/name> [--limit n]") + } + path = args[i] + } + if path == "" { + return c.fail(protocol.ExitUsage, "usage: webhook deliveries <owner/name> [--limit n]") + } + repo, code := resolveRepo(c, path, policy.CanAdmin) + if code >= 0 { + return code + } + ds, err := c.Store.ListDeliveries(repo.ID, limit) + if err != nil { + return c.fail(protocol.ExitFailure, "%v", err) + } + type out struct { + ID int64 `json:"id"` + URL string `json:"url"` + Event string `json:"event"` + Status string `json:"status"` + Attempts int `json:"attempts"` + LastStatus int `json:"last_status,omitempty"` + LastError string `json:"last_error,omitempty"` + } + var rows []out + for _, d := range ds { + rows = append(rows, out{d.ID, d.URL, d.EventKind, d.Status, d.Attempts, d.LastStatus, d.LastError}) + } + return c.emit(rows, func(w io.Writer) { + for _, d := range rows { + extra := "" + if d.LastError != "" { + extra = "\t" + d.LastError + } + fmt.Fprintf(w, "%d\t%s\t%s\t%s (%d attempts)%s\n", d.ID, d.Event, d.URL, d.Status, d.Attempts, extra) + } + }) +} + +func runWebhookRedeliver(c *Ctx, args []string) int { + if len(args) != 2 { + return c.fail(protocol.ExitUsage, "usage: webhook redeliver <owner/name> <delivery-id>") + } + repo, code := resolveRepo(c, args[0], policy.CanAdmin) + if code >= 0 { + return code + } + id, err := strconv.ParseInt(args[1], 10, 64) + if err != nil { + return c.fail(protocol.ExitUsage, "bad delivery id %q", args[1]) + } + if err := c.Store.Redeliver(repo.ID, id); err != nil { + if errors.Is(err, store.ErrNotFound) { + return c.fail(protocol.ExitNotFound, "no delivery %d on %s", id, repo.Path()) + } + return c.fail(protocol.ExitFailure, "%v", err) + } + return c.emit(map[string]any{"requeued": id}, func(w io.Writer) { + fmt.Fprintf(w, "delivery %d requeued\n", id) + }) +} @@ -167,6 +167,11 @@ func (s *Server) preReceive(req Request, dec *json.Decoder, enc *json.Encoder) { // place a hook writes outside its own repository. func (s *Server) postReceive(req Request) { for _, u := range req.Updates { + // Every ref update is an event webhooks can subscribe to. + s.st.RecordEvent(req.RepoID, req.UserID, "push", fmt.Sprintf( + `{"ref":%q,"old":%q,"new":%q,"forced":%v,"deleted":%v}`, + u.Ref, u.Old, u.New, u.IsForce, u.IsDelete)) + branch, ok := cutHeads(u.Ref) if !ok { continue @@ -220,13 +220,34 @@ func (s *Store) SetIssueAssignee(issueID, userID int64, add bool) error { return nil } -// RecordEvent appends to the event log (the forward hook CI will consume). +// RecordEvent appends to the event log and enqueues a delivery for every +// active webhook on the repo whose event filter matches. func (s *Store) RecordEvent(repoID, actorID int64, kind, dataJSON string) error { if dataJSON == "" { dataJSON = "{}" } - _, err := s.DB.Exec( + tx, err := s.DB.Begin() + if err != nil { + return err + } + defer tx.Rollback() + res, err := tx.Exec( "INSERT INTO events (repo_id, actor_id, kind, data_json) VALUES (?, ?, ?, ?)", repoID, actorID, kind, dataJSON) - return err + if err != nil { + return err + } + eventID, err := res.LastInsertId() + if err != nil { + return err + } + if _, err := tx.Exec(` + INSERT INTO webhook_deliveries (webhook_id, event_id) + SELECT id, ? FROM webhooks + WHERE repo_id = ? AND active = 1 + AND (events = '*' OR ',' || events || ',' LIKE '%,' || ? || ',%')`, + eventID, repoID, kind); err != nil { + return err + } + return tx.Commit() } new file mode 100644 @@ -0,0 +1,2 @@ +DROP TABLE webhook_deliveries; +DROP TABLE webhooks; new file mode 100644 @@ -0,0 +1,25 @@ +CREATE TABLE webhooks ( + id INTEGER PRIMARY KEY, + repo_id INTEGER NOT NULL REFERENCES repos(id) ON DELETE CASCADE, + url TEXT NOT NULL, + secret TEXT NOT NULL DEFAULT '', + events TEXT NOT NULL DEFAULT '*', + active INTEGER NOT NULL DEFAULT 1, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) +); +CREATE INDEX webhooks_repo ON webhooks(repo_id); + +CREATE TABLE webhook_deliveries ( + id INTEGER PRIMARY KEY, + webhook_id INTEGER NOT NULL REFERENCES webhooks(id) ON DELETE CASCADE, + event_id INTEGER NOT NULL REFERENCES events(id) ON DELETE CASCADE, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TEXT, + delivered_at TEXT, + failed_at TEXT, + last_status INTEGER, + last_error TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) +); +CREATE INDEX webhook_deliveries_due ON webhook_deliveries(next_attempt_at) + WHERE delivered_at IS NULL AND failed_at IS NULL; new file mode 100644 @@ -0,0 +1,174 @@ +package store + +import ( + "time" +) + +type Webhook struct { + ID int64 + URL string + Secret string + Events string // "*" or comma-separated kinds + Active bool + CreatedAt string +} + +type Delivery struct { + ID int64 + WebhookID int64 + URL string + Secret string + EventID int64 + EventKind string + RepoPath string + Actor string + DataJSON string + EventAt string + Attempts int +} + +type DeliveryStatus struct { + ID int64 + URL string + EventKind string + Status string // pending | delivered | failed + Attempts int + LastStatus int + LastError string + CreatedAt string +} + +func (s *Store) AddWebhook(repoID int64, url, secret, events string) (int64, error) { + res, err := s.DB.Exec( + "INSERT INTO webhooks (repo_id, url, secret, events) VALUES (?, ?, ?, ?)", + repoID, url, secret, events) + if err != nil { + return 0, err + } + return res.LastInsertId() +} + +func (s *Store) ListWebhooks(repoID int64) ([]Webhook, error) { + rows, err := s.DB.Query( + "SELECT id, url, secret, events, active, created_at FROM webhooks WHERE repo_id = ? ORDER BY id", repoID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Webhook + for rows.Next() { + var w Webhook + var active int + if err := rows.Scan(&w.ID, &w.URL, &w.Secret, &w.Events, &active, &w.CreatedAt); err != nil { + return nil, err + } + w.Active = active != 0 + out = append(out, w) + } + return out, rows.Err() +} + +func (s *Store) RemoveWebhook(repoID, hookID int64) error { + res, err := s.DB.Exec("DELETE FROM webhooks WHERE repo_id = ? AND id = ?", repoID, hookID) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} + +// DueDeliveries returns pending deliveries whose time has come, with the +// event and hook context needed to send them. +func (s *Store) DueDeliveries(limit int) ([]Delivery, error) { + rows, err := s.DB.Query(` + SELECT d.id, d.webhook_id, w.url, w.secret, d.event_id, e.kind, + COALESCE(u2.username, o.name, '') || '/' || COALESCE(r.name, ''), + COALESCE(u.username, ''), e.data_json, e.created_at, d.attempts + FROM webhook_deliveries d + JOIN webhooks w ON w.id = d.webhook_id + JOIN events e ON e.id = d.event_id + LEFT JOIN users u ON u.id = e.actor_id + LEFT JOIN repos r ON r.id = e.repo_id + LEFT JOIN users u2 ON r.owner_kind = 'user' AND u2.id = r.owner_id + LEFT JOIN orgs o ON r.owner_kind = 'org' AND o.id = r.owner_id + WHERE d.delivered_at IS NULL AND d.failed_at IS NULL + AND (d.next_attempt_at IS NULL OR d.next_attempt_at <= ?) + ORDER BY d.id LIMIT ?`, fmtTime(time.Now()), limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Delivery + for rows.Next() { + var d Delivery + if err := rows.Scan(&d.ID, &d.WebhookID, &d.URL, &d.Secret, &d.EventID, &d.EventKind, + &d.RepoPath, &d.Actor, &d.DataJSON, &d.EventAt, &d.Attempts); err != nil { + return nil, err + } + out = append(out, d) + } + return out, rows.Err() +} + +func (s *Store) MarkDelivered(id int64, status int) error { + _, err := s.DB.Exec(` + UPDATE webhook_deliveries SET delivered_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), + attempts = attempts + 1, last_status = ?, last_error = NULL WHERE id = ?`, status, id) + return err +} + +// MarkAttemptFailed records a failed attempt; nextAt nil dead-letters it. +func (s *Store) MarkAttemptFailed(id int64, status int, errMsg string, nextAt *time.Time) error { + if nextAt == nil { + _, err := s.DB.Exec(` + UPDATE webhook_deliveries SET failed_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), + attempts = attempts + 1, last_status = ?, last_error = ? WHERE id = ?`, status, errMsg, id) + return err + } + _, err := s.DB.Exec(` + UPDATE webhook_deliveries SET attempts = attempts + 1, last_status = ?, last_error = ?, + next_attempt_at = ? WHERE id = ?`, status, errMsg, fmtTime(*nextAt), id) + return err +} + +func (s *Store) ListDeliveries(repoID int64, limit int) ([]DeliveryStatus, error) { + rows, err := s.DB.Query(` + SELECT d.id, w.url, e.kind, + CASE WHEN d.delivered_at IS NOT NULL THEN 'delivered' + WHEN d.failed_at IS NOT NULL THEN 'failed' + ELSE 'pending' END, + d.attempts, COALESCE(d.last_status, 0), COALESCE(d.last_error, ''), d.created_at + FROM webhook_deliveries d + JOIN webhooks w ON w.id = d.webhook_id + JOIN events e ON e.id = d.event_id + WHERE w.repo_id = ? ORDER BY d.id DESC LIMIT ?`, repoID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []DeliveryStatus + for rows.Next() { + var d DeliveryStatus + if err := rows.Scan(&d.ID, &d.URL, &d.EventKind, &d.Status, &d.Attempts, &d.LastStatus, &d.LastError, &d.CreatedAt); err != nil { + return nil, err + } + out = append(out, d) + } + return out, rows.Err() +} + +// Redeliver resets a delivery for an immediate retry. +func (s *Store) Redeliver(repoID, deliveryID int64) error { + res, err := s.DB.Exec(` + UPDATE webhook_deliveries SET delivered_at = NULL, failed_at = NULL, next_attempt_at = NULL + WHERE id = ? AND webhook_id IN (SELECT id FROM webhooks WHERE repo_id = ?)`, deliveryID, repoID) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} new file mode 100644 @@ -0,0 +1,182 @@ +// Package webhook delivers events to registered endpoints: HMAC-signed +// JSON POSTs with bounded retries, exponential backoff, and dead-lettering. +package webhook + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "net/url" + "time" + + "gitbay.org/gitbay/internal/store" +) + +// ValidateURL rejects URLs a webhook must not target: non-HTTP schemes and, +// unless allowLocal, anything resolving to loopback, private, or link-local +// addresses (SSRF). +func ValidateURL(raw string, allowLocal bool) error { + u, err := url.Parse(raw) + if err != nil { + return fmt.Errorf("invalid URL: %w", err) + } + if u.Scheme != "http" && u.Scheme != "https" { + return fmt.Errorf("webhook URLs must be http or https") + } + if u.Hostname() == "" { + return fmt.Errorf("webhook URL has no host") + } + if allowLocal { + return nil + } + ips, err := net.LookupIP(u.Hostname()) + if err != nil { + return fmt.Errorf("cannot resolve %s: %w", u.Hostname(), err) + } + for _, ip := range ips { + if isForbidden(ip) { + return fmt.Errorf("webhook target %s resolves to a private or local address; refusing (SSRF)", u.Hostname()) + } + } + return nil +} + +func isForbidden(ip net.IP) bool { + return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || + ip.IsLinkLocalMulticast() || ip.IsUnspecified() +} + +type Deliverer struct { + St *store.Store + AllowLocal bool + RetryBase time.Duration // first retry delay; doubles per attempt + MaxAttempts int + client *http.Client +} + +// New builds a deliverer whose dialer re-checks resolved addresses at +// connect time, so a DNS answer that changes after ValidateURL still cannot +// reach private space. +func New(st *store.Store, allowLocal bool, retryBase time.Duration) *Deliverer { + d := &Deliverer{St: st, AllowLocal: allowLocal, RetryBase: retryBase, MaxAttempts: 5} + dialer := &net.Dialer{Timeout: 5 * time.Second} + d.client = &http.Client{ + Timeout: 10 * time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse // never follow redirects + }, + Transport: &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + host, port, err := net.SplitHostPort(addr) + if err != nil { + return nil, err + } + ips, err := net.DefaultResolver.LookupIP(ctx, "ip", host) + if err != nil { + return nil, err + } + for _, ip := range ips { + if !allowLocal && isForbidden(ip) { + return nil, fmt.Errorf("refusing connection to private address %s", ip) + } + } + return dialer.DialContext(ctx, network, net.JoinHostPort(ips[0].String(), port)) + }, + }, + } + return d +} + +// Run polls for due deliveries until ctx is done. +func (d *Deliverer) Run(ctx context.Context) { + tick := time.NewTicker(2 * time.Second) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + due, err := d.St.DueDeliveries(20) + if err != nil { + slog.Error("webhook: listing due deliveries", "err", err) + continue + } + for _, dl := range due { + d.deliver(ctx, dl) + } + } + } +} + +type payload struct { + Event string `json:"event"` + Repo string `json:"repo"` + Actor string `json:"actor,omitempty"` + CreatedAt string `json:"created_at"` + Data json.RawMessage `json:"data"` +} + +func (d *Deliverer) deliver(ctx context.Context, dl store.Delivery) { + body, err := json.Marshal(payload{ + Event: dl.EventKind, Repo: dl.RepoPath, Actor: dl.Actor, + CreatedAt: dl.EventAt, Data: json.RawMessage(dl.DataJSON), + }) + if err != nil { + d.fail(dl, 0, "marshal: "+err.Error()) + return + } + req, err := http.NewRequestWithContext(ctx, "POST", dl.URL, bytes.NewReader(body)) + if err != nil { + d.fail(dl, 0, "request: "+err.Error()) + return + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "gitbay-webhook") + req.Header.Set("X-Gitbay-Event", dl.EventKind) + req.Header.Set("X-Gitbay-Delivery", fmt.Sprint(dl.ID)) + if dl.Secret != "" { + mac := hmac.New(sha256.New, []byte(dl.Secret)) + mac.Write(body) + req.Header.Set("X-Gitbay-Signature-256", "sha256="+hex.EncodeToString(mac.Sum(nil))) + } + + resp, err := d.client.Do(req) + if err != nil { + d.fail(dl, 0, err.Error()) + return + } + io.Copy(io.Discard, io.LimitReader(resp.Body, 4096)) + resp.Body.Close() + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + if err := d.St.MarkDelivered(dl.ID, resp.StatusCode); err != nil { + slog.Error("webhook: marking delivered", "err", err) + } + return + } + d.fail(dl, resp.StatusCode, fmt.Sprintf("endpoint returned %d", resp.StatusCode)) +} + +// fail schedules a retry with exponential backoff, dead-lettering after +// MaxAttempts. +func (d *Deliverer) fail(dl store.Delivery, status int, msg string) { + attempt := dl.Attempts + 1 // the one that just happened + if attempt >= d.MaxAttempts { + if err := d.St.MarkAttemptFailed(dl.ID, status, msg, nil); err != nil { + slog.Error("webhook: dead-lettering", "err", err) + } + slog.Warn("webhook dead-lettered", "delivery", dl.ID, "url", dl.URL, "err", msg) + return + } + next := time.Now().Add(d.RetryBase << (attempt - 1)) + if err := d.St.MarkAttemptFailed(dl.ID, status, msg, &next); err != nil { + slog.Error("webhook: scheduling retry", "err", err) + } +}