forgejo/services/webhook/deliver.go
2024-03-24 10:35:06 +01:00

252 lines
7.2 KiB
Go

// Copyright 2019 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package webhook
import (
"context"
"crypto/tls"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"sync"
"time"
webhook_model "code.gitea.io/gitea/models/webhook"
"code.gitea.io/gitea/modules/graceful"
"code.gitea.io/gitea/modules/hostmatcher"
"code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/process"
"code.gitea.io/gitea/modules/proxy"
"code.gitea.io/gitea/modules/queue"
"code.gitea.io/gitea/modules/setting"
"code.gitea.io/gitea/modules/timeutil"
webhook_module "code.gitea.io/gitea/modules/webhook"
"github.com/gobwas/glob"
)
// Deliver creates the [http.Request] (depending on the webhook type), sends it
// and records the status and response.
func Deliver(ctx context.Context, t *webhook_model.HookTask) error {
w, err := webhook_model.GetWebhookByID(ctx, t.HookID)
if err != nil {
return err
}
defer func() {
err := recover()
if err == nil {
return
}
// There was a panic whilst delivering a hook...
log.Error("PANIC whilst trying to deliver webhook task[%d] to webhook %s Panic: %v\nStacktrace: %s", t.ID, w.URL, err, log.Stack(2))
}()
t.IsDelivered = true
handler := GetWebhookHandler(w.Type)
if handler == nil {
return fmt.Errorf("GetWebhookHandler %q", w.Type)
}
if t.PayloadVersion == 1 {
handler = defaultHandler{true}
}
req, body, err := handler.NewRequest(ctx, w, t)
if err != nil {
return fmt.Errorf("cannot create http request for webhook %s[%d %s]: %w", w.Type, w.ID, w.URL, err)
}
// Record delivery information.
t.RequestInfo = &webhook_model.HookRequest{
URL: req.URL.String(),
HTTPMethod: req.Method,
Headers: map[string]string{},
Body: string(body),
}
for k, vals := range req.Header {
t.RequestInfo.Headers[k] = strings.Join(vals, ",")
}
// Add Authorization Header
authorization, err := w.HeaderAuthorization()
if err != nil {
return fmt.Errorf("cannot get Authorization header for webhook %s[%d %s]: %w", w.Type, w.ID, w.URL, err)
}
if authorization != "" {
req.Header.Set("Authorization", authorization)
t.RequestInfo.Headers["Authorization"] = "******"
}
t.ResponseInfo = &webhook_model.HookResponse{
Headers: map[string]string{},
}
// OK We're now ready to attempt to deliver the task - we must double check that it
// has not been delivered in the meantime
updated, err := webhook_model.MarkTaskDelivered(ctx, t)
if err != nil {
log.Error("MarkTaskDelivered[%d]: %v", t.ID, err)
return fmt.Errorf("unable to mark task[%d] delivered in the db: %w", t.ID, err)
}
if !updated {
// This webhook task has already been attempted to be delivered or is in the process of being delivered
log.Trace("Webhook Task[%d] already delivered", t.ID)
return nil
}
// All code from this point will update the hook task
defer func() {
t.Delivered = timeutil.TimeStampNanoNow()
if t.IsSucceed {
log.Trace("Hook delivered: %s", t.UUID)
} else if !w.IsActive {
log.Trace("Hook delivery skipped as webhook is inactive: %s", t.UUID)
} else {
log.Trace("Hook delivery failed: %s", t.UUID)
}
if err := webhook_model.UpdateHookTask(ctx, t); err != nil {
log.Error("UpdateHookTask [%d]: %v", t.ID, err)
}
// Update webhook last delivery status.
if t.IsSucceed {
w.LastStatus = webhook_module.HookStatusSucceed
} else {
w.LastStatus = webhook_module.HookStatusFail
}
if err = webhook_model.UpdateWebhookLastStatus(ctx, w); err != nil {
log.Error("UpdateWebhookLastStatus: %v", err)
return
}
}()
if setting.DisableWebhooks {
return fmt.Errorf("webhook task skipped (webhooks disabled): [%d]", t.ID)
}
if !w.IsActive {
log.Trace("Webhook %s in Webhook Task[%d] is not active", w.URL, t.ID)
return nil
}
resp, err := webhookHTTPClient.Do(req.WithContext(ctx))
if err != nil {
t.ResponseInfo.Body = fmt.Sprintf("Delivery: %v", err)
return fmt.Errorf("unable to deliver webhook task[%d] in %s due to error in http client: %w", t.ID, w.URL, err)
}
defer resp.Body.Close()
// Status code is 20x can be seen as succeed.
t.IsSucceed = resp.StatusCode/100 == 2
t.ResponseInfo.Status = resp.StatusCode
for k, vals := range resp.Header {
t.ResponseInfo.Headers[k] = strings.Join(vals, ",")
}
p, err := io.ReadAll(resp.Body)
if err != nil {
t.ResponseInfo.Body = fmt.Sprintf("read body: %s", err)
return fmt.Errorf("unable to deliver webhook task[%d] in %s as unable to read response body: %w", t.ID, w.URL, err)
}
t.ResponseInfo.Body = string(p)
return nil
}
var (
webhookHTTPClient *http.Client
once sync.Once
hostMatchers []glob.Glob
)
func webhookProxy(allowList *hostmatcher.HostMatchList) func(req *http.Request) (*url.URL, error) {
if setting.Webhook.ProxyURL == "" {
return proxy.Proxy()
}
once.Do(func() {
for _, h := range setting.Webhook.ProxyHosts {
if g, err := glob.Compile(h); err == nil {
hostMatchers = append(hostMatchers, g)
} else {
log.Error("glob.Compile %s failed: %v", h, err)
}
}
})
return func(req *http.Request) (*url.URL, error) {
for _, v := range hostMatchers {
if v.Match(req.URL.Host) {
if !allowList.MatchHostName(req.URL.Host) {
return nil, fmt.Errorf("webhook can only call allowed HTTP servers (check your %s setting), deny '%s'", allowList.SettingKeyHint, req.URL.Host)
}
return http.ProxyURL(setting.Webhook.ProxyURLFixed)(req)
}
}
return http.ProxyFromEnvironment(req)
}
}
// Init starts the hooks delivery thread
func Init() error {
timeout := time.Duration(setting.Webhook.DeliverTimeout) * time.Second
allowedHostListValue := setting.Webhook.AllowedHostList
if allowedHostListValue == "" {
allowedHostListValue = hostmatcher.MatchBuiltinExternal
}
allowedHostMatcher := hostmatcher.ParseHostMatchList("webhook.ALLOWED_HOST_LIST", allowedHostListValue)
webhookHTTPClient = &http.Client{
Timeout: timeout,
Transport: &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: setting.Webhook.SkipTLSVerify},
Proxy: webhookProxy(allowedHostMatcher),
DialContext: hostmatcher.NewDialContextWithProxy("webhook", allowedHostMatcher, nil, setting.Webhook.ProxyURLFixed),
},
}
hookQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "webhook_sender", handler)
if hookQueue == nil {
return fmt.Errorf("unable to create webhook_sender queue")
}
go graceful.GetManager().RunWithCancel(hookQueue)
go graceful.GetManager().RunWithShutdownContext(populateWebhookSendingQueue)
return nil
}
func populateWebhookSendingQueue(ctx context.Context) {
ctx, _, finished := process.GetManager().AddContext(ctx, "Webhook: Populate sending queue")
defer finished()
lowerID := int64(0)
for {
taskIDs, err := webhook_model.FindUndeliveredHookTaskIDs(ctx, lowerID)
if err != nil {
log.Error("Unable to populate webhook queue as FindUndeliveredHookTaskIDs failed: %v", err)
return
}
if len(taskIDs) == 0 {
return
}
lowerID = taskIDs[len(taskIDs)-1]
for _, taskID := range taskIDs {
select {
case <-ctx.Done():
log.Warn("Shutdown before Webhook Sending queue finishing being populated")
return
default:
}
if err := enqueueHookTask(taskID); err != nil {
log.Error("Unable to push HookTask[%d] to the Webhook Sending queue: %v", taskID, err)
}
}
}
}