alice.pds.demo.boltless.dev
This repository has no description
1package webhook
2
3import (
4 "bytes"
5 "context"
6 "crypto/hmac"
7 "crypto/sha256"
8 "encoding/hex"
9 "encoding/json"
10 "fmt"
11 "io"
12 "log/slog"
13 "net/http"
14 "time"
15
16 "github.com/avast/retry-go/v4"
17 "github.com/bluesky-social/indigo/atproto/syntax"
18 "github.com/google/uuid"
19 "tangled.org/core/appview/db"
20 "tangled.org/core/appview/models"
21 "tangled.org/core/appview/notify"
22 "tangled.org/core/hostutil"
23 "tangled.org/core/log"
24 "tangled.org/core/orm"
25)
26
27type Notifier struct {
28 notify.BaseNotifier
29 db *db.DB
30 baseUrl string
31 logger *slog.Logger
32 client *http.Client
33}
34
35func NewNotifier(database *db.DB, baseUrl string, dev bool) *Notifier {
36 return &Notifier{
37 db: database,
38 baseUrl: baseUrl,
39 logger: log.New("webhook-notifier"),
40 // user-supplied webhook URLs are untrusted: block internal address
41 // ranges and don't follow redirects to guard against SSRF.
42 client: hostutil.SafeClient(dev, 30*time.Second),
43 }
44}
45
46var _ notify.Notifier = &Notifier{}
47
48func (w *Notifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) {
49 webhooks, err := w.activeWebhooksForEvent(repo.RepoDid, models.WebhookEventPush)
50 if err != nil {
51 w.logger.Error("failed to get webhooks for repo", "repo_did", repo.RepoDid, "err", err)
52 return
53 }
54 if len(webhooks) == 0 {
55 return
56 }
57
58 payload := w.buildPushPayload(repo, ref, oldSha, newSha, committerDid)
59 payloadBytes, err := json.Marshal(payload)
60 if err != nil {
61 w.logger.Error("failed to marshal push payload", "repo_did", repo.RepoDid, "err", err)
62 return
63 }
64
65 userAgent := "Tangled-Hook/" + newSha[:7]
66 for _, webhook := range webhooks {
67 go w.sendWebhook(ctx, webhook, string(models.WebhookEventPush), payload.Repository.FullName, userAgent, payloadBytes)
68 }
69}
70
71func (w *Notifier) RenameRepo(ctx context.Context, actor syntax.DID, oldRepo, newRepo *models.Repo) {
72 webhooks, err := w.activeWebhooksForEvent(newRepo.RepoDid, models.WebhookEventRepoRenamed)
73 if err != nil {
74 w.logger.Error("failed to get webhooks for repo", "repo_did", newRepo.RepoDid, "err", err)
75 return
76 }
77 if len(webhooks) == 0 {
78 return
79 }
80
81 payload := &models.WebhookRenamePayload{
82 OldName: oldRepo.Name,
83 NewName: newRepo.Name,
84 Repository: buildWebhookRepository(newRepo),
85 Sender: models.WebhookUser{Did: actor.String()},
86 }
87 payloadBytes, err := json.Marshal(payload)
88 if err != nil {
89 w.logger.Error("failed to marshal rename payload", "repo_did", newRepo.RepoDid, "err", err)
90 return
91 }
92
93 userAgent := "Tangled-Hook/rename"
94 for _, webhook := range webhooks {
95 go w.sendWebhook(ctx, webhook, string(models.WebhookEventRepoRenamed), payload.Repository.FullName, userAgent, payloadBytes)
96 }
97}
98
99func (w *Notifier) NewPull(ctx context.Context, pull *models.Pull) {
100 w.pullRequestEvent(ctx, models.WebhookEventPullRequestCreated, "created", pull.OwnerDid, pull)
101}
102
103func (w *Notifier) ResubmitPull(ctx context.Context, pull *models.Pull) {
104 w.pullRequestEvent(ctx, models.WebhookEventPullRequestResubmitted, "resubmitted", pull.OwnerDid, pull)
105}
106
107func (w *Notifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {
108 event, action, ok := pullStateEvent(pull.State)
109 if !ok {
110 return
111 }
112 w.pullRequestEvent(ctx, event, action, actor.String(), pull)
113}
114
115// pullStateEvent maps a pull's state to the webhook event announcing the
116// transition into that state
117func pullStateEvent(state models.PullState) (models.WebhookEvent, string, bool) {
118 switch state {
119 case models.PullMerged:
120 return models.WebhookEventPullRequestMerged, "merged", true
121 case models.PullClosed:
122 return models.WebhookEventPullRequestClosed, "closed", true
123 case models.PullOpen:
124 return models.WebhookEventPullRequestReopened, "reopened", true
125 default:
126 return "", "", false
127 }
128}
129
130func (w *Notifier) pullRequestEvent(ctx context.Context, event models.WebhookEvent, action, sender string, pull *models.Pull) {
131 // pull request events originate from http handlers, whose context is
132 // canceled as soon as the handler returns; detach so in-flight
133 // deliveries are not cut short
134 ctx = context.WithoutCancel(ctx)
135
136 webhooks, err := w.activeWebhooksForEvent(string(pull.RepoDid), event)
137 if err != nil {
138 w.logger.Error("failed to get webhooks for repo", "repo_did", pull.RepoDid, "err", err)
139 return
140 }
141 if len(webhooks) == 0 {
142 return
143 }
144
145 repo, err := db.GetRepo(w.db, orm.FilterEq("repo_did", string(pull.RepoDid)))
146 if err != nil {
147 w.logger.Error("failed to get repo", "repo_did", pull.RepoDid, "err", err)
148 return
149 }
150
151 payload := buildPullRequestPayload(action, repo, pull, sender, w.baseUrl)
152 payloadBytes, err := json.Marshal(payload)
153 if err != nil {
154 w.logger.Error("failed to marshal pull request payload", "repo_did", pull.RepoDid, "err", err)
155 return
156 }
157
158 userAgent := "Tangled-Hook/pull_request"
159 for _, webhook := range webhooks {
160 go w.sendWebhook(ctx, webhook, string(event), payload.Repository.FullName, userAgent, payloadBytes)
161 }
162}
163
164func buildPullRequestPayload(action string, repo *models.Repo, pull *models.Pull, sender, baseUrl string) *models.WebhookPullRequestPayload {
165 htmlUrl := fmt.Sprintf("%s/%s/%s/pulls/%d", baseUrl, repo.Did, repo.Slug(), pull.PullId)
166
167 pullRequest := models.WebhookPullRequest{
168 Number: pull.PullId,
169 Title: pull.Title,
170 Body: pull.Body,
171 State: pull.State.String(),
172 TargetBranch: pull.TargetBranch,
173 Owner: models.WebhookUser{Did: pull.OwnerDid},
174 HtmlUrl: htmlUrl,
175 CreatedAt: pull.Created.Format(time.RFC3339),
176 }
177 if len(pull.Submissions) > 0 {
178 pullRequest.RoundNumber = pull.LastRoundNumber()
179 pullRequest.PatchUrl = fmt.Sprintf("%s/round/%d.patch", htmlUrl, pull.LastRoundNumber())
180 }
181 if pull.PullSource != nil {
182 source := &models.WebhookPullRequestSource{
183 Branch: pull.PullSource.Branch,
184 }
185 if len(pull.Submissions) > 0 {
186 source.Sha = pull.LatestSha()
187 }
188 if pull.IsForkBased() {
189 source.Repo = pull.PullSource.RepoDid.String()
190 }
191 pullRequest.Source = source
192 }
193
194 return &models.WebhookPullRequestPayload{
195 Action: action,
196 PullRequest: pullRequest,
197 Repository: buildWebhookRepository(repo),
198 Sender: models.WebhookUser{Did: sender},
199 }
200}
201
202// Redeliver re-sends a stored delivery via the live send-and-record path,
203// signing with the webhook's current secret.
204func (w *Notifier) Redeliver(ctx context.Context, webhook models.Webhook, prev models.WebhookDelivery) {
205 // recover the repo full name (X-Tangled-Repo header) from the stored payload
206 var meta struct {
207 Repository struct {
208 FullName string `json:"full_name"`
209 } `json:"repository"`
210 }
211 _ = json.Unmarshal([]byte(prev.RequestBody), &meta)
212
213 w.sendWebhook(ctx, webhook, prev.Event, meta.Repository.FullName, "Tangled-Hook/retry", []byte(prev.RequestBody))
214}
215
216func (w *Notifier) activeWebhooksForEvent(repoDid string, event models.WebhookEvent) ([]models.Webhook, error) {
217 webhooks, err := db.GetActiveWebhooksForRepo(w.db, repoDid)
218 if err != nil {
219 return nil, err
220 }
221 var matching []models.Webhook
222 for _, webhook := range webhooks {
223 if webhook.HasEvent(event) {
224 matching = append(matching, webhook)
225 }
226 }
227 return matching, nil
228}
229
230func buildWebhookRepository(repo *models.Repo) models.WebhookRepository {
231 repository := models.WebhookRepository{
232 Name: repo.Name,
233 FullName: fmt.Sprintf("%s/%s", repo.Did, repo.Rkey),
234 Description: repo.Description,
235 Fork: repo.Source != "",
236 HtmlUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Rkey),
237 CloneUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Rkey),
238 SshUrl: fmt.Sprintf("ssh://git@%s/%s/%s", repo.Knot, repo.Did, repo.Rkey),
239 CreatedAt: repo.Created.Format(time.RFC3339),
240 UpdatedAt: repo.Created.Format(time.RFC3339),
241 Owner: models.WebhookUser{
242 Did: repo.Did,
243 },
244 }
245 if repo.Website != "" {
246 repository.Website = repo.Website
247 }
248 if repo.RepoStats != nil {
249 repository.StarsCount = repo.RepoStats.StarCount
250 repository.OpenIssues = repo.RepoStats.IssueCount.Open
251 }
252 return repository
253}
254
255func (w *Notifier) buildPushPayload(repo *models.Repo, ref, oldSha, newSha, committerDid string) *models.WebhookPayload {
256 pusher := committerDid
257 if committerDid == "" {
258 pusher = repo.Did
259 }
260 return &models.WebhookPayload{
261 Ref: ref,
262 Before: oldSha,
263 After: newSha,
264 Repository: buildWebhookRepository(repo),
265 Pusher: models.WebhookUser{
266 Did: pusher,
267 },
268 }
269}
270
271func (w *Notifier) sendWebhook(ctx context.Context, webhook models.Webhook, event, repoFullName, userAgent string, payloadBytes []byte) {
272 deliveryId := uuid.New().String()
273
274 req, err := http.NewRequestWithContext(ctx, "POST", webhook.Url, bytes.NewReader(payloadBytes))
275 if err != nil {
276 w.logger.Error("failed to create webhook request", "webhook_id", webhook.Id, "err", err)
277 return
278 }
279
280 req.Header.Set("Content-Type", "application/json")
281 req.Header.Set("User-Agent", userAgent)
282 req.Header.Set("X-Tangled-Event", event)
283 req.Header.Set("X-Tangled-Hook-ID", fmt.Sprintf("%d", webhook.Id))
284 req.Header.Set("X-Tangled-Delivery", deliveryId)
285 req.Header.Set("X-Tangled-Repo", repoFullName)
286
287 if webhook.Secret != "" {
288 signature := w.computeSignature(payloadBytes, webhook.Secret)
289 req.Header.Set("X-Tangled-Signature-256", "sha256="+signature)
290 }
291
292 delivery := &models.WebhookDelivery{
293 WebhookId: webhook.Id,
294 Event: event,
295 DeliveryId: deliveryId,
296 Url: webhook.Url,
297 RequestBody: string(payloadBytes),
298 }
299
300 retryOpts := []retry.Option{
301 retry.Attempts(3),
302 retry.Delay(1 * time.Second),
303 retry.MaxDelay(10 * time.Second),
304 retry.DelayType(retry.BackOffDelay),
305 retry.LastErrorOnly(true),
306 retry.OnRetry(func(n uint, err error) {
307 w.logger.Info("retrying webhook delivery",
308 "webhook_id", webhook.Id,
309 "attempt", n+1,
310 "err", err)
311 }),
312 retry.Context(ctx),
313 retry.RetryIf(func(err error) bool {
314 return err != nil
315 }),
316 }
317
318 var resp *http.Response
319 err = retry.Do(func() error {
320 var err error
321 resp, err = w.client.Do(req)
322 if err != nil {
323 return err
324 }
325 if resp.StatusCode >= 500 {
326 defer resp.Body.Close()
327 return fmt.Errorf("server error: %d", resp.StatusCode)
328 }
329 return nil
330 }, retryOpts...)
331
332 if err != nil {
333 w.logger.Error("webhook request failed after retries", "webhook_id", webhook.Id, "err", err)
334 delivery.Success = false
335 delivery.ResponseBody = err.Error()
336 } else {
337 defer resp.Body.Close()
338
339 delivery.ResponseCode = resp.StatusCode
340 delivery.Success = resp.StatusCode >= 200 && resp.StatusCode < 300
341
342 bodyBytes, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024))
343 if err != nil {
344 w.logger.Warn("failed to read webhook response body", "webhook_id", webhook.Id, "err", err)
345 } else {
346 delivery.ResponseBody = string(bodyBytes)
347 }
348
349 if !delivery.Success {
350 w.logger.Warn("webhook delivery failed",
351 "webhook_id", webhook.Id,
352 "status", resp.StatusCode,
353 "url", webhook.Url)
354 } else {
355 w.logger.Info("webhook delivered successfully",
356 "webhook_id", webhook.Id,
357 "url", webhook.Url,
358 "delivery_id", deliveryId)
359 }
360 }
361
362 if err := db.AddWebhookDelivery(w.db, delivery); err != nil {
363 w.logger.Error("failed to record webhook delivery", "webhook_id", webhook.Id, "err", err)
364 }
365}
366
367func (w *Notifier) computeSignature(payload []byte, secret string) string {
368 mac := hmac.New(sha256.New, []byte(secret))
369 mac.Write(payload)
370 return hex.EncodeToString(mac.Sum(nil))
371}