package handlers import ( "encoding/json" "fmt" "io" "net/http" "github.com/go-chi/chi" "gorm.io/gorm" "sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/delivery" "sneak.berlin/go/webhooker/internal/logfield" ) const ( // maxWebhookBodySize is the maximum allowed webhook // request body (1 MB). maxWebhookBodySize = 1 << maxBodyShift ) // HandleWebhook handles incoming webhook requests at entrypoint // URLs. func (h *Handlers) HandleWebhook() http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { w.Header().Set("Allow", "POST") http.Error( w, "Method Not Allowed", http.StatusMethodNotAllowed, ) return } entrypointUUID := chi.URLParam(r, "uuid") if entrypointUUID == "" { http.NotFound(w, r) return } entrypoint, ok := h.lookupEntrypoint( w, r, entrypointUUID, ) if !ok { return } // Logged only once the UUID is known to name a real // entrypoint. The UUID comes straight out of the path on // the one unauthenticated endpoint, so logging it before // the lookup let a client write an INFO line per invented // path; the request itself is already in the access log // and a miss is already logged at DEBUG. h.log.Info("webhook request received", "entrypoint_uuid", entrypointUUID, "method", r.Method, "remote_addr", r.RemoteAddr, ) if !entrypoint.Active { http.Error(w, "Gone", http.StatusGone) return } h.processWebhookRequest(w, r, entrypoint) } } // processWebhookRequest reads the body, serializes headers, loads // targets, and delivers the event. // // Nothing about the request itself is authenticated: the entrypoint // UUID in the path is the credential, and reaching here means it // matched an active entrypoint. func (h *Handlers) processWebhookRequest( w http.ResponseWriter, r *http.Request, entrypoint database.Entrypoint, ) { body, ok := h.readWebhookBody(w, r) if !ok { return } headersJSON, err := json.Marshal(r.Header) if err != nil { h.serverError(w, "failed to serialize headers", err) return } targets, err := h.loadActiveTargets(entrypoint.WebhookID) if err != nil { h.serverError(w, "failed to query targets", err) return } h.createAndDeliverEvent( w, r, entrypoint, body, headersJSON, targets, ) } // loadActiveTargets returns all active targets for a webhook. func (h *Handlers) loadActiveTargets( webhookID string, ) ([]database.Target, error) { var targets []database.Target err := h.db.DB().Where( "webhook_id = ? AND active = ?", webhookID, true, ).Find(&targets).Error return targets, err } // lookupEntrypoint finds an entrypoint by UUID path. func (h *Handlers) lookupEntrypoint( w http.ResponseWriter, r *http.Request, entrypointUUID string, ) (database.Entrypoint, bool) { var entrypoint database.Entrypoint result := h.db.DB().Where( "path = ?", entrypointUUID, ).First(&entrypoint) if result.Error != nil { // The receiver is unauthenticated and /webhook/{uuid} // matches any single segment, so this value is entirely // client-chosen on exactly the branch where the lookup // failed. DEBUG is off by default; the cap is what keeps // turning it on from restoring an unbounded write. h.log.Debug( "entrypoint not found", "path", logfield.Truncate( entrypointUUID, logfield.MaxBytes, ), ) http.NotFound(w, r) return entrypoint, false } return entrypoint, true } // readWebhookBody reads and validates the request body size. func (h *Handlers) readWebhookBody( w http.ResponseWriter, r *http.Request, ) ([]byte, bool) { body, err := io.ReadAll( io.LimitReader(r.Body, maxWebhookBodySize+1), ) if err != nil { h.log.Error( "failed to read request body", "error", err, ) http.Error( w, "Bad request", http.StatusBadRequest, ) return nil, false } if len(body) > maxWebhookBodySize { http.Error( w, "Request body too large", http.StatusRequestEntityTooLarge, ) return nil, false } return body, true } // createAndDeliverEvent stores the received event, fans it out to the // webhook's targets, and answers the sender. func (h *Handlers) createAndDeliverEvent( w http.ResponseWriter, r *http.Request, entrypoint database.Entrypoint, body, headersJSON []byte, targets []database.Target, ) { event, tasks, err := h.createAndFanOut( requestEventSource(r, entrypoint, headersJSON, body), targets, ) if err != nil { h.serverError(w, "failed to store webhook event", err) return } h.finishWebhookResponse(w, event, entrypoint, tasks) } // eventSource carries the fields a new event is built from. The // receiver fills it from the live request; the resubmit handler fills // it from a stored event. Both then go through createAndFanOut, so an // event is constructed and fanned out in one place however it entered // the system. type eventSource struct { WebhookID string EntrypointID string Method string HeadersJSON string ContentType string Body []byte // ResubmittedFromID names the event this one copies. Only the // resubmit path sets it. ResubmittedFromID *string } // event builds the row this source stores. func (s eventSource) event() *database.Event { return &database.Event{ WebhookID: s.WebhookID, EntrypointID: s.EntrypointID, Method: s.Method, Headers: s.HeadersJSON, Body: string(s.Body), ContentType: s.ContentType, ResubmittedFromID: s.ResubmittedFromID, } } // requestEventSource describes the event a live receiver request // stores. func requestEventSource( r *http.Request, entrypoint database.Entrypoint, headersJSON, body []byte, ) eventSource { return eventSource{ WebhookID: entrypoint.WebhookID, EntrypointID: entrypoint.ID, Method: r.Method, HeadersJSON: string(headersJSON), ContentType: r.Header.Get("Content-Type"), Body: body, } } // createAndFanOut writes the event and one pending delivery per target // in a single transaction, then hands the tasks to the delivery // engine. It is the only path by which an event and its deliveries are // created, so a resubmitted event is retried, SSRF-guarded and // circuit-broken exactly as a received one is. // // The tasks are returned as well as queued, so a caller can report how // many targets the event went to. func (h *Handlers) createAndFanOut( src eventSource, targets []database.Target, ) (*database.Event, []delivery.Task, error) { webhookDB, err := h.dbMgr.GetDB(src.WebhookID) if err != nil { return nil, nil, fmt.Errorf( "getting webhook database: %w", err, ) } tx := webhookDB.Begin() if tx.Error != nil { return nil, nil, fmt.Errorf( "beginning transaction: %w", tx.Error, ) } event := src.event() err = tx.Create(event).Error if err != nil { tx.Rollback() return nil, nil, fmt.Errorf("creating event: %w", err) } tasks, err := buildDeliveryTasks( tx, event, targets, inlineBody(src.Body), ) if err != nil { tx.Rollback() return nil, nil, err } err = tx.Commit().Error if err != nil { return nil, nil, fmt.Errorf( "committing transaction: %w", err, ) } // Counted here, after the commit: an event exists once it is // durably stored, which is what the delivery counters are // compared against on a dashboard. A resubmitted event counts // too, because it produces deliveries that the delivery side // counts; the resubmit counter is what separates the two. h.mtr.EventReceived() if len(tasks) > 0 { h.notifier.Notify(tasks) } return event, tasks, nil } // inlineBody returns a pointer to body as a string if it fits // within the inline size limit, or nil otherwise. func inlineBody(body []byte) *string { if len(body) < delivery.MaxInlineBodySize { s := string(body) return &s } return nil } // finishWebhookResponse logs the event and writes the HTTP response. func (h *Handlers) finishWebhookResponse( w http.ResponseWriter, event *database.Event, entrypoint database.Entrypoint, tasks []delivery.Task, ) { h.log.Info("webhook event created", "event_id", event.ID, "webhook_id", entrypoint.WebhookID, "entrypoint_id", entrypoint.ID, "target_count", len(tasks), ) w.WriteHeader(http.StatusOK) _, err := w.Write([]byte(`{"status":"ok"}`)) if err != nil { h.log.Error( "failed to write response", "error", err, ) } } // buildDeliveryTasks creates one pending delivery per target in the // transaction and returns the tasks for the delivery engine. The // caller owns the transaction and rolls it back on error. func buildDeliveryTasks( tx *gorm.DB, event *database.Event, targets []database.Target, bodyPtr *string, ) ([]delivery.Task, error) { tasks := make([]delivery.Task, 0, len(targets)) for i := range targets { dlv := &database.Delivery{ EventID: event.ID, TargetID: targets[i].ID, Status: database.DeliveryStatusPending, } err := tx.Create(dlv).Error if err != nil { return nil, fmt.Errorf( "creating delivery for target %s: %w", targets[i].ID, err, ) } tasks = append(tasks, delivery.Task{ DeliveryID: dlv.ID, EventID: event.ID, WebhookID: event.WebhookID, EntrypointID: event.EntrypointID, TargetID: targets[i].ID, TargetName: targets[i].Name, TargetType: targets[i].Type, TargetConfig: targets[i].Config, MaxRetries: targets[i].MaxRetries, Method: event.Method, Headers: event.Headers, ContentType: event.ContentType, Body: bodyPtr, AttemptNum: 1, }) } return tasks, nil }